1. 订单簿快照与增量消息的时序对齐校验

为什么快照与增量必须做序列号对齐

交易所的订单簿行情通常由两条独立通道组成：一条是低频的快照（Snapshot），携带某一时刻的完整盘口和该时刻的最后更新序列号 lastUpdateId；另一条是高频的增量消息（Diff/Update），每条携带首尾序列号 U（first）与 u（last）。重建订单簿时，必须保证增量消息的序列号区间与快照版本严格衔接，否则会出现漏单或重复应用，导致盘口深度错误。

对齐的核心判据只有一条：第一条被应用的增量消息必须满足 U ≤ lastUpdateId + 1 ≤ u。若某条增量的 u ≤ lastUpdateId，说明它已被快照覆盖，应丢弃；若某条增量的 U > lastUpdateId + 1，说明快照与增量之间存在缺口，快照版本过旧，必须重新拉取快照。校验器要做的就是在应用前逐条判定，并输出可诊断的异常信息。

快照 Snapshot lastUpdateId = 1000 完整盘口 + 版本号 增量消息流（按到达顺序） U=980 u=995 已被快照覆盖 U=996 u=1005 衔接成功 U=1010 u=1020 存在缺口 判定规则： u ≤ lastUpdateId → 丢弃（已覆盖）　|　U ≤ lastUpdateId+1 ≤ u → 应用（衔接）　|　U > lastUpdateId+1 → 缺口，重拉快照 诊断输出 丢弃: seq 980-995 应用: seq 996-1005 缺口: 1006-1009 缺失 正常路径 快照版本与增量首号衔接 订单簿连续更新，无缺口 异常路径 快照版本过旧 触发重拉快照

校验器的数据结构与核心判定

校验器需要维护三个状态：快照版本 lastUpdateId、当前已应用到订单簿的序列号 currentSeq、以及是否已完成首次对齐的布尔标志 aligned。首次对齐前，所有增量都按快照版本判定；对齐后，则按 currentSeq 判定连续性。下面给出一个可直接运行的 Python 实现，包含诊断信息输出。

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

@dataclass
class DiffMessage:
    U: int          # 首序列号
    u: int          # 尾序列号
    bids: list = field(default_factory=list)
    asks: list = field(default_factory=list)

@dataclass
class Diagnostic:
    level: str      # INFO / WARN / ERROR
    code: str       # COVERED / ALIGNED / GAP / OVERLAP
    detail: str

class SequenceAligner:
    def __init__(self, snapshot_last_update_id: int):
        self.last_update_id = snapshot_last_update_id
        self.current_seq = snapshot_last_update_id
        self.aligned = False
        self.diagnostics: List[Diagnostic] = []

    def check(self, msg: DiffMessage) -> Optional[Diagnostic]:
        # 1. 已被快照覆盖
        if msg.u <= self.last_update_id:
            d = Diagnostic("INFO", "COVERED",
                f"seq {msg.U}-{msg.u} 已被快照 {self.last_update_id} 覆盖，丢弃")
            self.diagnostics.append(d)
            return d

        # 2. 首次对齐判定
        if not self.aligned:
            if msg.U <= self.last_update_id + 1 <= msg.u:
                self.aligned = True
                self.current_seq = msg.u
                d = Diagnostic("INFO", "ALIGNED",
                    f"首条增量 seq {msg.U}-{msg.u} 与快照 {self.last_update_id} 衔接成功")
                self.diagnostics.append(d)
                return d
            else:
                d = Diagnostic("ERROR", "GAP",
                    f"快照 {self.last_update_id} 与增量首号 {msg.U} 之间存在缺口 "
                    f"{self.last_update_id + 1}-{msg.U - 1}，需重拉快照")
                self.diagnostics.append(d)
                return d

        # 3. 对齐后的连续性检查
        if msg.U <= self.current_seq + 1 <= msg.u:
            self.current_seq = msg.u
            d = Diagnostic("INFO", "ALIGNED",
                f"seq {msg.U}-{msg.u} 连续衔接，currentSeq 更新为 {self.current_seq}")
            self.diagnostics.append(d)
            return d
        elif msg.U > self.current_seq + 1:
            d = Diagnostic("ERROR", "GAP",
                f"seq {msg.U}-{msg.u} 与 currentSeq {self.current_seq} 之间存在缺口 "
                f"{self.current_seq + 1}-{msg.U - 1}")
            self.diagnostics.append(d)
            return d
        else:
            d = Diagnostic("WARN", "OVERLAP",
                f"seq {msg.U}-{msg.u} 与 currentSeq {self.current_seq} 重叠，需幂等过滤")
            self.diagnostics.append(d)
            return d

