Chapter 02無料公開

スクレイピング

↑目次をクリックすると、それぞれのソースコードに飛ぶことができます。

この章では、競馬予想AIを作るのに必要なデータをnetkeiba.comから取得します。次の章以降で、それぞれのデータについてクラスを定義してその中でデータ加工などを行うので、スクレイピングする関数も、それぞれのクラスのscrape()メソッドとして定義します。

スクレイピングするデータは次の4種類です。

データ メソッド 保存ファイル名
レース結果データ Results.scrape() results.pickle
馬の過去成績データ HorseResults.scrape() horse_results.pickle
血統データ Peds.scrape() peds.pickle
払い戻し表データ Return.scrape() return_tables.pickle

またこの4つの関数の他に、データを更新するためのupdate_data関数もここで定義しておきます。

Results.scrape()

メインとなる訓練データである、レース結果データをスクレイピングするメソッドです。先にコードを載せ、その下で解説をしていきます。関数定義の直後にあるコメントは、Parametersで引数を、Returnsで返り値をそれぞれ説明しています。

class Results:
    @staticmethod
    def scrape(race_id_list):
        """
        レース結果データをスクレイピングする関数

        Parameters:
        ----------
        race_id_list : list
            レースIDのリスト

        Returns:
        ----------
        race_results_df : pandas.DataFrame
            全レース結果データをまとめてDataFrame型にしたもの
        """

        #race_idをkeyにしてDataFrame型を格納
        race_results = {}
        for race_id in tqdm(race_id_list):
	time.sleep(1)
            try:
                url = "https://db.netkeiba.com/race/" + race_id
                #メインとなるテーブルデータを取得
                df = pd.read_html(url)[0]

                html = requests.get(url)
                html.encoding = "EUC-JP"
                soup = BeautifulSoup(html.text, "html.parser")

                #天候、レースの種類、コースの長さ、馬場の状態、日付をスクレイピング
                texts = (
                    soup.find("div", attrs={"class": "data_intro"}).find_all("p")[0].text
                    + soup.find("div", attrs={"class": "data_intro"}).find_all("p")[1].text
                )
                info = re.findall(r'\w+', texts)
                for text in info:
                    if text in ["芝", "ダート"]:
                        df["race_type"] = [text] * len(df)
                    if "障" in text:
                        df["race_type"] = ["障害"] * len(df)
                    if "m" in text:
                        df["course_len"] = [int(re.findall(r"\d+", text)[0])] * len(df)
                    if text in ["良", "稍重", "重", "不良"]:
                        df["ground_state"] = [text] * len(df)
                    if text in ["曇", "晴", "雨", "小雨", "小雪", "雪"]:
                        df["weather"] = [text] * len(df)
                    if "年" in text:
                        df["date"] = [text] * len(df)

                #馬ID、騎手IDをスクレイピング
                horse_id_list = []
                horse_a_list = soup.find("table", attrs={"summary": "レース結果"}).find_all(
                    "a", attrs={"href": re.compile("^/horse")}
                )
                for a in horse_a_list:
                    horse_id = re.findall(r"\d+", a["href"])
                    horse_id_list.append(horse_id[0])
                jockey_id_list = []
                jockey_a_list = soup.find("table", attrs={"summary": "レース結果"}).find_all(
                    "a", attrs={"href": re.compile("^/jockey")}
                )
                for a in jockey_a_list:
                    jockey_id = re.findall(r"\d+", a["href"])
                    jockey_id_list.append(jockey_id[0])
                df["horse_id"] = horse_id_list
                df["jockey_id"] = jockey_id_list

                #インデックスをrace_idにする
                df.index = [race_id] * len(df)

                race_results[race_id] = df
            #存在しないrace_idを飛ばす
            except IndexError:
                continue
            #wifiの接続が切れた時などでも途中までのデータを返せるようにする
            except Exception as e:
                print(e)
                break
            #Jupyterで停止ボタンを押した時の対処
            except:
                break

        #pd.DataFrame型にして一つのデータにまとめる
        race_results_df = pd.concat([race_results[key] for key in race_results])

        return race_results_df

まず、メソッドの前に@staticmethodがついています。これはクラス内に書かれたメソッドでも

Results.scrape(race_id_list)

