はじめに
機械学習の評価で「CVスコアは高いのに本番でボロボロ」という経験はありませんか?
競馬データで KFold をそのまま使うと、未来のデータで学習して過去を予測するという「リーク」が発生します。これを防ぐには、時系列の構造を考慮したクロスバリデーション(CV)が必要です。
この記事では、競馬の時系列特性に合わせた GroupKFold および TimeSeriesSplit の実装を解説します。
なぜ通常のKFoldではダメか
from sklearn.model_selection import KFold
# これは競馬データでは使ってはいけない
kf = KFold(n_splits=5, shuffle=True, random_state=42)
# shuffle=True にすると 2025年のデータで学習して 2020年を予測する分割が起きる
競馬データでは:
- 馬の調子・馬場状態・騎手の調子はトレンドがある
- 「未来を使って過去を予測」するとリークが生じる
- CVスコアが実際より良く見える(過大評価)
方法1:TimeSeriesSplit(最もシンプル)
import numpy as np
import pandas as pd
import lightgbm as lgb
from sklearn.model_selection import TimeSeriesSplit
from sklearn.metrics import roc_auc_score
def cv_time_series(
df: pd.DataFrame,
feature_cols: list[str],
label_col: str,
n_splits: int = 5,
lgb_params: dict = None,
) -> list[float]:
"""
TimeSeriesSpritによる時系列CV。
dfは日付昇順でソート済みであること。
"""
if lgb_params is None:
lgb_params = {
'objective': 'binary',
'metric': 'auc',
'n_estimators': 1000,
'learning_rate': 0.05,
'num_leaves': 63,
'verbose': -1,
}
X = df[feature_cols].values
y = df[label_col].values
tscv = TimeSeriesSplit(n_splits=n_splits)
auc_scores = []
for fold, (train_idx, val_idx) in enumerate(tscv.split(X)):
X_train, X_val = X[train_idx], X[val_idx]
y_train, y_val = y[train_idx], y[val_idx]
model = lgb.LGBMClassifier(**lgb_params)
model.fit(
X_train, y_train,
eval_set=[(X_val, y_val)],
callbacks=[
lgb.early_stopping(50, verbose=False),
lgb.log_evaluation(period=-1),
]
)
val_pred = model.predict_proba(X_val)[:, 1]
auc = roc_auc_score(y_val, val_pred)
auc_scores.append(auc)
# 検証期間を確認(リークのデバッグに便利)
if 'race_date' in df.columns:
val_dates = df.iloc[val_idx]['race_date']
train_dates = df.iloc[train_idx]['race_date']
print(f"Fold {fold+1}: 学習 {train_dates.min()}〜{train_dates.max()}, "
f"検証 {val_dates.min()}〜{val_dates.max()}, AUC={auc:.4f}")
print(f"\nCV AUC: {np.mean(auc_scores):.4f} ± {np.std(auc_scores):.4f}")
return auc_scores
方法2:年単位のGroupKFold(競馬に最適)
競馬では年単位でデータを分割するのが最も自然です。「直近N年をテスト」という設計にすることで、実運用に近い評価ができます。
from sklearn.model_selection import GroupKFold
def cv_by_year(
df: pd.DataFrame,
feature_cols: list[str],
label_col: str,
year_col: str = 'year',
n_splits: int = 5,
lgb_params: dict = None,
) -> dict:
"""
年をグループとしたGroupKFold CV。
各foldで「特定の年」が丸ごと検証データになる。
"""
if lgb_params is None:
lgb_params = {
'objective': 'binary',
'metric': 'auc',
'n_estimators': 1000,
'learning_rate': 0.05,
'num_leaves': 63,
'verbose': -1,
}
X = df[feature_cols].values
y = df[label_col].values
groups = df[year_col].values
gkf = GroupKFold(n_splits=n_splits)
results = []
for fold, (train_idx, val_idx) in enumerate(gkf.split(X, y, groups)):
X_train, X_val = X[train_idx], X[val_idx]
y_train, y_val = y[train_idx], y[val_idx]
val_years = np.unique(groups[val_idx])
model = lgb.LGBMClassifier(**lgb_params)
model.fit(
X_train, y_train,
eval_set=[(X_val, y_val)],
callbacks=[
lgb.early_stopping(50, verbose=False),
lgb.log_evaluation(period=-1),
]
)
val_pred = model.predict_proba(X_val)[:, 1]
auc = roc_auc_score(y_val, val_pred)
results.append({
'fold': fold + 1,
'val_years': val_years.tolist(),
'n_train': len(train_idx),
'n_val': len(val_idx),
'auc': auc,
})
print(f"Fold {fold+1}: 検証年={val_years}, AUC={auc:.4f}")
auc_list = [r['auc'] for r in results]
print(f"\nCV AUC: {np.mean(auc_list):.4f} ± {np.std(auc_list):.4f}")
return results
方法3:ウォークフォワード検証(最も厳密)
本番運用に最も近い評価方法は「過去で学習して直近で検証」を繰り返すウォークフォワードです。
def walk_forward_cv(
df: pd.DataFrame,
feature_cols: list[str],
label_col: str,
year_col: str = 'year',
train_start_year: int = 1999,
test_start_year: int = 2021,
test_end_year: int = 2025,
lgb_params: dict = None,
) -> pd.DataFrame:
"""
ウォークフォワード検証。
各ステップで「それ以前の全データ」で学習し「1年分」を評価する。
"""
if lgb_params is None:
lgb_params = {'objective': 'binary', 'n_estimators': 500, 'verbose': -1}
results = []
for test_year in range(test_start_year, test_end_year + 1):
# 学習:test_year より前の全データ
train_mask = df[year_col] < test_year
val_mask = df[year_col] == test_year
if train_mask.sum() == 0 or val_mask.sum() == 0:
continue
X_train = df.loc[train_mask, feature_cols].values
y_train = df.loc[train_mask, label_col].values
X_val = df.loc[val_mask, feature_cols].values
y_val = df.loc[val_mask, label_col].values
model = lgb.LGBMClassifier(**lgb_params)
model.fit(X_train, y_train)
val_pred = model.predict_proba(X_val)[:, 1]
auc = roc_auc_score(y_val, val_pred)
results.append({
'test_year': test_year,
'n_train': len(X_train),
'n_val': len(X_val),
'auc': auc,
})
print(f"{test_year}年: 学習{len(X_train)}件, 検証{len(X_val)}件, AUC={auc:.4f}")
return pd.DataFrame(results)
GroupKFold vs TimeSeriesSplit の使い分け
| 方法 | 特徴 | 競馬での用途 |
|---|---|---|
TimeSeriesSplit |
シンプルで使いやすい | 特徴量の大まかな評価 |
GroupKFold(年) |
年単位で厳密に分割 | 年次トレンドがある特徴量の評価 |
| ウォークフォワード | 最も厳密・計算コスト高 | 最終モデルの性能評価 |
競馬AIの開発では、最終評価はウォークフォワードで行い、特徴量の実験は TimeSeriesSplit で素早く確認するのがおすすめです。
注意:GroupKFoldは順序を保証しない
# GroupKFold の注意点
# 「2021年が検証→2020年で学習」という逆転が起きることがある
# 競馬では必ず val_years > train_years になっているか確認する
for fold, (train_idx, val_idx) in enumerate(gkf.split(X, y, groups)):
train_years = np.unique(groups[train_idx])
val_years = np.unique(groups[val_idx])
# 検証年が学習年より全て後であることを確認
if val_years.min() <= train_years.max():
print(f"⚠ Fold {fold+1}: 時系列逆転の可能性あり")
print(f" 学習: {train_years}, 検証: {val_years}")
GroupKFold はグループを分割するだけで時系列順序は保証しません。競馬データでは TimeSeriesSplit か自前のウォークフォワードが安全です。
まとめ
- KFold + shuffle=True は競馬データでは使わない(リークが発生する)
- TimeSeriesSplit が最もシンプルで安全
- GroupKFold(年単位) で年次パターンを評価できる
- ウォークフォワード が本番に最も近い厳密な評価
- GroupKFold は時系列順序を保証しないため、逆転チェックを必ず行う
詳細な実装について
競馬AIの実際のCV設計(学習期間の決め方、テスト年の設定、OOF予測値の生成)に興味がある方へ。
詳細はnoteの有料シリーズで解説しています:
競馬AI診断(無料)
あなたが考えた買い目を、通算回収率121%のAIが診断するWebアプリ