数据源接入：交易所行情协议解析与文件格式处理
做量化回测系统，第一步就是搞定数据。这事儿听起来简单，做起来全是坑。我见过太多团队，策略写得漂漂亮亮，结果数据源一接进来就崩了——要么解析速度跟不上，要么历史数据格式不统一，要么被交易所限流搞得焦头烂额。
今天咱们就把这块彻底聊透。我会从三个维度展开：行情协议怎么解、文件格式怎么选、鉴权限流怎么扛。嗯，都是实战中摔出来的经验。
一、交易所行情协议解析（L1/L2）
先说说L1和L2的区别。说白了，L1就是快照数据，每秒拍一张照片给你看。L2是逐笔数据，每一笔成交、每一笔挂单都告诉你。你想想看，做高频策略肯定要L2，做日频策略L1就够了。
1. L1行情协议
国内交易所的L1行情，基本都走的是TCP或UDP组播。我习惯用UDP组播，延迟低，但要做好丢包重传。协议格式各家不同，但核心字段就这些：
| 字段 | 类型 | 说明 |
|---|---|---|
| 交易日期 | int32 | YYYYMMDD格式 |
| 合约代码 | char[8] | 如"rb2401" |
| 最新价 | int64 | 乘以精度因子 |
| 成交量 | int64 | 累计成交量 |
| 持仓量 | int64 | 当前持仓 |
解析的时候要注意字节序。国内交易所大多用大端序，但有些期货公司转发的数据是小端序。我在项目中遇到过这个问题，当时排查了半天，最后发现是字节序没对齐。
核心原则：先读协议头，判断版本号，再根据版本号选择对应的解析器。别写死，否则交易所一升级你就得改代码。
2. L2逐笔数据
L2的数据量比L1大两个数量级。举个例子，一个活跃的期货合约，L1每秒也就几十个字段，L2每秒可能上千笔成交。解析L2，我建议用零拷贝技术。
// 伪代码：零拷贝解析L2逐笔成交
struct TradeRecord {
uint64_t timestamp; // 纳秒级时间戳
uint32_t price; // 价格，乘以10000
uint32_t volume; // 成交量
uint8_t side; // 0=买, 1=卖
uint8_t flag; // 成交标志位
} __attribute__((packed));
// 直接内存映射，避免拷贝
void* mapped = mmap(fd, size, PROT_READ, MAP_SHARED, 0, 0);
TradeRecord* records = (TradeRecord*)mapped;
for (int i = 0; i < count; i++) {
process(records[i]);
}
为什么要用零拷贝？因为L2数据量大，每次拷贝都会触发CPU缓存失效。我曾经在一个项目中，把解析从memcpy改成mmap，吞吐量直接提升了3倍。
注意：L2数据的时间戳精度很高，通常是纳秒级。但不同交易所的时钟源不一样，跨交易所回测时一定要做时间对齐。我见过有人直接把两个交易所的纳秒戳拿来比较，结果差了整整1秒——因为一个用的是GPS时钟，另一个用的是NTP。
二、历史数据文件格式（CSV/Parquet/二进制）
历史数据存什么格式？这个问题我纠结了很久。三种格式各有优劣，我直接给你对比表：
| 格式 | 存储效率 | 读取速度 | 可读性 | 适用场景 |
|---|---|---|---|---|
| CSV | 低（文本存储） | 慢（需解析） | 高 | 小数据量、调试用 |
| Parquet | 高（列式压缩） | 快（谓词下推） | 中 | 大数据分析、Pandas |
| 二进制 | 最高（紧凑结构） | 最快（内存映射） | 低 | 高频回测、生产环境 |
1. CSV格式
CSV最直观，但坑也最多。我建议你注意三点：
- 时间戳格式统一：别一会儿用"2024-01-01 09:30:00"，一会儿用"20240101093000"。我习惯用Unix毫秒时间戳，排序和计算都方便。
- 缺失值处理：停牌期间的数据是空着还是填前值？我建议统一填NaN，回测时由策略自行决定。
- 浮点数精度：CSV存浮点数会丢失精度。我建议价格和成交量都存整数，比如价格乘以10000存int32。
小技巧：如果你非要用CSV，记得用gzip压缩。同样的数据，压缩后体积能减少80%。读取时用pandas的read_csv直接支持.gz文件，很方便。
2. Parquet格式
Parquet是我现在的主力格式。它的列式存储特性，让按列读取变得极快。比如你只需要"close"和"volume"两列，Parquet只读这两列的数据块，CSV得全读进来再过滤。
# Python示例：Parquet按列读取
import pandas as pd
# 只读取需要的列，速度极快
df = pd.read_parquet(
'data.parquet',
columns=['timestamp', 'close', 'volume']
)
# 支持谓词下推，只读取特定时间范围
df = pd.read_parquet(
'data.parquet',
filters=[('timestamp', '>=', 1704067200000)]
)
不过Parquet也有缺点。小文件场景下，它的元数据开销占比太大。我建议每个Parquet文件至少存100万行以上，否则不如用二进制。
3. 二进制格式
二进制格式是性能之王。我自己的回测引擎就用二进制，直接内存映射，读取延迟在微秒级。设计二进制格式时，我遵循这几个原则：
- 固定长度记录：每条记录长度固定，方便随机访问。比如每条行情记录固定64字节。
- 文件头包含元数据：记录数、起始时间、结束时间、字段偏移量。这样读取时不用扫描全文件。
- 支持版本升级：文件头里留一个版本号字段，方便以后加字段。
我的建议：如果你刚开始做回测系统，先用Parquet。等性能瓶颈出现了，再针对热点数据转成二进制。别一开始就上二进制，调试起来太痛苦。
三、数据源鉴权与限流处理
这块是很多人容易忽略的。数据源接入不只是解析协议，还得搞定鉴权和限流。我见过有人写了个回测脚本，跑得好好的，结果一上线就被交易所封了IP。
1. 鉴权机制
交易所的鉴权方式五花八门。我总结下来，主流就三种：
| 鉴权方式 | 原理 | 安全性 | 适用场景 |
|---|---|---|---|
| API Key + Secret | HMAC签名 | 高 | REST API |
| Token认证 | JWT或自定义Token | 中 | WebSocket |
| IP白名单 | 绑定固定IP | 低 | 专线接入 |
我个人习惯用API Key + Secret的方式。注意Secret绝对不能硬编码在代码里，要用环境变量或密钥管理服务。我曾经见过有人把Secret提交到GitHub上，结果被爬虫扫到，数据被盗用了三个月才发现。
# 正确的做法：从环境变量读取密钥
import os
import hmac
import hashlib
api_key = os.environ.get('EXCHANGE_API_KEY')
api_secret = os.environ.get('EXCHANGE_API_SECRET')
# 生成签名
message = timestamp + method + path + body
signature = hmac.new(
api_secret.encode(),
message.encode(),
hashlib.sha256
).hexdigest()
2. 限流处理
限流是数据源接入中最头疼的问题。交易所为了保护自己的服务器，会对每个连接做频率限制。常见的限流策略有：
- 令牌桶算法：每秒发放固定数量的令牌，请求消耗令牌。我常用这个，实现简单，还能应对突发流量。
- 滑动窗口：统计过去N秒内的请求数，超过阈值就拒绝。适合对实时性要求不高的场景。
- 指数退避：被限流后，等待时间指数增长。比如第一次等1秒，第二次等2秒，第三次等4秒。
避坑指南：我曾经在接入某期货交易所时，没注意它的限流规则是"每IP每秒100次请求"。我的程序开了10个线程同时请求，结果每个线程都以为自己没超限，合起来每秒1000次请求，直接被封了24小时。后来我改用全局令牌桶，所有线程共享一个限流器，问题才解决。
限流处理的核心是优雅降级。被限流了不要直接抛异常，而是：
- 记录日志，标记当前时间点数据缺失
- 等待重试，用指数退避策略
- 如果连续重试失败超过N次，切换到备用数据源
- 回测时，对缺失数据做插值或标记
知识体系总览
说了这么多，我画了一张图帮你梳理整个数据源接入的架构。你看一眼就明白了：
这张图把整个数据源接入的流程串起来了。从底层的数据源，到中间的协议解析，再到上层的鉴权限流，每一层都有各自的职责。你搭建系统时，按这个分层结构来设计，扩展性和可维护性都会好很多。
最后说一句：数据源接入是回测系统的地基。地基没打好，上面策略写得再漂亮也是空中楼阁。我建议你花时间把这块做扎实，后面会省很多事。
公众号:蓝海资料掘金营，微信deep3321