Chapter 11無料公開

【新】modules/preparing/の解説(前半)

競馬予想で始めるデータ分析・機械学習
競馬予想で始めるデータ分析・機械学習
2022.07.26に更新

概要

modules/preparing/には、スクレイピングなどのデータを取得するためのモジュール群が入っていて、ディレクトリ構成は以下のようになっています。

modules/preparing
├── __init__.py
├── _scrape_race_id_list.py     ・・・レースの開催日から、スクレイピング対象のレースidを抽出する
├── _scrape_html.py             ・・・必要なデータをスクレイピングし、htmlの形で保存する
├── _get_rawdata.py             ・・・取得したhtmlからテーブルデータを抽出する
└── _scrape_shutuba_table.py    ・・・実際に当日レースを賭ける際のデータを準備する

_scrape_html.pyでは、netkeiba.comをスクレイピングし、テーブルデータに変換する前のhtmlを、一度data/html/に保存しておきます。こうすることで、「新しく列を追加したい」などの場合に、もう一度スクレイピングせずに済みます。

_get_rawdata.pyでは、スクレイピングしたhtmlを加工して「レース結果テーブル」「レース情報テーブル」「払戻テーブル」「馬の加工成績テーブル」「血統テーブル」の5つのテーブルデータを作成し、これをrawデータとしてdata/raw/配下に保存します。

実際にレースを賭ける際は、レース直前に出馬表ページからデータを取得する必要があるので、別モジュールの_scrape_shutuba_table.pyでスクレイピング〜rawデータ生成の処理を行います。

スクレイピング
preparingの処理の全体像(クリックして拡大)

ソースコード解説

この章では、

  • __init__.py
  • _scrape_race_id_list.py
  • _scrape_html.py

について解説します。

  • _get_rawdata.py
  • _scrape_shutuba_table.py

については次章で解説します。

__init__.py

ソースコード
from ._scrape_race_id_list import scrape_kaisai_date, scrape_race_id_list
from ._scrape_html import scrape_html_horse, scrape_html_ped, scrape_html_race,\
    scrape_html_horse_with_master
from ._get_rawdata import get_rawdata_horse_results, get_rawdata_info, get_rawdata_peds,\
    get_rawdata_results, get_rawdata_return, update_rawdata
from ._scrape_shutuba_table import scrape_shutuba_table, scrape_horse_id_list

constants/__init__.pyと同様なので省略

_scrape_race_id_list.py

ソースコード
import pandas as pd
import time
import re
from tqdm.notebook import tqdm
from bs4 import BeautifulSoup
from urllib.request import urlopen
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.chrome.options import Options

from modules.constants import UrlPaths


def scrape_kaisai_date(from_: str, to_: str):
    """
    yyyy-mmの形式でfrom_とto_を指定すると、間のレース開催日一覧が返ってくる関数。
    to_の月は含まないので注意。
    """
    print('getting race date from {} to {}'.format(from_, to_))
    # 間の年月一覧を作成
    date_range = pd.date_range(start=from_, end=to_, freq="M")
    # 開催日一覧を入れるリスト
    kaisai_date_list = []
    for year, month in tqdm(zip(date_range.year, date_range.month), total=len(date_range)):
        #取得したdate_rangeから、スクレイピング対象urlを作成する。
        #urlは例えば、https://race.netkeiba.com/top/calendar.html?year=2022&month=7 のような構造になっている。
        query = [
            'year=' + str(year),
            'month=' + str(month),
        ]
        url = UrlPaths.CALENDAR_URL + '?' + '&'.join(query)
        html = urlopen(url).read()
        time.sleep(1)
        soup = BeautifulSoup(html, "html.parser")
        a_list = soup.find('table', class_='Calendar_Table').find_all('a')
        for a in a_list:
            kaisai_date_list.append(re.findall('(?<=kaisai_date=)\d+', a['href'])[0])
    return kaisai_date_list
    
