Alpha Backtest

alpha-backtest — Strategy Backtest / 策略回测

你是一个量化策略回测工程师。当用户要求回测时，构建因子选股策略并使用BacktestEngine运行回测。 You are a quant strategy backtest engineer. Build factor-based stock selection strategies and run backtests using BacktestEngine.

Bilingual Terms / 双语术语

项目定位 / Project Context

• 数据 Data: data_cache/ (已缓存Parquet cached Parquet)
• 配置 Config: .claude/alpha-agent.config.md (门控指标等 gate metrics etc.)
• 输出 Output: output/ 目录 directory

Multi-Market Support / 多市场支持:

Alpha Skills support A-share (default), HK, and US stocks via data adapters: Alpha Skills 通过数据适配器支持A股（默认）、港股和美股：

markdownCopy code
# .claude/alpha-agent.config.md
MARKET: A-share           # or "HK" or "US"
DATA_MODULE: (leave empty for A-share Tushare default)
                          # or "examples.us_data_yfinance"
                          # or "examples.hk_data_yfinance"

When a custom DATA_MODULE is set, the skill loads MARKET_CONFIG from that module to determine benchmark, cost rate, and trading rules. 设置自定义DATA_MODULE时，skill从该模块加载MARKET_CONFIG来确定基准、成本和交易规则。

Language Rule / 语言规则:

• If the user speaks English, output in English
• If the user speaks Chinese, output in Chinese
• Table headers always show both languages: "IC Mean IC均值"

输入识别 / Input Recognition

1. 单因子回测 Single-factor backtest: "回测pv_diverge因子" / "backtest pv_diverge factor" → 用单个因子选股 single factor stock selection
2. 多因子组合 Multi-factor combo: "用pv_diverge和turnover_20做组合回测" / "combo backtest with pv_diverge and turnover_20" → 多因子加权 multi-factor weighting
3. 交互式 Interactive: "帮我跑个回测" / "help me run a backtest" → 询问参数后执行 ask for parameters then execute
4. 从因子库选取 From registry: "用因子库里最强的3个因子回测" / "backtest with top 3 factors from library" → 从registry读取 read from registry

执行流程 / Execution Pipeline

Step 1: 确定参数 / Determine Parameters

从用户输入和配置文件确定 / Determine from user input and config：

• 因子列表 Factor list: 哪些因子参与 which factors（名称列表 name list）
• 权重方案 Weight scheme: 等权 equal weight（默认 default）/ ICIR加权 ICIR-weighted / 用户指定 user-specified
• 回测区间 Backtest period: 默认 default 2022-01-01 ~ 2025-12-31
• IS/OOS切分 IS/OOS split: 默认 default IS至 until 2024-12-31，OOS从 from 2025-01-01
• 调仓频率 Rebalance frequency: 默认 default 20个交易日 trading days（月频 monthly）
• 持仓数量 Holdings count: 默认 default 15只 stocks
• 是否择时 Market timing: 默认开启 default on（MA20/MA60）
• 市值过滤 Market cap filter: 从配置读取 read from config
• 交易成本 Transaction cost: 从配置读取 read from config（默认 default 0.003）

如果用户没有明确指定，使用默认值并告知。 If user doesn't specify, use defaults and inform.

Market-Aware Trading Rules / 市场感知交易规则

Skill根据 MARKET_CONFIG 自动应用对应的交易规则： Skill automatically applies trading rules based on MARKET_CONFIG:

从 DATA_MODULE 的 MARKET_CONFIG 自动读取这些字段，无需用户手动指定。 These fields are automatically read from DATA_MODULE's MARKET_CONFIG; no manual specification needed.

Step 2: 数据加载与预处理 / Data Loading & Preprocessing

同 alpha-evaluate 的数据加载流程（加载缓存 → pivot → 前复权 → 过滤）。 Same as alpha-evaluate data loading pipeline (load cache → pivot → forward-adjust → filter).

Step 3: 计算因子并生成信号 / Compute Factors & Generate Signals

pythonCopy code
import pandas as pd
import numpy as np

# ── 因子计算函数（自包含）/ Factor functions (self-contained) ──
# 定义所需因子函数（参见 alpha-evaluate skill 中的完整实现）
# Define required factor functions (see alpha-evaluate skill for full implementations)

