第四章：数据基础设施——实时行情数据源选择、历史数据回放、数据存储方案
做跨交易所套利，说白了就是跟数据打交道。你策略再牛，模型再复杂，数据源拉胯了，一切都是白搭。我见过太多团队，策略回测跑得飞起，一上实盘就崩，十有八九是数据基础设施没搭好。
这一章，咱们就聊聊数据基础设施的三个核心问题：实时行情从哪来？历史数据怎么回放？数据存到哪去？ 嗯，这三个问题搞定了，你的套利系统才算真正有了地基。
4.1 实时行情数据源：选对“水管”很重要
实时行情，就是你的“水管”。水管粗不粗、稳不稳、干不干净，直接决定了你能不能喝到水。
我个人习惯把数据源分成三个梯队：
| 梯队 | 代表 | 延迟 | 成本 | 适用场景 |
|---|---|---|---|---|
| 第一梯队 | 交易所直连（NASDAQ ITCH/OUCH, NYSE PITCH） | < 1ms | 极高（月费+设备+托管） | 高频做市、极速套利 |
| 第二梯队 | 专业数据商（Polygon, Databento, Exegy） | 1-10ms | 中等（按量或月费） | 中频套利、统计套利 |
| 第三梯队 | 免费/聚合源（Yahoo Finance, IEX Cloud, Alpaca） | 100ms+ | 低/免费 | 回测验证、个人学习 |
核心观点： 做跨交易所套利，我建议至少用第二梯队的数据源。第一梯队太贵，第三梯队太慢。你想想看，套利机会转瞬即逝，100ms的延迟，黄花菜都凉了。
我曾经在项目中踩过一个坑：用某个免费API同时拉取NYSE和NASDAQ的行情，结果发现两个交易所的时间戳差了200多毫秒。这哪是套利？这分明是送钱。后来换了Polygon的WebSocket，才把时间同步误差控制在5ms以内。
4.2 历史数据回放：你的“时光机”
历史数据回放，说白了就是让你的策略回到过去，看看它到底行不行。没有回放，你就是在闭着眼睛开车。
我常用的回放方案有两种：
- Tick级回放： 逐笔成交、逐笔订单簿。适合高频策略验证。数据量巨大，一天可能几个GB。
- Bar级回放： 1秒、1分钟、5分钟K线。适合中低频策略。数据量小，回放速度快。
我个人习惯用 pandas + numba 写一个轻量级的回放引擎。核心逻辑很简单：
import pandas as pd
import numpy as np
from numba import jit
@jit(nopython=True)
def replay_tick(timestamps, prices, volumes, window_ms=100):
"""
简单的tick级回放引擎
timestamps: 纳秒时间戳
prices: 价格序列
volumes: 成交量序列
"""
n = len(timestamps)
signals = np.zeros(n)
for i in range(1, n):
# 计算过去window_ms内的价格变化
j = i - 1
while j >= 0 and (timestamps[i] - timestamps[j]) < window_ms * 1_000_000:
j -= 1
if j >= 0:
price_change = (prices[i] - prices[j]) / prices[j]
# 简单的均值回归信号
if abs(price_change) > 0.001:
signals[i] = -np.sign(price_change)
return signals
# 使用示例
df = pd.read_parquet('spy_trades.parquet')
signals = replay_tick(
df['timestamp'].values,
df['price'].values,
df['volume'].values
)小技巧： 回放时一定要考虑“未来信息”问题。我曾经犯过一个低级错误：用当天的收盘价去计算开盘时的信号，结果回测收益翻倍，实盘亏成狗。记住，回放时只能看到“过去”的数据。
4.3 数据存储方案：InfluxDB vs Parquet
数据存哪？怎么存？这问题看似简单，实则暗藏玄机。我见过有人把所有行情数据塞进MySQL，查询一次要等半天。也见过有人用CSV存tick数据，一个文件几个GB，加载一次内存就爆了。
我的建议是：按场景选存储。
4.3.1 InfluxDB：时序数据的“瑞士军刀”
InfluxDB是专门为时序数据设计的。它的优势在于：
- 写入快： 每秒可以写入百万级数据点
- 查询快： 按时间范围查询，秒级响应
- 自动过期： 可以设置数据保留策略，自动清理旧数据
我一般在实盘环境中用InfluxDB存储近期的行情数据（比如最近30天）。这样，策略运行时可以快速查询最近的价格走势。
from influxdb_client import InfluxDBClient, Point
from influxdb_client.client.write_api import SYNCHRONOUS
# 连接到InfluxDB
client = InfluxDBClient(url="http://localhost:8086", token="my-token", org="my-org")
write_api = client.write_api(write_type=SYNCHRONOUS)
# 写入一条tick数据
point = Point("trades") \
.tag("symbol", "SPY") \
.field("price", 450.23) \
.field("volume", 100) \
.time("2024-01-15T09:30:00.123456789Z")
write_api.write(bucket="market_data", record=point)
# 查询最近1分钟的数据
query = '''
from(bucket: "market_data")
|> range(start: -1m)
|> filter(fn: (r) => r["_measurement"] == "trades")
|> filter(fn: (r) => r["symbol"] == "SPY")
'''
result = client.query_api().query(query)4.3.2 Parquet：历史数据的“压缩大师”
Parquet是一种列式存储格式。它的优势在于：
- 压缩率高： 同样的数据，Parquet比CSV小5-10倍
- 读取快： 只读取需要的列，IO开销小
- 生态好： Pandas、Spark、DuckDB都原生支持
我习惯用Parquet存储历史回测数据。比如，我会把每天的tick数据存成一个Parquet文件，按日期分区。这样，回测时只需要加载特定日期的数据，内存占用小，速度快。
import pandas as pd
import pyarrow.parquet as pq
# 写入Parquet文件
df = pd.DataFrame({
'timestamp': pd.date_range('2024-01-15', periods=1000000, freq='1ms'),
'price': np.random.uniform(450, 451, 1000000),
'volume': np.random.randint(1, 1000, 1000000)
})
df.to_parquet('data/spy_trades_20240115.parquet',
compression='snappy',
row_group_size=100000)
# 读取时只加载需要的列
df = pd.read_parquet(
'data/spy_trades_20240115.parquet',
columns=['timestamp', 'price']
)注意： Parquet虽然好，但不要用它做实时写入。每次写入都要重写整个文件，性能很差。我见过有人用Parquet做实时行情存储，结果磁盘IO直接打满。正确的做法是：实时数据用InfluxDB，历史数据用Parquet。
4.4 整体架构：一张图说清楚
说了这么多，咱们来看看整体架构长什么样。下面这张图，是我在实际项目中用过的方案，你可以直接拿来用。
这张图里，数据从交易所或数据商流入，经过清洗、对齐、重建，然后分流到不同的存储系统。实时数据进InfluxDB，历史数据进Parquet，热点数据进Redis做缓存。最后，策略引擎和回测系统各取所需。
总结一下： 数据基础设施没有银弹。我的经验是，实时用InfluxDB，历史用Parquet，中间加一层Redis做缓冲。这套组合拳，我用了三年，没出过大问题。
嗯，数据基础设施就聊这么多。记住，数据是套利系统的血液，基础设施就是血管。血管堵了，再牛的策略也白搭。