资产定价与因子模型

核心问题: 股票收益从哪里来？多少是市场给的（Beta），多少是能力创造的（Alpha）？

学习目标

理解有效市场假说及其与量化策略的关系
掌握 CAPM 模型，理解风险与收益的线性关系
学习 Fama-French 多因子模型的演进逻辑
理解 Barra 风险模型在业界的应用
用 Python 从零构建因子投资组合
核心: 明确 ML 模型预测的到底是什么

一、有效市场假说 (EMH)

1.1 什么是有效市场假说？

有效市场假说 (Efficient Market Hypothesis, EMH) 由 Eugene Fama 提出，核心观点：

资产价格已经反映所有可用信息，因此无法持续获得超额收益。

1.2 三种形式的有效市场

形式	信息集	启示	对量化策略的影响
弱式有效	历史价格、成交量	技术分析无效	纯价格策略难赚钱
半强式有效	公开信息（财报、新闻）	基本分析无效	需要挖掘非公开信息
强式有效	所有信息（包括内幕）	内幕交易也无效	几乎不可能获得 Alpha

1.3 为什么市场不完全有效？

如果市场完全有效，为什么还有人能赚钱？现实中的市场摩擦创造了机会：

流动性约束: 机构资金量大，难以快速调仓
卖空限制: A 股融券成本高、券源少
行为偏差: 散户非理性交易（追涨杀跌）
信息扩散慢: 复杂信息需要时间被市场消化
套利成本: 交易费用、价格冲击、风险限制

1.4 EMH 对量化 ML 的启示

# EMH 的分层启示
market_efficiency = {
    "弱式有效": {
        "含义": "历史价格信息已被利用",
        "ML策略": "不要只用价格序列做特征",
        "可行方向": "挖掘另类数据（新闻、情绪、基本面）"
    },
    "半强式有效": {
        "含义": "公开信息快速被定价",
        "ML策略": "比拼信息处理速度和质量",
        "可行方向": "NLP 处理非结构化信息、高频交易"
    },
    "现实": {
        "状态": "介于弱式和半强式之间",
        "机会": "短期低效（行为偏差）、长期低效（制度约束）",
        "ML优势": "发现人类难以察觉的复杂模式"
    }
}

量化策略生存法则: 寻找市场低效的角落，在有效市场中赚取风险溢价。

二、CAPM 模型

2.1 模型动机

CAPM (Capital Asset Pricing Model) 回答一个核心问题：

投资者承担风险，应该获得多少收益补偿？

2.2 模型公式

$E (R_{i}) = R_{f} + β_{i} \times [E (R_{m}) - R_{f}]$

其中：

$E (R_{i})$ : 资产 i 的预期收益
$R_{f}$ : 无风险收益率
$β_{i}$ : 资产 i 对市场风险的敏感度
$E (R_{m}) - R_{f}$ : 市场风险溢价

2.3 直观理解

预期收益 = 无风险收益 + β × 市场风险溢价

示例：
- 无风险收益 Rf = 3%
- 市场风险溢价 = 7%
- 股票 A 的 β = 1.2

则：E(Ra) = 3% + 1.2 × 7% = 11.4%

解释：
- 3% 是时间价值的补偿（等待）
- 8.4% 是风险的补偿（承担市场波动）

2.4 Beta 的计算

$β_{i} = \frac{C o v ( R _{i} , R _{m} )}{V a r ( R _{m} )} = ρ_{i, m} \times \frac{σ _{i}}{σ _{m}}$

Beta 的含义:

$β = 1$ : 与市场同步波动
$β > 1$ : 放大市场波动（进攻型）
$β < 1$ : 收敛市场波动（防御型）
$β < 0$ : 与市场反向（对冲工具）

2.5 Alpha 的定义

Alpha 是实际收益与 CAPM 预期收益的差值：

$α_{i} = R_{i} - [R_{f} + β_{i} \times (R_{m} - R_{f})]$

Alpha 的含义:

$α > 0$ : 跑赢预期（选股能力）
$α = 0$ : 符合预期（市场定价合理）
$α < 0$ : 跑输预期（选股失误）

2.6 Python 实现 CAPM

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
 
# 设置随机种子，确保结果可复现
np.random.seed(42)
 
# 模拟参数
n_days = 252  # 一年交易日
rf = 0.03     # 无风险收益率 3%
market_std = 0.20  # 市场年化波动率 20%
 
# 生成市场收益率（假设市场年化收益 10%）
market_daily_return = 0.10 / 252
market_daily_std = market_std / np.sqrt(252)
 
# 模拟市场收益
market_returns = np.random.normal(
    loc=market_daily_return,
    scale=market_daily_std,
    size=n_days
)
 
# 模拟三只不同 Beta 的股票
betas = {
    '防守型股票': 0.6,
    '市场股票': 1.0,
    '进攻型股票': 1.5
}
 
# 股票特异性波动率（不可分散的风险）
idiosyncratic_std = 0.15 / np.sqrt(252)
 
stock_returns = pd.DataFrame(index=range(n_days))
 
for name, beta in betas.items():
    # 股票收益 = β × 市场收益 + 特异性收益
    systematic_part = beta * market_returns
    idiosyncratic_part = np.random.normal(0, idiosyncratic_std, n_days)
    stock_returns[name] = systematic_part + idiosyncratic_part
 
# 计算累计收益
stock_cumulative = (1 + stock_returns).cumprod()
market_cumulative = (1 + pd.Series(market_returns)).cumprod()
 
# 计算统计指标
def calculate_capm_metrics(stock_returns, market_returns, rf_annual):
    """计算 CAPM 相关指标"""
    rf_daily = rf_annual / 252
 
    # 计算 Beta
    covariance = np.cov(stock_returns, market_returns)[0, 1]
    market_variance = np.var(market_returns)
    beta = covariance / market_variance
 
    # 计算 Alpha (年化)
    stock_annual_return = (1 + stock_returns.mean()) ** 252 - 1
    market_annual_return = (1 + market_returns.mean()) ** 252 - 1
    expected_return = rf_annual + beta * (market_annual_return - rf_annual)
    alpha = stock_annual_return - expected_return
 
    # 计算信息比率 (Alpha / Tracking Error)
    excess_returns = stock_returns - rf_daily - beta * (market_returns - rf_daily)
    tracking_error = np.std(excess_returns) * np.sqrt(252)
    information_ratio = alpha / tracking_error if tracking_error > 0 else 0
 
    return {
        'Beta': beta,
        'Alpha': alpha,
        '年化收益': stock_annual_return,
        '预期收益': expected_return,
        '信息比率': information_ratio
    }
 
# 计算每只股票的指标
metrics = {}
for stock in stock_returns.columns:
    metrics[stock] = calculate_capm_metrics(
        stock_returns[stock].values,
        market_returns,
        rf
    )
 