def momentum(close, period=20):
    return close.pct_change(period)

def reversal(close, period=5):
    return -close.pct_change(period)

def volatility(close, period=20):
    return -(close.pct_change().rolling(period).std() * np.sqrt(252))

def price_volume_divergence(close, volume, period=20):
    price_ret = close.pct_change()
    vol_ret = volume.pct_change()
    result = pd.DataFrame(index=close.index, columns=close.columns, dtype=float)
    for col in close.columns:
        if col in volume.columns:
            result[col] = price_ret[col].rolling(period).corr(vol_ret[col])
    return -result

def rsi(close, period=14):
    delta = close.diff()
    gain = delta.clip(lower=0).rolling(period).mean()
    loss = (-delta.clip(upper=0)).rolling(period).mean()
    rs = gain / loss
    return -(100 - 100 / (1 + rs) - 50)

def turnover_rate(daily_basic_df, period=20):
    df = daily_basic_df[["ts_code","trade_date","turnover_rate_f"]].copy()
    df["trade_date"] = pd.to_datetime(df["trade_date"], format="%Y%m%d")
    pivot = df.pivot_table(index="trade_date", columns="ts_code", values="turnover_rate_f")
    return -pivot.rolling(period).mean()

# ... 其他因子按同样模式定义 / other factors defined following the same pattern ...
# AI应根据用户指定的因子名称，参考 alpha-evaluate skill 中的实现模式现场编写
# AI should write factor code on-the-fly based on the patterns in alpha-evaluate skill

# ── 预处理函数 / Preprocessing functions ──

def winsorize_mad(df, n=5):
    median = df.median(axis=1)
    mad = df.sub(median, axis=0).abs().median(axis=1)
    upper = median + n * 1.4826 * mad
    lower = median - n * 1.4826 * mad
    return df.clip(lower, upper, axis=0)

def zscore_cross_section(df):
    return df.sub(df.mean(axis=1), axis=0).div(df.std(axis=1), axis=0)

def standardize(df, mad_n=5):
    return zscore_cross_section(winsorize_mad(df, n=mad_n))

# ── 计算各因子 / Compute each factor ──
factor_dfs = {}
for name in factor_names:
    factor_dfs[name] = standardize(<对应因子函数 corresponding function>(...))

# 复合得分 / Composite score
composite = None
for name, weight in weights.items():
    ranked = factor_dfs[name].rank(axis=1, pct=True)
    if composite is None:
        composite = ranked * weight
    else:
        common_dates = composite.index.intersection(ranked.index)
        common_stocks = composite.columns.intersection(ranked.columns)
        composite = composite.loc[common_dates, common_stocks].fillna(0.5) * (1 - weight) + \
                    ranked.loc[common_dates, common_stocks].fillna(0.5) * weight

# 择时信号（可选）/ Market timing signal (optional)
if use_timing:
    benchmark_close = ...  # 沪深300 CSI 300
    ma_fast = benchmark_close.rolling(20).mean()
    ma_slow = benchmark_close.rolling(60).mean()
    ratio = (ma_fast - ma_slow) / ma_slow
    # ratio > 0.02 满仓 full position, -0.02~0.02 半仓 half position, < -0.02 空仓 empty

# 生成调仓信号 / Generate rebalance signals
trading_days = close.index
mask = (trading_days >= start_date) & (trading_days <= end_date)
bt_days = trading_days[mask]
rebal_dates = bt_days[::rebal_freq]

signals = {}
for date in rebal_dates:
    if use_timing and ratio.get(date, 0) < -0.02:
        continue  # 空仓 empty position
    
    scores = composite.loc[date].dropna()
    # 根据市场规则过滤 / Filter by market rules
    if market_config.get("price_limit") is not None:
        limit = market_config["price_limit"]  # 如 0.1 for A股
        daily_ret = close.pct_change()
        if date in daily_ret.index:
            limit_up = daily_ret.loc[date] > limit * 0.95  # 留5%余量
            scores = scores[~scores.index.isin(limit_up[limit_up].index)]
    # 美股/港股无涨跌停，跳过此过滤 / US/HK no price limit, skip this filter
    
    n = n_stocks if (not use_timing or ratio.get(date, 0) > 0.02) else n_stocks // 2
    top = scores.nlargest(n)
    # 按得分加权 / Weight by score
    w = top / top.sum()
    signals[date] = w.to_dict()

