📚 课程目录第4章数据清洗与时间对齐

4. 数据清洗与时间对齐

在订单簿重建中，原始行情数据几乎从不以“理想形态”出现。交易所推送的增量消息可能因网络多路径、撮合引擎并发输出、行情网关批量转发等原因出现乱序；断线重连或快照与增量拼接时可能产生重复消息；网络抖动、订阅切换或交易所侧限流会造成消息缺失；而不同数据源（如逐笔委托与逐笔成交、快照与增量）若来自不同时钟域，还会引入时钟偏差。这些问题如果不处理，重建出的订单簿会在某些时刻出现价格档位错乱、数量为负、买卖交叉等“不可能状态”。本节的核心任务是：把原始事件流清洗为一条单调、无重复、可追溯的统一事件时间轴，使重建过程中的事件顺序与真实市场逻辑一致。

一、先明确“事件时间”与“到达时间”的区别

清洗与对齐的第一步是建立正确的时间参照。每条行情消息至少涉及两个时间：交易所撮合或行情系统赋予的事件时间（如交易所时间戳、序列号隐含的顺序），以及本地接收到的到达时间。重建订单簿必须按事件时间排序，到达时间只能用于诊断延迟和判断乱序程度，不能作为排序依据。若数据源只提供序列号而不提供精确时间戳，则应以序列号作为主排序键，时间戳作为辅助校验。全文及后续图示统一约定：时间轴从左到右递增，序列号越小表示事件越早。

二、乱序：检测、缓冲与重排

乱序的典型表现是：后到的消息序列号小于已处理的最大序列号。工程上通常维护一个“已见最大序列号”max_seq，当新消息的序列号 seq <= max_seq 时判定为乱序或重复。对于乱序，不能简单丢弃，也不能立即处理，而应引入一个重排缓冲窗口：消息先进入按序列号排序的优先队列，当队列头部连续可用（即从期望序列号 expected_seq 开始无缺口）时再依次出队处理。窗口大小需要权衡：太小会漏掉迟到消息，太大会增加重建延迟。实践中可依据历史乱序深度分布设定，例如取 P99 乱序跨度作为窗口容量，并设置超时强制冲刷机制，避免个别缺失消息永久阻塞后续事件。

三、重复：幂等判定与去重键设计

重复消息常见于快照与增量拼接、断线重连后的补发、以及多路行情汇聚。去重的关键是设计稳定的去重键。对于带序列号的增量消息，去重键可直接使用 (通道标识, 序列号)；对于逐笔委托/成交，可使用 (交易所, 合约, 业务类型, 交易所时间戳, 序号)。需要注意：同一序列号在不同通道（如不同合约或不同业务流）中可能重复出现，因此去重键必须包含通道维度。去重应在进入重排缓冲之前完成，避免重复消息占用窗口并干扰连续性判断。对于快照类数据，重复快照本身无害，但重复应用增量才是危险来源，因此增量去重优先级最高。

四、缺失：缺口识别与修复策略

缺失的判定依据是序列号不连续：当 expected_seq 与当前可用最小序列号之间出现空洞，即存在缺口。缺口处理没有“万能解”，需要按数据源能力选择策略：若交易所提供重传或补拉接口，应优先补拉缺失区间；若只能依赖快照，则应在缺口处标记“不可信区间”，用最近一次有效快照重建基准，再从缺口之后的第一条连续增量继续应用，并在结果中记录该区间的重建置信度下降。切忌在缺口处“猜测”中间状态，因为订单簿的档位数量变化具有路径依赖性，跳过事件直接应用后续增量会导致数量累积错误。

五、时钟偏差：对齐到统一事件时间轴

当快照时间戳与增量时间戳来自不同时钟域时，直接比较可能产生毫秒级甚至更大的偏差，导致“快照已包含某增量，但增量时间戳看起来更晚”这类逻辑冲突。处理方法是：以交易所序列号或交易所时间戳为权威顺序，本地时间仅作参考；对跨源数据，先估计固定偏移（如用同一事件在两源中的时间差中位数），再在统一时间轴上按权威顺序排列。若无法获得权威顺序，则应明确降级为“近似对齐”，并在回测结果中标注该段数据的顺序不确定性。

六、落地实现：一条可复现的处理链

下面给出一个简化的处理链示例（Python 伪代码，参数为教学示例假设值），用于说明关键判断逻辑。示例中假设通道内序列号连续递增，重排窗口容量为 64，超时阈值为 500 毫秒。

from collections import deque
import heapq

WINDOW = 64
TIMEOUT_MS = 500

class EventAligner:
    def __init__(self):
        self.expected = None          # 期望的下一个序列号
        self.heap = []                # (seq, event_time, payload)
        self.seen = set()             # 去重键集合
        self.gaps = []                # 记录缺口区间

    def push(self, seq, event_time, payload, channel):
        key = (channel, seq)
        if key in self.seen:          # 重复：直接丢弃
            return
        self.seen.add(key)
        if self.expected is None:
            self.expected = seq
        heapq.heappush(self.heap, (seq, event_time, payload))
        if len(self.heap) > WINDOW:   # 窗口满，强制冲刷
            self._flush_until_gap()

    def _flush_until_gap(self):
        while self.heap and self.heap[0][0] == self.expected:
            seq, t, payload = heapq.heappop(self.heap)
            self._apply(seq, t, payload)
            self.expected += 1
        # 若堆顶序列号大于 expected，说明存在缺口
        if self.heap and self.heap[0][0] > self.expected:
            self.gaps.append((self.expected, self.heap[0][0] - 1))
            self.expected = self.heap[0][0]   # 跳过缺口，标记不可信

这段代码体现了三个关键点：去重发生在入堆之前；出堆只在序列号与期望值严格相等时进行，保证单调性；缺口出现时记录区间并推进期望值，而不是无限等待。实际工程中还需补充超时冲刷、补拉回调以及缺口区间的置信度标注。

七、如何判断清洗结果是否可信

清洗完成后，应做几项可观察的校验：其一，输出事件序列的序列号严格单调递增（允许记录缺口区间）；其二，重建订单簿在任意时刻不出现负数量、买卖价交叉、档位数量与快照对账偏差超阈值等异常；其三，统计乱序率、重复率、缺口数量与总时长占比，作为数据质量指标随回测结果一并输出。若缺口区间占比过高，应降低该段回测结论的权重，而不是把清洗后的数据当作“干净真相”。统一事件时间轴的价值，正在于让每一次重建都可追溯、可解释、可复核。