3. 数据获取：WebSocket实时行情订阅与REST API历史数据抓取
做价差套利，数据就是你的眼睛。没有数据，策略再好也是瞎蒙。
我个人习惯把数据获取分成两条线：实时行情和历史数据。前者用来发现机会，后者用来回测验证。两条线缺一不可。
3.1 为什么需要两套数据接口？
你想想看，套利机会往往转瞬即逝。比如币安和OKX之间的价差，可能几秒钟就消失了。这时候你需要WebSocket——它像一根水管，源源不断地把最新价格推送给你。
但光有实时数据不够。你要回测策略，要分析历史价差规律，就得靠REST API去拉历史K线。我在项目中遇到过一个问题：只用WebSocket攒数据，结果程序重启后历史数据全丢了，回测根本没法做。
所以我的建议是：WebSocket负责“现在”，REST API负责“过去”。两者配合，才能构建完整的数据体系。
3.2 WebSocket实时行情订阅
WebSocket说白了就是一条长连接。客户端连上服务器后，服务器有数据就主动推过来。你不用反复去问“有新数据吗？”，省带宽，也省时间。
3.2.1 连接与订阅
以币安为例，订阅BTC/USDT的深度行情，代码大概长这样：
import websocket
import json
def on_message(ws, message):
data = json.loads(message)
# 处理实时行情数据
print(f"最新价: {data['c']}")
def on_error(ws, error):
print(f"连接出错: {error}")
def on_close(ws, close_status_code, close_msg):
print("连接已关闭")
def on_open(ws):
# 订阅深度行情
subscribe_msg = {
"method": "SUBSCRIBE",
"params": ["btcusdt@depth20"],
"id": 1
}
ws.send(json.dumps(subscribe_msg))
websocket.enableTrace(False)
ws = websocket.WebSocketApp(
"wss://stream.binance.com:9443/ws",
on_open=on_open,
on_message=on_message,
on_error=on_error,
on_close=on_close
)
ws.run_forever()嗯，这里要注意：订阅的频道不要太多。我曾经一次性订阅了20个交易对的深度数据，结果本地带宽被打满，程序直接卡死。后来我改成只订阅价差最大的几个交易对，问题就解决了。
3.2.2 数据清洗与缓存
WebSocket推送的数据很原始。你需要做两件事：
- 去重：同一笔成交可能推送多次，用时间戳+交易对做唯一键去重
- 缓存：用队列暂存数据，防止处理速度跟不上推送速度
3.3 REST API历史数据抓取
历史数据用来做回测。没有历史数据，你根本不知道策略在过去的市场里表现如何。
3.3.1 请求频率限制
交易所都有频率限制。币安REST API的限制是每分钟1200次请求。你如果一股脑发请求，很快就会被封IP。
我建议的做法是：用时间窗口控制请求频率。比如每秒钟最多发20次请求，超了就等一秒再发。
import time
import requests
class RateLimiter:
def __init__(self, max_calls, period):
self.max_calls = max_calls
self.period = period
self.calls = []
def wait_if_needed(self):
now = time.time()
# 移除过期的调用记录
self.calls = [t for t in self.calls if now - t < self.period]
if len(self.calls) >= self.max_calls:
sleep_time = self.calls[0] + self.period - now
if sleep_time > 0:
time.sleep(sleep_time)
self.calls.append(time.time())
limiter = RateLimiter(max_calls=20, period=1)
def fetch_klines(symbol, interval, limit=500):
limiter.wait_if_needed()
url = "https://api.binance.com/api/v3/klines"
params = {
"symbol": symbol,
"interval": interval,
"limit": limit
}
resp = requests.get(url, params=params)
return resp.json()3.3.2 数据存储策略
历史数据拉下来后，存哪里？我个人推荐用本地CSV文件+数据库双保险。
- CSV文件：方便人工查看和调试
- SQLite/MySQL：方便程序快速查询和回测
存储结构建议这样：
| 字段 | 类型 | 说明 |
|---|---|---|
| symbol | VARCHAR | 交易对，如BTCUSDT |
| open_time | BIGINT | 开盘时间戳（毫秒） |
| open | DECIMAL | 开盘价 |
| high | DECIMAL | 最高价 |
| low | DECIMAL | 最低价 |
| close | DECIMAL | 收盘价 |
| volume | DECIMAL | 成交量 |
3.4 数据获取整体架构
下面这张图是我自己项目里用的数据获取架构，你可以参考一下：
这个架构看起来简单，但实际跑起来坑不少。我刚开始做的时候，WebSocket断连了都不知道，等发现时已经错过了好几波行情。后来我加了心跳检测和自动重连机制，才算稳定下来。
3.5 避坑指南
最后分享几个我踩过的坑：
- WebSocket断连：交易所会定期断开空闲连接。一定要实现自动重连，不然数据流就断了
- 数据时间对齐：不同交易所的K线时间戳可能差几毫秒。做价差计算时，一定要先对齐时间
- 磁盘空间：历史数据越攒越多。我建议定期清理，只保留最近3个月的数据用于回测