📝 この記事は Zenn で先に公開した記事の再掲です。最新版とコメントは Zenn をご参照ください。
はじめに：時系列 ML の「未来参照」という古典的バグ
時系列データを使った機械学習でもっとも踏みやすい罠が データリーケージ（Data Leakage） です。とくに「過去の出来事を見て次の出来事を予測する」タイプのタスクでは、学習時に未来の情報を混入させてしまうことで、評価指標が現実離れして良くなる現象が頻発します。
筆者は競馬予想ツール UmaScore（フォワードテスト中、umascore.com）の開発で、Python + SQLite + 独自指数 + LightGBM を組み合わせたパイプラインを書いています。このパイプラインを書き始めた当初、バックテスト集計で「あれ、数字が良すぎる」と感じる瞬間が何度かありました。原因はほぼ毎回データリーケージでした。
この記事では、競馬データという題材を使いながら 時系列 ML 全般に通じる4つのリーク経路 を整理し、それぞれに対して race_date という基準点でフィルタを通す実装パターンを紹介します。コードは UmaScore の services/prediction_engine.py から実物を引用しています。
なお、本記事は技術検証の知見共有が目的で、馬券購入を勧めるものではありません。記事末尾の注意書きも合わせてご覧ください。
競馬データにおける時系列リークの4経路
UmaScore のスコアリングは複数の特徴量を合成して Master_Score を作る構造です。リークが発生しうる箇所を整理すると、以下の4経路に集約されました。
| # | リーク経路 | 何が混ざるか | 対策 |
|---|---|---|---|
| 1 | 走力指数（Base_Ability_Index）の集計 |
予測対象レース当日・以降の過去走 |
race_date < ? で過去走テーブルをフィルタ |
| 2 | 適性スコア（Aptitude_Score）のコース実績集計 |
予測対象レース当日・以降の同条件走 | 同上 |
| 3 | 血統経由の集計（種牡馬産駒の同条件成績） | 予測対象レース当日・以降に発生した産駒の走 |
pr.race_date < ? で JOIN 先にもフィルタ |
| 4 | 推論時の関数呼び出し |
race_date を渡し忘れて全期間集計 |
呼び出し側の規律（テストで担保） |
「全期間で集計してから date でスライス」のような後付け対策は、JOIN 先の参照や血統経由のクエリで簡単に取りこぼします。そのため、UmaScore では「集計クエリの WHERE 句に必ず race_date < ? を挿し込む」という単一ルールで4経路を統一しました。
1. calculate_base_ability_index の race_date フィルタ
走力指数（過去5走から算出する Base_Ability_Index、0〜40点）の実装です。ここがリークすると、「未来の走破タイムを根拠に強さを語る」ことになり、評価が一気に崩れます。
関数シグネチャに race_date を明示し、SQL 側で AND race_date < ? を挿し込む構造を取りました。
# services/prediction_engine.py: calculate_base_ability_index
def calculate_base_ability_index(
horse_id: int,
race_id: Optional[int] = None,
race_date: Optional[str] = None,
) -> tuple[float, list[str]]:
"""
Base_Ability_Index 算出（0〜40点）
過去5走の走破タイム・上がり3F・着差・着順・直接対決から合成。
race_date が渡された場合、それより前の走のみを参照する（リーク防止）。
"""
missing_flags = []
with get_db() as conn:
date_filter = ""
params = [horse_id]
if race_date:
date_filter = " AND race_date < ?"
params.append(race_date)
past_races = conn.execute(
f"""
SELECT race_date, venue, race_name, distance, track_type, track_condition,
weather, finish_position, finish_time, last_3f_time, margin
FROM past_races
WHERE horse_id = ?{date_filter}
ORDER BY race_date DESC
LIMIT 5
""",
params,
).fetchall()
...
ここでのポイントは3つです。
-
race_dateを Optional にしている: バックテストでは厳格に渡し、運用時の手動デバッグではNone（＝全期間）も許容します。ただし運用時の本線では必ず渡す（後述）。 -
f-string で WHERE 句を組み立てるが、値はプレースホルダ経由:
date_filterは文字列定数（" AND race_date < ?"）の追加のみで、ユーザー入力は混ざらないので SQL インジェクションは発生しません。日付値はparams.append(race_date)側で渡しています。 -
ORDER BY race_date DESC LIMIT 5: 過去5走に絞ることで「直近の状態」を反映。フィルタを忘れると 過去5走の中に未来走が含まれる 致命的な状態になります。
「テーブルに race_date カラムがあるから大丈夫」と思っていても、LIMIT N でソートした結果には未来走が紛れます。ソート＋LIMIT 系のクエリは特にリーク事故が起きやすいので、入口で race_date < ? を必ず通します。
2. calculate_aptitude_score のコース実績フィルタ
適性スコア（Aptitude_Score、0〜15点）は「同じ競馬場・近い距離での過去成績」を集計します。ここでも race_date 引数を必須化しています。
# services/prediction_engine.py: calculate_aptitude_score（コース実績部分）
def calculate_aptitude_score(
horse_id: int,
race_id: int,
race_date: Optional[str] = None,
) -> tuple[float, list[str]]:
"""Aptitude_Score 算出（0〜15点）"""
missing_flags = []
with get_db() as conn:
race = conn.execute(
"SELECT venue, distance, track_type, track_condition "
"FROM races WHERE id = ?",
(race_id,),
).fetchone()
horse = conn.execute(
"SELECT sire, dam_sire FROM horses WHERE id = ?",
(horse_id,),
).fetchone()
if not race:
missing_flags.append("race_info_missing")
return 7.5, missing_flags
# コース実績（予測対象レース以前のデータのみ参照）
with get_db() as conn:
date_filter = ""
params = [horse_id, race["venue"],
race["distance"] - 200, race["distance"] + 200]
if race_date:
date_filter = " AND race_date < ?"
params.append(race_date)
course_results = conn.execute(
f"""
SELECT finish_position FROM past_races
WHERE horse_id = ? AND venue = ? AND distance BETWEEN ? AND ?{date_filter}
ORDER BY race_date DESC LIMIT 5
""",
params,
).fetchall()
ここで意識したのは 「距離 ± 200m の窓」のような近接フィルタを使っても、時間フィルタは独立して必要 だということです。venue と distance BETWEEN だけだとリーク経路2が残ります。「距離が近い → 適性が高そう」という空間軸のフィルタと、「過去のものだけ → 因果が成立」という時間軸のフィルタは直交しており、両方とも明示的に書く必要があります。
3. 血統経由の集計に潜む第3のリーク
これが今回の記事で一番伝えたい部分です。コース実績が足りない馬（=新馬・転厩・条件外参戦）は、種牡馬産駒の同条件成績で適性を推定します。このとき past_races と horses を JOIN しますが、JOIN 先の past_races.race_date も未来の走を含みうるため、pr.race_date < ? を JOIN クエリ側にも入れる必要があります。
# services/prediction_engine.py: calculate_aptitude_score（血統推定部分）
else:
missing_flags.append("course_experience_missing")
# 血統から適性を推定（同種牡馬産駒の同条件成績を統計）
if horse and horse["sire"]:
with get_db() as conn:
sire_date_filter = ""
sire_params = [horse["sire"], race["track_type"],
race["distance"] - 400, race["distance"] + 400]
if race_date:
sire_date_filter = " AND pr.race_date < ?"
sire_params.append(race_date)
sire_results = conn.execute(
f"""SELECT pr.finish_position FROM past_races pr
JOIN horses h ON pr.horse_id = h.id
WHERE h.sire = ? AND pr.track_type = ?
AND pr.distance BETWEEN ? AND ?
AND pr.finish_position IS NOT NULL{sire_date_filter}
LIMIT 100""",
sire_params,
).fetchall()
JOIN を含むクエリでは、テーブル別名（ここでは pr）を WHERE 句で明示する習慣をつけると、複数テーブルにまたがるリーク経路を見落としづらくなります。pr.race_date < ? のように 常に「どのテーブルの race_date か」を明示し、自然キー JOIN を多用するときには JOIN 先テーブルの時間軸も独立して制限する、という規律です。
この第3経路は コードレビューで一番見落とされやすい箇所でした。理由は、関数の冒頭ですでに「コース実績」用の race_date < ? を入れているため、後段の血統経由クエリでもフィルタ済みのように錯覚するからです。「fallback ルートに同じガードが入っているか」を独立に確認するチェック項目を、コードレビューのテンプレに加えています。
4. 推論時の race_date 渡しを呼び出し側で徹底する
関数側でいくら race_date を受け取れるようにしても、呼び出し側で渡し忘れたら全部無意味です。UmaScore では「呼び出し側で必ず races.race_date を取得 → 各スコア関数に渡す」という規律を、推論パイプラインの一箇所に集約しました。
# services/prediction_engine.py: スコア合成パイプライン
with get_db() as conn:
entries = conn.execute(
"""
SELECT re.id, re.horse_id, h.horse_name, re.frame_number,
re.horse_number_in_race, re.odds_win, re.popularity,
re.running_style
FROM race_entries re
JOIN horses h ON re.horse_id = h.id
WHERE re.race_id = ?
""",
(race_id,),
).fetchall()
# 予測対象レースの race_date を取得（リーク防止の基準点）
race_date_row = conn.execute(
"SELECT race_date FROM races WHERE id = ?", (race_id,)
).fetchone()
race_date = race_date_row["race_date"] if race_date_row else None
...
# 各スコア関数に race_date を必ず渡す
base, m1 = calculate_base_ability_index(entry["horse_id"], race_id, race_date)
training, m2 = calculate_training_score(entry["horse_id"])
aptitude, m3 = calculate_aptitude_score(entry["horse_id"], race_id, race_date)
human, m4 = calculate_human_factor_score(entry["id"])
ここで採用した3つの規律は以下です。
-
基準点は
race_idから1回だけ取得: 呼び出しごとにrace_dateを取りに行くと、引数を取り違える事故が起きやすいので、パイプラインの入口で1度だけ確定させます。 -
Optional[str]のまま渡す: バックテストの初期段階や、過去レースの再評価ではNoneを許容する場面もあるため、引数自体は Optional のままにしています。ただし運用本線では必ず非 None。 -
呼び出し箇所を1ファイルに閉じ込める: 複数ファイルから個別に呼ぶと「ここだけ
race_dateを渡し忘れる」事故が起きます。スコア合成はprediction_engine.pyの特定関数に集約しました。
呼び出し側の規律はコードレビューでカバーするのが基本ですが、「calculate_base_ability_index を呼ぶ箇所で race_date を引数に渡しているか」を grep でチェックする CI ルールを入れておくと事故が減ります（実装は宿題）。
4経路をまとめて潰したあとの数値
UmaScore のバックテストは、2025年1月〜2026年4月の16ヶ月、4,409レース・61,036出走を対象に走らせています。race_date フィルタを入れる前と入れた後では、以下の質的な差がありました。
- フィルタ前: 「過去走の中に未来走を含む」状態が混入し、走力指数の分布が予測対象レース後に偏って高くなる。バックテスト集計の数値は「綺麗すぎる」状態で、フォワードテストとの乖離が大きくなる可能性が高い構造でした（フィルタ前のバックテスト集計値そのものは廃棄してログが残っていません）。
-
フィルタ後: 4経路すべてで
race_date < ?を通したあとに、EV >= 1.5戦略のバックテスト集計値が 回収率 118.4%（2025年1月〜2026年4月／16ヶ月／4,409レース／61,036出走）に落ち着きました。16ヶ月中10ヶ月がプラス（62.5%）で、月次の分散は大きく、2ヶ月連続で赤字となる期間もあります。
数値の捉え方として大事なのは「フィルタを入れて数字が下がること自体が正常」という点です。むしろ、フィルタを入れる前後で集計値が大きく変動しない場合は、対象の特徴量に時間軸の情報がほとんど効いていない（つまり特徴量として有意でない）か、あるいはまだ別経路のリークが残っているかを疑うべきです。
過去のバックテスト数値は将来の成績を保証するものではありません。実運用とのギャップを検証するため、UmaScore では4月18日から フォワードテストを公開しています（umascore.com）。
まとめ：時系列 ML 全般に通じる規律
競馬データを題材にしましたが、本記事で整理した内容は 時系列 ML 全般 に通じます。
- 集計クエリの WHERE 句に必ず「基準時刻 < ?」を入れる: 入れ忘れではなく、入れ忘れに気づける仕組みを優先する（CI / lint / コードレビューチェックリスト）。
- JOIN 先テーブルの時間軸も独立にフィルタする: 自然キー JOIN を多用するときに見落としやすい第3経路。テーブル別名を WHERE 句で明示する習慣が予防になる。
- fallback ルートを独立にチェックする: メインルートだけ確認するコードレビューでは、fallback の血統経由・近接補間・キャッシュフォールバックが素通りする。
- 基準時刻はパイプラインの入口で1回だけ取得する: 呼び出しごとに取りに行くと、引数の取り違え事故が起きやすい。
- フィルタ前後で数値が変わらない場合は別のリークか、無効な特徴量を疑う: 「数字が下がること」を品質確認の指標にする。
時系列リークは「気をつければ防げる」類のバグではなく、仕組みで防ぐ対象だと考えています。UmaScore も4経路を埋めるまでに何度かやり直しました。同じ落とし穴に向かっている方の参考になれば幸いです。
注意事項
- 本記事は技術検証の知見共有を目的としています。
- 公営競技は20歳以上の自己責任です。馬券購入の代理や投資助言は行いません。
- 過去の検証結果は将来の成績を保証するものではありません。
- 本記事の手法は特定のサービスへの加入を推奨するものではありません。
- 競馬は娯楽です。余裕資金の範囲でお楽しみください。
著者: フォワードテスト中の競馬予想ツール UmaScore（umascore.com）の開発・運営者
β1期間中は無料公開で、検証ログも公開しています。