Otama's Playground

AIで遊んだ結果などをつらつら載せていきます。

競馬予想AIのためのデータ収集!スクレイピングでレース結果を簡単取得する方法

競馬予想AIを学習させたらどれくらいの精度が出るか気になったので、前準備としてネット上からスクレイピングしてくるスクリプトを書いてみました。

使用するデータ

netkeibaさんからデータをいただくことにします。

www.netkeiba.com

公式からアナウンスがあるように、負荷をかけるようなリクエストをしてしまうと二度とアクセスできない体にされてしまう可能性があります(IP遮断とかでしょうか...?)。netkeibaさん側に迷惑をかけないためにも、自衛のためにも間隔を空けずに連続でリクエストするようなことはしないようにお願いします。

support.keiba.netkeiba.com

利用規約 - netkeiba.com -

どのデータを取得するか

  • レース条件(レース番号、レース名の下あたりにある諸々)
  • レース結果テーブル
  • 払い戻しテーブル

2歳未勝利|2024年8月11日 | 競馬データベース - netkeiba

コード

requestのラッパーを作成して、ラッパー経由でリクエストする形にしてます。

from module import RequestWrapper
import pandas as pd
import io
from bs4 import BeautifulSoup
import re
import time, requests

# リクエスト制御用のラッパー(基本的にnetkeibaしかアクセスしないため簡易的な制御)
class RequestWrapper(object):
    def __init__(self, request_interval_milli = 1000):
        self.request_interval_milli = request_interval_milli
        self.last_requested_at = 0

    def get(self, url):
        target_epochmilli = self.last_requested_at + self.request_interval_milli
        self.sleep_until(target_epochmilli)

        response = requests.get(url)
        self.last_requested_at = time.time()*1000
        
        return response

    def sleep_until(self, target_epochmilli):
        epochmilli = time.time()*1000
        sleep_sec = (target_epochmilli - epochmilli)/1000
        if(sleep_sec > 0):
            time.sleep(sleep_sec)

# レース情報を抽出する
def extract_race_info_from_soup(soup):
    all_info = soup.find("div", attrs={"class": "data_intro"})
    race_data = all_info.find("dl", attrs={"class": "racedata fc"})

    # レース番号の取得
    race_number = race_data.find("dt").get_text().replace("\n", "").replace(" ", "")

    # レースのコンディションを取得
    race_condition1 = race_data.find("dd").find("p").find("span").get_text()
    parsed_race_condition1 = race_condition1.split(u'\xa0/\xa0')

    # レースのコンディションを取得(小さいテキスト)
    race_condition2 = all_info.find("p", attrs={"class":"smalltxt"}).get_text()
    parsed_race_condition2 = race_condition2.split()

    merged =  [race_number] + parsed_race_condition1 + parsed_race_condition2
    return merged

# 結果テーブルを抽出する
def extract_result_table_from_soup(soup):
    # tableタグを抜き出す
    table = soup.find("table", attrs={"summary":"レース結果"})
    rows = table.findAll("tr")

    # カラム名を抽出
    column_names = [column_name.get_text().replace("\n", "") for column_name in rows[0].findAll("th")]

    # 行毎にパースする
    parsed_rows = []
    for row in rows[1:]:
        parsed_row = []
        for cell in row.findAll("td"):
            text = cell.get_text().replace("\n", "")

            a_tag = cell.find("a")
            if(a_tag):
                text += "(" + a_tag.get("href") + ")"
            parsed_row.append(text)
        parsed_rows.append(parsed_row)

    # データフレームに埋め込む
    extracted = pd.DataFrame(parsed_rows, columns = column_names)
    return extracted

# 払い戻しテーブルを抽出する
def extract_payout_table_from_soup(soup):
    # tableタグを抜き出す
    tables = soup.findAll("table", attrs={"class":"pay_table_01","summary":"払い戻し"})

    index_names = []
    parsed_rows = []
    for table in tables:
        for row in table.findAll("tr"):
            index_names += [th.get_text() for th in row.findAll("th")]

            parsed_row = []
            for cell in row.findAll("td"):
                for br in cell.findAll("br"):
                    br.replace_with(";")
                text = cell.get_text()
                parsed_row.append(text)
            parsed_rows.append(parsed_row)

    # データフレームに埋め込む
    extracted = pd.DataFrame(parsed_rows, index = index_names)

    return extracted

# レースIDからレース結果を取り出す
def scrape_race_info(requestWrapper: RequestWrapper, race_id: int):
    # リクエスト
    url = f"https://db.netkeiba.com/race/{race_id}"
    html = requestWrapper.get(url)
    html.encoding = "EUC-JP"
    soup = BeautifulSoup(html.text, "html.parser")

    # レース情報のパース
    extracted_race_info = extract_race_info_from_soup(soup)

    # 結果テーブルのパース
    extracted_result_table = extract_result_table_from_soup(soup)

    # 払い戻しテーブルのパース
    extracted_payout_table = extract_payout_table_from_soup(soup)

    return extracted_race_info, extracted_result_table, extracted_payout_table

requestWrapper = RequestWrapper(request_interval_milli=1000)
result = scrape_race_info(requestWrapper, 202404030303)

print(result[0])
print(result[1].to_csv())
print(result[2].to_csv())

実行結果

ほぼ生データのままですが、抽出してくることができました。

レース情報のパース結果

['3R', 'ダ左1200m', '天候 : 晴', 'ダート : 良', '発走 : 11:15', '2024年8月17日', '3回新潟3日目', '2歳未勝利', '(混)[指](馬齢)']

source: https://db.netkeiba.com/race/202404030303

レース結果テーブルのパース結果

人の名前とか入ってるので載せません。気になる方は実行して確かめてみてください。

払い戻しテーブルのパース結果

0 1 2
単勝 2 320 2
複勝 2; 5; 13 130; 120; 230 2; 1; 5
枠連 2 - 3 370 1
馬連 2 - 5 400 1
ワイド 2 - 5; 2 - 13; 5 - 13 210; 160; 730 1; 16; 8
馬単 2 → 5 890 2
三連複 2 - 5 - 13 3,560 9
三連単 2 → 5 → 13 14,730 35

source: https://db.netkeiba.com/race/202404030303

最後に

データセット作成に関するTODOとしては以下かなと思います。

  • 馬情報、騎手情報などのスクレイピングコード
  • スクレイピングしたデータに前処理を加えて保存するコード

そのうち記事にすると思うので興味ある方はまた見に来てください。