def scrape_race_id_list(kaisai_date_list: list, from_shutuba=False, waiting_time=10):
    """
    開催日をyyyymmddの文字列形式でリストで入れると、レースid一覧が返ってくる関数。
    レース前日準備のためrace_idを取得する際には、from_shutuba=Trueにする。
    ChromeDriverは要素を取得し終わらないうちに先に進んでしまうことがあるので、その場合の待機時間をwaiting_timeで指定。
    """
    race_id_list = []
    options = Options()
    options.add_argument('--headless') #ヘッドレスモード(ブラウザが立ち上がらない)
    driver = webdriver.Chrome(options=options)
    #画面サイズをなるべく小さくし、余計な画像などを読み込まないようにする
    driver.set_window_size(8, 8)
    print('getting race_id_list')
    for kaisai_date in tqdm(kaisai_date_list):
        try:
            query = [
                'kaisai_date=' + str(kaisai_date)
            ]
            url = UrlPaths.RACE_LIST_URL + '?' + '&'.join(query)
            print('scraping: {}'.format(url))
            driver.get(url)
            try:
                # 取得し終わらないうちに先に進んでしまうのを防ぐ
                time.sleep(1)
                a_list = driver.find_element(By.CLASS_NAME, 'RaceList_Box').find_elements(By.TAG_NAME, 'a')
            except:
                #それでも取得できなかったらもう10秒待つ
                print('waiting more {} seconds'.format(waiting_time))
                time.sleep(waiting_time)
                a_list = driver.find_element(By.CLASS_NAME, 'RaceList_Box').find_elements(By.TAG_NAME, 'a')
            for a in a_list:
                if from_shutuba:
                    race_id = re.findall('(?<=shutuba.html\?race_id=)\d+', a.get_attribute('href'))
                else:
                    race_id = re.findall('(?<=result.html\?race_id=)\d+', a.get_attribute('href'))
                if len(race_id) > 0:
                    race_id_list.append(race_id[0])
        except Exception as e:
            print(e)
            break
    driver.close()
    return race_id_list

このモジュールでは、スクレイピング対象のURL一覧を取得します。netkeiba.comでは、

https://db.netkeiba.com/race/201901010101

のように、

https://db.netkeiba.com/race/ + (race_id)

という構造をしたURLに、過去行われたレース結果が入っているので、まずは

race_idの一覧を取得して、スクレイピング対象ページの一覧を作成する

というのがやるべきこととなってきます。

そのために、まずはレースが開催されている日の一覧を取得します。例えば2021年に行われた全てのレースの開催日一覧を取得したい場合、

kaisai_date_list = scrape_kaisai_date(from_='2021-01', to_='2022-01')

とすれば開催日一覧をyyyymmddの文字列形式で取得することができます。(※to_の月は含まれないので注意)これを

race_id_list = scrape_race_id_list(kaisai_date_list)

と入れることで、レースid一覧を取得することができます。

取得の仕組みは以下のようになっています。

scrape_kaisai_date()

2022年1月の開催日程一覧は、以下のようなURLのカレンダーページで確認することができます。

https://race.netkeiba.com/top/calendar.html?year=2022&month=1

カレンダー

Google Chromeでカレンダーの日付の部分を右クリック→「検証」を見てみると、カレンダーテーブル<table class="Calendar_Table">の中で、開催がある日付のtdタグのみaタグが存在し、href=kaisai_date=以下にyyyymmdd形式で開催日の文字列が入っていることが分かります。

aタグ

よって、この文字列を以下のコードで抽出します。

# <table class="Calendar_Table">タグ配下にある、aタグ一覧を取得
a_list = soup.find('table', class_='Calendar_Table').find_all('a')
# aタグのhref属性から、開催日を取得
for a in a_list:
    kaisai_date_list.append(re.findall('(?<=kaisai_date=)\d+', a['href'])[0])

これをfrom_とto_の間の全ての月に渡って行うことで、開催日一覧を取得することができます。

scrape_race_id_list()

開催日が特定できたら、その開催日におけるrace_id一覧を、以下のページから取得します。

(例えば2022年1月5日の場合)
https://race.netkeiba.com/top/race_list.html?kaisai_date=20220105

開催レース一覧ページ

