9.2 KiB
9.2 KiB
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('模块导入成功')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)
roe_data = loader.load_factor('roe', START_DATE, END_DATE)
print(f'价格数据: {price_data.shape}')
print(f'PE(TTM)数据: {pe_data.shape}')
print(f'ROE数据: {roe_data.shape}')In [ ]:
# ---- 截面标准化函数 ----
def zscore_cross_section(df):
return (df - df.mean(axis=1)) / (df.std(axis=1) + 1e-12)
# ---- 1. 动量因子 (20日收益率) ----
momentum = price_data.pct_change(20)
z_mom = zscore_cross_section(momentum)
# ---- 2. 低波因子 (20日波动率倒数) ----
daily_ret = price_data.pct_change()
volatility = daily_ret.rolling(20).std()
z_lowvol = -zscore_cross_section(volatility) # 波动越小越好
# ---- 3. 价值因子 (EP = 1/PE) ----
ep = 1.0 / pe_data.where(pe_data > 0)
z_value = zscore_cross_section(ep)
# ---- 4. 质量因子 (ROE) ----
z_quality = zscore_cross_section(roe_data)
# ---- 汇总各因子 ----
factors = {
'动量': momentum,
'低波': -volatility,
'价值': ep,
'质量': roe_data,
}
# 未来 10 日收益率
fwd_10d = price_data.pct_change(10).shift(-10)
# 各因子 IC 对比
print('各因子 IC 汇总 (未来10日):')
ic_results = {}
for name, fac in factors.items():
a = FactorAnalyzer(fac.stack(), fwd_10d.stack())
s = a.ic_summary()
ic_results[name] = s
print(f' {name}: IC均值={s["IC_Mean"]:.4f}, ICIR={s["IR"]:.4f}, '
f'IC>0占比={s["IC>0_Ratio"]:.2%}')In [ ]:
# 等权合成综合得分
multi_factor = (z_mom + z_lowvol + z_value + z_quality) / 4
# 分层回测
analyzer = FactorAnalyzer(multi_factor.stack(), fwd_10d.stack())
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('多因子合成分层收益 (未来10日)')
plt.ylabel('平均收益率')
plt.grid(True, alpha=0.3)
plt.show()In [ ]:
# 回测配置
config = AlphaConfig(
initial_cash=1_000_000,
commission_rate=0.0003,
slippage=0.001,
stamp_tax=0.001,
)
# 多因子策略:持有综合得分最高的 20% 股票
strategy = Strategy(
name='多因子选股策略-动量+低波+价值+质量',
factors=[],
signal_generator=QuantileSignal(
n_quantiles=5, long_quantile=5, short_quantile=0,
),
weight_allocator=EqualWeightAllocator(max_positions=10),
description='每月持有动量/低波/价值/质量四因子综合得分最高的20%股票'
)
engine = BacktestEngine(config)
equity_curve = engine.run(
strategy=strategy,
price_data=price_data,
factor_data={'multi': multi_factor},
rebalance_freq='M',
)
print(f'回测完成, 共 {len(equity_curve)} 个交易日')
print(f'累计收益率: {(equity_curve["nav"].iloc[-1] - 1) * 100:.2f}%')In [ ]:
fig, axes = plt.subplots(2, 1, figsize=(14, 8))
# 净值曲线
axes[0].plot(equity_curve.index, equity_curve['nav'], label='策略净值', color='darkred')
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()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('指标')In [ ]:
# 单因子 vs 多因子 IC 对比
all_factors = {
'动量': z_mom,
'低波': z_lowvol,
'价值': z_value,
'质量': z_quality,
'多因子(等权)': multi_factor,
}
print(f"{'因子':<12} {'IC均值':>8} {'ICIR':>8} {'IC>0占比':>10}")
print('-' * 42)
for name, fac in all_factors.items():
a = FactorAnalyzer(fac.stack(), fwd_10d.stack())
s = a.ic_summary()
print(f"{name:<12} {s['IC_Mean']:>8.4f} {s['IR']:>8.4f} {s['IC>0_Ratio']:>10.2%}")