6.1 因子选择与重要性分析¶

这一节讲什么？

特征太多、信噪比低、多重共线性——这是量化机器学习的三大敌人。这一节教你用统计方法（互信息、LASSO）和领域知识筛选真正有预测力的因子，避免把噪音喂给模型。

学习目标¶

• 理解量化因子的概念与分类
• 用统计方法筛选有预测价值的因子
• 计算因子的 IC 值（信息系数）评估预测能力
• 识别因子间的多重共线性

1. 这节究竟解决什么实际问题？¶

6.1 因子选择与重要性分析 的具体任务是：用重要性、相关性和稳定性筛选因子，避免把噪声当信号。 本节将明确输入数据、计算过程、输出及其在后续研究中的用途。

2. 前置知识¶

掌握特征、标签、训练集/测试集和基本回测概念。 对公式的理解应包括变量定义、单位、时间索引和适用条件。

3. 数据说明¶

特征默认来自固定样本或模拟样本；每个标签必须标明预测期限及其可交易时间。 运行前先检查 shape、日期范围、列名与缺失值。数据看起来能画图，不代表它适合回答研究问题。

4. 学习目标与完成标准¶

完成本节后，你应能用自己的话解释 6.1 因子选择与重要性分析 的输入、计算步骤、输出和局限；尤其能说明：用重要性、相关性和稳定性筛选因子，避免把噪声当信号。 你还应能独立改动一个参数并解释结果为什么变化，指出至少一个会让结论失真的假设。

5. 核心概念与公式¶

机器学习模型从训练样本估计输入与目标之间的关系。训练集拟合程度不能代表样本外预测能力。

训练时只能看到训练区间；特征时间不得晚于决策时间，标签与测试区间必须隔离。 看到公式时依次确认：每个符号的单位是什么？它属于哪个时点？分母会不会为零？它是描述性统计量、预测量，还是交易规则？

6. 从零实现¶

原始代码按照“准备输入 → 计算中间量 → 得到结果 → 解释输出”的顺序组织。每个单元应检查变量类型、索引、维度和代表性数值。中间变量用于核对公式实现和定位计算错误。

7. 成熟库实现¶

sklearn/PyTorch 加速建模；管道、切分和经济解释仍由研究者负责。 使用库函数前先阅读参数含义、默认值和返回对象；库输出一个数字，并不说明这个数字适合你的交易假设。

8. 两种实现的差异¶

从零实现便于核对公式、调试和识别隐含假设；成熟库适合在计算逻辑已验证后复用。两种实现依赖相同的数据口径与时点假设，使用库函数不会改变输入数据的质量。

9. 预期结果与检查范围¶

你应看到维度正确的 Series/DataFrame、日期单调递增的索引，以及能支持“用重要性、相关性和稳定性筛选因子，避免把噪声当信号。”这一任务的表格或图形。金融日收益率均值通常接近 0；若出现极大的连续收益、空表、全 NaN 或索引错位，应先排查数据和对齐逻辑，而不是立即相信结果。

10. 常见错误与诊断¶

随机打乱时序、Scaler 在全样本 fit、目标泄漏、调参反复查看测试集、把准确率等同收益。

诊断顺序：先打印类型与 shape，再看 head()/tail()，接着检查日期和 NaN，最后才比较数值。这样能避免在错误的对象上反复改参数。

11. 研究偏差检查¶

时间泄漏、p-hacking、多重检验和样本选择会制造虚假的 Alpha；必须报告朴素基线。 结果至少应与一个简单基线比较，并明确哪些部分只是教学演示。

12. 三层练习与答案¶

• 基础练习：验证一个中间变量、补全一段小计算，并用一句话解释输出。
• 研究练习：改变窗口、样本区间、约束或标的，记录哪些结论仍成立。
• 开放项目：把本节方法嵌入模块项目，报告基线、失败结果和限制。

题目在 ../exercises/06_ml_trading/01_feature_selection_exercises.ipynb；完整推导、可运行参考实现与结果检查在 ../solutions/06_ml_trading/01_feature_selection_solutions.ipynb。请先独立作答，再打开答案。

13. 延伸阅读¶

延伸材料：scikit-learn 模型选择文档。阅读记录应包括研究假设、数据频率、样本范围和结论适用条件。

14. 下一节¶

下一节是 02_ml_prediction。本节的概念会成为理解它的输入，而不是孤立的名词。

In [1]:

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns
import yfinance as yf
from scipy import stats

