数据源接入：交易所API协议、WebSocket实时流、CSV/Parquet历史数据加载

做高频交易，第一件事不是写策略，而是搞定数据。我见过太多人花几个月写了个漂亮的策略回测，结果一上实盘就崩——为什么？数据源没接对。今天咱们就把这块地基打牢。

交易所API协议：REST vs WebSocket

交易所给咱们提供了两种主要的数据获取方式。说白了，一个像你去图书馆借书（REST），一个像订报纸每天送到家（WebSocket）。

特性	REST API	WebSocket
请求方式	HTTP请求-响应	长连接推送
延迟	50-200ms	1-10ms
适用场景	历史数据、下单	实时行情、订单簿
带宽消耗	低	高（持续连接）

我个人习惯是这样：启动时先用REST拉一波快照，然后切到WebSocket收增量更新。这样既保证了初始数据的完整性，又不会错过实时变化。

WebSocket实时流：心跳、重连、序列号

WebSocket看着简单，坑可不少。我曾经在实盘中被坑过一次——交易所WebSocket断开了，我这边没发现，结果订单簿数据停了整整30秒。嗯，那30秒的行情波动，够我喝一壶的。

这里有几个关键点：

• 心跳机制：每隔15-30秒发一个ping，收不到pong就准备重连
• 序列号校验：每条消息带一个递增的seq，发现跳号说明丢数据了
• 快照+增量：每N条增量后，主动请求一次全量快照做校验

代码示例（Python伪代码）：

import websocket
import json

class OrderBookStream:
    def __init__(self, exchange_url):
        self.ws = None
        self.seq = 0
        self.snapshot = None
        
    def on_message(self, ws, message):
        data = json.loads(message)
        # 校验序列号
        if data['seq'] != self.seq + 1:
            self.request_snapshot()  # 发现跳号，重新拉快照
        self.seq = data['seq']
        # 更新订单簿
        self.update_orderbook(data)
        
    def on_error(self, ws, error):
        print(f"连接异常: {error}")
        self.reconnect_with_delay()  # 延迟1-2秒重连

CSV/Parquet历史数据加载

回测时咱们需要历史数据。CSV是最常见的格式，但Parquet才是高频交易的首选。为什么？

• CSV：文本格式，体积大，加载慢。100GB的CSV，光解析就要半小时
• Parquet：列式存储，压缩率高，加载快。同样数据，Parquet只有CSV的1/5大小

你想想看，回测一次要加载100GB数据，CSV得等半小时，Parquet只要5分钟。这差距，在迭代策略时就是天壤之别。

加载代码示例：

import pandas as pd
import pyarrow.parquet as pq

# CSV加载（慢）
df_csv = pd.read_csv('orderbook_2024.csv', parse_dates=['timestamp'])

# Parquet加载（快）
df_pq = pq.read_table('orderbook_2024.parquet').to_pandas()

# 我建议：数据清洗后统一存Parquet
# 原始数据保留CSV作为备份

数据源接入的整体流程

咱们把上面这些串起来，看看一个完整的数据接入流程长什么样：

这个流程我用了好几年，基本没出过大问题。核心思路就一条：分层解耦。数据源只管收数据，清洗模块只管处理，存储模块只管存。各干各的，出了问题也好排查。

好了，数据源接入这块就聊到这儿。记住：REST拿历史，WebSocket接实时，Parquet存归档。这三板斧用好了，后面的策略开发就顺了。