# 打印结果
print("=" * 60)
print("CAPM 分析结果")
print("=" * 60)
print(f"{'指标':<15} {'防守型':<12} {'市场股票':<12} {'进攻型':<12}")
print("-" * 60)
for metric in ['Beta', 'Alpha', '年化收益', '信息比率']:
    row = [metric]
    for stock in stock_returns.columns:
        row.append(f"{metrics[stock][metric]:.4f}")
    print(f"{row[0]:<15} {row[1]:<12} {row[2]:<12} {row[3]:<12}")
print("=" * 60)
 
# 绘制累计收益曲线
plt.figure(figsize=(12, 6))
plt.plot(market_cumulative.values, label='市场', linewidth=2, linestyle='--')
for stock in stock_returns.columns:
    plt.plot(stock_cumulative[stock].values, label=stock, linewidth=2)
plt.title('不同 Beta 股票的累计收益对比', fontsize=14)
plt.xlabel('交易日', fontsize=12)
plt.ylabel('累计收益 (初始=1)', fontsize=12)
plt.legend(fontsize=11)
plt.grid(True, alpha=0.3)
plt.tight_layout()
plt.show()
 
# 绘制收益分布
plt.figure(figsize=(12, 5))
plt.subplot(1, 2, 1)
plt.hist(market_returns, bins=50, alpha=0.7, label='市场', edgecolor='black')
plt.hist(stock_returns['进攻型股票'], bins=50, alpha=0.5, label='进攻型 (β=1.5)', edgecolor='black')
plt.xlabel('日收益率')
plt.ylabel('频数')
plt.title('收益分布对比')
plt.legend()
 
plt.subplot(1, 2, 2)
# 散点图展示股票收益与市场收益的关系
plt.scatter(market_returns, stock_returns['进攻型股票'], alpha=0.5, s=20)
# 添加拟合线
z = np.polyfit(market_returns, stock_returns['进攻型股票'], 1)
p = np.poly1d(z)
plt.plot(market_returns, p(market_returns), "r--", linewidth=2, label=f'拟合线 (β≈{z[0]:.2f})')
plt.xlabel('市场收益率')
plt.ylabel('股票收益率')
plt.title('CAPM: 股票收益 vs 市场收益')
plt.legend()
plt.grid(True, alpha=0.3)
plt.tight_layout()
plt.show()

2.7 CAPM 的局限与启示

局限	说明	对量化策略的启示
单因子	只有市场风险因子	需要更多风险因子解释收益
假设投资者同质	所有投资者预期相同	投资者异质性产生机会
历史数据估计	Beta 不稳定	需要动态调整 Beta 暴露
无法解释异象	动量、价值效应存在	这些”异象”是因子投资的基础

三、Fama-French 多因子模型

3.1 从单因子到多因子

CAPM 只有一个市场因子，但实证发现很多异象 (Anomalies):

已发现的市场异象：
├── 价值效应: 低市盈率股票长期收益高
├── 规模效应: 小盘股长期收益高
├── 动量效应: 过去上涨股票继续上涨
├── 质量效应: 财务健康公司收益高
└── 低波效应: 低波动股票风险调整后收益高

Fama 和 French 认为这些不是”异象”，而是风险因子——投资者承担了某种系统性风险，需要补偿。

3.2 三因子模型 (1992)

$E (R_{i}) - R_{f} = β_{i} \times (R_{m} - R_{f}) + s_{i} \times S M B + h_{i} \times H M L$

新增因子:

SMB (Small Minus Big): 小盘股因子
HML (High Minus Low): 价值因子（高账面市值比 - 低账面市值比）

因子构造逻辑

def construct_ff_factors(data):
    """
    Fama-French 三因子构造逻辑
 
    参数:
        data: 包含市值(ME)和账面市值比(BE/ME)的DataFrame
 
    返回:
        SMB, HML 因子收益
    """
    # 1. 按市值中位数分大/小盘
    me_median = data['ME'].median()
    data['size_group'] = np.where(data['ME'] > me_median, 'B', 'S')
 
    # 2. 按账面市值比三分位数分价值/中性/成长
    bm_30 = data['BE_ME'].quantile(0.3)
    bm_70 = data['BE_ME'].quantile(0.7)
 
    def get_value_group(bm):
        if bm <= bm_30:
            return 'L'  # Low = 成长
        elif bm >= bm_70:
            return 'H'  # High = 价值
        else:
            return 'N'  # Neutral
 
    data['value_group'] = data['BE_ME'].apply(get_value_group)
 
    # 3. 形成 6 个投资组合
    # SL: 小盘低BM(成长) | SM: 小盘中 | SH: 小盘高(价值)
    # BL: 大盘低BM(成长) | BM: 大盘中 | BH: 大盘高(价值)
 
    # 4. SMB = 1/3(小盘价值+小盘中+小盘成长) - 1/3(大盘价值+大盘中+大盘成长)
    #    = 做多小盘，做空大盘
 
    # 5. HML = 1/2(小盘价值+大盘价值) - 1/2(小盘成长+大盘成长)
    #    = 做多价值，做空成长
 
    return smb, hml

3.3 五因子模型 (2015)

$E (R_{i}) - R_{f} = β \times M K T + s \times S M B + h \times H M L + r \times R M W + c \times C M A$

新增因子:

RMW (Robust Minus Weak): 盈利能力因子（高盈利 - 低盈利）
CMA (Conservative Minus Aggressive): 投资因子（低投资 - 高投资）

逻辑:

高盈利公司 → 持续赚钱能力强 → 应该有溢价
保守投资公司 → 资本配置谨慎 → 应该有溢价

3.4 Python 实现 Fama-French 回归

import numpy as np
import pandas as pd
import statsmodels.api as sm
from sklearn.linear_model import LinearRegression
 
# 设置随机种子
np.random.seed(42)
 
# 模拟参数
n_days = 252
n_stocks = 100
 
# 模拟因子收益 (年化)
factor_annual_returns = {
    'MKT': 0.08,   # 市场风险溢价
    'SMB': 0.03,   # 小盘股溢价
    'HML': 0.04,   # 价值溢价
    'RMW': 0.03,   # 盈利能力溢价
    'CMA': 0.02    # 保守投资溢价
}
 
# 模拟因子收益率序列
factor_daily_returns = pd.DataFrame({
    factor: np.random.normal(ann/252, 0.01, n_days)
    for factor, ann in factor_annual_returns.items()
})
 
# 确保市场因子(MKT)与实际市场风险溢价相关
factor_daily_returns['MKT'] = np.random.normal(0.08/252, 0.01, n_days)
 
