4. 数据采集与整合：交易数据源

做交易监控系统，最头疼的往往不是规则怎么写，而是数据从哪来、怎么来、来了之后怎么处理。我见过太多项目，规则写得天花乱坠，结果数据源没接好，监控成了摆设。今天咱们就聊聊这个基础但关键的问题。

4.1 交易数据源：三大核心数据流

交易监控系统要监控什么？说白了就是三样东西：订单簿、成交记录、行情数据。这三者构成了监控系统的数据基石。

4.1.1 订单簿（Order Book）

订单簿记录了市场上所有未成交的买卖挂单。它反映了市场的深度和流动性。我个人习惯把订单簿比作「市场的呼吸」——每一笔挂单的撤单、新增、修改，都透露着交易者的意图。

订单簿的数据结构通常包含：

• 买盘（Bid）：按价格从高到低排列
• 卖盘（Ask）：按价格从低到高排列
• 每个价格档位的挂单量
• 时间戳：精确到毫秒甚至微秒

4.1.2 成交记录（Trade Log）

成交记录是已经发生的交易。它是最权威的数据源，因为每一笔成交都对应着真实的资金流动。成交记录通常包含：

字段	说明	示例
交易ID	全局唯一标识	TXN202403150001
时间戳	成交时间	2024-03-15 09:30:00.123
交易对	买卖标的	BTC/USDT
价格	成交价格	65000.12
数量	成交数量	0.5
买卖方向	Buy/Sell	Buy

4.1.3 行情数据（Market Data）

行情数据是经过聚合和计算的衍生数据。比如每分钟的K线、涨跌幅、成交量等。它不像订单簿和成交记录那么原始，但胜在轻量和易用。

嗯，这里要注意：行情数据虽然方便，但它是「二手数据」。如果监控规则对精度要求极高，我建议直接用成交记录。我曾经踩过坑——用行情数据做监控，结果因为聚合窗口的偏差，漏掉了一笔异常交易。

4.2 数据清洗与标准化

数据源接进来了，但问题才刚刚开始。不同交易所、不同系统的数据格式千差万别。有的用时间戳是毫秒，有的是微秒；有的价格字段是字符串，有的是浮点数。不洗一下，根本没法用。

4.2.1 常见的数据脏问题

• 缺失值：某些字段为空，比如成交记录的买卖方向缺失
• 异常值：价格出现负数，或者数量超出合理范围
• 重复数据：同一笔成交被推送了两次
• 时序乱序：后发生的交易先到达系统

4.2.2 标准化的核心步骤

我一般把标准化分为三步：

1. 字段映射：把不同源的数据字段统一命名。比如A交易所的「price」和B交易所的「px」都映射为「trade_price」。
2. 类型转换：统一数据类型。时间戳全部转为毫秒级整数，价格和数量统一为Decimal类型（避免浮点数精度问题）。
3. 去重与排序：按交易ID去重，按时间戳排序。这一步在实时流处理中尤其重要。

// 标准化示例（伪代码）
function standardizeTrade(rawTrade) {
    return {
        tradeId: rawTrade.trade_id || rawTrade.tid,
        timestamp: normalizeTimestamp(rawTrade.ts || rawTrade.time),
        price: new Decimal(rawTrade.price || rawTrade.px),
        quantity: new Decimal(rawTrade.qty || rawTrade.vol),
        side: rawTrade.side === 'B' ? 'BUY' : 'SELL'
    };
}

4.3 实时流处理：Kafka 实战

数据清洗完了，怎么实时地喂给监控系统？这里就要请出我们的主角——Kafka。

为什么选Kafka？说白了，它天生就是干这个的。高吞吐、低延迟、可持久化，还能保证消息的顺序性。交易监控对实时性要求极高，Kafka的毫秒级延迟完全够用。

4.3.1 数据流架构设计

我习惯把数据流分为三层：

• 数据接入层：从交易所或数据源接收原始数据，写入Kafka的原始Topic
• 数据清洗层：消费原始Topic，清洗标准化后写入清洗Topic
• 监控消费层：监控规则引擎消费清洗Topic，进行实时检测

下面是我画的一个架构图，你看一眼就明白了：

4.3.2 Kafka 配置要点

在实际项目中，Kafka的配置直接影响监控的实时性和可靠性。我分享几个关键配置：

配置项	推荐值	说明
partitions	3-6	根据数据量调整，保证并行消费
replication.factor	3	保证数据不丢，至少3副本
acks	all	所有副本确认后才算写入成功
retries	3	写入失败重试次数

4.3.3 实时流处理的避坑指南

做实时流处理，有几个坑我踩过，你注意避开：

• 背压问题：数据量突然暴增时，消费速度跟不上生产速度。解决方案是增加分区数，或者使用Kafka的背压机制。
• 数据延迟：网络抖动导致数据延迟到达。我建议在监控规则中加入时间窗口判断，比如「如果某笔交易延迟超过5秒，标记为可疑」。
• 重复消费：消费者重启后可能重复消费。用幂等性设计来避免重复处理，或者在业务层面做去重。

嗯，说到重复消费，我记得有一次线上事故——消费者重启后重复处理了同一批数据，导致监控系统连续报警了半小时。后来我们在消费端加了去重缓存，才彻底解决这个问题。

4.4 小结

数据采集与整合，说白了就是三个字：接、洗、流。接进来，洗干净，流出去。每一步都有坑，但每一步也都有成熟的解决方案。你想想看，如果数据源都没搞定，后面的监控规则写得再好也是白搭。

我个人习惯在项目初期就把数据链路跑通，哪怕规则简单点，也要先保证数据是可靠的。数据可靠了，监控才有意义。