4、数据清洗：异常值过滤、时间戳对齐、乱序数据重排、去重策略
行情数据进了内存，你以为就能直接用了吗？
太天真了。我见过太多团队，数据还没跑完就崩了。原因很简单——原始行情数据，脏得离谱。
这一章，我们就来聊聊数据清洗的四个核心动作：异常值过滤、时间戳对齐、乱序数据重排、去重策略。说白了，就是让数据变得「干净、有序、可用」。
4.1 异常值过滤：别让一颗老鼠屎坏了一锅粥
什么是异常值？举个例子：某只股票突然出现一笔成交价是前一笔的100倍。这明显是交易所的撮合异常，或者是数据源传输错误。如果你不处理，策略会直接爆仓。
我个人习惯用三种方法过滤异常值：
- 阈值过滤：设定价格上下限，比如涨跌幅超过10%的直接丢弃。
- 统计过滤：用Z-score或IQR（四分位距）识别离群点。
- 业务规则过滤：比如买卖价差为负、成交量小于等于0，这些明显不合理。
核心原则：宁可漏掉一笔正常数据，也不要放进来一笔异常数据。高频交易中，一次错误信号可能让你亏掉一周的利润。
我在项目中遇到过一件事：某次回测结果异常漂亮，年化收益超过200%。我一开始还以为是策略牛逼，后来一查，原来是数据里混入了大量「虚拟成交」——交易所测试环境的数据没被过滤掉。嗯，从那以后，我每次清洗都会加一道「数据源校验」。
def filter_outliers(df, price_col='price', volume_col='volume', z_thresh=3):
"""
基于Z-score过滤异常价格和成交量
"""
from scipy import stats
# 价格异常过滤
price_z = np.abs(stats.zscore(df[price_col]))
df = df[price_z < z_thresh]
# 成交量异常过滤（成交量不能为负或零）
df = df[df[volume_col] > 0]
# 业务规则：买卖价差必须为正
if 'ask_price' in df.columns and 'bid_price' in df.columns:
df = df[df['ask_price'] > df['bid_price']]
return df避坑指南：我曾经用全局Z-score过滤，结果发现早盘和尾盘的波动率天然不同，导致大量正常数据被误杀。后来改成「滚动窗口Z-score」，效果好了很多。
4.2 时间戳对齐：让不同数据源「说同一种语言」
行情数据来自多个交易所、多个数据商。每个数据源的时间戳格式可能都不一样：有的用Unix毫秒，有的用字符串，有的甚至带时区偏移。
你想想看，如果A交易所的tick是14:30:00.123，B交易所的tick是14:30:00.456，你拿它们做套利策略，差这333毫秒，可能就错过了最佳开仓点。
时间戳对齐，我一般分三步走：
- 统一格式：全部转为Unix纳秒或毫秒整数，避免浮点数精度问题。
- 时区归一：统一转为UTC时间，或者统一转为交易所本地时间。
- 对齐粒度：根据策略需求，对齐到毫秒、微秒或纳秒级别。
| 数据源 | 原始格式 | 对齐后格式 |
|---|---|---|
| 交易所A | 2024-01-15 14:30:00.123 | 1705314600123000000 (纳秒) |
| 交易所B | 1705314600456 (毫秒) | 1705314600456000000 (纳秒) |
| 数据商C | 2024-01-15T06:30:00.456Z | 1705314600456000000 (纳秒) |
def align_timestamps(df, time_col='timestamp', target_unit='ns'):
"""
将时间戳统一对齐到目标精度
"""
# 如果是字符串格式，先解析
if df[time_col].dtype == 'object':
df[time_col] = pd.to_datetime(df[time_col])
# 转为纳秒整数
if target_unit == 'ns':
df[time_col] = df[time_col].astype(np.int64)
# 排序（为后续乱序重排做准备）
df = df.sort_values(time_col).reset_index(drop=True)
return df注意：不同数据源的时钟可能存在偏差。我见过某数据商的时间戳比交易所慢了整整2秒。建议定期用交易所的行情快照做时钟校准。
4.3 乱序数据重排：把「插队」的数据放回原位
网络延迟、数据包重传、多路并发写入……这些都会导致数据乱序。简单说，就是后发生的事件先到了，先发生的事件反而后到。
乱序数据对策略的影响有多大？举个例子：你收到一笔14:30:00.500的成交，然后收到一笔14:30:00.300的成交。如果你不重排，策略会以为价格先涨后跌，但实际上价格是先跌后涨。方向完全反了。
我建议的做法是：
- 滑动窗口排序：维护一个固定大小的窗口（比如100ms），窗口内的数据按时间戳排序后再输出。
- 延迟输出：故意延迟50-100ms再输出数据，给乱序数据一个「归队」的时间。
- 心跳检测：如果超过一定时间没收到新数据，强制输出当前窗口内的所有数据。
class ReorderBuffer:
"""
乱序数据重排缓冲区
"""
def __init__(self, window_ms=100, max_latency_ms=200):
self.buffer = []
self.window_ms = window_ms
self.max_latency_ms = max_latency_ms
self.last_output_time = 0
def add(self, tick):
self.buffer.append(tick)
self.buffer.sort(key=lambda x: x['timestamp'])
# 检查是否可以输出
now = tick['timestamp']
while self.buffer:
if now - self.buffer[0]['timestamp'] > self.window_ms * 1e6:
yield self.buffer.pop(0)
else:
break
def flush(self):
"""强制输出所有剩余数据"""
for tick in self.buffer:
yield tick
self.buffer = []避坑指南：我曾经把窗口设得太小（10ms），结果乱序数据根本来不及归队，策略信号频繁闪烁。后来改成50ms窗口+100ms延迟，效果稳定多了。具体数值要根据你的网络延迟和交易所数据频率来调。
4.4 去重策略：别让同一笔数据算两次
数据重复的原因很多：网络重传、数据商重复推送、多路订阅导致同一笔数据被接收多次。
重复数据会导致什么后果？成交量翻倍、价格被重复计算、策略信号失真。最要命的是，如果你用重复数据做回测，结果会严重偏离真实情况。
我常用的去重策略有三种：
- 基于唯一ID去重：交易所通常会给每笔成交分配一个唯一的TradeID或SequenceNumber。这是最可靠的方法。
- 基于时间戳+价格+成交量去重：如果三个字段完全一致，大概率是重复数据。
- 基于滑动窗口去重：维护一个最近N条数据的哈希集合，新数据进来先查重。
def deduplicate(df, id_col='trade_id', time_col='timestamp'):
"""
基于唯一ID去重
"""
before = len(df)
df = df.drop_duplicates(subset=[id_col])
after = len(df)
print(f"去重前: {before} 条, 去重后: {after} 条, 移除: {before - after} 条重复数据")
return df
def deduplicate_by_fields(df, fields=['timestamp', 'price', 'volume']):
"""
基于多字段组合去重（适用于没有唯一ID的场景）
"""
return df.drop_duplicates(subset=fields)核心原则：去重一定要在时间戳对齐和乱序重排之后做。否则，同一笔数据可能因为时间戳格式不同而被误判为不同数据。
知识体系总览
下面这张图，把数据清洗的四个步骤串起来了。你可以把它当作一个「清洗流水线」：
这张图展示了数据清洗的完整流程。你可能会问：为什么去重要放在最后？
原因很简单：如果先做去重，再做时间戳对齐和乱序重排，同一笔数据可能因为时间戳格式不同而被误判为不同数据，导致去重失败。顺序很重要，别搞反了。
个人经验：我建议你在清洗流水线的每个步骤都加一个「计数器」，记录过滤/对齐/重排/去重的数据量。这样一旦策略表现异常，你可以快速定位是哪个环节出了问题。
公众号:蓝海资料掘金营，微信deep3321