競馬予測AIの作成⑱(発走時刻の取得処理の改修)

Python

はじめに

競馬予測システムにおける発走時刻を取得するプログラムを改修しました。
主な変更点は下記となります。

  1. 関数ベースのプログラムからクラスベースに変更しました。
  2. 名前から用途が判るようにメソッド名と変数名を変更しました。
  3. コメントのスタイルをreStructuredTextスタイルに変更しました。

参考にした書籍

bookfan 1号店 楽天市場店
¥3,190 (2024/06/22 09:02時点 | 楽天市場調べ)

プログラムの目的

日本中央競馬会(JRA)のウェブサイトから各開催場所のレース情報を自動的に収集し、発走時刻をCSV形式で保存することです。

プログラムの動作概要

SeleniumとBeautifulSoupを用いてJRAのウェブサイトからデータをスクレイピングします。指定された開催場所ごとのレース情報を抽出し、発走時刻を含むデータを整形してCSVファイルとして保存します。また、各開催場所のリストをpickleファイルとして保存します。

プログラムの主な機能

  • 発走時刻を保存するディレクトリの作成および古いCSVファイルの削除
  • Seleniumを使用してChromeブラウザを自動操作
  • JRAの出馬表またはレース結果ページへの自動ナビゲーション
  • BeautifulSoupを使用して開催場所、ラウンド、レース情報を抽出
  • 抽出したデータを整形し、CSVおよびpickleファイルとして保存

プログラムで生成されるデータ

プログラムを実行すると下記のデータが生成されます。

データ 内容
CSVファイル 各開催場所ごとの発走時刻情報を含むCSVファイル
pickleファイル 開催場所のリストを含むpickleファイル

<CSVファイルのサンプル>

プログラムのユーザインターフェイス

GUIのユーザインターフェイスは用意していません。

プログラムの構成

以下の主なモジュールから構成されます。

モジュール 概要
JraRaceDataFetcherクラス データ取得および整形の主要な機能を提供
main関数 プログラムのエントリーポイントおよびメイン処理の実行

プログラムのフローチャート

<補足>

  • 茶色:最初に実行される箇所
  • マルチプロセスで動作するので最初に実行される箇所は2箇所となります。
  • 緑色:他の関数やメソッドを呼び出さない
  • 灰色:通常の関数、メソッド
  • 関数名、メソッド名の前の数字はプログラム内の行数

プログラムの関数とメソッドの概要

JraRaceDataFetcherクラス

メソッド名 行数 概要
__init__ 36 JraRaceDataFetcherクラスのコンストラクタ。初期設定ファイルを受け取り、開催場所の辞書と保存パスを設定する。
setup_directory 43 発走時刻を保存するディレクトリを設定し、古いCSVファイルを削除する。
setup_webdriver 55 WebDriverをセットアップし、Chromeブラウザのインスタンスを生成する。
navigate_to_jra_page 67 JRAの出馬表またはレース結果へのナビゲーションを行い、BeautifulSoupオブジェクトを返す。
extract_location_info 109 BeautifulSoupオブジェクトから開催場所情報を抽出する。
fetch_location_urls 122 BeautifulSoupオブジェクトから開催場所のURLを取得する。
preparing_raceinfo_data 135 レース情報ページから不要なタグや改行を削除し、BeautifulSoupオブジェクトを返す。
extract_round_urls 160 BeautifulSoupオブジェクトからラウンドのURLを抽出する。
get_race_times 174 レースの発走時刻を抽出し、データフレームとして返す。
save_cell_time 201 発走時刻を抽出してCSVに保存する。
get_URL_of_each_racetrack_round 216 開催場所ごとのラウンドのURLを取得する。
extract_and_format_of_race_info 242 レース情報を抽出し、整形してデータフレームに追記保存する。
save_location_name 266 開催場所の名前を保存する。

関数

関数名 行数 概要
main 286 プログラムのメイン処理を実行する。発走時刻のディレクトリ設定、URL取得、レース情報の抽出、データの保存、プログラムの実行時間を表示する。

プログラムの実行環境

本プログラムは、Pythonのバージョン3.12.2で実行できることを確認しています。

本プログラムを実行するために必要なパッケージをインストールします。
Pythonの仮想環境を起動し、pipコマンドで下記のパッケージをインストールしてください。

(venv)> pip install pandas
(venv)> pip install requests
(venv)> pip install beautifulsoup4
(venv)> pip install tqdm
(venv)> pip install selenium
(venv)> pip install webdriver_manager

プログラムの実行

本プログラムは競馬予測AIのmain.pyから呼び出されます。競馬予測AIをダウンロードし、web_scrapingフォルダ内のget_cellTime.pyを差し替えてください。