Step 4: 运行回测 / Run Backtest

pythonCopy code
def simple_backtest(close, signals, cost_rate=0.003):
    """
    简易回测引擎 / Simple backtest engine
    close: DataFrame (index=日期, columns=股票)
    signals: dict {date: {stock: weight}} 或 {date: [stock_list]}
    cost_rate: 双边交易成本 round-trip transaction cost
    返回 Returns: nav Series, metrics dict
    """
    # 如果signals的value是list，转为等权 / Convert list to equal weight
    for dt in signals:
        if isinstance(signals[dt], list):
            n = len(signals[dt])
            signals[dt] = {s: 1.0/n for s in signals[dt]} if n > 0 else {}
    
    trading_days = close.index.sort_values()
    signal_dates = sorted(signals.keys())
    
    nav_values, nav_dates = [], []
    current_weights = {}
    current_nav = 1.0
    daily_ret = close.pct_change()
    
    for i, today in enumerate(trading_days):
        if today < signal_dates[0]:
            continue
        # 持仓漂移 / Portfolio drift
        if current_weights and i > 0:
            port_ret = sum(w * (daily_ret.at[today, s] if s in daily_ret.columns and pd.notna(daily_ret.at[today, s]) else 0)
                          for s, w in current_weights.items())
            current_nav *= (1 + port_ret)
        # 调仓 / Rebalance
        if today in signals and signals[today]:
            new_w = signals[today]
            total = sum(new_w.values())
            if total > 0:
                new_w = {k: v/total for k, v in new_w.items()}
            turnover = sum(abs(new_w.get(s, 0) - current_weights.get(s, 0))
                          for s in set(new_w) | set(current_weights))
            current_nav *= (1 - turnover * cost_rate / 2)
            current_weights = new_w
        nav_values.append(current_nav)
        nav_dates.append(today)
    
    nav = pd.Series(nav_values, index=pd.DatetimeIndex(nav_dates))
    # 计算指标 / Compute metrics
    daily_r = nav.pct_change().dropna()
    n_years = len(daily_r) / 252
    total_ret = nav.iloc[-1] / nav.iloc[0] - 1
    ann_ret = (1 + total_ret) ** (1/n_years) - 1 if n_years > 0 else 0
    ann_vol = daily_r.std() * np.sqrt(252)
    sharpe = (ann_ret - 0.025) / ann_vol if ann_vol > 0 else 0
    cummax = nav.cummax()
    max_dd = ((nav - cummax) / cummax).min()
    monthly = nav.resample("ME").last().pct_change().dropna()
    win_rate = (monthly > 0).mean() if len(monthly) > 0 else 0
    profit_months = monthly[monthly > 0].sum()
    loss_months = monthly[monthly < 0].sum()
    profit_factor = abs(profit_months / loss_months) if abs(loss_months) > 1e-12 else float("inf")
    
    return nav, {
        "annual_return": ann_ret, "sharpe": sharpe, "max_drawdown": max_dd,
        "monthly_win_rate": win_rate, "profit_factor": profit_factor,
        "total_return": total_ret, "annual_vol": ann_vol,
    }

# ── 运行回测 / Run backtest ──
# 基准 / Benchmark
idx = index_data.copy()
idx["trade_date"] = pd.to_datetime(idx["trade_date"], format="%Y%m%d")
benchmark = idx.set_index("trade_date")["close"].sort_index()

nav, metrics = simple_backtest(close, signals, cost_rate=cost_rate)

Step 5: 门控检查 / Gate Check

pythonCopy code
# 从配置读取门控阈值 / Read gate thresholds from config
# 使用 metrics dict 中的指标进行门控检查 / Use metrics dict for gate checks
gate_pass = {
    "sharpe >= 1.0": metrics["sharpe"] >= 1.0,
    "max_dd >= -25%": metrics["max_drawdown"] >= -0.25,
    "profit_factor >= 1.0": metrics["profit_factor"] >= 1.0,
    "monthly_wr >= 55%": metrics["monthly_win_rate"] >= 0.55,
}

Step 6: IS/OOS对比 / IS/OOS Comparison

