📚 课程目录第4章数据清洗与预处理

4. 数据清洗与预处理：缺失值处理、异常值检测、去重、时间戳对齐、数据类型转换

实时交易数据在进入风控特征引擎之前，必须经过严格的清洗与预处理。原始数据流中常包含传感器空值、网络抖动导致的重复记录、不同交易所的时间偏差以及字段类型不匹配等问题。本章将系统讲解五大核心预处理步骤，确保特征计算的准确性与时效性。

核心原则： 实时风控场景下，预处理必须在毫秒级完成。优先使用向量化操作与流式处理框架（如 Kafka Streams、Flink），避免逐行循环。

4.1 缺失值处理

交易数据缺失通常由行情中断、数据源延迟或字段映射失败引起。处理策略需根据缺失率与业务含义选择：

策略	适用场景	实时实现方式	风险提示
前向填充	价格、成交量等连续序列	`df.ffill()` 或流式状态变量	长时间缺失会累积偏差
后向填充	回测或批处理场景	`df.bfill()`	实时中不可用（引入未来信息）
固定值填充	订单簿深度、标记位	`df.fillna(0)` 或 `-1`	可能掩盖异常状态
插值法	高频 tick 数据	`df.interpolate()`	计算开销较大，需控制窗口
丢弃行	缺失率 > 80% 或关键字段	`df.dropna()`	可能丢失交易信号

import pandas as pd
import numpy as np

# 模拟实时 tick 数据
tick_data = pd.DataFrame({
    'timestamp': pd.date_range('2025-01-01', periods=10, freq='100ms'),
    'bid': [100.1, np.nan, 100.3, np.nan, np.nan, 100.6, 100.7, np.nan, 100.9, 101.0],
    'ask': [100.2, 100.2, np.nan, 100.4, 100.5, np.nan, 100.8, 100.8, np.nan, 101.1],
    'volume': [100, 200, np.nan, 150, np.nan, 300, 250, np.nan, 180, 220]
})

# 前向填充 + 限制最大填充步数（防止陈旧数据）
tick_data['bid'] = tick_data['bid'].ffill(limit=3)
tick_data['ask'] = tick_data['ask'].ffill(limit=3)
tick_data['volume'] = tick_data['volume'].fillna(0)  # 成交量缺失视为 0

避坑： 实时风控中禁止使用 bfill() 或全局插值，它们会引入未来数据，导致回测过拟合与实盘失效。始终使用流式友好的前向填充或状态缓存。

4.2 异常值检测

异常值可能来自行情错误、网络闪断或恶意操纵。实时检测需兼顾灵敏度与计算效率，常用方法如下：

Z-Score 法： 适用于正态分布特征，实时维护滚动均值和标准差，阈值通常设为 3 或 3.5。
IQR 法： 对偏态分布鲁棒，使用滚动分位数（Q1, Q3），异常定义为超出 [Q1 - 1.5*IQR, Q3 + 1.5*IQR]。
固定阈值法： 根据业务规则硬编码，如单笔成交量超过历史均值 10 倍、买卖价差为负。
孤立森林（Isolation Forest）： 适合多维异常检测，但实时场景需使用增量版本或降采样。

def detect_anomalies_zscore(series, window=20, threshold=3.0):
    """滚动 Z-Score 异常检测"""
    rolling_mean = series.rolling(window=window).mean()
    rolling_std = series.rolling(window=window).std()
    z_scores = (series - rolling_mean) / rolling_std
    return z_scores.abs() > threshold

# 检测 bid 价格异常
anomaly_mask = detect_anomalies_zscore(tick_data['bid'], window=5, threshold=2.5)
tick_data['bid_anomaly'] = anomaly_mask
print(tick_data[tick_data['bid_anomaly']])

  实战建议： 将异常值标记为 NaN 或替换为前向有效值，而非直接丢弃。丢弃会破坏时间序列的连续性，影响后续特征如动量、波动率的计算。

4.3 去重

重复数据通常由网络重传、消息队列重复消费或交易所冗余推送引起。去重策略需区分精确去重与近似去重：

场景	去重键	方法	实时注意事项
同一交易所 tick	timestamp + symbol + exchange	`df.drop_duplicates(subset=[...])`	使用布隆过滤器减少内存
多源聚合数据	全局唯一 ID (UUID)	Redis 或内存 Set 去重	设置 TTL 防止内存泄漏
窗口聚合结果	窗口起始时间 + 分组键	幂等写入（覆盖模式）	确保窗口边界对齐

# 精确去重示例
tick_data = tick_data.drop_duplicates(subset=['timestamp', 'bid', 'ask'], keep='last')

# 流式场景使用布隆过滤器（需安装 pybloom_live）
from pybloom_live import ScalableBloomFilter
bloom = ScalableBloomFilter(mode=ScalableBloomFilter.SMALL_SET_GROWTH)
def is_duplicate(tick_id):
    if tick_id in bloom:
        return True
    bloom.add(tick_id)
    return False

4.4 时间戳对齐