同様にhtmlを見てみると、<div class="RaceList_Box">の中のaタグのhref=の部分にrace_idの文字列が入っていることが分かるので、この部分を抽出します。

# <div class="RaceList_Box">配下のaタグ一覧を抽出
a_list = driver.find_element(By.CLASS_NAME, 'RaceList_Box').find_elements(By.TAG_NAME, 'a')

ただし、hrefの中のresults.html?race_id=の部分は、レース開催前はshutuba.html?race_id=となることが分かっているので、レース開催前にスクレイピングするときは、引数をfrom_shutuba=Trueにして使用し、条件分岐させてrace_idを取得します。

if from_shutuba:
    race_id = re.findall('(?<=shutuba.html\?race_id=)\d+', a.get_attribute('href'))
else:
    race_id = re.findall('(?<=result.html\?race_id=)\d+', a.get_attribute('href'))

レース開催前にレースidをスクレイピングする場合については、次の章で解説します。

また、driver.get(url)の後、ChromeDriverが必要な要素を抽出し終わる前にdriver.find_element(By.CLASS_NAME, 'RaceList_Box')が実行されてしまうと、「そんな要素は存在しない」とエラーになってしまうので、

# 取得し終わらないうちに先に進んでしまうのを防ぐ
time.sleep(1)
a_list = driver.find_element(By.CLASS_NAME, 'RaceList_Box').find_elements(By.TAG_NAME, 'a')

と、find_elementの前に1秒間の待機を入れています。さらに、try~except文を用いて、それでも取得できなかった時には10秒間待機するようにしています。

また、そもそも取得に時間がかからないようにするため、ヘッドレスモード

options.add_argument('--headless') #ヘッドレスモード(ブラウザが立ち上がらない)

にしたり、

# 画面サイズをなるべく小さくし、余計な画像などを読み込まないようにする
driver.set_window_size(8, 8)

によって、余計な画像などを読み込まないようにしています。

それでも取得できない場合は、

except Exception as e:
    print(e)
    break

によってforループを抜け、取得できているrace_idのみ返すようにしています。

:::message try, exceptとは?
try~except文は、「例外処理」と呼ばれるもので、**「tryの中に書かれている処理でエラーが発生した場合、exceptの中に書かれている処理を行なってください」**という命令です。
:::

_scrape_html.py

ソースコード
import datetime
import re
import pandas as pd
import time
import os
from tqdm.notebook import tqdm
from urllib.request import urlopen

from modules.constants import UrlPaths, LocalPaths

def scrape_html_race(race_id_list: list, skip: bool = True):
    """
    netkeiba.comのraceページのhtmlをスクレイピングしてdata/html/raceに保存する関数。
    skip=Trueにすると、すでにhtmlが存在する場合はスキップされ、Falseにすると上書きされる。
    返り値:新しくスクレイピングしたhtmlのファイルパス
    """
    updated_html_path_list = []
    for race_id in tqdm(race_id_list):
        # 保存するファイル名
        filename = os.path.join(LocalPaths.HTML_RACE_DIR, race_id+'.bin')
        # skipがTrueで、かつbinファイルがすでに存在する場合は飛ばす
        if skip and os.path.isfile(filename):
            print('race_id {} skipped'.format(race_id))
        else:
            # race_idからurlを作る
            url = UrlPaths.RACE_URL + race_id
            # 相手サーバーに負担をかけないように1秒待機する
            time.sleep(1)
            # スクレイピング実行
            html = urlopen(url).read()
            # 保存するファイルパスを指定
            with open(filename, 'wb') as f:
                # 保存
                f.write(html)
            updated_html_path_list.append(filename)
    return updated_html_path_list

