PythagorasGoal/tests/test_gtaa_broker.py

"""Test della configurazione di esecuzione di GTAA01 su un broker a costo fisso
(scripts/research/r0727_gtaa_broker.py).

Il finding da bloccare e' UNO: una banda in dollari ASSOLUTI non scala col capitale, e a capitale
piccolo congela lo sleeve invece di renderlo frugale. Lo Sharpe non lo dice — resta accettabile —
lo dice la QUOTA DI TEMPO A MERCATO. E' la stessa firma del null de-levering che il progetto ha
gia' incontrato quattro volte, in una veste nuova.
"""
from __future__ import annotations

import re
import sys
from pathlib import Path

import numpy as np
import pytest

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

import r0726_gtaa_ucits as U  # noqa: E402
import r0727_gtaa_broker as B  # noqa: E402

RAW = ROOT / "data" / "raw"


@pytest.fixture(scope="module")
def ctx():
    for s in list(U.US) + list(U.SET_DEPLOY.values()):
        pre = "eq" if s in U.US else "eqx"
        if not (RAW / f"{pre}_{s.lower()}_1d.parquet").exists():
            pytest.skip("data/raw incompleto (gitignored)")
    us = {s: U.close_us(s) for s in U.US}
    dep = {s: U.close_uc(u, us[s])[0] for s, u in U.SET_DEPLOY.items()}
    cal = U.common_cal(list(us.values()) + list(dep.values()))
    start = max(U._norm(x).index[0] for x in dep.values())
    return dep, cal[cal >= start + np.timedelta64(380, "D")]


def _esposta(dep, cal, cap, band, every=5) -> float:
    _, P = U.sleeve(dep, dep, cal, cap, band, every, U.cost_form(fixed=1.0), want_pos=True)
    return float((P > 0).mean().mean())


# ===========================================================================
# il finding
# ===========================================================================
def test_banda_assoluta_larga_congela_lo_sleeve_a_capitale_piccolo(ctx):
    """IL test. A $3.000 una banda da $400 e' l'80% della gamba: lo sleeve smette di seguire il
    target. Lo Sharpe resta accettabile, quindi il controllo NON puo' essere sullo Sharpe."""
    dep, cal = ctx
    assert _esposta(dep, cal, 3_000.0, 400.0) < B.ESPOSTA_SANA
    assert B.ordini_anno(dep, cal, 3_000.0, 400.0, 5) < 10


def test_la_stessa_banda_a_capitale_grande_e_sana(ctx):
    """Il confronto che rende il test precedente informativo: non e' la banda a essere sbagliata,
    e' il fatto che sia espressa in dollari assoluti."""
    dep, cal = ctx
    assert _esposta(dep, cal, 10_000.0, 400.0) >= B.ESPOSTA_SANA


def test_la_banda_proporzionale_resta_sana_a_ogni_capitale(ctx):
    dep, cal = ctx
    for cap in (3_000.0, 10_000.0, 50_000.0):
        assert _esposta(dep, cal, cap, B.BAND_FRAC * cap / 6.0) >= B.ESPOSTA_SANA, \
            f"congelato a ${cap:,.0f}"


def test_la_banda_proporzionale_rende_gli_ordini_invarianti_al_capitale(ctx):
    """E' la proprieta' che serve: il costo annuo scala col capitale invece che restare fisso."""
    dep, cal = ctx
    n = [B.ordini_anno(dep, cal, c, B.BAND_FRAC * c / 6.0, 5) for c in (3_000., 10_000., 50_000.)]
    assert max(n) - min(n) <= 2, f"ordini/anno non invarianti: {n}"


def test_una_banda_piu_larga_non_puo_generare_piu_ordini(ctx):
    dep, cal = ctx
    n = [B.ordini_anno(dep, cal, 10_000.0, b, 5) for b in (25.0, 100.0, 400.0)]
    assert n == sorted(n, reverse=True), f"monotonia violata: {n}"


# ===========================================================================
# la soglia per il broker
# ===========================================================================
def test_la_soglia_per_ordine_cresce_col_capitale(ctx):
    """Con banda proporzionale gli ordini sono costanti ma il capitale cresce: la tariffa
    tollerabile deve crescere in proporzione."""
    dep, cal = ctx
    v = []
    for cap in (3_000.0, 10_000.0):
        s = B.soglia_per_ordine(dep, cal, cap, B.BAND_FRAC * cap / 6.0, 5)
        v.append(60.0 if s.startswith(">") else float(s[1:]))
    assert v[1] > 2 * v[0], f"la soglia non scala col capitale: {v}"


