数据管道设计：行情数据接入、数据归一化、实时数据总线、历史数据存储

数据管道，说白了就是高频交易系统的血管。行情数据能不能喂得进来、喂得快不快、喂得准不准，直接决定了你的策略能不能抢到那几微秒的利润。我见过太多团队，策略写得花里胡哨，结果数据管道一压测就崩，行情还没到策略模块呢，价格已经跳了三跳。

今天咱们就把数据管道的四个核心环节拆开揉碎讲清楚。嗯，这四块分别是：行情数据接入、数据归一化、实时数据总线、历史数据存储。每一块都有坑，我一个个说。

一、行情数据接入

行情接入是管道的入口。交易所的行情数据，通常通过两种方式下发：一种是快照（Snapshot），一种是增量（Incremental）。快照就是当前时刻的全量盘口数据，增量就是每次变化的那一小撮。

我个人习惯的做法是：先用快照初始化本地状态，然后靠增量流持续更新。为什么？因为增量数据量小，延迟低。但这里有个坑——增量数据可能会丢包。一旦丢了一个包，你的本地状态就跟交易所对不上了。

行情接入的协议，常见的有两种：

• TCP：可靠，但重传机制会导致延迟抖动。适合对数据完整性要求极高的场景。
• UDP组播：不可靠，但延迟低且稳定。适合高频交易场景，但需要自己处理丢包和乱序。

你想想看，如果做的是纳秒级套利，TCP那点重传延迟可能就让你错过行情了。所以大部分高频交易团队都选UDP组播，然后在应用层做校验和重传。

下面是一个简单的UDP行情接入代码示例：

// 伪代码：UDP行情接入
void* udp_listener(void* arg) {
    int sock = socket(AF_INET, SOCK_DGRAM, 0);
    struct sockaddr_in addr;
    addr.sin_family = AF_INET;
    addr.sin_port = htons(8888);
    addr.sin_addr.s_addr = inet_addr("224.0.0.1");
    bind(sock, (struct sockaddr*)&addr, sizeof(addr));
    
    // 加入组播组
    struct ip_mreq mreq;
    mreq.imr_multiaddr.s_addr = inet_addr("224.0.0.1");
    mreq.imr_interface.s_addr = htonl(INADDR_ANY);
    setsockopt(sock, IPPROTO_IP, IP_ADD_MEMBERSHIP, &mreq, sizeof(mreq));
    
    char buffer[1024];
    while (1) {
        int len = recvfrom(sock, buffer, sizeof(buffer), 0, NULL, NULL);
        // 解析行情数据
        parse_market_data(buffer, len);
    }
}

二、数据归一化

不同交易所的行情格式千奇百怪。有的用二进制，有的用JSON，有的用自定义协议。你的策略总不能为每个交易所写一套解析逻辑吧？所以需要数据归一化。

归一化，说白了就是把各种格式的行情数据，统一成一种内部标准格式。我建议用Protocol Buffers或者FlatBuffers。为什么？因为解析速度快，序列化后体积小。

归一化要做三件事：

1. 字段映射：把交易所的字段名映射成内部统一的字段名。比如上交所的"LastPx"和深交所的"last_price"，都映射成内部的"last_price"。
2. 类型转换：把字符串转成数值，把时间戳转成统一格式。比如把"2024-01-15 09:30:00.123"转成纳秒级时间戳。
3. 数据校验：检查数据是否合法。比如价格不能为负，成交量不能为0。

归一化后的数据格式示例：

// 内部统一行情格式
message MarketData {
    int64 timestamp_ns = 1;      // 纳秒级时间戳
    string symbol = 2;           // 合约代码
    double last_price = 3;       // 最新价
    int64 volume = 4;            // 成交量
    double bid_price = 5;        // 买一价
    double ask_price = 6;        // 卖一价
    int32 bid_volume = 7;        // 买一量
    int32 ask_volume = 8;        // 卖一量
}

三、实时数据总线

归一化后的数据，需要分发给多个消费者。比如策略模块、风控模块、监控模块、历史存储模块。如果每个模块都直接去行情接入模块拉数据，那接入模块就成了瓶颈。

所以需要一条实时数据总线。我推荐用基于共享内存的发布-订阅模式。为什么不用消息队列？因为Kafka、RabbitMQ这些消息队列的延迟太高了，动不动就几毫秒，高频交易根本受不了。

共享内存的延迟可以做到微秒级。具体做法是：

• 在内存中开辟一块环形缓冲区（Ring Buffer）。
• 行情接入模块作为生产者，往环形缓冲区里写数据。
• 各个消费者模块作为订阅者，从环形缓冲区里读数据。

这里要注意并发控制。多个消费者同时读同一个位置怎么办？我建议用无锁队列（Lock-Free Queue），用CAS操作来保证原子性。

下面是一个简单的无锁环形缓冲区实现：

// 伪代码：无锁环形缓冲区
template<typename T, size_t SIZE>
class LockFreeRingBuffer {
    std::atomic<size_t> head_{0};
    std::atomic<size_t> tail_{0};
    T buffer_[SIZE];
    
public:
    bool push(const T& item) {
        size_t tail = tail_.load(std::memory_order_relaxed);
        size_t next = (tail + 1) % SIZE;
        if (next == head_.load(std::memory_order_acquire)) {
            return false; // 缓冲区满
        }
        buffer_[tail] = item;
        tail_.store(next, std::memory_order_release);
        return true;
    }
    
    bool pop(T& item) {
        size_t head = head_.load(std::memory_order_relaxed);
        if (head == tail_.load(std::memory_order_acquire)) {
            return false; // 缓冲区空
        }
        item = buffer_[head];
        head_.store((head + 1) % SIZE, std::memory_order_release);
        return true;
    }
};

四、历史数据存储

实时数据用完了，不能丢。历史数据是策略回测和事后分析的命根子。但高频交易的数据量太大了，一天可能产生几十GB甚至几百GB的行情数据。怎么存？

我建议用列式存储格式，比如Parquet或ORC。为什么？因为行式存储（比如CSV）在查询时会把整行数据都读出来，而列式存储只读你需要的列，IO效率高得多。

存储策略上，我一般分两层：

层级	存储介质	数据范围	用途
热数据	SSD + 内存映射	最近7天	策略回测、实时分析
冷数据	HDD 或 对象存储	7天以上	长期归档、合规审计

热数据用内存映射文件（mmap），查询时直接访问内存，速度极快。冷数据压缩后存到对象存储，比如S3或MinIO，成本低。

数据压缩也很关键。行情数据有很强的规律性，比如价格变化通常很小。用差分编码（Delta Encoding）可以把数据量压缩到原来的1/5甚至1/10。具体做法是：不存原始价格，只存价格的变化量。

// 差分编码示例
原始价格: 100.01, 100.02, 100.01, 100.03, 100.02
差分编码: 100.01, +0.01, -0.01, +0.02, -0.01

整体架构图

下面我用一张SVG图把整个数据管道的架构画出来。你一看就明白了。

从图上可以看得很清楚：行情数据从交易所进来，经过接入、归一化，进入实时数据总线。总线把数据分发给策略、风控和历史存储。历史存储又分热数据和冷数据两层。整个链路环环相扣，任何一个环节出问题，都会影响整个系统的性能。

---

公众号:蓝海数据掘金营，微信deep3321