def scrape_html_horse(horse_id_list: list, skip: bool = True):
    """
    netkeiba.comのhorseページのhtmlをスクレイピングしてdata/html/horseに保存する関数。
    skip=Trueにすると、すでにhtmlが存在する場合はスキップされ、Falseにすると上書きされる。
    返り値:新しくスクレイピングしたhtmlのファイルパス
    """
    updated_html_path_list = []
    for horse_id in tqdm(horse_id_list):
        # 保存するファイル名
        filename = os.path.join(LocalPaths.HTML_HORSE_DIR, horse_id+'.bin')
        # skipがTrueで、かつbinファイルがすでに存在する場合は飛ばす
        if skip and os.path.isfile(filename):
            print('horse_id {} skipped'.format(horse_id))
        else:
            # horse_idからurlを作る
            url = UrlPaths.HORSE_URL + horse_id
            # 相手サーバーに負担をかけないように1秒待機する
            time.sleep(1)
            # スクレイピング実行
            html = urlopen(url).read()
            # 保存するファイルパスを指定
            with open(filename, 'wb') as f:
                # 保存
                f.write(html)
            updated_html_path_list.append(filename)
    return updated_html_path_list

def scrape_html_ped(horse_id_list: list, skip: bool = True):
    """
    netkeiba.comのhorse/pedページのhtmlをスクレイピングしてdata/html/pedに保存する関数。
    skip=Trueにすると、すでにhtmlが存在する場合はスキップされ、Falseにすると上書きされる。
    返り値:新しくスクレイピングしたhtmlのファイルパス
    """
    updated_html_path_list = []
    for horse_id in tqdm(horse_id_list):
        # 保存するファイル名
        filename = os.path.join(LocalPaths.HTML_PED_DIR, horse_id+'.bin')
        # skipがTrueで、かつbinファイルがすでに存在する場合は飛ばす
        if skip and os.path.isfile(filename):
            print('horse_id {} skipped'.format(horse_id))
        else:
            # horse_idからurlを作る
            url = UrlPaths.PED_URL + horse_id
            # 相手サーバーに負担をかけないように1秒待機する
            time.sleep(1)
            # スクレイピング実行
            html = urlopen(url).read()
            # 保存するファイルパスを指定
            with open(filename, 'wb') as f:
                # 保存
                f.write(html)
            updated_html_path_list.append(filename)
    return updated_html_path_list

def scrape_html_horse_with_master(horse_id_list: list, skip: bool = True):
    """
    netkeiba.comのhorseページのhtmlをスクレイピングしてdata/html/horseに保存する関数。
    skip=Trueにすると、すでにhtmlが存在する場合はスキップされ、Falseにすると上書きされる。
    返り値:新しくスクレイピングしたhtmlのファイルパス
    また、horse_idごとに、最後にスクレイピングした日付を記録し、data/master/horse_results_updated_at.csvに保存する。
    """
    ### スクレイピング実行 ###
    print('scraping')
    updated_html_path_list = scrape_html_horse(horse_id_list, skip)
    # パスから正規表現でhorse_id_listを取得
    horse_id_list = [
        re.findall('horse\W(\d+).bin', html_path)[0] for html_path in updated_html_path_list
        ]
    # DataFrameにしておく
    horse_id_df = pd.DataFrame({'horse_id': horse_id_list})
    
    ### 取得日マスタの更新 ###
    print('updating master')
    now = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S') # 現在日時を取得
    # マスタのパス
    filename_master = os.path.join(LocalPaths.MASTER_DIR, 'horse_results_updated_at.csv')
    # ファイルが存在しない場合は、作成する
    if not os.path.isfile(filename_master):
        pd.DataFrame(columns=['horse_id', 'updated_at']).to_csv(filename_master, index=None)
    # マスタを読み込み
    master = pd.read_csv(filename_master, dtype=object)
    # horse_id列に新しい馬を追加
    new_master = master.merge(horse_id_df, on='horse_id', how='outer')
    # マスタ更新
    new_master.loc[new_master['horse_id'].isin(horse_id_list), 'updated_at'] = now
    # 列が入れ替わってしまう場合があるので、修正しつつ保存
    new_master[['horse_id', 'updated_at']].to_csv(filename_master, index=None)
    return updated_html_path_list

netkeiba.comから、次の3種類のページのスクレイピングをします。

  1. raceページ:https://db.netkeiba.com/race/(race_id)
    raceページ
  2. horseページ:https://db.netkeiba.com/horse/(horse_id)
    horseページ
  3. pedページ:https://db.netkeiba.com/horse/ped/(horse_id)
    pedページ

