PythagorasGoal/scripts/research/r0901c_pavimento_leva.py

"""PAVIMENTO-LEVA — il pavimento come LICENZA DI TAGLIA, non come sconto sul drawdown.

DOMANDA (operatore, 2026-09-01): "possiamo usare quanto conosciamo del pavimento per studiare
una strategia?"

COSA SAPPIAMO DEL PAVIMENTO (COLLAR01, §71, stesso giorno):
  - FUNZIONA: con beta 1,0 il maxDD scende in 36/48 celle (§46 lo negava per un libro a beta 0,076);
  - a premio ZERO vale moltissimo: maxDD 51,83% -> 36,76%, drift +10,04% -> +35,69% (il pranzo gratis);
  - cio' che uccise il collar fu il TETTO, non il pavimento (C9: il tetto tagliava il 46% dei cicli
    vincenti, il pavimento parava il 6,7% dei perdenti);
  - la put e' l'ala CARA (skew: IV_put/ATM 1,25 a d0,10) e il DVOL sta il 32% sopra la RV: comprare
    il pavimento significa pagare il premio di varianza. Qui il prezzo a DVOL e' REALISTICO (e'
    cio' che il mercato chiede): M29 corregge chi VENDE a modello, non chi compra.

L'INVERSIONE. COLLAR01 chiese "quanto DD mi risparmia il pavimento a taglia fissa?" e perse contro il
de-levering. La domanda speculare e' "quanta TAGLIA mi autorizza il pavimento a DD fisso?". Sembra
la stessa (a iso-DD e' un solo confronto), ma non lo e' per due ragioni misurabili:
  (1) il de-levering riduce taglia e rischio IN PROPORZIONE; un pavimento cambia la FORMA — se la
      forma taglia la coda sinistra piu' di quanto costi, scalare in su la struttura protetta puo'
      battere la base nuda alla stessa DD. E' l'unica cosa che il de-levering NON puo' comprare;
  (2) il tetto di leva del progetto NON e' fissato dallo Sharpe ma dal COSTO DI UN EPISODIO di
      disaster-SL: n_asset * frac * scala * sl_pct <= 0,50 => k <= 1,67x (GATE SCALA-01, §4.5).
      Il disaster-SL e' ROTOLANTE e non limita la perdita (peggior DD dall'ingresso misurato
      -60,6%). Una put limita la perdita PER COSTRUZIONE — dentro la sua finestra.

ATTESE A PRIORI (scritte prima di misurare):
  B1  A premio reale il pavimento LICENZIA taglia (k_f > 1) ma il drift della struttura scalata
      resta SOTTO quello della base nuda: il premio costa piu' di quanto la licenza renda. Previsione
      centrale: il pavimento perde il test della licenza nella maggioranza delle celle.
  B2  La riduzione di DD viene dai crolli VELOCI (dentro una finestra di opzione), non dai declini
      LENTI: in un grind di settimane la put scade OTM ogni volta e si paga premio E perdita.
      Il maxDD di BTC 2021-2026 e' il 2022, che fu un grind => il pavimento lo protegge poco.
  B3  La "licenza" del disaster-SL e' di CARTA: l'invariante limita UN episodio, la put limita UNA
      finestra, e il vero massimo su finestre consecutive NON e' limitato. Sostituire sl_pct con la
      distanza del pavimento nell'invariante sarebbe un errore — e va scritto ora, prima che
      qualcuno lo proponga.
  B4  Il pavimento gated su IV a buon mercato (DVOL/RV basso) migliora ma non basta a passare.

NESSUN ORDINE. Libro, pesi, cron, config INVARIATI.
"""
from __future__ import annotations

import math
import sys
from pathlib import Path

import numpy as np
import pandas as pd

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

import r0901_btc_collar as C  # noqa: E402  — il motore, riusato e non riscritto

DPY = C.DPY

G_DPUT = (0.05, 0.10, 0.20, 0.30)
G_TENOR = (7, 14)


def k_up_for_same_dd(r: np.ndarray, target_dd: float, kmax: float = 6.0) -> float:
    """La scala >=1 che porta il maxDD della struttura PROTETTA esattamente a quello della base NUDA.
    E' il null M5 letto al contrario: invece di ridurre la base al DD del protetto, si ALZA il
    protetto al DD della base. maxDD(k*r) e' crescente in k: bisezione."""
    if C._dd(r) >= target_dd:
        return 1.0
    lo, hi = 1.0, kmax
    if C._dd(hi * r) < target_dd:
        return kmax
    for _ in range(60):
        mid = 0.5 * (lo + hi)
        if C._dd(mid * r) > target_dd:
            hi = mid
        else:
            lo = mid
    return 0.5 * (lo + hi)


