cyb50-quant/research/chinext50_regime_project/backtest/engine.py

from __future__ import annotations

from typing import Any

import numpy as np
import pandas as pd


def _max_drawdown(equity: pd.Series) -> float:
    peak = equity.cummax()
    drawdown = equity / peak - 1.0
    return float(drawdown.min())


def compute_metrics(
    strategy_returns: pd.Series,
    benchmark_returns: pd.Series,
    turnover: pd.Series | None = None,
    tracking_difference: pd.Series | None = None,
    annualization: int = 252,
) -> dict[str, float]:
    strategy_returns = strategy_returns.dropna()
    benchmark_returns = benchmark_returns.reindex(strategy_returns.index).fillna(0.0)
    turnover = turnover.reindex(strategy_returns.index).fillna(0.0) if turnover is not None else pd.Series(0.0, index=strategy_returns.index)
    tracking_difference = (
        tracking_difference.reindex(strategy_returns.index).fillna(0.0)
        if tracking_difference is not None
        else strategy_returns - benchmark_returns
    )

    if strategy_returns.empty:
        return {
            'annual_return': 0.0,
            'annual_vol': 0.0,
            'sharpe': 0.0,
            'max_drawdown': 0.0,
            'calmar': 0.0,
            'benchmark_sharpe': 0.0,
            'sharpe_delta': 0.0,
            'benchmark_max_drawdown': 0.0,
            'drawdown_improvement_ratio': 0.0,
            'upside_capture': 0.0,
            'downside_capture': 0.0,
            'annual_turnover': 0.0,
            'tracking_diff_mean': 0.0,
            'tracking_diff_abs_mean': 0.0,
            'tracking_error_20_p95': 0.0,
        }

    def annual_return(returns: pd.Series) -> float:
        total = float((1.0 + returns).prod())
        n = len(returns)
        return total ** (annualization / max(n, 1)) - 1.0

    def annual_vol(returns: pd.Series) -> float:
        return float(returns.std(ddof=0) * np.sqrt(annualization))

    strategy_ann = annual_return(strategy_returns)
    strategy_vol = annual_vol(strategy_returns)
    strategy_sharpe = strategy_ann / strategy_vol if strategy_vol > 0 else 0.0
    strategy_equity = (1.0 + strategy_returns).cumprod()
    strategy_mdd = abs(_max_drawdown(strategy_equity))
    strategy_calmar = strategy_ann / strategy_mdd if strategy_mdd > 0 else 0.0

    bench_ann = annual_return(benchmark_returns)
    bench_vol = annual_vol(benchmark_returns)
    bench_sharpe = bench_ann / bench_vol if bench_vol > 0 else 0.0
    bench_equity = (1.0 + benchmark_returns).cumprod()
    bench_mdd = abs(_max_drawdown(bench_equity))

    up_mask = benchmark_returns > 0
    down_mask = benchmark_returns < 0
    upside_capture = (strategy_returns[up_mask].mean() / benchmark_returns[up_mask].mean()) if up_mask.any() else 0.0
    downside_capture = (strategy_returns[down_mask].mean() / benchmark_returns[down_mask].mean()) if down_mask.any() else 0.0
    drawdown_improvement = (bench_mdd - strategy_mdd) / bench_mdd if bench_mdd > 0 else 0.0
    annual_turnover = float(turnover.mean() * annualization)
    tracking_diff_mean = float(tracking_difference.mean())
    tracking_diff_abs_mean = float(tracking_difference.abs().mean())
    tracking_error_20 = tracking_difference.rolling(20).std().dropna()
    tracking_error_20_p95 = float(tracking_error_20.quantile(0.95)) if not tracking_error_20.empty else 0.0

    return {
        'annual_return': float(strategy_ann),
        'annual_vol': float(strategy_vol),
        'sharpe': float(strategy_sharpe),
        'max_drawdown': float(strategy_mdd),
        'calmar': float(strategy_calmar),
        'benchmark_return': float(bench_ann),
        'benchmark_vol': float(bench_vol),
        'benchmark_sharpe': float(bench_sharpe),
        'benchmark_max_drawdown': float(bench_mdd),
        'sharpe_delta': float(strategy_sharpe - bench_sharpe),
        'drawdown_improvement_ratio': float(drawdown_improvement),
        'upside_capture': float(upside_capture),
        'downside_capture': float(downside_capture),
        'annual_turnover': annual_turnover,
        'tracking_diff_mean': tracking_diff_mean,
        'tracking_diff_abs_mean': tracking_diff_abs_mean,
        'tracking_error_20_p95': tracking_error_20_p95,
    }


