3. 重复增量消息的幂等过滤与状态一致性验证

为什么重复增量必须被过滤

交易所的增量推送通道在弱网重连、心跳超时补发、多路复用切换等场景下，可能把同一序列号的消息投递两次。若直接应用，价格档位的数量会被重复加减，订单簿出现“凭空多出/少掉”的挂单量，且这种偏差会随每次重复累积。幂等过滤器的职责是：以序列号为唯一键，只让每个序列号被应用一次，同时保证被丢弃的消息不改变任何状态。

本例采用“已应用序列号水位 + 重复计数”的轻量方案：维护 lastAppliedSeq（已成功应用的最大序列号）与 seen 集合（用于识别水位以下的迟到重复）。判定规则如下：

• seq <= lastAppliedSeq 且命中 seen：判定为重复，丢弃并计数。
• seq == lastAppliedSeq + 1：正常应用，推进水位。
• seq > lastAppliedSeq + 1：缺口，交由乱序缓冲处理（本例不展开，仅标记为 GAP 不应用）。

幂等过滤器的实现

下面是一个可直接运行的 Python 实现。订单簿用 dict 保存价格档位数量，apply 负责按方向加减数量；过滤器在应用前完成判定，重复消息直接返回 False 且不触碰订单簿。

from collections import deque

class OrderBook:
    def __init__(self):
        self.bids = {}   # price -> qty
        self.asks = {}

    def apply(self, msg):
        book = self.bids if msg["side"] == "B" else self.asks
        px, qty = msg["px"], msg["qty"]
        new_qty = book.get(px, 0) + qty
        if new_qty <= 0:
            book.pop(px, None)
        else:
            book[px] = new_qty

    def snapshot_hash(self):
        # 稳定序列化后取哈希，用于一致性对比
        import hashlib, json
        payload = json.dumps(
            {"bids": sorted(self.bids.items()),
             "asks": sorted(self.asks.items())},
            sort_keys=True)
        return hashlib.sha256(payload.encode()).hexdigest()[:16]

class IdempotentFilter:
    def __init__(self, start_seq=0, seen_window=4096):
        self.last_applied = start_seq
        self.seen = deque(maxlen=seen_window)
        self.seen_set = set()
        self.dup_count = 0
        self.gap_count = 0

    def _remember(self, seq):
        if len(self.seen) == self.seen.maxlen:
            self.seen_set.discard(self.seen[0])
        self.seen.append(seq)
        self.seen_set.add(seq)

    def feed(self, msg, book):
        seq = msg["seq"]
        if seq <= self.last_applied and seq in self.seen_set:
            self.dup_count += 1
            return "DUP"          # 重复，丢弃
        if seq == self.last_applied + 1:
            book.apply(msg)
            self.last_applied = seq
            self._remember(seq)
            return "APPLIED"
        if seq > self.last_applied + 1:
            self.gap_count += 1
            return "GAP"          # 缺口，交由缓冲处理
        # seq 低于水位但不在 seen 中：超出窗口的迟到消息
        self.dup_count += 1
        return "STALE"

构造含重复的测试流并验证一致性

验证思路是双路径对比：路径 A 把原始消息流（含重复）直接喂给一个“无过滤”订单簿，路径 B 经过幂等过滤器再喂给另一个订单簿。若过滤器正确，两条路径的最终状态哈希必须相等，且路径 B 的 dup_count 应等于注入的重复条数。

def build_stream():
    base = [
        {"seq": 1041, "side": "B", "px": 100.0, "qty": 5},
        {"seq": 1042, "side": "B", "px": 100.0, "qty": 3},
        {"seq": 1043, "side": "S", "px": 101.0, "qty": 4},
        {"seq": 1044, "side": "B", "px": 99.5,  "qty": 2},
        {"seq": 1045, "side": "S", "px": 101.0, "qty": -1},
    ]
    # 注入重复：1042 再来一次，1044 再来两次
    return base + [base[1], base[3], base[3]]

raw_book = OrderBook()
for m in build_stream():
    raw_book.apply(m)          # 路径A：不过滤

filtered_book = OrderBook()
f = IdempotentFilter(start_seq=1040)
results = [f.feed(m, filtered_book) for m in build_stream()]

print("路径A哈希:", raw_book.snapshot_hash())
print("路径B哈希:", filtered_book.snapshot_hash())
print("判定序列:", results)
print("重复计数:", f.dup_count, "缺口计数:", f.gap_count)

预期现象（由给定输入直接推出）：results 中第 6、7、8 条为 DUP，其余为 APPLIED；dup_count == 3；两条路径哈希相等。若哈希不等，说明过滤器放行了重复消息或误丢了正常消息，需回查 seen 窗口与水位推进逻辑。