第二章：交易所数据源解析——WebSocket连接、REST API快照、增量数据流

做量化交易，第一道坎就是怎么把交易所的数据拿过来。

你可能觉得，不就是连个API嘛，有什么难的？

嗯，我刚开始也这么想。直到有一次，我的订单簿在盘中突然出现了一个巨大的价差缺口——买单和卖单之间差了整整两个tick。当时我冷汗都下来了，赶紧查日志，发现是增量数据丢了一条。

从那以后，我对数据源解析这件事，再也不敢掉以轻心。

2.1 数据源的整体架构

交易所的数据源，说白了就两种模式：

• REST API：你主动去问，交易所给你一个快照。适合拿全量数据，但实时性差。
• WebSocket：交易所主动推给你。适合拿增量数据，实时性好。

实际生产中，我们通常把两者结合起来用。我习惯叫它「快照+增量」模式。

为什么不能只用WebSocket？

因为WebSocket连接可能会断。断连之后，你本地维护的订单簿就和交易所的不一致了。这时候就需要重新拿一次快照，再继续接增量。

2.2 WebSocket连接：实时性的基石

WebSocket这东西，说白了就是一个长连接。建立之后，交易所会持续往你这边推数据。

我个人习惯用 websocket-client 这个库，Python环境下很稳定。

import websocket
import json

def on_message(ws, message):
    data = json.loads(message)
    # 处理增量数据
    process_increment(data)

def on_error(ws, error):
    print(f"WebSocket错误: {error}")

def on_close(ws, close_status_code, close_msg):
    print("WebSocket连接已关闭")
    # 自动重连逻辑
    reconnect()

def on_open(ws):
    print("WebSocket连接已建立")
    # 订阅订单簿增量
    subscribe_msg = {
        "op": "subscribe",
        "args": ["orderbook.BTC-USDT"]
    }
    ws.send(json.dumps(subscribe_msg))

ws = websocket.WebSocketApp(
    "wss://api.exchange.com/ws",
    on_open=on_open,
    on_message=on_message,
    on_error=on_error,
    on_close=on_close
)
ws.run_forever()

2.3 REST API快照：重建的锚点

快照就是某一时刻订单簿的全量数据。你想想看，如果没有快照，只靠增量，万一中间丢了一条，后面的数据就全歪了。

所以快照的作用，就是给你一个「锚点」。

import requests

def get_orderbook_snapshot(symbol="BTC-USDT"):
    url = f"https://api.exchange.com/v1/orderbook/{symbol}"
    params = {"depth": 100}  # 取前100档
    
    resp = requests.get(url, params=params, timeout=5)
    data = resp.json()
    
    # 快照结构示例
    snapshot = {
        "bids": data["bids"],  # [[价格, 数量], ...]
        "asks": data["asks"],
        "timestamp": data["ts"],
        "sequence": data["seq"]  # 序列号，用于对齐增量
    }
    return snapshot

2.4 增量数据流：实时更新的血液

增量数据，就是交易所告诉你「刚才发生了什么变化」。

比如有人下了一笔买单，交易所会推一条消息：

{
    "type": "increment",
    "symbol": "BTC-USDT",
    "bids": [["50000.00", "1.5"]],  // 新增或更新的买单
    "asks": [],
    "sequence": 12345678
}

这里有个关键点：增量的更新方式是替换，不是累加。

什么意思呢？如果某个价格档位已经有挂单了，增量里来了同价格的新数据，那就直接覆盖掉旧数据。如果数量为0，就表示这个档位被撤单了，要删掉。

2.5 订单簿重建：把碎片拼成完整画面

好了，现在我们有快照和增量了。怎么拼起来？

我一般用红黑树或者跳表来维护订单簿，但为了演示，这里用Python的字典加排序：

class OrderBook:
    def __init__(self):
        self.bids = {}  # 价格 -> 数量
        self.asks = {}
        self.last_seq = 0
    
    def apply_snapshot(self, snapshot):
        """应用快照"""
        self.bids.clear()
        self.asks.clear()
        
        for price, qty in snapshot["bids"]:
            self.bids[float(price)] = float(qty)
        for price, qty in snapshot["asks"]:
            self.asks[float(price)] = float(qty)
        
        self.last_seq = snapshot["sequence"]
        print(f"快照已应用，序列号: {self.last_seq}")
    
    def apply_increment(self, inc):
        """应用增量"""
        # 检查序列号连续性
        if inc["sequence"] != self.last_seq + 1:
            print(f"警告：序列号不连续！期望 {self.last_seq + 1}，实际 {inc['sequence']}")
            # 这里应该触发重新获取快照
            return False
        
        # 处理买单增量
        for price, qty in inc.get("bids", []):
            price_f = float(price)
            qty_f = float(qty)
            if qty_f == 0:
                self.bids.pop(price_f, None)  # 删除
            else:
                self.bids[price_f] = qty_f  # 更新或新增
        
        # 处理卖单增量
        for price, qty in inc.get("asks", []):
            price_f = float(price)
            qty_f = float(qty)
            if qty_f == 0:
                self.asks.pop(price_f, None)
            else:
                self.asks[price_f] = qty_f
        
        self.last_seq = inc["sequence"]
        return True
    
    def get_top_n(self, n=10):
        """获取前n档买卖盘口"""
        sorted_bids = sorted(self.bids.items(), reverse=True)[:n]
        sorted_asks = sorted(self.asks.items())[:n]
        return sorted_bids, sorted_asks

2.6 实战中的几个坑

问题	表现	解决方案
序列号跳跃	增量丢失，订单簿偏差	检测到跳跃后立即重新获取快照
WebSocket断连	收不到增量	自动重连 + 重连后重新拿快照
快照与增量时序错乱	快照拿到后，增量已经包含了快照之前的变化	快照请求时记录时间戳，丢弃该时间戳之前的增量
数据延迟	增量到达时间晚于实际成交时间	在策略中引入时间戳比对，标记延迟数据

嗯，这里要注意一点：不同交易所的数据格式差异很大。有的用数组，有的用对象，有的序列号是字符串，有的是数字。我建议在解析层就统一成标准格式，不要让上层策略感知到这些差异。

说白了，数据源解析这件事，80%的工作都是在处理边界情况和异常恢复。正常流程写起来很简单，但真正考验功力的，是那些「万一」的情况。

我个人习惯在开发阶段就模拟各种异常场景：断网、数据乱序、重复推送、空数据……把这些都测一遍，上线后才睡得着觉。

---

公众号:蓝海数据掘金营，微信deep3321