def run_backtest(df: pd.DataFrame, config: dict[str, Any] | None = None) -> tuple[pd.DataFrame, dict[str, float]]:
    out = df.copy()
    trading_cfg = (config or {}).get('trading', {})
    annualization = int(trading_cfg.get('annualization', 252))

    if 'target_exposure' not in out.columns:
        raise ValueError('target_exposure column is required for backtest.')

    if 'open' in out.columns:
        asset_exec_return = out['open'].shift(-1) / out['open'] - 1.0
    else:
        asset_exec_return = out['close'].pct_change().shift(-1)

    executed_exposure = out['target_exposure'].shift(1).fillna(0.0)
    previous_executed = executed_exposure.shift(1).fillna(0.0)
    turnover = (executed_exposure - previous_executed).abs()

    one_way_cost_bps = float(trading_cfg.get('fee_bps_roundtrip', 8)) / 2.0 + float(trading_cfg.get('slippage_bps_oneway', 4))
    cost_rate = one_way_cost_bps / 10000.0
    extreme_move_threshold = float(trading_cfg.get('extreme_day_move_threshold', 0.03))
    extreme_day_cost_multiplier = float(trading_cfg.get('extreme_day_cost_multiplier', 1.0))
    gap_slippage_factor = float(trading_cfg.get('gap_slippage_factor', 0.0))

    extreme_day_flag = asset_exec_return.abs() >= extreme_move_threshold
    effective_multiplier = pd.Series(1.0, index=out.index)
    effective_multiplier.loc[extreme_day_flag.fillna(False)] = extreme_day_cost_multiplier

    trading_cost_base = turnover * cost_rate * effective_multiplier
    if 'gap_open' in out.columns:
        gap_open = out['gap_open'].fillna(0.0)
    else:
        gap_open = (out['open'] / out['close'].shift(1) - 1.0).fillna(0.0) if {'open', 'close'}.issubset(out.columns) else pd.Series(0.0, index=out.index)
    gap_shock_cost = turnover * gap_open.abs() * gap_slippage_factor
    trading_cost = trading_cost_base + gap_shock_cost

    out['asset_exec_return'] = asset_exec_return
    out['executed_exposure'] = executed_exposure
    out['turnover'] = turnover
    out['extreme_day_flag'] = extreme_day_flag.fillna(False)
    out['execution_cost_multiplier'] = effective_multiplier
    out['trading_cost_base'] = trading_cost_base
    out['gap_shock_cost'] = gap_shock_cost
    out['trading_cost'] = trading_cost
    out['strategy_return_gross'] = executed_exposure * asset_exec_return
    out['strategy_return_net'] = out['strategy_return_gross'] - trading_cost
    out['tracking_difference'] = out['strategy_return_net'] - out['strategy_return_gross']
    out['tracking_error_20'] = out['tracking_difference'].rolling(20).std()
    out['strategy_equity'] = (1.0 + out['strategy_return_net'].fillna(0.0)).cumprod()
    out['benchmark_equity'] = (1.0 + out['asset_exec_return'].fillna(0.0)).cumprod()

    metrics = compute_metrics(
        strategy_returns=out['strategy_return_net'],
        benchmark_returns=out['asset_exec_return'],
        turnover=out['turnover'],
        tracking_difference=out['tracking_difference'],
        annualization=annualization,
    )
    return out, metrics