プログラム実行後のCSVファイルとpklファイルの出力先

config.iniの下記の箇所で設定することができます。

# 開催場所ごとの発走時刻情報を保存しているパス
cell_time_path = ./data/celltime/

プログラムの全コード

# JRAから開催場所ごとのレースの発走時刻を取得する
# 呼び出し元:
#   main.py
# 実行前の設定:
#   config.iniのcell_time_pathで取得した情報の保存先を指定しておく
# 実行方法:
#   main.pyから呼び出されるが、個別に実行することも可能
#   個別に実行する場合は、仮想環境で下記のコマンドを実行する
#   (venv) python get_cellTime.py
# 出力内容:
#   location_list.pkl
#   開催場所ごとの発走時刻を記載したcsvファイル

import os.path
import os
import time
import pickle
import glob
import pandas as pd
import requests
import re
import urllib
import configparser

import selenium
from bs4 import BeautifulSoup
from tqdm import tqdm

from selenium import webdriver
from selenium.webdriver.chrome.service import Service as ChromeService
from selenium.webdriver.common.by import By
from webdriver_manager.chrome import ChromeDriverManager


class JraRaceDataFetcher:
    def __init__(self, inifile):
        # 開催場所のローマ字表記と日本語表記の組み合わせ
        self.loc_dict = dict(inifile['loc_romaji_jp'])
        self.cell_time_path = inifile['settings']['cell_time_path']
        self.error_count = 0
        self.max_errors = 3  # 最大エラー許容回数

    def setup_directory(self):
        """
        発走時刻を保存するディレクトリを設定し、古いCSVファイルを削除する。
        :return: None
        """
        # 発走時刻を保存するディレクトリがなければ作成する
        os.makedirs(self.cell_time_path, exist_ok=True)

        # 過去に取得した発走時刻のCSVを消去する
        for i in glob.glob(self.cell_time_path + '*.csv'):
            os.remove(i)

    def setup_webdriver(self):
        """
        WebDriverをセットアップし、Chromeブラウザのインスタンスを生成する。
        :return:ChromeブラウザのWebDriverインスタンス
        :rtype:selenium.webdriver.Chrome
        """
        # ChromeDriverを自動インストールし、サービスとして開始する
        service = ChromeService(executable_path=ChromeDriverManager().install())
        browser = webdriver.Chrome(service=service)
        browser.implicitly_wait(20)  # 最大待機時間(秒)
        return browser

    def navigate_to_jra_page(self, access_type):
        """
        JRAの出馬表またはレース結果へのナビゲーションし、BeautifulSoupオブジェクトを返す。
        1. WebDriverをセットアップして、JRAのトップページを開く。
        2. 指定されたページタイプ(出馬表またはレース結果)に応じて、該当ページにナビゲートする。
        3. ナビゲーション後のページを取得し、BeautifulSoupオブジェクトを生成して返す。

        :param str access_type: アクセスするページの種類。'entries'(出馬表)または 'results'(レース結果)のいずれか。
        :return: ナビゲート後のページのBeautifulSoupオブジェクト
        :rtype: bs4.BeautifulSoup
        """
        try:
            browser = self.setup_webdriver()
            browser.get('https://www.jra.go.jp/')
            # 出馬表またはレース結果ページへのナビゲーション
            navigation_steps = {
                'entries': ('//*[@id="quick_menu"]/div/ul/li[2]/a',
                            '//*[@id="main"]/div[2]/div/div/div[1]/a',
                            '//*[@id="race_list"]/tbody/tr[1]/th/a'),
                'results': ('//*[@id="quick_menu"]/div/ul/li[2]/a',
                            '//*[@id="main"]/div[2]/div/div/div[1]/a',
                            '//*[@id="race_list"]/tbody/tr[1]/td[8]/a')
            }
            for step in navigation_steps[access_type]:
                browser.find_element(By.XPATH, step).click()
            # BeautifulSoupオブジェクトの作成
            res = requests.get(browser.current_url)
            browser.quit()
            return BeautifulSoup(res.content, 'lxml')
        except selenium.common.exceptions.NoSuchElementException as err:
            self.error_count += 1
            print(f'エラーが発生しました: {err}')
            if self.error_count >= self.max_errors:
                print('最大エラー回数に達したため、処理を終了します。')
                exit(1)
            else:
                print(f'リトライします ({self.error_count}/{self.max_errors})')
                return self.navigate_to_jra_page(access_type)
        except Exception as err:
            print(f'予期しないエラーが発生しました: {err}')
            exit(1)

    def extract_location_info(self, soup):
        """
        開催場所情報を抽出する。
        1. 指定されたBeautifulSoupオブジェクトから開催場所の情報を含むテキストを取得する。
        2. 不要な文字(数字、回、日)を削除し、開催場所をリストとして返す。

        :param bs4.BeautifulSoup soup: BeautifulSoupオブジェクト
        :return: 開催場所のリスト
        :rtype: list
        """
        locations = soup.find('div', class_='link_list multi div3 center mid narrow').text
        return re.sub(r'[0-9回日]', '', locations).split()

    def fetch_location_urls(self, soup):
        """
        開催場所のURLを取得する。
        1. 指定されたBeautifulSoupオブジェクトから開催場所のリンクを含むaタグを全て取得する。
        2. 各リンクのhref属性を基にフルURLを生成し、リストとして返す。

        :param bs4.BeautifulSoup soup: BeautifulSoupオブジェクト
        :return: 開催場所のURLリスト
        :rtype: list
        """
        base_url = 'https://www.jra.go.jp'
        return [urllib.parse.urljoin(base_url, link['href']) for link in soup.select('.link_list.multi.div3.center.mid.narrow a')]

    def preparing_raceinfo_data(self, url):
        """
        レース情報ページから不要なタグや改行を削除し、BeautifulSoupオブジェクトを返す。
        1. 指定されたURLからページを取得し、BeautifulSoupオブジェクトを生成する。
        2. ブリンカー着用の馬のタグ(horse_icon blinker)を削除する。
        3. 不要な改行を削除する。

        :param str url: レース情報ページのURL
        :return: 整形されたレース情報のBeautifulSoupオブジェクト
        :rtype: bs4.BeautifulSoup
        """
        response = requests.get(url)
        response.encoding = response.apparent_encoding
        soup = BeautifulSoup(response.text, 'html5lib')

        # ブリンカー着用の馬のタグを削除
        for tag in soup.find_all('span', class_='horse_icon blinker'):
            tag.decompose()

        # 不要な改行を削除
        for tag in soup.find_all('td', class_='num'):
            tag.string = tag.get_text(strip=True)

        return soup

    def extract_round_urls(self, soup):
        """
        ラウンドのURLを抽出する。
        1. 指定されたBeautifulSoupオブジェクトからラウンド情報を含むulタグを取得する。
        2. 各ラウンドのリンクを含むaタグのhref属性を基にフルURLを生成し、リストとして返す。

        :param bs4.BeautifulSoup soup: BeautifulSoupオブジェクト
        :return: ラウンドのURLリスト
        :rtype: list
        """
        base_url = 'https://www.jra.go.jp'
        rounds_info = soup.find('ul', attrs={'class': 'nav race-num mt15'})
        return [urllib.parse.urljoin(base_url, a['href']) for a in rounds_info.find_all('a')]

    def get_race_times(self, soup, url):
        """
        レースの発走時刻を抽出し、データフレームとして返す。
        1. 指定されたBeautifulSoupオブジェクトから開催場所と日付の情報を抽出する。
        2. 発走時刻を抽出し、整形する。
        3. 開催場所、日付、発走時刻、レース情報のURLを含むデータフレームを作成する。

        :param bs4.BeautifulSoup soup: BeautifulSoupオブジェクト
        :param str url: レース情報ページのURL
        :return: 開催場所、日付、発走時刻、レース情報のURLを含むデータフレーム
        :rtype: pandas.DataFrame
        """
        # 開催場所と日付の情報を抽出
        race_base_info = soup.find('div', class_='cell date').text.split()
        race_location, race_date = race_base_info[1].replace('日', '日目'), race_base_info[0].split('(')[0]

        # 発走時刻の抽出と整形
        cell_time = (soup.find('div', class_='cell time').text.replace('発走時刻:', '').
                     replace('\n', '').replace('分', '').replace(' ', '').
                     replace('時', ':'))

        # データフレームの作成
        cell_time_info = pd.DataFrame([[race_location, race_date, cell_time, url]],
                                      columns=['location', 'race_date', 'cell_time', 'race_info_url'])

        return cell_time_info

    def save_cell_time(self, cell_time_info, loc_name):
        """
        発走時刻を抽出してCSVに保存する。
        1. 指定された開催場所の発走時刻情報をフィルタリングする。
        2. フィルタリングされた発走時刻情報をCSVファイルとして保存する。

        :param pandas.DataFrame cell_time_info: 発走時刻情報を含むデータフレーム
        :param str loc_name: 保存する開催場所の名前
        :return: None
        """
        # 発走時刻を抽出してCSVに保存
        csv_file_path = os.path.join(self.cell_time_path, f'cell_time_info_{loc_name}.csv')
        filtered_info = cell_time_info[cell_time_info['location'].str.contains(loc_name)][['race_date', 'cell_time', 'race_info_url']]
        filtered_info.to_csv(csv_file_path, encoding='cp932', index=False, errors='ignore')

    def get_URL_of_each_racetrack_round(self):
        """
        開催場所ごとのラウンドのURLを取得する。
        1. 出走表ページにナビゲートし、開催場所の情報とそのURLを取得する。
        2. 各開催場所ごとのラウンドのURLを取得する。

        :return: ラウンドのURLリスト、開催場所のリスト、開催場所のURLリスト
        :rtype: tuple (list, list, list)
        """
        # 出走表に取得する
        soup = self.navigate_to_jra_page('entries')

        # レース開催の場所、レース開催場所のURLを取得する
        location_list = self.extract_location_info(soup)
        location_url = self.fetch_location_urls(soup)

        # 開催場所ごとのラウンドのURLを取得する
        round_url = []
        for i in location_url:
            res = requests.get(i)
            res.encoding = res.apparent_encoding  # resに含まれるテキストから文字コードを自動設定する
            soup = BeautifulSoup(res.text, 'html5lib')
            round_url.extend(self.extract_round_urls(soup))

        return round_url, location_list, location_url

    def extract_and_format_of_race_info(self, round_url):
        """
        レース情報を抽出し、整形してデータフレームに追記保存する。
        1. 指定されたラウンドURLリストを順に処理し、各URLからレース情報を抽出する。
        2. 抽出した情報を整形し、発走時刻情報をデータフレームに追加する。

        :param list round_url: ラウンドのURLリスト
        :return: 発走時刻情報を含むデータフレーム
        :rtype: pandas.DataFrame
        """
        # レース情報の抽出と整形を行い、データフレームに追記保存する
        cell_time_info = pd.DataFrame()
        print('発走時刻の取得を開始します')

        for url in tqdm(round_url):
            soup = self.preparing_raceinfo_data(url)
            # 発走時刻を取得する
            try:
                cell_time_info = pd.concat([cell_time_info, self.get_race_times(soup, url)])
            except BaseException as err:
                print(f'予測に必要なデータを取得できません:{err}')

        return cell_time_info

    def save_location_name(self, location_list):
        """
        開催場所の名前を保存する。
        1. 開催場所のリストからローマ字表記のリストを作成する。
        2. ローマ字表記のリストをpickleファイルとして保存する。

        :param list location_list: 開催場所のリスト
        :return: None
        """
        # 開催場所(中山、東京など)のloc_dictのvalue(nakayama,tokyoなど)を取得する
        loc_list = []
        for location in location_list:
            key = [k for k, v in self.loc_dict.items() if v == location][0]
            loc_list.append(key)

        # 開催場所のリストを保存する
        with open(self.cell_time_path + 'location_list.pkl', 'wb') as f:
            pickle.dump(loc_list, f)


