第四章：行情数据接入：CFETS/外汇交易中心行情接口、行情数据格式与解析分发引擎设计
各位同学，今天我们来聊聊做市系统里最基础、也最要命的一环——行情数据接入。
我常说一句话：行情是做市系统的血液。没有行情，你的定价模型再牛、算法再快，都是空中楼阁。我自己刚入行那会儿，就吃过行情延迟的亏，差点让公司亏掉一套房的首付。嗯，从那以后，我对行情接入这块就特别较真。
4.1 行情数据源：CFETS/外汇交易中心接口
国内银行间市场，行情数据主要来自中国外汇交易中心（CFETS）。说白了，就是官方指定的数据源。你想想看，银行间市场的报价、成交、深度数据，都得从这里拿。
CFETS 提供的接口主要有两种：
- FIX 协议接口：国际通用的金融信息交换协议。外汇交易中心支持 FIX 4.4 版本。我个人习惯用这个，因为兼容性好，对接海外系统也方便。
- STEP 协议接口：中国版的 FIX，基于 FIX 协议做了本地化改造。国内很多银行间系统用这个。
- 自定义接口：有些老系统或者特殊需求，会用交易中心提供的专有格式。嗯，这个我建议尽量少碰，维护成本高。
核心要点：无论用哪种接口，你都得搞清楚行情数据的生命周期——从交易中心发出，到你的系统收到，中间经历了哪些环节。我曾经排查过一个延迟问题，发现是网络链路上多了一跳，白白浪费了 2 毫秒。
4.2 行情数据格式：FIX / STEP / 自定义
咱们一个一个来看。
4.2.1 FIX 协议格式
FIX 协议的消息是纯文本的，用标签-值对表示。比如一条简单的行情快照：
8=FIX.4.4|9=78|35=W|49=CFETS|56=YOUR_SYSTEM|34=2|52=20240315-10:30:00.123|55=USD/CNY|268=1|269=0|270=7.1234|271=1000000|10=123|这里每个字段都有含义：
8：协议版本35：消息类型（W 表示行情快照）55：交易品种268：行情条目数269：行情类型（0=买价，1=卖价）270：价格271：数量
我的经验：FIX 协议虽然可读性好，但解析起来有坑。比如分隔符不是普通的竖线，而是 SOH（0x01）字符。我第一次写解析器时，直接用字符串 split，结果死活解析不对。后来才发现是分隔符的问题。
4.2.2 STEP 协议格式
STEP 协议和 FIX 很像，但有一些中国特色的扩展。比如增加了人民币相关的字段，支持中文标签等。格式上基本一致，但字段定义略有不同。
8=STEP.1.0.0|9=85|35=W|49=CFETS|56=YOUR_SYSTEM|34=5|52=20240315-10:30:01.456|55=USD/CNY|268=2|269=0|270=7.1235|271=2000000|269=1|270=7.1236|271=1500000|10=456|你看，STEP 的版本号是 STEP.1.0.0，而 FIX 是 FIX.4.4。其他结构大同小异。
4.2.3 自定义格式
有些老系统或者特定场景，会用交易中心提供的专有格式。比如二进制定长报文、XML 格式等。我建议能不用就不用，除非你不得不对接遗留系统。
避坑指南：我曾经接手过一个项目，对方用了自定义的二进制格式，文档还不全。结果解析时发现字段偏移量算错了，导致所有价格都读反了。嗯，从那以后，我要求所有新系统必须用标准协议。
4.3 行情解析引擎设计
行情解析引擎，说白了就是把原始报文转成系统内部能用的数据结构。我设计过好几版，踩过不少坑，这里分享一个比较成熟的方案。
4.3.1 整体架构
先看一张图，这是我个人比较喜欢的架构：
你看，整个流程分四层：
- 网络接入层：负责和 CFETS 建立连接，处理心跳、重连、序列号校验等。我建议用 Netty 或者 Vert.x 这种异步框架，性能好。
- 协议解析层：把 FIX/STEP/自定义报文转成统一的对象。这里要特别注意性能，因为行情数据量很大。
- 数据清洗层：过滤掉脏数据、重复数据、过期数据。嗯，这一步很多人忽略，但很重要。
- 分发引擎：把清洗后的行情数据分发给各个业务模块。我一般用内存队列 + 多播的方式。
4.3.2 核心代码示例
这里给一个 FIX 协议解析的简化版代码，用 Java 写的：
public class FixMessageParser {
private static final char SOH = '\u0001';
public MarketDataSnapshot parse(String rawMessage) {
MarketDataSnapshot snapshot = new MarketDataSnapshot();
String[] fields = rawMessage.split(String.valueOf(SOH));
for (String field : fields) {
if (field.isEmpty()) continue;
String[] kv = field.split("=", 2);
if (kv.length != 2) continue;
int tag = Integer.parseInt(kv[0]);
String value = kv[1];
switch (tag) {
case 35: // 消息类型
snapshot.setMsgType(value);
break;
case 55: // 交易品种
snapshot.setSymbol(value);
break;
case 268: // 行情条目数
snapshot.setNoMdEntries(Integer.parseInt(value));
break;
case 269: // 行情类型
snapshot.setMdEntryType(value);
break;
case 270: // 价格
snapshot.setMdEntryPx(Double.parseDouble(value));
break;
case 271: // 数量
snapshot.setMdEntrySize(Long.parseLong(value));
break;
// 其他字段...
}
}
return snapshot;
}
}性能优化建议：
- 不要用 String.split，用循环遍历字符数组，能快 3-5 倍
- 预分配对象池，避免频繁 GC
- 用直接内存（DirectBuffer）存储原始报文，减少拷贝
4.3.3 分发引擎设计要点
行情分发引擎，说白了就是把解析好的行情数据，快速、可靠地送给各个消费者。我设计过几个版本，这里说几个关键点：
| 设计维度 | 推荐方案 | 踩过的坑 |
|---|---|---|
| 传输方式 | 内存队列 + 多播 | 用 TCP 单播，延迟高，CPU 占用大 |
| 队列选型 | Disruptor / Aeron | 用 LinkedBlockingQueue，吞吐量上不去 |
| 序列化 | FlatBuffers / SBE | 用 JSON，解析慢，带宽浪费 |
| 容错机制 | 序列号校验 + 重传 | 没做校验，丢数据都不知道 |
我个人比较推荐用 Disruptor 作为核心队列。为什么呢？因为它无锁、无竞争，延迟能控制在微秒级。我在一个项目中，用 Disruptor 替换了原来的 BlockingQueue，吞吐量直接翻了 5 倍。
重要提醒：行情分发一定要做背压处理。什么意思呢？就是当消费者处理不过来时，不能无限制地往队列里塞数据。我曾经见过一个系统，因为定价引擎卡住了，结果分发队列爆了，OOM 直接挂掉。嗯，从那以后，我都在分发引擎里加了水位监控和降级策略。
4.4 实战经验总结
最后，分享几个我在项目中积累的经验：
- 连接管理：CFETS 的连接是长连接，要处理好心跳和重连。我建议用指数退避策略，避免频繁重连把对方打挂。
- 数据校验：行情数据偶尔会有异常值，比如价格为 0 或者负数。一定要做校验，否则你的定价模型会算出离谱的结果。
- 监控告警：行情延迟、丢包、连接断开，都要有实时告警。我一般用 Prometheus + Grafana 做监控，阈值设得很敏感。
- 回放机制：行情数据要落盘，方便事后分析和回测。我建议用列式存储，比如 Parquet 格式，查询效率高。
好了，行情数据接入这块就讲到这里。记住一句话：行情是根，根扎稳了，上面的系统才能跑得快、跑得稳。
公众号:蓝海资料掘金营，微信deep3321