2. 盘口快照字段清洗与异常值处理

为什么跳水场景对脏数据格外敏感

盘中快速跳水时，盘口会在几百毫秒内连续变化：买盘档位被迅速打掉、卖盘价格下压、部分档位在快照中缺失。如果原始快照里混有缺失档位、价格倒挂（买一价 ≥ 卖一价）、量值为负这类异常，回放时很容易被误读成"买盘瞬间消失"或"卖压突然放大"，从而把脏数据当成真实跳水信号。因此清洗的目标不是"修得好看"，而是让异常显式化、可追溯，且不改变真实行情的时序与量价关系。

清洗规则与标记策略

对每条盘口快照，按以下顺序逐条校验。核心原则：能安全修复的修复并打标，不能安全修复的置空并打标，绝不静默丢弃整条快照（否则会破坏回放时间轴）。

• 缺失档位：某档价格或量为空/NaN。若仅个别档缺失，保留其余档，缺失档置为 None 并标记 MISSING_LEVEL；若买一或卖一缺失，整条快照标记 CRITICAL_MISSING，回放时该快照不参与价差计算。
• 价格倒挂：买一价 ≥ 卖一价。不修改价格（可能是真实瞬时撮合），但标记 CROSSED_BOOK，回放统计中单独计数，避免被当作正常价差。
• 量值为负：委托量不可能为负。视为采集/传输错误，将该档量置 0 并标记 NEG_VOLUME；若同一快照负量档位超过 2 个，整条标记 SUSPECT_SNAPSHOT。
• 档位价格非单调：买盘价格应随档位递减、卖盘递增。违反则标记 NON_MONOTONIC，保留原值供人工复核。

可运行的清洗实现

下面用 Python 实现上述规则。快照用字典表示，bids/asks 为 [(price, volume), ...] 列表，按档位从优到劣排列。flags 用整数位标志累加。

from dataclasses import dataclass, field

# 位标志定义
MISSING_LEVEL     = 1 << 0
CRITICAL_MISSING  = 1 << 1
CROSSED_BOOK      = 1 << 2
NEG_VOLUME        = 1 << 3
SUSPECT_SNAPSHOT  = 1 << 4
NON_MONOTONIC     = 1 << 5

FLAG_NAMES = {
    MISSING_LEVEL: "MISSING_LEVEL",
    CRITICAL_MISSING: "CRITICAL_MISSING",
    CROSSED_BOOK: "CROSSED_BOOK",
    NEG_VOLUME: "NEG_VOLUME",
    SUSPECT_SNAPSHOT: "SUSPECT_SNAPSHOT",
    NON_MONOTONIC: "NON_MONOTONIC",
}

@dataclass
class CleanSnapshot:
    ts: int
    bids: list
    asks: list
    flags: int = 0
    def add(self, f):
        self.flags |= f
    def flag_names(self):
        return [n for b, n in FLAG_NAMES.items() if self.flags & b]

def _is_bad(x):
    return x is None or (isinstance(x, float) and x != x)  # None 或 NaN

def clean_snapshot(raw):
    ts = raw["ts"]
    bids = [list(lv) for lv in raw.get("bids", [])]
    asks = [list(lv) for lv in raw.get("asks", [])]
    snap = CleanSnapshot(ts=ts, bids=bids, asks=asks)

    # ① 缺失档位
    for side in (bids, asks):
        for lv in side:
            if _is_bad(lv[0]) or _is_bad(lv[1]):
                lv[0] = None if _is_bad(lv[0]) else lv[0]
                lv[1] = None if _is_bad(lv[1]) else lv[1]
                snap.add(MISSING_LEVEL)
    if not bids or bids[0][0] is None or not asks or asks[0][0] is None:
        snap.add(CRITICAL_MISSING)

    # ② 价格倒挂
    if bids and asks and bids[0][0] is not None and asks[0][0] is not None:
        if bids[0][0] >= asks[0][0]:
            snap.add(CROSSED_BOOK)

    # ③ 量值为负
    neg_cnt = 0
    for side in (bids, asks):
        for lv in side:
            if lv[1] is not None and lv[1] < 0:
                lv[1] = 0
                neg_cnt += 1
    if neg_cnt:
        snap.add(NEG_VOLUME)
        if neg_cnt > 2:
            snap.add(SUSPECT_SNAPSHOT)

    # ④ 档位单调性（仅对有效价格检查）
    def monotonic(side, descending):
        prices = [lv[0] for lv in side if lv[0] is not None]
        for a, b in zip(prices, prices[1:]):
            if (descending and a < b) or (not descending and a > b):
                return False
        return True
    if not monotonic(bids, True) or not monotonic(asks, False):
        snap.add(NON_MONOTONIC)

    return snap

用构造的异常样本验证

下面三条快照分别覆盖不同异常组合，属于教学构造数据，用于验证标记是否正确，不代表真实行情。

samples = [
    # 买一缺失 + 卖一正常
    {"ts": 1000, "bids": [[None, 300], [99.98, 200]],
     "asks": [[100.02, 150], [100.05, 400]]},
    # 价格倒挂 + 一档负量
    {"ts": 1001, "bids": [[100.10, 500], [100.08, 300]],
     "asks": [[100.05, 200], [100.09, 100]]},
    # 三档负量 + 买盘价格非单调
    {"ts": 1002, "bids": [[100.00, -10], [100.02, -5], [99.99, -1]],
     "asks": [[100.03, 120], [100.06, 80]]},
]

for s in samples:
    c = clean_snapshot(s)
    print(c.ts, c.flag_names(), "bids=", c.bids, "asks=", c.asks)

预期输出（由上述规则直接推出）：

• 1000 ['MISSING_LEVEL', 'CRITICAL_MISSING']，买一价被置 None，其余档保留。
• 1001 ['CROSSED_BOOK', 'NEG_VOLUME']，价格未改动，负量档置 0。
• 1002 ['NEG_VOLUME', 'SUSPECT_SNAPSHOT', 'NON_MONOTONIC']，三档负量触发可疑快照，买盘 100.00→100.02 违反递减。

回放阶段的判读约定