跳到主要内容

金融交叉验证：Purged K-Fold 与禁运

直觉​

普通机器学习的 KK 折交叉验证默认打乱数据——这在 IID 场景没问题。但金融数据是时间相关的：昨天和今天的收益高度相关，而且当你用多期未来收益做标签时，相邻样本的标签会重叠（都覆盖了同一段未来）。一旦打乱，训练集里就会出现「标签和测试集标签几乎同一段未来」的样本——模型等于偷看了答案，交叉验证分数被严重高估。López de Prado 提出的 Purged K-Fold + 禁运（Embargo） 就是为修这个问题而生。

标签重叠与 Purge​

设标签是 hh 期远期收益，则样本 ii 的标签覆盖时间窗 [i, i+h][i,\, i+h]。两个样本的窗口一旦相交，就共享了信息：

重叠  ⟺  [i, i+h] ∩ [j, j+h]≠∅\text{重叠} \iff [i,\,i+h]\,\cap\,[j,\,j+h]\ne\varnothing

Purge：对每个测试折，把训练集中标签窗口与测试标签窗口相交的样本全部剔除。Embargo（禁运）：测试折结束后的若干个样本也剔除——因为它们的标签可能由测试期内的数据算出，仍带轻微泄漏。

「人话」解释：为什么普通 KFold 在金融里会「作弊」？

假设标签是「未来 3 个月收益」。4 月的样本标签覆盖 4–6 月，5 月的样本标签覆盖 5–7 月——两者都包含 5、6 月。如果 5 月样本在训练集、4 月样本在测试集，模型训练时已经「见过」测试标签里的一大部分未来，验证分数当然虚高。 Purge 就是「测试折动到的未来，训练集一概不许碰」，把这种共享未来的样本删掉，还原出诚实的样本外分数。

可运行案例：泄漏让 IC 虚高多少？​

用 3 个月远期收益作为标签（标签重叠 h=3h=3），对比两种 5 折交叉验证的平均 Rank IC：打乱 KFold（泄漏） vs Purged + 禁运（诚实）。

import numpy as np
import pandas as pd
from sklearn.ensemble import RandomForestRegressor

prices = pd.read_csv('/data/sp500_universe_daily.csv', parse_dates=['date']).set_index('date')
meta = pd.read_csv('/data/sp500_metadata.csv').set_index('ticker').loc[prices.columns]
dret = np.log(prices / prices.shift(1)); monthly = prices.resample('ME').last()

mom = lambda n: np.log(monthly / monthly.shift(n)).stack()
feats = {'mom_1': mom(1), 'mom_3': mom(3), 'mom_6': mom(6), 'mom_12': mom(12),
       'vol_3': (dret.rolling(63).std() * np.sqrt(252)).resample('ME').last().stack()}
F = pd.DataFrame(feats).join(meta['quality'], on='ticker')
h = 3
y = np.log(monthly.shift(-h) / monthly).stack()          # 3 月远期收益(标签重叠)
data = F.join(y.rename('y')).dropna().sort_index()
cols = list(F.columns)
dates = data.index.get_level_values('date').unique().sort_values()
D = len(dates)                                            # 月数 = 时间步数

def fold_dates(order, n_splits=5, h=3, embargo=1):
  fs = D // n_splits
  for k in range(n_splits):
      ti = np.array(order[k*fs:(k+1)*fs if k < n_splits-1 else D])  # 测试月(按给定顺序)
      pos = {d: i for i, d in enumerate(dates)}
      lo = pos[dates[ti.min()]]; hi = pos[dates[ti.max()]] + h       # 测试标签覆盖的时间步范围
      tr = []
      for i, d in enumerate(dates):
          if d in set(dates[ti]): continue
          if i + h >= lo and i <= hi: continue                        # purge: 标签窗口重叠
          tr.append(d)
      tr = [d for d in tr if not (pos[dates[ti.max()]] < pos[d] <= pos[dates[ti.max()]] + embargo)]  # 禁运
      yield list(dates[ti]), tr

def evaluate(order, purge=True, embargo=1):
  ics = []
  for te_d, tr_d in fold_dates(order, h=h, embargo=embargo if purge else 0):
      dtr = data[data.index.get_level_values('date').isin(tr_d or te_d)] if tr_d else data.iloc[:0]
      dte = data[data.index.get_level_values('date').isin(te_d)]
      if len(dtr) == 0: continue
      rf = RandomForestRegressor(n_estimators=100, max_depth=4, random_state=0).fit(dtr[cols], dtr['y'])
      p = pd.Series(rf.predict(dte[cols]), index=dte.index)
      ics.append(p.groupby('date').apply(lambda s: s.rank().corr(dte.loc[s.index,'y'].rank())).mean())
  return np.nanmean(ics)

chrono = list(range(D))
shuffled = list(np.random.default_rng(0).permutation(D))
naive = evaluate(shuffled, purge=False)
purged = evaluate(chrono, purge=True, embargo=1)
print(f"打乱 KFold 平均 IC  = {naive:+.3f}   ← 标签重叠泄漏, 虚高")
print(f"Purged+禁运 平均 IC = {purged:+.3f}   ← 诚实样本外")
print(f"\n泄漏让 IC 虚增约 {naive - purged:+.3f} → 这部分在实盘里根本赚不到。")

小结​

  • 金融数据时间相关 + 多期标签重叠，使打乱 KFold 严重泄漏、虚高样本外分数；
  • Purged K-Fold：剔除训练集中标签窗口与测试折重叠的样本；
  • 禁运（Embargo）：再删测试折后若干样本，防标签由测试期数据算出；
  • 永远在时间有序 + Purge 下评估模型，否则上线即缩水。