4、数据采集模块：多源数据接入
数据采集，说白了就是给交易系统装上一双「千里眼」。没有数据，一切策略都是空中楼阁。我在做套利系统时，第一个踩的坑就是数据源单一——某个交易所突然断流，整个系统直接瘫痪。从那以后，我对数据采集模块的要求就一个字：稳。
4.1 多源数据接入：别把鸡蛋放一个篮子里
套利交易对数据实时性要求极高。你想想看，两个交易所之间的价差可能只存在几十毫秒。如果数据源挂了，或者延迟高了，策略就废了。
我个人习惯接入三类数据源：
- 交易所官方API：比如Binance、OKX、Coinbase的WebSocket接口。这是最直接、延迟最低的方式。
- 第三方数据商：像彭博、路透、或者一些加密数据聚合商。它们提供清洗后的数据，但延迟会高一些。
- 自建节点：对于区块链数据，我建议跑自己的全节点。我在项目中遇到过，第三方节点在行情剧烈波动时经常限流。
核心原则：至少两个独立数据源互为备份。主数据源断线时，自动切换备用源，切换时间控制在100ms以内。
4.2 数据清洗与对齐：脏数据比没数据更可怕
数据进来了，但能用吗？不一定。交易所返回的数据经常有各种问题：
- 重复推送（同一个tick发了两遍）
- 时间戳错乱（服务器时钟不同步）
- 缺失字段（某些交易所不返回交易ID）
- 异常值（价格突然跳变几个数量级）
我曾经因为没做清洗，系统根据一个错误的价格触发了套利订单，直接亏损了5个BTC。嗯，从那以后我写了个清洗流水线：
class DataCleaner:
def __init__(self):
self.seen_ids = set() # 去重用
self.last_price = None
def clean_tick(self, tick):
# 1. 去重
if tick['id'] in self.seen_ids:
return None
self.seen_ids.add(tick['id'])
# 2. 时间戳对齐（统一转成UTC纳秒）
tick['ts'] = self._normalize_ts(tick['ts'])
# 3. 价格合理性检查
if self.last_price:
change = abs(tick['price'] - self.last_price) / self.last_price
if change > 0.1: # 单笔波动超过10%，标记异常
self._report_anomaly(tick)
return None
self.last_price = tick['price']
return tick避坑指南：时间戳对齐是最大的坑。不同交易所的时钟可能差几秒，我建议用NTP服务器统一校准，并在数据入库前做线性插值对齐。
4.3 Tick级与K线级数据存储
数据存哪里？怎么存？这取决于你的使用场景。
| 数据类型 | 存储方案 | 典型容量 | 查询场景 |
|---|---|---|---|
| Tick级（逐笔成交） | 时序数据库（InfluxDB、ClickHouse） | 每天约500GB（全市场） | 回放、延迟分析、高频策略回测 |
| K线级（1分钟/5分钟） | 关系型数据库（PostgreSQL） | 每天约10GB | 策略信号计算、风控检查 |
| 快照级（Order Book） | 对象存储（S3/MinIO） | 每天约2TB | 深度分析、流动性建模 |
我个人习惯用分层存储策略：
- 热数据（最近7天）：放在SSD上，用InfluxDB存Tick级数据，查询延迟控制在10ms以内。
- 温数据（7-90天）：放在普通HDD上，用ClickHouse存聚合后的K线数据。
- 冷数据（90天以上）：压缩后存到S3，用Parquet格式，需要时再加载。
注意：Tick级数据不要用JSON存！我见过有人把每个tick存成一行JSON，结果一个月的日志文件就几百GB，查询慢得要命。用列式存储（Parquet/ORC）或者专门的时序数据库，压缩比能到10:1以上。
4.4 数据采集架构总览
说了这么多，咱们用一张图把整个数据采集模块串起来：
这张图其实就说了三件事：数据从哪来、怎么洗干净、存到哪去。每个环节都有坑，但只要你把「容错」和「对齐」这两个词刻在脑子里，基本不会出大问题。
我的经验：刚开始做数据采集时，别追求完美。先跑通一条链路，再逐步加冗余。我见过太多团队一开始就想搞「全市场全数据」，结果半年了还在做数据清洗。
公众号:蓝海资料掘金营，微信deep3321