多因子模型实战指南：从理论到Python实现

import pandas as pd
import numpy as np
import tushare as ts
from sklearn.preprocessing import StandardScaler
from statsmodels.api import OLS

# 获取股票数据
def get_stock_data(stock_code, start_date, end_date):
    pro = ts.pro_api('your_token')
    df = pro.daily(ts_code=stock_code, start_date=start_date, end_date=end_date)
    return df

# 计算因子暴露
def calculate_factor_exposures(stock_returns, factor_data):
    """
    计算股票对各个因子的暴露度
    stock_returns: 个股收益率序列
    factor_data: 因子收益率数据框
    """
    # 合并数据
    merged_data = pd.merge(stock_returns, factor_data, 
                          left_index=True, right_index=True, 
                          how='inner')
    
    # 标准化因子
    scaler = StandardScaler()
    factor_columns = ['MKT', 'SMB', 'HML', 'UMD']
    merged_data[factor_columns] = scaler.fit_transform(merged_data[factor_columns])
    
    # 回归分析
    X = merged_data[factor_columns]
    y = merged_data['return']
    model = OLS(y, X).fit()
    
    return model.params, model.rsquared

def factor_effectiveness_test(factor_scores, forward_returns, periods=1):
    """
    检验因子的预测能力
    factor_scores: 因子得分（如市值、账面市值比等）
    forward_returns: 未来收益
    periods: 预测周期
    """
    # 分组回测
    factor_scores = factor_scores.rank(method='first')
    groups = pd.qcut(factor_scores, 5, labels=False)
    
    # 计算各组平均收益
    group_returns = forward_returns.groupby(groups).mean()
    
    # 计算多空组合收益
    long_short_return = group_returns.iloc[-1] - group_returns.iloc[0]
    
    # IC分析
    ic = factor_scores.corr(forward_returns, method='spearman')
    
    return {
        'group_returns': group_returns,
        'long_short_return': long_short_return,
        'ic': ic
    }

def factor_preprocessing(factor_data):
    """
    因子预处理流程
    """
    # 1. 去极值（Winsorize）
    factor_data = factor_data.clip(
        lower=factor_data.quantile(0.01),
        upper=factor_data.quantile(0.99)
    )
    
    # 2. 标准化
    scaler = StandardScaler()
    factor_data = pd.DataFrame(
        scaler.fit_transform(factor_data),
        columns=factor_data.columns,
        index=factor_data.index
    )
    
    # 3. 行业中性化
    factor_data = industry_neutralization(factor_data, industry_codes)
    
    # 4. 填充缺失值
    factor_data = factor_data.fillna(factor_data.median())
    
    return factor_data