2. 变量衍生基础：原始数据清洗、缺失值处理、异常值识别与截尾

原始数据是因子研究的起点，但金融数据天生不完美：缺失、异常、噪声无处不在。本章将系统讲解变量衍生前的数据预处理三步骤——清洗、缺失值处理、异常值识别与截尾，确保后续衍生变量建立在可靠的数据基础之上。

2.1 原始数据清洗：从脏数据到可用数据

数据清洗是预处理的第一步，主要解决格式不一致、重复记录、逻辑错误等问题。金融数据常见的脏数据类型包括：

2.2 缺失值处理：识别、分类与填充策略

金融数据中的缺失值并非随机出现，往往与停牌、退市、数据采集故障等因素相关。处理缺失值前，需先识别缺失模式：

• 完全随机缺失（MCAR）：缺失与观测数据无关，如随机传输错误。可直接删除或简单填充。
• 随机缺失（MAR）：缺失依赖于其他观测变量，如小市值股票更易缺失分析师覆盖数据。需用模型预测填充。
• 非随机缺失（MNAR）：缺失与未观测到的因素相关，如业绩差的公司主动不披露数据。需谨慎处理，避免引入偏差。

常用缺失值填充方法对比：

# Python 示例：缺失值处理流程
import pandas as pd
import numpy as np

# 加载日频股票数据
df = pd.read_csv('stock_data.csv', parse_dates=['date'])
df.set_index(['stock_id', 'date'], inplace=True)

# 步骤1：识别缺失模式
missing_ratio = df.isnull().mean().sort_values(ascending=False)
print("缺失率排名：\n", missing_ratio[missing_ratio > 0])

# 步骤2：按变量类型选择填充策略
# 价格类变量：前向填充 + 线性插值
df['close'] = df.groupby('stock_id')['close'].ffill().interpolate(method='linear')

# 成交量类变量：前向填充，若仍缺失则填充0
df['volume'] = df.groupby('stock_id')['volume'].ffill().fillna(0)

# 财务指标类变量：截面中位数填充
df['pe_ratio'] = df.groupby('date')['pe_ratio'].transform(
    lambda x: x.fillna(x.median())
)

# 步骤3：标记仍存在的缺失（用于后续分析）
df['close_missing_flag'] = df['close'].isnull().astype(int)

2.3 异常值识别：统计方法与领域知识结合

异常值（Outlier）是偏离正常分布的数据点，可能由数据错误、极端市场事件或真实但罕见的交易行为引起。识别异常值需结合统计方法和金融领域知识。

常用异常值识别方法：

• Z-Score 方法：假设数据服从正态分布，将 |Z| > 3 的点视为异常。适用于对称分布数据。
• IQR（四分位距）方法：将低于 Q1 - 1.5×IQR 或高于 Q3 + 1.5×IQR 的点视为异常。对偏态分布更稳健。
• MAD（中位数绝对偏差）方法：使用中位数替代均值，对异常值本身不敏感，适合金融数据。
• 百分比截尾法：直接截取上下 1% 或 5% 的分位数，简单但可能丢失信息。

# Python 示例：异常值识别与可视化
from scipy import stats

# 计算日收益率
df['return'] = df.groupby('stock_id')['close'].pct_change()

# 方法1：Z-Score 识别
z_scores = np.abs(stats.zscore(df['return'].dropna()))
outlier_z = z_scores > 3
print(f"Z-Score 方法识别异常值数量：{outlier_z.sum()}")

# 方法2：IQR 方法
Q1 = df['return'].quantile(0.25)
Q3 = df['return'].quantile(0.75)
IQR = Q3 - Q1
lower_bound = Q1 - 1.5 * IQR
upper_bound = Q3 + 1.5 * IQR
outlier_iqr = (df['return'] < lower_bound) | (df['return'] > upper_bound)
print(f"IQR 方法识别异常值数量：{outlier_iqr.sum()}")

# 方法3：MAD 方法
median = df['return'].median()
mad = np.median(np.abs(df['return'] - median))
modified_z = 0.6745 * (df['return'] - median) / mad
outlier_mad = np.abs(modified_z) > 3.5
print(f"MAD 方法识别异常值数量：{outlier_mad.sum()}")

2.4 异常值截尾：截断与缩尾技术

识别异常值后，需决定如何处理。截尾（Truncation）和缩尾（Winsorization）是两种主流方法：

# Python 示例：缩尾处理
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%缩尾
df['return_winsorized'] = df.groupby('date')['return'].transform(
    lambda x: winsorize_series(x, limits=[0.01, 0.01])
)

# 对比处理前后的统计量
print("原始收益率统计：")
print(df['return'].describe())
print("\n缩尾后收益率统计：")
print(df['return_winsorized'].describe())

2.5 数据预处理流程总览

以下 SVG 图展示了从原始数据到可用变量的完整预处理流程，以及各步骤之间的迭代关系：

2.6 实战案例：A股日频数据预处理

以下案例展示完整的预处理流程在 A 股日频数据上的应用：

# 完整预处理流程示例
import pandas as pd
import numpy as np
from scipy import stats

def preprocess_stock_data(df, date_col='date', stock_col='stock_id',
                          price_cols=['open','high','low','close'],
                          vol_col='volume',
                          outlier_method='mad',
                          winsor_limits=[0.01, 0.01]):
    """
    股票数据完整预处理流程
    """
    # 1. 数据清洗
    # 去重
    df = df.drop_duplicates(subset=[stock_col, date_col])
    # 逻辑检查：价格必须为正
    for col in price_cols:
        df = df[df[col] > 0]
    # 成交量检查
    df = df[df[vol_col] >= 0]

    # 2. 缺失值处理
    # 价格类：前向填充 + 线性插值
    for col in price_cols:
        df[col] = df.groupby(stock_col)[col].ffill().interpolate(method='linear')
    # 成交量：前向填充，剩余填充0
    df[vol_col] = df.groupby(stock_col)[vol_col].ffill().fillna(0)

    # 3. 计算收益率
    df['return'] = df.groupby(stock_col)['close'].pct_change()

    # 4. 异常值识别与截尾
    if outlier_method == 'mad':
        # MAD 方法
        median = df['return'].median()
        mad = np.median(np.abs(df['return'] - median))
        modified_z = 0.6745 * (df['return'] - median) / mad
        df['return_outlier'] = np.abs(modified_z) > 3.5
    elif outlier_method == 'iqr':
        Q1 = df['return'].quantile(0.25)
        Q3 = df['return'].quantile(0.75)
        IQR = Q3 - Q1
        lower = Q1 - 1.5 * IQR
        upper = Q3 + 1.5 * IQR
        df['return_outlier'] = (df['return'] < lower) | (df['return'] > upper)

    # 5. 缩尾处理（截面维度）
    df['return_winsor'] = df.groupby(date_col)['return'].transform(
        lambda x: x.clip(
            lower=x.quantile(winsor_limits[0]),
            upper=x.quantile(1 - winsor_limits[1])
        )
    )

    # 6. 标记缺失率高的股票（可选）
    missing_threshold = 0.2  # 20% 缺失率
    stock_missing = df.groupby(stock_col)['close'].apply(
        lambda x: x.isnull().mean()
    )
    high_missing_stocks = stock_missing[stock_missing > missing_threshold].index
    df['high_missing'] = df[stock_col].isin(high_missing_stocks)

    return df

# 使用示例
# df_clean = preprocess_stock_data(raw_df)

2.7 预处理质量评估指标

完成预处理后，需通过以下指标评估数据质量改善效果：