4、数据采集与接入：API对接、WebSocket实时流、文件批量导入、数据校验与清洗
做市商系统里，数据就是命根子。这话一点都不夸张。
我见过太多团队，策略模型跑得飞起，结果数据源一断，整个系统直接瘫痪。所以今天咱们聊聊数据采集与接入——说白了，就是怎么把外面乱七八糟的数据，变成系统能吃的干净粮食。
4.1 数据接入的三种主流方式
做市商的数据源，无非就三类：交易所的REST API、WebSocket实时流、还有历史数据的文件导入。每种方式都有自己的脾气，咱们一个一个说。
4.1.1 REST API 对接
REST API 是最常见的接入方式。交易所给你一个HTTP接口，你发请求，它返回JSON。简单吧？但坑不少。
我个人的习惯是：
- 限频控制：交易所都有请求频率限制。比如每秒最多5次。你超了，它直接封你IP。我在项目中遇到过，一个同事没做限频，结果整个机房的IP都被封了，那叫一个惨。
- 重试机制：网络波动是常态。请求失败后，等几秒再试。一般重试3次，间隔指数增长。
- 数据缓存：别每次都去拉全量数据。增量更新才是王道。
# 伪代码示例：带限频和重试的API请求
import time
import requests
def fetch_with_retry(url, max_retries=3):
for i in range(max_retries):
try:
resp = requests.get(url, timeout=5)
if resp.status_code == 200:
return resp.json()
except Exception as e:
print(f"第{i+1}次请求失败: {e}")
time.sleep(2 ** i) # 指数退避
raise Exception("请求彻底失败")
⚠️ 注意：千万别把API Key硬编码在代码里。用环境变量或者配置中心管理。我曾经见过有人把密钥提交到GitHub，结果被机器人扫到，账户直接被盗刷。
4.1.2 WebSocket 实时流
做市商对实时性要求极高。REST API 延迟太高，动不动几百毫秒。这时候就得用 WebSocket。
WebSocket 是长连接，交易所主动往你这边推数据。你想想看，行情一变，数据瞬间就到，这才是做市商该有的节奏。
接入WebSocket的几个要点：
- 心跳维持：连接久了容易断。交易所一般要求你每隔30秒发一个ping包。你不发，它就当你死了，直接断开。
- 断线重连：网络波动导致断开是家常便饭。我建议你写个自动重连逻辑，断开后立即重连，别让数据断流超过1秒。
- 数据去重：有时候交易所会重复推送同一条数据。你收到后得做个去重，不然策略会重复下单。
# 伪代码示例：WebSocket心跳与重连
import websocket
def on_message(ws, message):
# 处理行情数据
process_tick(message)
def on_close(ws, close_status_code, close_msg):
print("连接断开，准备重连...")
time.sleep(1)
ws.run_forever() # 自动重连
ws = websocket.WebSocketApp(
"wss://exchange.com/ws",
on_message=on_message,
on_close=on_close
)
ws.run_forever(ping_interval=30) # 每30秒发一次心跳
💡 小技巧：WebSocket 数据量大的时候，别在主线程里处理。用队列把数据丢给后台线程，主线程只管收。不然处理慢了，数据会堆积，延迟越来越高。
4.1.3 文件批量导入
历史数据怎么办？比如你要回测一个策略，需要过去一年的订单簿数据。这时候就得靠文件批量导入。
常见的文件格式有CSV、Parquet、甚至交易所自己的二进制格式。我个人偏爱Parquet，因为它压缩率高，读取快，还支持列式存储。
导入流程：
- 文件校验：先检查文件完整性。比如MD5校验，防止文件传输过程中损坏。
- 格式转换：把原始数据转成系统内部统一的Schema。
- 分批写入：别一次性加载几十GB的文件到内存。分批次读，每批10万条，写入数据库。
# 伪代码示例：分批导入CSV文件
import pandas as pd
chunk_size = 100000
for chunk in pd.read_csv("history.csv", chunksize=chunk_size):
# 数据清洗
chunk = chunk.dropna()
chunk = chunk[chunk['price'] > 0] # 过滤无效价格
# 写入数据库
db.insert(chunk)
4.2 数据校验与清洗
数据进来了，但能用吗？不一定。交易所的数据也可能有错。比如价格字段出现负数，时间戳格式不对，甚至数据重复。
我把它总结成三步：
4.2.1 格式校验
检查每个字段的类型和范围。比如：
- 价格必须是正数
- 时间戳必须是毫秒级整数
- 交易量不能为0
4.2.2 逻辑校验
有些数据格式对，但逻辑不对。比如：
- 卖一价低于买一价（这明显不合理）
- 时间戳比当前时间还晚（数据穿越了？）
- 同一笔交易出现两次（重复数据）
4.2.3 异常处理
发现异常数据怎么办？我建议：
- 丢弃：明显错误的数据直接扔掉
- 标记：可疑数据打个标签，后续人工审核
- 告警：异常率超过阈值，立刻通知运维
📌 核心原则：脏数据进，脏数据出。数据清洗不是可选项，是必选项。我见过一个团队，因为没做数据校验，策略根据错误数据下了几十笔废单，亏了几十万。
4.3 数据接入的整体架构
说了这么多，咱们画张图看看整体流程。
这张图展示了数据从源头到清洗完成的完整链路。你看，数据源有三种，经过接入层处理，再到校验层清洗，最后输出干净数据给下游策略使用。
4.4 实战中的避坑指南
最后，分享几个我踩过的坑：
- 别信交易所的文档：文档写的和实际返回的数据经常对不上。我建议你写个测试脚本，先跑一遍看看真实数据长什么样。
- 数据延迟要监控：WebSocket 偶尔会卡住，数据延迟从几毫秒变成几秒。你得有个监控，延迟超过阈值就告警。
- 文件导入要幂等：同一个文件导入两次，结果应该一样。不然回测结果对不上，你都不知道该信哪个。
💡 我的经验：数据接入这块，花80%的时间做校验和容错，只花20%的时间写接入代码。因为数据不出错，后面的策略才能跑得稳。
好了，数据采集与接入就聊到这儿。记住一句话：数据质量决定策略上限。把数据管好了，你的做市商系统就成功了一半。