optimizer/vol_optimizer.py

from __future__ import annotations

import pandas as pd

from backtest.vol_backtest import backtest_realized_vol_signal


def optimize_volatility_signal(
    prices: pd.Series,
    signal: str = "long_vol",
    short_windows: tuple[int, ...] = (5, 10, 15),
    long_windows: tuple[int, ...] = (20, 30, 60),
    holding_days_options: tuple[int, ...] = (3, 5, 10),
) -> dict:
    runs = []
    for short_window in short_windows:
        for long_window in long_windows:
            if short_window >= long_window:
                continue
            for holding_days in holding_days_options:
                result = backtest_realized_vol_signal(
                    prices=prices,
                    short_window=short_window,
                    long_window=long_window,
                    holding_days=holding_days,
                    signal=signal,
                )
                runs.append(
                    {
                        "short_window": short_window,
                        "long_window": long_window,
                        "holding_days": holding_days,
                        "trade_count": result["trade_count"],
                        "win_rate": result["win_rate"],
                        "total_return_proxy": result["total_return_proxy"],
                        "max_drawdown_proxy": result["max_drawdown_proxy"],
                        "avg_trade_pnl_proxy": result["avg_trade_pnl_proxy"],
                    }
                )

    runs.sort(
        key=lambda run: (
            run["total_return_proxy"],
            -abs(run["max_drawdown_proxy"]),
            run["win_rate"],
        ),
        reverse=True,
    )
    best = runs[0] if runs else None
    baseline = next(
        (
            run
            for run in runs
            if run["short_window"] == 10 and run["long_window"] == 30 and run["holding_days"] == 5
        ),
        runs[0] if runs else None,
    )
    return {
        "signal": signal,
        "best": best,
        "baseline": baseline,
        "top_runs": runs[:10],
        "metrics_delta": (
            {
                "total_return_proxy_delta": best["total_return_proxy"] - baseline["total_return_proxy"],
                "win_rate_delta": best["win_rate"] - baseline["win_rate"],
                "max_drawdown_proxy_delta": best["max_drawdown_proxy"] - baseline["max_drawdown_proxy"],
            }
            if best and baseline
            else None
        ),
        "anti_overfit_note": (
            "This is an in-sample parameter scan. Use walk-forward or out-of-sample validation "
            "before trusting optimized parameters."
        ),
    }