跳到正文
本地阅读 · 时间验证与模型选择

时间验证与模型选择

机器学习 · 机器学习 · P0
本章目录

学完后，你应能独立完成

  • 推导不同持有期和截止政策的训练边界
  • 设计日期 walk-forward 与最终留出
  • 检查折内预处理、早停、OOF 和预测键

前置建议：研究问题与数据约定 · Ridge、Lasso 与因子合成

OBSERVE / 交互实验

改变一个条件，观察发生什么

浏览器内即时教学计算；Python 实验结果另列于代码区。图形与数值一起核对。

01

先决定要泛化到哪里

随机行 K-fold 评估混合历史状态中的新行；股票分组回答新股票；未来日期验证回答以后市场中的预测。它们不是可互换的分数。TimeSeriesSplit 按行位置切，长表的 gap=5 只代表五行，不能自动代表五个交易日。

先从唯一日期生成折，再把日期映射回全部股票行。同一日期不跨训练与验证；进一步检查标签实际可得性。Pipeline 保护折内 fit，却不能修复已经错位的标签或非法 splitter。

02

从时点推导最新训练标签

验证第一信号日期位置为 s，训练在 s−1 收盘。标签退出为 t+k+1 开盘，且假设当天开盘标签立即到达。因此 t+k+1≤s−1，即 t≤s−k−2。k=5 时保留到 s−7，删除最后六个候选信号日期。

如果训练改为 s 日收盘，或者供应商晚一个交易日到达，边界会变化。应比较真实 timestamp，而不是固定写“删五天”。页面的延迟定义为退出开盘后额外整数交易日到达，便于核对。

\[\mathrm{feature\_available}\le\mathrm{signal},\quad \mathrm{label\_available}\le\mathrm{cutoff},\quad t\le s-k-2\]
03

purge、embargo 与成熟过滤

成熟过滤直接检查在线训练能否看到标签；purge 删除训练中与留出事件区间重叠的标签。某些 purged K-fold 允许用测试段之后作为训练，还需要声明测试后 embargo。纯过去训练的额外前置 gap 是另一项政策，不能只因为用了同样名字就宣称等同所有文献的 embargo。

共享价格端点、共享收益增量和共享市场状态要区分。没有共同收益增量也不意味着独立；重叠简单收益或非线性 IC 不一般严格是 MA(k−1)。处理推断要考虑实际依赖，而不是机械代入一个阶数。

04

rolling 与 expanding 都是一项选择

Expanding 使用全部成熟历史，rolling 使用最近窗口。短窗口减少旧状态影响，也减少训练信息。窗长、再训练频率、标签延迟和额外 gap 都要事先定义或只在开发期选择。

后来训练包含之前的验证段，只要标签届时已成熟，并不会污染已保存的早期未来预测。真正要评价的是整个选择和更新政策。反复在同一开发集挑最高分会过拟合，需外层或最终未参与选择的留出。

05

从 Pipeline 到时间合法 OOF

每折在训练内拟合填补、scaler、PCA、特征筛选、模型。开发折决定参数和轮数，最终测试前锁政策。时间 OOF 不仅要求该行没参与拟合，还要求生成它的基模型没有使用该行之后的不可用标签。

元模型训练时，它的目标也要成熟。最早没有足够历史产生 OOF 的段留空，不能用训练内预测补齐。对误差分布和政策变化保持警惕，不把“折外”当作自动合法。

06

让错误在入口就失败

下方新实现会验证重复键、缺失可得时点、晚到特征与已知标签的未知可得时点。评价以 (date,stock) 合并预测，要求键集合完全相同。乱序可以对齐，缺键或重复键要报错，而不是静默丢行。

原辅助代码的时间切分依赖生成器上游校验，评价又按行序读取。这里把这两项迁移条件落实为接口检查。断言和小样本验证不能保证任意输入正确，但能让主要研究约定可执行。

IMPLEMENT / 完整代码

从算法到一次真实运行

代码在本地 Python 中执行，HTML 内展示参考实现和本轮运行结果。请先独立完成练习，再展开完整参考文件。

validation.py / 核心函数
def make_panel(seed=42, T=180, N=24, k=5, mechanism="linear"):
    """Illustrative weekdays, 6 AR factors, future open-to-open k-day label."""
    if mechanism not in {"null","linear","interaction","regime"}:
        raise ValueError("Unknown mechanism")
    rng=np.random.default_rng(seed)
    dates=pd.bdate_range("2020-01-02",periods=T)
    factors=np.empty((T,N,6)); factors[0]=rng.normal(size=(N,6))
    for t in range(1,T):
        factors[t]=.6*factors[t-1]+.8*rng.normal(size=(N,6))
    factors[:,:,3]=.9*factors[:,:,0]+.15*rng.normal(size=(T,N))
    prices=np.full((T,N),100.)
    for t in range(T-1):
        x=factors[max(0,t-1)]
        signal=np.zeros(N)
        if mechanism!="null": signal=.0015*x[:,0]
        if mechanism in {"interaction","regime"}: signal += .0015*x[:,1]*x[:,2]
        if mechanism=="regime" and t>=int(.65*T): signal -= .002*x[:,0]
        noise=.006*rng.standard_t(5,N)/np.sqrt(5/3)+rng.normal(0,.002)
        prices[t+1]=prices[t]*np.exp(signal+noise)
    panel=pd.DataFrame(factors.reshape(T*N,6),columns=[f"x{i}" for i in range(6)])
    panel["date"]=np.repeat(dates,N); panel["stock"]=np.tile([f"S{i:03}" for i in range(N)],T)
    panel["tpos"]=np.repeat(np.arange(T),N); panel["open"]=prices.ravel()
    panel["signal_time"]=panel.date+pd.Timedelta(hours=15)
    panel["feature_available_time"]=panel.signal_time
    labels=panel.groupby("stock").open
    panel["y"]=labels.shift(-(k+1))/labels.shift(-1)-1
    exit_dates=pd.Series(dates).shift(-(k+1)).to_numpy()[panel.tpos]
    panel["label_available_time"]=exit_dates+pd.Timedelta(hours=9,minutes=30)
    return panel