用示例数据验证判定结果

假设快照 lastUpdateId = 1000，依次送入三条增量消息，观察校验器的判定与诊断输出。示例数据为教学构造，结果由上述逻辑直接推出。

aligner = SequenceAligner(snapshot_last_update_id=1000)

msgs = [
    DiffMessage(U=980,  u=995),    # 已被快照覆盖
    DiffMessage(U=996,  u=1005),   # 与快照衔接
    DiffMessage(U=1010, u=1020),   # 存在缺口 1006-1009
]

for m in msgs:
    d = aligner.check(m)
    print(f"[{d.level}] {d.code}: {d.detail}")

print("aligned =", aligner.aligned)
print("currentSeq =", aligner.current_seq)

预期输出（由给定输入直接推出）：

[INFO] COVERED: seq 980-995 已被快照 1000 覆盖，丢弃
[INFO] ALIGNED: 首条增量 seq 996-1005 与快照 1000 衔接成功
[ERROR] GAP: seq 1010-1020 与 currentSeq 1005 之间存在缺口 1006-1009
aligned = True
currentSeq = 1005
判定顺序很重要：必须先判“已覆盖”，再判“首次对齐”，最后判“对齐后连续性”。若把覆盖判定放到对齐之后，首条 u ≤ lastUpdateId 的消息会被误判为缺口，导致不必要的快照重拉。

异常场景的诊断信息设计

诊断信息要能让运维或开发人员一眼定位问题类型与缺口范围。建议每条诊断至少包含：级别、错误码、涉及的序列号区间、当前基准序列号、以及建议动作。下表给出四类典型场景的判定与处置。

场景判定条件错误码建议动作
增量已被快照覆盖u ≤ lastUpdateIdCOVERED丢弃，继续下一条
首条增量与快照衔接U ≤ lastUpdateId+1 ≤ uALIGNED应用并置 aligned=True
快照版本过旧U > lastUpdateId+1GAP重拉快照，重置校验器
增量区间重叠u ≤ currentSeq 且未覆盖OVERLAP幂等过滤，仅应用新部分
重拉快照后必须用新的 lastUpdateId 重建校验器，并把 aligned 重置为 False。若沿用旧 currentSeq，后续增量会被误判为重叠或缺口，订单簿将无法恢复一致。

验证校验器是否按预期工作

用断言把上述示例固化为可重复执行的测试，覆盖“覆盖丢弃”“首次衔接”“缺口报错”三条路径，并检查 currentSeq 未被缺口消息污染。

def test_aligner():
    a = SequenceAligner(1000)
    d1 = a.check(DiffMessage(980, 995))
    assert d1.code == "COVERED" and not a.aligned

    d2 = a.check(DiffMessage(996, 1005))
    assert d2.code == "ALIGNED" and a.aligned and a.current_seq == 1005

    d3 = a.check(DiffMessage(1010, 1020))
    assert d3.code == "GAP" and a.current_seq == 1005   # 缺口不推进基准

    # 缺口补齐后应恢复
    d4 = a.check(DiffMessage(1006, 1015))
    assert d4.code == "ALIGNED" and a.current_seq == 1015

test_aligner()
print("all assertions passed")

通过条件：四条断言全部成立，且 currentSeq 在缺口消息后仍为 1005，说明校验器没有把不连续的消息错误地应用到订单簿基准上。若断言失败，优先检查判定顺序与 lastUpdateId + 1 的边界是否写成了 lastUpdateId。

← 上一章 📖 返回目录 下一章 →