PythagorasGoal/scripts/research/r0726_venue_response.py

#!/usr/bin/env python
"""r0726_venue_response.py — quanto costa RISPONDERE a un allarme di venue, e quanto salva.

Un tripwire senza il costo della risposta non e' un sistema: e' un allarme che non si sa se
valga la pena ascoltare. Qui si misura il lato economico, che e' quello che decide.

DUE NUMERI, e sono asimmetrici di tre ordini di grandezza:

  (A) COSTO DI UN FALSO ALLARME — l'allarme scatta, si va flat e si esce dal venue per N giorni,
      poi si rientra. Misurato sul book REALE (TP01+SKH01 75/25) forzando flat una finestra di N
      giorni a OGNI possibile data d'inizio: la distribuzione di quel costo e' la risposta.
      ⚠️ Non e' un costo medio-zero: uscire e rientrare paga fee, e soprattutto il book e' un
      trend-follower — uscire durante un trend costa il trend, non l'interesse di N giorni.

  (B) VALORE DI UN VERO POSITIVO — quanto del saldo si salva uscendo con `lead_hours` di anticipo,
      dato che il prelievo non e' istantaneo (`withdraw_hours`). Se il lead e' minore del tempo di
      prelievo, il tripwire non serve a niente: e' il vincolo che decide se il sistema esiste.

  ⚠️ `withdraw_hours` NON e' stimato da me: e' il parametro che l'operatore deve misurare con un
  prelievo di prova (vedi runbook). Qui se ne mostra la SENSIBILITA', come per `p` nel venue-risk.

    uv run python scripts/research/r0726_venue_response.py
"""
from __future__ import annotations

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"))

from src.portfolio.portfolio import combine_outer, metrics  # noqa: E402
from src.portfolio.sleeves import deribit_book_sleeves      # noqa: E402

EXIT_DAYS = (1, 2, 3, 5, 7, 14, 30)
# Ore di anticipo del segnale sul blocco definitivo dei prelievi. Ancore storiche:
# FTX ~72h fra il primo blocco visibile e la bancarotta; Mt.Gox mesi; QuadrigaCX settimane.
LEAD_HOURS = (6, 12, 24, 72, 168, 720)
WITHDRAW_HOURS = (1, 6, 12, 24, 48)


def book_series() -> pd.Series:
    """Il book Deribit REALE (TP01 0.75 + SKH01 0.25), quello che sta sul conto a rischio."""
    sl = deribit_book_sleeves()
    return combine_outer({s.name: s.daily() for s in sl}, {s.name: s.weight for s in sl})


def cost_of_exit(r: pd.Series, days: int, fee_rt: float = 0.001) -> np.ndarray:
    """Costo (in % di equity finale) di andare flat per `days` giorni a ogni data d'inizio.

    Il confronto e' fra la stessa serie con e senza il buco: cio' che si perde e' il RITORNO
    COMPOSTO della finestra, piu' un round-trip di fee per uscire e rientrare. Un valore NEGATIVO
    e' un costo, POSITIVO significa che il falso allarme ha per caso evitato una perdita.
    """
    v = r.values.astype(float)
    n = len(v)
    if n <= days:
        return np.array([])
    out = np.empty(n - days)
    for i in range(n - days):
        window = np.prod(1.0 + v[i:i + days])       # cio' che il book avrebbe fatto
        out[i] = (1.0 / window) * (1.0 - fee_rt) - 1.0   # saltarla, piu' il round-trip
    return out


def salvage(lead_h: float, withdraw_h: float) -> float:
    """Frazione del saldo che esce prima del blocco.

    Modello dichiarato: il prelievo e' un processo con un tempo fisso `withdraw_h`; se il segnale
    arriva con `lead_h` di anticipo si riesce a completare `floor(lead/withdraw)` prelievi, e ogni
    prelievo puo' portare via al piu' TUTTO il saldo (Deribit non impone un cap giornaliero
    rilevante a questa taglia). Quindi il salvataggio e' binario nella pratica:
      lead >= withdraw -> 100%, altrimenti 0%.
    Non e' una semplificazione pigra: e' la forma vera del problema a $600-20k, dove il saldo entra
    in un singolo prelievo. A taglie dove servono N prelievi il modello va rifatto lineare.
    """
    return 1.0 if lead_h >= withdraw_h else 0.0


