【競馬AI-6】過去5走を横持ちにして学習データを整える

競馬AIを作るときに欠かせないのが「過去のレース成績をどのように扱うか」という視点です。1走前だけを見るのでは足りず、直近の5走分をまとめて特徴量にしたい場面は多いでしょう。そこで今回は、全ての馬のレース結果に対して「過去5走を横持ちで持たせる処理」を紹介します。

処理概要
プログラム全文
解説
ポイント
コードの呼び出し方
まとめ

処理概要

今回の処理では、まず全ての馬についてレースを最新から順に並べることから始めます。そこから直近の1走前〜5走前までの情報を取り出し、同じ行に横持ちで追加していきます。

結果として「ある馬のこのレース」という1行に、その馬の直近5走分の情報が付与されます。これを作っておけば、学習モデルに過去成績も含めて入力することが簡単になります。

プログラム全文

以下が実際のコードです。
（db/init_work_tables.py に追記する）

def create_joined_race_table():
    # カラムとMySQL型を明示的に定義
    merge_columns_with_types = {
        "RACE_ID": "BIGINT(20)",
        "KAKUTEI_CHAKUJUN": "CHAR(2)",
        "TANSHO_ODDS_REAL": "DECIMAL(5,1)",
        "WAKUBAN": "CHAR(1)",
        "UMABAN": "CHAR(2)",
        "FUTAN_JURYO": "CHAR(3)",
        "BLINKER_SHIYO_KUBUN": "CHAR(2)",
        "KISHU_CODE": "CHAR(5)",
        "BATAIJU": "CHAR(3)",
        "ZOGEN_FUGO": "CHAR(1)",
        "ZOGEN_SA": "CHAR(3)",
        "SOHA_TIME": "VARCHAR(32)",
        "CHAKUSA_CODE1": "CHAR(3)",
        "CORNER1_JUNI": "CHAR(2)",
        "CORNER2_JUNI": "CHAR(2)",
        "CORNER3_JUNI": "CHAR(2)",
        "CORNER4_JUNI": "CHAR(2)",
        "TANSHO_NINKIJUN": "CHAR(2)",
        "KOHAN_4F": "CHAR(3)",
        "KOHAN_3F": "CHAR(3)",
        "TIME_SA": "CHAR(4)",
        "KYAKUSHITSU_HANTEI": "CHAR(1)"
    }

    merge_columns = list(merge_columns_with_types.keys())

    print("🚀 データ取得開始...")
    with get_connection() as conn:
        with conn.cursor() as cursor:
            cursor.execute("SELECT * FROM w_ranked_race_temp ORDER BY RACE_ID DESC")
            rows = cursor.fetchall()
            columns = [desc[0] for desc in cursor.description]
            df = pd.DataFrame(rows, columns=columns)
            print(f"✅ {len(df)} 件のデータを取得しました。")

            print("🧮 race_row カラム生成中...")
            df["race_row"] = df.groupby("KETTO_TOROKU_BANGO").cumcount() + 1

            print("📊 ソート実行中...")
            df.sort_values(by=["KETTO_TOROKU_BANGO", "RACE_ID"], ascending=[True, False], inplace=True)

            print("🔁 過去5走データの付与を開始...")
            for i in range(1, 6):
                print(f"　➡️ {i}/5 回目の shift 実行中...", end="\r")
                shifted = df.groupby("KETTO_TOROKU_BANGO", group_keys=False)[merge_columns].shift(-i)
                shifted.columns = [f"PREV{i}_{col}" for col in merge_columns]
                df = pd.concat([df, shifted], axis=1)
            print("\n✅ 過去5走の結合完了。")

        with conn.cursor() as cursor:
            print("🧹 w_joined_race を削除中...")
            cursor.execute("DROP TABLE IF EXISTS w_joined_race")
            cursor.execute("CREATE TABLE w_joined_race LIKE w_ranked_race_temp")

            print("🛠 カラム追加（ALTER TABLE）中...")
            alter_statements = []
            for i in range(1, 6):
                for col, mysql_type in merge_columns_with_types.items():
                    col_name = f"PREV{i}_{col}"
                    alter_statements.append(f"ADD COLUMN {col_name} {mysql_type}")

            if alter_statements:
                alter_sql = f"ALTER TABLE w_joined_race {', '.join(alter_statements)}"
                cursor.execute(alter_sql)
            conn.commit()
            print("✅ カラム追加完了")

            insert_columns = df.columns.tolist()
            insert_sql = f"""
                INSERT INTO w_joined_race ({','.join(insert_columns)})
                VALUES ({','.join(['%s'] * len(insert_columns))})
            """

            print("🧹 NaN → None 変換中...")
            df = df.where(pd.notnull(df), None)

            records = df.to_records(index=False)
            total = len(records)
            batch_size = 1000
            total_batches = (total + batch_size - 1) // batch_size

            print(f"📤 データ挿入開始：{total} 件をバッチサイズ {batch_size} で挿入します。")
            for i in range(0, total, batch_size):
                batch_index = i // batch_size + 1
                batch = [tuple(r) for r in records[i:i + batch_size]]
                cursor.executemany(insert_sql, batch)
                conn.commit()
                print(f"📦 バッチ {batch_index} / {total_batches} を挿入中...", end="\r")
            print(f"\n✅ 全 {total} 行を w_joined_race に挿入完了。")

解説

この処理の肝は 「過去の走行データをシフトして横持ち化する」 部分です。

まず、race_row というカラムで「この馬にとって何走前なのか」を順番に番号付けします。その上で、pandasの groupby + shift を使って、1走前から5走前までをそれぞれの行に追加します。

例えばある馬が「2025年春天」を走った場合、その行には「2025年大阪杯」「2025年京都記念」といった過去の履歴が PREV1_〜 の形で横持ちされます。

これがあるだけで「過去5走を特徴量に入れる」という処理が一気に楽になります。

ポイント

SQLだけでやらない
自己結合で過去5走を取ろうとするとSQLが非常に重くなります。pandasの shift を使う方がシンプルで安定します。
カラム定義を明示する
PREV1〜PREV5のカラムはALTER TABLEで追加。型を揃えておくことで後工程のミスを減らせます。
バッチ挿入で安定化
1,000件ずつ分割して挿入。大量データを一気に処理せず、安定したジョブ実行が可能になります。

コードの呼び出し方

この関数は前回のコードと同じくinit_db.py から実行します。
複数の作業テーブルをまとめて作成する流れの中で create_ranked_race_temp() が呼ばれます。以下のようなコードになります。

from db.init_work_tables import (
    create_ranked_race_base,
    create_ranked_race_temp,
    create_joined_race_table
)

if __name__ == "__main__":
    print("🛠️ ranked_race_base を作成中...")
    create_ranked_race_base()

    print("🛠️ ranked_race_temp を作成中...")
    create_ranked_race_temp()

    print("🛠️ joined_race を作成中...")
    create_joined_race_table()

    print("✅ すべて完了しました。")