PythagorasGoal/tests/test_gtaa_band_gate.py

"""Test del gate onesto sulla banda GTAA01 (scripts/research/r0727_gtaa_band_gate.py).

Congela i tre esiti del gate e — soprattutto — l'ANNUALIZZAZIONE. Una serie su giorni di borsa
passata a un annualizzatore a 365 esce con lo Sharpe ×1.20 e il CAGR ×1.45: e' l'errore del
25/07, ed e' l'unico difetto qui che produrrebbe numeri sbagliati ma plausibili.

`test_la_proposta_non_e_selezionata_sull_hold_out` e' il test di metodo: se un giorno la cella
proposta risultasse in rango MIGLIORE sull'hold-out che in-sample, sarebbe la firma della
selezione-sull'hold-out e la validazione andrebbe rifatta.
"""
from __future__ import annotations

import sys
from pathlib import Path

import numpy as np
import pandas as pd
import pytest

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))
sys.path.insert(0, str(ROOT / "scripts" / "research"))
sys.path.insert(0, str(ROOT / "scripts" / "research" / "alt"))

import r0727_gtaa_band_gate as BG  # noqa: E402


@pytest.fixture(scope="module")
def grid():
    """Griglia ridotta alla sola cadenza della proposta: basta per i gate (B) e (C) e tiene
    il test veloce. La griglia piena vive nello script."""
    ref = BG.run_cell(1, 0.0, BG.CAP_REF)
    out = {}
    for frac in BG.FRAC_GRID:
        s = BG.run_cell(BG.PROPOSTA[0], frac, BG.CAP_REF)
        j = pd.concat([s, ref], axis=1, join="inner").dropna()
        out[frac] = dict(series=s, met=BG.met(s),
                         corr=float(j.iloc[:, 0].corr(j.iloc[:, 1])))
    return ref, out


# ===========================================================================
# annualizzazione — l'errore che produce numeri plausibili e sbagliati
# ===========================================================================
def test_annualizza_a_giorni_di_borsa():
    assert BG.DPY == 252.0


def test_lo_sharpe_e_calcolato_a_252_non_a_365():
    idx = pd.bdate_range("2020-01-01", periods=1000)
    r = pd.Series(np.linspace(0.0004, 0.0006, 1000), index=idx)
    v = r.values
    atteso = v.mean() / v.std() * np.sqrt(252.0)     # std di popolazione, come nel resto del progetto
    assert BG.met(r)["sharpe"] == pytest.approx(atteso, rel=1e-12)
    assert BG.met(r)["sharpe"] != pytest.approx(v.mean() / v.std() * np.sqrt(365.0), rel=1e-6)


# ===========================================================================
# la parametrizzazione: e' questa la proposta
# ===========================================================================
def test_la_banda_e_una_frazione_della_gamba_non_un_importo():
    """E' l'intera ragione della proposta: un importo assoluto ha effetti diversi a capitali
    diversi, una frazione no."""
    from src.portfolio.gtaa import EQ_UNIVERSE
    for cap in (3_000.0, 10_000.0, 50_000.0):
        assert 0.25 * cap / len(EQ_UNIVERSE) == pytest.approx(0.25 * cap / 6)


def test_la_produzione_usa_ancora_la_banda_ASSOLUTA():
    """Il gate valida una proposta che NON e' ancora in produzione. Se un giorno la produzione
    passasse alla banda frazionaria, questo test si rompe e va riletto tutto il confronto."""
    from src.portfolio.gtaa import REBAL_BAND_USD, REBAL_EVERY
    assert (REBAL_EVERY, REBAL_BAND_USD) == BG.CANONICO == (5, 50.0)


