第三章：数据源接入——交易所API对接（REST与WebSocket），数据清洗与对齐
做跨市场套利，第一步不是写策略，而是搞定数据。
我见过太多团队，策略逻辑写得天花乱坠，结果一上线就亏钱。为什么？数据源没接好。要么延迟高，要么数据对不齐，要么清洗环节漏了脏数据。说白了，数据是套利交易的「米」，没有米，再好的厨子也做不出饭。
这一章，我们就来聊聊怎么把交易所的数据「搬」到自己的系统里，并且搬得干净、搬得及时。
3.1 REST API：稳，但慢
REST API 就像你去银行柜台办业务。你提交请求，等柜员处理，然后拿到结果。这个过程是同步的，你发一个请求，等一个响应。
在量化交易里，REST 通常用来做两件事：
- 获取历史数据：比如拉取过去1000根K线，用来回测或初始化策略状态。
- 执行交易指令：下单、撤单、查询持仓。
我个人习惯，把 REST 当作「慢通道」。它不追求极致的速度，但要求绝对的可靠。
核心要点：REST 请求一定要做「重试机制」和「限速控制」。交易所都有频率限制，你打得太快会被封IP。
来看一段伪代码，展示如何用 REST 拉取币安的最新价格：
import requests
import time
def fetch_price(symbol):
url = f"https://api.binance.com/api/v3/ticker/price?symbol={symbol}"
for retry in range(3):
try:
resp = requests.get(url, timeout=2)
if resp.status_code == 200:
return float(resp.json()['price'])
else:
time.sleep(0.5 * (retry + 1))
except Exception as e:
print(f"请求失败，重试第{retry+1}次: {e}")
return None嗯，这里要注意。重试间隔不要固定，最好用指数退避。我第一次写的时候没注意，结果交易所一波动，我的重试请求直接把API打挂了。后来被交易所发邮件警告，才老老实实加了退避逻辑。
3.2 WebSocket：快，但容易断
WebSocket 就不一样了。它像你打开水龙头，水一直流。你不需要每次去拧开关，只要接住流过来的数据就行。
对于套利交易，WebSocket 是主力。因为我们需要实时行情，毫秒级的延迟差异可能就是盈亏的分水岭。
WebSocket 有几个坑，我一个个说：
- 断线重连：网络波动、交易所维护、连接超时，都会导致WebSocket断开。你必须实现自动重连逻辑。
- 心跳维持：很多交易所要求客户端定期发送ping帧，否则会主动断开连接。
- 数据积压：如果行情太快，你的处理速度跟不上，数据会在缓冲区堆积。这时候要么丢数据，要么内存爆掉。
避坑指南：我曾经在对接某二线交易所时，它的WebSocket每隔45分钟就会自动断开一次，而且不发送任何断开通知。我的程序傻傻地等着，直到半小时后才发现数据停了。从那以后，我所有WebSocket连接都加了一个「静默超时检测」——如果超过60秒没有收到任何数据，就主动重连。
下面是一个带自动重连的WebSocket示例：
import websocket
import json
import threading
class MarketDataStream:
def __init__(self, url, callback):
self.url = url
self.callback = callback
self.ws = None
self.keep_running = True
def on_message(self, ws, message):
data = json.loads(message)
self.callback(data)
def on_error(self, ws, error):
print(f"WebSocket错误: {error}")
def on_close(self, ws, close_status_code, close_msg):
print(f"连接关闭，状态码: {close_status_code}")
if self.keep_running:
self.reconnect()
def on_open(self, ws):
print("连接已建立")
# 订阅行情
subscribe_msg = {
"method": "SUBSCRIBE",
"params": ["btcusdt@ticker"],
"id": 1
}
ws.send(json.dumps(subscribe_msg))
def reconnect(self):
print("尝试重连...")
threading.Timer(3.0, self.connect).start()
def connect(self):
self.ws = websocket.WebSocketApp(
self.url,
on_message=self.on_message,
on_error=self.on_error,
on_close=self.on_close,
on_open=self.on_open
)
self.ws.run_forever(ping_interval=30, ping_timeout=10)
def stop(self):
self.keep_running = False
if self.ws:
self.ws.close()3.3 数据清洗：脏数据是策略的毒药
数据到了，但别急着用。交易所返回的数据，远比你想象的脏。
我遇到过的情况包括：
- 价格字段出现负数（交易所内部测试数据误推送）
- 时间戳是字符串格式，而且不同交易所格式还不一样
- 成交量突然变成0，但价格还在跳动（明显是数据缺失）
- 同一笔订单的成交记录重复推送了两次
清洗规则，我总结了三步：
- 格式校验：检查字段类型是否正确。价格必须是正数，时间戳必须是整数或标准时间格式。
- 逻辑校验：检查数据是否合理。比如买一价必须小于卖一价，成交量不能为负。
- 去重：根据交易ID或时间戳+价格+数量的组合，去除重复数据。
小技巧：清洗后的数据，我习惯打一个「清洗标记」字段。这样在后续分析时，可以快速区分原始数据和清洗后的数据，方便排查问题。
3.4 数据对齐：套利的核心难点
数据清洗完了，接下来是最关键的一步——对齐。
为什么需要对齐？因为不同交易所的行情到达时间不一样。假设币安在T1时刻推送了BTC价格，OKX在T2时刻推送了同一时刻的价格。如果你直接用这两个价格计算价差，结果可能是错的——因为这两个价格对应的「市场时刻」不同。
对齐的方法，我常用两种：
| 对齐方式 | 原理 | 适用场景 | 缺点 |
|---|---|---|---|
| 时间戳对齐 | 以交易所返回的时间戳为准，将同一秒内的数据归为一组 | 低频套利（秒级） | 交易所时间戳可能不同步 |
| 事件驱动对齐 | 以本地接收到数据的时间为准，设定一个时间窗口（如100ms） | 高频套利（毫秒级） | 网络延迟会影响对齐精度 |
我个人更倾向于「事件驱动对齐」。因为交易所的时间戳，说实话，我不太信得过。有些交易所的服务器时间会漂移，你拿它的时间戳对齐，反而对不准。
具体做法是这样的：
- 本地维护一个「行情缓存」，按交易所+交易对分类
- 每收到一条新数据，打上本地接收时间戳
- 每隔一个固定时间窗口（比如50ms），扫描所有缓存，取出最新的一条数据
- 用这些「对齐后」的数据计算价差
注意：时间窗口不能设得太小，否则可能等不到所有交易所的数据；也不能设得太大，否则套利机会就溜走了。我一般根据交易所的网络延迟来动态调整，通常在50ms到200ms之间。
3.5 知识体系总览
说了这么多，我们来画一张图，把整个数据接入的流程串起来。
这张图把整个流程分成了四层：数据源层、接入层、清洗层、对齐层。每一层都有自己需要解决的问题。你想想看，如果其中任何一层出了问题，后面的策略逻辑再漂亮也没用。
好了，数据源接入这块，核心内容就是这些。从REST到WebSocket，从清洗到对齐，每一步都有坑，但也都有解法。做套利交易，数据是地基，地基稳了，楼才能盖得高。