def main(inifile):
    """
    プログラムのメイン処理を実行する。
    1. プログラムの開始時刻を取得する。
    2. JraRaceDataFetcherオブジェクトを作成する。
    3. 発走時刻を保存するフォルダがなければ作成し、過去のCSVファイルを消去する。
    4. 開催場所ごとのラウンドのURLを取得する。
    5. ラウンドURLからレース情報を抽出し、整形する。
    6. 各開催場所の発走時刻をCSVファイルとして保存する。
    7. 開催場所の名前をpickleファイルとして保存する。
    8. プログラムの実行時間を表示する。

    :param dict inifile: 初期設定ファイル
    :return: None
    """
    # プログラムの開始時刻を取得する
    start_time = time.time()

    jra_data_fetcher = JraRaceDataFetcher(inifile)

    # 発走時刻を保存するフォルダがなければ作成する
    # 過去に取得した発走時刻のCSVを消去する
    jra_data_fetcher.setup_directory()

    # 開催場所ごとのラウンドのURLを取得する
    round_url, location_list, location_url = jra_data_fetcher.get_URL_of_each_racetrack_round()

    # レース情報の抽出と整形を行う
    cell_time_info = jra_data_fetcher.extract_and_format_of_race_info(round_url)

    # 開催場所の発走時刻を保存する
    for location in location_list:
        jra_data_fetcher.save_cell_time(cell_time_info, location)

    # 開催場所の名前を保存する
    jra_data_fetcher.save_location_name(location_list)

    # プログラムの実行時間を表示する
    print(os.path.basename(__file__), f'実行時間: {time.time() - start_time:.2f}秒')


if __name__ == "__main__":
    # 設定ファイルを読み込む
    inifile = configparser.ConfigParser(interpolation=None)
    inifile.read('./config.ini', 'UTF-8')
    main(inifile)

以上です。

コメント

タイトルとURLをコピーしました