PythagorasGoal/tests/test_vrp_profit_take.py

"""Test della misura del profit-take su VRP01 (scripts/research/r0730_vrp_profit_take.py).

Tre cose vanno congelate qui, e una sola e' il risultato:

* **La replica bit-exact.** Tutto il confronto poggia su una re-implementazione del sleeve:
  se qualcuno tocca `_vrp_weekly_asset`, il test deve rompersi invece di far passare un delta
  misurato fra due cose diverse.
* **La trappola dell'inner-join.** Le due varianti escono in date DIVERSE, quindi un
  `join="inner"` fra le due serie confronta solo le settimane senza uscita anticipata e
  risponde "identiche" con potenza zero. E' l'errore che ho commesso in sessione; il test lo
  rende irripetibile.
* **Il verdetto** (PT50 non batte il canonico) — con il **controllo positivo** obbligatorio sul
  null del de-levering: un test che dice sempre REFUTED e' indistinguibile da uno rotto.
"""
from __future__ import annotations

import importlib.util
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))

_spec = importlib.util.spec_from_file_location(
    "r0730_pt", ROOT / "scripts" / "research" / "r0730_vrp_profit_take.py")
PT = importlib.util.module_from_spec(_spec)
_spec.loader.exec_module(PT)

from src.portfolio.sleeves import VRP_CFG, _vrp_weekly_asset      # noqa: E402


# ---------------------------------------------------------------------------
# cache: ogni corsa rilegge 7 anni di storia
# ---------------------------------------------------------------------------
@pytest.fixture(scope="module")
def canon():
    return PT.combo(pt=None, f=1.0, real_fee=True)[0]


@pytest.fixture(scope="module")
def pt50():
    return PT.combo(pt=0.50, f=1.0, real_fee=True)[0]


# ===========================================================================
# la fondazione: senza questa nessun delta significa niente
# ===========================================================================
@pytest.mark.parametrize("asset", ["BTC", "ETH"])
def test_la_replica_e_bit_exact_col_sleeve(asset):
    mine, _ = PT.run_asset(asset, pt=None, f=VRP_CFG["f"], real_fee=False)
    ref = _vrp_weekly_asset(asset)
    j = pd.concat({"m": mine, "r": ref}, axis=1, join="inner")
    assert len(j) == len(ref) == len(mine)           # nessuna data persa da nessuna delle due
    assert float((j["m"] - j["r"]).abs().max()) == 0.0


# ===========================================================================
# l'errore di metodo commesso in sessione, congelato
# ===========================================================================
def test_confrontare_per_data_di_uscita_da_una_risposta_falsa(canon, pt50):
    """Un inner-join fra canonico e PT tiene solo le settimane in cui il PT NON e' scattato,
    dove per costruzione i due valori coincidono: la conclusione sarebbe 'nessuna differenza'."""
    a = canon[canon != 0.0]
    b = pt50[pt50 != 0.0]
    j = pd.concat({"a": a, "b": b}, axis=1, join="inner")
    assert len(j) < 0.9 * min(len(a), len(b))        # il join butta via la maggior parte dei dati
    # ...e su cio' che resta le due sono IDENTICHE: la risposta sarebbe "nessuna differenza"
    assert float((j["a"] - j["b"]).abs().max()) == 0.0
    # mentre la differenza vera, misurata appaiata, e' enorme
    assert abs(PT.stats(a)["sh"] - PT.stats(b)["sh"]) > 1.0


def test_il_confronto_appaiato_per_ingresso_vede_la_differenza():
    """La lente giusta: stesso ingresso, due esiti. Qui la differenza c'e' ed e' grande."""
    R = PT.paired_trades("ETH", pt=0.50)
    assert len(R) > 50
    assert R["early"].mean() > 0.5                    # il trigger e' raggiungibile
    assert R["pt"].sum() < R["hold"].sum()            # e costa


# ===========================================================================
# il meccanismo
# ===========================================================================
@pytest.mark.parametrize("asset", ["BTC", "ETH"])
def test_il_profit_take_scatta_sui_vincenti_non_sui_perdenti(asset):
    """E' il motivo per cui perde: tronca i trade che sarebbero finiti bene e non salva
    quelli che finiscono male."""
    R = PT.paired_trades(asset, pt=0.50)
    W, L = R[R["hold"] > 0], R[R["hold"] <= 0]
    assert W["early"].mean() > 0.75                   # >75% dei vincenti escono prima
    assert L["early"].mean() < 0.40                   # <40% dei perdenti
    assert W["early"].mean() > 2.0 * L["early"].mean()
    assert W["pt"].mean() < W["hold"].mean()          # i vincenti vengono troncati