plt.rcParams['figure.figsize'] = (12, 5)
print('库加载完成 ')

import numpy as np import pandas as pd import matplotlib.pyplot as plt import seaborn as sns import yfinance as yf from scipy import stats plt.rcParams['figure.figsize'] = (12, 5) print('库加载完成 ')

库加载完成 

In [ ]:

# 默认使用固定样本，保证课程与 CI 不依赖实时接口。
USE_OFFLINE_DATA = True
if USE_OFFLINE_DATA:
    from pathlib import Path
    import sys
    docs_root = next(parent for parent in [Path.cwd(), *Path.cwd().parents]
                     if (parent / 'utils' / 'offline_data.py').exists())
    sys.path.insert(0, str(docs_root))
    from utils.offline_data import configure_yfinance_offline
    configure_yfinance_offline(yf)
else:
    print('在线下载为扩展步骤；请核对数据字段、时区和下载日期。')

# 默认使用固定样本，保证课程与 CI 不依赖实时接口。 USE_OFFLINE_DATA = True if USE_OFFLINE_DATA: from pathlib import Path import sys docs_root = next(parent for parent in [Path.cwd(), *Path.cwd().parents] if (parent / 'utils' / 'offline_data.py').exists()) sys.path.insert(0, str(docs_root)) from utils.offline_data import configure_yfinance_offline configure_yfinance_offline(yf) else: print('在线下载为扩展步骤；请核对数据字段、时区和下载日期。')

In [2]:

import matplotlib.pyplot as plt

# 1. 设置系统自带的中文字体（这里使用黑体 SimHei）
plt.rcParams['font.sans-serif'] = ['SimHei']  # 如果你想用微软雅黑，可以改成 ['Microsoft YaHei']

# 2. 解决更换字体后，负号（-）显示为方块的问题
plt.rcParams['axes.unicode_minus'] = False

import matplotlib.pyplot as plt # 1. 设置系统自带的中文字体（这里使用黑体 SimHei） plt.rcParams['font.sans-serif'] = ['SimHei'] # 如果你想用微软雅黑，可以改成 ['Microsoft YaHei'] # 2. 解决更换字体后，负号（-）显示为方块的问题 plt.rcParams['axes.unicode_minus'] = False

1. 量化因子分类¶

类型	示例因子	说明
动量因子	20日/60日收益率	强者恒强
反转因子	5日/短期收益率	短期超跌反弹
波动率因子	历史波动率、ATR	低波动溢价
技术因子	RSI、MACD、布林带	技术分析信号
成交量因子	量比、换手率	资金流向
基本面因子	PE、PB、ROE	价值/质量

本节重点介绍技术因子的构建与筛选。

2. 构建因子库¶

In [3]:

# 下载多只股票
tickers = ['AAPL', 'MSFT', 'GOOGL', 'AMZN', 'TSLA',
           'JPM', 'BAC', 'XOM', 'JNJ', 'PG']
prices = yf.download(tickers, start='2018-01-01', end='2024-01-01',
                     progress=False)['Close'].dropna()
volume = yf.download(tickers, start='2018-01-01', end='2024-01-01',
                     progress=False)['Volume'].dropna()

# 为每只股票构建因子面板
all_factors = []

for ticker in tickers:
    p = prices[ticker].dropna()
    v = volume[ticker].dropna()
    df = pd.DataFrame(index=p.index)

    # 动量 / 反转
    df['mom_5d']  = p.pct_change(5)
    df['mom_20d'] = p.pct_change(20)
    df['mom_60d'] = p.pct_change(60)

    # 波动率
    df['vol_20d'] = p.pct_change().rolling(20).std() * np.sqrt(252)

    # RSI
    delta = p.diff()
    gain = delta.clip(lower=0).ewm(com=13).mean()
    loss = (-delta.clip(upper=0)).ewm(com=13).mean()
    df['rsi_14'] = 100 - 100 / (1 + gain / loss)

    # 布林带位置
    ma20 = p.rolling(20).mean()
    std20 = p.rolling(20).std()
    df['bb_pct'] = (p - (ma20 - 2 * std20)) / (4 * std20)  # 在带内的位置 [0, 1]

    # 成交量比
    df['vol_ratio'] = v / v.rolling(20).mean()
    df['price_ma_ratio'] = p / p.rolling(20).mean() - 1

    # 目标变量（下期20日收益）
    df['fwd_ret_20d'] = p.pct_change(20).shift(-20)

    df['ticker'] = ticker
    all_factors.append(df)

