数据清洗基础：缺失值处理、异常值检测与时间序列对齐

原始金融数据几乎不可能直接用于回测。缺失值、异常值以及时间戳错位是三大常见污染源。本章从实战角度出发，建立一套可复用的清洗流水线，确保进入策略的信号与价格序列干净、连续、对齐。

2.1 缺失值处理

金融时间序列的缺失通常由停牌、数据源中断、节假日差异或撮合间隙造成。处理方式分为三类：删除、填充与插值。选择哪种方法取决于缺失比例与策略对数据完整性的敏感度。

2.1.1 前向填充（Forward Fill）

将最后一个非缺失值复制到后续缺失位置，直到遇到下一个有效值。这是回测中最安全的填充方式，因为它不会引入未来数据。

import pandas as pd
import numpy as np

# 模拟日频收盘价，含缺失
dates = pd.date_range('2024-01-01', periods=10, freq='D')
prices = [100.0, 101.5, np.nan, np.nan, 103.0, 104.2, np.nan, 106.8, 107.5, np.nan]
series = pd.Series(prices, index=dates)

# 前向填充
ffilled = series.ffill()
print(ffilled)

2.1.2 插值法（Interpolation）

当缺失值位于两个有效值之间且缺失长度可控时，线性插值能提供更自然的过渡。对于非等间隔时间序列，推荐使用时间加权插值。

# 线性插值
linear_interp = series.interpolate(method='linear')

# 时间加权插值（考虑实际时间间隔）
time_interp = series.interpolate(method='time')

# 仅对连续缺失不超过3个的点插值，超过则保留NaN
limited_interp = series.interpolate(method='linear', limit=3)

2.2 异常值检测

异常值在金融数据中可能代表真实极端事件（如闪崩、财报冲击），也可能是数据录入错误或撮合异常。检测后需要区分“清洗”与“保留”的边界。

2.2.1 3-Sigma 法则

假设数据服从正态分布，将均值 ± 3 倍标准差之外的点标记为异常。适用于收益率、因子值等近似对称分布。

def detect_3sigma(series, n_sigma=3):
    mean = series.mean()
    std = series.std()
    lower = mean - n_sigma * std
    upper = mean + n_sigma * std
    return series[(series < lower) | (series > upper)]

# 示例：检测日收益率异常
returns = series.pct_change().dropna()
outliers = detect_3sigma(returns)
print(f"检测到 {len(outliers)} 个异常点")

2.2.2 IQR 方法

基于四分位距，不受极端值影响，适合偏态分布。将低于 Q1 - 1.5×IQR 或高于 Q3 + 1.5×IQR 的点视为异常。

def detect_iqr(series, multiplier=1.5):
    Q1 = series.quantile(0.25)
    Q3 = series.quantile(0.75)
    IQR = Q3 - Q1
    lower = Q1 - multiplier * IQR
    upper = Q3 + multiplier * IQR
    return series[(series < lower) | (series > upper)]

outliers_iqr = detect_iqr(returns)
print(f"IQR方法检测到 {len(outliers_iqr)} 个异常点")

2.3 时间序列对齐

多资产回测中最隐蔽的错误来源：不同数据源的时间戳精度不一致，导致信号与价格错位。对齐的目标是让所有序列在同一个时间轴上严格对齐。

2.3.1 重采样与频率统一

将不同频率的数据统一到目标频率。日频数据对齐到交易日历，分钟数据对齐到交易所撮合时间。

# 将5分钟数据重采样为15分钟
df_5min = pd.DataFrame({
    'price': [100, 100.5, 101, 100.8, 101.2, 101.5]
}, index=pd.date_range('2024-01-01 09:30', periods=6, freq='5min'))

df_15min = df_5min.resample('15min').agg({
    'price': 'last'  # 使用最后价格作为该时段收盘价
})
print(df_15min)

2.3.2 时间戳标准化

不同交易所或数据商的时间戳可能包含时区、毫秒差异。统一转换为 UTC 或交易所本地时间，并截断到目标精度。

# 标准化时间戳：统一到交易所本地时间，截断到秒
df['timestamp'] = pd.to_datetime(df['timestamp'], utc=True)
df['timestamp'] = df['timestamp'].dt.tz_convert('Asia/Shanghai')
df['timestamp'] = df['timestamp'].dt.floor('1s')

2.3.3 对齐检查与漂移修正

对齐完成后，必须验证所有序列的时间索引完全一致。常见问题包括：节假日差异导致某资产缺失一天、盘中停牌导致分钟数据断层。

# 检查对齐完整性
def check_alignment(*series_list, freq='D'):
    """检查多个时间序列是否完全对齐"""
    base_index = series_list[0].index
    for i, s in enumerate(series_list[1:], 1):
        if not s.index.equals(base_index):
            missing = base_index.difference(s.index)
            extra = s.index.difference(base_index)
            print(f"序列 {i}: 缺失 {len(missing)} 个时间点, 多余 {len(extra)} 个时间点")
            return False
    return True

# 对齐修正：取所有序列时间索引的交集
common_index = df1.index.intersection(df2.index).intersection(df3.index)
df1_aligned = df1.loc[common_index]
df2_aligned = df2.loc[common_index]

2.4 清洗流水线实战模板

将上述步骤整合为一个可复用的清洗函数，按顺序执行：对齐 → 缺失值处理 → 异常值检测 → 标记/替换。

def clean_pipeline(df, price_col='close', freq='D', 
                   fill_method='ffill', outlier_method='iqr'):
    """
    完整数据清洗流水线
    参数:
        df: DataFrame，必须包含时间索引
        price_col: 价格列名
        freq: 目标频率
        fill_method: 'ffill' 或 'linear'
        outlier_method: '3sigma' 或 'iqr'
    返回:
        清洗后的DataFrame，新增 outlier_flag 列
    """
    # 1. 时间序列对齐：重采样到目标频率
    df = df.resample(freq).last()
    
    # 2. 缺失值处理
    if fill_method == 'ffill':
        df[price_col] = df[price_col].ffill()
    elif fill_method == 'linear':
        df[price_col] = df[price_col].interpolate(method='linear', limit=3)
    
    # 3. 异常值检测
    returns = df[price_col].pct_change().dropna()
    if outlier_method == '3sigma':
        mean = returns.mean()
        std = returns.std()
        lower = mean - 3 * std
        upper = mean + 3 * std
    else:  # iqr
        Q1 = returns.quantile(0.25)
        Q3 = returns.quantile(0.75)
        IQR = Q3 - Q1
        lower = Q1 - 1.5 * IQR
        upper = Q3 + 1.5 * IQR
    
    # 标记异常，但不删除（保留决策权给策略）
    df['outlier_flag'] = ((returns < lower) | (returns > upper)).reindex(df.index, fill_value=False)
    
    return df

# 使用示例
cleaned_df = clean_pipeline(raw_df, price_col='close', freq='D',
                            fill_method='ffill', outlier_method='iqr')