滚动前向验证

按你本会交易的方式检验

像普通机器学习那样把训练行和测试行随机划分，在时间序列上会出问题：训练集里有晚于测试日的日期，相邻日期又彼此相关，于是测试在一定程度上是训练的副本。滚动前向验证尊重时间：用过去的一段训练，在紧随其后的一段上测试，然后窗口前移并重复。每个预测都来自从未见过未来的模型，这恰好就是实盘交易的情形。

import numpy as np

n, train_size, test_size = 10, 4, 2
start = 0
while start + train_size + test_size <= n:
    train = np.arange(start, start + train_size)
    test = np.arange(start + train_size, start + train_size + test_size)
    print("train", train.tolist(), "test", test.tolist())
    start += test_size                                   # slide forward by one test block

滚动还是扩展，以及禁运期

有两种形式。滚动窗口保持训练长度固定，丢掉最老的行：能适应市场状态变化，成本恒定。扩展窗口总是从第 0 行开始，训练集越来越大：用到每个观测，但旧数据可能已过时。第三个要素是间隔（禁运期）：如果某个特征或目标向前或向后看 h 天，那么紧挨测试期之前的行就与它共享信息，所以训练与测试之间要跳过几行。

def walk_forward_splits(n, train_size, test_size, gap=0):
    splits, start = [], 0
    while start + train_size + gap + test_size <= n:
        train = list(range(start, start + train_size))
        test = list(range(start + train_size + gap, start + train_size + gap + test_size))
        splits.append((train, test))
        start += test_size
    return splits

for train, test in walk_forward_splits(12, 5, 2, gap=1):
    print(train, "->", test)                         # the row between them is skipped

给每一折打分

对每一折，用训练行拟合、在测试行上打分，于是得到每折一个误差，而不是单一数字：它们的离散程度说明模型随时间的稳定性。一个有用的基准是平凡预测，这里是对每个测试日都用训练均值。逐折打印误差：平均表现好但在某个阶段糟透的模型，与始终尚可的模型并不相同。

import numpy as np

def walk_forward_splits(n, train_size, test_size):
    out, start = [], 0
    while start + train_size + test_size <= n:
        out.append((np.arange(start, start + train_size), np.arange(start + train_size, start + train_size + test_size)))
        start += test_size
    return out

rng = np.random.default_rng(2)
returns = rng.normal(0.0004, 0.01, size=300)
errors = []
for train, test in walk_forward_splits(len(returns), 100, 50):
    errors.append(np.mean((returns[test] - returns[train].mean()) ** 2))
print(np.round(np.array(errors) * 1e4, 3))                      # squared error in basis points squared, fold by fold

发现泄露

只有当每个训练行都严格早于每个测试行时，划分才是干净的。检查 train.max() < test.min() 只要一行，就能抓住最常见的错误：未排序或被打乱的索引、错位一位的切片、取自默认会打乱数据的库函数的划分。把这样的检查放进构建各折的代码里，让泄露直接中止运行，而不是产出好看的结果。

import numpy as np

def leaks_future(train_idx, test_idx):
    return bool(np.max(train_idx) >= np.min(test_idx))

train = np.arange(0, 100)
print(leaks_future(train, np.arange(100, 120)))                     # False: clean
shuffled = np.random.default_rng(1).permutation(120)                # what a shuffling splitter does
print(leaks_future(shuffled[:100], shuffled[100:]))                 # True: training rows after test rows

• 滚动前向：用过去一段训练、在下一段测试、窗口前移；每个测试行都在其全部训练行之后。
• 滚动窗口保持训练规模固定；扩展窗口从第 0 行开始；当特征跨越多日时，训练与测试之间留间隔（禁运期）。
• 分别给每折打分并看离散程度；检查 train.max() < test.min()；绝不在你要报告的折上调参。

下一节：限价订单簿，价格真正形成的数据结构。

实验：评分内容

• 用 walk_forward_splits 让窗口沿时间滑动25 分
• 用 expanding_splits 保留全部历史25 分
• 用 forecast_mse 逐折检验预测25 分
• 用 leaks_future 发现泄露25 分