競馬でデータ分析をしていくためにまず必要となるのがデータの取得。
今回はPythonでオッズの取得をやってみた。
Playwright
データの取得はいろいろ考えられて、Web APIが使えればそれが一番早いんだけど、残念ながら使えなさそう （公式でJRA-VANというのはあって、Web APIもあるっぽいんだけど、macだと動くか怪しい・・・）。
そこでスクレイピングでWebサイトから情報をとってくることにした。
オッズはJRA公式サイトから取得できて、一応スクレイピングは禁止されてないようだった
（robots.txtでDisallowが空で、規約にも禁止の文言は見当たらない）。
もちろん、過度なアクセスは避けるべきだけど。
Pythonでスクレイピングする場合、SeleniumやPlaywrightを使うことが考えられて、今回はPlaywrightを使うことにした：
- Playwrightのメリット
- 使用するブラウザが公式で用意されていて簡単にダウンロードできる
- ツールを使ってブラウザ操作からコードを生成できる
- 画面遷移待ち、ダウンロード待ちなどをAPIでサポートしてる
- Playwrightのデメリット
- async/await汚染が酷い
メリットの3つ目はちょっと分かりにくいと思うけど、Seleniumの場合、画面遷移やダウンロードなど時間がかかる処理について、その完了を待つAPIは用意されてないので、自前でポーリングするコードを書く必要があったりする。 これが地味に手間。 その点、Playwrightはちゃんと操作が完了するまで待ってくれるAPIになっていて、コードが書きやすい。
そしてデメリットについては非同期APIをサポートしているということで、人によってはメリットに感じるかと思う。 まぁ、個人的にはasync/awaitは大嫌いなんだけど。
Playwrightのインストールとセットアップ
Playwrightのインストールは簡単で、pip installするだけ：
$ pip install playwright
なお、pytestとセットで使う場合はpytest-playwrightをインストールした方がいいっぽいけど、そうでなければ上のように単体でOK。
Playwrightをインストールしたら必要なブラウザをインストールしておく：
$ playwright install
ブラウザ操作によるコード生成
スクレイピングのコードを一から書いてもいいんだけど、Playwrightにはブラウザ操作からコードを生成するツールがついているので、これを使うとちょっと便利。 まぁ最終的には調整が必要だけど。
このツールを起動するには、コマンドラインで次のコマンドを叩く：
$ playwright codegen --target python-async <URL>
なお、オプションで--target python-asyncをつけてるのはasyncio用のコードを出力するため。
これがないとJupyterで動かしたときにエラーになる可能性がある。
詳しくは前述のasyncioに関する記事を参照。
このツールが起動すると、操作するためのブラウザと、そこでの操作がコードに書き起こされるウィンドウが開かれる。 ブラウザでリンクをクリックしたりするとそれに応じてコードが生成されていくのが分かると思う。
今回はJRA公式からデータを取りたかったので、URLとしてhttps://jra.jp/を指定してツールを起動。
そのあと、ページ上部の「オッズ」をクリックし、オッズを見たい開催（たとえば「1回中山5日」）のボタンをクリック。
そしてレースとして「1R」を選び、「単勝・複勝」のオッズのページ、「馬連」のオッズのページを見る。
さらにページ上部にある「2R」を選び、同様に単勝・複勝のオッズ、馬連のオッズを見る、という操作をやってみた。
それで生成されたコード（に少しコメントを追加したもの）が以下：
import asyncio from playwright.async_api import Playwright, async_playwright, expect async def run(playwright: Playwright) -> None: browser = await playwright.chromium.launch(headless=False) context = await browser.new_context() page = await context.new_page() await page.goto("https://jra.jp/") await page.get_by_role("link", name="オッズ").click() await page.get_by_role("link", name="1回中山5日").click() await page.get_by_role("row", name="1レース 10時10分 3歳未勝利牝 単勝複勝 枠連 馬連 ワイド 馬単 3連複 3連単").get_by_role("link", name="1レース").click() # もう少しいい指定をしたい await page.get_by_role("link", name="単勝・複勝").click() # ここで単勝のオッズを読む必要あり await page.get_by_role("link", name="馬連").click() # ここで馬連のオッズを読む必要あり await page.get_by_role("link", name="2レース").first.click() await page.get_by_role("link", name="単勝・複勝").click() # ここで単勝のオッズを読む必要あり await page.get_by_role("link", name="馬連").click() # ここで馬連のオッズを読む必要あり # 以下、最終レースまで同様 # 1回中京5日、1回小倉1日も同様 # --------------------- await context.close() await browser.close() async def main() -> None: async with async_playwright() as playwright: await run(playwright) asyncio.run(main())
ここからさらに手を入れる必要はあるものの、大雑把にコードが作られるので最初の一歩は踏み出しやすいと思う。 簡単なスクレイピングならこれだけで十分かも。
コンテキストマネージャ
Playwrightでのスクレイピングの流れは次のようになっている：
- Playwrightのインスタンスを用意
- Playwrightのインスタンスでブラウザを開く
- ブラウザで新しいコンテキストを用意する
- コンテキストでページを開く
- ページで目的のサイトに行ったり要素にアクセスして情報を集める
- コンテキストを閉じる
- ブラウザを閉じる
ここで、ブラウザを開くのと閉じるの、コンテキストを用意するのと閉じるのはペアになっている必要がある。 で、上記のコードだと閉じるのは単独でやってるけど、実際には例外が発生した場合にもちゃんと閉じる必要があるので、try-finallyなどを使った方がいい。
そういう場合にいいのがwith文を使う方法で、コンテキストマネージャとして特定のメソッドを実装するとwith文を抜けるときにリソースの解放を忘れずに行うことができる。
上記ではしっかりとコンテキストマネージャを実装しているけど、contextlibを使うとRubyのブロック呼び出しと同じ感覚でwith文を使えるようになるので、そっちの方が便利かもしれない。
contextlibを使って用意したコンテキストマネージャが以下：
import contextlib @contextlib.asynccontextmanager async def open_browser(playwright, headless=True): browser = await playwright.chromium.launch(headless=headless) try: yield browser finally: await browser.close() @contextlib.asynccontextmanager async def open_context(browser): context = await browser.new_context() try: yield context finally: await context.close()
こうするとwith文に続くブロックが関数内のyieldの部分で実行される感じになる。
なお、関数内で非同期関数を呼んでいるのでコンテキストマネージャも非同期にする必要があり、with文も async with文にする必要がある。
汚染が酷い。。。
そしてwith文の入れ子が深くなるので、それを浅くするコンテキストマネージャも用意した：
from playwright.async_api import async_playwright @contextlib.asynccontextmanager async def playwright_context(headless=True): async with async_playwright() as playwright: async with open_browser(playwright, headless) as browser: async with open_context(browser) as context: yield browser, context
async/await汚染マジでヤバいよなぁ・・・
オッズ取得
さて、外側のコンテキストマネージャは用意できたので、実際にオッズを取得する部分を作っていく。
とりあえず必要なライブラリのインポートとか。
import numpy as np import pandas as pd import asyncio from pathlib import Path import traceback # asyncioの入れ子を許す import nest_asyncio nest_asyncio.apply() # 取得した結果をodds_csv/以下に保存する save_dir = Path("odds_csv")
nest_asyncioはasynioの入れ子を許すためのライブラリ。
これを使わないとJupyterとスクリプトの両方で動くようにできない。
まずはJRAのサイトに行って指定された開催の1Rのページに行く関数：
# JRA → オッズ → 開催 → 1Rへ移動 async def select_1st_race(page, place): # placeは"1回中山5日"など await page.goto("https://jra.jp/") async with page.expect_navigation(): await page.get_by_role("link", name="オッズ").click() async with page.expect_navigation(): await page.get_by_role("link", name=place).click() async with page.expect_navigation(): await page.locator("#race_list").locator("tbody tr a").first.click()
ここで使ってるasync with page.expect_navigation()はブロックの終わりでページの遷移が終わるまで待ってくれるもので、ページが変わるときには呼んでおいた方が安全。
あとlocator()は要素を探してくれるメソッドで、CSSセレクタを指定するとそれにマッチする要素を見つけてくれる。
firstとすれば見つかったものの先頭、all()とすればすべて（のイテレータ）を返してくれる。
また、見つかったものに対してlocator()を呼ぶと、その子孫のみを対象にして探してくれる。
上記の例だとpage.locator("#race_list")でページ全体からIDが#race_listの要素を探し、その子孫からtbody tr aでヒットするものを探し、その一番最初の要素をクリックしていることになる。
次は1Rのページに行った状態で、レース数を取得するための関数：
# 1R以降でレース数を取得 async def get_race_count(page): return await page.locator("ul.race-num").first.locator("li").count()
count()は見つかった要素の数を返すメソッド。
そして1Rのページに入った状態で、他のレースへ移動するための関数：
# 1R以降で他のレースへ移動 async def select_race(page, race): async with page.expect_navigation(): await page.get_by_role("link", name=f"{race}レース").first.click()
あとはテーブルからオッズを読みたいんだけど、出走取消になった場合には小数ではなく「取消」という文字が入っていることがあったので、そういう不正な数字の場合にはnp.nanを返す関数を用意しておく：
# オッズでfloat以外（「取消」など）のことがあったので、その対策 def float_or_nan(str_value): try: return float(str_value) except: return np.nan
単勝のオッズを取得するコードは以下：
# 単勝オッズ取得 async def get_odds_tansho(page, place, race): number_list = [] odds_list = [] async with page.expect_navigation(): await page.get_by_role("link", name="単勝・複勝").click() odds_table = page.locator("#odds_list table") for tr in await odds_table.locator("tbody tr").all(): number = int(await tr.locator("td.num").inner_text()) odds = float_or_nan(await tr.locator("td.odds_tan").inner_text()) number_list.append(number) odds_list.append(odds) table = pd.DataFrame({ "number": number_list, "odds": odds_list, }) table["place"] = place table["race"] = race # 列の並び替え columns = ["place", "race", "number", "odds"] table = table[columns] return table
テーブルの各行について処理することで馬番とオッズを取っている。
この中のinner_text()は要素のテキストを返すメソッドで、それをint()で整数にしたり、上記のfloat_or_nan()で小数にして、DataFrameにして返している。
同様に、馬連のオッズを取得するコードは以下：
# 馬連オッズ取得 async def get_odds_umaren(page, place, race): number1_list = [] number2_list = [] odds_list = [] async with page.expect_navigation(): await page.get_by_role("link", name="馬連").click() odds_tables = page.locator("#odds_list table.umaren") for odds_table in await odds_tables.all(): number1 = int(await odds_table.locator("caption").inner_text()) for tr in await odds_table.locator("tbody tr").all(): number2 = int(await tr.locator("th").inner_text()) odds = float_or_nan(await tr.locator("td").inner_text()) number1_list.append(number1) number2_list.append(number2) odds_list.append(odds) table = pd.DataFrame({ "number1": number1_list, "number2": number2_list, "odds": odds_list, }) table["place"] = place table["race"] = race # 列の並び替え columns = ["place", "race", "number1", "number2", "odds"] table = table[columns] return table
あとはこれらを呼び出していくだけ：
# 開催のオッズを取得する async def get_odds(place): odds_tansho_list = [] odds_umaren_list = [] async with playwright_context() as (browser, context): page = await context.new_page() await select_1st_race(page, place) race_count = await get_race_count(page) for race in range(1, race_count+1): await select_race(page, race) odds_tansho = await get_odds_tansho(page, place, race) odds_umaren = await get_odds_umaren(page, place, race) odds_tansho_list.append(odds_tansho) odds_umaren_list.append(odds_umaren) odds_tansho = pd.concat(odds_tansho_list, ignore_index=True) odds_umaren = pd.concat(odds_umaren_list, ignore_index=True) return odds_tansho, odds_umaren
これを実行すると各開催の単勝、馬連のオッズのDataFrameが得られるんだけど、それをCSVに保存する同期関数を用意してやる：
# オッズを取得して保存 def get_odds_and_save_csv(place, save_dir, csv_base, version): if not isinstance(save_dir, Path): save_dir = Path(save_dir) if not save_dir.exists(): save_dir.mkdir(parents=True) tansho_path = save_dir / f"{csv_base}_tansho_{version}.csv" umaren_path = save_dir / f"{csv_base}_umaren_{version}.csv" if tansho_path.exists() or umaren_path.exists(): print("Already exists.") return try: odds_tansho, odds_umaren = asyncio.run(get_odds(place)) odds_tansho.to_csv(tansho_path, index=False) odds_umaren.to_csv(umaren_path, index=False) except: traceback.print_exc() print("Failed to get odds.")
上記のasynio.run()としてるのが実際に非同期関数を走らせてる部分。
ちなみに、最初に生成したコードからはだいぶ変わってると思うけど、まぁちゃんとやろうとするとそれなりに大変よね（＾＾；
呼び出しは次のような感じ：
get_odds_and_save_csv("1回中山5日", save_dir, "20230114_nakayama", "0930")
これを実行すると、呼び出した時点での「1回中山5日」の単勝オッズ、馬連オッズがodds_csv/の下に20230114_nakayama_tansho_0930.csvおよび20230114_nakayama_umaren_0930.csvとして保存される。
なお、0930としてるのは9時半時点でのオッズを取ろうとしてるから。
こんな感じでPlaywrightを使ってオッズ取得できた。
コンテキストマネージャとかasyncioの話も出てきてるのでPythonをそれなりに使ってないと難しい部分もあるかもしれないけど、比較的簡単にスクレイピングできている（Seleniumだとポーリング用のクラスを用意する必要があったりでもうちょっと大変・・・）。
今日はここまで！
