4. 数据清洗与标准化：去重、乱序处理、时间戳对齐、缺失值填充，以及统一的数据模型定义
数据清洗，说白了就是给行情数据「洗澡」。
你想想看，交易所推送过来的原始数据，那叫一个乱。重复的、乱序的、缺胳膊少腿的，什么都有。我刚开始做实时行情系统那会儿，就吃过这个亏——数据没洗干净就直接喂给策略引擎，结果回测曲线漂亮得不行，实盘一跑就崩。嗯，从那以后，我把数据清洗列为了最高优先级。
4.1 去重：别让同一笔成交算两次
为什么会有重复数据？原因很多。比如网络重传、交易所网关的冗余推送、或者我们自己的采集节点做了主备切换。我个人习惯的做法是：在数据入口处就干掉重复，不要等到后面再处理。
核心思路：用「交易所代码 + 产品代码 + 时间戳 + 序列号」作为唯一键。
来看一段我常用的去重逻辑（伪代码）：
// 基于布隆过滤器的去重，适合高吞吐场景
class Deduplicator {
BloomFilter<String> filter = new BloomFilter<>(1000000, 0.001);
boolean isDuplicate(TickData tick) {
String key = tick.exchange + ":" + tick.symbol
+ ":" + tick.timestamp + ":" + tick.seqNo;
if (filter.mightContain(key)) {
return true; // 大概率重复
}
filter.put(key);
return false;
}
}避坑指南：我曾经用HashMap做去重，结果内存爆了。布隆过滤器虽然有小概率误判，但胜在省内存。如果你对精度要求极高，可以用Redis的Set结构，配合过期时间自动清理。
4.2 乱序处理：时间旅行者必须被纠正
行情数据乱序，是分布式系统的常态。比如上海和深圳的行情，经过不同的网络路径到达你的服务器，时间戳晚的反而先到。这很常见。
我处理乱序的思路很简单：给数据一个「等待窗口」。
// 乱序处理：等待窗口机制
class OutOfOrderHandler {
long windowMs = 50; // 等待50毫秒
PriorityQueue<TickData> buffer = new PriorityQueue<>(
(a, b) -> Long.compare(a.timestamp, b.timestamp)
);
void handle(TickData tick) {
buffer.offer(tick);
// 清理窗口外的数据
while (!buffer.isEmpty() &&
buffer.peek().timestamp < System.currentTimeMillis() - windowMs) {
emit(buffer.poll()); // 按时间顺序输出
}
}
}注意：窗口大小不是固定的。我建议根据网络延迟的P99值来动态调整。比如你的网络P99延迟是30ms，那窗口设到50ms就够用了。设太大，实时性会受影响；设太小，乱序数据还是漏过去了。
4.3 时间戳对齐：让不同数据源说同一种「时间语言」
不同交易所的时间戳格式五花八门。有的用Unix毫秒，有的用纳秒，还有的用字符串格式的"2024-01-15 09:30:00.123"。更头疼的是，有些交易所的时钟本身就不准。
我的做法是：统一转为纳秒级时间戳，并做时钟同步校正。
| 数据源 | 原始格式 | 标准化后 |
|---|---|---|
| 上海交易所 | 毫秒时间戳 | 纳秒时间戳 |
| 深圳交易所 | 字符串格式 | 纳秒时间戳 |
| 期货CTP | 微秒时间戳 | 纳秒时间戳 |
// 时间戳对齐工具
class TimestampNormalizer {
long normalize(String rawTimestamp, String source) {
switch (source) {
case "SH":
return Long.parseLong(rawTimestamp) * 1_000_000L;
case "SZ":
return parseStringTimestamp(rawTimestamp); // 字符串转纳秒
case "CTP":
return Long.parseLong(rawTimestamp) * 1_000L;
default:
throw new IllegalArgumentException("未知数据源");
}
}
}一个小技巧：我习惯在标准化后的时间戳里保留原始时间戳的副本。万一后面排查问题，还能追溯回去。这个习惯救过我不少次。
4.4 缺失值填充：别让空数据搞崩你的策略
行情数据缺失，最常见的是某只股票突然停牌，或者网络闪断导致几笔数据没收到。直接跳过？不行。策略引擎需要连续的数据流。
我常用的填充策略有三种：
- 前值填充：用上一笔数据填充。适合价格变化不大的场景。
- 线性插值：根据前后数据计算中间值。适合成交量等连续指标。
- 标记填充：填充一个特殊值（如-1），让下游自己决定怎么处理。
// 缺失值填充器
class MissingValueFiller {
TickData lastValid = null;
TickData fill(TickData current) {
if (current.price == null) {
if (lastValid != null) {
current.price = lastValid.price; // 前值填充
current.volume = 0; // 成交量填0
} else {
current.price = 0.0; // 没有历史数据，填0
}
}
lastValid = current;
return current;
}
}我曾经踩过的坑：用前值填充时，如果连续缺失超过5笔，价格可能已经严重偏离真实值。建议设置一个「最大连续填充次数」，超过这个次数就标记为异常数据，而不是继续填充。
4.5 统一数据模型：给所有数据一个「家」
数据清洗的最终目的，是让所有数据都遵循同一个数据模型。我习惯用Protocol Buffers来定义，因为它序列化快、跨语言兼容好。
// 统一行情数据模型（Protobuf定义）
message UnifiedTick {
string exchange = 1; // 交易所代码
string symbol = 2; // 产品代码
int64 timestamp_ns = 3; // 标准化后的纳秒时间戳
double price = 4; // 最新成交价
int64 volume = 5; // 成交量
double bid_price = 6; // 买一价
double ask_price = 7; // 卖一价
int32 bid_volume = 8; // 买一量
int32 ask_volume = 9; // 卖一量
int64 seq_no = 10; // 交易所序列号
int32 quality = 11; // 数据质量标记：0=正常，1=填充，2=异常
}为什么这么设计？quality字段是我后来加的。有了它，下游策略可以知道哪些数据是「原装」的，哪些是「加工」过的。这个设计思路，我建议你也用上。
4.6 整体流程：一张图看懂
下面这张图，是我做数据清洗的标准流程。每个环节都经过实战检验。
数据清洗这件事，看起来琐碎，但它是整个实时行情系统的基石。我见过太多团队在算法和策略上花大功夫，结果数据源没洗干净，一切白搭。记住一句话：垃圾进，垃圾出。把数据清洗做好，后面的工作才能事半功倍。
最后分享一个经验：数据清洗的代码，一定要写单元测试。我每次改清洗逻辑，都会用历史数据跑一遍回归测试。别嫌麻烦，这个习惯能帮你省下无数排查问题的时间。
公众号:蓝海数据掘金营，微信deep3321