4. 数据获取与预处理：构建高质量因子计算基础

量化因子的计算依赖干净、对齐的行情数据。本章将系统讲解如何获取股票日线与分钟线数据，并进行清洗、对齐与重采样，为后续XGBoost因子计算奠定数据基础。

4.1 数据获取：日线与分钟线行情

行情数据通常通过量化平台API或本地数据库获取。以下对比两种常用数据源的特征：

以下代码展示从本地数据库获取日线数据的标准流程：

import pandas as pd
import numpy as np
from sqlalchemy import create_engine

# 数据库连接
engine = create_engine('sqlite:///stock_data.db')

# 获取指定股票池的日线数据
def fetch_daily_data(stock_list, start_date, end_date):
    query = f"""
    SELECT ts_code, trade_date, open, high, low, close, vol, amount
    FROM daily_data
    WHERE ts_code IN ({','.join([f"'{s}'" for s in stock_list])})
      AND trade_date BETWEEN '{start_date}' AND '{end_date}'
    ORDER BY ts_code, trade_date
    """
    df = pd.read_sql(query, engine)
    df['trade_date'] = pd.to_datetime(df['trade_date'])
    return df

# 示例：获取5只股票2023年数据
stocks = ['000001.SZ', '600000.SH', '000002.SZ', '600036.SH', '601318.SH']
df_daily = fetch_daily_data(stocks, '2023-01-01', '2023-12-31')
print(f"获取数据量: {len(df_daily)} 条")
print(df_daily.head())

4.2 数据清洗：处理缺失值与异常值

原始行情数据常存在缺失值和异常值，直接影响因子计算的准确性。清洗流程分为三个步骤：

1. 缺失值检测：识别停牌、新股上市前、退市等导致的空值。
2. 缺失值处理：根据缺失比例选择填充或删除。
3. 异常值识别：检测价格跳空、成交量异常、复权错误等。

以下表格列出不同缺失场景的处理策略：

def clean_market_data(df):
    """清洗行情数据：处理缺失值和异常值"""
    df = df.copy()
    
    # 1. 按股票分组处理
    for code, group in df.groupby('ts_code'):
        idx = group.index
        
        # 2. 缺失值处理：前向填充停牌日
        df.loc[idx, ['open', 'high', 'low', 'close', 'vol', 'amount']] = \
            group[['open', 'high', 'low', 'close', 'vol', 'amount']].ffill()
        
        # 3. 异常值检测：价格超过前后日20%视为异常
        price_cols = ['open', 'high', 'low', 'close']
        for col in price_cols:
            pct_change = group[col].pct_change()
            outlier_mask = (pct_change.abs() > 0.20) & (pct_change.notna())
            df.loc[idx[outlier_mask], col] = np.nan
        
        # 4. 再次填充异常值（用前后均值）
        df.loc[idx, price_cols] = df.loc[idx, price_cols].interpolate(method='linear')
    
    # 5. 删除仍然存在NaN的行（如上市首日）
    df = df.dropna(subset=['close'])
    return df

df_clean = clean_market_data(df_daily)
print(f"清洗后数据量: {len(df_clean)} 条")

4.3 数据对齐：多股票时间序列对齐

因子计算要求所有股票在同一时间截面上具有可比性。数据对齐解决两个问题：

• 交易日对齐：不同股票因停牌导致交易日不一致。
• 字段对齐：确保每只股票在同一日期具有相同的字段集合。

标准对齐流程如下：

def align_market_data(df):
    """将多股票数据对齐到统一交易日历"""
    # 1. 获取全市场统一交易日历
    all_dates = df['trade_date'].unique()
    all_dates = sorted(all_dates)
    
    # 2. 构建股票-日期完整网格
    all_stocks = df['ts_code'].unique()
    full_index = pd.MultiIndex.from_product(
        [all_stocks, all_dates], 
        names=['ts_code', 'trade_date']
    )
    
    # 3. 对齐：将原始数据映射到完整网格
    df_aligned = df.set_index(['ts_code', 'trade_date'])
    df_aligned = df_aligned.reindex(full_index)
    
    # 4. 重置索引
    df_aligned = df_aligned.reset_index()
    df_aligned = df_aligned.sort_values(['ts_code', 'trade_date'])
    
    return df_aligned

df_aligned = align_market_data(df_clean)
print(f"对齐后数据量: {len(df_aligned)} 条")
print(f"股票数: {df_aligned['ts_code'].nunique()}, 交易日数: {df_aligned['trade_date'].nunique()}")