# 生成股票的因子暴露 (Beta 值)
# 每只股票对不同因子有不同的敏感度
np.random.seed(42)
factor_loadings = pd.DataFrame({
    'MKT': np.random.uniform(0.5, 1.5, n_stocks),
    'SMB': np.random.uniform(-0.5, 1.0, n_stocks),
    'HML': np.random.uniform(-0.5, 1.0, n_stocks),
    'RMW': np.random.uniform(-0.3, 0.8, n_stocks),
    'CMA': np.random.uniform(-0.3, 0.6, n_stocks),
}, index=[f'Stock_{i}' for i in range(n_stocks)])
 
# 生成股票收益率
stock_returns = pd.DataFrame(index=range(n_days))
alphas = {}  # 每只股票的真实 Alpha
 
for stock in factor_loadings.index:
    loadings = factor_loadings.loc[stock]
 
    # 系统性收益 = Σ(因子暴露 × 因子收益)
    systematic_return = (
        loadings['MKT'] * factor_daily_returns['MKT'] +
        loadings['SMB'] * factor_daily_returns['SMB'] +
        loadings['HML'] * factor_daily_returns['HML'] +
        loadings['RMW'] * factor_daily_returns['RMW'] +
        loadings['CMA'] * factor_daily_returns['CMA']
    )
 
    # 特异性收益 (不能被因子解释的部分)
    idiosyncratic = np.random.normal(0, 0.008, n_days)
 
    # 真实 Alpha (年化 3% 的正态分布，有些股票有，有些没有)
    true_alpha = np.random.normal(0.02, 0.015)
    alphas[stock] = true_alpha
 
    # 最终收益 = Alpha + 系统性收益 + 特异性收益
    stock_returns[stock] = true_alpha/252 + systematic_return + idiosyncratic
 
# 选取一只股票进行详细分析
sample_stock = 'Stock_0'
y = stock_returns[sample_stock].values
X = factor_daily_returns.values
X_with_const = sm.add_constant(X)
 
# 三因子回归
model_3factor = sm.OLS(
    y,
    sm.add_constant(factor_daily_returns[['MKT', 'SMB', 'HML']].values)
).fit()
 
# 五因子回归
model_5factor = sm.OLS(y, X_with_const).fit()
 
# 打印回归结果
print("=" * 80)
print(f"股票 {sample_stock} 的因子分析")
print("=" * 80)
 
print("\n【三因子模型回归结果】")
print("-" * 80)
print(f"{'因子':<15} {'系数':<12} {'t值':<12} {'p值':<12} {'解释':<20}")
print("-" * 80)
factor_names_3 = ['Alpha', 'MKT', 'SMB', 'HML']
for i, name in enumerate(factor_names_3):
    coef = model_3factor.params[i]
    t_val = model_3factor.tvalues[i]
    p_val = model_3factor.pvalues[i]
 
    explanations = [
        '超额收益能力',
        '市场风险暴露',
        '小盘股暴露',
        '价值股暴露'
    ]
    print(f"{name:<15} {coef:>10.4f}   {t_val:>10.2f}   {p_val:>10.4f}   {explanations[i]}")
 
print(f"\nR-squared: {model_3factor.rsquared:.4f}")
print(f"Adjusted R-squared: {model_3factor.rsquared_adj:.4f}")
 
print("\n" + "=" * 80)
print("【五因子模型回归结果】")
print("-" * 80)
print(f"{'因子':<15} {'系数':<12} {'t值':<12} {'p值':<12} {'显著吗':<10}")
print("-" * 80)
factor_names_5 = ['Alpha', 'MKT', 'SMB', 'HML', 'RMW', 'CMA']
for i, name in enumerate(factor_names_5):
    coef = model_5factor.params[i]
    t_val = model_5factor.tvalues[i]
    p_val = model_5factor.pvalues[i]
    significant = "***" if p_val < 0.01 else "**" if p_val < 0.05 else "*" if p_val < 0.1 else ""
    print(f"{name:<15} {coef:>10.4f}   {t_val:>10.2f}   {p_val:>10.4f}   {significant:<10}")
 
print(f"\nR-squared: {model_5factor.rsquared:.4f}")
print(f"Adjusted R-squared: {model_5factor.rsquared_adj:.4f}")
print("=" * 80)
 
# 比较三因子和五因子
print("\n【模型对比】")
print(f"三因子 R²: {model_3factor.rsquared:.4f}")
print(f"五因子 R²: {model_5factor.rsquared:.4f}")
print(f"R² 提升: {(model_5factor.rsquared - model_3factor.rsquared)*100:.2f}%")
 
# 因子收益贡献分解
print("\n【收益来源分解】（五因子模型，年化）")
print("-" * 60)
total_return = (stock_returns[sample_stock].mean() * 252)
contributions = {
    'Alpha (年化)': model_5factor.params[0] * 252,
    'MKT 贡献': model_5factor.params[1] * factor_annual_returns['MKT'],
    'SMB 贡献': model_5factor.params[2] * factor_annual_returns['SMB'],
    'HML 贡献': model_5factor.params[3] * factor_annual_returns['HML'],
    'RMW 贡献': model_5factor.params[4] * factor_annual_returns['RMW'],
    'CMA 贡献': model_5factor.params[5] * factor_annual_returns['CMA'],
}
 
for source, contrib in contributions.items():
    pct = contrib / total_return * 100 if total_return != 0 else 0
    print(f"{source:<15}: {contrib:>7.2%} ({pct:>6.2f}%)")
print("-" * 60)
print(f"{'总收益':<15}: {total_return:>7.2%}")
print("=" * 80)
 
# 可视化
fig, axes = plt.subplots(2, 2, figsize=(14, 10))
 
# 1. 累计收益对比
ax1 = axes[0, 0]
cumulative_stock = (1 + stock_returns[sample_stock]).cumprod()
cumulative_predicted = (1 + (
    model_5factor.params[0] +
    model_5factor.params[1] * factor_daily_returns['MKT'] +
    model_5factor.params[2] * factor_daily_returns['SMB'] +
    model_5factor.params[3] * factor_daily_returns['HML'] +
    model_5factor.params[4] * factor_daily_returns['RMW'] +
    model_5factor.params[5] * factor_daily_returns['CMA']
)).cumprod()
ax1.plot(cumulative_stock.values, label='实际收益', linewidth=2)
ax1.plot(cumulative_predicted.values, label='因子模型预测', linewidth=2, linestyle='--')
ax1.set_title('实际收益 vs 因子模型预测', fontsize=12)
ax1.set_xlabel('交易日')
ax1.set_ylabel('累计收益')
ax1.legend()
ax1.grid(True, alpha=0.3)
 
# 2. 残差分析
ax2 = axes[0, 1]
residuals = model_5factor.resid
ax2.hist(residuals, bins=50, edgecolor='black', alpha=0.7)
ax2.axvline(0, color='red', linestyle='--', linewidth=2)
ax2.set_title('残差分布（特异性收益）', fontsize=12)
ax2.set_xlabel('残差')
ax2.set_ylabel('频数')
ax2.grid(True, alpha=0.3)
 
