《股票多因子模型实战：Python 核心代码解析》 笔记
所属分类 quant
浏览量 13
《股票多因子模型实战：Python 核心代码解析》 陆一潇
聚焦于股票多因子模型的实战落地，以 Python 为工具，从因子构建、数据处理、模型验证到实盘应用，
系统拆解了多因子模型的核心逻辑与代码实现，是量化投资入门与进阶的实用参考书
量化投研全流程
数据→因子→模型→回测→优化
量化投研环境搭建（Python 库、数据来源）；
多因子模型基础（因子分类、有效因子特征）；
因子数据处理（清洗、标准化、中性化）；
单因子检验（IC 值、分层回测、换手率）；
多因子组合构建（因子加权、正交化、风险控制）；
模型回测与绩效评估；
实战优化（因子衰减、参数调优、实盘适配）
第一章：量化投研环境与数据基础
核心工具栈（Python）
类别 核心库 用途
数据获取 Tushare/Wind/JoinQuant A 股行情、财务、行业数据获取
数据处理 Pandas/Numpy 清洗、转换、因子计算
因子分析 Scikit-learn 标准化、回归、中性化
回测与可视化 Matplotlib/Backtrader 回测框架、收益曲线绘制
高性能计算 Dask/Swifter 大规模因子数据并行处理
大规模数据建议用Parquet格式存储（比 CSV 节省 70% 空间，读取速度提升 5 倍）
第二章：多因子模型基础理论
2.1 因子分类（A 股核心有效因子）
因子类型 典型代表 逻辑说明 收益特征
量价因子 动量（20/60 日）、反转 趋势延续 / 均值回归 高频、收益波动大
价值因子 PE/PB/PCF 低估股票长期收益更高 低频、抗跌
成长因子 营收增速、净利润增速 高成长企业盈利增长带动股价 中高频、行业分化
质量因子 ROE/ROIC/ 资产负债率 盈利质量高的企业长期表现好 低频、稳健
风险因子 市值（Size）、波动率 小市值 / 低波动股票超额收益 长期有效、易拥挤
情绪因子 换手率、融资融券余额 市场情绪反映资金流向 高频、易衰减
2.2 有效因子的核心特征
单调性：因子值与股票收益呈稳定的正 / 负相关；
持续性：因子有效性跨周期（至少 1 年以上）；
可解释性：有基本面 / 市场逻辑支撑（避免数据挖掘陷阱）；
低相关性：多因子间相关性 < 0.5（避免共线性）；
可交易性：考虑流动性（换手率 > 1%）、交易成本。
第三章：因子构建（代码 + 逻辑拆解）
3.1 量价因子（高频因子）
3.1.1 动量因子（避免未来函数）
def build_momentum_factor(df, window=20, lag=1):
"""
构建动量因子：window日累计收益率，滞后lag期（避免未来函数）
:param df: 含close_adj的股票数据框
:param window: 统计窗口（20/60/120日）
:param lag: 滞后期（至少1期，避免用当日数据预测当日）
:return: 含动量因子的DataFrame
"""
# 计算window日收益率
df['momentum_raw'] = df.groupby('ts_code')['close_adj'].pct_change(window)
# 滞后lag期，消除未来函数
df[f'momentum_{window}d'] = df.groupby('ts_code')['momentum_raw'].shift(lag)
# 剔除极端值（3σ原则）
df = remove_outlier(df, f'momentum_{window}d')
return df
def remove_outlier(df, col, sigma=3):
"""3σ剔除极端值：避免单只股票异常值影响"""
mean = df[col].mean()
std = df[col].std()
df.loc[(df[col] > mean + sigma*std) | (df[col] < mean - sigma*std), col] = np.nan
return df
3.1.2 换手率因子（情绪类）
def build_turnover_factor(df, window=5):
"""构建5日平均换手率因子（标准化）"""
# 计算滚动平均换手率
df['turnover_5d'] = df.groupby('ts_code')['turnover_rate'].rolling(window).mean().reset_index(0, drop=True)
# 标准化（跨股票可比）
df['turnover_5d_standard'] = df.groupby('trade_date')['turnover_5d'].apply(
lambda x: (x - x.mean()) / x.std()
)
return df
3.2 财务因子（低频因子）
3.2.1 核心问题：财务数据对齐
财务数据按 “报告期” 披露，需映射到 “交易日期”，书籍提供核心对齐逻辑：
def map_finance_to_trade(fin_df, trade_df):
"""
财务数据对齐交易日期：
- 年报/中报/季报披露后，才能用于因子计算
"""
# 1. 定义报告期对应的最早可用交易日期（如2023年报，2024-04-30后可用）
fin_df['report_period'] = pd.to_datetime(fin_df['end_date'])
fin_df['available_date'] = fin_df['report_period'].apply(
lambda x: pd.Timestamp(f"{x.year+1}-04-30") if x.month == 12 # 年报
else pd.Timestamp(f"{x.year}-08-31") if x.month == 6 # 中报
else pd.Timestamp(f"{x.year}-10-31") if x.month == 9 # 三季报
else pd.Timestamp(f"{x.year}-04-30") # 一季报
)
# 2. 合并交易数据与财务数据（仅保留财务数据可用后的交易记录）
merge_df = pd.merge(trade_df, fin_df[['ts_code', 'pe', 'roe', 'available_date']], on='ts_code', how='left')
merge_df = merge_df[merge_df['trade_date'] >= merge_df['available_date']]
return merge_df
3.2.2 ROE 因子（质量类）
def build_roe_factor(fin_df, trade_df):
"""构建ROE因子（TTM，滚动四个季度）"""
# 1. 计算ROE-TTM（避免单季度波动）
fin_df['roe_ttm'] = fin_df.groupby('ts_code')['roe'].rolling(4).sum().reset_index(0, drop=True)
# 2. 对齐交易日期
merge_df = map_finance_to_trade(fin_df, trade_df)
# 3. 填充缺失值（用最新可用的ROE-TTM）
merge_df['roe_ttm'] = merge_df.groupby('ts_code')['roe_ttm'].fillna(method='ffill')
return merge_df
3.3 风险因子（市值 / 波动率）
def build_risk_factors(df):
"""构建市值+波动率风险因子"""
# 1. 市值因子（总市值取对数，消除量纲）
df['size'] = np.log(df['total_mv'])
# 2. 波动率因子（20日收益率标准差）
df['volatility_20d'] = df.groupby('ts_code')['return'].rolling(20).std().reset_index(0, drop=True)
return df
第四章：因子预处理（核心步骤）
4.1 预处理全流程
原始因子值
缺失值处理
极端值剔除
标准化
中性化
正交化
4.2 缺失值处理（分场景）
场景 处理方法 代码示例
单只股票短期缺失 前向填充（ffill）
df['factor'] = df.groupby('ts_code')['factor'].ffill()
行业 / 整体大量缺失 行业均值填充
df['factor'] = df.groupby('industry')['factor'].fillna(df.groupby('industry')['factor'].mean())
无有效数据（退市） 直接剔除
df = df[df['factor'].notna()]
4.3 标准化（消除量纲）
4.3.1 横截面标准化（每日 / 每月）
def cross_section_standardize(df, factor_col, date_col='trade_date'):
"""
横截面标准化：同一时间点，所有股票的因子值标准化
（避免时间序列标准化的“幸存者偏差”）
"""
def standardize_group(x):
scaler = StandardScaler()
x[factor_col + '_std'] = scaler.fit_transform(x[[factor_col]])
return x
# 按日期分组标准化
df = df.groupby(date_col).apply(standardize_group).reset_index(drop=True)
return df
4.3.2 行业内标准化（增强因子纯度）
def industry_standardize(df, factor_col, industry_col, date_col='trade_date'):
"""行业内横截面标准化：先按行业分组，再标准化"""
df[factor_col + '_ind_std'] = df.groupby([date_col, industry_col])[factor_col].apply(
lambda x: (x - x.mean()) / x.std()
)
return df
4.4 中性化（剔除干扰因素）
4.4.1 行业中性化（核心）
def industry_neutralize_v2(df, factor_col, industry_col):
"""
进阶行业中性化：回归法（更精准）
因子值 = 行业虚拟变量 * 系数 + 残差（中性化因子）
"""
# 1. 生成行业虚拟变量
industry_dummies = pd.get_dummies(df[industry_col], prefix='ind')
df_with_dummies = pd.concat([df, industry_dummies], axis=1)
# 2. 回归：因子值 ~ 行业虚拟变量
X = df_with_dummies[industry_dummies.columns]
y = df_with_dummies[factor_col]
model = LinearRegression()
model.fit(X, y)
# 3. 残差即为中性化因子
df[factor_col + '_neutral'] = y - model.predict(X)
return df
4.4.2 市值中性化（剔除市值干扰）
def size_neutralize(df, factor_col, size_col='size'):
"""市值中性化：因子值对市值回归，残差为中性化因子"""
X = df[[size_col]]
y = df[factor_col]
model = LinearRegression()
model.fit(X, y)
df[factor_col + '_size_neutral'] = y - model.predict(X)
return df
4.5 正交化（消除因子间共线性）
def factor_orthogonalize(df, target_factor, other_factors):
"""
因子正交化：目标因子剔除其他因子的影响
例：动量因子正交化（剔除市值+行业影响）
"""
X = df[other_factors]
y = df[target_factor]
model = LinearRegression()
model.fit(X, y)
df[target_factor + '_ortho'] = y - model.predict(X)
return df
第五章：单因子检验（有效性验证）
5.1 核心检验指标
指标 计算公式 判定标准
IC 值 Spearman (因子值，下期收益) 绝对值 > 0.05，且显著（t 检验 p<0.05）
IR 值 IC 均值 / IC 标准差 >0.5 为有效，>1 为优秀
分层收益 最高层 - 最低层累计收益 持续为正，且单调性明显
换手率 组合月度调仓比例 <30%（过高则交易成本侵蚀收益）
最大回撤 累计收益的最大下跌幅度 <10%（单因子）
5.2 IC 分析全流程（书籍核心代码）
def ic_analysis_full(df, factor_col, return_col='next_return', date_col='trade_date'):
"""
完整IC分析：月度IC、IC均值、IR、t检验、胜率
:param return_col: 下期收益率（如未来5/10/20日收益）
"""
# 1. 确保收益是“下期”（关键：避免未来函数）
df[return_col] = df.groupby('ts_code')['return'].shift(-5) # 未来5日收益
# 2. 按月份分组计算IC
df['month'] = df[date_col].dt.to_period('M')
ic_results = []
for month, group in df.groupby('month'):
# 剔除缺失值
group_clean = group.dropna(subset=[factor_col, return_col])
if len(group_clean) < 50: # 样本量不足跳过
continue
# 计算Spearman IC
ic = group_clean[[factor_col, return_col]].corr(method='spearman').iloc[0,1]
ic_results.append({
'month': month,
'ic': ic,
'sample_size': len(group_clean)
})
ic_df = pd.DataFrame(ic_results)
# 3. 计算核心统计量
ic_mean = ic_df['ic'].mean()
ic_std = ic_df['ic'].std()
ic_ir = ic_mean / ic_std if ic_std != 0 else 0
# t检验：IC均值是否显著不为0
t_stat = ic_mean / (ic_std / np.sqrt(len(ic_df)))
p_value = ttest_1samp(ic_df['ic'], 0)[1]
# 胜率：IC>0的月份占比
win_rate = len(ic_df[ic_df['ic'] > 0]) / len(ic_df)
# 4. 输出结果
result = {
'ic_mean': round(ic_mean, 4),
'ic_std': round(ic_std, 4),
'ic_ir': round(ic_ir, 4),
't_stat': round(t_stat, 4),
'p_value': round(p_value, 4),
'win_rate': round(win_rate, 4),
'ic_series': ic_df
}
return result
# 调用示例
ic_result = ic_analysis_full(df, 'momentum_20d_std_neutral', 'next_return')
print(f"IC均值：{ic_result['ic_mean']}，IR：{ic_result['ic_ir']}，胜率：{ic_result['win_rate']}")
5.3 分层回测（可视化 + 统计）
def quantile_backtest_full(df, factor_col, return_col='next_return', quantile=5, date_col='trade_date'):
"""
5分层回测：
- 按因子值从小到大分为Q1（最低）到Q5（最高）
- 计算各层累计收益、夏普比率、最大回撤
"""
# 1. 每日分层
df['quantile'] = df.groupby(date_col)[factor_col].rank(pct=True).apply(
lambda x: int(x * quantile) + 1
)
# 2. 计算各层每日收益（等权）
daily_return = df.groupby([date_col, 'quantile'])[return_col].mean().reset_index()
daily_return_pivot = daily_return.pivot(index=date_col, columns='quantile', values=return_col)
# 3. 计算累计收益
cumulative_return = (1 + daily_return_pivot).cumprod()
# 4. 计算各层绩效指标
performance = {}
for q in range(1, quantile+1):
ret_series = daily_return_pivot[q]
# 夏普比率（无风险利率取3%年化）
sharpe = np.sqrt(252) * ret_series.mean() / ret_series.std()
# 最大回撤
cum_ret = cumulative_return[q]
drawdown = (cum_ret / cum_ret.cummax() - 1).min()
# 年化收益
annual_return = (cum_ret.iloc[-1] ** (252/len(cum_ret))) - 1
performance[f'Q{q}'] = {
'annual_return': round(annual_return, 4),
'sharpe': round(sharpe, 4),
'max_drawdown': round(drawdown, 4),
'total_return': round(cum_ret.iloc[-1] - 1, 4)
}
# 5. 多空收益（Q5 - Q1）
daily_return_pivot['long_short'] = daily_return_pivot[5] - daily_return_pivot[1]
cumulative_return['long_short'] = (1 + daily_return_pivot['long_short']).cumprod()
return cumulative_return, performance, daily_return_pivot
# 可视化分层收益
def plot_quantile_return(cumulative_return):
plt.figure(figsize=(12, 6))
for col in cumulative_return.columns:
plt.plot(cumulative_return.index, cumulative_return[col], label=f'{col}')
plt.title('因子分层累计收益')
plt.xlabel('日期')
plt.ylabel('累计收益（初始1）')
plt.legend()
plt.grid(True)
plt.show()
5.4 因子衰减分析（时效性）
def factor_decay_analysis(df, factor_col, max_lag=20):
"""
因子衰减分析：计算不同滞后期的IC值
例：lag=0（当期因子）、lag=1（滞后1期）...lag=20
"""
decay_results = []
for lag in range(max_lag+1):
# 因子滞后lag期
df[f'factor_lag_{lag}'] = df.groupby('ts_code')[factor_col].shift(lag)
# 计算IC
ic = ic_analysis_full(df, f'factor_lag_{lag}')['ic_mean']
decay_results.append({
'lag': lag,
'ic_mean': ic
})
decay_df = pd.DataFrame(decay_results)
# 可视化衰减曲线
plt.figure(figsize=(10, 5))
plt.plot(decay_df['lag'], decay_df['ic_mean'], marker='o')
plt.title('因子衰减曲线')
plt.xlabel('滞后期（交易日）')
plt.ylabel('IC均值')
plt.grid(True)
plt.show()
return decay_df
第六章：多因子组合构建
6.1 因子加权方法对比
加权方法 计算逻辑 适用场景 代码示例
等权法 各因子得分等权相加 因子有效性相近 score = (f1 + f2 + f3)/3
IC 加权法 按 IC 均值绝对值加权 因子有效性差异大 weights = IC_abs / sum(IC_abs)
回归加权法 收益～因子值 回归系数加权 追求收益最大化 见下方代码
机器学习加权 XGBoost/LightGBM 拟合因子权重 因子间非线性关系 进阶拓展
6.1.1 回归加权法（书籍核心）
def regression_weight(df, factor_cols, return_col='next_return'):
"""
回归加权：通过截面回归计算因子权重
每日回归：下期收益 = w1*f1 + w2*f2 + ... + wn*fn + ε
"""
# 1. 按日期分组回归
weight_list = []
for date, group in df.groupby('trade_date'):
group_clean = group.dropna(subset=factor_cols + [return_col])
if len(group_clean) < 100:
continue
X = group_clean[factor_cols]
y = group_clean[return_col]
model = LinearRegression()
model.fit(X, y)
# 存储每日权重
weights = dict(zip(factor_cols, model.coef_))
weights['trade_date'] = date
weight_list.append(weights)
weight_df = pd.DataFrame(weight_list)
# 2. 计算平均权重（用于组合构建）
avg_weights = weight_df[factor_cols].mean().to_dict()
# 3. 计算多因子得分
df['multi_factor_score'] = 0
for factor, weight in avg_weights.items():
df['multi_factor_score'] += df[factor] * weight
return df, avg_weights
6.2 多因子组合构建流程
def build_multi_factor_portfolio(df, score_col='multi_factor_score', top_n=100):
"""
构建多因子组合：
1. 每日按多因子得分排序，选前N只股票
2. 等权配置，月度调仓
"""
# 1. 按月调仓（每月最后一个交易日调仓）
df['month'] = df['trade_date'].dt.to_period('M')
df['is_month_end'] = df.groupby('month')['trade_date'].rank(ascending=False) == 1
# 2. 筛选调仓日，选前top_n股票
rebalance_dates = df[df['is_month_end']]['trade_date'].unique()
portfolio = []
for date in rebalance_dates:
# 当日股票得分
score_df = df[df['trade_date'] == date].dropna(subset=[score_col])
# 按得分降序选前top_n
top_stocks = score_df.nlargest(top_n, score_col)['ts_code'].tolist()
portfolio.append({
'rebalance_date': date,
'stocks': top_stocks
})
portfolio_df = pd.DataFrame(portfolio)
# 3. 计算组合收益（持有至下月调仓）
def calculate_port_return(row):
# 调仓日至下月调仓日的收益
start_date = row['rebalance_date']
end_date = portfolio_df[portfolio_df['rebalance_date'] > start_date]['rebalance_date'].min()
if pd.isna(end_date):
end_date = df['trade_date'].max()
# 筛选持仓股票在该时间段的收益
hold_df = df[(df['ts_code'].isin(row['stocks'])) &
(df['trade_date'] >= start_date) &
(df['trade_date'] <= end_date)]
# 等权日收益
daily_ret = hold_df.groupby('trade_date')['return'].mean()
# 累计收益
cum_ret = (1 + daily_ret).prod() - 1
return pd.Series({
'start_date': start_date,
'end_date': end_date,
'cum_return': cum_ret,
'daily_ret_mean': daily_ret.mean()
})
portfolio_perf = portfolio_df.apply(calculate_port_return, axis=1)
return portfolio_df, portfolio_perf
6.3 风险控制（书籍进阶内容）
6.3.1 行业暴露控制
def control_industry_exposure(df, score_col, industry_col, max_exposure=0.2):
"""
行业暴露控制：单行业权重不超过max_exposure
"""
# 1. 初始得分排序
df_sorted = df.sort_values(score_col, ascending=False)
# 2. 按行业分组，计算权重上限
industry_cap = df_sorted.groupby(industry_col).size() * max_exposure
# 3. 按行业限额筛选股票
selected_stocks = []
for industry in df_sorted[industry_col].unique():
industry_stocks = df_sorted[df_sorted[industry_col] == industry]
n_select = int(industry_cap[industry])
selected = industry_stocks.head(n_select)
selected_stocks.append(selected)
selected_df = pd.concat(selected_stocks)
return selected_df
6.3.2 市值暴露控制
def control_size_exposure(df, score_col, size_col='size', target_size=0):
"""
市值暴露控制：组合市值均值接近目标值（如0，即市场中性）
"""
# 1. 初始筛选
df_sorted = df.sort_values(score_col, ascending=False).head(200)
# 2. 计算当前组合市值均值
current_size = df_sorted[size_col].mean()
# 3. 调整筛选，使市值均值接近target_size
if current_size > target_size:
# 多选小市值股票
df_sorted['size_adjusted_score'] = df_sorted[score_col] - (df_sorted[size_col] - target_size)
else:
# 多选大市值股票
df_sorted['size_adjusted_score'] = df_sorted[score_col] + (target_size - df_sorted[size_col])
# 重新排序选前100
selected_df = df_sorted.sort_values('size_adjusted_score', ascending=False).head(100)
return selected_df
第七章：回测与绩效评估
7.1 回测核心注意事项
避免未来函数：因子值必须在调仓日前可获取；
交易成本：佣金（0.03%）+ 滑点（0.05%）+ 印花税（0.1%，卖出）；
流动性约束：剔除换手率 < 1%、流通市值 < 50 亿的股票；
调仓频率：月度 / 双周调仓（高频调仓成本高）
7.2 绩效评估指标体系
def evaluate_portfolio(return_series):
"""
组合绩效评估：年化收益、夏普、最大回撤、胜率等
:param return_series: 日收益序列
"""
# 基础指标
annual_return = (1 + return_series).prod() ** (252 / len(return_series)) - 1
sharpe = np.sqrt(252) * return_series.mean() / return_series.std()
# 最大回撤
cum_return = (1 + return_series).cumprod()
drawdown = (cum_return / cum_return.cummax() - 1).min()
# 胜率
win_rate = len(return_series[return_series > 0]) / len(return_series)
# 盈亏比
profit = return_series[return_series > 0].mean()
loss = return_series[return_series < 0].mean()
profit_loss_ratio = abs(profit / loss) if loss != 0 else np.inf
# 信息比率（相对沪深300）
hs300_return = get_hs300_return(return_series.index) # 需提前获取沪深300收益
excess_return = return_series - hs300_return
ir = excess_return.mean() / excess_return.std() * np.sqrt(252)
return {
'annual_return': round(annual_return, 4),
'sharpe_ratio': round(sharpe, 4),
'max_drawdown': round(drawdown, 4),
'win_rate': round(win_rate, 4),
'profit_loss_ratio': round(profit_loss_ratio, 4),
'information_ratio': round(ir, 4)
}
第八章：实战优化与常见坑点
8.1 因子失效原因与应对
失效原因 典型表现 应对方法
市场风格切换 IC 值突然反转 动态因子权重（如月度更新 IC 权重）
因子拥挤 分层收益收窄 加入换手率约束，避开热门股票
数据挖掘陷阱 样本内有效，样本外失效 样本外验证（分训练集 / 测试集）
交易成本侵蚀 回测收益高，实盘收益低 提高调仓周期，控制换手率
8.2 实战优化技巧
因子融合：将量价因子（高频）与财务因子（低频）结合，兼顾短期与长期；
动态调仓：根据市场波动率调整仓位（高波动时降仓）；
样本外验证：按时间划分训练集（2018-2022）、测试集（2023-2024），避免过拟合；
实盘适配：回测时加入真实交易成本，模拟滑点（如买入价 + 0.02%，卖出价 - 0.02%）
8.3 常见代码坑点
坑点 表现 解决方案
未来函数 回测收益极高，实盘亏损 所有因子滞后至少 1 期，收益用下期
数据对齐错误 财务因子用了未披露的数据 严格映射 “报告期 - 可用日期”
幸存者偏差 只包含当前上市的股票 用复权数据，保留退市股票记录
行业分类过时 因子中性化失效 每月更新申万行业分类数据
注意点：
因子的 “有效性” 是动态的，需持续监控 IC、分层收益；
预处理（标准化 / 中性化）是因子有效与否的关键；
回测必须贴近实盘，考虑交易成本、流动性、调仓频率；
多因子模型的核心是 “因子互补”，而非简单堆砌。
建议学习路径：
先复现单因子检验流程，
再尝试 2-3 个低相关因子构建组合，
最后加入风险控制与实盘适配，
逐步从 “回测有效” 走向 “实盘盈利”
一 其他示例代码
数据获取与清洗
import tushare as ts
import pandas as pd
import numpy as np
# 1. 初始化Tushare（需申请token）
ts.set_token("你的Tushare Token")
pro = ts.pro_api()
# 2. 获取股票日度行情数据
def get_stock_data(ts_code, start_date, end_date):
df = pro.daily(ts_code=ts_code, start_date=start_date, end_date=end_date)
# 数据清洗：处理缺失值、排序
df = df.dropna(subset=['close', 'vol']).sort_values('trade_date').reset_index(drop=True)
# 转换日期格式
df['trade_date'] = pd.to_datetime(df['trade_date'])
# 计算日收益率
df['return'] = df['close'].pct_change()
return df
# 3. 获取沪深300成分股（因子分析常用标的）
def get_hs300_stocks(date):
hs300 = pro.index_weight(index_code='000300.SH', start_date=date, end_date=date)
return hs300['con_code'].tolist()
# 示例调用
hs300_codes = get_hs300_stocks('20240101')
stock_data = get_stock_data('000001.SZ', '20230101', '20240101')
二 量价因子：动量因子（20 日累计收益率）
def build_momentum_factor(df, window=20):
"""构建20日动量因子"""
df['momentum_20d'] = df['close'].pct_change(window).shift(1) # 滞后1期避免未来函数
return df
# （2）财务因子：市盈率（PE）
def build_pe_factor(stock_code, date):
"""获取单只股票的PE因子"""
fina_indicator = pro.fina_indicator(ts_code=stock_code, start_date=date, end_date=date)
pe = fina_indicator['pe'].iloc[0] if not fina_indicator.empty else np.nan
return pe
# （3）风险因子：市值因子
def build_size_factor(stock_code, date):
"""构建市值因子（总市值取对数）"""
daily_basic = pro.daily_basic(ts_code=stock_code, trade_date=date)
size = np.log(daily_basic['total_mv'].iloc[0]) if not daily_basic.empty else np.nan
return size
三 因子预处理（标准化 + 中性化）
因子的有效性依赖于合理的预处理，书籍重点讲解了 Z-score 标准化和行业中性化：
from sklearn.preprocessing import StandardScaler
def standardize_factor(factor_df):
"""Z-score标准化：消除量纲影响"""
scaler = StandardScaler()
factor_df['factor_standard'] = scaler.fit_transform(factor_df[['factor_value']])
return factor_df
def industry_neutralize(factor_df, industry_col):
"""行业中性化：剔除行业对因子的影响"""
# 按行业分组，计算组内残差（因子值 - 行业均值）
factor_df['industry_mean'] = factor_df.groupby(industry_col)['factor_value'].transform('mean')
factor_df['factor_neutral'] = factor_df['factor_value'] - factor_df['industry_mean']
return factor_df
四. 单因子检验（核心：IC 值 + 分层回测）
单因子检验是验证因子有效性的关键，书籍详细实现了信息系数（IC）和分层回测：
（1）信息系数（IC）：因子与下期收益的相关性
def calculate_ic(factor_df, return_col, factor_col):
"""计算IC值（Spearman相关系数）"""
# 确保因子与收益对应下期（避免未来函数）
ic = factor_df[[factor_col, return_col]].corr(method='spearman').iloc[0,1]
return ic
# 批量计算月度IC
def monthly_ic_analysis(factor_df):
factor_df['month'] = factor_df['trade_date'].dt.to_period('M')
ic_series = factor_df.groupby('month').apply(
lambda x: calculate_ic(x, 'next_return', 'factor_neutral')
)
# 输出IC均值、IR（信息比率=IC均值/IC标准差）
ic_mean = ic_series.mean()
ic_ir = ic_mean / ic_series.std()
return {'ic_mean': ic_mean, 'ic_ir': ic_ir, 'ic_series': ic_series}
（2）分层回测：验证因子的收益区分能力
def quantile_backtest(factor_df, quantile=5):
"""分层回测：将股票按因子值分为N层，计算各层收益"""
# 按因子值排序，分层
factor_df['quantile'] = factor_df.groupby('trade_date')['factor_neutral'].rank(pct=True).apply(
lambda x: int(x * quantile) + 1
)
# 计算各层日均收益
quantile_return = factor_df.groupby(['trade_date', 'quantile'])['next_return'].mean().reset_index()
# 透视表：日期为行，分层为列
quantile_pivot = quantile_return.pivot(index='trade_date', columns='quantile', values='next_return')
# 计算各层累计收益
cumulative_return = (1 + quantile_pivot).cumprod()
return cumulative_return, quantile_pivot
五. 多因子组合构建
在单因子有效基础上，书籍讲解了多因子加权（等权、IC 加权、回归加权）和风险约束：
def multi_factor_weight(factor_dict, weight_method='ic_weight'):
"""多因子加权：IC加权法"""
# factor_dict: {因子名: 因子值, ...}
factor_df = pd.DataFrame(factor_dict)
# 计算各因子的IC均值（提前通过单因子检验得到）
ic_means = {
'momentum_20d': 0.08,
'pe': -0.06,
'size': -0.05
} # 示例IC均值，实际需计算
if weight_method == 'ic_weight':
# 按IC绝对值加权（归一化）
ic_abs = np.abs(list(ic_means.values()))
weights = ic_abs / sum(ic_abs)
elif weight_method == 'equal':
weights = [1/len(factor_df.columns)] * len(factor_df.columns)
# 计算多因子综合得分
factor_df['multi_factor_score'] = np.dot(factor_df.values, weights)
return factor_df
上一篇
均值回归策略实例
python @classmethod 和 @staticmethod 区别
股票多因子模型实战