4. 数据清洗与预处理：缺失值处理、异常值检测、数据对齐与重采样

高频数据在采集、传输与存储过程中极易受到市场微观结构噪声、网络延迟、交易中断等因素影响，导致数据质量下降。本章聚焦于高频数据清洗与预处理的四大核心环节：缺失值处理、异常值检测、数据对齐与重采样。掌握这些技术是构建稳健量化策略的基础。

4.1 缺失值处理

高频数据中的缺失值通常由交易暂停、数据推送中断或记录错误引起。处理策略需根据缺失模式与业务场景选择。

4.2 异常值检测

高频数据中的异常值可能源于错误报价、闪电崩盘或数据录入错误。常用检测方法包括Winsorization与Z-score。

4.2.1 Winsorization（缩尾处理）

将超出指定百分位数的极端值替换为该百分位数的边界值，而非直接删除。适用于保留样本量同时抑制极端值影响。

import numpy as np
import pandas as pd

def winsorize_series(s, limits=[0.01, 0.01]):
    """
    对序列进行缩尾处理
    limits: [lower_percentile, upper_percentile]
    """
    lower = s.quantile(limits[0])
    upper = s.quantile(1 - limits[1])
    return s.clip(lower, upper)

# 示例：对1分钟收益率进行1%缩尾
returns = pd.Series([0.001, -0.05, 0.002, 0.1, -0.003, 0.15])
cleaned_returns = winsorize_series(returns, limits=[0.05, 0.05])
print(cleaned_returns)

4.2.2 Z-score方法

基于数据点与均值的标准差距离识别异常值。适用于近似正态分布的数据。

def zscore_outlier_detection(s, threshold=3):
    """
    基于Z-score检测异常值
    返回：布尔序列，True表示异常
    """
    z_scores = (s - s.mean()) / s.std()
    return np.abs(z_scores) > threshold

# 示例：检测收益率中的异常值
outliers = zscore_outlier_detection(returns, threshold=3)
print("异常值索引:", returns[outliers].index.tolist())

4.3 数据对齐

高频数据通常来自多个交易所或数据源，时间戳不一致、时钟偏差、延迟差异导致对齐困难。对齐的核心目标是使不同序列在统一时间轴上可比。

• 精确时间戳对齐：将每个数据点分配到最近的整数时间戳（如毫秒级）。
• 时钟偏差校正：通过NTP同步或统计方法估计并消除系统间时钟偏移。
• 多源数据对齐：以主数据源时间轴为基准，对其他数据源进行前向填充或插值。

# 多源数据对齐示例
import pandas as pd

# 模拟两个数据源
ts1 = pd.to_datetime(['2024-01-01 09:30:00.100', '2024-01-01 09:30:00.200'])
ts2 = pd.to_datetime(['2024-01-01 09:30:00.150', '2024-01-01 09:30:00.250'])

df1 = pd.DataFrame({'price': [100.1, 100.3]}, index=ts1)
df2 = pd.DataFrame({'volume': [500, 700]}, index=ts2)

# 以100ms为基准对齐
aligned = pd.DataFrame(index=pd.date_range('09:30:00', '09:30:01', freq='100ms'))
aligned['price'] = df1['price'].reindex(aligned.index, method='ffill')
aligned['volume'] = df2['volume'].reindex(aligned.index, method='ffill')
print(aligned)

4.4 重采样

重采样将不规则高频数据转换为固定频率的时间序列，是特征工程与模型输入的必备步骤。

# 重采样为1分钟K线
import pandas as pd
import numpy as np

# 模拟Tick数据
np.random.seed(42)
tick_index = pd.date_range('2024-01-01 09:30:00', periods=1000, freq='100ms')
tick_data = pd.Series(np.random.randn(1000).cumsum() + 100, index=tick_index, name='price')

# 重采样为1分钟OHLC
minute_ohlc = tick_data.resample('1min').ohlc()
print(minute_ohlc.head())

4.5 实战流程总结

完整的高频数据清洗与预处理流水线应包含以下步骤：

1. 原始数据加载：读取Tick级或分钟级数据，检查时间戳完整性。
2. 缺失值处理：根据字段类型选择前向填充或插值。
3. 异常值检测：先Winsorization抑制极端值，再Z-score或MAD标记异常。
4. 数据对齐：统一时间轴，校正时钟偏差。
5. 重采样：按策略需求生成固定频率K线或快照。
6. 质量验证：检查清洗后数据的统计特征与原始数据的一致性。