第2章数据源接入

数据源接入：交易所行情网关对接、UDP组播数据接收、TCP重传机制处理

行情数据源接入，是整个高频引擎的「水龙头」。

水龙头要是拧不开，或者水流忽大忽小，后面再好的管道、再快的处理器，全白搭。我见过不少团队，策略模型写得漂漂亮亮，结果一上实盘，行情延迟抖得像心电图——最后查出来，是网关对接那层没做好。

说白了，数据源接入就三件事：连得上、收得全、补得快。咱们一个一个拆开聊。

交易所行情网关对接

每家交易所的行情网关，都有自己的脾气。

上交所用STEP协议，深交所用MDDP，中金所又是另一套。我个人的习惯是，不要试图写一个「万能网关」——那玩意儿维护成本太高，而且容易出兼容性问题。

正确的做法是：

抽象一层统一的行情接口：不管底层是STEP还是MDDP，对外暴露的API应该是一样的。比如 subscribe(symbol, callback)、unsubscribe(symbol)。
每个交易所单独实现一个适配器：适配器负责协议解析、心跳维持、登录认证这些脏活累活。
连接管理要独立：网关断连、重连、切换主备，这些逻辑不要和业务代码混在一起。

核心要点：网关对接的成败，不在于你多快能解析一条行情，而在于你多快能发现连接异常，并自动恢复。

我曾经遇到过一个坑：某交易所的网关，每天收盘后会自动断开连接，但不会发任何通知。我们的系统傻傻地等着，直到第二天开盘才发现行情断了。后来我加了一个心跳超时检测，3秒没收到心跳就主动重连，这才解决问题。

UDP组播数据接收

交易所的行情数据，绝大多数走的是UDP组播。

为什么用UDP？因为快。TCP有三次握手、拥塞控制，延迟不稳定。UDP只管发，不管收，延迟低且可控。

但UDP也有它的毛病——丢包。网络稍微抖动一下，包就丢了。你想想看，如果丢了关键的快照数据，你的策略可能就会做出错误决策。

接收UDP组播，我建议注意以下几点：

设置合适的接收缓冲区大小：Linux默认的UDP接收缓冲区可能只有几百KB，对于高频行情来说远远不够。我一般会设置到 16MB 甚至更大。
使用SO_REUSEPORT：允许多个进程/线程绑定到同一个端口，实现并行接收。这在多核机器上能显著提升吞吐量。
绑定CPU核心：把接收线程绑定到固定的CPU核心上，避免上下文切换带来的延迟抖动。

import socket
import struct

def setup_udp_multicast(mcast_group, port, iface_ip):
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    
    # 设置接收缓冲区为16MB
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 16 * 1024 * 1024)
    
    # 加入组播组
    mreq = struct.pack("4s4s", socket.inet_aton(mcast_group), socket.inet_aton(iface_ip))
    sock.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, mreq)
    
    sock.bind(('', port))
    return sock

小技巧：接收UDP数据时，尽量用 recvmsg 而不是 recv。前者能拿到辅助数据，比如时间戳，对后续的延迟分析很有帮助。

TCP重传机制处理

UDP丢包了怎么办？靠TCP重传。

交易所通常会提供两条通道：一条是UDP组播的「快通道」，一条是TCP的「补通道」。UDP丢了包，就从TCP通道请求重传。

这里有个关键点：重传请求不能太频繁，也不能太慢。

太频繁：网络带宽被重传请求占满，反而影响正常行情接收。
太慢：行情已经过了时效，重传回来的数据没有意义。

我个人的经验是：维护一个滑动窗口，记录最近收到的包序号。一旦发现序号不连续，立即发起重传请求。同时设置一个超时时间，比如10毫秒，超时未收到重传数据，就放弃这次请求，继续处理后续数据。

class GapDetector:
    def __init__(self, window_size=1000):
        self.window = set()
        self.window_size = window_size
        self.last_seq = 0
        
    def on_packet(self, seq_num):
        if seq_num <= self.last_seq:
            return  # 重复包，忽略
        if seq_num - self.last_seq > 1:
            # 发现缺口，发起重传请求
            missing = list(range(self.last_seq + 1, seq_num))
            self.request_retransmit(missing)
        self.last_seq = seq_num
        self.window.add(seq_num)
        if len(self.window) > self.window_size:
            self.window.pop()

注意：TCP重传通道本身也可能丢包或延迟。我建议给重传请求加上序列号，并记录请求时间。如果某个请求超时未响应，可以尝试从备用通道获取数据，或者直接跳过。

整体架构图

下面这张图，展示了数据源接入层的整体结构。你可以看到UDP和TCP两条通道如何协同工作。

这张图里，你看到的是典型的「双通道」架构。UDP负责快，TCP负责稳。两者配合，才能在高频场景下做到低延迟 + 高可靠。

一句话总结：数据源接入的核心，不是把数据拿回来，而是稳定地、低延迟地、不丢不重地把数据拿回来。网关对接、UDP接收、TCP重传，这三板斧缺一不可。

嗯，数据源这块就聊到这儿。下一层，咱们该聊聊数据到了之后，怎么在内存里高效流转了。

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

📚 课程目录

数据源接入：交易所行情网关对接、UDP组播数据接收、TCP重传机制处理

交易所行情网关对接

UDP组播数据接收

TCP重传机制处理

整体架构图