# 3. 因子暴露条形图
ax3 = axes[1, 0]
factor_exposure = model_5factor.params[1:]
colors = ['steelblue' if x > 0 else 'coral' for x in factor_exposure]
ax3.barh(factor_names_5[1:], factor_exposure, color=colors, edgecolor='black')
ax3.axvline(0, color='black', linewidth=0.8)
ax3.set_title('因子暴露（Beta 系数）', fontsize=12)
ax3.set_xlabel('暴露值')
ax3.grid(True, alpha=0.3, axis='x')
 
# 4. 预测 vs 实际散点图
ax4 = axes[1, 1]
predicted = model_5factor.predict(X_with_const)
ax4.scatter(predicted, y, alpha=0.5, s=20)
min_val = min(predicted.min(), y.min())
max_val = max(predicted.max(), y.max())
ax4.plot([min_val, max_val], [min_val, max_val], 'r--', linewidth=2)
ax4.set_title('预测收益 vs 实际收益', fontsize=12)
ax4.set_xlabel('因子模型预测收益')
ax4.set_ylabel('实际收益')
ax4.grid(True, alpha=0.3)
 
plt.tight_layout()
plt.show()

3.5 因子相关性分析

import seaborn as sns
 
# 计算因子相关系数矩阵
factor_corr = factor_daily_returns.corr()
 
# 绘制相关性热力图
plt.figure(figsize=(10, 8))
sns.heatmap(
    factor_corr,
    annot=True,
    fmt='.2f',
    cmap='RdBu_r',
    center=0,
    vmin=-1,
    vmax=1,
    square=True,
    linewidths=1,
    cbar_kws={'label': '相关系数'}
)
plt.title('Fama-French 五因子相关性矩阵', fontsize=14, pad=20)
plt.tight_layout()
plt.show()
 
print("【因子相关性分析】")
print("-" * 50)
print("因子相关系数反映了风险因子的独立性和共线性")
print("高相关性 (>0.5) 说明因子可能有重叠")
print("低相关性说明因子捕捉了不同的风险源")
print("-" * 50)

四、APT 套利定价理论

4.1 与 CAPM 的区别

特性	CAPM	APT
理论基础	均衡模型	套利无边界
因子数量	1个（市场）	多个（不确定）
因子识别	明确（市场收益）	待定（实证发现）
假设强度	强（投资者同质等）	弱（无套利机会）

4.2 APT 公式

$E (R_{i}) = R_{f} + β_{i 1} \times R P_{1} + β_{i 2} \times R P_{2} + \dots + β_{in} \times R P_{n}$

其中 $R P_{j}$ 是因子 j 的风险溢价。

4.3 APT 的实用价值

APT 不告诉你具体有哪些因子，但提供因子投资的哲学基础：

收益来自对多个系统性风险的暴露
因子需要经济学直觉支持，不能只是数据挖掘
因子之间应该低相关，否则重复计算风险

五、Barra 风险模型

5.1 业界标准

Barra 风险模型（现 MSCI Barra）是机构投资者的核心工具：

收益分解方程：

r_i = α_i + Σ(β_ij × f_j) + ε_i

其中：
- r_i: 股票 i 的收益
- α_i: 特异性收益（选股 Alpha）
- β_ij: 股票 i 对因子 j 的暴露
- f_j: 因子 j 的收益
- ε_i: 残差（不能被因子解释）

5.2 Barra 因子体系

因子类型	具体因子	计算方式
风格因子	规模	ln(市值)
	波动率	日收益波动率
	动量	过去12个月累计收益
	价值	市净率倒数
	盈利质量	ROE、ROA
	成长	营收/净利润增长率
	杠杆	总资产/股东权益
行业因子	中信一级行业	虚拟变量（属于=1，不属于=0）
市场因子	市场收益	Cap加权市场收益

5.3 Barra 模型 Python 实现

import numpy as np
import pandas as pd
from sklearn.linear_model import Ridge
from sklearn.preprocessing import StandardScaler
 
np.random.seed(42)
 
# 模拟参数
n_stocks = 200
n_days = 252
 
# 1. 生成风格因子暴露 (每个股票的基本面特征)
style_factors = pd.DataFrame({
    'Size': np.random.normal(0, 1, n_stocks),          # 规模因子
    'Beta': np.random.normal(1, 0.3, n_stocks),        # Beta因子
    'Momentum': np.random.normal(0, 1, n_stocks),      # 动量因子
    'Value': np.random.normal(0, 1, n_stocks),         # 价值因子
    'Volatility': np.random.normal(0, 1, n_stocks),    # 波动率因子
    'Profit': np.random.normal(0, 1, n_stocks),        # 盈利能力
    'Growth': np.random.normal(0, 1, n_stocks),        # 成长因子
    'Leverage': np.random.normal(0, 1, n_stocks),      # 杠杆因子
}, index=[f'Stock_{i}' for i in(n_stocks)])
 
# 2. 生成行业暴露 (每个股票只属于一个行业)
industries = ['银行', '地产', '医药', '科技', '消费', '能源', '材料', '工业']
industry_exposure = pd.DataFrame(
    np.zeros((n_stocks, len(industries))),
    columns=industries,
    index=style_factors.index
)
# 随机分配行业
for stock in style_factors.index:
    assigned_industry = np.random.choice(industries)
    industry_exposure.loc[stock, assigned_industry] = 1
 
# 3. 模拟因子收益率序列
factor_returns = pd.DataFrame({
    'Size': np.random.normal(0.0002, 0.005, n_days),
    'Beta': np.random.normal(0.0001, 0.003, n_days),
    'Momentum': np.random.normal(0.0003, 0.006, n_days),
    'Value': np.random.normal(0.0001, 0.004, n_days),
    'Volatility': np.random.normal(-0.0001, 0.004, n_days),
    'Profit': np.random.normal(0.0002, 0.003, n_days),
    'Growth': np.random.normal(0.0001, 0.005, n_days),
    'Leverage': np.random.normal(-0.0001, 0.003, n_days),
})
 
# 添加行业因子收益
for industry in industries:
    factor_returns[industry] = np.random.normal(0.0002, 0.004, n_days)
 
# 4. 生成股票收益率
stock_returns = pd.DataFrame(index=range(n_days), columns=style_factors.index)
 
# 每只股票的特异性波动率
specific_vol = np.random.uniform(0.008, 0.02, n_stocks)
 
