2. 市场数据获取：对接交易所API、WebSocket实时行情、RESTful历史数据
做套利交易，第一步就是搞定数据。没有数据，策略就是空中楼阁。
我个人习惯把数据获取分成三个层次：实时行情、历史数据、账户信息。每个层次都有不同的技术选型。今天咱们就聊聊怎么把这些数据从交易所里「掏」出来。
2.1 交易所API选型：选对接口少踩坑
市面上交易所很多，但API设计思路大同小异。我建议你重点关注三个维度：
- 文档质量：接口文档写得清不清楚？有没有示例代码？
- 限频策略：每秒能请求多少次？超过限制会怎样？
- 数据精度：价格、数量的小数位数是否满足套利需求？
我在项目中遇到过最坑的事：某交易所文档写着「支持WebSocket」，结果连上后每30秒断一次。后来发现是他们负载均衡配置有问题。所以，选接口前一定要做压力测试。
2.2 RESTful API：拉取历史数据
RESTful API适合获取历史K线、订单簿快照这类「一次性」数据。说白了就是：你问一次，它答一次。
以币安为例，获取BTC/USDT的1小时K线：
import requests
import time
def fetch_klines(symbol, interval, limit=500):
url = "https://api.binance.com/api/v3/klines"
params = {
"symbol": symbol,
"interval": interval,
"limit": limit
}
response = requests.get(url, params=params)
data = response.json()
# 解析数据：时间戳、开、高、低、收、成交量...
klines = []
for item in data:
kline = {
"timestamp": item[0],
"open": float(item[1]),
"high": float(item[2]),
"low": float(item[3]),
"close": float(item[4]),
"volume": float(item[5])
}
klines.append(kline)
return klines
# 调用示例
klines = fetch_klines("BTCUSDT", "1h", 100)
print(f"获取到 {len(klines)} 条K线数据")
你想想看，这里有个细节：limit参数最大能设多少？币安是1000条。如果你需要更多历史数据，就得用startTime和endTime参数分批拉取。
2.3 WebSocket：实时行情推送
套利交易对时效性要求极高。RESTful轮询太慢了，WebSocket才是王道。
WebSocket建立连接后，交易所会主动推送数据给你。你只需要「躺平」接收就行。
import websocket
import json
def on_message(ws, message):
data = json.loads(message)
# 解析深度数据
if 'b' in data: # 买单
bids = data['b']
print(f"最新买单：{bids[0]}")
if 'a' in data: # 卖单
asks = data['a']
print(f"最新卖单：{asks[0]}")
def on_error(ws, error):
print(f"WebSocket错误：{error}")
def on_close(ws, close_status_code, close_msg):
print("WebSocket连接关闭")
def on_open(ws):
# 订阅BTC/USDT的深度数据
subscribe_msg = {
"method": "SUBSCRIBE",
"params": ["btcusdt@depth20@100ms"],
"id": 1
}
ws.send(json.dumps(subscribe_msg))
print("订阅成功")
# 启动WebSocket
ws_url = "wss://stream.binance.com:9443/ws"
ws = websocket.WebSocketApp(ws_url,
on_open=on_open,
on_message=on_message,
on_error=on_error,
on_close=on_close)
ws.run_forever()
嗯，这里要注意：WebSocket断线重连是必须处理的。我见过太多人因为网络抖动导致数据中断，策略还在傻傻地跑。结果呢？套利机会没抓到，反而亏了手续费。
2.4 数据清洗与对齐
数据拿到手了，但能用吗？不一定。
不同交易所的数据格式、时间戳、精度都不一样。比如币安的时间戳是毫秒级，而OKX是微秒级。你得把它们统一成标准格式。
| 交易所 | 时间戳格式 | 价格精度 | 数量精度 |
|---|---|---|---|
| 币安 | 毫秒 | 2位小数 | 6位小数 |
| OKX | 微秒 | 4位小数 | 8位小数 |
| Bybit | 毫秒 | 2位小数 | 4位小数 |
我建议你写一个统一的数据适配层。不管从哪个交易所拿数据，都转成统一的内部格式。这样策略代码就不用关心底层数据源了。
class UnifiedKline:
def __init__(self, exchange, symbol):
self.exchange = exchange
self.symbol = symbol
self.timestamp = None
self.open = 0.0
self.high = 0.0
self.low = 0.0
self.close = 0.0
self.volume = 0.0
@staticmethod
def from_binance(data):
kline = UnifiedKline("binance", data["symbol"])
kline.timestamp = data[0] # 毫秒
kline.open = float(data[1])
kline.high = float(data[2])
kline.low = float(data[3])
kline.close = float(data[4])
kline.volume = float(data[5])
return kline
@staticmethod
def from_okx(data):
kline = UnifiedKline("okx", data["symbol"])
kline.timestamp = data[0] // 1000 # 微秒转毫秒
kline.open = float(data[1])
kline.high = float(data[2])
kline.low = float(data[3])
kline.close = float(data[4])
kline.volume = float(data[5])
return kline
2.5 数据存储方案
数据存哪里？这是个好问题。
我个人习惯分层存储：
- 热数据（最近1小时）：存内存里，用Redis或者Python的deque。读写快，适合实时计算。
- 温数据（最近7天）：存SQLite或者PostgreSQL。方便回测和分析。
- 冷数据（历史数据）：存CSV文件或者Parquet格式。压缩率高，适合长期归档。
你想想看，如果所有数据都放内存，程序跑几天就爆了。如果都放数据库，实时查询又太慢。分层存储是最优解。
2.6 数据质量监控
数据出问题，策略必死。我吃过这个亏。
有一次，交易所的WebSocket推送了重复数据，我的策略以为出现了套利机会，连续开了10个仓位。结果呢？全是假信号，亏了一笔。
从那以后，我加了三道防线：
- 时间戳校验：如果新数据的时间戳比上一条还早，直接丢弃。
- 价格合理性检查：如果价格波动超过5%，触发告警，人工确认。
- 数据完整性统计：每分钟统计一次接收到的数据条数，如果低于阈值，说明连接可能断了。
def validate_tick(tick, last_tick):
# 时间戳校验
if tick['timestamp'] <= last_tick['timestamp']:
return False
# 价格合理性检查
price_change = abs(tick['price'] - last_tick['price']) / last_tick['price']
if price_change > 0.05: # 超过5%
print(f"警告：价格异常波动 {price_change*100:.2f}%")
return False
return True
2.7 本章知识体系
下面这张图展示了市场数据获取的整体架构。你可以看到数据从交易所到策略的完整流转路径。
从这张图你能看到，数据从交易所出发，经过不同的获取方式，再到清洗对齐，最后分层存储。每一步都有坑，每一步都需要精心设计。
好了，关于市场数据获取就聊这么多。记住一句话：数据质量决定策略上限。把数据搞扎实了，后面的套利逻辑才能跑得稳。
公众号:蓝海资料掘金营，微信deep3321