quanxiel/alpha/价值投资策略.ipynb

价值投资策略 — 构建与回测

策略思路

价值投资（Value Investing）：买入相对于基本面被低估的股票。 本策略使用 市盈率 PE 倒数（EP） 与 市净率 PB 倒数（BP） 合成价值因子， 每月调仓，持有估值最低（价值因子最高）的一批股票。

研究流程

1. 加载数据（价格 + PE/PB 估值指标）
2. 价值因子计算与 IC 分析
3. 分层回测验证因子有效性
4. 构建策略并回测
5. 绩效评估

In [ ]:

import sys
sys.path.insert(0, '..')

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt

from alpha.config import AlphaConfig
from alpha.factors import FactorAnalyzer
from alpha.strategy import Strategy, QuantileSignal, EqualWeightAllocator
from alpha.backtest import BacktestEngine
from alpha.evaluation import PerformanceEvaluator
from alpha.data_loader import DataLoader

%matplotlib inline
plt.rcParams['font.sans-serif'] = ['SimHei']
plt.rcParams['axes.unicode_minus'] = False

print('模块导入成功')

1. 加载数据

价值因子需要估值数据（PE / PB），通过 DataLoader.load_factor 加载。

In [ ]:

START_DATE = '2020-01-01'
END_DATE   = '2024-12-31'

loader = DataLoader()

# 价格数据 + 估值数据
price_data = loader.load_prices(START_DATE, END_DATE)
pe_data  = loader.load_factor('pe_ttm', START_DATE, END_DATE)   # 市盈率 TTM
pb_data  = loader.load_factor('pb', START_DATE, END_DATE)       # 市净率

print(f'价格数据: {price_data.shape}')
print(f'PE(TTM)数据: {pe_data.shape}')
print(f'PB数据: {pb_data.shape}')

# 展示某天的估值截面
sample = pd.DataFrame({
    'PE_TTM': pe_data.iloc[-1],
    'PB': pb_data.iloc[-1],
}).dropna()
sample.head(10)

2. 价值因子计算与 IC 分析

价值因子 = 标准化后的 (1/PE + 1/PB)：

Value = Z\left(\frac{1}{PE}\right) + Z\left(\frac{1}{PB}\right)

PE/PB 越低 → 价值因子越高 → 越被低估。

In [ ]:

# 剔除 PE/PB 非正值（亏损或净资产为负）
ep = 1.0 / pe_data.where(pe_data > 0)
bp = 1.0 / pb_data.where(pb_data > 0)

# 截面标准化后合成
def zscore_cross_section(df):
    return (df - df.mean(axis=1)) / (df.std(axis=1) + 1e-12)

value_factor = zscore_cross_section(ep) + zscore_cross_section(bp)

# 未来 20 日收益率（价值因子预测周期较长）
fwd_20d = price_data.pct_change(20).shift(-20)

factor_stacked = value_factor.stack()
fwd_stacked = fwd_20d.stack()

analyzer = FactorAnalyzer(factor_stacked, fwd_stacked)
ic = analyzer.compute_ic(method='rank')

print('价值因子 IC 汇总:')
for k, v in analyzer.ic_summary().items():
    print(f'  {k}: {v:.4f}')

ic.plot(figsize=(12, 3), title='价值因子 Rank IC 时间序列')
plt.axhline(y=0, color='r', linestyle='--')
plt.show()

3. 分层回测

In [ ]:

quantile_ret = analyzer.quantile_returns(n_quantiles=5)
print('各分位组平均收益 (Q1=最贵, Q5=最便宜):')
print(quantile_ret)

quantile_ret['avg_return'].plot(kind='bar', figsize=(8, 4), color='steelblue')
plt.title('价值因子分层收益 (未来20日)')
plt.ylabel('平均收益率')
plt.grid(True, alpha=0.3)
plt.show()

4. 构建策略并回测

In [ ]:

# 回测配置
config = AlphaConfig(
    initial_cash=1_000_000,
    commission_rate=0.0003,
    slippage=0.001,
    stamp_tax=0.001,
)

# 价值策略：持有价值因子最高的 20% 股票（最被低估）
strategy = Strategy(
    name='价值投资策略-EP+BP',
    factors=[],
    signal_generator=QuantileSignal(
        n_quantiles=5, long_quantile=5, short_quantile=0,
    ),
    weight_allocator=EqualWeightAllocator(max_positions=10),
    description='每月持有EP+BP合成的价值因子最高的20%股票'
)

engine = BacktestEngine(config)
equity_curve = engine.run(
    strategy=strategy,
    price_data=price_data,
    factor_data={'value': value_factor},
    rebalance_freq='M',
)

print(f'回测完成, 共 {len(equity_curve)} 个交易日')
print(f'累计收益率: {(equity_curve["nav"].iloc[-1] - 1) * 100:.2f}%')

5. 净值曲线与回撤

In [ ]:

fig, axes = plt.subplots(2, 1, figsize=(14, 8))

# 净值曲线
axes[0].plot(equity_curve.index, equity_curve['nav'], label='策略净值', color='purple')
axes[0].axhline(y=1.0, color='gray', linestyle='--')
axes[0].set_title('价值投资策略净值曲线')
axes[0].legend()
axes[0].grid(True, alpha=0.3)

# 回撤曲线
nav = equity_curve['nav']
running_max = nav.cummax()
drawdown = (nav - running_max) / running_max
axes[1].fill_between(equity_curve.index, 0, drawdown.values, color='red', alpha=0.3)
axes[1].set_title('回撤曲线')
axes[1].grid(True, alpha=0.3)

plt.tight_layout()
plt.show()

6. 绩效评估

In [ ]:

# 基准：股票池等权组合
bench_ret = price_data.pct_change().mean(axis=1)

evaluator = PerformanceEvaluator(equity_curve, benchmark_returns=bench_ret, risk_free_rate=0.03)
print(evaluator.summary())

report_df = evaluator.full_report()
pd.DataFrame(list(report_df.items()), columns=['指标', '数值']).set_index('指标')

---

单独估值指标对比

对比单独使用 EP、BP 与合成价值因子的效果：

In [ ]:

# 单独因子的 IC 对比
for name, fac in [('EP', ep), ('BP', bp), ('EP+BP', value_factor)]:
    a = FactorAnalyzer(fac.stack(), fwd_stacked)
    s = a.ic_summary()
    print(f'{name}: IC均值={s["IC_Mean"]:.4f}, ICIR={s["IR"]:.4f}, '
          f'IC>0占比={s["IC>0_Ratio"]:.2%}')