for i, stock in enumerate(style_factors.index):
    # 风格因子贡献
    style_contribution = (
        style_factors.loc[stock, 'Size'] * factor_returns['Size'] +
        style_factors.loc[stock, 'Beta'] * factor_returns['Beta'] +
        style_factors.loc[stock, 'Momentum'] * factor_returns['Momentum'] +
        style_factors.loc[stock, 'Value'] * factor_returns['Value'] +
        style_factors.loc[stock, 'Volatility'] * factor_returns['Volatility'] +
        style_factors.loc[stock, 'Profit'] * factor_returns['Profit'] +
        style_factors.loc[stock, 'Growth'] * factor_returns['Growth'] +
        style_factors.loc[stock, 'Leverage'] * factor_returns['Leverage']
    )
 
    # 行业因子贡献
    industry_contribution = sum(
        industry_exposure.loc[stock, ind] * factor_returns[ind]
        for ind in industries
    )
 
    # 特异性收益
    specific_return = np.random.normal(0, specific_vol[i], n_days)
 
    # 添加一些真实的 Alpha（选股能力）
    true_alpha = np.random.normal(0.0001, 0.0002)
 
    stock_returns[stock] = (
        true_alpha + style_contribution +
        industry_contribution + specific_return
    )
 
# 5. 横截面回归：每天做一次回归，提取因子收益
estimated_factor_returns = []
factor_covariances = []  # 因子协方差矩阵
specific_risks = []      # 特异性风险
 
for day in range(n_days):
    # 当天的股票收益率
    y = stock_returns.iloc[day].values
 
    # 合并风格因子和行业因子暴露
    X = pd.concat([
        style_factors,
        industry_exposure
    ], axis=1).values
 
    # 加权最小二乘回归（用市值作为权重）
    # 这里简单用等权重
    weights = np.ones(n_stocks)
 
    try:
        # 使用岭回归避免多重共线性
        model = Ridge(alpha=0.1)
        model.fit(X, y, sample_weight=weights)
 
        # 记录因子收益（去掉截距项即 Alpha）
        factor_return = pd.Series(
            model.coef_,
            index=list(style_factors.columns) + industries
        )
        estimated_factor_returns.append(factor_return)
 
        # 计算特异性风险
        residuals = y - model.predict(X)
        specific_std = np.std(residuals)
        specific_risks.append(specific_std)
 
    except Exception as e:
        # 如果回归失败，用上一天的值
        if estimated_factor_returns:
            estimated_factor_returns.append(estimated_factor_returns[-1])
            specific_risks.append(specific_risks[-1])
 
# 转换为 DataFrame
estimated_factor_returns = pd.DataFrame(estimated_factor_returns)
 
# 6. 分析结果
print("=" * 80)
print("Barra 风险模型分析结果")
print("=" * 80)
 
# 因子收益率统计
print("\n【风格因子收益率统计】（年化）")
print("-" * 80)
style_cols = list(style_factors.columns)
factor_stats = estimated_factor_returns[style_cols].describe()
factor_annual = factor_stats.loc[['mean', 'std']] * 252
factor_annual.loc['夏普比率'] = factor_annual.loc['mean'] / factor_annual.loc['std'] * np.sqrt(252)
 
print(f"{'因子':<15} {'年化收益':<12} {'年化波动':<12} {'夏普比率':<12}")
print("-" * 80)
for factor in style_cols:
    ann_return = factor_annual.loc['mean', factor]
    ann_std = factor_annual.loc['std', factor]
    sharpe = factor_annual.loc['夏普比率', factor]
    print(f"{factor:<15} {ann_return:>10.2%}   {ann_std:>10.2%}   {sharpe:>10.2f}")
 
print("\n【行业因子收益率统计】（年化）")
print("-" * 80)
industry_stats = estimated_factor_returns[industries].describe()
industry_annual = industry_stats.loc['mean'] * 252
 
print(f"{'行业':<10} {'年化收益':<12}")
print("-" * 80)
for ind in industries:
    print(f"{ind:<10} {industry_annual[ind]:>10.2%}")
 
# 因子相关性分析
print("\n【风格因子相关性矩阵】")
print("-" * 80)
factor_corr = estimated_factor_returns[style_cols].corr()
print(factor_corr.round(2))
 
# 7. 风险归因示例：对某只股票进行风险分解
sample_stock = 'Stock_0'
stock_return_mean = stock_returns[sample_stock].mean() * 252
 
# 风格因子贡献
style_contribution = sum(
    style_factors.loc[sample_stock, factor] * estimated_factor_returns[factor].mean() * 252
    for factor in style_cols
)
 
# 行业因子贡献
stock_industry = industry_exposure.loc[sample_stock].idxmax()
industry_contribution = estimated_factor_returns[stock_industry].mean() * 252
 
# 特异性贡献（用平均特异性波动率估算）
specific_contribution = np.mean(specific_risks) * np.sqrt(252) * 0.5  # 假设半倍标准差
 
print("\n【风险归因分析】")
print("-" * 80)
print(f"股票: {sample_stock}")
print(f"所属行业: {stock_industry}")
print("-" * 80)
print(f"{'组成部分':<20} {'年化贡献':<15} {'占比':<10}")
print("-" * 80)
print(f"{'风格因子合计':<20} {style_contribution:>13.2%}   {style_contribution/stock_return_mean*100:>6.1f}%")
print(f"{'行业因子贡献':<20} {industry_contribution:>13.2%}   {industry_contribution/stock_return_mean*100:>6.1f}%")
print(f"{'特异性收益估计':<20} {specific_contribution:>13.2%}   {specific_contribution/stock_return_mean*100:>6.1f}%")
print("-" * 80)
print(f"{'总收益':<20} {stock_return_mean:>13.2%}   {100.0:>6.1f}%")
 
# 8. 可视化
fig, axes = plt.subplots(2, 2, figsize=(14, 10))
 
# 累计因子收益
ax1 = axes[0, 0]
for factor in ['Momentum', 'Value', 'Size']:
    cumulative = (1 + estimated_factor_returns[factor]).cumprod()
    ax1.plot(cumulative.values, label=factor, linewidth=2)
ax1.set_title('主要风格因子累计收益', fontsize=12)
ax1.set_xlabel('交易日')
ax1.set_ylabel('累计收益')
ax1.legend()
ax1.grid(True, alpha=0.3)
 
# 因子收益率分布
ax2 = axes[0, 1]
factor_means = estimated_factor_returns[style_cols].mean() * 252
factor_stds = estimated_factor_returns[style_cols].std() * np.sqrt(252)
ax2.scatter(factor_means, factor_stds, s=100, alpha=0.7, edgecolors='black')
for factor in style_cols:
    ax2.annotate(
        factor,
        (factor_means[factor], factor_stds[factor]),
        fontsize=9, ha='center'
    )
ax2.axhline(0, color='red', linestyle='--', alpha=0.5)
ax2.axvline(0, color='red', linestyle='--', alpha=0.5)
ax2.set_xlabel('年化收益率')
ax2.set_ylabel('年化波动率')
ax2.set_title('因子风险-收益分布', fontsize=12)
ax2.grid(True, alpha=0.3)
 
