quanxiel/alpha/阿尔法策略研究示例.ipynb

阿尔法策略回测 — 完整示例

本 Notebook 演示如何使用 quanxiel.alpha 模块进行：

1. 因子计算与 IC 分析
2. 策略信号生成
3. 事件驱动回测
4. 绩效评估

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 FactorRegistry, FactorAnalyzer
from alpha.strategy import Strategy, QuantileSignal, EqualWeightAllocator
from alpha.backtest import BacktestEngine
from alpha.evaluation import PerformanceEvaluator, ReportGenerator

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

1. 准备数据

从本地数据库或 Tushare 获取量价数据与财务数据。这里用随机数据演示流程。

In [ ]:

# ---- 生成示例数据 -------
np.random.seed(42)
dates = pd.date_range('2022-01-01', '2024-12-31', freq='B')
stocks = [f'{i:06d}.SH' for i in range(600000, 600050)]

# 价格数据
price_data = pd.DataFrame(
    np.cumprod(1 + np.random.randn(len(dates), len(stocks)) * 0.02, axis=0),
    index=dates, columns=stocks
)

# 因子数据（模拟技术面因子值）
factor_df = pd.DataFrame(
    np.random.randn(len(dates), len(stocks)),
    index=dates, columns=stocks
)

print(f'价格数据: {price_data.shape}')
print(f'因子数据: {factor_df.shape}')

2. 因子 IC 分析

In [ ]:

# 计算未来 1 期收益率
fwd_returns = price_data.pct_change().shift(-1).stack()
factor_stacked = factor_df.stack()

analyzer = FactorAnalyzer(factor_stacked, fwd_returns)

# Rank IC
ic = analyzer.compute_ic(method='rank')
print('IC 汇总:')
for k, v in analyzer.ic_summary().items():
    print(f'  {k}: {v:.4f}')

# IC 曲线
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('各分位组平均收益:')
print(quantile_ret)

4. 构建策略 & 回测

In [ ]:

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

# 策略组合
strategy = Strategy(
    name='技术因子-分位数策略',
    factors=[],
    signal_generator=QuantileSignal(n_quantiles=5, long_quantile=5, short_quantile=1),
    weight_allocator=EqualWeightAllocator(max_positions=20),
    description='买入因子值最高的分位组，等权持仓'
)

# 回测引擎
engine = BacktestEngine(config)
equity_curve = engine.run(
    strategy=strategy,
    price_data=price_data,
    factor_data={'technical': factor_df},
    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='steelblue')
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, color='red', alpha=0.3)
axes[1].set_title('回撤曲线')
axes[1].grid(True, alpha=0.3)

plt.tight_layout()
plt.show()

6. 绩效评估

In [ ]:

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

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

7. 交易记录分析

In [ ]:

trades = engine.portfolio.trades
if trades:
    trade_df = pd.DataFrame([
        {'日期': t.date, '股票': t.stock, '方向': t.side,
         '数量': t.quantity, '价格': t.price, '佣金': t.commission}
        for t in trades
    ])
    print(f'总交易数: {len(trade_df)}')
    print(f'买入: {(trade_df["方向"]=="buy").sum()}, 卖出: {(trade_df["方向"]=="sell").sum()}')
    display(trade_df.head(20))

---

使用真实数据

将以上模拟数据替换为从 quanxiel.quantitative_data 模块加载的真实行情：

from quantitative_data.importer import DataImporter

importer = DataImporter()
price_data = importer.load_daily_prices('2020-01-01', '2024-12-31')