2、数据源与采集：主流Tick数据提供商对比（交易所、第三方）、数据采集架构设计、实时采集与历史数据回放
做Tick级回测，第一步就是搞定数据。没有高质量的数据，后面所有策略都是空中楼阁。我这些年踩过的坑，有一半都跟数据源有关。今天咱们就把这块掰开揉碎了讲清楚。
2.1 主流Tick数据提供商对比
先说说数据从哪来。市面上主要有两类：交易所直连和第三方聚合。我个人的经验是，没有绝对的好坏，关键看你的场景。
| 提供商 | 数据类型 | 延迟 | 成本 | 适用场景 |
|---|---|---|---|---|
| 交易所直连（如CTP、XTP） | 原始Tick | 微秒级 | 高（需席位费） | 高频交易、做市 |
| 第三方（如Wind、聚宽、Tushare） | 清洗后Tick | 秒级 | 中低 | 策略回测、中低频 |
| 开源数据（如Quandl、Binance API） | 部分Tick | 分钟级 | 免费 | 学习、原型验证 |
核心观点：如果你做的是高频策略，别犹豫，直接上交易所直连。如果是普通回测，第三方数据完全够用。我曾经为了省成本用免费数据做回测，结果策略在实盘上完全失效——数据精度差太多了。
2.2 数据采集架构设计
数据采集架构，说白了就是怎么把数据从源头搬到你的存储里。我建议分三层设计：采集层、缓冲层、存储层。
我的建议：缓冲层千万别省。我早期做采集时直接往数据库写，结果网络抖动导致数据丢失，回测结果全偏了。加上Kafka之后，数据可靠性提升了一个量级。
2.3 实时采集与历史数据回放
实时采集和历史回放，其实是同一套架构的两个模式。你想想看，回测本质上就是「把历史数据当成实时数据来喂」。我习惯用同一个接口处理这两种场景。
实时采集核心逻辑
实时采集的关键是低延迟和可靠性。我一般用WebSocket连接交易所，然后异步写入缓冲层。下面是一个简化版的采集代码：
import asyncio
import websockets
import json
from kafka import KafkaProducer
class TickCollector:
def __init__(self, symbol, exchange_url):
self.symbol = symbol
self.url = exchange_url
self.producer = KafkaProducer(
bootstrap_servers=['localhost:9092'],
value_serializer=lambda v: json.dumps(v).encode('utf-8')
)
async def collect(self):
async with websockets.connect(self.url) as ws:
while True:
data = await ws.recv()
tick = json.loads(data)
# 数据校验
if self.validate(tick):
self.producer.send('ticks', tick)
def validate(self, tick):
# 检查价格、数量是否合理
if tick['price'] <= 0 or tick['volume'] <= 0:
return False
return True注意：实时采集最怕的就是断连。我曾经在生产环境遇到过交易所WebSocket断开后自动重连，但重连期间的数据全丢了。一定要实现断点续传机制。
历史数据回放设计
历史回放的核心是「时间控制」。你不能一股脑把数据全塞进去，得按真实的时间间隔来。我常用的做法是：
- 数据加载：从存储层按时间顺序读取Tick数据
- 时间戳对齐：用系统时间或模拟时间控制回放速度
- 事件驱动：每个Tick触发一次策略计算
import time
from datetime import datetime
class TickReplayer:
def __init__(self, data_path, speed=1.0):
self.data = self.load_data(data_path)
self.speed = speed # 回放速度倍率
def replay(self):
prev_time = None
for tick in self.data:
current_time = datetime.fromtimestamp(tick['timestamp'])
if prev_time:
# 计算真实时间间隔
interval = (current_time - prev_time).total_seconds()
# 按速度倍率等待
time.sleep(interval / self.speed)
# 触发策略
self.on_tick(tick)
prev_time = current_time
def on_tick(self, tick):
# 策略逻辑
pass关键点：回放速度倍率很重要。调试时用10倍速快速验证，正式回测用1倍速保证精度。我一般会加一个「暂停/继续」功能，方便排查问题。
2.4 数据质量保障
数据质量是回测的生命线。我总结了几条必须做的检查：
- 缺失值检测：检查时间戳是否连续，有没有跳秒
- 异常值过滤：价格突变超过阈值要标记
- 重复数据去重：同一个Tick出现两次要剔除
- 时间戳对齐：不同数据源的时间戳要统一到同一时区
避坑指南：我曾经遇到过第三方数据的时间戳是本地时间，而交易所用的是UTC。回测时没注意，结果策略在开盘前就触发了交易。从那以后，我所有数据都强制转成UTC+0存储。
嗯，数据源和采集这块就讲这么多。记住一句话：数据质量决定回测可信度。下一节咱们聊聊数据存储的具体方案，包括怎么用Parquet和InfluxDB高效存储Tick数据。