第一章：数据获取与存储——WebSocket实时数据流接入、REST API历史数据获取、Tick数据的高效存储方案
做套利策略，说白了就是跟时间赛跑。你比别人快0.1秒，可能就多赚几个点。但这一切的前提是什么？数据。没有干净、及时、完整的数据，再牛的策略也是空中楼阁。
我个人习惯把数据这块分成三个环节：实时接入、历史回补、高效存储。今天咱们就一个一个拆开聊。
1.1 WebSocket实时数据流接入
为什么用WebSocket而不是轮询？你想想看，套利机会往往就出现在价格跳变的瞬间。如果用REST API每秒轮询一次，等你拿到数据，黄花菜都凉了。
WebSocket是全双工通信，交易所一有Tick变动，立马推送到你本地。延迟通常在10-50毫秒级别，比轮询快一个数量级。
核心要点：WebSocket连接建立后，需要维护心跳。交易所一般要求每30秒发一次ping，否则会断开连接。我在项目中遇到过因为心跳超时导致数据断流，回测时发现少了一堆Tick，排查了半天才发现是这个问题。
下面是一个简单的WebSocket接入示例，以币安为例：
import websocket
import json
def on_message(ws, message):
data = json.loads(message)
# 解析Tick数据
tick = {
'symbol': data['s'],
'price': float(data['c']),
'volume': float(data['v']),
'timestamp': data['E']
}
print(f"收到Tick: {tick}")
def on_error(ws, error):
print(f"WebSocket错误: {error}")
def on_close(ws, close_status_code, close_msg):
print("WebSocket连接关闭")
def on_open(ws):
print("WebSocket连接已建立")
# 订阅BTC/USDT的实时行情
subscribe_msg = {
"method": "SUBSCRIBE",
"params": ["btcusdt@ticker"],
"id": 1
}
ws.send(json.dumps(subscribe_msg))
# 启动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连接，结果订阅了20多个交易对，数据量一大直接卡死。后来改成每个交易对单独开一个连接，或者用连接池管理，问题就解决了。建议一个连接不要订阅超过10个交易对。
1.2 REST API历史数据获取
实时数据有了，但回测怎么办？你需要历史数据。REST API就是干这个的。
大部分交易所都提供历史K线或Tick数据接口。但要注意，历史Tick数据通常只保留最近几天，更早的数据需要付费或者自己攒。
我一般这样设计数据获取流程：
- 首次运行：用REST API拉取最近7天的历史数据
- 日常运行：WebSocket实时接收，同时每5分钟用REST API做一次数据校验
- 数据补全：如果发现缺失，从REST API补拉
下面是一个获取历史K线的示例：
import requests
import pandas as pd
def fetch_historical_klines(symbol, interval, limit=1000):
url = "https://api.binance.com/api/v3/klines"
params = {
'symbol': symbol,
'interval': interval,
'limit': limit
}
response = requests.get(url, params=params)
data = response.json()
# 转换为DataFrame
df = pd.DataFrame(data, columns=[
'timestamp', 'open', 'high', 'low', 'close', 'volume',
'close_time', 'quote_asset_volume', 'number_of_trades',
'taker_buy_base_vol', 'taker_buy_quote_vol', 'ignore'
])
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
return df
# 获取BTC/USDT最近1000根1分钟K线
df = fetch_historical_klines('BTCUSDT', '1m', 1000)
print(df.head())注意：REST API有频率限制。币安的限制是每分钟1200次请求。我曾经一次性拉取100个交易对的历史数据，结果被限流了半小时。建议加个请求队列，控制请求频率在每秒20次以内。
1.3 Tick数据的高效存储方案
数据拿到了，存哪儿？这是个大学问。
Tick数据的特点是：高频、海量、时间序列。一个交易对一天就能产生几十万条Tick，多个交易对加起来轻松上亿。普通的关系型数据库（比如MySQL）根本扛不住。
我推荐两种方案：
| 方案 | 适用场景 | 优点 | 缺点 |
|---|---|---|---|
| InfluxDB | 实时查询、监控 | 写入快、查询快、自带时间序列优化 | 存储成本高、不适合复杂分析 |
| Parquet | 历史回测、数据分析 | 压缩率高、列式存储、适合批量处理 | 不支持实时写入 |
我个人习惯是：实时数据用InfluxDB，历史数据转存Parquet。这样既保证了实时查询的速度，又节省了存储成本。
InfluxDB写入示例
from influxdb import InfluxDBClient
import json
client = InfluxDBClient(host='localhost', port=8086)
client.create_database('tick_data')
def write_tick_to_influx(tick):
json_body = [
{
"measurement": "tick",
"tags": {
"symbol": tick['symbol']
},
"time": tick['timestamp'],
"fields": {
"price": tick['price'],
"volume": tick['volume']
}
}
]
client.write_points(json_body)
# 示例：写入一条Tick
tick = {'symbol': 'BTCUSDT', 'price': 50000.0, 'volume': 1.5, 'timestamp': '2024-01-01T00:00:00Z'}
write_tick_to_influx(tick)Parquet存储示例
import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq
def save_tick_to_parquet(tick_list, file_path):
df = pd.DataFrame(tick_list)
table = pa.Table.from_pandas(df)
pq.write_table(table, file_path, compression='snappy')
# 示例：保存1000条Tick到Parquet文件
tick_list = [
{'symbol': 'BTCUSDT', 'price': 50000.0, 'volume': 1.5, 'timestamp': '2024-01-01T00:00:00Z'}
# ... 更多数据
]
save_tick_to_parquet(tick_list, 'btc_tick_20240101.parquet')经验之谈：Parquet文件建议按日期分区存储，比如 data/2024/01/01/btc_tick.parquet。这样查询某一天的数据时，只需要加载对应的文件，效率高很多。我刚开始把所有数据塞到一个文件里，结果回测时加载一次要等5分钟，后来改成按天分区，加载时间降到了10秒以内。
1.4 整体架构图
下面这张图展示了数据获取与存储的整体流程。你可以看到，WebSocket负责实时数据，REST API负责历史数据，两者最终汇入InfluxDB和Parquet。
嗯，到这里数据获取与存储的核心内容就讲完了。记住一句话：数据质量决定策略上限。WebSocket保证实时性，REST API保证完整性，InfluxDB和Parquet保证存储效率。这三板斧用好了，你的套利策略就成功了一半。
公众号:蓝海资料掘金营，微信deep3321