一、数据管道概述：量化交易的数据命脉

做量化交易这些年，我越来越觉得——策略模型再牛，没有靠谱的数据管道，一切都是空中楼阁。说白了，数据管道就是你的交易系统里，那条让数据从源头流到策略引擎的「高速公路」。

今天我们就来聊聊，这条高速公路到底长什么样。

1.1 什么是量化交易数据管道？

量化交易数据管道，是一整套从数据采集、清洗、存储到计算、服务的自动化流程。它的任务只有一个：把原始数据变成策略能用的「燃料」。

我个人习惯把数据管道比作「炼油厂」。原油（原始行情）进来，经过蒸馏（清洗）、催化（计算）、储存（存储），最后变成汽油（信号）送到引擎（策略）里。任何一个环节出问题，车都跑不起来。

1.2 核心目标：低延迟、高可用、可扩展

这三个词，是衡量数据管道好坏的「金标准」。我一个个说。

低延迟

做高频交易的朋友最懂这个。行情数据从交易所出来，到你策略收到，中间每多1微秒，可能就少赚一笔钱。我记得有个项目，客户要求行情端到端延迟不超过10微秒。我们折腾了整整两周，最后发现是网卡中断合并（interrupt coalescing）没关，白白多了5微秒延迟。

避坑指南：我曾经以为只要用C++写采集程序就够快了，结果发现内存分配策略不对，延迟抖动大得离谱。后来改用预分配内存池，才把抖动压下去。

高可用

数据管道不能断。哪怕断1秒，你的策略可能就错过了关键行情。我见过最惨的案例——某团队数据管道挂了5分钟，恰好那5分钟出了个大新闻，市场剧烈波动。等他们恢复，策略已经亏了7位数。

高可用怎么做？冗余、容错、自动切换。说白了就是「鸡蛋别放一个篮子里」。

可扩展

你今天处理10个品种，明天可能要处理100个。今天每秒1万笔行情，明天可能10万笔。管道设计时就要考虑「加机器就能扛」。

嗯，这里要注意：可扩展不是「以后再说」，而是架构层面一开始就要留好接口。我见过太多团队，数据量一上来就重构管道，那叫一个痛苦。

目标	关键指标	常见瓶颈
低延迟	端到端延迟 < 1ms（高频）	网络、序列化、内存分配
高可用	可用性 > 99.99%	单点故障、网络分区
可扩展	水平扩展线性度 > 0.9	数据分片、状态共享

1.3 典型架构：五层模型

一个成熟的数据管道，通常分为五层。我画了张图，你看一眼就明白了。

采集层

这是管道的「水龙头」。行情数据从交易所、数据商、甚至爬虫进来。我建议至少做两路冗余——主路断了自动切备路，别让单点故障搞死你。

采集层常见的技术栈：C++/Rust写高性能采集器，或者用成熟的中间件比如NATS、Kafka做缓冲。我个人偏好Kafka，因为它的持久化机制能帮你扛住下游消费慢的情况。

清洗层

原始数据脏得很。重复数据、乱序数据、错误数据，什么都有。清洗层就是「过滤器」。

举个例子：某交易所的行情偶尔会发重复的tick。如果你不做去重，策略可能同一个价格交易两次。我见过有人用时间戳+价格+成交量做hash去重，效果不错。

清洗层还有个重要任务——时间对齐。不同数据源的时间戳格式可能不一样，有的用纳秒，有的用毫秒。统一成一种格式，不然后面计算层会疯掉。

存储层

清洗完的数据要存下来。这里我强烈推荐时序数据库，比如InfluxDB、ClickHouse、TimescaleDB。别用MySQL存行情数据，查询性能差得离谱。

存储层有个经典问题：冷热数据怎么处理？热数据（最近1天的）放SSD，冷数据（1个月前的）放HDD或者对象存储。我习惯用分层存储策略，能省不少钱。

-- 冷热分层示例（ClickHouse）
CREATE TABLE trades (
    symbol String,
    price Float64,
    volume UInt64,
    ts DateTime
) ENGINE = MergeTree
PARTITION BY toYYYYMM(ts)
ORDER BY (symbol, ts)
TTL ts + INTERVAL 30 DAY TO VOLUME 'cold';

计算层

这是策略工程师最关心的层。实时计算因子、聚合指标、生成信号。计算层分两种模式：

• 实时计算：用Flink、Spark Streaming或者自研的流计算引擎。延迟要求高的场景，我建议用Flink，它的状态管理做得很好。
• 批量计算：回测、因子分析用批处理。Spark或者Dask都行，看你的数据量。

嗯，这里有个坑：实时计算和批量计算的结果要能对得上。我见过团队实时算的因子和回测算的因子差了0.1%，排查了三天发现是浮点数精度问题。后来统一用Decimal128才解决。

服务层

最后一层，把计算好的数据「喂」给策略。服务层通常提供REST API或者WebSocket接口。我个人更推荐WebSocket，因为行情数据是流式的，轮询太浪费带宽。

服务层还要做权限控制——谁可以订阅什么数据，订阅多少路，都要管。不然某个策略订阅了所有品种的深度行情，带宽直接被打满。

1.4 一个简单的数据管道示例

说了这么多理论，来个实际例子。假设我们要搭建一个处理A股Level-2行情的数据管道：

# 伪代码：数据管道核心流程
def data_pipeline():
    # 1. 采集
    raw_ticks = collect_from_exchange("sh.600519")
    
    # 2. 清洗
    clean_ticks = dedup(raw_ticks)          # 去重
    clean_ticks = align_timestamp(clean_ticks)  # 时间对齐
    
    # 3. 存储
    store_to_clickhouse(clean_ticks)
    
    # 4. 计算
    vwap = calculate_vwap(clean_ticks, window=10)
    volatility = calculate_volatility(clean_ticks, window=20)
    
    # 5. 服务
    publish_to_strategy(vwap, volatility)

这个例子虽然简单，但五脏俱全。实际生产环境里，每一层都要做大量优化。比如采集层要处理几十个品种的并发连接，清洗层要应对每秒几万笔tick的去重，存储层要保证写入不丢数据……

好了，第一章就聊到这儿。数据管道的概念和架构，你心里应该有个谱了。后面我们会一步步深入每一层的实现细节。

---

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