def test_la_banda_proporzionale_alza_la_soglia_rispetto_al_canonico(ctx):
    """La ragione operativa di tutto il filone: a $3.000 si passa da ~$2 a ~$5 per ordine."""
    dep, cal = ctx
    can = B.soglia_per_ordine(dep, cal, 3_000.0, 50.0, 5)
    prop = B.soglia_per_ordine(dep, cal, 3_000.0, B.BAND_FRAC * 3_000.0 / 6.0, 5)
    assert float(prop[1:]) > 2 * float(can[1:])


# ===========================================================================
# cio' che l'operatore copia e incolla
# ===========================================================================
def test_le_isin_sono_ben_formate_e_coprono_le_sei_gambe():
    """Le ISIN finiscono in un campo di ricerca su un conto reale: un refuso qui compra un'altra
    cosa. Formato ISO 6166 e domicilio irlandese (e' il motivo per cui sono acquistabili nell'UE)."""
    assert set(B.ISIN) == set(U.SET_DEPLOY.values())
    for t, (isin, desc) in B.ISIN.items():
        assert re.fullmatch(r"IE00[A-Z0-9]{8}", isin), f"{t}: ISIN malformata {isin}"
        assert desc.strip()


def test_le_soglie_di_lettura_sono_dichiarate():
    assert B.BAND_FRAC == 0.25
    assert B.ESPOSTA_SANA == 0.60


# ===========================================================================
# equivalenze fra listini — la stessa ISIN su borse diverse e' lo STESSO fondo
# ===========================================================================
def test_le_equivalenze_dichiarate_condividono_la_isin_del_veicolo_scelto():
    """ZPRR (Xetra EUR) e R2US (Londra USD) sono la stessa classe di quote: se le ISIN non
    coincidono l'equivalenza e' falsa e l'operatore comprerebbe un altro fondo."""
    assert {a for a, _, _, _ in B.EQUIVALENZE} == set(B.ISIN), \
        "ogni gamba deve avere una linea in euro dichiarata: su un conto in EUR e' cio' che evita la conversione"
    for a, b, isin, _ in B.EQUIVALENZE:
        assert a in B.ISIN, f"{a} non e' un veicolo del set DEPLOY"
        assert B.ISIN[a][0] == isin, f"{a}: la ISIN dell'equivalenza non e' quella del veicolo"
        assert re.fullmatch(r"IE00[A-Z0-9]{8}", isin)
        if b != a:
            assert b not in B.ISIN, f"{b} e' un ALIAS di {a}, non un veicolo distinto"


def test_ogni_isin_compare_una_volta_sola_fra_le_equivalenze():
    """Due gambe con la stessa ISIN sarebbero lo stesso fondo comprato due volte: il vol-target
    girerebbe su cinque esposizioni invece di sei senza che nulla lo segnali."""
    isins = [i for _, _, i, _ in B.EQUIVALENZE]
    assert len(set(isins)) == len(isins)
    assert len(set(B.ISIN[t][0] for t in B.ISIN)) == 6


def test_il_ripiego_smallcap_e_dichiarato_come_indice_DIVERSO():
    """SPY4 e' S&P 400 MID cap: e' un ripiego misurato, non un equivalente. Il test esiste
    perche' la differenza fra 'equivalente' e 'ripiego accettabile' e' proprio cio' che si
    perde quando si copiano ticker da una lista."""
    tic, isin, desc = B.RIPIEGO_SMALLCAP
    assert re.fullmatch(r"IE00[A-Z0-9]{8}", isin)
    assert isin != B.ISIN["R2US"][0], "un ripiego con la stessa ISIN sarebbe un'equivalenza"
    assert "Mid" in desc


def test_il_ripiego_smallcap_e_indistinguibile_dalla_gamba_scelta(ctx):
    """Il finding della sezione (6): sostituire mid a small lascia lo sleeve al 99% uguale.
    Se un giorno smettesse di essere vero, la raccomandazione 'usa SPY4 se non trovi R2US'
    andrebbe ritirata."""
    dep, cal = ctx
    us = {s: U.close_us(s) for s in U.US}
    mid = dict(dep)
    mid["IWM"], _ = U.close_uc(B.RIPIEGO_SMALLCAP[0], us["IWM"])
    cl = U.common_cal(list(dep.values()) + [mid["IWM"]])
    cl = cl[cl >= cal[0]]
    band = B.BAND_FRAC * 10_000.0 / 6.0
    a = U.sleeve(dep, dep, cl, 10_000.0, band, 5, U.cost_form(fixed=1.0))
    b = U.sleeve(mid, mid, cl, 10_000.0, band, 5, U.cost_form(fixed=1.0))
    assert a.corr(b) > 0.95, f"il ripiego non e' piu' equivalente: corr {a.corr(b):.3f}"