# 行业因子收益条形图
ax3 = axes[1, 0]
industry_annual_sorted = industry_annual.sort_values(ascending=False)
colors = ['green' if x > 0 else 'red' for x in industry_annual_sorted.values]
ax3.barh(industry_annual_sorted.index, industry_annual_sorted.values, color=colors, edgecolor='black')
ax3.axvline(0, color='black', linewidth=0.8)
ax3.set_xlabel('年化收益率')
ax3.set_title('行业因子收益率', fontsize=12)
ax3.grid(True, alpha=0.3, axis='x')
 
# 特异性风险分布
ax4 = axes[1, 1]
ax4.hist(specific_risks, bins=50, edgecolor='black', alpha=0.7)
ax4.axvline(np.mean(specific_risks), color='red', linestyle='--', linewidth=2, label=f'均值: {np.mean(specific_risks):.4f}')
ax4.set_xlabel('日特异性波动率')
ax4.set_ylabel('频数')
ax4.set_title('特异性风险分布', fontsize=12)
ax4.legend()
ax4.grid(True, alpha=0.3)
 
plt.tight_layout()
plt.show()

六、因子投资的实践问题

6.1 因子拥挤 (Factor Crowding)

定义: 当太多资金追逐同一因子，因子收益会下降甚至反转。

检测指标:

def detect_factor_crowding(factor_returns, prices):
    """
    检测因子拥挤度
 
    指标:
    1. 因子相关性上升
    2. 因子波动率下降（同质化交易）
    3. 因子收益偏度负化（拥挤时更容易崩盘）
    """
    # 滚动相关性
    rolling_corr = factor_returns.rolling(60).corr()
 
    # 滚动波动率
    rolling_vol = factor_returns.rolling(60).std()
 
    # 滚动偏度
    rolling_skew = factor_returns.rolling(60).skew()
 
    return {
        'correlation_trend': rolling_corr.iloc[-1].mean(),
        'volatility_decline': rolling_vol.iloc[-1] / rolling_vol.mean(),
        'skewness_warning': rolling_skew.iloc[-1] < -1
    }

6.2 因子衰减 (Factor Decay)

现象: 发现的因子在使用一段时间后收益消失。

原因:

数据挖掘偏差: 原本就是假象
市场学习: 其他机构发现并交易，套利机会消失
制度变化: 市场规则改变（如 T+0 → T+1）

应对:

样本外测试严格
因子组合需要动态调整
理解因子的经济学逻辑

6.3 风格轮动 (Style Rotation)

不同时期表现最好的风格不同：

经济周期 → 风格轮动路径:

衰退 → 复苏 → 扩张 → 滞胀
  ↓        ↓        ↓        ↓
防守      价值      成长      质量
(高股息)  (低估值)  (高增长)  (稳定盈利)

识别方法: 宏观指标（PMI、利率、信用利差）→ 预测下一阶段强势风格

七、动量 vs 反转

7.1 两种时间序列模式

价格路径示意:

动量效应 (Momentum):
  价格 ─────────────────────╮
                            ╰╮╭─ 上涨持续
                              ╰

反转效应 (Reversal):
  价格 ────────╭╮
              ╰╯────── 回落

7.2 时间序列动量 vs 截面动量

类型	定义	策略
时间序列动量	过去上涨的资产未来继续上涨	做多过去赢家，做空过去输家
截面动量	相对其他资产表现好的资产未来继续好	买入最强势的，卖出最弱势的

7.3 Python 实现动量策略

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
 
np.random.seed(42)
 
# 生成模拟价格数据（带有动量特征）
n_days = 500
n_stocks = 10
 
# 基础收益
base_returns = np.random.normal(0.0005, 0.02, (n_days, n_stocks))
 
# 添加动量效应：过去上涨的股票倾向于继续上涨
momentum_signal = pd.DataFrame(index=range(n_days), columns=range(n_stocks))
for t in range(20, n_days):
    # 过去20天的累计收益作为动量信号
    past_returns = base_returns[t-20:t]
    cumulative_momentum = np.sum(past_returns, axis=0)
 
    # 动量越强，未来收益越高（效应强度为信号方差的20%）
    momentum_effect = 0.2 * cumulative_momentum / 20
    base_returns[t] += momentum_effect
 
# 生成价格序列
prices = pd.DataFrame(
    np.cumprod(1 + base_returns, axis=0),
    columns=[f'Stock_{i}' for i in range(n_stocks)]
)
 
# 1. 时间序列动量策略
def time_series_momentum(prices, lookback=20, holding=5):
    """
    时间序列动量策略
 
    参数:
        prices: 价格 DataFrame
        lookback: 动量回望期
        holding: 持有期
    """
    returns = prices.pct_change()
    signals = pd.DataFrame(index=prices.index, columns=prices.columns)
 
    for i in range(lookback, len(prices)):
        # 计算过去 lookback 天的累计收益率
        past_cumret = (prices.iloc[i-lookback:i] / prices.iloc[i-lookback] - 1).iloc[-1]
 
        # 正收益做多，负收益做空
        signals.iloc[i] = np.sign(past_cumret)
 
    # 计算 holding 期后的收益
    strategy_returns = pd.Series(index=prices.index)
    for i in range(lookback, len(prices) - holding):
        if i % holding == 0:  # 每 holding 天调仓
            current_signal = signals.iloc[i]
            future_returns = returns.iloc[i:i+holding]
            # 等权重做多或做空
            strategy_returns.iloc[i] = (
                current_signal * future_returns
            ).mean().mean()  # 横截面平均
 
    return strategy_returns, signals
 
# 2. 截面动量策略
def cross_sectional_momentum(prices, lookback=20, top_n=3):
    """
    截面动量策略
 
    参数:
        prices: 价格 DataFrame
        lookback: 动量回望期
        top_n: 买入前 n 名，做空后 n 名
    """
    returns = prices.pct_change()
    signals = pd.DataFrame(index=prices.index, columns=prices.columns)
 
    for i in range(lookback, len(prices)):
        # 计算每只股票的过去累计收益
        momentum = (prices.iloc[i-lookback:i] / prices.iloc[i-lookback] - 1).iloc[-1]
 
        # 排名
        ranked = momentum.rank()
 
        # 买入前 top_n，做空后 top_n
        signals.iloc[i] = np.where(
            ranked >= len(prices.columns) - top_n + 1,  # 前几名
            1,  # 做多
            np.where(ranked <= top_n, -1, 0)  # 后几名做空，其他不持有
        )
 
    # 计算策略收益
    strategy_returns = (signals.shift(1) * returns).mean(axis=1)
 
    return strategy_returns, signals
 
