2. 数据采集层设计：实时行情数据源接入与存储
数据采集层，说白了就是量化系统的「耳朵」和「眼睛」。
你想想看，没有数据，再牛的策略也是白搭。我做了这么多年量化架构，见过太多团队在策略上花了大把精力，结果数据层一塌糊涂——行情延迟高、数据对不齐、回测和实盘结果天差地别。嗯，这些坑我都踩过。
2.1 实时行情数据源接入
实时行情接入，核心就两个路子：交易所API和WebSocket。我个人习惯把前者叫「拉模式」，后者叫「推模式」。
2.1.1 交易所API（RESTful）
REST API适合做历史数据补录、账户查询这类低频操作。但你要是用它做实时行情，那就等着被市场甩下车吧。为什么？因为每次请求都要建立HTTP连接，延迟通常在几十到几百毫秒。高频交易里，这时间够价格跳好几个档位了。
2.1.2 WebSocket（推荐方案）
WebSocket才是实时行情的正确姿势。它建立一次连接，服务端主动推送数据给你。延迟能压到1-5毫秒，这才是做量化该有的样子。
这里贴一段我常用的WebSocket行情接入代码框架：
import websocket
import json
class MarketDataFeed:
def __init__(self, url, symbols):
self.url = url
self.symbols = symbols
self.ws = None
def on_message(self, ws, message):
# 解析Tick数据
tick = json.loads(message)
# 这里做数据清洗和对齐
self.process_tick(tick)
def on_error(self, ws, error):
# 我曾经因为没处理断线重连，导致一整个交易时段数据全丢了
print(f"WebSocket error: {error}")
self.reconnect()
def on_close(self, ws, close_status_code, close_msg):
print("Connection closed, reconnecting...")
self.reconnect()
def on_open(self, ws):
# 订阅行情
subscribe_msg = {
"type": "subscribe",
"symbols": self.symbols
}
ws.send(json.dumps(subscribe_msg))
def connect(self):
self.ws = websocket.WebSocketApp(
self.url,
on_open=self.on_open,
on_message=self.on_message,
on_error=self.on_error,
on_close=self.on_close
)
self.ws.run_forever()
def reconnect(self):
# 指数退避重连策略
import time
for i in range(5):
try:
time.sleep(2 ** i)
self.connect()
break
except:
continue2.2 历史数据存储方案
数据存哪里？这问题我纠结了很久。InfluxDB和ClickHouse，我两个都用过，各有千秋。
2.2.1 InfluxDB：时序数据库的老牌选手
InfluxDB天生为时序数据而生。写入快、查询简单，尤其适合做实时监控和短周期分析。
| 特性 | InfluxDB | ClickHouse |
|---|---|---|
| 写入速度 | 极快（百万点/秒） | 快（数十万行/秒） |
| 查询延迟 | 毫秒级（短时间范围） | 毫秒级（大规模数据） |
| 存储压缩比 | 3:1 ~ 5:1 | 5:1 ~ 10:1 |
| SQL兼容性 | 类SQL（Flux查询） | 标准SQL |
| 适用场景 | 实时监控、短周期分析 | 大规模历史分析、复杂聚合 |
我个人习惯用InfluxDB做「热数据」存储——比如最近7天的Tick数据。查询快，写入也扛得住。
2.2.2 ClickHouse：分析型数据库的王者
ClickHouse是真正的「大数据」选手。如果你要存几年的Tick数据，InfluxDB会变得很慢，但ClickHouse依然游刃有余。
为什么？因为ClickHouse的列式存储和向量化执行引擎，对大规模聚合查询简直是降维打击。
-- ClickHouse建表示例：存储Tick级行情
CREATE TABLE tick_data (
symbol String,
timestamp DateTime64(3),
price Float64,
volume Float64,
bid_price Float64,
ask_price Float64,
bid_volume Float64,
ask_volume Float64
) ENGINE = MergeTree()
PARTITION BY toYYYYMM(timestamp)
ORDER BY (symbol, timestamp);
-- 查询某只股票某天的最高价
SELECT symbol, max(price) as high
FROM tick_data
WHERE symbol = 'BTCUSDT'
AND timestamp >= '2024-01-01'
AND timestamp < '2024-01-02'
GROUP BY symbol;2.3 数据清洗与对齐
数据清洗，这活儿看着简单，做起来全是坑。我刚开始做量化时，以为拿到交易所的原始数据就能直接用。结果回测跑出来收益惊人，实盘一跑就亏——后来才发现是数据没对齐。
2.3.1 Tick级数据去重
为什么会有重复数据？两个原因：
- 网络重传： WebSocket断线重连后，可能会收到重复的Tick
- 多源接入： 同时接入多个交易所，同一时刻的行情可能重复
去重策略其实不复杂，核心就是「时间戳+价格+成交量」作为唯一键。
import pandas as pd
def deduplicate_ticks(ticks_df):
"""
Tick级去重
我曾经因为没做去重，导致回测里同一个Tick被算了两次成交量
结果策略的滑点模型完全不准
"""
# 按时间戳和价格去重
deduped = ticks_df.drop_duplicates(
subset=['timestamp', 'price', 'volume'],
keep='first'
)
# 按时间戳排序
deduped = deduped.sort_values('timestamp')
# 检查时间戳是否严格递增
if not deduped['timestamp'].is_monotonic_increasing:
# 处理乱序数据
deduped = deduped.sort_values('timestamp').reset_index(drop=True)
return deduped2.3.2 数据对齐：不同时间粒度的统一
不同交易所的Tick时间戳精度不一样。有的精确到毫秒，有的精确到微秒。如果你不做对齐，回测和实盘的数据口径就不一致。
我常用的做法是：统一将时间戳对齐到毫秒级，然后做「时间桶」聚合。
def align_ticks_to_millisecond(ticks_df):
"""
将Tick数据对齐到毫秒级时间桶
"""
# 截断到毫秒
ticks_df['aligned_ts'] = ticks_df['timestamp'].dt.floor('ms')
# 按时间桶聚合
aligned = ticks_df.groupby(['symbol', 'aligned_ts']).agg({
'price': 'last', # 取最后一笔价格
'volume': 'sum', # 成交量累加
'bid_price': 'last',
'ask_price': 'last'
}).reset_index()
return aligned2.4 数据采集层的整体架构
说了这么多，咱们用一张图把整个数据采集层的架构串起来：
这张图把整个数据采集层的流程串起来了。从数据源接入，到清洗对齐，再到分层存储，每一步都有讲究。
公众号:蓝海资料掘金营，微信deep3321