データリーク(data leakage、データ漏洩)と先読みバイアス(look-ahead bias)は、バックテストや時系列予測で 見栄えのする偽の成績 を生む典型的な原因です。本記事では、実例コードで「どこに未来情報が混入しているか」と「どう防ぐか」を一段ずつ確認します。

目次

  1. データリークの定義
  2. サンプルデータの準備
  3. A. 同日約定リーク
  4. B. 全期間統計リーク
  5. C. 未来ラベルリーク
  6. D. データ補完リーク
  7. 自動チェックの工夫
  8. チェックリスト

データリークの定義

データリークとは、本来「未来」にあるはずの情報が、訓練・検証・バックテストの 過去サイドに混入する現象 です。具体的には次の 4 系統に分類できます。

系統内容起きやすい場所
A. 同日約定当日のシグナルを当日の終値で約定バックテストのリターン計算
B. 全期間統計全期間の平均・標準偏差で標準化特徴量エンジニアリング
C. 未来ラベル将来 N 日のリターンを特徴量に混ぜる機械学習の入力作成
D. データ補完未来の値で欠損補完(後方補完)前処理

サンプルデータの準備

リーク確認用に、ランダムウォーク的な株価と日次リターンを作ります。

import numpy as np
import pandas as pd
rng = np.random.default_rng(seed=17)
n_days = 800
ret = rng.normal(loc=0.0003, scale=0.012, size=n_days)
price = 1500 * np.exp(np.cumsum(ret))
df = pd.DataFrame({"C": price}, index=pd.date_range("2022-01-04", periods=n_days, freq="B"))
df["ret"] = df["C"].pct_change()

A. 同日約定リーク

最も古典的なバグです。当日終値で計算したシグナルが、その日のリターンに掛けられている状態です。

悪い例

# 当日終値ベースの単純ルール
df["signal"] = (df["C"] > df["C"].rolling(20).mean()).astype(int)
# NG: 当日のシグナルで当日のリターンを取る(未来を使っている)
df["bad_strategy_ret"] = df["signal"] * df["ret"]

signal は当日の終値の情報を含み、ret は前日終値から当日終値へのリターンです。両者を同日に掛けると、当日終値の情報が当日のリターン計算に流れ込みます。

直し方

シグナルを 1 日 shift(1) し、ポジションとして翌営業日から有効化します。

df["position"] = df["signal"].shift(1)
df["good_strategy_ret"] = df["position"] * df["ret"]

検算として、両者の累積リターンを比べると、悪い例の方が不自然に好成績になることが多くあります。

bad_eq = (1 + df["bad_strategy_ret"].fillna(0)).cumprod()
good_eq = (1 + df["good_strategy_ret"].fillna(0)).cumprod()
print(f"NG 累積: {bad_eq.iloc[-1]:.4f}")
print(f"OK 累積: {good_eq.iloc[-1]:.4f}")

差が大きいときは、リーク防止が効いていない実装の可能性があります。

B. 全期間統計リーク

特徴量を標準化(z-score)するときに、訓練期間と検証期間を区別せず全期間の平均・標準偏差を使うと、訓練データに未来の統計情報が漏れます。

悪い例

df["feat"] = df["C"].rolling(10).mean() / df["C"].rolling(60).mean() - 1
# NG: 全期間統計で標準化
mu, sd = df["feat"].mean(), df["feat"].std(ddof=0)
df["feat_bad"] = (df["feat"] - mu) / sd

訓練期間に対して feat_bad を当てるとき、musd には検証期間の値も含まれます。これは未来情報のリークです。

直し方 1: 訓練期間で fit、検証期間に transform

split_date = df.index[int(len(df) * 0.7)]
train = df.loc[df.index < split_date].copy()
test = df.loc[df.index >= split_date].copy()
mu, sd = train["feat"].mean(), train["feat"].std(ddof=0)
train["feat_z"] = (train["feat"] - mu) / sd
test["feat_z"] = (test["feat"] - mu) / sd

sklearn を使う場合、fit を訓練データだけに、transform を訓練・検証の両方に適用するのが原則です。

from sklearn.preprocessing import StandardScaler
scaler = StandardScaler()
train_X = scaler.fit_transform(train[["feat"]])
test_X = scaler.transform(test[["feat"]])

直し方 2: ローリング統計量

過去 60 日だけを使うローリング標準化なら、未来情報を一切使いません。

window = 60
roll_mu = df["feat"].rolling(window).mean()
roll_sd = df["feat"].rolling(window).std(ddof=0)
df["feat_roll_z"] = (df["feat"] - roll_mu) / roll_sd

実装が簡潔で、訓練・検証の境界を意識しなくて良いのが利点です。窓幅は十分長く取り、初期の NaN を許容します。

C. 未来ラベルリーク

機械学習で「翌日の上がり下がり」を予測するモデルを作るとき、特徴量に 将来情報 を混ぜてしまう例です。

悪い例

# ラベル: 翌日の上昇 1 / 下降 0
df["label"] = (df["ret"].shift(-1) > 0).astype(int)
# 特徴量: 当日の SMA(これは OK)
df["feat_a"] = df["C"].rolling(20).mean()
# NG: 将来 5 日のリターン平均を特徴量にしている
df["feat_b_bad"] = df["ret"].rolling(5).mean().shift(-5)

feat_b_bad は「現在から見て未来 5 日」の平均リターンであり、ラベル label と未来情報を共有しています。学習すると訓練でほぼ完璧に当たりますが、検証や本番では機能しません。

