第3章实时数据采集

3. 实时数据采集：对接交易所行情API（WebSocket/Feed Handler），数据清洗与对齐

做高频做市，最怕什么？

不是策略亏钱，而是你拿到的行情数据是错的。或者更糟——数据是对的，但慢了50微秒。

我见过太多团队，策略模型跑得飞起，结果一上实盘就崩。查到最后，问题全出在数据采集这一层。说白了，地基没打牢，楼盖得再高也得塌。

这一章，我们就来聊聊怎么把交易所的行情数据稳稳当当地接进来，再洗干净、对齐好。

3.1 选对协议：WebSocket vs Feed Handler

交易所给咱们提供了两种主流方式：WebSocket 和 Feed Handler。怎么选？

特性	WebSocket	Feed Handler
延迟	1-10ms（受网络抖动影响）	10-100μs（硬件级加速）
部署复杂度	低，几行代码搞定	高，需要FPGA或专用网卡
数据完整性	依赖心跳重连机制	硬件序列号校验，几乎无丢包
成本	几乎为零	数万到数十万/年
适用场景	回测、模拟盘、低延迟策略	高频做市、套利、做市商直连

我个人习惯是：回测和模拟盘用WebSocket，实盘高频必须上Feed Handler。为什么？

有一次我在项目中用WebSocket接币安的深度数据，平时跑得好好的。结果某天行情剧烈波动，WebSocket连接突然断了，重连花了3秒。这3秒里，策略还在按旧数据下单，直接亏了6位数。嗯，从那以后，实盘我再也不敢只用WebSocket了。

⚠️ 避坑指南： 我曾经踩过一个坑——WebSocket的ping/pong机制。很多交易所要求客户端定期发ping，否则会断开连接。但如果你在Python里用asyncio，别忘了加个定时器。否则，半夜行情一冷清，连接就断了，第二天早上才发现。

3.2 数据清洗：别让脏数据毁了你的策略

行情数据从交易所出来，到你的策略引擎，中间要过好几道关。每一关都可能引入脏数据。

常见的脏数据有这几类：

重复数据：同一个tick被推送了两次
乱序数据：后发的tick先到了（网络延迟导致）
缺失数据：某个时间点的快照丢了
异常数据：价格突然跳变（比如从100跳到10000）

怎么处理？我一般分三步走：

第一步：去重

每个tick都有一个唯一的序列号（Sequence Number）。维护一个last_seq变量，只处理比它大的序列号。重复的直接丢弃。

# 伪代码示例
last_seq = 0
def on_tick(tick):
    if tick.seq <= last_seq:
        return  # 重复数据，直接丢弃
    last_seq = tick.seq
    process_tick(tick)

第二步：排序

乱序数据比较头疼。我的做法是：维护一个长度为N的滑动窗口。窗口内的tick按序列号排序，窗口外的tick如果序列号比窗口内最小的还小，就丢弃。

💡 小技巧： N怎么选？我一般设成100。太小了容易丢数据，太大了延迟会增加。你可以根据交易所的推送频率来调。比如币安深度数据每秒推送100次，那N=100就够用了。

第三步：异常检测

价格跳变怎么处理？简单粗暴：设定一个阈值。比如，当前tick的价格与上一个tick相比，变化超过5%，就标记为异常，先缓存起来，等下一个tick来确认。

def is_anomaly(current_price, last_price, threshold=0.05):
    change = abs(current_price - last_price) / last_price
    if change > threshold:
        return True
    return False

你想想看，如果直接把这个异常tick喂给策略，策略可能会瞬间下出天量订单。我见过有人因为这个被交易所罚款的。

3.3 数据对齐：让不同数据源“说同一种语言”

做高频做市，你通常要同时接多个交易所的数据。比如，你在币安做市，但也要看OKX的行情来套利。

问题来了：不同交易所的时间戳格式不一样，价格精度不一样，甚至买卖盘的结构都不一样。

怎么对齐？

时间戳对齐

交易所给的时间戳，有的是毫秒级，有的是微秒级。我建议统一转成纳秒级，用int64存储。为什么？因为纳秒级精度足够做高频了，而且int64做比较运算非常快。

# 统一转成纳秒
def normalize_timestamp(ts, unit='ms'):
    if unit == 'ms':
        return ts * 1_000_000
    elif unit == 'us':
        return ts * 1_000
    elif unit == 'ns':
        return ts
    else:
        raise ValueError(f"Unknown unit: {unit}")

价格精度对齐

不同交易所对同一个币种的价格精度可能不同。比如，币安的BTC/USDT价格精度是0.01，而OKX是0.1。

我的做法是：统一用最小价格变动单位（Tick Size）的整数倍来表示。这样，不管哪个交易所的数据进来，都能直接比较。

  核心原则：
  数据对齐的本质，是让不同来源的数据在同一个坐标系下可比。时间、价格、数量，这三个维度必须统一。

买卖盘对齐

这个最麻烦。不同交易所的买卖盘结构不一样：

币安：按价格档位推送，每次只推送变化的部分
OKX：全量推送，每次都是完整的买卖盘
Coinbase：增量+全量混合

我建议统一维护一个本地的Order Book快照。不管交易所怎么推，你都在本地维护一个完整的买卖盘。这样，策略层看到的永远是一个统一格式的数据。

class OrderBook:
    def __init__(self):
        self.bids = {}  # 价格 -> 数量
        self.asks = {}
    
    def update(self, price, quantity, side):
        if quantity == 0:
            # 删除该档位
            if side == 'bid':
                self.bids.pop(price, None)
            else:
                self.asks.pop(price, None)
        else:
            # 更新该档位
            if side == 'bid':
                self.bids[price] = quantity
            else:
                self.asks[price] = quantity

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

说了这么多，咱们来看看整体架构长什么样。

你看，整个流程其实就三层：

采集层：负责跟交易所建立连接，收数据
清洗层：把脏数据过滤掉，保证数据质量
对齐层：把不同来源的数据统一格式，喂给策略

每一层都只做一件事，层与层之间通过队列解耦。这样，即使某一层出了问题，也不会影响其他层。

💡 我的经验： 层与层之间用无锁队列（Lock-Free Queue）来传递数据。Python里可以用queue.Queue，但性能要求高的话，建议用collections.deque加原子操作。我在一个项目中用deque实现了单线程下每秒处理50万笔tick，延迟不到1微秒。

嗯，数据采集这块，说白了就是“接进来、洗干净、对齐好”。每一步都不难，但每一步都容易出问题。你想想看，如果数据采集这层没做好，后面的策略再牛也没用。

我曾经在一个项目中，就因为数据对齐没做好，两个交易所的买卖盘差了0.1个tick，导致套利策略一直在假信号里打转。查了三天才找到原因。从那以后，我每次上线新交易所，都会先跑一周的模拟盘，专门验证数据质量。

好了，这一章就到这里。记住：数据是策略的命根子。把数据采集这层做扎实了，后面的路就好走了。

公众号:蓝海资料掘金营，微信deep3321

📚 课程目录

3. 实时数据采集：对接交易所行情API（WebSocket/Feed Handler），数据清洗与对齐

3.1 选对协议：WebSocket vs Feed Handler

3.2 数据清洗：别让脏数据毁了你的策略

第一步：去重

第二步：排序

第三步：异常检测

3.3 数据对齐：让不同数据源“说同一种语言”

时间戳对齐

价格精度对齐

买卖盘对齐

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