regime-aware-multi-factor/validation.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
from __future__ import annotations

from dataclasses import replace

import numpy as np
import pandas as pd
from sklearn.metrics import balanced_accuracy_score, f1_score

from Backtest_utils import BacktestConfig
from Regime_forecast import has_xgboost
from main_update import run_backtest


def compute_regime_classification_metrics(detail_df: pd.DataFrame) -> dict:
    if detail_df.empty:
        return {
            "balanced_accuracy": np.nan,
            "macro_f1": np.nan,
            "forecast_count": 0,
        }

    regime_df = detail_df.dropna(subset=["forecasted_regime", "realized_next_regime"]).copy()
    if regime_df.empty:
        return {
            "balanced_accuracy": np.nan,
            "macro_f1": np.nan,
            "forecast_count": 0,
        }

    y_true = regime_df["realized_next_regime"].astype(int)
    y_pred = regime_df["forecasted_regime"].astype(int)
    return {
        "balanced_accuracy": balanced_accuracy_score(y_true, y_pred),
        "macro_f1": f1_score(y_true, y_pred, average="macro"),
        "forecast_count": len(regime_df),
    }


def run_regime_walk_forward_validation(
    price_df: pd.DataFrame,
    returns_df: pd.DataFrame,
    ff5_realized_df: pd.DataFrame,
    ff5_signal_df: pd.DataFrame,
    benchmark_df: pd.DataFrame,
    base_config: BacktestConfig,
    n_clusters_grid: tuple[int, ...] = (2, 3, 4),
    stickiness_grid: tuple[float, ...] = (0.70, 0.85, 0.95),
    classifier_grid: tuple[str, ...] = ("rf", "xgb"),
) -> pd.DataFrame:
    rows = []
    for classifier_name in classifier_grid:
        if classifier_name == "xgb" and not has_xgboost():
            continue

        for n_clusters in n_clusters_grid:
            for stickiness in stickiness_grid:
                config = replace(
                    base_config,
                    n_clusters=n_clusters,
                    regime_classifier=classifier_name,
                    hmm_stickiness=stickiness,
                )
                detail_df, metrics_df = run_backtest(
                    price_df=price_df,
                    returns_df=returns_df,
                    ff5_realized_df=ff5_realized_df,
                    benchmark_df=benchmark_df,
                    config=config,
                    ff5_signal_df=ff5_signal_df,
                    show_progress=False,
                )

                strategy_metrics = metrics_df.loc["signal_tilt"].to_dict() if not metrics_df.empty else {}
                class_metrics = compute_regime_classification_metrics(detail_df)
                rows.append(
                    {
                        "n_clusters": n_clusters,
                        "hmm_stickiness": stickiness,
                        "regime_classifier": classifier_name,
                        "balanced_accuracy": class_metrics["balanced_accuracy"],
                        "macro_f1": class_metrics["macro_f1"],
                        "forecast_count": class_metrics["forecast_count"],
                        "months": strategy_metrics.get("months"),
                        "cagr": strategy_metrics.get("cagr"),
                        "annual_vol": strategy_metrics.get("annual_vol"),
                        "sharpe": strategy_metrics.get("sharpe"),
                        "max_drawdown": strategy_metrics.get("max_drawdown"),
                        "annual_turnover": strategy_metrics.get("annual_turnover"),
                    }
                )

    result_df = pd.DataFrame(rows)
    if result_df.empty:
        return result_df

    return result_df.sort_values(
        by=["sharpe", "cagr", "balanced_accuracy"],
        ascending=[False, False, False],
    ).reset_index(drop=True)