不同数据源的时间戳可能存在微秒级偏差，对齐是计算跨市场价差、订单簿同步等特征的前提。对齐策略包括：

最近邻对齐： 将每个 tick 对齐到最近的整秒/整毫秒边界，使用 pd.DatetimeIndex.floor()。
线性插值对齐： 在固定频率时间点上插值，适用于高频连续数据。
异步消息队列对齐： 使用 Kafka 时间戳提取器或 Flink 的 Watermark 机制。

# 对齐到 100ms 时间桶
tick_data['timestamp'] = pd.to_datetime(tick_data['timestamp'])
tick_data.set_index('timestamp', inplace=True)
aligned = tick_data.resample('100ms').ffill().bfill()  # 先前向后向后填充边界

# 多源对齐示例：合并两个交易所的 tick
exchange_a = tick_data[['bid']].rename(columns={'bid': 'bid_a'})
exchange_b = tick_data[['ask']].rename(columns={'ask': 'ask_b'})
merged = pd.merge_asof(exchange_a, exchange_b, left_index=True, right_index=True, direction='nearest', tolerance=pd.Timedelta('50ms'))

性能提示： merge_asof 是实时对齐的利器，时间复杂度 O(n log n)，远优于笛卡尔积。在流式框架中，可使用 IntervalJoin 或 TemporalJoin 实现类似效果。

4.5 数据类型转换

原始数据流中的字段类型常与特征计算要求不匹配，需在预处理阶段统一转换：

原始类型	目标类型	转换方法	常见问题
字符串数字	float64	`pd.to_numeric(..., errors='coerce')`	非数字字符导致 NaN
int64	float64	`df['col'].astype('float64')`	大整数精度丢失
Unix 毫秒时间戳	datetime64[ns]	`pd.to_datetime(..., unit='ms')`	时区未指定
object 分类	category	`df['col'].astype('category')`	类别过多导致内存不降反升
bool	int8	`df['col'].astype('int8')`	缺失值无法转换

# 完整数据类型转换流水线
def convert_dtypes_pipeline(df):
    # 时间戳统一
    if 'timestamp' in df.columns:
        df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms', utc=True)
    # 价格字段转为 float64
    price_cols = ['bid', 'ask', 'last', 'high', 'low']
    for col in price_cols:
        if col in df.columns:
            df[col] = pd.to_numeric(df[col], errors='coerce')
    # 成交量转为 float32（节省内存）
    if 'volume' in df.columns:
        df['volume'] = df['volume'].astype('float32')
    # 标记位转为 int8
    flag_cols = ['is_buy', 'is_anomaly']
    for col in flag_cols:
        if col in df.columns:
            df[col] = df[col].astype('int8')
    return df

tick_data = convert_dtypes_pipeline(tick_data)
print(tick_data.dtypes)

4.6 综合实战：实时清洗流水线

将上述步骤整合为一个可复用的清洗函数，适用于 Kafka 消息的逐条处理：

def realtime_cleanse(msg: dict, state: dict) -> dict:
    """
    单条消息清洗函数，适用于流式处理
    state 维护滚动统计量（均值、标准差、布隆过滤器等）
    """
    # 1. 类型转换
    msg['bid'] = pd.to_numeric(msg.get('bid'), errors='coerce')
    msg['ask'] = pd.to_numeric(msg.get('ask'), errors='coerce')
    msg['volume'] = float(msg.get('volume', 0))

    # 2. 缺失值处理（前向填充）
    if pd.isna(msg['bid']):
        msg['bid'] = state.get('last_bid', 0.0)
    if pd.isna(msg['ask']):
        msg['ask'] = state.get('last_ask', 0.0)
    state['last_bid'] = msg['bid']
    state['last_ask'] = msg['ask']

    # 3. 异常值检测（滚动 Z-Score）
    state['bid_history'] = state.get('bid_history', []) + [msg['bid']]
    if len(state['bid_history']) > 20:
        state['bid_history'].pop(0)
    if len(state['bid_history']) >= 5:
        mean = np.mean(state['bid_history'])
        std = np.std(state['bid_history'])
        if std > 0 and abs(msg['bid'] - mean) / std > 3.0:
            msg['bid_anomaly'] = 1
            msg['bid'] = state['last_bid']  # 替换为前向值
        else:
            msg['bid_anomaly'] = 0

    # 4. 去重（布隆过滤器）
    tick_id = f"{msg['timestamp']}_{msg['symbol']}"
    if tick_id in state.get('bloom', set()):
        return None  # 丢弃重复
    state.setdefault('bloom', set()).add(tick_id)

    # 5. 时间戳对齐（向下取整到 100ms）
    ts = pd.to_datetime(msg['timestamp'], unit='ms')
    msg['timestamp_aligned'] = ts.floor('100ms').value // 10**6

    return msg

  生产级建议： 上述函数可直接嵌入 pandas_udf（PySpark）或 ProcessFunction（Flink）。对于超高频场景（> 10万 tick/秒），建议用 Cython 或 Numba 编译关键路径，或使用 Rust/Python 桥接库（如 Polars）。