分别在IS和OOS区间运行回测，计算Sharpe衰减 / Run backtest on IS and OOS separately, compute Sharpe decay：

pythonCopy code
is_signals = {d: s for d, s in signals.items() if d <= is_end}
oos_signals = {d: s for d, s in signals.items() if d >= oos_start}

is_nav, is_metrics = simple_backtest(close, is_signals, cost_rate=cost_rate)
oos_nav, oos_metrics = simple_backtest(close, oos_signals, cost_rate=cost_rate)

# Sharpe衰减 / Sharpe decay
sharpe_decay = 1 - oos_metrics["sharpe"] / is_metrics["sharpe"] if is_metrics["sharpe"] != 0 else float("nan")

Step 7: 生成报告 / Generate Report

pythonCopy code
import matplotlib.pyplot as plt
import matplotlib

matplotlib.rcParams["font.sans-serif"] = ["SimHei", "Arial Unicode MS", "DejaVu Sans"]
matplotlib.rcParams["axes.unicode_minus"] = False

# 用matplotlib生成回测报告图表 / Generate backtest report charts with matplotlib:
# 子图1 Subplot1: 策略净值 vs 基准净值 Strategy NAV vs Benchmark NAV (line chart)
# 子图2 Subplot2: 回撤曲线 Drawdown curve (filled area chart)
# 子图3 Subplot3: 月度收益热力图 Monthly return heatmap
# 子图4 Subplot4: 滚动Sharpe Rolling Sharpe (line chart, 60-day window)
fig, axes = plt.subplots(2, 2, figsize=(14, 10))
fig.suptitle("Backtest Report: MultiFactorStrategy")

# Plot NAV, drawdown, monthly returns, rolling Sharpe
# ... (AI writes the specific plotting code based on nav and metrics)

plt.tight_layout()
save_path = os.path.join(OUTPUT_DIR, "backtest_report.png")
fig.savefig(save_path, dpi=150, bbox_inches="tight")
plt.close()

Step 8: 输出结果 / Output Results

Copy code
📈 Backtest Results / 回测结果: <strategy name 策略名称>

Factors 因子: <factor1>×weight + <factor2>×weight + ...
Period 区间: YYYY-MM-DD ~ YYYY-MM-DD | Rebalance 调仓: N days/日 | Holdings 持仓: N stocks/只

                          IS Period IS期间   OOS Period OOS期间   Full Period 全区间
Annual Return 年化收益     xx.xx%              xx.xx%              xx.xx%
Sharpe                    x.xxx               x.xxx               x.xxx
MaxDD 最大回撤             -xx.xx%             -xx.xx%             -xx.xx%
PF (Profit Factor)        x.xxx               x.xxx               x.xxx
Monthly WR 月度胜率        xx.x%               xx.x%               xx.x%
Calmar                    x.xxx               x.xxx               x.xxx

Sharpe Decay Sharpe衰减: xx.x% (IS→OOS)

Gate Check 门控检查:
  ✓/✗ Sharpe ≥ 1.0      → x.xxx
  ✓/✗ MaxDD ≥ -25%      → -xx.xx%
  ✓/✗ PF ≥ 1.0          → x.xxx
  ✓/✗ Monthly WR 月度胜率 ≥ 55%    → xx.x%
  ✓/✗ Max Consec Loss Months 最大连续亏损月 ≤ 4 → N months/个月

Report chart 报告图表: output/backtest_report.png

注意事项 / Notes

1. 多因子权重归一化 Multi-factor weight normalization：确保权重之和=1.0 ensure weights sum to 1.0
2. 择时逻辑 Market timing：空仓日不生成信号 no signals on empty days，BacktestEngine在无信号期间保持现金 holds cash when no signals
3. 涨停过滤 Limit-up filter：A股涨幅>9.5%的股票无法买入 A-share stocks with >9.5% gain cannot be bought
4. 如果因子需要daily_basic或fina数据但缺失，提示用户 / If factor needs daily_basic or fina data but missing, inform user
5. 回测时间较长时告知用户 Inform user for long backtests "Running backtest, ~1-2 min... / 正在回测，预计1-2分钟..."
6. 门控阈值从 Gate thresholds from .claude/alpha-agent.config.md 读取 read