4. 数据获取与存储

高频回测的起点是可靠、低延迟的行情数据。本章聚焦两个核心环节：从交易所API实时获取Tick数据与使用列式存储格式高效持久化。错误的获取策略会导致数据缺失或延迟失真；低效的存储格式则会让回测I/O成为瓶颈。

4.1 交易所API实时行情获取架构

高频行情获取需要同时处理多品种、高吞吐的增量数据。典型的架构采用WebSocket长连接接收推送，配合REST API做补录与校验。

4.2 WebSocket实时Tick数据获取实战

以币安(Binance)现货行情为例，使用websockets库订阅深度快照与增量数据。核心步骤包括：

1. 建立连接：通过wss://stream.binance.com:9443/ws建立WebSocket连接。
2. 订阅主题：发送JSON订阅消息，如{"method":"SUBSCRIBE","params":["btcusdt@depth20@100ms"],"id":1}。
3. 异步接收：使用asyncio事件循环持续接收消息，避免阻塞。
4. 解析与校验：提取bids/asks数组、更新时间戳，校验数据连续性。
5. 数据分发：将解析后的Tick对象送入内存队列，供回测引擎或存储模块消费。

4.3 Tick数据存储格式对比

传统CSV或JSON格式在高频场景下存在严重的I/O瓶颈。Parquet与Feather作为列式存储格式，专为分析型工作负载优化。下表对比了三种格式在Tick数据场景下的关键差异：

4.4 使用Parquet/Feather存储Tick数据

推荐策略：实时写入使用Feather（低延迟、零拷贝），长期归档使用Parquet（高压缩、跨平台兼容）。以下代码演示了两种格式的写入与读取：

import pandas as pd
import pyarrow.feather as feather
import pyarrow.parquet as pq
import pyarrow as pa
from datetime import datetime

# 模拟Tick数据
tick_data = pd.DataFrame({
    'symbol': ['BTCUSDT'] * 1000,
    'timestamp': pd.date_range('2025-01-01', periods=1000, freq='100ms'),
    'bid_price': [50000.0 + i * 0.1 for i in range(1000)],
    'ask_price': [50001.0 + i * 0.1 for i in range(1000)],
    'bid_volume': [0.5 + i * 0.01 for i in range(1000)],
    'ask_volume': [0.6 + i * 0.01 for i in range(1000)]
})

# 写入Feather（适合实时追加）
feather.write_feather(tick_data, 'tick_20250101.feather')
print(f"Feather写入完成，文件大小: {os.path.getsize('tick_20250101.feather') / 1024:.2f} KB")

# 写入Parquet（适合归档，启用Snappy压缩）
table = pa.Table.from_pandas(tick_data)
pq.write_table(table, 'tick_20250101.parquet', compression='snappy')
print(f"Parquet写入完成，文件大小: {os.path.getsize('tick_20250101.parquet') / 1024:.2f} KB")

# 读取Feather（极速加载）
df_feather = feather.read_feather('tick_20250101.feather')
print(f"Feather读取耗时: {timeit.timeit(lambda: feather.read_feather('tick_20250101.feather'), number=10)/10:.4f}秒")

# 读取Parquet（支持列裁剪）
df_parquet = pq.read_table('tick_20250101.parquet', columns=['timestamp', 'bid_price']).to_pandas()
print(f"Parquet列裁剪读取耗时: {timeit.timeit(lambda: pq.read_table('tick_20250101.parquet', columns=['timestamp', 'bid_price']), number=10)/10:.4f}秒")

4.5 数据存储与回测流水线集成

将数据获取、存储与回测引擎串联，形成完整的离线回测流水线。下图展示了数据从交易所到回测结果的全链路：

4.6 数据质量校验与异常处理

高频数据易受网络抖动、交易所限流、时钟偏差等因素影响。在存储前必须执行以下校验：

• 时间戳单调性检查：确保每个品种的Tick时间戳严格递增，出现回退时标记并丢弃。
• 价格合理性校验：过滤超出合理价格区间的异常Tick（如价格突变超过阈值）。
• 数据连续性检测：检查相邻Tick时间间隔是否超过预设阈值（如1秒），触发补录流程。
• 重复数据去重：基于交易所消息ID或本地哈希去重，避免重复写入。