Pythonで競馬のレース結果を取得してみた。

前回はPythonでPlaywrightを使ってJRA公式サイトからオッズの取得をしてみた。

今回は同様にJRA公式からレース結果を取得してみる。

準備

といってもコードはほとんど前回のオッズ取得と同じ。

なのでまずはライブラリのインポートなど必要な準備から。

import numpy as np
import pandas as pd

import asyncio
import contextlib
from pathlib import Path
import traceback

# asyncioの入れ子を許す
import nest_asyncio
nest_asyncio.apply()

from playwright.async_api import async_playwright

save_dir = Path("result_csv")

# コンテキストマネージャ
@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()

@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

コンテキストマネージャについては前回の記事を参照。

レース結果取得

レース結果はJRA公式のページ上部にある「レース結果」から見ることができる。

オッズのときと同様に、開催を選んで、そこに含まれる全レースの結果を取得する。

# 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()

# 1R以降でレース数を取得
async def get_race_count(page):
    return await page.locator("ul.race-num").first.locator("li").count()

# 1R以降で他のレースへ移動
async def select_race(page, race):
    async with page.expect_navigation():
        await page.get_by_role("link", name=f"{race}レース").first.click()

今回の取得対象にしたいのは、着順と払戻。 とくに払戻は「単勝」「馬連」「3連複」を見ることにする。

# 着順でint以外（「取消」など）のことがあったので、その対策
def int_or_none(str_value):
    try:
        return int(str_value)
    except:
        return None

# 着順取得
async def get_rank(page, place, race):
    rank_list = []
    number_list = []

    result_table = page.locator("#race_result table").first
    for tr in await result_table.locator("tbody tr").all():
        rank = int_or_none(await tr.locator("td.place").inner_text())
        if rank is None:
            continue
        number = int(await tr.locator("td.num").inner_text())
        rank_list.append(rank)
        number_list.append(number)

    table = pd.DataFrame({
        "rank": rank_list,
        "number": number_list,
    })
    table["place"] = place
    table["race"] = race

    # 列の並び替え
    columns = ["place", "race", "rank", "number"]
    table = table[columns]

    return table

# 金額を取得する（桁区切りのカンマと単位がついてくるので除く）
def int_from_yen(yen_str):
    str_value = yen_str[:-1]  # 単位の「円」を取り除く
    str_value = str_value.replace(",", "")  # 桁区切りのカンマを取り除く
    return int(str_value)

# 払戻取得（単勝、馬連、3連複）
async def get_refund(page, place, race):
    tansho = int_from_yen(await page.locator("li.win .yen").inner_text())
    umaren = int_from_yen(await page.locator("li.umaren .yen").inner_text())
    trio = int_from_yen(await page.locator("li.trio .yen").inner_text())

    return pd.DataFrame({
        "place": [place],
        "race": [race],
        "tansho": [tansho],
        "umaren": [umaren],
        "trio": [trio],
    })

前回と同様に、ページ内で見たい要素を見つけて、その値をまとめ、DataFrameにする感じ。 なお、着順については出走取消や競争中止などになっていた番号はとらないことにした。

あとはこれらを呼び出してCSVに保存する関数を用意：

# 開催のレース結果を取得する
async def get_result(place):
    rank_list = []
    refund_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)
            rank = await get_rank(page, place, race)
            refund = await get_refund(page, place, race)
            rank_list.append(rank)
            refund_list.append(refund)
    rank = pd.concat(rank_list, ignore_index=True)
    refund = pd.concat(refund_list, ignore_index=True)
    return rank, refund

# レース結果を取得して保存
def get_result_and_save_csv(place, save_dir, csv_base):
    if not isinstance(save_dir, Path):
        save_dir = Path(save_dir)
    if not save_dir.exists():
        save_dir.mkdir(parents=True)
    rank_path = save_dir / f"{csv_base}_rank.csv"
    refund_path = save_dir / f"{csv_base}_refund.csv"
    if rank_path.exists() or refund_path.exists():
        print("Already exists.")
        return

    try:
        rank, refund = asyncio.run(get_result(place))
        rank.to_csv(rank_path, index=False)
        refund.to_csv(refund_path, index=False)
    except:
        traceback.print_exc()
        print("Failed to get result.")

これをたとえば次のように呼び出してやる：

get_result_and_save_csv("1回中山5日", save_dir, "20230114_nakayama")

これでresult_csv/の下に「1回中山5日」のレース結果（着順、払戻）が20230114_nakayama_rank.csvおよび20230114_nakayama_refund.csvとして保存される。

---

前回と今回でレースのオッズと結果を取得できるようにしたので、次はいろいろ分析していきたいと思う。

今日はここまで！