テストでAUC 0.87、本番で勝率48%だった話
少し前に、トヨタ（7203）の翌日騰落を機械学習で予測しようとLightGBMを回しました。テストデータでAUC 0.87と表示されて「これは勝ったな」と一瞬思ったのですが、1ヶ月実運用して勝率48%。手数料込みでマイナスでした。
原因を突き止めるのに数日かかりました。リークは1箇所ではなく、3箇所ありました。そのうち1つは、ネット上のサンプルコードにも普通に載っている書き方です。しかも見た目には何の問題もありません。
この記事では、その3つを再現コード付きで説明します。同じように「テストでは良いのに実運用でダメ」という人は、たぶんどれかに当てはまります。
犯人1：全期間でスケーリングしていた
これは有名なやつです。標準化を分割前にやると、テスト期間の平均と分散が学習に混ざります。
import numpy as np
import pandas as pd
from sklearn.preprocessing import StandardScaler
# ❌ 全期間でfit
scaler = StandardScaler().fit(X)
X_scaled = scaler.transform(X)
# ✅ 学習データだけでfit
scaler = StandardScaler().fit(X.iloc[train_idx])
X_tr = scaler.transform(X.iloc[train_idx])
X_te = scaler.transform(X.iloc[test_idx])木系のモデル（LightGBM、ランダムフォレスト）は標準化の影響を受けないので、これだけならAUC 0.87にはなりません。ただ、同じ発想で「全期間の分位点でビン分割」や「全期間の順位（rank）」を特徴量にすると、木系でも一気にスコアが跳ねます。自分はここもやっていました。
# ❌ 全期間の順位。未来の分布を知っていることになる
f["ret_rank"] = f["ret1"].rank(pct=True)
# ✅ 過去だけで計算する順位
f["ret_rank"] = f["ret1"].rolling(252).rank(pct=True)犯人2：特徴量が翌日を見ていた
# ❌ centerを付けると、中心の前後を使う＝未来を含む
f["ma20"] = df["Close"].rolling(20, center=True).mean()
# ❌ bfillは未来の値で埋める
f["vol"] = df["Volume"].bfill()
# ❌ 差分を取ったあとにshift(-1)が残っている
f["momentum"] = df["Close"].pct_change().shift(-1)意図的に書く人はいませんが、試行錯誤の途中で入れたshift(-1)が残ることは普通に起こります。自分の場合、ボラティリティ特徴量を作るときにcenter=Trueを付けたままでした。
検知は簡単で、各特徴量と目的変数の相関を見るだけです。
def leak_scan(f: pd.DataFrame, target: str = "target",
thresh: float = 0.15) -> None:
cols = [c for c in f.columns
if c != target and pd.api.types.is_numeric_dtype(f[c])]
corr = f[cols].apply(lambda s: s.corr(f[target])).abs()
corr = corr.sort_values(ascending=False)
print(corr.round(4).head(10))
bad = corr[corr > thresh]
if len(bad):
print(f"\n⚠ 相関が高すぎる特徴量: {list(bad.index)}")
print(" 翌日方向と0.15を超えて相関する変数は、まず存在しません")翌日の方向と0.15以上相関する特徴量は、現実には存在しません。出たら喜ぶ前に疑う。これを習慣にしてから、事故が激減しました。
犯人3：複数銘柄をconcatしてTimeSeriesSplitしていた
これが本命でした。しかも、いろいろな記事で見かける書き方です。
# 一見なんの問題もなさそうなコード
frames = []
for t in ["7203.T", "6501.T", "4063.T"]:
d = yf.download(t, period="5y", progress=False)
f = make_features(d)
f["ticker"] = t
frames.append(f)
data = pd.concat(frames).sort_index() # ← ここ
for tr, te in TimeSeriesSplit(n_splits=5).split(X):
...何が起きているか。3銘柄を連結して日付でソートすると、同じ日付の行が3つ並びます。ところがTimeSeriesSplitは行番号で分割するので、境界が日付の途中に落ちます。
つまり、2024年5月1日のトヨタは学習データ、同じ日の日立はテストデータという状態になります。日本の大型製造株は日々の値動きが強く相関しているので、これは「テスト日の相場が上げだったか下げだったか」を学習データで教えているのと同じです。
def show_boundary_leak(data: pd.DataFrame, n_splits: int = 5) -> None:
idx = data.index
for i, (tr, te) in enumerate(TimeSeriesSplit(n_splits=n_splits).split(data)):
overlap = set(idx[tr]) & set(idx[te])
print(f"fold{i}: 学習と検証で日付が重複 = {len(overlap)}日")
if overlap:
print(f" 例: {sorted(overlap)[:3]}")
show_boundary_leak(data)重複日数が0でなければアウトです。自分の場合、各フォールドで2日ぶんが重複していました。たった2日ですが、相関の高い銘柄では十分に効きます。
さらに深刻な問題があります。重複がゼロでも、まだリークします。特徴量にret20（20日リターン）を使っていると、テスト期間の最初の20日は学習期間の価格を参照しています。目的変数も1日先を見ているので、境界をまたいだ情報が漏れます。
正しい分割：日付でグループ化して間隔を空ける
解決策は2つの組み合わせです。日付単位で分割する（グループ分割）と、学習と検証の間に空白期間を置く（パージ／エンバーゴ）。
from dataclasses import dataclass
@dataclass
class Split:
train: np.ndarray
test: np.ndarray
def purged_date_split(dates: pd.Series, n_splits: int = 5,
embargo: int = 25) -> list[Split]:
"""日付単位で時系列分割し、学習と検証の間にembargo日の空白を置く。"""
uniq = np.array(sorted(dates.unique()))
n = len(uniq)
fold = n // (n_splits + 1)
out = []
for i in range(1, n_splits + 1):
tr_end = fold * i
te_start = tr_end + embargo # 空白を挟む
te_end = min(te_start + fold, n)
if te_start >= te_end:
break
tr_dates = set(uniq[:tr_end])
te_dates = set(uniq[te_start:te_end])
assert not (tr_dates & te_dates), "日付が重複しています"
out.append(Split(
train=np.where(dates.isin(tr_dates))[0],
test=np.where(dates.isin(te_dates))[0],
))
return outembargoはいちばん長い特徴量の窓に合わせます。ret20や20日ボラティリティを使うなら25日程度。ここを0にすると、境界付近の行が学習期間の情報を持ったままテストされます。
assertを入れているのが大事です。リークしない実装を書くのではなく、リークしたら止まる実装を書く。人間の注意力は当てになりません。
やり直した検証コード
import yfinance as yf
import lightgbm as lgb
from sklearn.metrics import roc_auc_score
from scipy import stats
TICKERS = ["7203.T", "6501.T", "4063.T", "6861.T"] # トヨタ 日立 信越 キーエンス
def make_features(df: pd.DataFrame) -> pd.DataFrame:
c, v = df["Close"], df["Volume"]
r = c.pct_change()
f = pd.DataFrame(index=df.index)
f["ret1"] = r
f["ret5"] = c.pct_change(5)
f["ret20"] = c.pct_change(20)
f["vol20"] = r.rolling(20).std()
f["vol_ratio"] = r.rolling(5).std() / r.rolling(60).std()
f["turnover"] = v / v.rolling(20).mean()
f["ma20_dev"] = c / c.rolling(20).mean() - 1
f["hl_range"] = (df["High"] - df["Low"]) / c
f["fwd"] = r.shift(-1) # 評価用（特徴量には使わない）
f["target"] = (f["fwd"] > 0).astype(int)
return f.dropna()
def load_panel(tickers: list[str], period: str = "10y") -> pd.DataFrame:
frames = []
for t in tickers:
d = yf.download(t, period=period, auto_adjust=True, progress=False)
if isinstance(d.columns, pd.MultiIndex):
d.columns = d.columns.get_level_values(0)
f = make_features(d)
f["ticker"] = t
frames.append(f.reset_index().rename(columns={"Date": "date"}))
return pd.concat(frames, ignore_index=True).sort_values(
["date", "ticker"]).reset_index(drop=True)
panel = load_panel(TICKERS)
FEATS = ["ret1", "ret5", "ret20", "vol20", "vol_ratio",
"turnover", "ma20_dev", "hl_range"]
leak_scan(panel[FEATS + ["target"]])def run(panel: pd.DataFrame, embargo: int = 25,
cost: float = 0.001) -> pd.DataFrame:
X = panel[FEATS].to_numpy()
y = panel["target"].to_numpy()
fwd = panel["fwd"].to_numpy()
rows = []
for i, sp in enumerate(purged_date_split(panel["date"], 5, embargo)):
m = lgb.LGBMClassifier(
n_estimators=600, learning_rate=0.02,
num_leaves=15, min_child_samples=100,
subsample=0.8, subsample_freq=1, colsample_bytree=0.7,
reg_lambda=5.0, random_state=42, verbosity=-1,
)
m.fit(X[sp.train], y[sp.train])
p = m.predict_proba(X[sp.test])[:, 1]
yt, ft = y[sp.test], fwd[sp.test]
pos = (p > 0.5).astype(float)
pnl = pos * ft - np.abs(np.diff(np.r_[0.0, pos])) * cost
hits = int(((p > 0.5) == yt).sum())
ci = stats.binomtest(hits, len(yt), 0.5).proportion_ci(0.95)
rows.append({
"fold": i,
"n": len(yt),
"AUC": round(roc_auc_score(yt, p), 4),
"精度": round(hits / len(yt), 4),
"CI下限": round(ci.low, 3),
"ベース": round(max(yt.mean(), 1 - yt.mean()), 3),
"戦略%": round(((1 + pnl).prod() - 1) * 100, 2),
"B&H%": round(((1 + ft).prod() - 1) * 100, 2),
})
return pd.DataFrame(rows).set_index("fold")
print(run(panel))これで測り直したところ、AUCは0.51〜0.55、精度はベースラインとほぼ同じでした。0.87はどこにもありません。
エンバーゴの効果を確かめる
「本当にそんなに変わるのか」と思ったので、条件を変えて比較しました。
def compare_splits(panel: pd.DataFrame) -> None:
print("=== 行番号で分割（よくある書き方）===")
X, y = panel[FEATS].to_numpy(), panel["target"].to_numpy()
aucs = []
for tr, te in TimeSeriesSplit(n_splits=5).split(X):
m = lgb.LGBMClassifier(n_estimators=600, learning_rate=0.02,
num_leaves=15, random_state=42, verbosity=-1)
m.fit(X[tr], y[tr])
aucs.append(roc_auc_score(y[te], m.predict_proba(X[te])[:, 1]))
print(f"AUC平均 = {np.mean(aucs):.4f}")
for emb in (0, 5, 25, 60):
r = run(panel, embargo=emb)
print(f"日付分割 embargo={emb:2d}日 → AUC平均 = {r['AUC'].mean():.4f}")
compare_splits(panel)行番号分割のほうが常に高いAUCを出します。そしてエンバーゴを伸ばすほどスコアは下がります。下がるのが正常です。上がったぶんは全部、境界からの漏れでした。
| 分割方法 | AUCの傾向 | 実運用との一致 |
|---|---|---|
| ランダム分割（KFold） | 0.7以上 | まったく一致しない |
| 行番号でTimeSeriesSplit | 0.56〜0.60 | 過大評価 |
| 日付分割・エンバーゴなし | 0.54前後 | やや過大 |
| 日付分割＋エンバーゴ25日 | 0.51〜0.55 | おおむね一致 |
AUC 0.55は「しょぼい」のか
最初は0.55という数字にがっかりしました。でも今は、正しく測って0.55なら十分だと思っています。理由は2つあります。
1つは、翌日の方向を素朴な価格特徴量だけで予測しているから。使っている情報が価格と出来高しかない以上、これ以上出るほうが不自然です。
もう1つは、AUC 0.55でも、確信度の高い一部だけを取れば使える可能性があることです。全日取引する必要はありません。
def selective(panel: pd.DataFrame, top_pct: float = 0.1,
cost: float = 0.001) -> None:
"""確信度が上位top_pctの日だけエントリーする。"""
X, y = panel[FEATS].to_numpy(), panel["target"].to_numpy()
fwd = panel["fwd"].to_numpy()
for sp in purged_date_split(panel["date"], 5, 25)[-1:]:
m = lgb.LGBMClassifier(n_estimators=600, learning_rate=0.02,
num_leaves=15, min_child_samples=100,
random_state=42, verbosity=-1)
m.fit(X[sp.train], y[sp.train])
p = m.predict_proba(X[sp.test])[:, 1]
ft = fwd[sp.test]
thr = np.quantile(p, 1 - top_pct)
sel = p >= thr
print(f"全体 : n={len(ft):5d} 平均={ft.mean() * 100:+.4f}%")
print(f"上位{top_pct:.0%}: n={sel.sum():5d} "
f"平均={ft[sel].mean() * 100:+.4f}% "
f"コスト後={(ft[sel].mean() - 2 * cost) * 100:+.4f}%")
t, pv = stats.ttest_1samp(ft[sel], 0.0)
print(f" t={t:.2f} p={pv:.3f}")
selective(panel)上位10%の日の平均リターンが全体平均より明確に高く、かつ往復コスト0.2%を引いてもプラスなら、意味があります。ただし、自分のケースではコストを引くとほぼゼロでした。そこが現実です。
リークを二度とやらないための仕組み
気をつける、では防げません。テストにします。
def test_no_leak(panel: pd.DataFrame) -> None:
"""CIやスクリプト冒頭で毎回走らせる検査。"""
# 1. 学習と検証で日付が重複していないか
for sp in purged_date_split(panel["date"], 5, 25):
d_tr = set(panel["date"].iloc[sp.train])
d_te = set(panel["date"].iloc[sp.test])
assert not (d_tr & d_te), "日付が重複"
assert min(d_te) > max(d_tr), "検証期間が学習期間より前にある"
# 2. 特徴量が目的変数と異常に相関していないか
for c in FEATS:
r = abs(panel[c].corr(panel["target"]))
assert r < 0.15, f"{c} の相関が高すぎる: {r:.3f}"
# 3. 特徴量にNaNや無限大が残っていないか
assert np.isfinite(panel[FEATS].to_numpy()).all(), "非有限値がある"
# 4. 目的変数をシャッフルしたらAUCが0.5付近に落ちるか
rng = np.random.default_rng(0)
y_shuf = rng.permutation(panel["target"].to_numpy())
sp = purged_date_split(panel["date"], 5, 25)[-1]
X = panel[FEATS].to_numpy()
m = lgb.LGBMClassifier(n_estimators=300, num_leaves=15,
random_state=0, verbosity=-1)
m.fit(X[sp.train], y_shuf[sp.train])
auc = roc_auc_score(y_shuf[sp.test], m.predict_proba(X[sp.test])[:, 1])
assert 0.45 < auc < 0.55, f"シャッフルでAUC={auc:.3f}。配管がおかしい"
print("リーク検査: 問題なし")
test_no_leak(panel)4番目が特に効きます。目的変数をランダムに並べ替えたのにAUCが0.6を超えるなら、検証の仕組み自体が壊れています。モデルの問題ではなく、コードの問題です。ここで引っかかったことが実際に一度ありました。
まとめ
- AUC 0.87の正体は3種類のリーク。全期間スケーリング、未来を見る特徴量、そして分割の境界
- 複数銘柄を
concatして行番号で分割すると、同じ日が学習と検証に分かれる - 日付単位で分割し、最長の特徴量窓ぶんのエンバーゴを空ける
- 特徴量と目的変数の相関が0.15を超えたら疑う
- ラベルをシャッフルしてAUCが0.5に戻るかを毎回テストする
- 正しく測ったAUC 0.55は、しょぼくない。0.8が出たら壊れている
機械学習トレードで大事なのは、派手に勝つモデルより期待どおりに動く地味なモデルです。そして期待どおりかどうかは、検証の仕組みが正しくないと分かりません。
同じようにハマっている人は、まずtest_no_leakのようなテストを書いてみてください。自分はこれを書いた日に、過去のモデルが全部おかしかったことを知りました。落ち込みましたが、それ以降は実運用との乖離がほとんどなくなりました。
