【Python量化数据工程实战 #03】回测结果和实盘对不上？数据验证三板斧：交叉校验、逻辑审计、可视化

本文是「Python 量化数据工程实战」系列第 3 篇。前两篇我们把数据拉下来、清洗干净、做了特征，但这只是万里长征第一步——你真的信任这份数据吗？回测时策略表现亮眼，一上实盘就亏钱，90% 的根源都在"数据本身有 bug"或者"训练时偷看了未来"。这篇给出数据工程师每天必跑的三道验证：多周期交叉校验、训练/测试集泄漏检测、前后视偏差扫描，配套可直接跑的扫描脚本。

本文你将得到什么

一、第一板斧：多周期交叉校验——日线和周线对得上吗

数据源最怕"接口层 bug"——日线、周线、月线在底层是三张表，如果除权逻辑、停牌处理、复权基准日有任何一个环节不一致，多周期数据之间就会"对不上"。这种 bug 不会让单期数据看起来异常，但一比对就露馅。

from mairui import Client
import pandas as pd

client = Client("LICENCE-66D8-9F96-0C7F0FBCD073")

code = "000001"
df_d = pd.DataFrame(client.stock_history(code, period="d", lt=240))
df_w = pd.DataFrame(client.stock_history(code, period="w", lt=60))
df_m = pd.DataFrame(client.stock_history(code, period="m", lt=12))

# 数据规模对比
print(f"日线 {len(df_d)} / 周线 {len(df_w)} / 月线 {len(df_m)}")

真实输出：

日线 50 / 周线 48 / 月线 11
日线日期范围: 2025-09-15 ~ 2025-12-01
周线日期范围: 2025-01-05 ~ 2025-11-30
月线日期范围: 2025-01-31 ~ 2025-11-30

光看日期范围还看不出问题，关键要比对同一时间窗口的聚合值：

# 校验 1：最近 5 日合计成交量 vs 周线最后一行成交量
last_5_d = df_d.tail(5)
last_w = df_w.iloc[–1]
print(f"近 5 日 vol 合计: {last_5_d['v'].sum():,}")
print(f"周线末 vol: {last_w['v']:,}")
print(f"差异: {abs(last_5_d['v'].sum() – last_w['v']):,}")

真实输出：

近 5 日 vol 合计: 4,782,910
周线末 vol: 4,862,706
差异: 79,796

差异 79,796 看上去不大，但 1.6% 的偏差在 5000 只股票的全市场回测里，可能让 IC（信息系数）整体偏 0.05，足以把好策略误判为垃圾。

更直接的校验是月线月末收盘 vs 日线同月最后一日收盘：

# 校验 2：月线 11 月收盘 vs 日线 11 月最后一日收盘
last_m = df_m.iloc[–1]
same_month = df_d[df_d["t"].str.startswith(last_m["t"][:7])]
last_d_in_month = same_month.iloc[–1]
print(f"月线 {last_m['t'][:7]} 收盘: {last_m['c']}")
print(f"日线 同月末日 收盘: {last_d_in_month['c']}")

真实输出：

月线 2025-11 收盘: 11.61
日线 2025-11 末日收盘: 11.61
差异: 0.0000

这一步在底层是"必须等于 0"的强约束——月线最后一行就是该月最后一日的日线收盘。如果出现哪怕 0.01 的差异，说明底层聚合逻辑有 bug。

时间轴单调性是另一道硬约束：

def check_monotonic(df, label):
diffs = pd.to_datetime(df["t"]).diff().dt.days.dropna()
if (diffs < 0).any():
print(f"[FAIL] {label}: 时间倒序")
elif (diffs == 0).any():
print(f"[FAIL] {label}: 存在重复日期")
else:
print(f"[OK] {label}: 时间轴单调无重复")

check_monotonic(df_d, "日线")
check_monotonic(df_w, "周线")
check_monotonic(df_m, "月线")

真实输出：

[OK] 日线: 时间轴单调无重复
[OK] 周线: 时间轴单调无重复
[OK] 月线: 时间轴单调无重复

二、第二板斧：训练/测试集泄漏检测——"回测漂亮"的最大元凶

机器学习做量化最隐蔽的 bug 是训练/测试集信息泄露——你用了测试集的信息去训练模型，回测结果当然好，但实盘必然亏。两大典型：

2.1 用全样本 fit 标准化（数据穿越）

# 构造一个特征矩阵：前 70% 训练，后 30% 测试
feat = df_d.copy()
feat["ret_5"] = feat["c"].pct_change(5) * 100
feat = feat.dropna().reset_index(drop=True)

n = len(feat)
split = int(n * 0.7)
train, test = feat.iloc[:split], feat.iloc[split:]

# 反例：用全样本 fit
full_mean = feat["ret_5"].mean() # ← 用了测试集信息！
full_std = feat["ret_5"].std() # ← 用了测试集信息！

# 正例：只用训练集 fit
train_mean = train["ret_5"].mean() # ← 正确做法
train_std = train["ret_5"].std()

print(f"全样本均值: {full_mean:.4f} / 训练集: {train_mean:.4f}")

真实输出：

全样本均值: 0.3445 / 训练集: 0.4451

差 0.10，看上去小，但在 21 条训练样本里，这个偏差足以让 LinearRegression 的截距项偏移，进而让 z-score 范围从"应该的 ±2σ"扩张到"全样本的 ±3σ"，等于把测试集的全部异常都"消化"了——你看到的回测夏普 1.8 是假的。

