第3章数据采集层设计

第3章：数据采集层设计：从CTP/易盛等API获取行情与成交数据，使用ZeroMQ进行低延迟传输

数据采集层，说白了就是整个高频交易系统的“耳朵”和“嘴巴”。

耳朵用来听市场的声音——行情数据；嘴巴用来表达我们的意图——下单成交。这一层如果设计不好，后面再牛的策略也是白搭。我见过太多团队，策略模型跑得飞起，结果数据源延迟高了10微秒，整个系统就废了。

3.1 行情数据源：CTP与易盛的选择

国内期货市场，主流的API就两个：CTP（上期技术）和易盛。我个人习惯把CTP作为首选，因为它的市场占有率最高，接口文档也相对规范。但易盛在某些品种上的速度确实有优势，尤其是郑州商品交易所的品种。

这里有个坑，我曾经踩过：CTP的行情接口是推送式的，但它的推送频率并不是固定的。你想想看，如果市场剧烈波动，CTP可能会在1毫秒内推送几十笔数据；如果市场冷清，可能几秒钟都没动静。所以，你的接收端必须能处理这种“突发流量”。

  核心要点： 行情数据采集不是简单的“收数据”，而是要处理数据流的“尖峰”和“低谷”。

3.2 成交数据：别小看它

很多人只关注行情，忽略了成交数据。其实，成交数据才是检验策略是否有效的“照妖镜”。

我记得有一次，策略在回测中表现完美，实盘一跑就亏钱。查了半天，发现是成交数据的延迟问题——CTP返回的成交确认比行情慢了整整50毫秒。这50毫秒里，策略已经根据错误的状态下了好几笔废单。

成交数据包括：

委托回报：你的订单是否被交易所接收
成交回报：订单是否成交，成交价和量是多少
撤单回报：撤单是否成功

这些数据必须和行情数据严格对齐时间戳。我建议使用交易所的原始时间戳，而不是本地时间。因为本地时钟可能漂移，而交易所的时间是统一的。

3.3 ZeroMQ：低延迟传输的利器

数据采集到了，怎么传给策略层？很多人直接用TCP Socket，或者用消息队列如RabbitMQ。但在高频交易场景下，这些方案延迟太高了。

ZeroMQ（简称ZMQ）是我用了多年的方案。它本质上是一个高性能的异步消息库，但比传统消息队列轻量得多。说白了，它就是一个“快递员”，专门负责把数据从A点送到B点，而且速度极快。

ZMQ的核心模式有三种：

模式	说明	适用场景
PUB-SUB	发布-订阅模式，一个发布者，多个订阅者	行情数据广播
PUSH-PULL	管道模式，数据从一端流向另一端	成交数据传递
REQ-REP	请求-应答模式	控制指令

我个人最常用的是PUB-SUB模式来传输行情数据。为什么？因为行情数据是“一对多”的——一个数据采集节点，需要同时供给策略节点、风控节点、监控节点。PUB-SUB天然支持这种广播。

小技巧： ZMQ的PUB-SUB模式有个“慢订阅者”问题。如果某个订阅者处理速度慢，它会阻塞整个管道。我的解决方案是：给每个订阅者设置独立的缓冲区，或者使用“丢弃旧数据”策略。

3.4 数据采集层的架构图

下面这张图是我在实际项目中使用的数据采集层架构。你看一眼就明白了：

你看，数据从CTP/易盛进来后，统一经过ZMQ的PUB节点进行广播。策略、风控、监控各自订阅自己需要的数据。这样做的好处是：数据采集和数据处理完全解耦。哪怕策略节点挂了，数据采集依然在跑，不会丢数据。

3.5 代码示例：ZMQ行情发布与订阅

光说不练假把式。下面是我在实际项目中用的代码片段，你直接拿去改改就能用。

发布端（数据采集节点）：

import zmq
import json

class MarketDataPublisher:
    def __init__(self, pub_port=5555):
        self.context = zmq.Context()
        self.socket = self.context.socket(zmq.PUB)
        self.socket.bind(f"tcp://*:{pub_port}")
        print(f"行情发布者已启动，端口: {pub_port}")
    
    def publish_tick(self, tick_data):
        """
        发布Tick行情数据
        tick_data格式: {"symbol":"rb2501","price":3800,"volume":100,"timestamp":1234567890}
        """
        topic = "tick"
        message = json.dumps(tick_data)
        # ZMQ的PUB-SUB模式，先发主题，再发数据
        self.socket.send_string(f"{topic} {message}")
    
    def publish_trade(self, trade_data):
        """
        发布成交数据
        """
        topic = "trade"
        message = json.dumps(trade_data)
        self.socket.send_string(f"{topic} {message}")

订阅端（策略节点）：

import zmq
import json

class MarketDataSubscriber:
    def __init__(self, pub_host="localhost", pub_port=5555):
        self.context = zmq.Context()
        self.socket = self.context.socket(zmq.SUB)
        self.socket.connect(f"tcp://{pub_host}:{pub_port}")
        # 订阅所有主题，也可以只订阅特定主题如 "tick"
        self.socket.subscribe("")
        print(f"行情订阅者已连接，地址: {pub_host}:{pub_port}")
    
    def receive_data(self):
        """接收数据，非阻塞模式"""
        try:
            message = self.socket.recv_string(flags=zmq.NOBLOCK)
            topic, data_str = message.split(" ", 1)
            data = json.loads(data_str)
            return topic, data
        except zmq.Again:
            return None, None

注意： 上面的代码用了 zmq.NOBLOCK 标志，这是非阻塞接收。在高频场景下，绝对不能阻塞主线程。我曾经见过有人用阻塞接收，结果行情一快，整个策略线程卡死，直接爆仓。

3.6 延迟优化：几个关键点

数据采集层的延迟，直接决定了你的交易速度。我总结了几条经验：

避免数据拷贝：ZMQ内部会做内存拷贝，尽量减少。可以用 zmq.NOBLOCK 配合零拷贝技术。
线程绑定CPU：把数据采集线程绑定到独立的CPU核心上，避免被其他进程干扰。Linux下用 taskset 命令。
使用共享内存：如果数据采集和策略在同一个机器上，可以考虑用共享内存代替ZMQ。但这样会失去灵活性，我一般只在极端追求速度时用。
控制数据量：不是所有行情数据都需要传输。比如，你可以只传输价格变化超过某个阈值的Tick，而不是每个Tick都传。

嗯，这里要特别提一下：延迟优化是个无底洞。我见过有人为了省1微秒，把代码写成天书，结果维护成本暴涨。我的建议是：先做到10微秒以内，再考虑要不要继续优化。别一开始就追求极致。

3.7 避坑指南

最后，分享几个我踩过的坑：

坑1：ZMQ的HWM（高水位线）。默认情况下，ZMQ的发送缓冲区是无限的。如果订阅者处理不过来，内存会爆。一定要设置 socket.setsockopt(zmq.SNDHWM, 1000) 限制缓冲区大小。
坑2：CTP的登录超时。CTP的登录过程可能长达几秒，如果网络不稳定，甚至会超时。我建议在登录时加一个重试机制，最多重试3次。
坑3：时间戳对齐。CTP返回的时间戳是交易所时间，但本地时间可能有偏差。我习惯在数据采集层就把时间戳统一转换成UTC，避免后续处理混乱。

数据采集层设计好了，你的高频交易系统就成功了一半。剩下的，就是策略和风控的事了。

公众号:蓝海资料掘金营，微信deep3321

📚 课程目录