def peggior_finestra(nav: pd.Series, giorni: int) -> float:
    """La peggior perdita su una finestra di `giorni` (quella che UNA put puo' coprire)."""
    r = nav / nav.shift(giorni) - 1.0
    return float(r.min())


def main() -> None:
    print("=" * 96)
    print("  PAVIMENTO-LEVA — il pavimento come licenza di taglia")
    print("=" * 96)
    print(__doc__.split("ATTESE A PRIORI")[1].split("NESSUN ORDINE")[0])

    cal = C.calibra_catena()
    P = C.serie_giornaliera()
    print(f"  LENTE LUNGA {P.index[0].date()} -> {P.index[-1].date()} ({len(P)/DPY:.2f} anni). "
          f"Prezzo a DVOL x skew x termine misurati sulla catena (realistico per chi COMPRA).")

    righe = []
    for forte in (True, False):
        dirs = C.gate_trend(P, forte)
        nm = "FORTE" if forte else "LARGO"
        base = C.collar(P, cal, dput=0.2, dcall=0.2, tenor=7, dirs=dirs, senza_opzioni=True)
        rb, nb = base["ret"], base["nav"]
        dd_b, dr_b, sh_b = C._dd(rb), C._drift(rb), C._sh(rb)
        print(f"\n  ── gate {nm}: BASE nuda Sharpe {sh_b:.3f} · maxDD {100*dd_b:.2f}% · drift {100*dr_b:+.2f}%/a"
              f" · peggior 7g {100*peggior_finestra(nb,7):.1f}% · peggior 14g {100*peggior_finestra(nb,14):.1f}%")
        print(f"     {'ten':>4}{'dput':>6}{'Sharpe':>8}{'maxDD':>8}{'drift':>9} | "
              f"{'k_f':>6}{'drift@k_f':>11}{'vs base':>9} | {'k null':>7}{'drift null':>11}{'vs':>4} | "
              f"{'pegg7':>7}{'pegg14':>7}{'pav.para':>9}")
        for tenor in G_TENOR:
            for dp in G_DPUT:
                o = C.collar(P, cal, dput=dp, dcall=0.2, tenor=tenor, dirs=dirs, solo_pavimento=True)
                r, nav = o["ret"], o["nav"]
                dd, dr, sh = C._dd(r), C._drift(r), C._sh(r)
                # (a) LICENZA: alzo il protetto fino al DD della base
                kf = k_up_for_same_dd(r.values, dd_b)
                dr_kf = C._drift(kf * r.values)
                lic = dr_kf > dr_b
                # (b) NULL M5 classico: abbasso la base fino al DD del protetto
                kn = C.k_for_same_dd(rb.values, dd)
                dr_n = C._drift(kn * rb.values)
                m5 = dr > dr_n and dd < dd_b
                cic = pd.DataFrame(o["cicli"])
                para = 100.0 * cic["floor"].mean() if len(cic) else float("nan")
                righe.append(dict(gate=nm, forte=forte, tenor=tenor, dp=dp, sh=sh, dd=dd, drift=dr,
                                  kf=kf, drift_kf=dr_kf, licenza=lic, kn=kn, drift_null=dr_n, m5=m5,
                                  pegg7=peggior_finestra(nav, 7), pegg14=peggior_finestra(nav, 14),
                                  para=para, ret=r, nav=nav))
                print(f"     {tenor:>4}{dp:>6.2f}{sh:>8.3f}{100*dd:>7.2f}%{100*dr:>8.2f}% | "
                      f"{kf:>6.3f}{100*dr_kf:>10.2f}%{('VINCE' if lic else 'perde'):>9} | "
                      f"{kn:>7.3f}{100*dr_n:>10.2f}%{('V' if m5 else 'p'):>4} | "
                      f"{100*peggior_finestra(nav,7):>6.1f}%{100*peggior_finestra(nav,14):>6.1f}%{para:>8.1f}%")

    R = pd.DataFrame(righe)
    n_lic = int(R["licenza"].sum())
    print("\n" + "=" * 96)
    print(f"  B1  il pavimento a premio reale VINCE il test della LICENZA in {n_lic}/{len(R)} celle "
          f"({'REFUTATA' if n_lic > len(R)//2 else 'CONFERMATA'})")
    print(f"      (e il null M5 classico in {int(R['m5'].sum())}/{len(R)}: le due letture "
          f"{'concordano' if abs(n_lic-int(R['m5'].sum()))<=2 else 'DIVERGONO — va capito perche'}) ")

    # ---- B2: da dove viene la riduzione di DD? crolli veloci o grind? ------------------------
    print("\n  B2  MECCANISMO — la put copre UNA finestra: protegge i crolli veloci, non i grind.")
    for forte in (True, False):
        dirs = C.gate_trend(P, forte)
        nm = "FORTE" if forte else "LARGO"
        base = C.collar(P, cal, dput=0.2, dcall=0.2, tenor=7, dirs=dirs, senza_opzioni=True)
        nb = base["nav"]
        best = R[(R["forte"] == forte)].sort_values("sh", ascending=False).iloc[0]
        nf = best["nav"]
        # il maxDD: quando, e quanto e' durato
        for lbl, nav in (("base", nb), ("pavimento", nf)):
            pk = nav.cummax()
            ddser = nav / pk - 1.0
            i_min = ddser.idxmin()
            i_pk = nav[:i_min].idxmax()
            durata = (i_min - i_pk).days
            print(f"     {nm} {lbl:>9}: maxDD {100*ddser.min():.2f}% da {i_pk.date()} a {i_min.date()} "
                  f"— {durata} giorni ({'GRIND' if durata > 3*best['tenor'] else 'crollo dentro ~1 finestra'})")
        # perdita "dentro finestra" contro "fra finestre": quanto del DD e' coperto dalla put
        pegg_b = peggior_finestra(nb, best["tenor"])
        pegg_f = peggior_finestra(nf, best["tenor"])
        print(f"     {nm} peggior finestra di {best['tenor']}g: base {100*pegg_b:.1f}% -> "
              f"pavimento {100*pegg_f:.1f}% (la put fa QUESTO); il maxDD invece "
              f"{100*C._dd(nb/nb.shift(1)-1):.1f}% -> {100*best['dd']:.1f}% (i grind restano)")

    # ---- B3: l'invariante del disaster-SL con un pavimento: licenza di carta ------------------
    print("\n  B3  L'INVARIANTE DEL DISASTER-SL con un pavimento — licenza di CARTA.")
    frac, sl = 0.50, 0.30
    k_inv = 0.50 / (2 * frac * sl)
    print(f"     oggi: n_asset 2 x frac {frac} x scala x sl {sl} <= 0,50  =>  scala <= {k_inv:.2f}x")
    for forte in (True,):
        best = R[R["forte"] == forte].sort_values("sh", ascending=False).iloc[0]
        cic = pd.DataFrame(best["cicli"] if "cicli" in best else [])
        dirs = C.gate_trend(P, forte)
        o = C.collar(P, cal, dput=best["dp"], dcall=0.2, tenor=best["tenor"], dirs=dirs, solo_pavimento=True)
        cic = pd.DataFrame(o["cicli"])
        # distanza mediana del pavimento dallo spot all'apertura
        cic["dist"] = (cic["S0"] - cic["Kl"]) / cic["S0"] * cic["dirn"]
        dist = float(cic["dist"].abs().median())
        k_carta = 0.50 / (2 * frac * dist)
        print(f"     con la put a d{best['dp']:.2f}/{best['tenor']}g il pavimento sta a {100*dist:.1f}% "
              f"dallo spot: sostituire sl con la distanza darebbe scala <= {k_carta:.2f}x ({k_carta/k_inv:.1f}x in piu')")
        # ma la perdita su finestre CONSECUTIVE non e' limitata: la misuro
        nf = best["nav"]
        for n_fin in (1, 2, 4, 8):
            g = n_fin * best["tenor"]
            print(f"       peggior perdita su {n_fin} finestre consecutive ({g:>3}g): "
                  f"{100*peggior_finestra(nf, g):.1f}%  "
                  f"{'<= una finestra: la put copre' if n_fin == 1 else '-> la put NON copre: e'' un grind'}")
        print(f"     ⇒ la licenza {k_carta:.2f}x vale per UN crollo dentro UNA finestra; su un grind di "
              f"{8*best['tenor']} giorni la perdita e' {100*abs(peggior_finestra(nf, 8*best['tenor'])):.0f}%, "
              f"e a {k_carta:.1f}x farebbe {100*k_carta*abs(peggior_finestra(nf, 8*best['tenor'])):.0f}% dell'equity.")

    # ---- B4: pavimento GATED su IV a buon mercato -------------------------------------------
    print("\n  B4  PAVIMENTO GATED sull'IV — compro la put solo quando il premio di varianza e' sottile.")
    print("     (§46: 'nessuna copertura dinamica gated su regime e'' stata provata' — qui lo e'.)")
    lr = np.log(P["px"] / P["px"].shift(1))
    rv30 = (lr.rolling(30).std() * math.sqrt(DPY))
    ratio = (P["iv"] / rv30)                       # DVOL / RV trailing: quanto e' cara la vol OGGI
    pct = ratio.rolling(365, min_periods=90).rank(pct=True)
    for forte in (True, False):
        dirs = C.gate_trend(P, forte)
        nm = "FORTE" if forte else "LARGO"
        base = C.collar(P, cal, dput=0.2, dcall=0.2, tenor=7, dirs=dirs, senza_opzioni=True)
        rb = base["ret"]; dd_b, dr_b = C._dd(rb), C._drift(rb)
        best = R[R["forte"] == forte].sort_values("sh", ascending=False).iloc[0]
        for soglia in (0.25, 0.50):
            # NOTA: il gate di IV si applica alla PUT, non alla posizione. Lo simulo con due
            # strutture (protetta / nuda) e le incollo giorno per giorno sul segnale di IV, pagando
            # lo spread alle transizioni tramite il motore stesso (ricalcolo con dirs modulati).
            economico = (pct.shift(1) <= soglia).reindex(P.index).fillna(False).values
            r_prot = best["ret"].reindex(P.index)
            r_nuda = rb.reindex(P.index)
            r_mix = pd.Series(np.where(economico, r_prot, r_nuda), index=P.index).dropna()
            dd, dr = C._dd(r_mix), C._drift(r_mix)
            kf = k_up_for_same_dd(r_mix.values, dd_b)
            dr_kf = C._drift(kf * r_mix.values)
            quota = 100.0 * np.mean(economico[np.asarray(dirs) != 0]) if np.any(np.asarray(dirs) != 0) else 0.0
            print(f"     {nm} put ON se DVOL/RV sotto il {100*soglia:.0f}° pctl (ON il {quota:.0f}% dei giorni a mercato): "
                  f"Sh {C._sh(r_mix):.3f} DD {100*dd:.2f}% drift {100*dr:+.2f}% | licenza k_f {kf:.3f} "
                  f"-> drift {100*dr_kf:+.2f}% vs base {100*dr_b:+.2f}% {'VINCE' if dr_kf > dr_b else 'perde'}")
    print("     ⚠️ approssimazione dichiarata: l'incollatura ignora lo spread delle transizioni ON/OFF "
          "della put (a favore del gated). Se perde COSI', perde a maggior ragione.")

    # ---- IL PREMIO DI PAREGGIO: quanto dovrebbe costare la put perche' il pavimento protegga? --
    print("\n  PREMIO DI PAREGGIO — a quale frazione del premio REALE il pavimento pareggia il maxDD nudo.")
    print("  (Il DVOL sta 1,32x sopra la RV-forward: un prezzo 'equo' sarebbe ~76% del reale.)")
    pareggi = []
    for forte in (True, False):
        dirs = C.gate_trend(P, forte)
        nm = "FORTE" if forte else "LARGO"
        base = C.collar(P, cal, dput=0.2, dcall=0.2, tenor=7, dirs=dirs, senza_opzioni=True)["ret"]
        ddb = C._dd(base)
        for tenor, dp in ((14, 0.05), (14, 0.10), (7, 0.10), (14, 0.20)):
            lo, hi = 0.0, 1.0
            for _ in range(25):
                md = 0.5 * (lo + hi)
                r = C.collar(P, cal, dput=dp, dcall=0.2, tenor=tenor, dirs=dirs,
                             solo_pavimento=True, prem_scale=md)["ret"]
                if C._dd(r) < ddb:
                    lo = md
                else:
                    hi = md
            pareggi.append(0.5 * (lo + hi))
            print(f"     {nm} d{dp:.2f}/{tenor}g: pareggio del maxDD a **{100*0.5*(lo+hi):.0f}%** del premio reale")
    print(f"  ⇒ banda {100*min(pareggi):.0f}-{100*max(pareggi):.0f}%: il mercato dovrebbe scontare la put del "
          f"{100*(1-max(pareggi)):.0f}-{100*(1-min(pareggi)):.0f}% perche' il pavimento INIZI a proteggere.")

    # ---- VERDETTO A RUNTIME -----------------------------------------------------------------
    print("\n" + "=" * 96)
    if n_lic == 0:
        v = ("IL PAVIMENTO NON LICENZIA TAGLIA A PREMIO REALE: 0 celle. Cio' che il de-levering non "
             "puo' comprare, la put lo vende a un prezzo che nessuna cella ripaga")
    elif n_lic <= len(R) // 4:
        v = f"MARGINALE: {n_lic}/{len(R)} celle — dentro il rumore di griglia, e sul bordo"
    else:
        v = f"IL PAVIMENTO LICENZIA TAGLIA in {n_lic}/{len(R)} celle — serve M2/DSR e il null di permutazione"
    print(f"  VERDETTO (calcolato a runtime): {v}")
    print("=" * 96)


if __name__ == "__main__":
    main()