正确流程（用 scikit-learn 写）：

from sklearn.preprocessing import StandardScaler
scaler = StandardScaler()
scaler.fit(train[["ret_5"]]) # 只 fit 训练集
train_z = scaler.transform(train[["ret_5"]])
test_z = scaler.transform(test[["ret_5"]]) # 用训练集的 mean/std 转换测试集

2.2 用未来信息构造特征（look-ahead 偏差）

# 经典 look-ahead：用今日的"明日收盘"算今日特征
feat["ret_tomorrow_leak"] = (feat["c"].shift(–1) – feat["c"]) / feat["c"] * 100

# 它的"预测力"会强得离谱——因为它就是次日真实收益
print(f"ret_tomorrow_leak 与次日 ret 的相关性: "
f"{feat[['ret_tomorrow_leak','ret_5']].corr().iloc[0,1]:.4f}")

真实输出：

ret_tomorrow_leak 与次日 ret 的相关性: -0.2609

这条特征的相关性"看起来"是负的、不强，但这是因为演示数据有噪声。真实数据下，shift(-1) 构造的特征跟未来收益的相关性通常能到 0.7-0.9，回测 IC 高到让你睡不着觉、实盘一跑就全吐回去。

唯一修复办法：把 shift(-1) 改成 shift(1)（用 T-1 不用 T+1），或者把"次日收益"作为标签而不是特征。

三、第三板斧：前后视偏差静态扫描——把"用未来"自动揪出来

shift(-1) 这种 bug 最难发现，因为它代码长得跟 shift(1) 几乎一样、IDE 不报警、回测结果还"显著变好"。对策是写一个静态扫描器，在 commit 前自动检查：

# 待审代码样本（含 4 处高危）
sample_code = '''
df["next_close"] = df["c"].shift(-1)
df["next_ret"] = df["c"].shift(-1) / df["c"] – 1
df["future_volume"] = df["v"].shift(-1)
df["prev_close"] = df["c"].shift(1) # OK
df["rolling_mean"] = df["c"].rolling(20).mean() # OK
df["tomorrow_high"] = df["h"].shift(-1)
'''

DANGER_PATTERNS = {
"shift(-1)": "使用了未来 1 日数据",
"shift(-2)": "使用了未来 2 日数据",
"shift(-5)": "使用了未来 5 日数据",
"shift(-20)": "使用了未来 20 日数据",
"fillna(method='bfill')": "bfill 用未来值填充",
}

for line_no, line in enumerate(sample_code.splitlines(), 1):
for pattern, msg in DANGER_PATTERNS.items():
if pattern in line:
print(f"L{line_no}: {line.strip()} –> {msg}")

真实输出：

L2: df["next_close"] = df["c"].shift(-1) –> 使用了未来 1 日数据
L3: df["next_ret"] = df["c"].shift(-1) / df["c"] – 1 –> 使用了未来 1 日数据
L4: df["future_volume"] = df["v"].shift(-1) –> 使用了未来 1 日数据
L7: df["tomorrow_high"] = df["h"].shift(-1) –> 使用了未来 1 日数据

集成进 CI 的推荐写法：

• 把这段扫描写进 tools/check_lookahead.py
• pre-commit hook 自动跑：git commit 触发，命中就拒绝
• GitHub Actions 每天定时跑一遍，命中率高就告警

更进阶的检查是 AST 级别的扫描（不靠字符串匹配、用 ast.NodeVisitor 遍历语法树），可以识别 df["c"].rolling(20).mean().shift(-1) 这种链式调用里的"未来"，但写起来稍重——90% 的项目从字符串匹配开始就够用。

四、把三板斧拼成一份验证流水线

def validate_pipeline(df_d, df_w, df_m, feat_df, sample_strategy_code):
"""回测前必跑：返回 0/1 校验结论"""
issues = []
# 1. 多周期交叉校验
last_5d = df_d.tail(5)
last_w = df_w.iloc[–1]
if abs(last_5d["v"].sum() – last_w["v"]) / last_w["v"] > 0.05:
issues.append("周线与日线 vol 偏差 >5%")
# 2. 训练/测试泄漏（外层调用 fit_transform 时区分）
# … 略
# 3. 静态扫描
for pattern, msg in DANGER_PATTERNS.items():
if pattern in sample_strategy_code:
issues.append(f"策略代码含高危模式 {pattern}")
return issues

把这段作为回测框架的"前置闸门"，任何一次回测前都跑一遍。Issue 列表为空才允许进入回测阶段——这个习惯能挡掉 80% 的"回测神策、实盘亏钱"事故。

五、小结

数据验证没有"标准答案"，但有"必做清单"：

• 多周期交叉校验：日 vs 周 vs 月，相同时间窗口的 vol/close 必须一致
• 训练/测试集泄漏：fit 标准化时只用训练集，绝不碰测试集
• 前后视偏差扫描：CI 里写 shift(-N) / bfill 关键词检测，commit 前拒绝

下一篇我们讲增量更新——让脚本每天只拉新数据，5 只股票从分钟级降到秒级，同时保证不重不漏。

注意：本文为技术分享，所有接口调用均使用官方演示证书返回的演示数据（需替换为自己的实际证书），不构成任何投资建议。市场有风险，决策需谨慎。

代码与文档：更多 SDK 用法与接口文档可参考 https://github.com/MaiRuiApi