factors = pd.concat(all_factors).dropna()
feature_cols = ['mom_5d', 'mom_20d', 'mom_60d', 'vol_20d', 'rsi_14',
                'bb_pct', 'vol_ratio', 'price_ma_ratio']
print(f'因子面板形状: {factors.shape}')
print(f'样本数: {len(factors)}')

# 下载多只股票 tickers = ['AAPL', 'MSFT', 'GOOGL', 'AMZN', 'TSLA', 'JPM', 'BAC', 'XOM', 'JNJ', 'PG'] prices = yf.download(tickers, start='2018-01-01', end='2024-01-01', progress=False)['Close'].dropna() volume = yf.download(tickers, start='2018-01-01', end='2024-01-01', progress=False)['Volume'].dropna() # 为每只股票构建因子面板 all_factors = [] for ticker in tickers: p = prices[ticker].dropna() v = volume[ticker].dropna() df = pd.DataFrame(index=p.index) # 动量 / 反转 df['mom_5d'] = p.pct_change(5) df['mom_20d'] = p.pct_change(20) df['mom_60d'] = p.pct_change(60) # 波动率 df['vol_20d'] = p.pct_change().rolling(20).std() * np.sqrt(252) # RSI delta = p.diff() gain = delta.clip(lower=0).ewm(com=13).mean() loss = (-delta.clip(upper=0)).ewm(com=13).mean() df['rsi_14'] = 100 - 100 / (1 + gain / loss) # 布林带位置 ma20 = p.rolling(20).mean() std20 = p.rolling(20).std() df['bb_pct'] = (p - (ma20 - 2 * std20)) / (4 * std20) # 在带内的位置 [0, 1] # 成交量比 df['vol_ratio'] = v / v.rolling(20).mean() df['price_ma_ratio'] = p / p.rolling(20).mean() - 1 # 目标变量（下期20日收益） df['fwd_ret_20d'] = p.pct_change(20).shift(-20) df['ticker'] = ticker all_factors.append(df) factors = pd.concat(all_factors).dropna() feature_cols = ['mom_5d', 'mom_20d', 'mom_60d', 'vol_20d', 'rsi_14', 'bb_pct', 'vol_ratio', 'price_ma_ratio'] print(f'因子面板形状: {factors.shape}') print(f'样本数: {len(factors)}')

因子面板形状: (14290, 10)
样本数: 14290

3. 信息系数 IC（Information Coefficient）¶

$$IC = \text{Spearman}\left(\text{因子值}_t,\ \text{下期收益}_{t+n}\right)$$

• $|IC| > 0.05$：因子有实用价值
• 月度 IC 序列：ICIR = mean(IC) / std(IC) > 0.5 为稳定因子

In [4]:

def compute_ic(factors_df, feature_cols, target='fwd_ret_20d', freq='ME'):
    """按时间截面计算 Spearman IC"""
    ic_records = []
    for date, group in factors_df.groupby(pd.Grouper(freq=freq, level=0)):
        if len(group) < 5:
            continue
        row = {'date': date}
        for col in feature_cols:
            if group[col].std() == 0:
                row[col] = np.nan
                continue
            ic_val, _ = stats.spearmanr(group[col].rank(), group[target].rank())
            row[col] = ic_val
        ic_records.append(row)
    return pd.DataFrame(ic_records).set_index('date')

ic_df = compute_ic(factors, feature_cols)

ic_summary = pd.DataFrame({
    'Mean IC': ic_df.mean(),
    'Std IC': ic_df.std(),
    'ICIR': ic_df.mean() / ic_df.std(),
    '|IC| > 0.05 比例': (ic_df.abs() > 0.05).mean()
}).round(3)

# 按 |Mean IC| 排序
ic_summary = ic_summary.reindex(ic_summary['Mean IC'].abs().sort_values(ascending=False).index)
print('因子 IC 汇总（按 |Mean IC| 排序）：')
print(ic_summary.to_string())

