数据源接入：WebSocket实时行情接入、REST API历史数据获取、数据清洗与对齐
做价差回归策略，数据是命根子。这话一点都不夸张。
我见过太多人，策略逻辑写得漂漂亮亮，结果一上线就亏钱。为什么？数据源没处理好。要么实时行情断流了，要么历史数据有坑，要么两个合约的时间戳对不上。今天咱们就把这块硬骨头啃下来。
实时行情接入：WebSocket 才是王道
做实时监控，千万别用 REST API 去轮询。为什么？延迟高、浪费带宽、还容易被交易所封IP。
我个人习惯，所有实时数据都用 WebSocket。它就像一根管道，交易所那边有数据变化，直接推过来，我们这边被动接收就行。
核心要点：WebSocket 建立一次连接，后续数据全走这条通道。延迟通常在 10ms 以内，比 REST 轮询快一个数量级。
下面是我常用的 WebSocket 接入模板。以币安为例，订阅两个合约的深度数据：
import websocket
import json
import threading
class RealTimeFeed:
def __init__(self, symbols):
self.symbols = symbols
self.ws = None
self.latest_data = {}
def on_message(self, ws, message):
data = json.loads(message)
# 解析数据，提取买一卖一价
symbol = data['s']
bid = float(data['b'])
ask = float(data['a'])
self.latest_data[symbol] = {
'bid': bid,
'ask': ask,
'mid': (bid + ask) / 2,
'timestamp': data['E']
}
def on_error(self, ws, error):
print(f"WebSocket 出错了: {error}")
# 自动重连逻辑
self.reconnect()
def start(self):
url = "wss://stream.binance.com:9443/ws"
streams = "/".join([f"{s.lower()}@bookTicker" for s in self.symbols])
full_url = f"{url}/{streams}"
self.ws = websocket.WebSocketApp(
full_url,
on_message=self.on_message,
on_error=self.on_error
)
# 跑在独立线程里
wst = threading.Thread(target=self.ws.run_forever)
wst.daemon = True
wst.start()
# 使用示例
feed = RealTimeFeed(['BTCUSDT', 'ETHUSDT'])
feed.start()避坑指南：我曾经在项目里忘了加自动重连，结果半夜交易所重启，我的策略傻等了一晚上数据。记住，WebSocket 一定要有重连机制，最好加上指数退避。
历史数据获取：REST API 的优雅用法
实时数据有了，但策略回测、参数优化还得靠历史数据。REST API 虽然慢，但胜在稳定、数据全。
这里有个坑：很多交易所的 REST API 有频率限制。你想想看，如果一次请求只能拿 500 根 K 线，要拿三年数据得发多少请求？
我的做法是分片拉取，然后本地缓存：
import requests
import time
from datetime import datetime, timedelta
def fetch_historical_klines(symbol, interval, start_time, end_time):
"""
分片拉取历史K线数据
"""
base_url = "https://api.binance.com/api/v3/klines"
limit = 500 # 每次最多500根
all_klines = []
current_start = start_time
while current_start < end_time:
params = {
'symbol': symbol,
'interval': interval,
'startTime': int(current_start.timestamp() * 1000),
'limit': limit
}
resp = requests.get(base_url, params=params)
if resp.status_code != 200:
print(f"请求失败: {resp.status_code}")
time.sleep(1)
continue
klines = resp.json()
if not klines:
break
all_klines.extend(klines)
# 更新起始时间
last_kline_time = klines[-1][0]
current_start = datetime.fromtimestamp(last_kline_time / 1000)
# 礼貌性等待
time.sleep(0.1)
return all_klines
# 获取最近30天的1小时K线
end = datetime.now()
start = end - timedelta(days=30)
data = fetch_historical_klines('BTCUSDT', '1h', start, end)注意：不同交易所的 K 线数据结构不一样。币安返回的是数组，OKX 返回的是对象。写代码前一定先看文档，别想当然。
数据清洗与对齐：最容易被忽视的环节
数据拿到手了，你以为就能直接用了吗？太天真了。
我遇到过的情况：两个交易所的 BTC 永续合约，一个用 UTC 时间，一个用本地时间。直接算价差？算出来的全是错的。
数据清洗，说白了就三件事：
- 去重：WebSocket 偶尔会推送重复数据，用时间戳去重
- 补缺失：某个时间点数据没到，用前一个值填充
- 对齐时间戳：把两个合约的数据对齐到同一个时间轴上
下面是我常用的对齐函数：
import pandas as pd
import numpy as np
def align_spread_data(data1, data2, freq='1s'):
"""
将两个合约的数据对齐到同一时间轴
"""
# 转成 DataFrame
df1 = pd.DataFrame(data1).set_index('timestamp')
df2 = pd.DataFrame(data2).set_index('timestamp')
# 重采样到固定频率
df1_resampled = df1['mid'].resample(freq).last().ffill()
df2_resampled = df2['mid'].resample(freq).last().ffill()
# 合并
aligned = pd.DataFrame({
'symbol1': df1_resampled,
'symbol2': df2_resampled
})
# 计算价差
aligned['spread'] = aligned['symbol1'] - aligned['symbol2']
# 去掉两边有空值的行
aligned = aligned.dropna()
return aligned
# 使用示例
aligned_data = align_spread_data(
feed.latest_data.get('BTCUSDT'),
feed.latest_data.get('ETHUSDT'),
freq='1s'
)整体架构：一张图说清楚
上面讲了三个模块，它们怎么配合？我画了张图，你看一眼就明白了：
你看，整个流程其实很清晰。左边是实时数据，右边是历史数据，中间经过清洗对齐，最后输出干净的价差序列。
我的经验：数据对齐这一步，千万别用循环去逐条比对。用 pandas 的 resample 和 merge，一行代码顶你写 50 行循环。性能差几十倍。
实战中的几个坑
最后分享几个我踩过的坑，你遇到了能少走弯路：
- 时间戳单位不一致：有的交易所用毫秒，有的用微秒，还有的用秒。统一转成毫秒再处理。
- 节假日数据缺失：某些合约在节假日不交易，数据会断档。用 ffill 填充时要小心，别把几小时前的价格当成当前价。
- WebSocket 断线重连：别只重连一次。我一般重连 5 次，每次间隔递增，还不行就发告警。
- REST API 限频：有些交易所的限频是每分钟 1200 次，算好你的请求量，别超了被拉黑。
数据源这块，说白了就是「接进来、洗干净、对齐好」。这三步做好了，你的策略就成功了一半。剩下的，就是策略逻辑的事了。