4. 数据准备：获取股票行情数据（日频收益率）、获取因子数据（以动量因子为例）、数据对齐与清洗

数据是量化分析的基石。在构建 Rank IC 监控系统之前，必须完成三部分核心工作：获取股票日频行情数据、计算因子数据（以动量因子为例）、数据对齐与清洗。本章将使用 Python 的 pandas、numpy 和 akshare（或 tushare）库，演示完整的数据准备流程。

4.1 获取股票日频收益率数据

日频收益率是计算 Rank IC 的基础。我们以沪深 300 成分股为例，获取最近 1 年的日频行情数据，并计算对数收益率。

import akshare as ak
import pandas as pd
import numpy as np
from datetime import datetime, timedelta

# 1. 获取沪深300成分股列表
stock_300 = ak.index_stock_cons_csindex("000300")
stock_list = stock_300['成分券代码'].tolist()
print(f"沪深300成分股数量: {len(stock_list)}")

# 2. 设置时间范围
end_date = datetime.now().strftime('%Y%m%d')
start_date = (datetime.now() - timedelta(days=365)).strftime('%Y%m%d')

# 3. 批量获取日频行情（示例取前5只，实际全量需循环）
def fetch_daily_data(stock_code, start, end):
    try:
        df = ak.stock_zh_a_hist(symbol=stock_code, period="daily",
                                start_date=start, end_date=end, adjust="qfq")
        df['stock_code'] = stock_code
        df['trade_date'] = pd.to_datetime(df['日期'])
        df['close'] = df['收盘'].astype(float)
        df['ret'] = np.log(df['close'] / df['close'].shift(1))
        return df[['trade_date', 'stock_code', 'close', 'ret']]
    except:
        return pd.DataFrame()

# 实际使用时建议用多线程或分批获取
sample_stocks = stock_list[:5]
all_data = []
for code in sample_stocks:
    data = fetch_daily_data(code, start_date, end_date)
    if not data.empty:
        all_data.append(data)

price_df = pd.concat(all_data, ignore_index=True)
price_df = price_df.dropna(subset=['ret'])
print(price_df.head())

4.2 计算动量因子（Momentum Factor）

动量因子是最经典的截面因子之一。我们定义 过去 20 个交易日的累计收益率 作为动量信号，即 MOM_20 = (close_t / close_{t-20}) - 1。

# 计算动量因子：过去20日累计收益率
def calc_momentum(df, window=20):
    df = df.sort_values(['stock_code', 'trade_date']).copy()
    df['mom_20'] = df.groupby('stock_code')['close'].transform(
        lambda x: x / x.shift(window) - 1
    )
    return df

price_df = calc_momentum(price_df, window=20)
factor_df = price_df[['trade_date', 'stock_code', 'mom_20']].dropna()
print(factor_df.head())

4.3 数据对齐与清洗

数据对齐是确保因子值和未来收益率在时间上严格匹配的关键步骤。我们需要将因子数据与下一交易日的收益率对齐，并清洗异常值。

4.3.1 时间对齐：因子 vs 未来收益率

Rank IC 衡量的是 t 日因子值 与 t+1 日收益率 的截面秩相关性。因此，必须将因子数据向前对齐到未来收益率。

# 对齐：将因子数据与未来1日收益率合并
def align_factor_with_future_ret(price_df, factor_df):
    # 先获取未来1日收益率
    ret_df = price_df[['trade_date', 'stock_code', 'ret']].copy()
    ret_df['future_date'] = ret_df['trade_date']  # 保留原日期用于对齐

    # 将收益率向前移动1天（即 t 日的因子对应 t+1 日的收益率）
    ret_df['trade_date'] = ret_df.groupby('stock_code')['trade_date'].shift(-1)

    # 合并因子与未来收益率
    merged = pd.merge(factor_df, ret_df, on=['trade_date', 'stock_code'], how='inner')
    merged = merged.dropna(subset=['mom_20', 'ret'])
    return merged

aligned_df = align_factor_with_future_ret(price_df, factor_df)
print(f"对齐后数据量: {len(aligned_df)} 行")
print(aligned_df.head())

4.3.2 清洗规则

清洗是保证因子质量的重要环节。我们执行以下标准清洗流程：

• 缺失值处理： 删除因子或收益率为 NaN 的行。
• 极端值处理： 对因子值进行 MAD（中位数绝对偏差） 去极值，剔除超过 5 倍 MAD 的样本。
• 停牌股处理： 剔除当日收益率为 0 且前一日收盘价未变的股票（连续停牌）。
• ST 股处理： 剔除股票代码中包含 ST 或 *ST 的股票。

def clean_data(df):
    # 1. 删除缺失值
    df = df.dropna(subset=['mom_20', 'ret']).copy()

    # 2. MAD去极值（按日期分组）
    def mad_clip(group):
        median = group['mom_20'].median()
        mad = (group['mom_20'] - median).abs().median()
        upper = median + 5 * mad
        lower = median - 5 * mad
        group['mom_20'] = group['mom_20'].clip(lower, upper)
        return group

    df = df.groupby('trade_date', group_keys=False).apply(mad_clip)

    # 3. 剔除停牌股（当日ret为0且close未变）
    df = df[~((df['ret'] == 0) & (df['close'] == df['close'].shift(1)))]

    # 4. 剔除ST股（假设stock_code为纯数字，ST股代码含字母）
    df = df[df['stock_code'].str.match(r'^\d{6}$')]

    return df.reset_index(drop=True)

cleaned_df = clean_data(aligned_df)
print(f"清洗后数据量: {len(cleaned_df)} 行")

4.4 数据准备流程总览

下图展示了从原始行情到可用于 Rank IC 计算的完整数据流水线：

4.5 数据质量检查清单

在进入 Rank IC 计算之前，建议执行以下检查，确保数据质量：

至此，我们已经完成了从原始行情到清洗后因子数据的全部准备工作。下一阶段将基于这些数据，正式计算 Rank IC 序列并评估其稳定性。