3. 逐笔成交与盘口快照的合并回放流

为什么需要“合并回放流”

逐笔成交（Trade）和盘口快照（Snapshot）是两条独立的数据流：成交流记录“谁在什么价格成交了多少”，快照流记录“某一时刻买卖各档的挂单状态”。跳水回放要还原的是“盘口变化 → 成交触发 → 盘口再变化”的因果链，因此必须把两条流按时间戳合并成一条事件驱动的回放流，逐条推进。合并的核心难点不在排序本身，而在同一时间戳下成交与快照的先后顺序——顺序错了，回放中就会出现“先看到价格跌了、再看到卖单堆积”的倒因为果。

合并规则：同一时间戳的定序

交易所行情中，快照通常是按固定周期（如每 3 秒）切片生成的，其时间戳标记的是切片时刻；而逐笔成交的时间戳是成交撮合时刻。两者落在同一毫秒时，业务含义是：该快照反映的是切片时刻之前已发生的成交结果。因此合并时采用如下定序约定（本案例统一使用，正文与代码保持一致）：

• 同一时间戳：成交事件排在快照之前（Trade 先于 Snapshot），保证快照中的累计成交量、最新价已包含该笔成交。
• 不同时间戳：严格按时间升序；时间戳相同且同为成交时，按原始序列号（seq）稳定排序，避免同毫秒多笔成交被打乱。
• 快照与快照同时间戳属异常数据，保留首次出现、标记重复。

合并实现：稳定排序 + 定序键

最稳妥的做法不是“先合并再排序”，而是给每条事件构造一个可比较的定序键，再对合并后的列表做一次稳定排序。定序键由三部分组成：时间戳、类型优先级、原始序号。类型优先级让同一时间戳下 Trade 排在 Snapshot 之前；原始序号保证同类型同时间戳的成交保持原始撮合顺序。

from dataclasses import dataclass, field
from typing import List, Literal

@dataclass
class Event:
    ts: int                      # 统一为毫秒整数，便于比较
    kind: Literal["trade", "snap"]
    seq: int                     # 成交用原始序号；快照用切片序号
    payload: dict = field(default_factory=dict)

# 定序键：时间戳升序 → 同 ts 下 trade(0) 先于 snap(1) → 同类型按 seq
def order_key(e: Event):
    kind_rank = 0 if e.kind == "trade" else 1
    return (e.ts, kind_rank, e.seq)

def merge_streams(trades: List[Event], snaps: List[Event]) -> List[Event]:
    merged = trades + snaps
    # Python 的 sorted 是稳定排序，但这里键已含 seq，稳定性作为兜底
    merged.sort(key=order_key)
    # 去除同 ts 重复快照（保留首条）
    out, seen_snap_ts = [], set()
    for e in merged:
        if e.kind == "snap":
            if e.ts in seen_snap_ts:
                continue
            seen_snap_ts.add(e.ts)
        out.append(e)
    return out

注意 ts 统一为毫秒整数是合并的前提——若成交用微秒、快照用秒，直接比较会得到错误顺序。这一步应在数据接入阶段完成（时间戳统一），本案例只消费已对齐的整数时间戳。

逐条推进：事件驱动的回放指针

合并完成后，回放器持有一个指针 idx，每次 step() 取出当前事件并更新状态。成交事件更新“最新价、累计成交量、最近成交方向”；快照事件整体替换盘口十档。由于定序键已保证 Trade 先于同 ts 的 Snap，快照替换时累计量已包含该笔成交，不会出现“快照累计量落后于成交”的矛盾。

class ReplayEngine:
    def __init__(self, events: List[Event]):
        self.events = events
        self.idx = 0
        self.last_price = None
        self.cum_vol = 0
        self.orderbook = None

    def step(self) -> Event | None:
        if self.idx >= len(self.events):
            return None
        e = self.events[self.idx]
        self.idx += 1
        if e.kind == "trade":
            self.last_price = e.payload["price"]
            self.cum_vol += e.payload["vol"]
        else:
            self.orderbook = e.payload["levels"]
        return e

    def run_to(self, ts_ms: int):
        # 推进到指定时间戳（含），用于慢放/定位
        while self.idx < len(self.events) and self.events[self.idx].ts <= ts_ms:
            self.step()

验证：同时间戳顺序是否真的生效

用一组构造数据验证定序键。构造 2 笔成交与 1 条快照落在同一毫秒 ts=3000，期望输出顺序为 Trade(seq=4) → Trade(seq=5) → Snap。同时构造一条重复快照验证去重。

trades = [
    Event(120, "trade", 1, {"price": 10.02, "vol": 300}),
    Event(120, "trade", 2, {"price": 10.01, "vol": 500}),
    Event(450, "trade", 3, {"price": 10.00, "vol": 800}),
    Event(3000, "trade", 4, {"price": 9.98, "vol": 1200}),
    Event(3000, "trade", 5, {"price": 9.97, "vol": 900}),
]
snaps = [
    Event(0,    "snap", 0, {"levels": "初始"}),
    Event(3000, "snap", 1, {"levels": "卖压"}),
    Event(3000, "snap", 1, {"levels": "卖压-重复"}),
    Event(6000, "snap", 2, {"levels": "跳水"}),
]

merged = merge_streams(trades, snaps)
for e in merged:
    print(e.ts, e.kind, e.seq)

# 预期输出（ts, kind, seq）：
# 0    snap  0
# 120  trade 1
# 120  trade 2
# 450  trade 3
# 3000 trade 4
# 3000 trade 5
# 3000 snap  1
# 6000 snap  2

回放中值得观察的判读点

合并流跑通后，回放时重点观察“成交与快照的对应关系”是否自洽：

• 累计量一致性：快照中的累计成交量应 ≥ 此前所有成交的累计量；若小于，说明定序或数据源有问题。
• 最新价一致性：快照的最新价应等于此前最后一笔成交价（或在其附近），若快照最新价明显偏离，检查是否把快照排到了成交之前。
• 卖压与成交的先后：跳水段中，若先出现卖一量骤增的快照、再出现大额卖向成交，属正常“卖压先行”；若顺序相反，需回查定序键。