4、实时行情处理：行情订阅、数据清洗、Tick级数据对齐

做套利的人都知道一句话：行情是命，延迟是病。

我做了这么多年跨交易所套利，见过太多策略本身没问题，最后死在行情处理上的案例。说白了，你算法再牛，如果拿到的Tick数据是脏的、乱的、不同步的，那一切都是白搭。

这一章，我们就来聊聊实时行情处理的核心三件事：怎么订阅、怎么清洗、怎么对齐。

4.1 行情订阅：别一股脑全收

很多新手上来就想订阅所有交易所的所有品种。嗯，这里要注意——行情数据量比你想象的大得多。

以币安和OKX为例，光BTC/USDT一个交易对，高峰期每秒能产生几十甚至上百笔Tick。如果你同时订阅50个交易对，带宽和CPU直接拉满。

我的建议是：只订阅你策略需要的交易对。比如你做BTC和ETH的价差套利，那就只订阅这两个币种在目标交易所的深度和成交数据。

订阅方式上，我个人习惯用WebSocket。RESTful轮询在Tick级场景下基本不可用，延迟太高。

// 伪代码示例：WebSocket订阅
ws.connect('wss://exchange1.com/ws')
ws.subscribe({
  channel: 'ticker',
  symbol: 'BTC/USDT'
})

ws.on('message', function(data) {
  // 收到行情数据
  processTick(data)
})

4.2 数据清洗：脏数据是隐形杀手

你以为交易所发过来的数据都是干净的吗？太天真了。

我在项目中遇到过好几次，交易所偶尔会推送一些异常数据：价格突然跳变、时间戳错乱、甚至重复推送同一笔成交。如果不做清洗，策略会做出错误决策。

数据清洗我一般做三件事：

• 去重：同一笔成交ID出现两次，只保留第一次
• 时间戳校验：如果当前Tick的时间戳比上一个还早，直接丢弃
• 价格合理性检查：价格偏离最近N笔成交均值超过3个标准差，标记为异常

def clean_tick(tick, last_tick):
    # 去重
    if tick['trade_id'] == last_tick['trade_id']:
        return None
    
    # 时间戳校验
    if tick['timestamp'] <= last_tick['timestamp']:
        return None
    
    # 价格合理性检查
    price_change = abs(tick['price'] - last_tick['price']) / last_tick['price']
    if price_change > 0.03:  # 超过3%
        log.warning(f"异常价格跳变: {tick}")
        return None
    
    return tick

4.3 Tick级数据对齐：套利的灵魂

好了，行情订阅了，数据也清洗了。接下来是最关键的一步：对齐。

你想想看，交易所A和交易所B的Tick到达时间是不一样的。网络延迟、处理延迟、甚至交易所自身的撮合速度都不同。如果你拿交易所A在T1时刻的价格，去对比交易所B在T2时刻的价格，那算出来的价差毫无意义。

怎么做对齐？我常用的方法有两种：

1. 时间戳对齐法：以统一的时间基准（比如NTP同步后的本地时间）为准，将两个交易所的Tick对齐到同一个时间窗口内。
2. 事件驱动对齐法：当收到交易所A的Tick时，等待交易所B在极短时间窗口内（比如10ms）的最近一笔Tick，然后配对。

我个人更推荐第二种。为什么？因为时间戳对齐依赖交易所的时间戳精度，而不同交易所的时间戳可能偏差很大。事件驱动对齐更贴近实际交易场景。

class TickAligner:
    def __init__(self, window_ms=10):
        self.window_ms = window_ms
        self.buffer = {'exchange_a': [], 'exchange_b': []}
    
    def on_tick(self, tick, exchange):
        self.buffer[exchange].append(tick)
        self._try_align()
    
    def _try_align(self):
        # 取两个交易所最近的一笔Tick
        a = self.buffer['exchange_a'][-1] if self.buffer['exchange_a'] else None
        b = self.buffer['exchange_b'][-1] if self.buffer['exchange_b'] else None
        
        if a and b:
            time_diff = abs(a['timestamp'] - b['timestamp'])
            if time_diff <= self.window_ms:
                # 对齐成功，计算价差
                spread = a['price'] - b['price']
                self.on_spread(spread)

4.4 整体架构：一张图说清楚

说了这么多，我们来看一张整体流程图。这是我做套利系统时常用的行情处理链路。

从这张图你可以看到，整个链路是串行的。任何一个环节出问题，下游都会受影响。所以我一直强调：行情处理不是锦上添花，而是生死攸关。

4.5 避坑指南

最后，分享几个我亲身踩过的坑：

• 不要依赖交易所的时间戳。我曾经发现某交易所的时间戳比真实时间慢了整整2秒。从那以后，我所有对齐逻辑都基于本地NTP时间。
• 行情断连要自动重连。WebSocket偶尔会断开，必须实现心跳检测和自动重连机制。我见过有人没做重连，策略在断连期间一直用旧数据交易，亏了十几万。
• 数据清洗要记录日志。每次丢弃异常数据，都要记录原因。这样复盘时才能知道是交易所的问题还是自己代码的问题。