のように「普通の関数」として実行できる、「静的メソッド」というものです。よくわからない方はとりあえず、「ただの関数」だと思っておいて問題ありません。

レース結果データは、netkeiba.comの「データベース」のページにあり、下の画像は2019年7月27日のレース結果データの例です。
あ
GoogleChromeで右クリック→検証からhtmlタグを見たときに、tableタグがついているデータはpandasのread_htmlを使うと、1行でスクレイピングできるので便利です。

pd.read_html("https://db.netkeiba.com/race/201901010101")[0]


netkeiba.comではレースごとにレースIDが与えられていて、

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

という構造になっているので、レースIDのリストを入れたらそれぞれのレース結果をスクレイピングしてPandasのDataFrame型で返してくれる関数を作ります。

実際にレースIDのリストを作って、2019年の全レース結果データをスクレイピングしてみましょう。

race_id_list = []
for place in range(1, 11, 1):
    for kai in range(1, 13, 1):
        for day in range(1, 13, 1):
            for r in range(1, 13, 1):
                race_id = "2019" + str(place).zfill(2) + str(kai).zfill(2) + str(day).zfill(2) + str(r).zfill(2)
                race_id_list.append(race_id)
		
results = Results.scrape(race_id_list)

結構時間がかかると思いますが、これでresultsという変数にpandasのDataFrame型でデータが入っているはずです。

もう一度スクレイピングするのは面倒なのでデータを保存しておきましょう。DataFrameはto_pickleを使えば1行のコードでpickleファイルに保存できるので便利です。

results.to_pickle('results.pickle')

逆に、読み込むときはpandasのread_pickleです。

results = pd.read_pickle('results.pickle')

これで、メインとなるレース結果データのスクレイピングが完了しました。

HorseResults.scrape()

次に、馬の過去成績データをスクレイピングします。

#馬の過去成績データを処理するクラス
class HorseResults:
    @staticmethod
    def scrape(horse_id_list):
        """
        馬の過去成績データをスクレイピングする関数

        Parameters:
        ----------
        horse_id_list : list
            馬IDのリスト

        Returns:
        ----------
        horse_results_df : pandas.DataFrame
            全馬の過去成績データをまとめてDataFrame型にしたもの
        """

        #horse_idをkeyにしてDataFrame型を格納
        horse_results = {}
        for horse_id in tqdm(horse_id_list):
            try:
                url = 'https://db.netkeiba.com/horse/' + horse_id
                df = pd.read_html(url)[3]
                #受賞歴がある馬の場合、3番目に受賞歴テーブルが来るため、4番目のデータを取得する
                if df.columns[0]=='受賞歴':
                    df = pd.read_html(url)[4]
                df.index = [horse_id] * len(df)
                horse_results[horse_id] = df
                time.sleep(1)
            except IndexError:
                continue
            except Exception as e:
                print(e)
                break
            except:
                break

        #pd.DataFrame型にして一つのデータにまとめる        
        horse_results_df = pd.concat([horse_results[key] for key in horse_results])

        return horse_results_df

下の画像はゴルゴンダのページの例です。

馬にも固有の馬IDが与えられているので(上の例だと2017105318)、そのリストを入れたらDataFrameで返してくれる関数を作ります。基本的なコードの構造は上と同じです。

この関数を用いて、さっき取得したresultsにある全部の馬の過去成績データを取得します。

horse_id_list = results['horse_id'].unique()
horse_results = HorseResults.scrape(horse_id_list)
horse_results #jupyterで出力


同様に、pickleファイルに保存します。

horse_results.to_pickle('horse_results.pickle')

Peds.scrape()

馬ごとに5世代分の血統データをスクレイピングする関数を作ります。

