はじめに
「この騎手の勝率は何%か」という特徴量は競馬AIで重要ですが、計算方法を間違えるとデータリークが生じます。
全期間の勝率を計算して特徴量にすると「未来の結果」が含まれてしまいます。正しくは「そのレースより前のデータだけ」で計算する必要があります。この記事では expanding_mean と rolling_mean、グループ別集計でのリーク防止を実装します。
よくある間違い：全期間集計
import pandas as pd
import numpy as np
# 悪い例：全データで勝率を計算してから特徴量に使う
df['jockey_win_rate'] = df.groupby('jockey_code')['won'].transform('mean')
# → バックテスト時に「未来のレース結果」が含まれる（リーク！）
Expanding Mean（累積平均）：正しいリーク防止
def add_jockey_stats_expanding(df: pd.DataFrame) -> pd.DataFrame:
"""
騎手の勝率・複勝率を expanding mean（累積平均）で計算する。
日付でソート後、各行は「その行より前の行だけ」で計算される。
前提: df は race_date の昇順でソート済みであること。
"""
df = df.copy()
# 必ず日付でソートしてから処理する
df = df.sort_values('race_date').reset_index(drop=True)
for stat_col, new_col in [
('won', 'jockey_win_rate'),
('place2', 'jockey_place2_rate'),
('place3', 'jockey_place3_rate'),
]:
if stat_col not in df.columns:
continue
# shift(1): 自分自身のレース結果を含めない（重要！）
df[new_col] = (
df.groupby('jockey_code')[stat_col]
.transform(lambda x: x.shift(1).expanding().mean())
)
return df
# 同様に調教師の勝率も計算
def add_trainer_stats_expanding(df: pd.DataFrame) -> pd.DataFrame:
df = df.copy()
df = df.sort_values('race_date').reset_index(drop=True)
for stat_col, new_col in [
('won', 'trainer_win_rate'),
('place3', 'trainer_place3_rate'),
]:
if stat_col not in df.columns:
continue
df[new_col] = (
df.groupby('trainer_code')[stat_col]
.transform(lambda x: x.shift(1).expanding().mean())
)
return df
Rolling Mean：直近N走の勝率
直近の調子を見るには固定ウィンドウの rolling が適しています。
def add_rolling_stats(
df: pd.DataFrame,
window: int = 30,
min_periods: int = 5,
) -> pd.DataFrame:
"""
直近 window 走の勝率を計算する。
Args:
window: ローリングウィンドウのサイズ（走数）
min_periods: 有効な値を返すための最低データ数
"""
df = df.copy()
df = df.sort_values('race_date').reset_index(drop=True)
for jockey_col in ['jockey_code', 'trainer_code']:
prefix = 'jockey' if jockey_col == 'jockey_code' else 'trainer'
for stat_col, new_col in [
('won', f'{prefix}_win_rate_last{window}'),
('place3', f'{prefix}_place3_rate_last{window}'),
]:
if stat_col not in df.columns:
continue
df[new_col] = (
df.groupby(jockey_col)[stat_col]
.transform(lambda x: x.shift(1).rolling(
window=window,
min_periods=min_periods
).mean())
)
return df
# 複数のウィンドウサイズで計算
def add_multi_window_stats(df: pd.DataFrame) -> pd.DataFrame:
"""
直近10走・30走・100走の勝率を全て特徴量に追加する。
"""
df = df.copy()
df = df.sort_values('race_date').reset_index(drop=True)
windows = [10, 30, 100]
for w in windows:
df[f'jockey_win_rate_last{w}'] = (
df.groupby('jockey_code')['won']
.transform(lambda x: x.shift(1).rolling(w, min_periods=3).mean())
)
return df
コース別・条件別の集計
単純な勝率だけでなく、「特定のコースでの勝率」を計算するとより精度が上がります。
def add_jockey_course_stats(df: pd.DataFrame) -> pd.DataFrame:
"""
騎手×コース別の勝率を計算する（expanding mean）。
コース = 会場（東京・阪神等）+ 芝/ダート + 距離帯
"""
df = df.copy()
df = df.sort_values('race_date').reset_index(drop=True)
# コース識別子を作成
df['course_key'] = (
df['venue_code'].astype(str) + '_' +
df['race_type'].astype(str) + '_' +
pd.cut(df['distance'], bins=[0, 1400, 1800, 2200, 9999],
labels=['sprint', 'mile', 'inter', 'long']).astype(str)
)
# 騎手×コース別の勝率
df['jockey_course_win_rate'] = (
df.groupby(['jockey_code', 'course_key'])['won']
.transform(lambda x: x.shift(1).expanding().mean())
)
return df
件数が少ない場合の補正（ベイズ平均）
新人騎手は過去のデータが少なく、勝率が0%または100%になりがちです。全体平均で補正するベイズ平均が有効です。
def bayesian_win_rate(
individual_wins: float,
individual_races: float,
global_win_rate: float,
prior_strength: float = 50.0,
) -> float:
"""
ベイズ平均による勝率の補正。
Args:
individual_wins: その騎手の勝利数
individual_races: その騎手のレース数
global_win_rate: 全騎手の平均勝率
prior_strength: 事前分布の強さ（大きいほど全体平均に引き寄せる）
Returns:
補正済み勝率
"""
numerator = individual_wins + global_win_rate * prior_strength
denominator = individual_races + prior_strength
return numerator / denominator
def add_bayesian_jockey_rate(df: pd.DataFrame) -> pd.DataFrame:
"""ベイズ平均補正した騎手勝率を追加する"""
df = df.copy()
df = df.sort_values('race_date').reset_index(drop=True)
# 累積の勝利数・レース数を計算
df['jockey_cum_wins'] = (
df.groupby('jockey_code')['won']
.transform(lambda x: x.shift(1).expanding().sum())
)
df['jockey_cum_races'] = (
df.groupby('jockey_code')['won']
.transform(lambda x: x.shift(1).expanding().count())
)
# 全体平均勝率（全期間）
global_rate = df['won'].mean()
# ベイズ補正
df['jockey_win_rate_bayes'] = df.apply(
lambda row: bayesian_win_rate(
row['jockey_cum_wins'] or 0,
row['jockey_cum_races'] or 0,
global_rate,
) if pd.notna(row['jockey_cum_races']) else global_rate,
axis=1
).astype(np.float32)
return df
パフォーマンス比較：apply vs transform
import time
# apply は遅い
start = time.time()
df['jockey_win_rate_apply'] = df.apply(
lambda row: compute_win_rate(row['jockey_code'], row['race_date'], df),
axis=1
)
print(f"apply: {time.time()-start:.1f}秒")
# transform は速い
start = time.time()
df['jockey_win_rate_transform'] = (
df.groupby('jockey_code')['won']
.transform(lambda x: x.shift(1).expanding().mean())
)
print(f"transform: {time.time()-start:.1f}秒")
# transform が apply より数十倍速い
まとめ
-
shift(1).expanding().mean()でリークなしの累積平均勝率を計算する -
sort_values('race_date')を transform の前に必ず実行する - 直近の調子は
rolling(window=30, min_periods=5)で計算する - 件数が少ない騎手にはベイズ平均で全体平均に引き寄せる補正をする
-
applyよりtransformが大幅に速い
詳細な実装について
競馬AIで使っている騎手・調教師・馬主・父系統別の集計特徴量（コース別・条件別・季節別）の完全実装に興味がある方へ。
詳細はnoteの有料シリーズで解説しています：