# 运行策略
ts_momentum_returns, ts_signals = time_series_momentum(prices)
cs_momentum_returns, cs_signals = cross_sectional_momentum(prices)
 
# 计算累计收益
ts_cumulative = (1 + ts_momentum_returns.fillna(0)).cumprod()
cs_cumulative = (1 + cs_momentum_returns.fillna(0)).cumprod()
market_cumulative = (1 + prices.pct_change().mean(axis=1)).cumprod()
 
# 计算策略指标
def calculate_performance_metrics(returns):
    """计算策略绩效指标"""
    returns = returns.dropna()
    if len(returns) == 0:
        return {}
 
    annual_return = (1 + returns.mean()) ** 252 - 1
    annual_vol = returns.std() * np.sqrt(252)
    sharpe = annual_return / annual_vol if annual_vol > 0 else 0
 
    # 最大回撤
    cumulative = (1 + returns).cumprod()
    running_max = cumulative.expanding().max()
    drawdown = (cumulative - running_max) / running_max
    max_drawdown = drawdown.min()
 
    return {
        '年化收益': annual_return,
        '年化波动': annual_vol,
        '夏普比率': sharpe,
        '最大回撤': max_drawdown
    }
 
ts_metrics = calculate_performance_metrics(ts_momentum_returns)
cs_metrics = calculate_performance_metrics(cs_momentum_returns)
market_metrics = calculate_performance_metrics(prices.pct_change().mean(axis=1))
 
# 打印结果
print("=" * 80)
print("动量策略回测结果")
print("=" * 80)
 
print("\n【策略对比】")
print("-" * 80)
print(f"{'指标':<15} {'时间序列动量':<18} {'截面动量':<15} {'买入持有':<15}")
print("-" * 80)
 
for metric in ['年化收益', '年化波动', '夏普比率', '最大回撤']:
    ts_val = ts_metrics.get(metric, 0)
    cs_val = cs_metrics.get(metric, 0)
    mkt_val = market_metrics.get(metric, 0)
 
    if metric == '年化收益' or metric == '年化波动':
        print(f"{metric:<15} {ts_val:>15.2%}  {cs_val:>15.2%}  {mkt_val:>15.2%}")
    else:
        print(f"{metric:<15} {ts_val:>15.2f}  {cs_val:>15.2f}  {mkt_val:>15.2f}")
 
print("=" * 80)
 
# 可视化
fig, axes = plt.subplots(2, 2, figsize=(14, 10))
 
# 1. 累计收益对比
ax1 = axes[0, 0]
ax1.plot(ts_cumulative.values, label='时间序列动量', linewidth=2)
ax1.plot(cs_cumulative.values, label='截面动量', linewidth=2)
ax1.plot(market_cumulative.values, label='等权市场', linewidth=2, linestyle='--')
ax1.set_title('累计收益对比', fontsize=12)
ax1.set_xlabel('交易日')
ax1.set_ylabel('累计收益')
ax1.legend()
ax1.grid(True, alpha=0.3)
 
# 2. 动量信号热力图（截面动量）
ax2 = axes[0, 1]
# 取最后 100 天的信号
signal_sample = cs_signals.iloc[-100:]
im = ax2.imshow(signal_sample.T, aspect='auto', cmap='RdBu_r', vmin=-1, vmax=1)
ax2.set_title('截面动量信号热力图（最近100天）', fontsize=12)
ax2.set_xlabel('交易日')
ax2.set_ylabel('股票')
plt.colorbar(im, ax=ax2, label='信号（红=做多，蓝=做空）')
 
# 3. 滚动夏普比率
ax3 = axes[1, 0]
window = 60
ts_rolling_sharpe = (
    ts_momentum_returns.rolling(window).mean() /
    ts_momentum_returns.rolling(window).std() * np.sqrt(252)
)
cs_rolling_sharpe = (
    cs_momentum_returns.rolling(window).mean() /
    cs_momentum_returns.rolling(window).std() * np.sqrt(252)
)
ax3.plot(ts_rolling_sharpe.values, label='时间序列动量', linewidth=2)
ax3.plot(cs_rolling_sharpe.values, label='截面动量', linewidth=2)
ax3.axhline(0, color='black', linestyle='--', alpha=0.5)
ax3.set_title(f'滚动夏普比率（{window}天窗口）', fontsize=12)
ax3.set_xlabel('交易日')
ax3.set_ylabel('夏普比率')
ax3.legend()
ax3.grid(True, alpha=0.3)
 
# 4. 动量因子收益率分布
ax4 = axes[1, 1]
# 计算每天的动量因子收益（多空组合收益）
momentum_factor_returns = pd.Series({
    '时间序列': ts_momentum_returns.mean(),
    '截面': cs_momentum_returns.mean()
})
# 计算标准差
momentum_factor_std = pd.Series({
    '时间序列': ts_momentum_returns.std(),
    '截面': cs_momentum_returns.std()
})
# 年化
momentum_annual = momentum_factor_returns * 252
momentum_std_annual = momentum_factor_std * np.sqrt(252)
 
x_pos = np.arange(len(momentum_annual))
bars = ax4.bar(x_pos, momentum_annual.values,
               yerr=momentum_std_annual.values,
               capsize=5, alpha=0.7, edgecolor='black')
colors = ['green' if x > 0 else 'red' for x in momentum_annual.values]
for bar, color in zip(bars, colors):
    bar.set_color(color)
ax4.set_xticks(x_pos)
ax4.set_xticklabels(momentum_annual.index)
ax4.set_ylabel('年化收益率')
ax4.set_title('动量因子年化收益（±1标准差）', fontsize=12)
ax4.axhline(0, color='black', linewidth=0.8)
ax4.grid(True, alpha=0.3, axis='y')
 
plt.tight_layout()
plt.show()

7.4 反转策略

def short_term_reversal(prices, lookback=5, holding=5):
    """
    短期反转策略
 
    逻辑: 过去下跌的股票倾向于短期反弹
    """
    returns = prices.pct_change()
    signals = pd.DataFrame(index=prices.index, columns=prices.columns)
 
    for i in range(lookback, len(prices)):
        # 过去 lookback 天的累计收益
        past_cumret = (prices.iloc[i-lookback:i] / prices.iloc[i-lookback] - 1).iloc[-1]
 
        # 负收益做多（抄底），正收益做空（止盈）
        signals.iloc[i] = -np.sign(past_cumret)
 
    # 计算策略收益
    strategy_returns = pd.Series(index=prices.index)
    for i in range(lookback, len(prices) - holding):
        if i % holding == 0:
            current_signal = signals.iloc[i]
            future_returns = returns.iloc[i:i+holding]
            strategy_returns.iloc[i] = (
                current_signal * future_returns
            ).mean().mean()
 
    return strategy_returns
 
