第四节：盘口数据清洗实战——去重与排序、买卖盘价差修复、零值/负值处理、涨跌停板特殊处理
盘口数据，说白了就是交易所实时推送的「买卖十档」信息。这东西看着简单，但你要是直接拿原始数据跑策略，十有八九要翻车。我做了这么多年量化，踩过的坑比吃过的盐还多。今天咱们就把盘口清洗的四个硬骨头啃下来。
核心原则：盘口数据清洗的目标不是「完美」，而是「可用」。我们要在保留有效信息的同时，剔除那些会搞崩模型的脏数据。
4.1 去重与排序：别让重复数据骗了你
先说说去重。你可能觉得这有什么难的？pandas 里 drop_duplicates 一调不就完了？嗯，没那么简单。
重复数据的来源：
- 交易所快照推送频率高于你的采样频率（比如你每秒采一次，但交易所每 100ms 推一次）
- 网络重传导致同一笔数据收到两次
- 数据源拼接时产生的重复行
我个人的习惯是：先按时间戳排序，再按「时间戳 + 买卖档位」联合去重。为什么先排序？因为原始数据的时间戳可能是乱序的，直接去重会丢掉有效数据。
import pandas as pd
def clean_duplicates(df):
"""
盘口数据去重函数
策略：按时间戳排序后，基于时间戳+买卖档位去重
"""
# 先确保时间戳是 datetime 类型
df['timestamp'] = pd.to_datetime(df['timestamp'])
# 按时间戳排序
df = df.sort_values('timestamp').reset_index(drop=True)
# 定义去重键：时间戳精确到毫秒 + 买卖方向 + 档位
df['dedup_key'] = df['timestamp'].astype(str) + '_' + df['side'] + '_' + df['level'].astype(str)
# 保留第一次出现的记录
df_clean = df.drop_duplicates(subset='dedup_key', keep='first')
# 删除辅助列
df_clean = df_clean.drop(columns=['dedup_key'])
print(f"去重前 {len(df)} 行，去重后 {len(df_clean)} 行，移除 {len(df) - len(df_clean)} 行重复")
return df_clean避坑指南：我曾经遇到过一个数据源，同一毫秒内出现两条完全相同的盘口数据。用 keep='first' 保留第一条没问题。但如果你发现同一毫秒内两条数据价格不同，那可能是交易所的「价格更新」——这种情况不应该去重，而是保留两条。
4.2 买卖盘价差修复：当买一价高于卖一价时
正常情况下，买一价应该低于卖一价，中间有个价差。但实际数据里，偶尔会出现买一价 ≥ 卖一价的情况。这要么是数据错误，要么是极端行情下的瞬间状态。
为什么会这样？说白了，就是数据推送的时序问题。比如卖一价先更新了，买一价还没来得及更新，你刚好在这个间隙采到了数据。
我的处理策略分三种情况：
| 情况 | 判断条件 | 处理方式 |
|---|---|---|
| 轻微倒挂 | 买一价 - 卖一价 ≤ 最小变动价位 × 2 | 取两者均值作为买卖中间价，修复档位 |
| 严重倒挂 | 买一价 - 卖一价 > 最小变动价位 × 2 | 标记为异常，用前一条有效数据填充 |
| 连续倒挂 | 连续 3 条以上数据都倒挂 | 可能是数据源问题，整段标记并剔除 |
def fix_spread(df, tick_size=0.01):
"""
修复买卖盘价差异常
tick_size: 最小变动价位，比如股票是0.01，比特币是0.1
"""
df = df.copy()
# 提取买一和卖一
bid1 = df[df['side'] == 'bid'].groupby('timestamp')['price'].max().reset_index()
ask1 = df[df['side'] == 'ask'].groupby('timestamp')['price'].min().reset_index()
merged = pd.merge(bid1, ask1, on='timestamp', suffixes=('_bid', '_ask'))
# 计算价差
merged['spread'] = merged['price_ask'] - merged['price_bid']
# 轻微倒挂修复
mask_minor = (merged['spread'] < 0) & (merged['spread'].abs() <= 2 * tick_size)
merged.loc[mask_minor, 'mid_price'] = (merged['price_bid'] + merged['price_ask']) / 2
merged.loc[mask_minor, 'price_bid'] = merged['mid_price'] - tick_size
merged.loc[mask_minor, 'price_ask'] = merged['mid_price'] + tick_size
# 严重倒挂：用前一条有效数据
mask_severe = merged['spread'] < -2 * tick_size
merged.loc[mask_severe, 'price_bid'] = merged['price_bid'].shift(1)
merged.loc[mask_severe, 'price_ask'] = merged['price_ask'].shift(1)
print(f"修复轻微倒挂 {mask_minor.sum()} 条，严重倒挂 {mask_severe.sum()} 条")
return merged注意：修复后的数据一定要打上「已修复」标签。我见过有人修复完直接当原始数据用，回测跑得漂漂亮亮，实盘一跑就崩——因为修复逻辑在实盘里根本不存在。
4.3 零值/负值处理：这些数据有毒
盘口数据里出现零值或负值，基本可以断定是数据错误。价格不可能为零，成交量也不可能为负。但现实就是这么魔幻——我就遇到过某数据源在系统初始化时，把未更新的字段默认填了 0。
处理原则：
- 价格字段：零值或负值 → 直接剔除该条记录
- 成交量字段：零值可以保留（表示该档位无挂单），负值 → 取绝对值或剔除
- 买卖方向：如果 side 字段为空或异常值 → 根据价格与中间价的关系推断
def handle_zero_negative(df):
"""
处理零值和负值
返回清洗后的数据和异常记录数
"""
initial_len = len(df)
anomalies = []
# 价格字段：剔除零值和负值
price_cols = [col for col in df.columns if 'price' in col.lower()]
for col in price_cols:
mask = (df[col] <= 0)
anomalies.append(mask.sum())
df = df[~mask]
# 成交量字段：负值取绝对值
volume_cols = [col for col in df.columns if 'volume' in col.lower() or 'qty' in col.lower()]
for col in volume_cols:
neg_mask = df[col] < 0
df.loc[neg_mask, col] = df.loc[neg_mask, col].abs()
anomalies.append(neg_mask.sum())
removed = initial_len - len(df)
print(f"移除 {removed} 条价格异常记录，修正 {sum(anomalies)} 个负值成交量")
return df个人经验：零值处理要分场景。如果是「盘口深度」的零值，比如卖五档之后全是零，那其实是正常现象——表示没有更深的价格了。但如果是「成交价」出现零，那绝对是脏数据。
4.4 涨跌停板特殊处理：别把涨停当异常
涨跌停板是 A 股特有的机制，也是盘口清洗里最容易出问题的地方。你想想看，涨停的时候买一价等于涨停价，卖一价可能为空或者也是涨停价——这在正常逻辑里是「异常」，但在涨停板逻辑里是「正常」。
处理要点：
- 识别涨跌停状态：通过昨收价和涨跌幅限制计算涨停价和跌停价
- 涨停时：卖一价可能缺失，买一价等于涨停价，此时不进行价差修复
- 跌停时：买一价可能缺失，卖一价等于跌停价，同样跳过修复
- 封板后：成交量可能骤降，盘口深度变浅，这些都不是数据错误
def handle_limit_up_down(df, prev_close, limit_pct=0.1):
"""
处理涨跌停板数据
prev_close: 昨收价
limit_pct: 涨跌幅限制，A股一般是0.1（10%）
"""
# 计算涨跌停价格
limit_up = prev_close * (1 + limit_pct)
limit_down = prev_close * (1 - limit_pct)
# 标记涨跌停状态
df['is_limit_up'] = (df['side'] == 'bid') & (df['price'] >= limit_up - 0.01)
df['is_limit_down'] = (df['side'] == 'ask') & (df['price'] <= limit_down + 0.01)
# 涨跌停时，跳过价差修复
df['skip_spread_fix'] = df['is_limit_up'] | df['is_limit_down']
# 对于涨停板，如果卖一缺失，用涨停价填充
limit_up_mask = df['is_limit_up'] & (df['side'] == 'ask')
df.loc[limit_up_mask, 'price'] = limit_up
# 对于跌停板，如果买一缺失，用跌停价填充
limit_down_mask = df['is_limit_down'] & (df['side'] == 'bid')
df.loc[limit_down_mask, 'price'] = limit_down
print(f"识别涨停 {df['is_limit_up'].sum()} 条，跌停 {df['is_limit_down'].sum()} 条")
return df我曾经犯过的错：刚开始做 A 股策略时，我没处理涨跌停板。结果涨停那天，我的策略检测到「买一价等于卖一价」，自动触发了价差修复逻辑，把价格往下调了。第二天开盘直接低开，策略亏得亲妈都不认识。从那以后，涨跌停板处理成了我清洗流程里的必修课。
知识体系总览
下面这张图概括了盘口数据清洗的完整流程，你可以把它当作操作手册来用：
嗯，这四个步骤走完，你的盘口数据基本就干净了。但记住，清洗只是第一步。下一阶段我们要做的是特征工程——从这些干净数据里挖出真正有用的信号。不过那是后话了，先把今天的代码跑通再说。