PythagorasGoal/tests/test_gtaa_ucits.py

"""Test della rivalidazione di GTAA01 sui veicoli UCITS (scripts/research/r0726_gtaa_ucits.py).

Tre cose vanno bloccate qui, e sono di natura diversa:
  1. il MOTORE generalizzato deve essere la produzione (bit-exact), altrimenti le tre lenti
     misurano una strategia che non e' quella del progetto;
  2. lo STIMATORE del drag: la media delle differenze giornaliere NON e' la deviazione fra due
     veicoli sullo stesso indice, e il test lo dimostra su un caso costruito dove la risposta
     giusta e' nota;
  3. il FINDING operativo: a $3.000 l'insieme di veicoli a prezzo unitario alto perde due gambe,
     quello a prezzo basso no. E' la ragione per cui il frazionamento del broker non serve.
"""
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"))

from src.portfolio import gtaa as G  # noqa: E402

import r0726_gtaa_ucits as U  # noqa: E402

RAW = ROOT / "data" / "raw"


def _need(prefix: str, syms) -> None:
    for s in syms:
        if not (RAW / f"{prefix}_{s.lower()}_1d.parquet").exists():
            pytest.skip(f"{prefix}_{s.lower()}_1d.parquet assente (data/raw e' gitignored)")


# ===========================================================================
# 1. il motore E' la produzione
# ===========================================================================
def test_il_motore_generalizzato_riproduce_gtaa_returns_bit_exact():
    """IL test di questo file. `sleeve_native` deve dare ESATTAMENTE gtaa.gtaa_returns: se
    diverge, ogni confronto fra lenti sta misurando un'altra strategia."""
    _need("eq", U.US)
    us = {s: U.close_us(s) for s in U.US}
    mine = U.sleeve_native(us, us, G.GTAA_DEFAULT_CAPITAL, G.REBAL_BAND_USD, G.REBAL_EVERY,
                           U.cost_ib_us)
    prod = U._norm(G.gtaa_returns(capital=G.GTAA_DEFAULT_CAPITAL))
    both = pd.concat([mine, prod], axis=1).dropna()
    assert len(both) > 5000
    assert float((both.iloc[:, 0] - both.iloc[:, 1]).abs().max()) == 0.0


def test_segnale_e_veicolo_separati_coincidono_quando_sono_la_stessa_serie():
    _need("eq", ["SPY"])
    _need("eqx", ["CSPX"])
    spy, cspx = U.close_us("SPY"), U.close_uc("CSPX", U.close_us("SPY"))[0]
    cal = U.common_cal([spy, cspx])
    a = U.leg_returns(spy, spy, cal, 1000.0, 50.0, 5, U.cost_ib_us)[0]
    b = U.leg_returns(spy, spy, cal, 1000.0, 50.0, 5, U.cost_ib_us)[0]
    assert float((a - b).abs().max()) == 0.0
    # con veicolo diverso il risultato DEVE cambiare, altrimenti il parametro e' ignorato
    c = U.leg_returns(spy, cspx, cal, 1000.0, 50.0, 5, U.cost_ib_us)[0]
    assert float((a - c).abs().max()) > 1e-6


# ===========================================================================
# 2. lo stimatore del drag
# ===========================================================================
def _due_veicoli(n=1500, drag_annuo=0.0, rumore=0.0, seed=3):
    """Due veicoli sullo STESSO indice: b = a x (drift noto) x (rumore di campionamento).
    Il rumore e' i.i.d. e stazionario — e' l'analogo del fatto che Londra chiude 4h30 prima."""
    rng = np.random.default_rng(seed)
    idx = pd.bdate_range("2015-01-01", periods=n)
    p = 100 * np.exp(np.cumsum(rng.normal(0.0003, 0.011, n)))
    a = pd.Series(p, index=idx)
    b = a * (1 + drag_annuo) ** (np.arange(n) / 252.0)
    if rumore:
        b = b * (1 + rng.normal(0, rumore, n))
    return a, b


def test_il_drag_cumulato_recupera_un_drag_noto():
    a, b = _due_veicoli(drag_annuo=-0.0030)
    assert U.drag_cumulato(a, b, a.index) == pytest.approx(-0.0030, abs=2e-4)


def test_il_drag_cumulato_e_immune_al_rumore_di_campionamento():
    """Il rumore stazionario non e' un drag: uno stimatore corretto non lo deve vedere."""
    a, b = _due_veicoli(drag_annuo=0.0, rumore=0.01)
    assert abs(U.drag_cumulato(a, b, a.index)) < 0.003