def compute_ic(factors_df, feature_cols, target='fwd_ret_20d', freq='ME'): """按时间截面计算 Spearman IC""" ic_records = [] for date, group in factors_df.groupby(pd.Grouper(freq=freq, level=0)): if len(group) < 5: continue row = {'date': date} for col in feature_cols: if group[col].std() == 0: row[col] = np.nan continue ic_val, _ = stats.spearmanr(group[col].rank(), group[target].rank()) row[col] = ic_val ic_records.append(row) return pd.DataFrame(ic_records).set_index('date') ic_df = compute_ic(factors, feature_cols) ic_summary = pd.DataFrame({ 'Mean IC': ic_df.mean(), 'Std IC': ic_df.std(), 'ICIR': ic_df.mean() / ic_df.std(), '|IC| > 0.05 比例': (ic_df.abs() > 0.05).mean() }).round(3) # 按 |Mean IC| 排序 ic_summary = ic_summary.reindex(ic_summary['Mean IC'].abs().sort_values(ascending=False).index) print('因子 IC 汇总（按 |Mean IC| 排序）：') print(ic_summary.to_string())

因子 IC 汇总（按 |Mean IC| 排序）：
                Mean IC  Std IC   ICIR  |IC| > 0.05 比例
rsi_14           -0.119   0.245 -0.485           0.841
price_ma_ratio   -0.118   0.248 -0.475           0.826
bb_pct           -0.115   0.221 -0.523           0.812
vol_20d           0.098   0.298  0.328           0.913
mom_20d          -0.093   0.259 -0.360           0.884
mom_5d           -0.088   0.205 -0.429           0.768
mom_60d          -0.040   0.301 -0.131           0.884
vol_ratio         0.006   0.170  0.035           0.754

In [5]:

# IC 随时间变化
fig, ax = plt.subplots(figsize=(13, 5))
top_factors = ic_summary.head(4).index.tolist()
for col in top_factors:
    ic_cumsum = ic_df[col].fillna(0).cumsum()
    ax.plot(ic_df.index, ic_cumsum.values, label=col, linewidth=1.5)
ax.axhline(0, color='black', linewidth=0.8)
ax.set_title('Top 因子 IC 累积和（持续上升=稳定正向因子）', fontsize=13)
ax.legend()
ax.grid(alpha=0.3)
plt.tight_layout()
plt.show()

# IC 随时间变化 fig, ax = plt.subplots(figsize=(13, 5)) top_factors = ic_summary.head(4).index.tolist() for col in top_factors: ic_cumsum = ic_df[col].fillna(0).cumsum() ax.plot(ic_df.index, ic_cumsum.values, label=col, linewidth=1.5) ax.axhline(0, color='black', linewidth=0.8) ax.set_title('Top 因子 IC 累积和（持续上升=稳定正向因子）', fontsize=13) ax.legend() ax.grid(alpha=0.3) plt.tight_layout() plt.show()

4. 因子相关性与多重共线性¶

In [6]:

factor_corr = factors[feature_cols].corr()

fig, ax = plt.subplots(figsize=(8, 6))
mask = np.triu(np.ones_like(factor_corr, dtype=bool), k=1)
sns.heatmap(factor_corr, annot=True, fmt='.2f', cmap='coolwarm',
            center=0, square=True, linewidths=0.5, ax=ax, mask=mask)
ax.set_title('因子相关性矩阵', fontsize=13)
plt.tight_layout()
plt.show()

# 高相关因子对
high_corr = []
for i, c1 in enumerate(feature_cols):
    for j, c2 in enumerate(feature_cols):
        if i < j and abs(factor_corr.loc[c1, c2]) > 0.6:
            high_corr.append((c1, c2, factor_corr.loc[c1, c2]))

if high_corr:
    print('  高度相关的因子对（|corr| > 0.6，可能冗余）：')
    for c1, c2, corr in sorted(high_corr, key=lambda x: -abs(x[2])):
        print(f'  {c1} ↔ {c2}: {corr:.2f}')
else:
    print(' 无高度相关因子对')