#血統データを処理するクラス
class Peds:
    @staticmethod
    def scrape(horse_id_list):
        """
        血統データをスクレイピングする関数

        Parameters:
        ----------
        horse_id_list : list
            馬IDのリスト

        Returns:
        ----------
        peds_df : pandas.DataFrame
            全血統データをまとめてDataFrame型にしたもの
        """

        peds_dict = {}
        for horse_id in tqdm(horse_id_list):
            try:
                url = "https://db.netkeiba.com/horse/ped/" + horse_id
                df = pd.read_html(url)[0]

                #重複を削除して1列のSeries型データに直す
                generations = {}
                for i in reversed(range(5)):
                    generations[i] = df[i]
                    df.drop([i], axis=1, inplace=True)
                    df = df.drop_duplicates()
                ped = pd.concat([generations[i] for i in range(5)]).rename(horse_id)

                peds_dict[horse_id] = ped.reset_index(drop=True)
                time.sleep(1)
            except IndexError:
                continue
            except Exception as e:
                print(e)
                break
            except:
                break

        #列名をpeds_0, ..., peds_61にする
        peds_df = pd.concat([peds_dict[key] for key in peds_dict], axis=1).T.add_prefix('peds_')

        return peds_df

画像はゴルゴンダの例です

同様にpandasのread_htmlを使うのですが、普通にやると下のようにデータが重複しています。

しかし、5世代分で62個のデータがあれば良いので、重複を消して1列のSeries型にする処理を行なっています。

スクレイピングできたら、同様にpeds.pickleに保存してください。

Return.scrape()

回収率の計算をするために、レース結果データと同じページにある払い戻し表をスクレイピングします。

#払い戻し表データを処理するクラス
class Return:
    @staticmethod
    def scrape(race_id_list):
        """
        払い戻し表データをスクレイピングする関数

        Parameters:
        ----------
        race_id_list : list
            レースIDのリスト

        Returns:
        ----------
        return_tables_df : pandas.DataFrame
            全払い戻し表データをまとめてDataFrame型にしたもの
        """

        return_tables = {}
        for race_id in tqdm(race_id_list):
            try:
                url = "https://db.netkeiba.com/race/" + race_id

                #普通にスクレイピングすると複勝やワイドなどが区切られないで繋がってしまう。
                #そのため、改行コードを文字列brに変換して後でsplitする
                f = urlopen(url)
                html = f.read()
                html = html.replace(b'<br />', b'br')
                dfs = pd.read_html(html)

                #dfsの1番目に単勝〜馬連、2番目にワイド〜三連単がある
                df = pd.concat([dfs[1], dfs[2]])

                df.index = [race_id] * len(df)
                return_tables[race_id] = df
                time.sleep(1)
            except IndexError:
                continue
            except Exception as e:
                print(e)
                break
            except:
                break

        #pd.DataFrame型にして一つのデータにまとめる
        return_tables_df = pd.concat([return_tables[key] for key in return_tables])
        return return_tables_df

pd.read_html(url)[1]で左側(単勝〜馬連)表が、pd.read_html(url)[2]で右側(ワイド〜三連単)の表がスクレイピングできます。

しかし、払い戻し表は少し厄介で、普通にスクレイピングすると改行タグが消えてしまい、複勝やワイドのデータが繋がってしまいます。

url = "https://db.netkeiba.com/race/201902010101"
pd.read_html(url)[1]


なので、

f = urlopen(url)
html = f.read()

でまずはhtmlを取得し、

html = html.replace(b'<br />', b'br')

で改行タグを文字列型brに置き換えてからスクレイピングすることで、あとでこの文字を頼りにsplit関数などで区切ることができます。

スクレイピングできたら、return_tables.pickleに保存しましょう。

update_data関数

馬の過去成績データなどを新しくスクレイピングして古いデータに追加するときに必要になるものです。

def update_data(old, new):
    """
    Parameters:
    ----------
    old : pandas.DataFrame
        古いデータ
    new : pandas.DataFrame
        新しいデータ
    """

    filtered_old = old[~old.index.isin(new.index)]
    return pd.concat([filtered_old, new])

馬の過去成績データは馬が新しくレースをするたびに、その馬のページに出走データが追加されていきます。 例えば、horse_idが2017104475のデータ

を、2020年6月21日以前にスクレイピングしていた場合、次のようになるはずです。

2020年6月21日以降にこのページを新しくスクレイピングしてhorse_resultsを更新しようと思った時、普通にpd.concatで繋げてしまうと、古い3つの行は重複してしまいます。
そこで、古いデータと新しいデータ両方に存在するhorse_idについては、新しくスクレイピングしたものを使い、それ以外は古いものを使うようにこの関数で処理しています。


これで必要なデータは揃いました。次の章ではそれぞれのクラスに前処理をするコードを追加し、データを機械学習モデルに入れることができる形に加工します。