直し方

特徴量に使う shift常に正の値 であることを確認します。

# OK: 過去 5 日のリターン平均
df["feat_b"] = df["ret"].rolling(5).mean()
# OK: 過去のリターンを 1 日ずらす(当日終値の情報すら使わないなら shift(1))
df["feat_b_safe"] = df["ret"].rolling(5).mean().shift(1)

「特徴量に shift(-N) が含まれていないか」「rolling の閉区間が当日を含むかどうか」をコードレビューで必ず確認します。

D. データ補完リーク

欠損値の補完で、未来の値を使ってしまう例です。

悪い例

df_missing = df.copy()
df_missing.loc[df_missing.index[100:103], "C"] = np.nan
# NG: 後方補完(未来の値で埋める)
df_missing["close_bad"] = df_missing["C"].bfill()

bfill(backward fill)は、欠損を 次に出てくる非欠損値 で埋めます。これは未来情報の利用です。

直し方

時系列の前から処理する場合は、ffill(forward fill、過去の値で埋める)を使うか、欠損のまま残してモデル側で扱います。

# OK: 前方補完(過去の値で埋める)
df_missing["close_good"] = df_missing["C"].ffill()

訓練データのみで欠損補完用のモデル(中央値など)を作り、検証データに適用するのも安全な選択です。

自動チェックの工夫

リークの有無を機械的に検証するのは難しい一方、簡単な仕掛けで多くを防げます。

1. シャッフルテスト

「ラベルをランダムにシャッフルしたデータで学習・予測する」と、まともなモデルなら精度はチャンスレベル(2 値分類で約 50%)になります。これを大きく超える精度が出る場合、特徴量にラベルが漏れている可能性があります。

from sklearn.linear_model import LogisticRegression
from sklearn.metrics import accuracy_score
X = df[["feat_a", "feat_b"]].dropna()
y = df.loc[X.index, "label"].dropna()
X = X.loc[y.index]
# ラベルだけシャッフル
y_shuffled = y.sample(frac=1.0, random_state=0).reset_index(drop=True)
y_shuffled.index = y.index
model = LogisticRegression()
model.fit(X.iloc[:500], y_shuffled.iloc[:500])
pred = model.predict(X.iloc[500:])
print("shuffle accuracy:", accuracy_score(y_shuffled.iloc[500:], pred))

50% から大きく外れる場合、実装を疑う材料になります。

2. 時系列順での検証

訓練・検証を時系列順に分け、過去だけで学習し未来を予測 する基本原則を守ります。詳しくは#10-9「時系列クロスバリデーションと訓練/検証分割」

split = int(len(X) * 0.7)
X_train, X_test = X.iloc[:split], X.iloc[split:]
y_train, y_test = y.iloc[:split], y.iloc[split:]

3. パイプラインでの fit / transform 分離

sklearn の Pipeline を使うと、fittransform が自動的に正しく分離されます。前処理を Pipeline に含めることで、全期間統計リークを防ぎやすくなります。

from sklearn.pipeline import Pipeline
pipe = Pipeline([
("scaler", StandardScaler()),
("clf", LogisticRegression()),
])
pipe.fit(X_train, y_train)
print("test acc:", pipe.score(X_test, y_test))

チェックリスト

実装直後に確認すべき項目です。

[ ] シグナル → ポジションの間に shift(1) が入っている
[ ] 特徴量に shift(-N) を含む列がない(目的が明確な場合のみ許可)
[ ] 標準化・スケーリングは訓練データで fit、検証データに transform
[ ] 欠損補完は ffill / 訓練データに基づくモデルで行う
[ ] 訓練・検証は時系列順に分割している
[ ] シャッフルテストで精度がチャンスレベル付近に落ちる

注意点

  • リークは「ありとあらゆる場所」で起きます。前処理・特徴量・分割の各層で疑うこと
  • 一度クリーンになっても、特徴量を 1 つ追加するだけで再びリークが入る場合があります
  • バックテストの結果が過剰に良い(年率シャープ 3 を超えるなど)ときは、まずリークを疑うのが安全です

本記事のコードはバックテスト・特徴量検証用で、実発注 API は呼びません。

生成AI へのプロンプト例

特徴量生成コードに対するリークチェックを自動化したい場合の例です。

次の特徴量生成コードを受け取って、リークの可能性を箇条書きで指摘してください。
指摘は以下のチェックポイントに沿って行ってください。
チェックポイント:
1. shift(N) の N が負の値になっている列はないか
2. rolling の窓に「未来」を含む書き方(reverse=True など)がないか
3. 標準化・正規化の fit が全期間で行われていないか
4. 欠損補完で bfill / interpolate(method=...) を使っていないか
5. ラベルと特徴量の Date が同日であり、特徴量にその日の C が含まれていないか
入力:
[ここにコードを貼る]
出力:
- 指摘箇所の行番号と理由
- 修正案(コードスニペット)

まとめ

  • データリークは「同日約定」「全期間統計」「未来ラベル」「データ補完」の 4 系統が中心
  • それぞれの典型バグは shift(1) / fit/transform の分離 / rolling の正しい向き / ffill で防げる
  • リークが疑われるときは「シャッフルテスト」「時系列順分割」「Pipeline 化」の 3 点で機械的に確認
  • 過剰に好成績なバックテストは、まずリークを疑う