def validate_panel(frame):
    if frame.duplicated(["date","stock"]).any():
        raise ValueError("Duplicate (date,stock) keys")
    if frame[["signal_time","feature_available_time"]].isna().any().any():
        raise ValueError("Missing availability timestamps")
    if (frame.feature_available_time>frame.signal_time).any():
        raise ValueError("Feature was unavailable at decision time")
    if (frame.y.notna() & frame.label_available_time.isna()).any():
        raise ValueError("Known label has unknown availability time")

def time_fold(frame, start, end, train_start=0):
    """Train at prior signal day's 15:00; dates [start,end) are held out."""
    validate_panel(frame)
    dates=frame.date.drop_duplicates().sort_values().reset_index(drop=True)
    if not 1<=start<end<=len(dates) or not 0<=train_start<start:
        raise ValueError("Invalid date fold")
    cutoff=dates.iloc[start-1]+pd.Timedelta(hours=15)
    train=frame[(frame.date>=dates.iloc[train_start]) & (frame.date<dates.iloc[start])
                & frame.y.notna() & (frame.label_available_time<=cutoff)].copy()
    valid=frame[frame.date.isin(dates.iloc[start:end]) & frame.y.notna()].copy()
    if train.empty or valid.empty: raise ValueError("Empty fold")
    return train,valid,cutoff

def keyed_metrics(frame, predictions):
    """predictions is a (date,stock,pred) table; exact keys required, order free."""
    if frame.duplicated(["date","stock"]).any() or predictions.duplicated(["date","stock"]).any():
        raise ValueError("Duplicate prediction/evaluation keys")
    aligned=frame[["date","stock","y"]].merge(predictions[["date","stock","pred"]],
        on=["date","stock"],how="outer",validate="1:1",indicator=True)
    if not aligned._merge.eq("both").all(): raise ValueError("Prediction keys do not match")
    daily=[]
    for date,group in aligned.groupby("date"):
        finite=np.isfinite(group.y)&np.isfinite(group.pred)
        g=group.loc[finite]; ic=rank_ic=spread=np.nan
        if len(g)>=3 and g.y.nunique()>1 and g.pred.nunique()>1:
            ic=g.y.corr(g.pred); rank_ic=g.y.rank().corr(g.pred.rank())
            ordered=g.sort_values(["pred","stock"]); size=max(1,len(g)//5)
            spread=ordered.tail(size).y.mean()-ordered.head(size).y.mean()
        daily.append({"date":date,"n":len(g),"ic":ic,"rank_ic":rank_ic,"spread":spread})
    table=pd.DataFrame(daily)
    valid=np.isfinite(aligned.y)&np.isfinite(aligned.pred)
    mse=float(np.mean((aligned.loc[valid,"y"]-aligned.loc[valid,"pred"])**2)) if valid.any() else np.nan
    return {"mse":mse,"ic":float(table.ic.mean()),"rank_ic":float(table.rank_ic.mean()),
            "coverage":float(table.ic.notna().mean()),"rows":int(valid.sum()),"dates":len(table)},table
下载本节 .pypython validation.py依赖版本见首页，示例只使用合成数据。
查看本轮本地运行记录 · 2026-10-12
{
  "cutoff": "2020-01-22 15:00:00",
  "last_mature_signal": 8,
  "shuffled_prediction_mse": 0.0,
  "late_feature_rejected": true,
  "missing_prediction_key_rejected": true
}
RECALL / 闭卷自测

先回答，再看提示和答案

Q01

在 s=20、k=5、截止19收盘、退出次日才到达时，最新训练信号是多少？

一级提示

先加到 label_available，不是只改验证行号。

参考答案与条件

t+6+1≤19，因此 t≤12。须声明额外延迟是一个交易日。

Q02

后一个训练折包含前一个验证折，必然泄漏吗？

一级提示

先写出目标、信息集和所需条件。

参考答案与条件

不必然。后来标签成熟后可以合法使用，已保存的前一预测不会因此变成训练内预测。

Q03

测试看过后根据 SHAP 换特征，还能称最终测试吗？

一级提示

先写出目标、信息集和所需条件。

参考答案与条件

不能继续当作独立留出。可描述旧测试，但新的选择需要新的未用评价样本。

TRANSFER / 独立练习

把知识迁移到一个新任务

给训练起点、持有期、时点与延迟各做一个变式；列出实际保留的键。故意加入晚到特征、重复键和乱序预测，分别验证报错与正确对齐。

这里保存学习笔记，不在浏览器执行 Python。A/B/C 由你按独立完成程度填写。

参考与进一步阅读

在线文档可能比本地固定版本更新；API 行为以已验证代码和版本记录为准。

掌握状态与复习

关掉资料独立完成后记 A；提示后完成记 B；需要重学记 C。

记录尽量保存在当前浏览器，建议定期导出备份。