# ===========================================================================
# il trigger non dipende da f — proprieta', non coincidenza
# ===========================================================================
def test_il_trigger_e_invariante_in_f():
    """`f*debit <= (1-pt)*f*credito` non contiene f: cambiare il prezzo del modello sposta il
    PnL ma non QUANDO si esce. Se qualcuno applicasse f asimmetricamente, questo si rompe."""
    a = PT.paired_trades("ETH", pt=0.50, f=1.00)
    b = PT.paired_trades("ETH", pt=0.50, f=0.73)
    assert len(a) == len(b)
    assert (a["early"].values == b["early"].values).all()


# ===========================================================================
# fee: la forma del listino, non solo il livello
# ===========================================================================
def test_la_fee_di_gamba_usa_il_sottostante_quando_il_premio_e_alto():
    # opzione cara: 12.5% del premio (12.5) e' piu' del 3bps del sottostante (0.57) -> vince il 3bps
    assert PT.leg_fee(1900.0, 100.0) == pytest.approx(0.0003 * 1900.0)


def test_la_fee_di_gamba_e_tappata_al_premio_quando_l_opzione_e_quasi_senza_valore():
    # opzione da 1 dollaro: il cap al 12.5% (0.125) morde, altrimenti si pagherebbe piu' del premio
    assert PT.leg_fee(1900.0, 1.0) == pytest.approx(0.125)


def test_la_fee_non_e_mai_negativa_su_premio_negativo():
    assert PT.leg_fee(1900.0, -5.0) == 0.0


# ===========================================================================
# il verdetto, e il controllo positivo che lo rende credibile
# ===========================================================================
def test_pt50_non_batte_il_canonico(canon, pt50):
    for match in ("vol", "dd"):
        r = PT.delever_null(canon, pt50, match)
        assert not r["batte"], f"match={match}: il verdetto e' cambiato, va rivisto il diario"
        assert r["d_sh"] < -0.5


def test_controllo_positivo_il_null_sa_dire_di_si(canon):
    """Obbligatorio: un null che risponde sempre REFUTED e' indistinguibile da uno rotto.
    Una variante COSTRUITA migliore (stesso rendimento, meta' della volatilita' di rumore)
    deve passare."""
    better = canon.copy()
    better[better < 0] *= 0.5                                  # perdite dimezzate = davvero meglio
    r = PT.delever_null(canon, better, "vol")
    assert r["batte"] and r["d_sh"] > 0


# ===========================================================================
# eseguibilita': a $3.000 la domanda viene prima del rendimento
# ===========================================================================
@pytest.fixture(scope="module")
def ex3k():
    return PT.executability(3000.0)


def test_a_3000_le_opzioni_btc_sono_fuori(ex3k):
    """min_trade_amount 0.1 BTC = ~$6.000 di collaterale cash-secured per lotto."""
    assert ex3k["BTC"]["lots"] == 0
    assert ex3k["BTC"]["collat"] > 3000.0


def test_a_3000_eth_entra_per_un_solo_lotto(ex3k):
    assert ex3k["ETH"]["lots"] == 1
    assert ex3k["ETH"]["collat"] < 3000.0 < 2 * ex3k["ETH"]["collat"]


def test_al_peso_di_book_lo_sleeve_non_e_collocabile(ex3k):
    """VRP01 pesa il 12%: $360 su $3.000, sotto il collaterale di un solo lotto ETH.
    E' il vincolo che decide, non lo Sharpe."""
    assert int(3000.0 * 0.12 // ex3k["ETH"]["collat"]) == 0


def test_il_lotto_minimo_e_quello_letto_dal_venue():
    """Congelato dal listino Deribit del 2026-07-30: se cambia, i numeri di questo studio
    cambiano e il test deve dirlo."""
    assert PT.MIN_TRADE == {"BTC": 0.1, "ETH": 1.0}
    assert PT.FEE_UNDERLYING == 0.0003 and PT.FEE_PREMIUM_CAP == 0.125