def main() -> None:
    print("=" * 100)
    print("  r0726 — IL LATO RISPOSTA: quanto costa un falso allarme, quanto salva un vero")
    print("=" * 100)

    r = book_series()
    m = metrics(r)
    print(f"\n  book Deribit reale: {len(r):,} giorni  Sharpe {m['sharpe']:.2f}  "
          f"CAGR {m['cagr']*100:.1f}%  maxDD {m['maxdd']*100:.1f}%")

    # ------------------------------------------------------- (A) falso allarme
    print("\n  (A) COSTO DI UN FALSO ALLARME — flat per N giorni, a ogni data d'inizio possibile")
    print(f"\n  {'giorni':>7}{'costo medio':>14}{'mediana':>10}{'p5 (peggio)':>14}"
          f"{'p95 (meglio)':>14}{'% dannosi':>11}")
    costs = {}
    for d in EXIT_DAYS:
        c = cost_of_exit(r, d) * 100
        costs[d] = c
        print(f"  {d:>7}{c.mean():>13.3f}%{np.median(c):>9.3f}%{np.percentile(c, 5):>13.3f}%"
              f"{np.percentile(c, 95):>13.3f}%{(c < 0).mean()*100:>10.1f}%")

    print("\n    Lettura: il costo MEDIO e' piccolo (il book e' flat o quasi buona parte del")
    print("    tempo), ma la CODA non lo e' — uscire durante un trend costa il trend. Il p5 e' il")
    print("    numero da usare per decidere, non la media.")

    # ------------------------------------------------------- (B) vero positivo
    print("\n  (B) VALORE DI UN VERO POSITIVO — serve che il segnale anticipi il tempo di prelievo")
    print(f"\n  {'lead segnale':>14} |" + "".join(f"{w:>6}h" for w in WITHDRAW_HOURS))
    for lh in LEAD_HOURS:
        lab = f"{lh}h" if lh < 48 else f"{lh//24}g"
        cells = "".join(f"{'SALVO' if salvage(lh, w) else '  --':>7}" for w in WITHDRAW_HOURS)
        print(f"  {lab:>14} |{cells}")
    print("\n    colonne = quanto ci mette un prelievo a completarsi (DA MISURARE con un prelievo")
    print("    di prova: e' l'unico parametro che decide se il sistema funziona).")

    # ------------------------------------------------------- asimmetria
    print("\n  (C) L'ASIMMETRIA — e' questa che decide se il tripwire vale la pena")
    d_ref = 3
    c = costs[d_ref]
    mean_cost = abs(c.mean())          # per il VALORE ATTESO si usa la MEDIA...
    tail_cost = abs(np.percentile(c, 5))   # ...il p5 dice quanto puo' fare male UNA volta
    print(f"\n    costo ATTESO di un falso allarme (uscita 3g, media):   {mean_cost:.3f}% equity")
    print(f"    costo di CODA dello stesso (p5, quanto puo' bruciare): {tail_cost:.3f}% equity")
    print(f"    perdita evitata da un vero positivo:                  100.000% equity")
    print(f"\n    ⚠️ Il break-even si calcola sulla MEDIA, non sulla coda: e' un valore atteso")
    print(f"       ripetuto nel tempo, non una scommessa singola. (Con la coda il numero esce")
    print(f"       {tail_cost/mean_cost:.1f}x piu' severo e la conclusione si ribalta.)")
    print(f"\n    Il tripwire conviene se  p_annua > (falsi allarmi/anno) x {mean_cost/100:.5f}:")
    rows = []
    for fp_year in (0.125, 1, 4, 12, 52):
        be = fp_year * mean_cost / 100.0
        lab = "1 ogni 8 anni" if fp_year < 1 else f"{fp_year:g}/anno"
        rows.append((lab, be))
        print(f"      {lab:>16}  ->  serve p > {be*100:>7.3f}%")
    # verdetto CALCOLATO, non pre-scritto: confronto col ventaglio di `p` del venue-risk
    P_GRID = (0.005, 0.01, 0.02, 0.05)
    print(f"\n    Confronto col ventaglio di `p` di r0726_venue_risk ({', '.join(f'{p:.1%}' for p in P_GRID)}):")
    for lab, be in rows:
        ok = [f"{p:.1%}" for p in P_GRID if p > be]
        verdict = f"conviene per p in {{{', '.join(ok)}}}" if ok else "NON conviene a nessun p del ventaglio"
        print(f"      {lab:>16}  {verdict}")
    print(f"\n    Il tasso di falsi allarmi misurato (r0726_venue_tripwire) e' ZERO su 8 anni:")
    print(f"    anche contando 1 ogni 8 anni come limite superiore, il break-even e'")
    print(f"    p > {rows[0][1]*100:.3f}% — un ordine di grandezza sotto la piu' ottimistica")
    print(f"    delle ipotesi (0.5%). A 12 falsi allarmi/anno invece NON converrebbe: il")
    print(f"    valore del sistema sta tutto nella SPECIFICITA', non nella sensibilita'.")

    print("\n  ⚠️ CIO' CHE QUESTO NON DICE: che il segnale ARRIVI. L'asimmetria rende conveniente")
    print("     ASCOLTARE l'allarme, non garantisce che l'allarme suoni prima del blocco. Un")
    print("     fallimento improvviso (furto di chiavi, sequestro) non ha finestra e nessun")
    print("     tripwire lo prende: quella parte di `p` resta scoperta e va detta.")


if __name__ == "__main__":
    main()