# 运行反转策略
reversal_returns = short_term_reversal(prices)
reversal_cumulative = (1 + reversal_returns.fillna(0)).cumprod()
 
print("\n【短期反转策略】")
print("-" * 60)
reversal_metrics = calculate_performance_metrics(reversal_returns)
for metric, value in reversal_metrics.items():
    if '收益' in metric or '波动' in metric:
        print(f"{metric}: {value:.2%}")
    else:
        print(f"{metric}: {value:.2f}")

7.5 动量与反转的 ML 应用

# ML 模型预测时需要考虑:
# 1. 时间序列特征: 过去收益、波动率
# 2. 截面特征: 相对排名、行业相对表现
# 3. 交互特征: 市场状态（高波动时动量更强？）
 
def create_momentum_features(prices, lookback_periods=[5, 20, 60]):
    """
    构造动量相关特征供 ML 使用
    """
    returns = prices.pct_change()
    features = pd.DataFrame(index=prices.index)
 
    for period in lookback_periods:
        # 时间序列动量
        features[f'momentum_{period}'] = (
            prices.pct_change(period).shift(1)
        ).mean(axis=1)
 
        # 动量强度（过去收益的绝对值）
        features[f'momentum_strength_{period}'] = (
            returns.rolling(period).apply(lambda x: np.abs(x).mean())
        ).mean(axis=1).shift(1)
 
        # 动量一致性（正收益天数占比）
        features[f'momentum_consistency_{period}'] = (
            (returns.rolling(period).apply(lambda x: (x > 0).sum()) / period
        ).mean(axis=1).shift(1)
        )
 
    # 添加市场状态特征
    market_return = returns.mean(axis=1)
    features['market_volatility'] = market_return.rolling(20).std().shift(1)
    features['market_trend'] = market_return.rolling(60).mean().shift(1)
 
    return features
 
# 使用示例
momentum_features = create_momentum_features(prices)
print("\n【动量特征示例】")
print(momentum_features.dropna().head())

八、价值 vs 成长

8.1 定义与度量

因子	定义	常用指标
价值	相对内在价值被低估	市净率(P/B)、市盈率(P/E)、市销率(P/S)、EV/EBITDA
成长	盈利/营收高速增长	营收增长率、净利润增长率、ROE 变化

8.2 价值溢价之谜

现象: 价值股长期收益高于成长股，但并非持续。

解释:

风险补偿: 价值股往往财务状况较差
行为偏差: 投资者过度追捧成长股（漂亮股票）
周期性: 经济复苏期价值股表现更好

8.3 Python 实现

def value_factor_analysis(prices, fundamental_data):
    """
    价值因子分析
 
    参数:
        prices: 价格数据
        fundamental_data: 包含估值指标的基本面数据
    """
    # 计算估值因子
    # 市净率倒数 = 账面价值 / 市值
    pb_inverse = fundamental_data['book_value'] / prices
 
    # 市盈率倒数 = 每股收益 / 股价
    pe_inverse = fundamental_data['eps'] / prices
 
    # 综合价值得分（Z-score 标准化后平均）
    pb_z = (pb_inverse - pb_inverse.mean()) / pb_inverse.std()
    pe_z = (pe_inverse - pe_inverse.mean()) / pe_inverse.std()
 
    value_score = (pb_z + pe_z) / 2
 
    # 分组回测
    quintiles = pd.qcut(value_score, 5, labels=['低估值', '', '', '', '高估值'])
 
    # 计算每组收益
    group_returns = {}
    for group in ['低估值', '高估值']:
        mask = quintiles == group
        group_returns[group] = (
            prices.pct_change().mean(axis=1)[mask]
        ).mean()
 
    # 价值因子收益 = 高估值组 - 低估值组
    value_premium = group_returns['高估值'] - group_returns['低估值']
 
    return value_premium, group_returns

九、你用 ML 预测的到底是什么？

9.1 收益分解视角

股票收益 = Alpha + Beta_1 * 因子1 + Beta_2 * 因子2 + ... + 残差

ML 模型预测的可能是:
1. 真正的 Alpha（选股能力）
2. 某些已知因子的暴露（换一种方式计算 Beta）
3. 过拟合的噪声

9.2 区分 Alpha 与 Beta

def analyze_prediction_source(model_predictions, factor_returns, factor_loadings):
    """
    分析 ML 模型预测的收益来源
 
    参数:
        model_predictions: ML 模型预测的股票收益
        factor_returns: 因子收益率
        factor_loadings: 股票的因子暴露
    """
    # 将预测收益对因子暴露做回归
    # 预测收益 = α + Σ(β_i × f_i) + ε
 
    # 如果回归 R² 很高，说明模型主要是在预测 Beta
    # 如果 α 仍然显著，说明模型有真正的选股能力
 
    from sklearn.linear_model import LinearRegression
 
    X = factor_loadings.values
    y = model_predictions.values
 
    model = LinearRegression()
    model.fit(X, y)
 
    # 计算每个股票的预测收益中，多少来自因子，多少来自 Alpha
    factor_contribution = model.predict(X)
    alpha_contribution = y - factor_contribution
 
    # 方差分解
    var_total = np.var(y)
    var_factor = np.var(factor_contribution)
    var_alpha = np.var(alpha_contribution)
 
    return {
        'beta_explained_ratio': var_factor / var_total,
        'alpha_ratio': var_alpha / var_total,
        'model_r2': model.score(X, y)
    }

9.3 实践建议

回归测试: 对模型预测做因子回归，检查因子暴露
中性化: 如果模型暴露了 unwanted factor，进行因子中性化
理解模型: 分析模型学到了什么，而不是只看预测准确率

十、核心知识点总结

理论	核心观点	对量化 ML 的启示
EMH	价格反映信息，市场有效	寻找市场低效的角落，理解信息扩散机制
CAPM	收益 = 无风险 + β × 市场溢价	区分 Alpha 和 Beta，理解风险溢价
FF 三因子	市场风险 + 规模 + 价值	因子投资的鼻祖，风格暴露很重要
Barra	多因子风险模型	行业风险管理，组合风险控制
APT	收益来自多个风险因子	因子需要有经济学直觉，不只是统计

关键公式:

CAPM: E(R) = Rf + β × (Rm - Rf)
FF3:  E(R) = Rf + β×MKT + s×SMB + h×HML
Alpha: α = 实际收益 - 预期收益
IR:   信息比率 = α / σ(α)

实践要点:

因子拥挤会导致收益衰减，需要持续研究新因子
风格轮动要求动态调整因子暴露
ML 预测需要区分是 Alpha 还是 Beta
因子相关性高时需要正交化处理

下一步: 继续阅读 02-市场微观结构.md，理解真实市场的交易机制。

MindCarver Blog

MindCarver

探索

01-资产定价与因子模型