数据采集与存储：WebSocket实时订阅、REST API历史数据拉取、Parquet与HDF5存储格式选择

做市商这个行当，说白了就是跟时间赛跑。

你比别人快1毫秒，可能就多赚一笔。你比别人慢1毫秒，可能就被对手盘吃掉。所以数据采集和存储，是整个策略的命门。我见过太多团队，策略模型写得漂漂亮亮，结果数据链路一塌糊涂，实盘跑起来直接崩掉。

今天咱们就把这块硬骨头啃下来。

一、实时数据订阅：WebSocket才是王道

做市商需要实时盯盘口。盘口数据变化极快，每秒可能几百次甚至上千次。用REST API轮询？别闹了，那玩意儿延迟太高，而且容易被交易所限流。

我个人习惯，所有实时数据都用WebSocket。它建立一次连接，服务端主动推送，延迟通常在1-10毫秒以内。

举个例子，币安的WebSocket深度流：

import websocket
import json

def on_message(ws, message):
    data = json.loads(message)
    # 处理增量数据
    bids = data['b']  # 买单
    asks = data['a']  # 卖单
    # 更新本地订单簿
    update_orderbook(bids, asks)

def on_error(ws, error):
    print(f"连接异常: {error}")
    # 自动重连逻辑
    reconnect()

ws = websocket.WebSocketApp(
    "wss://stream.binance.com:9443/ws/btcusdt@depth20@100ms",
    on_message=on_message,
    on_error=on_error
)
ws.run_forever()

这里有个坑，我曾经踩过。交易所的WebSocket偶尔会断连，或者推送的数据包顺序错乱。你想想看，如果本地订单簿更新顺序错了，整个策略就废了。

二、历史数据拉取：REST API的优雅用法

实时数据搞定了，但策略回测、模型训练都需要历史数据。这时候REST API就派上用场了。

REST API的特点是「请求-响应」模式。你发一个请求，它返回一段数据。适合拉取过去几小时、几天甚至几个月的数据。

我一般这样设计拉取逻辑：

1. 确定时间范围：比如从2024-01-01到2024-06-30
2. 分片拉取：每次只拉取1小时的数据，避免请求超时
3. 并发控制：交易所通常有频率限制，比如每秒最多10次请求
4. 断点续传：记录上次拉取到的位置，中断后可以继续

import requests
import time
from datetime import datetime, timedelta

def fetch_klines(symbol, start_time, end_time):
    url = "https://api.binance.com/api/v3/klines"
    all_data = []
    current = start_time
    
    while current < end_time:
        params = {
            'symbol': symbol,
            'interval': '1m',
            'startTime': int(current.timestamp() * 1000),
            'limit': 1000
        }
        resp = requests.get(url, params=params)
        data = resp.json()
        
        if not data:
            break
            
        all_data.extend(data)
        # 更新当前时间到最后一根K线的时间
        current = datetime.fromtimestamp(data[-1][0] / 1000) + timedelta(minutes=1)
        time.sleep(0.1)  # 控制请求频率
    
    return all_data

三、存储格式选择：Parquet vs HDF5

数据拉下来了，存哪儿？怎么存？

做市商的数据量巨大。一个币种，一天就有1440根1分钟K线，加上盘口快照，轻松上百万条记录。存成CSV？那玩意儿又慢又占空间。

我主要用两种格式：Parquet 和 HDF5。它们都是列式存储，压缩率高，读取快。

特性	Parquet	HDF5
压缩率	极高（Snappy/ZSTD）	高（GZIP/LZF）
读取速度	快（列裁剪、谓词下推）	快（支持分块读取）
生态兼容	Pandas、Spark、Dask	Pandas、NumPy、PyTorch
适用场景	大规模数据分析、数据湖	科学计算、深度学习
元数据管理	弱（依赖外部目录）	强（内置层级结构）

我个人习惯：

• 盘口快照数据：用Parquet。因为盘口数据字段多（几十档买卖单），Parquet的列裁剪能只读取需要的列，节省大量IO。
• K线数据：用HDF5。因为K线数据通常需要按时间序列分析，HDF5支持高效的切片和索引。
• 策略回测结果：用Parquet。方便后续用Spark或Dask做大规模回测。

四、知识体系总览

说了这么多，咱们用一张图把整个数据采集与存储的脉络理清楚。

这张图把整个流程串起来了。从交易所数据源出发，实时数据走WebSocket，历史数据走REST API。采集到的盘口数据和K线数据，分别用Parquet和HDF5存储。最终服务于策略回测、模型训练和实盘监控。

---

公众号:蓝海资料掘金营，微信deep3321