2. 盘口数据缺失与时间戳错乱的清洗验证

脏数据的三种典型形态与判定口径

盘口快照进入回测引擎前，必须先通过一道“数据体检”。本例聚焦三类高频脏数据：快照间隔不均（相邻快照时间差偏离正常采样周期）、时间戳回退（后一条快照时间早于前一条）、买卖档缺失（买一/卖一价格或数量为空、为0、或档位数量不足）。判定口径要写死成可执行规则，否则清洗结果无法复现。

下面这张图给出清洗与校验的整体数据流：原始快照先做时间轴校验，再分叉处理回退与间隔异常，最后做盘口完整性校验，输出可信快照集与清洗日志。

清洗规则的可执行实现

把上面的判定口径落成代码，核心是三个函数：时间轴校验、间隔填充、盘口完整性过滤。下面给出可直接运行的 Python 实现，输入为含 ts（毫秒时间戳）、bid_px_1..5、ask_px_1..5、bid_sz_1..5、ask_sz_1..5 的 DataFrame。

import pandas as pd
import numpy as np

T = 500          # 正常采样周期(ms)
GAP_MULT = 3     # 间隔异常倍数
MIN_LEVELS = 5   # 最小有效档位数

def validate_timeline(df):
    """标记时间戳回退与间隔异常"""
    df = df.sort_values("ts").reset_index(drop=True)
    df["dt"] = df["ts"].diff()
    df["bad_ts"] = df["dt"] < 0
    df["gap"] = df["dt"] > GAP_MULT * T
    return df

def fill_gaps(df):
    """对间隔异常做前向填充，并记录填充标记"""
    df["filled"] = False
    gap_idx = df.index[df["gap"]]
    for i in gap_idx:
        if i == 0:
            continue
        df.loc[i, df.columns.difference(["ts","dt","bad_ts","gap","filled"])] = \
            df.loc[i-1, df.columns.difference(["ts","dt","bad_ts","gap","filled"])]
        df.loc[i, "filled"] = True
    return df

def check_book(df):
    """盘口完整性：买一卖一非空、价格量>0、档位数≥N"""
    ok = pd.Series(True, index=df.index)
    for side in ["bid","ask"]:
        for lv in range(1, MIN_LEVELS+1):
            px = f"{side}_px_{lv}"; sz = f"{side}_sz_{lv}"
            ok &= df[px].notna() & (df[px] > 0) & df[sz].notna() & (df[sz] > 0)
    df["book_ok"] = ok
    return df

def clean(df):
    df = validate_timeline(df)
    df = fill_gaps(df)
    df = check_book(df)
    clean_df = df[df["book_ok"] & ~df["bad_ts"]].copy()
    log = {
        "total": len(df),
        "bad_ts": int(df["bad_ts"].sum()),
        "gap": int(df["gap"].sum()),
        "book_bad": int((~df["book_ok"]).sum()),
        "kept": len(clean_df),
    }
    return clean_df, log

清洗前后对策略信号与成交模拟的影响验证

清洗是否“有效”，不能只看剔除了多少行，而要看策略信号与成交模拟是否发生实质变化。验证方法是：用同一策略分别跑原始数据与清洗后数据，对比信号触发次数、成交笔数、成交均价偏差。下面给出一个最小可复现的对比脚本，策略逻辑为“买一量突增 2 倍即触发买入信号”。

def signal_on_bid_surge(df, mult=2.0):
    """买一量较前值放大 mult 倍触发信号"""
    prev = df["bid_sz_1"].shift(1)
    return (df["bid_sz_1"] > mult * prev).fillna(False)

def simulate(df, signal_col):
    """简化成交模拟：信号出现时以卖一价成交"""
    trades = df[df[signal_col]]
    if trades.empty:
        return {"n_trades": 0, "avg_px": np.nan}
    return {"n_trades": len(trades), "avg_px": trades["ask_px_1"].mean()}

raw = pd.read_csv("snapshots_raw.csv")
clean_df, log = clean(raw)

raw["sig"] = signal_on_bid_surge(raw)
clean_df["sig"] = signal_on_bid_surge(clean_df)

r_raw = simulate(raw, "sig")
r_cln = simulate(clean_df, "sig")

print("清洗日志:", log)
print("原始数据成交:", r_raw)
print("清洗后成交:", r_cln)
print("成交均价偏差:", abs(r_raw["avg_px"] - r_cln["avg_px"]))

清洗日志与剔除清单（可直接交付）

每次清洗必须产出可审计的日志，否则压力测试输入无法追溯。下表是清洗日志的推荐字段与示例值，可直接作为交付物模板。

结果判读与通过条件

• 保留率：清洗后保留行数 / 原始行数 ≥ 99%，低于此值说明脏数据比例过高，需先排查采集链路。
• 信号一致性：清洗前后信号触发次数差异 ≤ 5%，否则检查填充是否制造了虚假的买一量突增。
• 成交均价偏差：≤ 1 个最小变动价位，超出则说明缺失档位被错误填充。
• 可追溯性：每条被剔除或填充的行都必须能在日志中定位到原始行号与原因。

满足以上四条，清洗后的快照集才可作为后续滑点扫描、费率压力测试的可信输入。