# ===========================================================================
# (A) selezione — il gate di metodo
# ===========================================================================
def test_la_proposta_non_e_selezionata_sull_hold_out():
    """Rango in-sample <= rango hold-out: la proposta non migliora passando all'hold-out.
    Il contrario sarebbe la firma della selezione-sull'hold-out."""
    rows = []
    for every in BG.EVERY_GRID:
        for frac in BG.FRAC_GRID:
            s = BG.run_cell(every, frac, BG.CAP_REF)
            rows.append(dict(every=every, frac=frac,
                             sh_in=BG.met(s.loc[: BG.HOLDOUT])["sharpe"],
                             sh_oos=BG.met(s.loc[BG.HOLDOUT:])["sharpe"]))
    G = pd.DataFrame(rows)
    p = G[(G["every"] == BG.PROPOSTA[0]) & (G["frac"] == BG.PROPOSTA[1])].iloc[0]
    rank_in = int((G["sh_in"] > p["sh_in"]).sum()) + 1
    rank_oos = int((G["sh_oos"] > p["sh_oos"]).sum()) + 1
    assert rank_in <= rank_oos, (
        f"la proposta e' {rank_oos}a sull'hold-out ma {rank_in}a in-sample: "
        "sta meglio dove non doveva essere guardata")


def test_l_hold_out_usato_e_quello_documentato_di_gtaa01():
    """2015+: l'OOS equity della validazione 22/06, indipendente dall'hold-out crypto."""
    assert BG.HOLDOUT == "2015-01-01"


# ===========================================================================
# (B) deflated Sharpe
# ===========================================================================
def test_la_proposta_sopravvive_alla_deflazione(grid):
    from altlib import deflated_sharpe
    ref, cells = grid
    all_sr = [c["met"]["sharpe"] for c in cells.values()]
    s = cells[BG.PROPOSTA[1]]["series"]
    dsr, _ = deflated_sharpe(BG.met(s)["sharpe"], all_sr, s.dropna().values, dpy=BG.DPY)
    assert dsr >= 0.95


# ===========================================================================
# (C) tracking — e il modo di fallire, che NON e' il de-levering
# ===========================================================================
def test_al_25_percento_lo_sleeve_segue_ancora_se_stesso(grid):
    _, cells = grid
    assert cells[0.25]["corr"] >= BG.MIN_CORR


def test_oltre_il_25_percento_il_tracking_si_rompe(grid):
    _, cells = grid
    assert cells[0.40]["corr"] < BG.MIN_CORR
    assert cells[0.60]["corr"] < cells[0.40]["corr"]


def test_la_banda_larga_non_de_leva(grid):
    """Il null de-levering classico NON morde qui: allargando la banda la vol non scende.
    Se un giorno scendesse, il fallimento sarebbe di un altro tipo e il verdetto va riscritto."""
    ref, cells = grid
    vref = BG.met(ref)["vol"]
    assert cells[0.60]["met"]["vol"] / vref > 0.95


def test_il_25_percento_e_al_bordo_non_al_centro(grid):
    """Onesta' sul margine: la proposta passa, ma di poco. Congelarlo evita che in futuro si
    citi come un plateau largo."""
    _, cells = grid
    assert BG.MIN_CORR <= cells[0.25]["corr"] < BG.MIN_CORR + 0.02


# ===========================================================================
# perche' la proposta esiste: invarianza al capitale
# ===========================================================================
def test_la_banda_frazionaria_batte_quella_fissa_a_capitale_piccolo():
    from src.portfolio.gtaa import gtaa_returns
    cap = 3_000.0
    frazionaria = BG.met(BG.run_cell(BG.PROPOSTA[0], BG.PROPOSTA[1], cap))["sharpe"]
    fissa = BG.met(gtaa_returns(capital=cap, band_usd=50.0, every=5))["sharpe"]
    assert frazionaria > fissa


def test_a_capitale_grande_le_due_configurazioni_convergono():
    """A $50k la banda fissa smette di essere una tassa: il vantaggio della proposta e' al
    capitale PICCOLO, ed e' li' che va citato."""
    from src.portfolio.gtaa import gtaa_returns
    cap = 50_000.0
    frazionaria = BG.met(BG.run_cell(BG.PROPOSTA[0], BG.PROPOSTA[1], cap))["sharpe"]
    fissa = BG.met(gtaa_returns(capital=cap, band_usd=50.0, every=5))["sharpe"]
    assert abs(frazionaria - fissa) < abs(
        BG.met(BG.run_cell(BG.PROPOSTA[0], BG.PROPOSTA[1], 3_000.0))["sharpe"]
        - BG.met(gtaa_returns(capital=3_000.0, band_usd=50.0, every=5))["sharpe"])