1. 盘口快照数据结构与字段校验

为什么先做字段校验

逐笔盘口快照是后续所有事件识别（价差异常、量比突变、撤单潮等）的原料。原始快照常见四类脏数据：字段缺失（如某档 bid_px 为空）、时间戳乱序或重复、价量为负、价格档位不合理（买价高于卖价、档位不递增、价格偏离涨跌停区间）。若不先清洗，后续阈值统计会被污染。本例对单只股票的快照序列做四类校验，输出结构化异常清单。

快照数据结构与校验规则

约定一条快照记录包含：ts（毫秒时间戳）、bid_px[1..5]/bid_vol[1..5]、ask_px[1..5]/ask_vol[1..5]、last_px、limit_up、limit_down。校验规则如下：

• 完整性：必需字段非空，五档价量齐全。
• 时间戳单调性：ts 严格递增，重复或回退均记异常。
• 非负性：所有价格、成交量 ≥ 0。
• 档位合理性：买价从一档到五档递减，卖价递增；买一 ≤ 卖一（允许相等表示无成交挂单）；价格落在 [limit_down, limit_up] 内。

实现：单条快照校验函数

用 Python 实现，返回该条记录的异常列表；空列表表示通过。价格档位用 bid_px 递减、ask_px 递增判断，并检查买一卖一关系与涨跌停区间。

from dataclasses import dataclass, field

REQUIRED = ["ts", "last_px", "limit_up", "limit_down"]
for i in range(1, 6):
    REQUIRED += [f"bid_px{i}", f"bid_vol{i}", f"ask_px{i}", f"ask_vol{i}"]

def check_snapshot(rec, prev_ts=None):
    """返回异常列表，每项为 (字段, 规则, 原因)。"""
    errs = []

    # ① 完整性
    for f in REQUIRED:
        if f not in rec or rec[f] is None:
            errs.append((f, "完整性", "字段缺失"))

    # ② 时间戳单调性
    ts = rec.get("ts")
    if ts is not None and prev_ts is not None and ts <= prev_ts:
        errs.append(("ts", "单调性", f"ts={ts} 未大于前值 {prev_ts}"))

    # ③ 非负性
    for f in REQUIRED:
        if f == "ts":
            continue
        v = rec.get(f)
        if v is not None and v < 0:
            errs.append((f, "非负性", f"负值 {v}"))

    # ④ 档位合理性
    bid_px = [rec.get(f"bid_px{i}") for i in range(1, 6)]
    ask_px = [rec.get(f"ask_px{i}") for i in range(1, 6)]
    if all(v is not None for v in bid_px):
        for i in range(4):
            if bid_px[i] < bid_px[i + 1]:
                errs.append((f"bid_px{i+1}", "档位", "买价未递减"))
    if all(v is not None for v in ask_px):
        for i in range(4):
            if ask_px[i] > ask_px[i + 1]:
                errs.append((f"ask_px{i+1}", "档位", "卖价未递增"))
    if bid_px[0] is not None and ask_px[0] is not None and bid_px[0] > ask_px[0]:
        errs.append(("bid_px1", "档位", "买一高于卖一"))
    lo, hi = rec.get("limit_down"), rec.get("limit_up")
    if lo is not None and hi is not None:
        for f in [f"bid_px{i}" for i in range(1, 6)] + [f"ask_px{i}" for i in range(1, 6)]:
            v = rec.get(f)
            if v is not None and not (lo <= v <= hi):
                errs.append((f, "档位", f"{v} 超出[{lo},{hi}]"))
    return errs

批量校验与异常清单输出

按时间顺序遍历快照，维护 prev_ts 实现单调性检查，把每条异常连同行号写入清单。

def validate_series(snapshots):
    report = []
    prev_ts = None
    for idx, rec in enumerate(snapshots):
        errs = check_snapshot(rec, prev_ts)
        for f, rule, reason in errs:
            report.append({"row": idx, "ts": rec.get("ts"),
                           "field": f, "rule": rule, "reason": reason})
        if rec.get("ts") is not None:
            prev_ts = rec["ts"]
    return report

# 示例数据（教学构造）
snaps = [
    {"ts": 93000120, "last_px": 10.20, "limit_up": 11.22, "limit_down": 9.18,
     "bid_px1": 10.20, "bid_vol1": 500, "bid_px2": 10.19, "bid_vol2": 300,
     "bid_px3": 10.18, "bid_vol3": 200, "bid_px4": 10.17, "bid_vol4": 100,
     "bid_px5": 10.16, "bid_vol5": 80,
     "ask_px1": 10.21, "ask_vol1": 400, "ask_px2": 10.22, "ask_vol2": 350,
     "ask_px3": 10.23, "ask_vol3": 250, "ask_px4": 10.24, "ask_vol4": 150,
     "ask_px5": 10.25, "ask_vol5": 90},
    {"ts": 93000120, "last_px": 10.20, "limit_up": 11.22, "limit_down": 9.18,
     "bid_px1": 10.20, "bid_vol1": 500, "bid_px2": 10.19, "bid_vol2": 300,
     "bid_px3": 10.18, "bid_vol3": 200, "bid_px4": 10.17, "bid_vol4": 100,
     "bid_px5": 10.16, "bid_vol5": 80,
     "ask_px1": 10.21, "ask_vol1": 400, "ask_px2": 10.22, "ask_vol2": 350,
     "ask_px3": 10.23, "ask_vol3": 250, "ask_px4": 10.24, "ask_vol4": 150,
     "ask_px5": 10.25, "ask_vol5": 90},
    {"ts": 93000420, "last_px": 10.20, "limit_up": 11.22, "limit_down": 9.18,
     "bid_px1": -0.01, "bid_vol1": 500, "bid_px2": 10.19, "bid_vol2": 300,
     "bid_px3": 10.18, "bid_vol3": 200, "bid_px4": 10.17, "bid_vol4": 100,
     "bid_px5": 10.16, "bid_vol5": 80,
     "ask_px1": 10.22, "ask_vol1": 400, "ask_px2": 10.22, "ask_vol2": 350,
     "ask_px3": 10.23, "ask_vol3": 250, "ask_px4": 10.24, "ask_vol4": 150,
     "ask_px5": 10.25, "ask_vol5": 90},
    {"ts": 93000720, "last_px": 10.20, "limit_up": 11.22, "limit_down": 9.18,
     "bid_px1": 10.25, "bid_vol1": 500, "bid_px2": 10.19, "bid_vol2": 300,
     "bid_px3": 10.18, "bid_vol3": 200, "bid_px4": 10.17, "bid_vol4": 100,
     "bid_px5": 10.16, "bid_vol5": 80,
     "ask_px1": 10.24, "ask_vol1": 400, "ask_px2": 10.22, "ask_vol2": 350,
     "ask_px3": 10.23, "ask_vol3": 250, "ask_px4": 10.24, "ask_vol4": 150,
     "ask_px5": 10.25, "ask_vol5": 90},
]

for r in validate_series(snaps):
    print(r)

结果判读与通过条件

对上述示例数据，预期输出（教学构造，非实测）：

通过条件：清单为空即该序列可直接进入下游；非空时按 row 隔离异常记录，其余记录保留。注意第 2 行 bid_px1=-0.01 会同时触发非负性与区间两条规则，这是预期行为——同一字段可命中多条规则，便于定位根因。校验器只做判定不做修复，修复策略（丢弃、前值填充、标记）由下游按业务决定。