factor_corr = factors[feature_cols].corr() fig, ax = plt.subplots(figsize=(8, 6)) mask = np.triu(np.ones_like(factor_corr, dtype=bool), k=1) sns.heatmap(factor_corr, annot=True, fmt='.2f', cmap='coolwarm', center=0, square=True, linewidths=0.5, ax=ax, mask=mask) ax.set_title('因子相关性矩阵', fontsize=13) plt.tight_layout() plt.show() # 高相关因子对 high_corr = [] for i, c1 in enumerate(feature_cols): for j, c2 in enumerate(feature_cols): if i < j and abs(factor_corr.loc[c1, c2]) > 0.6: high_corr.append((c1, c2, factor_corr.loc[c1, c2])) if high_corr: print(' 高度相关的因子对（|corr| > 0.6，可能冗余）：') for c1, c2, corr in sorted(high_corr, key=lambda x: -abs(x[2])): print(f' {c1} ↔ {c2}: {corr:.2f}') else: print(' 无高度相关因子对')

  高度相关的因子对（|corr| > 0.6，可能冗余）：
  rsi_14 ↔ bb_pct: 0.89
  mom_20d ↔ price_ma_ratio: 0.85
  rsi_14 ↔ price_ma_ratio: 0.81
  bb_pct ↔ price_ma_ratio: 0.80
  mom_5d ↔ price_ma_ratio: 0.76
  mom_20d ↔ rsi_14: 0.75
  mom_5d ↔ bb_pct: 0.68
  mom_20d ↔ bb_pct: 0.62
  mom_5d ↔ rsi_14: 0.61

5. 因子分组回测（分层测试）¶

将股票按因子值分成 5 组，看各组的平均下期收益是否单调。

In [7]:

best_factor = ic_summary.index[0]  # 取 IC 最大的因子

# 按月截面分层
quintile_returns = []
for date, group in factors.groupby(pd.Grouper(freq='ME', level=0)):
    if len(group) < 5:
        continue
    group = group.copy()
    group['quintile'] = pd.qcut(group[best_factor], 5, labels=[1, 2, 3, 4, 5])
    for q, qgroup in group.groupby('quintile', observed=True):
        quintile_returns.append({'date': date, 'quintile': int(q),
                                  'avg_fwd_ret': qgroup['fwd_ret_20d'].mean()})

qdf = pd.DataFrame(quintile_returns)
group_means = qdf.groupby('quintile')['avg_fwd_ret'].mean()

fig, ax = plt.subplots(figsize=(7, 4))
colors = ['red', 'salmon', 'gray', 'lightgreen', 'green']
bars = ax.bar(group_means.index, group_means.values * 100, color=colors, alpha=0.85)
ax.axhline(0, color='black', linewidth=0.8)
ax.set_xlabel('因子分组（1=最低，5=最高）')
ax.set_ylabel('平均20日远期收益率 (%)')
ax.set_title(f'因子分层测试: {best_factor}', fontsize=13)
for bar, val in zip(bars, group_means.values):
    ax.text(bar.get_x() + bar.get_width() / 2,
             bar.get_height() + 0.01, f'{val:.2%}', ha='center', fontsize=10)
plt.tight_layout()
plt.show()
print('样本中的分组收益单调性需要结合样本外稳定性和交易成本解释。')

best_factor = ic_summary.index[0] # 取 IC 最大的因子 # 按月截面分层 quintile_returns = [] for date, group in factors.groupby(pd.Grouper(freq='ME', level=0)): if len(group) < 5: continue group = group.copy() group['quintile'] = pd.qcut(group[best_factor], 5, labels=[1, 2, 3, 4, 5]) for q, qgroup in group.groupby('quintile', observed=True): quintile_returns.append({'date': date, 'quintile': int(q), 'avg_fwd_ret': qgroup['fwd_ret_20d'].mean()}) qdf = pd.DataFrame(quintile_returns) group_means = qdf.groupby('quintile')['avg_fwd_ret'].mean() fig, ax = plt.subplots(figsize=(7, 4)) colors = ['red', 'salmon', 'gray', 'lightgreen', 'green'] bars = ax.bar(group_means.index, group_means.values * 100, color=colors, alpha=0.85) ax.axhline(0, color='black', linewidth=0.8) ax.set_xlabel('因子分组（1=最低，5=最高）') ax.set_ylabel('平均20日远期收益率 (%)') ax.set_title(f'因子分层测试: {best_factor}', fontsize=13) for bar, val in zip(bars, group_means.values): ax.text(bar.get_x() + bar.get_width() / 2, bar.get_height() + 0.01, f'{val:.2%}', ha='center', fontsize=10) plt.tight_layout() plt.show() print('样本中的分组收益单调性需要结合样本外稳定性和交易成本解释。')

样本中的分组收益单调性需要结合样本外稳定性和交易成本解释。

练习¶

1. 计算「价格/52周高点」因子的 IC，与 mom_60d 比较。
2. 增加更多股票（如标普500全部成分股），IC 结果会更稳定吗？
3. 用分层测试检验「低波动率因子」：波动率最低组的股票表现是否优于高波动率组？

---

下一节 → 02_ml_prediction.ipynb

In [ ]: