2. 数据准备与清洗：金融数据源介绍、数据对齐与去极值、缺失值处理与标准化

在量化因子研究中，数据质量直接决定因子有效性与策略稳定性。本章系统讲解金融数据源的选择标准、多源数据对齐方法、异常值处理技术以及数据标准化流程，为后续变量衍生与IC分析奠定坚实基础。

2.1 金融数据源介绍

量化研究常用的金融数据源可分为三类：

数据类别	典型数据源	频率	关键字段
行情数据	Wind、Tushare、聚宽、RiceQuant	日频/分钟频	开盘价、收盘价、最高价、最低价、成交量、成交额
财务数据	Wind、CSMAR、东方财富Choice	季频/年频	营业收入、净利润、总资产、净资产、每股收益
另类数据	通联数据、朝阳永续、舆情API	日频/事件驱动	分析师预期、龙虎榜、新闻情绪、供应链数据

2.2 数据对齐：多源异构数据的统一处理

不同数据源在时间戳、复权方式、停牌处理上存在差异，必须进行严格对齐。对齐流程包含以下步骤：

1. 时间戳标准化： 统一为 YYYY-MM-DD 格式，去除盘后数据中的多余时间标记。
2. 复权处理： 全部使用后复权价格，保证历史收益率计算的一致性。
3. 停牌/退市标记： 停牌日价格填充为 NaN，退市股票在退市日之后剔除。
4. 索引对齐： 以交易日历为基准，使用 pandas.merge 或 join 进行外连接，缺失交易日填充 NaN。

import pandas as pd

# 示例：行情数据与财务数据对齐
trade_calendar = pd.read_csv('trade_dates.csv', parse_dates=['date'])
price_data = pd.read_csv('daily_price.csv', parse_dates=['date'])
fundamental_data = pd.read_csv('quarterly_fundamentals.csv', parse_dates=['report_date'])

# 将财务数据向前填充至交易日
fundamental_aligned = fundamental_data.set_index('report_date').reindex(
    trade_calendar['date'], method='ffill'
).reset_index()

# 合并行情与财务数据
merged_data = pd.merge(price_data, fundamental_aligned, on=['date', 'stock_code'], how='inner')

2.3 去极值：MAD法与百分位法

金融数据常包含极端报价、数据录入错误或异常波动，去极值是因子计算前的必要步骤。常用方法有两种：

方法	原理	适用场景	参数
MAD法	基于中位数绝对偏差，对异常值不敏感	偏态分布、含较多极端值的数据	通常取 n=3~5 倍MAD
百分位法	直接截断上下百分位，简单直观	正态分布或近似正态的数据	常用 1%/99% 或 0.5%/99.5%

import numpy as np

def mad_winsorize(series, n=5):
    """MAD法去极值：将超出 n 倍 MAD 的值替换为边界值"""
    median = series.median()
    mad = np.median(np.abs(series - median))
    upper = median + n * mad
    lower = median - n * mad
    return series.clip(lower, upper)

def percentile_winsorize(series, lower=0.01, upper=0.99):
    """百分位法去极值"""
    lo = series.quantile(lower)
    hi = series.quantile(upper)
    return series.clip(lo, hi)

2.4 缺失值处理

缺失值来源包括停牌、新股上市、财务数据未更新等。处理策略需根据缺失比例和业务含义选择：

• 删除法： 缺失比例 < 1% 且随机分布时，直接删除该行。
• 截面填充： 使用同行业、同市值股票的均值或中位数填充。
• 时序填充： 使用前向填充（ffill）或后向填充（bfill），适用于财务数据。
• 模型预测： 缺失比例较高时，使用线性回归或KNN预测填充。

2.5 标准化：Z-score与Rank标准化

标准化使不同量纲的因子具有可比性，是因子合成与IC分析的前提。两种主流方法对比如下：

方法	公式	输出范围	对异常值敏感度	适用场景
Z-score	(x - μ) / σ	理论上无界	高	数据近似正态、已去极值
Rank标准化	rank(x) / N	[0, 1]	低	偏态分布、含异常值

from scipy.stats import rankdata

def zscore_normalize(series):
    """Z-score标准化"""
    return (series - series.mean()) / series.std()

def rank_normalize(series):
    """Rank标准化：将因子值映射到[0,1]均匀分布"""
    ranks = rankdata(series)
    return (ranks - 1) / (len(ranks) - 1)

2.6 完整数据清洗流水线

将上述步骤组合为可复用的清洗流水线，确保每次因子计算前数据状态一致：

def data_cleaning_pipeline(df, group_col='date', value_col='factor_value',
                           outlier_method='mad', mad_n=5,
                           fill_method='cross_sectional',
                           norm_method='zscore'):
    """
    完整数据清洗流水线
    步骤：去极值 → 缺失值填充 → 标准化
    """
    # 1. 截面去极值
    if outlier_method == 'mad':
        df[value_col] = df.groupby(group_col)[value_col].transform(
            lambda x: mad_winsorize(x, n=mad_n))
    elif outlier_method == 'percentile':
        df[value_col] = df.groupby(group_col)[value_col].transform(
            lambda x: percentile_winsorize(x))

    # 2. 缺失值处理
    if fill_method == 'cross_sectional':
        df[value_col] = df.groupby(group_col)[value_col].transform(
            lambda x: x.fillna(x.median()))
    elif fill_method == 'ffill':
        df[value_col] = df.groupby('stock_code')[value_col].transform(
            lambda x: x.fillna(method='ffill'))

    # 3. 标准化
    if norm_method == 'zscore':
        df[value_col] = df.groupby(group_col)[value_col].transform(
            lambda x: zscore_normalize(x))
    elif norm_method == 'rank':
        df[value_col] = df.groupby(group_col)[value_col].transform(
            lambda x: rank_normalize(x))

    return df