def test_la_media_delle_differenze_giornaliere_NON_e_lo_stimatore_giusto():
    """⚠️ Il test che giustifica `drag_cumulato`. Due veicoli con lo STESSO drift geometrico ma
    campionati diversamente: la media delle differenze semplici vede un drag che non esiste,
    perche' e' contaminata dalla diversa varianza giornaliera (drag di varianza), mentre il
    rapporto cumulato — che e' la domanda vera, 'quanto mi costa il veicolo' — vede ~0.

    E' il difetto che nella prima stesura dava -0.47%/anno su CSPX invece di -0.06%.
    """
    a, b = _due_veicoli(drag_annuo=0.0, rumore=0.012)
    cum = U.drag_cumulato(a, b, a.index)
    media = float((b.pct_change() - a.pct_change()).dropna().mean() * 252)
    assert abs(cum) < 0.003, "il rapporto cumulato dovrebbe vedere ~0"
    assert abs(media) > 5 * abs(cum), \
        f"il caso costruito non separa i due stimatori (cum {cum:.4%}, media {media:.4%})"


# ===========================================================================
# 3. il finding operativo: prezzo unitario e gambe vive
# ===========================================================================
def test_azioni_intere_non_possono_aumentare_l_esposizione():
    """Un vincolo arrotonda IN GIU': se l'esposizione sale, il modello e' sbagliato (e uno
    Sharpe piu' alto sarebbe null de-levering travestito da miglioramento)."""
    _need("eqx", set(U.SET_DEPLOY.values()))
    _need("eq", U.US)
    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(dep.values()))[-800:]
    for cap in (3_000.0, 20_000.0):
        _, pf = U.sleeve(dep, dep, cal, cap, G.REBAL_BAND_USD, G.REBAL_EVERY,
                         U.cost_form(fixed=1.0), want_pos=True)
        _, pi = U.sleeve(dep, dep, cal, cap, G.REBAL_BAND_USD, G.REBAL_EVERY,
                         U.cost_form(fixed=1.0), integer_shares=True, want_pos=True)
        assert pi.sum().sum() <= pf.sum().sum() + 1e-9


def test_a_3000_l_insieme_a_prezzo_alto_perde_gambe_e_quello_a_prezzo_basso_no():
    """IL finding operativo, bloccato. CSPX ($802) ed EQQQ ($693) non sono comprabili con una
    gamba da $500: senza frazionamento quelle due gambe semplicemente non esistono."""
    _need("eq", U.US)
    _need("eqx", set(U.SET_STORIA.values()) | set(U.SET_DEPLOY.values()))
    us = {s: U.close_us(s) for s in U.US}
    storia = {s: U.close_uc(u, us[s])[0] for s, u in U.SET_STORIA.items()}
    deploy = {s: U.close_uc(u, us[s])[0] for s, u in U.SET_DEPLOY.items()}
    cal = U.common_cal(list(storia.values()) + list(deploy.values()))[-700:]

    def gambe_vive(v):
        _, pos = U.sleeve(v, v, cal, 3_000.0, G.REBAL_BAND_USD, G.REBAL_EVERY,
                          U.cost_form(fixed=1.0), integer_shares=True, want_pos=True)
        return int((pos.abs().sum(axis=0) > 0).sum())

    assert gambe_vive(storia) < 6, "l'insieme a prezzo alto dovrebbe perdere gambe a $3.000"
    assert gambe_vive(deploy) == 6, "l'insieme a prezzo basso deve tenerle tutte e sei"


def test_i_veicoli_deploy_costano_meno_per_azione_di_quelli_storia():
    _need("eq", U.US)
    _need("eqx", set(U.SET_STORIA.values()) | set(U.SET_DEPLOY.values()))
    us = {s: U.close_us(s) for s in U.US}
    for s in U.US:
        a = float(U._norm(U.close_uc(U.SET_STORIA[s], us[s])[0]).iloc[-1])
        b = float(U._norm(U.close_uc(U.SET_DEPLOY[s], us[s])[0]).iloc[-1])
        assert b <= a + 1e-9, f"{s}: il veicolo 'deploy' non e' piu' economico ({b} > {a})"


# ===========================================================================
# scelte dichiarate
# ===========================================================================
def test_i_veicoli_scelti_sono_indici_equivalenti_non_fattori():
    """RTWO e' Russell 2000 *quality*, CSUSS e' MSCI USA Small Cap *ESG*, IDP6 e' S&P 600:
    tutti hanno piu' storia dei veicoli scelti, e nessuno e' la stessa esposizione."""
    scelti = set(U.SET_STORIA.values()) | set(U.SET_DEPLOY.values())
    assert not (scelti & {"RTWO", "CSUSS", "IDP6"})


def test_le_due_mappe_coprono_le_stesse_sei_gambe():
    assert tuple(U.SET_STORIA) == U.US and tuple(U.SET_DEPLOY) == U.US
    assert len(U.US) == 6