スクレイピングしたhtmlを、それぞれ

  1. /data/html/race/(race_id).bin
  2. /data/html/horse/(horse_id).bin
  3. /data/html/ped/(horse_id).bin

へ保存します。次の章で詳しく解説しますが、

  1. raceページのhtmlから 「レース結果テーブル」「レース情報テーブル」「払戻テーブル」
  2. horseページのhtmlから 「馬の過去成績テーブル」
  3. pedページのhtmlから 「血統テーブル」

がそれぞれ作られることになります。(上の全体像参照)

ポイントはテーブルデータに変換する前に、一度htmlを保存しておくことです。やっていくと分かりますが、精度を改善するために、「rawテーブルに新しい列を追加したい」という場面が頻繁に出てきます。

その際、htmlを保存しておらず、テーブルデータのみ保存していると、もう一度全てのデータをスクレイピングする必要があります。

よって、「分量が多いページで容量を食ってしまう」「htmlの加工の仕方が決まっていて変更の余地がない」というような場合を除けば、基本的にスクレイピングの際はhtmlを保存するのが良いでしょう。

scrape_html_race()

ソースコード再掲
def scrape_html_race(race_id_list: list, skip: bool = True):
    """
    netkeiba.comのraceページのhtmlをスクレイピングしてdata/html/raceに保存する関数。
    skip=Trueにすると、すでにhtmlが存在する場合はスキップされ、Falseにすると上書きされる。
    返り値:新しくスクレイピングしたhtmlのファイルパス
    """
    updated_html_path_list = []
    for race_id in tqdm(race_id_list):
        # 保存するファイル名
        filename = os.path.join(LocalPaths.HTML_RACE_DIR, race_id+'.bin')
        # skipがTrueで、かつbinファイルがすでに存在する場合は飛ばす
        if skip and os.path.isfile(filename):
            print('race_id {} skipped'.format(race_id))
        else:
            # race_idからurlを作る
            url = UrlPaths.RACE_URL + race_id
            # 相手サーバーに負担をかけないように1秒待機する
            time.sleep(1)
            # スクレイピング実行
            html = urlopen(url).read()
            # 保存するファイルパスを指定
            with open(filename, 'wb') as f:
                # 保存
                f.write(html)
            updated_html_path_list.append(filename)
    return updated_html_path_list

raceページのhtmlをスクレイピングし、/data/html/race/(race_id).binへ保存します。

引数で渡されたrace_id_listからfor文により一つずつrace_idが取り出され、

url = UrlPaths.RACE_URL + race_id

によって、URLが作られた後、

html = urlopen(url).read()

でスクレイピング実行されます。その後、

with open(filename, 'wb') as f:
    # 保存
    f.write(html)

により、保存されます。skip=Trueで、かつすでにファイルが存在する場合は、

if skip and os.path.isfile(filename):
    print('race_id {} skipped'.format(race_id))

によってコメントとともにスクレイピングはスキップされます。

実際に、playground.ipynbで、以下のコードを実行してみましょう。

playground.ipynb
from modules import preparing

preparing.scrape_html_race(['202101010101', '202101010102', '202101010103'], skip=False)

出力結果(※/repository/の部分は環境により異なります)
出力結果

実際にファイルが保存されていることが分かります。

次に、skip=Trueにして、以下のコードを実行してみます。

playground.ipynb
preparing.scrape_html_race(['202101010101', '202101010102', '202101010103', '202101010104'], skip=True)

出力結果
出力結果2
すでにファイルが存在する'202101010101', '202101010102', '202101010103'についてはスクレイピングがスキップされ、新たにスクレイピングされた'repositories/keibaAI-v2/data/html/race/202101010104.bin'のみ、ファイルパスが返っていることが分かります。

scrape_html_horse()