4.4 重采样：分钟线转日线与日线转周线

重采样用于将高频数据转换为低频数据，或统一不同数据源的频率。分钟线转日线是高频因子计算中的常见操作。

重采样规则如下：

def resample_minute_to_daily(df_minute):
    """将分钟线数据重采样为日线数据"""
    # 假设df_minute包含: ts_code, trade_time, open, high, low, close, vol, amount
    df_minute = df_minute.copy()
    df_minute['trade_date'] = df_minute['trade_time'].dt.date
    
    # 按股票和日期分组重采样
    def resample_group(group):
        return pd.Series({
            'open': group['open'].iloc[0],
            'high': group['high'].max(),
            'low': group['low'].min(),
            'close': group['close'].iloc[-1],
            'vol': group['vol'].sum(),
            'amount': group['amount'].sum()
        })
    
    df_daily = df_minute.groupby(['ts_code', 'trade_date']).apply(resample_group)
    df_daily = df_daily.reset_index()
    df_daily['trade_date'] = pd.to_datetime(df_daily['trade_date'])
    
    return df_daily

def resample_daily_to_weekly(df_daily):
    """将日线数据重采样为周线数据"""
    df_daily = df_daily.copy()
    df_daily['week'] = df_daily['trade_date'].dt.isocalendar().week.astype(int)
    df_daily['year'] = df_daily['trade_date'].dt.year
    
    def resample_weekly(group):
        return pd.Series({
            'open': group['open'].iloc[0],
            'high': group['high'].max(),
            'low': group['low'].min(),
            'close': group['close'].iloc[-1],
            'vol': group['vol'].sum(),
            'amount': group['amount'].sum(),
            'week_start': group['trade_date'].iloc[0]
        })
    
    df_weekly = df_daily.groupby(['ts_code', 'year', 'week']).apply(resample_weekly)
    df_weekly = df_weekly.reset_index()
    df_weekly['trade_date'] = df_weekly['week_start']
    
    return df_weekly

# 示例：分钟线转日线
# df_minute = fetch_minute_data(stocks, '2023-01-01', '2023-01-31')
# df_daily_from_minute = resample_minute_to_daily(df_minute)

4.5 实战案例：完整数据预处理流水线

将上述步骤整合为一个可复用的数据预处理流水线，直接输出可用于因子计算的DataFrame：

def data_preprocessing_pipeline(stock_list, start_date, end_date, 
                                 source='daily', resample_to=None):
    """
    完整数据预处理流水线
    
    Parameters:
    -----------
    stock_list : list
        股票代码列表
    start_date, end_date : str
        日期范围
    source : str
        'daily' 或 'minute'
    resample_to : str or None
        重采样目标频率：'daily', 'weekly', 'monthly'
    
    Returns:
    --------
    pd.DataFrame : 清洗对齐后的行情数据
    """
    # 1. 数据获取
    if source == 'daily':
        df = fetch_daily_data(stock_list, start_date, end_date)
    else:
        df = fetch_minute_data(stock_list, start_date, end_date)
        df = resample_minute_to_daily(df)
    
    # 2. 数据清洗
    df = clean_market_data(df)
    
    # 3. 数据对齐
    df = align_market_data(df)
    
    # 4. 重采样（如果需要）
    if resample_to == 'weekly':
        df = resample_daily_to_weekly(df)
    elif resample_to == 'monthly':
        df = resample_daily_to_monthly(df)
    
    # 5. 添加辅助字段
    df['returns'] = df.groupby('ts_code')['close'].pct_change()
    df['log_returns'] = np.log(df['close'] / df.groupby('ts_code')['close'].shift(1))
    df['turnover'] = df['vol'] / df.groupby('ts_code')['vol'].rolling(20).mean()
    
    return df

# 使用示例
df_ready = data_preprocessing_pipeline(
    stock_list=['000001.SZ', '600000.SH'],
    start_date='2023-01-01',
    end_date='2023-12-31',
    source='daily',
    resample_to=None
)
print(f"最终数据形状: {df_ready.shape}")
print(df_ready[['ts_code', 'trade_date', 'close', 'returns']].head())

完成本章的数据预处理后，你将获得一个干净、对齐、频率统一的行情数据集，可直接用于第5章的因子计算与XGBoost模型训练。