regime-aware-multi-factor/Backtest_utils.py at main · TorresYue/regime-aware-multi-factor · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
from __future__ import annotations

from dataclasses import dataclass

import numpy as np
import pandas as pd


@dataclass(frozen=True)
class BacktestConfig:
    top_k: int = 50
    beta_lookback: int = 60
    cov_lookback: int = 12
    risk_model: str = "factor"
    min_cov_obs: int = 6
    idio_var_floor: float = 1e-4
    ridge_alpha: float = 0.0
    n_clusters: int = 3
    cluster_method: str = "gmm"
    regime_classifier: str = "rf"
    use_regime_probability: bool = False
    regime_feature_lags: tuple[int, ...] = (1, 3, 6)
    regime_feature_rolling_windows: tuple[int, ...] = (3, 6, 12)
    min_regime_train_obs: int = 24
    rf_n_estimators: int = 200
    rf_max_depth: int | None = 4
    rf_min_samples_leaf: int = 2
    xgb_n_estimators: int = 200
    xgb_max_depth: int = 3
    xgb_learning_rate: float = 0.05
    w_max: float = 0.10
    lambda_signal: float = 0.005
    gamma_div: float = 0
    cost_bps: float = 0
    hmm_n_iter: int = 100
    hmm_tol: float = 1e-4
    hmm_stickiness: float = 0.85
    hmm_covariance_floor: float = 1e-4
    random_state: int = 42


def post_return_weights(weights: pd.Series | None, realized_returns: pd.Series | None) -> pd.Series | None:
    if weights is None or realized_returns is None or weights.empty:
        return weights

    aligned_returns = realized_returns.reindex(weights.index).fillna(0.0)
    gross_values = weights * (1.0 + aligned_returns)
    total_value = gross_values.sum()
    if total_value <= 0:
        return pd.Series(dtype=float)
    return gross_values / total_value


def compute_turnover(
    new_weights: pd.Series,
    previous_weights: pd.Series | None = None,
    previous_period_returns: pd.Series | None = None,
) -> float:
    new_weights = new_weights[new_weights > 0].copy()
    if new_weights.empty:
        return 0.0

    if previous_weights is None or previous_weights.empty:
        return float(new_weights.sum())

    previous_live = post_return_weights(previous_weights, previous_period_returns)
    if previous_live is None or previous_live.empty:
        return float(new_weights.sum())

    universe = previous_live.index.union(new_weights.index)
    previous_live = previous_live.reindex(universe, fill_value=0.0)
    new_weights = new_weights.reindex(universe, fill_value=0.0)
    return float(0.5 * np.abs(new_weights - previous_live).sum())


def compute_performance_metrics(
    returns: pd.Series,
    benchmark_returns: pd.Series,
    risk_free: pd.Series,
    turnover: pd.Series | None = None,
) -> dict:
    returns = returns.dropna()
    benchmark_returns = benchmark_returns.reindex(returns.index).dropna()
    common_index = returns.index.intersection(benchmark_returns.index)
    returns = returns.loc[common_index]
    benchmark_returns = benchmark_returns.loc[common_index]
    risk_free = risk_free.reindex(common_index).fillna(0.0)

    if returns.empty:
        return {
            "months": 0,
            "cagr": np.nan,
            "annual_vol": np.nan,
            "sharpe": np.nan,
            "max_drawdown": np.nan,
            "hit_rate": np.nan,
            "avg_monthly_turnover": np.nan,
            "annual_turnover": np.nan,
            "benchmark_cagr": np.nan,
            "benchmark_sharpe": np.nan,
            "benchmark_max_drawdown": np.nan,
        }

    num_months = len(returns)
    num_years = num_months / 12.0

    nav = (1.0 + returns).cumprod()
    bench_nav = (1.0 + benchmark_returns).cumprod()

    running_max = nav.cummax()
    bench_running_max = bench_nav.cummax()

    excess_returns = returns - risk_free
    benchmark_excess = benchmark_returns - risk_free

    annual_vol = returns.std(ddof=1) * np.sqrt(12) if num_months > 1 else np.nan
    sharpe = (
        excess_returns.mean() / excess_returns.std(ddof=1) * np.sqrt(12)
        if num_months > 1 and excess_returns.std(ddof=1) > 0
        else np.nan
    )
    benchmark_sharpe = (
        benchmark_excess.mean() / benchmark_excess.std(ddof=1) * np.sqrt(12)
        if num_months > 1 and benchmark_excess.std(ddof=1) > 0
        else np.nan
    )

    avg_turnover = turnover.mean() if turnover is not None and not turnover.empty else np.nan

    return {
        "months": num_months,
        "cagr": nav.iloc[-1] ** (1 / num_years) - 1 if num_years > 0 else np.nan,
        "annual_vol": annual_vol,
        "sharpe": sharpe,
        "max_drawdown": (nav / running_max - 1).min(),
        "hit_rate": (returns > 0).mean(),
        "avg_monthly_turnover": avg_turnover,
        "annual_turnover": avg_turnover * 12 if pd.notna(avg_turnover) else np.nan,
        "benchmark_cagr": bench_nav.iloc[-1] ** (1 / num_years) - 1 if num_years > 0 else np.nan,
        "benchmark_sharpe": benchmark_sharpe,
        "benchmark_max_drawdown": (bench_nav / bench_running_max - 1).min(),
    }