第四章：数据源获取——交易所API、WebSocket实时行情、历史数据回放
做量化交易，第一件事不是写策略，而是搞定数据。
我见过太多人，策略写得天花乱坠，结果一跑实盘，数据源先崩了。嗯，数据是策略的命根子，这话一点不夸张。
这一章，咱们就聊聊怎么把数据稳稳地拿到手。说白了，就三个场景：API拉取、WebSocket实时推送、历史数据回放。
核心观点：数据获取的稳定性，决定了策略的上限。别在数据源上省钱，也别在代码健壮性上偷懒。
4.1 交易所API——最基础的数据获取方式
交易所API，说白了就是HTTP请求。你发一个GET，它返回JSON。就这么简单。
但我个人习惯，从来不用裸的requests去调。为什么？因为限频、重试、签名这些东西，写起来太容易出错了。
我的经验：用现成的SDK，比如ccxt，能省掉80%的坑。但要注意，ccxt的版本更新很快，有些交易所的接口会变，记得锁定版本号。
来看一个典型的API调用示例：
import ccxt
import time
# 初始化交易所
exchange = ccxt.binance({
'apiKey': 'YOUR_API_KEY',
'secret': 'YOUR_SECRET',
'enableRateLimit': True, # 自动限频，这个很重要
})
# 获取现货行情
def fetch_ticker(symbol='BTC/USDT'):
try:
ticker = exchange.fetch_ticker(symbol)
return {
'symbol': ticker['symbol'],
'bid': ticker['bid'],
'ask': ticker['ask'],
'last': ticker['last'],
'timestamp': ticker['timestamp']
}
except Exception as e:
print(f'获取行情失败: {e}')
return None
# 循环拉取
while True:
data = fetch_ticker()
if data:
print(data)
time.sleep(1) # 控制频率
这里有个坑，我曾经踩过。有一次我忘了开enableRateLimit，结果请求太频繁，IP直接被交易所封了半小时。嗯，那半小时我啥也干不了，只能干瞪眼。
警告：API限频不是闹着玩的。每个交易所的规则不一样，有的按IP限，有的按API Key限。建议在代码里加一个本地计数器，双重保险。
4.2 WebSocket实时行情——毫秒级的较量
做套利策略，API拉取的速度根本不够。你想想看，API轮询最快也就100ms一次，但WebSocket是推送的，延迟能到1ms以内。
我个人习惯，所有需要实时响应的策略，一律用WebSocket。API只用来做下单和查询账户。
来看一个WebSocket的订阅示例：
import websocket
import json
def on_message(ws, message):
data = json.loads(message)
# 处理深度数据
if 'bids' in data:
print(f"买一价: {data['bids'][0][0]}, 量: {data['bids'][0][1]}")
print(f"卖一价: {data['asks'][0][0]}, 量: {data['asks'][0][1]}")
def on_error(ws, error):
print(f"WebSocket错误: {error}")
def on_close(ws, close_status_code, close_msg):
print("连接关闭，准备重连...")
def on_open(ws):
# 订阅深度数据
subscribe_msg = {
"method": "SUBSCRIBE",
"params": ["btcusdt@depth20@100ms"],
"id": 1
}
ws.send(json.dumps(subscribe_msg))
print("订阅成功")
# 启动WebSocket
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()
为什么会这样？因为WebSocket一旦断开，数据流就断了。我见过有人写策略，WebSocket断了半小时都没发现，结果策略还在跑，用的全是过期数据。那画面，惨不忍睹。
避坑指南：我曾经在WebSocket重连逻辑上栽过跟头。一定要加心跳检测，比如每30秒发一个ping。如果连续3次ping没收到pong，强制重连。别指望交易所的WebSocket永远不掉线。
4.3 历史数据回放——策略的试金石
历史数据回放，说白了就是模拟过去的行情，看看你的策略能不能赚钱。
我见过很多人，策略在回测里赚得盆满钵满，一上实盘就亏。为什么？因为回放的数据质量不行。
历史数据通常有两种格式：
| 数据类型 | 精度 | 数据量 | 适用场景 |
|---|---|---|---|
| Tick数据 | 毫秒级 | 极大（每天几GB） | 高频套利、做市策略 |
| Bar数据（1分钟/5分钟） | 分钟级 | 中等 | 趋势策略、中低频策略 |
我个人习惯，做套利策略至少要用Tick级别的数据。因为套利机会往往就存在几毫秒之间，用分钟Bar根本看不出效果。
来看一个简单的历史数据回放框架：
import pandas as pd
import time
class DataReplayer:
def __init__(self, csv_path):
self.data = pd.read_csv(csv_path)
self.data['timestamp'] = pd.to_datetime(self.data['timestamp'])
self.index = 0
def next_tick(self):
"""模拟下一个Tick数据"""
if self.index >= len(self.data):
return None
tick = self.data.iloc[self.index]
self.index += 1
# 模拟真实时间间隔
if self.index > 1:
prev_time = self.data.iloc[self.index-2]['timestamp']
curr_time = tick['timestamp']
time_diff = (curr_time - prev_time).total_seconds()
time.sleep(min(time_diff, 0.1)) # 最多等100ms
return tick
def reset(self):
self.index = 0
# 使用示例
replayer = DataReplayer('btc_tick_data.csv')
while True:
tick = replayer.next_tick()
if tick is None:
print("回放结束")
break
# 在这里跑你的策略逻辑
print(f"时间: {tick['timestamp']}, 价格: {tick['price']}, 量: {tick['volume']}")
小技巧：回放的时候，别忘了加一个时间缩放因子。比如你想跑10倍速，就把sleep时间除以10。这样既能快速验证，又能模拟真实的时间节奏。
4.4 三种方式的选型建议
你可能会问，到底该用哪种？我直接给结论：
- 做回测：用历史数据回放，别用API，别用WebSocket。回测要的是确定性，不是实时性。
- 做实盘监控：用WebSocket。延迟低，数据准。
- 做下单和账户查询：用API。WebSocket不适合做交易指令，丢包了你就亏大了。
再次提醒：数据源是策略的基石。我见过有人为了省几块钱，用免费的公共API，结果数据延迟了5秒。套利策略5秒的延迟，黄花菜都凉了。该花的钱，别省。
好了，数据源这块就聊到这儿。记住一句话：数据稳，策略才能稳。下一章咱们聊聊怎么把这些数据组织起来，构建一个高效的数据管道。
公众号:蓝海数据掘金营，微信deep3321