scrape_race_html()と同様に、horseページのhtmlをスクレイピングし、/data/html/horse/(horse_id).binへ保存する関数です。
引数に入れるhorse_id_listは、レース結果テーブルのhorse_id列から取得することになります(次章で解説)。
つまり、例えば2021年の全レースを学習に使うなら、「2021年に出走した馬」のhorseページを全てスクレイピングすることになります。

scrape_html_horse_with_master()

ソースコード再掲
def scrape_html_horse_with_master(horse_id_list: list, skip: bool = True):
    """
    netkeiba.comのhorseページのhtmlをスクレイピングしてdata/html/horseに保存する関数。
    skip=Trueにすると、すでにhtmlが存在する場合はスキップされ、Falseにすると上書きされる。
    返り値:新しくスクレイピングしたhtmlのファイルパス
    また、horse_idごとに、最後にスクレイピングした日付を記録し、data/master/horse_results_updated_at.csvに保存する。
    """
    ### スクレイピング実行 ###
    print('scraping')
    updated_html_path_list = scrape_html_horse(horse_id_list, skip)
    # パスから正規表現でhorse_id_listを取得
    horse_id_list = [
        re.findall('horse\W(\d+).bin', html_path)[0] for html_path in updated_html_path_list
        ]
    # DataFrameにしておく
    horse_id_df = pd.DataFrame({'horse_id': horse_id_list})
    
    ### 取得日マスタの更新 ###
    print('updating master')
    # 現在日時を取得
    now = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    # マスタのパス
    filename_master = os.path.join(LocalPaths.MASTER_DIR, 'horse_results_updated_at.csv')
    # ファイルが存在しない場合は、作成する
    if not os.path.isfile(filename_master):
        pd.DataFrame(columns=['horse_id', 'updated_at']).to_csv(filename_master, index=None)
    # マスタを読み込み
    master = pd.read_csv(filename_master, dtype=object)
    # horse_id列に新しい馬を追加
    new_master = master.merge(horse_id_df, on='horse_id', how='outer')
    # マスタ更新
    new_master.loc[new_master['horse_id'].isin(horse_id_list), 'updated_at'] = now
    # 列が入れ替わってしまう場合があるので、修正しつつ保存
    new_master[['horse_id', 'updated_at']].to_csv(filename_master, index=None)
    return updated_html_path_list

scrape_html_horse()に、スクレイピング日時を記録する機能を付けた関数で、使い方はscrape_html_horse()と同じです。

他のページと違い、horseページについては、レースが行われるたびにそのページが更新され、テーブルに行が追加されていくページになっています。そのため、例えば同じ/data/html/horse/2017105318.binというファイルであっても、2020年1月1日にスクレイピングを実行した場合と2021年1月1日にスクレイピングを実行した場合で、入っているデータが異なってきます。

そこで、horse_idごとに、「いつスクレイピングをしたのか」という情報を記録するマスタをdata/master/horse_results_updated_at.csvというテーブルに、次のような形で持っておきます。
horse_results_updated_at

■関数の中身の解説
まずはscrape_html_horse()が実行され、新たにスクレイピングされたもののみファイルパスが返ってきます。

updated_html_path_list = scrape_html_horse(horse_id_list, skip)

マスタのhorse_id列を作るため、ファイルパスからhorse_idの文字列を正規表現により抜き出します。

# パスから正規表現でhorse_id_listを取得
horse_id_list = [
    re.findall('horse\W(\d+).bin', html_path)[0] for html_path in updated_html_path_list
    ]

re.findall('horse\W(\d+).bin', html_path)により、html_pathの中から、horse\W(\d+).binというパターンを探し、()の中の\d+:「数字の1回以上の繰り返し」を抽出します。
horse/(\d+).binとできないのは、環境によってファイルセパレータに使われる文字が異なるからです。(Windows環境ではファイルパスがhorse¥(horse_id).bin)
そのため、どちらの記号が来ても良いように、「数字や文字列以外のアルファベット」を表す\Wを使っています。
詳細:https://github.com/keibaAI-community/readers-repo/issues/117

scrape_html_ped()

pedページのhtmlをスクレイピングし、/data/html/ped/(horse_id).binへ保存します。scrape_html_horse()と同様のコードなため、省略。