4、事件源与数据接入：市场数据源（行情、订单簿、新闻）、数据接入层设计、数据清洗与标准化
做量化交易这些年，我踩过最大的坑，就是数据。你策略再牛，模型再漂亮，数据源一崩，全白搭。今天咱们就聊聊数据接入这件事——说白了，就是怎么把外面乱七八糟的市场数据，变成你系统里能用的干净事件流。
4.1 市场数据源：行情、订单簿、新闻
先说说数据源。我个人习惯把市场数据分成三类：行情数据、订单簿数据、新闻数据。这三类数据的特点完全不同，接入方式也天差地别。
4.1.1 行情数据（Tick & K线）
行情数据是最基础的。Tick数据就是每一笔成交的快照，包含价格、成交量、时间戳。K线数据则是聚合后的结果，比如1分钟、5分钟K线。
这里有个坑——时间戳的精度问题。我在项目中遇到过，某交易所给的时间戳是毫秒级的，但另一家是微秒级的。你如果直接拿来做事件对齐，会出大问题。我建议统一转成纳秒级时间戳，或者至少保证精度一致。
4.1.2 订单簿数据（Order Book）
订单簿数据比行情数据复杂得多。它包含买一卖一到买N卖N的完整深度信息。我见过不少团队，只接行情不接订单簿，结果策略在流动性差的时候直接崩了。
订单簿数据有两个关键点：
- 快照 vs 增量：全量快照数据量大，但容易对齐；增量更新效率高，但容易丢数据。我建议两者结合——先拿快照，再用增量更新。
- 深度层级：一般接10-20层就够了，太深的数据噪音大，反而影响策略。
4.1.3 新闻数据（News Feed）
新闻数据是典型的非结构化数据。它不像行情那样有固定的格式，可能是文本、图片、甚至视频。做事件驱动策略时，新闻数据往往用来做事件检测——比如“美联储加息”、“公司财报超预期”。
接入新闻数据时，我建议关注三点：
- 数据源可靠性：新闻源可能随时断流，要有备用源。
- 延迟要求：新闻数据对延迟要求不高，但必须保证顺序。
- 结构化处理：原始新闻需要做NLP处理，提取关键事件。
4.2 数据接入层设计
数据接入层，说白了就是你的数据“水管”。水管设计得好，数据流就顺畅；设计得不好，到处漏水。
我一般把接入层设计成三层架构：
| 层级 | 职责 | 技术选型 |
|---|---|---|
| 连接层 | 管理数据源连接、心跳、重连 | WebSocket、TCP、REST |
| 协议层 | 解析数据格式、校验完整性 | Protobuf、JSON、Thrift |
| 分发层 | 将数据路由到不同消费者 | Kafka、RabbitMQ、ZeroMQ |
嗯，这里要注意——连接层最容易出问题。你想想看，交易所的WebSocket连接可能因为网络波动断开，如果没有自动重连机制，你的策略就会一直用旧数据跑，那后果...你懂的。
4.2.1 连接管理
我习惯用连接池来管理数据源连接。每个数据源对应一个连接池，池子里维护多个连接。这样做的好处是：
- 一个连接断了，自动切换到另一个
- 可以同时订阅多个数据流
- 方便做负载均衡
举个例子，如果你同时接入币安和OKX的行情，每个交易所开2个WebSocket连接，总共4个连接。连接池会自动管理这些连接的健康状态。
4.2.2 数据缓冲与背压
数据接入层必须处理背压问题。说白了就是——数据来得太快，下游处理不过来怎么办？
我建议用有界队列做缓冲。队列满了就丢弃旧数据，或者直接拒绝新数据。千万别用无界队列，否则内存会爆。
// 伪代码示例：带背压的数据接入
class DataIngestion {
private val buffer = new BoundedQueue[MarketData](capacity = 10000)
def onData(data: MarketData): Unit = {
if (!buffer.offer(data)) {
// 队列满了，丢弃旧数据
logger.warn("Buffer full, dropping data")
}
}
def consume(): MarketData = {
buffer.poll() // 非阻塞获取
}
}4.3 数据清洗与标准化
数据清洗，是我觉得最枯燥但又最重要的一步。你想想看，不同交易所的数据格式千奇百怪——有的用Unix时间戳，有的用ISO 8601；有的价格用字符串，有的用整数。不统一的话，策略根本没法跑。
4.3.1 常见数据问题
我在项目中遇到过这些典型问题：
- 缺失值：某段时间没有行情数据，可能是交易所宕机了
- 异常值：价格突然跳变100倍，明显是数据错误
- 重复数据：同一个Tick被推送了两次
- 乱序数据：后发生的事件先到达
针对这些问题，我一般这样处理：
| 问题类型 | 处理方式 |
|---|---|
| 缺失值 | 前向填充或插值，超过阈值则标记为无效 |
| 异常值 | 基于统计方法（如3σ）或业务规则过滤 |
| 重复数据 | 基于唯一ID去重 |
| 乱序数据 | 使用时间窗口排序，延迟处理 |
4.3.2 标准化模型
我建议定义一个统一的数据模型，所有数据源都转换成这个模型。这样下游策略只需要处理一种格式。
// 统一的市场数据模型
case class MarketData(
source: String, // 数据源标识
symbol: String, // 交易对
eventType: String, // 事件类型：tick/orderbook/news
timestamp: Long, // 纳秒级时间戳
price: Double, // 统一为Double
volume: Double, // 统一为Double
extra: Map[String, Any] // 扩展字段
)标准化过程中，我特别强调时间戳的统一。所有数据源的时间戳，我都转成UTC纳秒级。为什么？因为不同时区、不同精度的时间戳混在一起，事件排序会乱得一塌糊涂。
4.4 知识体系总览
最后，我用一张图来总结本章的核心逻辑。这张图展示了从数据源到标准化事件流的完整链路。
这张图里，从上到下就是数据从源头到策略的完整链路。每个环节都有对应的技术方案和注意事项。我个人觉得，数据接入层和清洗层是最容易出问题的，但也是最能体现架构师功力的地方。
好了，这一章的内容就到这里。数据接入这件事，说白了就是“把脏活累活干好”。你把这些基础打牢了，后面的策略开发才能事半功倍。