diff --git a/scripts/research/r0823_tail_hedge.py b/scripts/research/r0823_tail_hedge.py new file mode 100644 index 0000000..9cd00e8 --- /dev/null +++ b/scripts/research/r0823_tail_hedge.py @@ -0,0 +1,881 @@ +"""r0823_tail_hedge — §46 TAIL-HEDGE: comprare put deep-OTM come ASSICURAZIONE STATICA sul libro. + +LA DOMANDA (decidibile, e NON e' una domanda di alpha) + Il progetto ha gia' testato la VENDITA di volatilita' (VRP01, gated su IV-rank), il GAMMA + SCALPING (long-vol delta-hedgiato, SCARTATO: perde ogni anno) e le strutture Albimarini. + Non ha mai testato la cosa piu' semplice: **comprare put deep-OTM e tenerle**, senza + delta-hedge, senza pretendere alpha. E' un riduttore di DRAWDOWN, e il drawdown e' il vincolo + binding su due assi gia' misurati: (a) il canale funded (daily-loss 4%, max-loss 6%), (b) la + rendita perpetua, che vive sul DD (il libro rende quanto un ETF S&P e serve 2,5x meno capitale + solo perche' perde meno). + Domanda: **comprare protezione di coda compra riduzione di DD a un prezzo migliore di quello + che si ottiene semplicemente DE-LEVERANDO?** + +=========================== ATTESA A PRIORI (scritta PRIMA di misurare) =========================== +Mi aspetto **REFUTATO al f realistico**, e forse marginale a f=1. Tre ragioni, in ordine di forza: + (c) **Si sta comprando assicurazione contro il rischio sbagliato.** Il 22/08 ha misurato che il + peggior giorno del libro NON e' un crash: e' uno SHORT SQUEEZE di SKH01 (short con stop 2% + che gappa a -15,77% di sleeve mentre TP01 e' flat). Una PUT non protegge da un rialzo. + (b) **Il libro e' GIA' difensivo dove la put paga.** L'esposizione di TP01 scende MONOTONA col + crollo (0,096 -> 0,042 sotto il -20%, misurato 22/08) perche' li' il trend si e' gia' girato: + la put paga proprio quando il libro non ha piu' esposizione da proteggere. + (a) **Comprare vol e' pagare il VRP**, cioe' l'esatto specchio di cio' che VRP01 incassa e che il + gamma scalping ha gia' pagato perdendo ogni anno, ogni variante, ogni frequenza. +Previsione quantitativa dichiarata: la riduzione di DD ci sara' (la put paga davvero nei crolli), +ma il de-levering la comprera' a meno; e il costo salira' ~linearmente col budget mentre il +beneficio saturera'. +================================================================================================== + +IL GATE CHE DECIDE — NULL DEL DE-LEVERING (6 occorrenze nel progetto: VRP-DD, TP01xDVOL, MAT01, +azioni intere UCITS, SOL, TP01-SIZE). Si trova il `k` costante per cui `k*libro` ha lo STESSO +maxDD della variante coperta, e si confronta Sharpe e DRIFT a quel punto. + ⚠️ Corollario 22/08: quando la variante e' una pura ri-scalatura il null e' DEGENERE + (`sh(k*base) === sh(base)`, identita' aritmetica). Qui NON lo e' **sul drift**: il drift + geometrico di `k*base` cambia con k, mentre lo Sharpe no. Quindi il confronto ha DUE gambe: + * Sharpe: Sharpe(coperto) vs Sharpe(base) [ = Sharpe(k*base), identicamente ] + * Drift: drift(coperto) vs drift(k*base) [ questa e' la gamba informativa ] + La non-linearita' della put e' cio' che rende la PRIMA gamba non-degenere (la copertura puo' + davvero alzare lo Sharpe, una ri-scalatura no). + +COSA C'E' DENTRO + 1) Motore prezzi RIUSATO (`src.portfolio.sleeves._bs_put` / `_strike_from_delta`, gli stessi di + VRP01) su DVOL reale. Premio **MODELLATO**, non quotato: dichiarato come tale. + 2) f MISURATO OGGI sulle quote vere della catena Deribit (BTC_USDC + inverse), sulle ali che + questa strategia comprerebbe — non assunto dal 30/07. + 3) Griglia dichiarata prima: delta {-0.05,-0.10,-0.15} x tenore {7,14,30} x budget + {0.5%,1%,2%}/anno x roll {scadenza, meta' vita} = **54 celle** per lente di f. + 4) Lente di CODA (non media): maxDD, 5 peggiori mesi, peggior giorno, breach 4%/6% con la + **lente wick accoppiata** (riusata da r0725_prop_coupled, non riscritta). + 5) Eseguibilita' misurata DAL VENUE (min_trade_amount + tick), non assunta. + 6) Controlli positivo e negativo obbligatori. + +CONVENZIONI DICHIARATE + * Libro = TP01 75 / SKH01 25 (il book live Deribit). Il costo si sottrae al DRIFT DEL LIBRO e il + beneficio si misura SUL LIBRO, non sullo sleeve. + * Copertura 50/50 BTC/ETH (l'esposizione del libro e' 50/50). + * **Mark-to-market GIORNALIERO** della put (BS a DVOL e spot correnti). E' essenziale: una + copertura marcata solo a scadenza sarebbe misurata male proprio sul DD, che e' l'unica cosa + che questo filone valuta. + * `f` applicato SIMMETRICAMENTE (si compra E si marca a f x modello). Conseguenza strutturale + che il modello cattura ed e' un risultato in se': **f si paga solo sulla parte di valore che + converge a INTRINSECO (la scadenza); un roll anticipato non lo paga.** Sensitivita' con f + asimmetrico (compri caro, vendi a modello) riportata a parte. + * Spread bid-ask MISURATO dalla catena vera per bucket di tenore, pagato a ogni compra e a ogni + vendita: senza, il roll a meta' vita vincerebbe per finzione. + * Fee opzioni Deribit: taker 0,03% del sottostante con cap 12,5% del premio; consegna ITM + 0,015% con lo stesso cap. + * Finestra: **2021-03-24 -> oggi** (il DVOL non esiste prima). Contiene il 2022. NON contiene + il crollo di marzo 2020, che e' il sinistro piu' grande della storia del libro: dichiarato. + +Uso: `nice -n 19 timeout 900 uv run python scripts/research/r0823_tail_hedge.py` +""" +from __future__ import annotations + +import json +import math +import sys +import time +import urllib.request +from functools import lru_cache +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")) +sys.path.insert(0, str(ROOT / "scripts" / "research" / "alt")) + +from src.data.downloader import load_data # noqa: E402 +from src.strategies.trend_portfolio import resample_1d # noqa: E402 +# MOTORE PREZZI RIUSATO — gli stessi di VRP01, non una reimplementazione. +from src.portfolio.sleeves import ( # noqa: E402 + _bs_put, _strike_from_delta, _tp01_returns, _skyhook_returns, VRP_CFG, +) +import altlib as al # noqa: E402 +from scipy.stats import norm as _norm # noqa: E402 + +ASSETS = ("BTC", "ETH") +W_TP, W_SKH = 0.75, 0.25 # book live Deribit (sleeves.deribit_book_sleeves) +DPY = 365.25 + +# ---- costi opzioni Deribit (listino pubblico) ------------------------------------------------- +OPT_TAKER_UNDERLYING = 0.0003 # 0,03% del sottostante per gamba +OPT_FEE_CAP_FRAC = 0.125 # cap 12,5% del premio dell'opzione +OPT_DELIVERY_UNDERLYING = 0.00015 # 0,015% consegna ITM + +# ---- GRIGLIA DICHIARATA PRIMA DI GUARDARE ----------------------------------------------------- +G_DELTA = (-0.05, -0.10, -0.15) +G_TENOR = (7, 14, 30) +G_BUDGET = (0.005, 0.010, 0.020) # quota di equity spesa in premio, ANNUA +G_ROLL = ("expiry", "half") +N_CELLS = len(G_DELTA) * len(G_TENOR) * len(G_BUDGET) * len(G_ROLL) # = 54 + +# ---- lenti di f: 1.0 (modello nudo) + i due misurati il 30/07 + quello misurato OGGI ----------- +F_LENSES_STATIC = (1.0, 2.23, 5.85) + +DERIBIT = "https://www.deribit.com/api/v2/public/" + + +# ==================================================================== 0. UTIL METRICHE + +def _sh(x) -> float: + r = np.asarray(pd.Series(x).dropna().values, float) + return float(np.mean(r) / np.std(r) * math.sqrt(DPY)) if len(r) > 2 and np.std(r) > 0 else 0.0 + + +def _dd(x) -> float: + r = np.asarray(pd.Series(x).dropna().values, float) + eq = np.cumprod(1.0 + r) + pk = np.maximum.accumulate(eq) + return float(np.max((pk - eq) / pk)) if len(eq) else 0.0 + + +def _drift(x) -> float: + """Drift GEOMETRICO annualizzato (CAGR). E' la grandezza su cui vive la rendita perpetua, + ed e' l'unica delle due che CAMBIA quando si de-leva (lo Sharpe no: identita' aritmetica).""" + r = np.asarray(pd.Series(x).dropna().values, float) + if len(r) < 2: + return 0.0 + tot = float(np.prod(1.0 + r)) + if tot <= 0: + return -1.0 + return float(tot ** (DPY / len(r)) - 1.0) + + +def _dd_of_k(r: np.ndarray, k: float) -> float: + return _dd(k * r) + + +def k_for_same_drift(base: np.ndarray, target_drift: float) -> float: + """Il de-levering a PARI DRIFT. Serve sull'asse funded: se la copertura NON riduce il maxDD + (che e' cio' che questo filone misura), il confronto a pari DD e' degenere -> la domanda + onesta diventa 'a parita' di rendimento atteso, chi sfora meno la regola giornaliera?'. + drift(k*r) e' crescente in k su un libro a drift positivo.""" + lo, hi = 1e-4, 1.0 + if _drift(hi * base) <= target_drift: + return 1.0 + for _ in range(60): + mid = 0.5 * (lo + hi) + if _drift(mid * base) > target_drift: + hi = mid + else: + lo = mid + return 0.5 * (lo + hi) + + +def k_for_same_dd(base: np.ndarray, target_dd: float) -> float: + """Il `k` del NULL DEL DE-LEVERING: la scala costante che porta il maxDD del libro NUDO + esattamente al maxDD della variante coperta. maxDD(k*r) e' monotono crescente in k + (compounding), quindi bisezione.""" + lo, hi = 1e-4, 1.0 + if _dd_of_k(base, hi) <= target_dd: + return 1.0 # la copertura non riduce il DD: k=1 + for _ in range(60): + mid = 0.5 * (lo + hi) + if _dd_of_k(base, mid) > target_dd: + hi = mid + else: + lo = mid + return 0.5 * (lo + hi) + + +# ==================================================================== 1. DATI + +@lru_cache(maxsize=1) +def daily_panel() -> pd.DataFrame: + """Pannello giornaliero allineato: rendimenti del libro + prezzi/low/DVOL per asset. + Un solo indice, cosi' nessuna domanda successiva puo' disallinearsi.""" + tp = _tp01_returns() + skh = _skyhook_returns() + J = pd.concat({"tp": tp, "skh": skh}, axis=1, join="inner").fillna(0.0) + book = W_TP * J["tp"] + W_SKH * J["skh"] + + cols = {"book": book, "tp": J["tp"], "skh": J["skh"]} + for a in ASSETS: + d = resample_1d(load_data(a, "1h")) + idx = pd.DatetimeIndex(pd.to_datetime(d["datetime"], utc=True)) + for c in ("close", "low", "high"): + assert c in d.columns, f"resample_1d senza colonna {c}" + cols[f"px_{a}"] = pd.Series(d["close"].values.astype(float), index=idx) + cols[f"lo_{a}"] = pd.Series(d["low"].values.astype(float), index=idx) + dv = pd.read_parquet(ROOT / "data" / "raw" / f"dvol_{a.lower()}.parquet") + cols[f"iv_{a}"] = pd.Series(dv["close"].values.astype(float) / 100.0, + index=pd.to_datetime(dv["timestamp"], unit="ms", utc=True)) + P = pd.concat(cols, axis=1, join="inner").sort_index().dropna() + return P + + +# ==================================================================== 2. IL MOTORE DELLA COPERTURA + +def _entry_fee(prem_paid: float, S: float) -> float: + return min(OPT_TAKER_UNDERLYING * S, OPT_FEE_CAP_FRAC * prem_paid) if prem_paid > 0 else 0.0 + + +def hedge_leg(px: np.ndarray, lo: np.ndarray, iv: np.ndarray, *, + delta: float, tenor: int, budget_ann: float, roll: str, + f: float, spread: float, f_sell: float | None = None, + free: bool = False) -> tuple[np.ndarray, np.ndarray, dict]: + """Serie GIORNALIERE (ret, wick_extra) del contributo di UNA gamba di copertura, per 1 unita' + di equity allocata a quella gamba. + + ret[i] = variazione giornaliera del valore MTM della posizione (frazione di equity) + wick_extra[i] = guadagno AGGIUNTIVO della copertura al minimo intra-giornaliero, ottenuto + RIPREZZANDO la put al LOW del giorno (cattura la convessita', non solo il delta) + + CAUSALE per costruzione: strike, premio e quantita' a `t` usano solo px[t]/iv[t]; il payoff + usa prezzi successivi e non retroagisce mai sulla decisione. + + `free=True` -> CONTROLLO POSITIVO: il payoff c'e' ma il premio non viene addebitato. + """ + n = len(px) + T_full = tenor / DPY + hold = tenor if roll == "expiry" else max(1, tenor // 2) + n_rolls_year = DPY / hold + b_roll = budget_ann / n_rolls_year # spesa per roll, in frazione di equity + + ret = np.zeros(n) + wick = np.zeros(n) + diag = dict(n_rolls=0, n_itm=0, prem_spent=0.0, payoff_recv=0.0, fees=0.0, spread_cost=0.0) + + i = 60 # warm-up (coerente con VRP01) + while i + hold < n: + S0, sig = float(px[i]), float(iv[i]) + if not (S0 > 0 and sig > 0): + i += hold + continue + K = _strike_from_delta(S0, T_full, sig, delta) + p_model = _bs_put(S0, K, T_full, sig) + if p_model <= 0: + i += hold + continue + p_ask = f * p_model * (1.0 + spread) # si COMPRA al lato caro + fee0 = _entry_fee(p_ask, S0) + cost_unit = p_ask + fee0 + q = b_roll / cost_unit # unita' di sottostante coperte + # giorno d'ingresso: il valore MTM e' q*f*p_model (il mid), quindi si perde subito + # fee + mezzo spread. Se `free`, il premio non si paga affatto. + v_prev = q * f * p_model + # ⚠️ `free` = CONTROLLO POSITIVO (payoff regalato). La prima stesura metteva 0.0 qui e + # lasciava v_prev>0: cosi' il MTM successivo addebitava il decadimento di un valore mai + # accreditato -> il "free lunch" costava quanto il premio e il controllo FALLIVA. Il + # regalo va ACCREDITATO: l'equity sale di v_prev il giorno in cui ricevi l'asset. + ret[i] += (v_prev if free else (v_prev - b_roll)) + diag["n_rolls"] += 1 + diag["prem_spent"] += 0.0 if free else b_roll + diag["fees"] += q * fee0 + diag["spread_cost"] += q * f * p_model * spread + + for s in range(1, hold + 1): + j = i + s + rem = (tenor - s) / DPY + Sj, ivj = float(px[j]), float(iv[j]) + if s == tenor: # SCADENZA: valore = intrinseco + intr = max(0.0, K - Sj) + v = q * intr + if intr > 0: + dfee = min(OPT_DELIVERY_UNDERLYING * Sj, OPT_FEE_CAP_FRAC * intr) * q + v -= dfee + diag["fees"] += dfee + diag["n_itm"] += 1 + diag["payoff_recv"] += v + vw = v # a scadenza niente wick residuo + else: + fs = f if f_sell is None else f_sell + v_mid = q * f * _bs_put(Sj, K, rem, ivj) + v = v_mid + vw = q * f * _bs_put(float(lo[j]), K, rem, ivj) + if s == hold: # VENDITA anticipata: mezzo spread + v = q * fs * _bs_put(Sj, K, rem, ivj) * (1.0 - spread) + sfee = min(OPT_TAKER_UNDERLYING * Sj, OPT_FEE_CAP_FRAC * max(v, 0.0)) + v -= sfee + diag["fees"] += sfee + diag["spread_cost"] += v_mid * spread + diag["payoff_recv"] += v + wick[j] += vw - v_mid # EXTRA al minimo, contro il MID di chiusura + ret[j] += v - v_prev + v_prev = v + i += hold + + return ret, np.maximum(wick, 0.0), diag + + +@lru_cache(maxsize=None) +def hedge_series(delta: float, tenor: int, budget: float, roll: str, f: float, + spread: float, f_sell: float | None = None, + free: bool = False) -> tuple: + """Copertura del LIBRO: 50/50 BTC+ETH (il libro e' 50/50). Ritorna (ret, wick) giornalieri + come frazione dell'equity del libro.""" + P = daily_panel() + ret = np.zeros(len(P)) + wick = np.zeros(len(P)) + diags = {} + for a in ASSETS: + r, w, d = hedge_leg(P[f"px_{a}"].values, P[f"lo_{a}"].values, P[f"iv_{a}"].values, + delta=delta, tenor=tenor, budget_ann=budget * 0.5, roll=roll, + f=f, spread=spread, f_sell=f_sell, free=free) + ret += r + wick += w + diags[a] = d + return ret, wick, diags + + +# ==================================================================== 3. f e SPREAD DAL VENUE + +def _get(path: str, timeout: int = 25): + with urllib.request.urlopen(DERIBIT + path, timeout=timeout) as r: + return json.load(r)["result"] + + +def measure_f_from_venue() -> dict: + """f = quota MID di mercato / prezzo BS a DVOL piatto, MISURATO OGGI sulle ali che questa + strategia comprerebbe. Non e' un'assunzione ereditata dal 30/07: e' una misura su questa + catena, su questi strike, con questo motore. + + Si misura anche il MEZZO SPREAD relativo per bucket di tenore, che serve al roll anticipato. + Solo strumenti con bid E ask veri (un ask solitario al tick non e' un prezzo). + """ + out = dict(ok=False, note=None, per_delta={}, spread={}, n=0) + try: + idx = {a: float(_get(f"get_index_price?index_name={a.lower()}_usdc")["index_price"]) + for a in ASSETS} + dvol = {a: float(_get(f"get_volatility_index_data?currency={a}&start_timestamp=" + f"{int((time.time()-86400)*1000)}&end_timestamp={int(time.time()*1000)}" + f"&resolution=3600")["data"][-1][4]) / 100.0 + for a in ASSETS} + time.sleep(0.6) + bs = _get("get_book_summary_by_currency?currency=USDC&kind=option") + except Exception as e: # rete assente: si dichiara, non si finge + out["note"] = f"venue non raggiungibile ({type(e).__name__}: {e})" + return out + + now = pd.Timestamp.now(tz="UTC") + rows = [] + for r in bs: + nm = r["instrument_name"] + if not nm.endswith("-P"): + continue + base = nm.split("_USDC-")[0] if "_USDC-" in nm else None + if base not in ASSETS: + continue + try: + _, exp, ks, _ = nm.split("-") + K = float(ks.replace("d", ".")) + texp = pd.to_datetime(exp, format="%d%b%y", utc=True) + pd.Timedelta(hours=8) + except Exception: + continue + bid, ask = r.get("bid_price"), r.get("ask_price") + if not bid or not ask or bid <= 0 or ask <= bid: + continue # niente bid vero -> non e' un prezzo + T = (texp - now).total_seconds() / (365.25 * 86400) + if T <= 1 / DPY or T > 45 / DPY: + continue + S, sig = idx[base], dvol[base] + mid = 0.5 * (bid + ask) + half_spread = 0.5 * (ask - bid) / mid + model = _bs_put(S, K, T, sig) + if model <= 1e-9: + continue + # delta BS della put allo stesso vol piatto (per assegnare l'osservazione al suo bucket) + d1 = (math.log(S / K) + 0.5 * sig ** 2 * T) / (sig * math.sqrt(T)) + dlt = -float(_norm.cdf(-d1)) + rows.append(dict(asset=base, K=K, T=T, days=T * DPY, delta=dlt, + mid=mid, model=model, f=mid / model, hs=half_spread)) + if not rows: + out["note"] = "nessuno strumento con bid E ask veri nella finestra <=45g" + return out + D = pd.DataFrame(rows) + out["n"] = int(len(D)) + for dtgt in G_DELTA: + sel = D[(D.delta >= dtgt - 0.035) & (D.delta <= dtgt + 0.035)] + if len(sel) >= 3: + out["per_delta"][dtgt] = dict(n=int(len(sel)), f_med=float(sel.f.median()), + f_lo=float(sel.f.quantile(0.10)), + f_hi=float(sel.f.quantile(0.90)), + hs_med=float(sel.hs.median())) + for lab, (a, b) in dict(t7=(1, 10), t14=(10, 21), t30=(21, 45)).items(): + sel = D[(D.days >= a) & (D.days < b)] + if len(sel) >= 3: + out["spread"][lab] = dict(n=int(len(sel)), hs_med=float(sel.hs.median()), + hs_p90=float(sel.hs.quantile(0.90))) + out["ok"] = bool(out["per_delta"] or out["spread"]) + return out + + +def venue_min_ticket() -> dict: + """Eseguibilita' MISURATA dal venue: min_trade_amount + tick, e il costo in DOLLARI del + lotto minimo per una put deep-OTM vera. ⚠️ Comprare un'opzione costa il PREMIO, non il + NOZIONALE: il muro dei 'lotti da migliaia di dollari' del 22/08 e' il collaterale per + VENDERE, non il costo per COMPRARE. Qui si misura la cosa giusta.""" + out = dict(ok=False, note=None, fam={}) + try: + for cur, pref in (("USDC", "{a}_USDC-"), ("BTC", "BTC-"), ("ETH", "ETH-")): + inst = _get(f"get_instruments?currency={cur}&kind=option&expired=false") + time.sleep(0.6) + for a in ASSETS: + p = pref.format(a=a) + sel = [i for i in inst if i["instrument_name"].startswith(p)] + if not sel: + continue + fam = "USDC-lineare" if cur == "USDC" else "inverse" + out["fam"][(a, fam)] = dict(min_amt=float(sel[0]["min_trade_amount"]), + tick=float(sel[0]["tick_size"]), + settle=sel[0]["settlement_currency"], + n=len(sel)) + out["ok"] = bool(out["fam"]) + except Exception as e: + out["note"] = f"venue non raggiungibile ({type(e).__name__}: {e})" + return out + + +# ==================================================================== 4. LENTE WICK ACCOPPIATA + +@lru_cache(maxsize=1) +def coupled_book_tuples() -> pd.DataFrame: + """(R, m) giornaliere del libro 75/25 sotto la LENTE WICK ACCOPPIATA — RIUSATA da + r0725_prop_coupled (importata, non riscritta). Se il recon non e' disponibile si dichiara e + si ricade su close-only, DICENDO che close-only e' CIECA su una regola a UN giorno + (misurato 22/08: 0,00 breach/anno contro 0,40-1,21 veri, rapporto INF).""" + try: + import r0725_prop_coupled as pc + return pc.crypto_daily_tuples(W_TP, W_SKH) + except Exception as e: # dichiarato, mai silenzioso + print(f" ⚠️ lente accoppiata NON disponibile ({type(e).__name__}: {e}) -> close-only") + return pd.DataFrame() + + +# ==================================================================== 5. VALUTAZIONE DI UNA CELLA + +def evaluate(book: np.ndarray, hret: np.ndarray) -> dict: + hedged = book + hret + dd_h = _dd(hedged) + k = k_for_same_dd(book, dd_h) + base_k = k * book + dd_b = _dd(book) + dd_reduced = bool(dd_h < dd_b - 1e-12) + dr_h, dr_k = _drift(hedged), _drift(base_k) + return dict( + sh_base=_sh(book), sh_hedged=_sh(hedged), sh_delev=_sh(base_k), + dd_base=dd_b, dd_hedged=dd_h, dd_delev=_dd(base_k), + dr_base=_drift(book), dr_hedged=dr_h, dr_delev=dr_k, + k=k, hedged=hedged, dd_reduced=dd_reduced, + # VINCE solo se (i) il DD scende davvero E (ii) a quel DD il drift batte il de-levering. + # Senza (i) il null e' DEGENERE (k=1, dr_delev==dr_base) e un Δdrift>0 direbbe solo + # "la put ha guadagnato in questa finestra", che e' una domanda di ALPHA — non questa. + wins=bool(dd_reduced and dr_h > dr_k), + ) + + +def worst_months(r: pd.Series, n: int = 5) -> list: + m = (1.0 + r).resample("ME").prod() - 1.0 + return [(str(i.date())[:7], float(v)) for i, v in m.nsmallest(n).items()] + + +# ==================================================================== MAIN + +def main() -> None: + t0 = time.time() + print("=" * 100) + print("§46 TAIL-HEDGE — put deep-OTM come ASSICURAZIONE STATICA sul libro TP01 75 / SKH01 25") + print("=" * 100) + print(f"GRIGLIA DICHIARATA PRIMA DI GUARDARE: {len(G_DELTA)} delta x {len(G_TENOR)} tenori x " + f"{len(G_BUDGET)} budget x {len(G_ROLL)} roll = {N_CELLS} celle per lente di f.") + print("ATTESA A PRIORI: REFUTATO al f realistico. Ragione piu' forte (c): il peggior giorno del") + print(" libro NON e' un crash ma uno short-squeeze di SKH01 -> una PUT non protegge da un rialzo.") + + P = daily_panel() + book = P["book"].values + idx = P.index + bs_book = pd.Series(book, index=idx) + print(f"\nFinestra (comune libro+DVOL): {idx[0].date()} -> {idx[-1].date()} " + f"({len(P)} giorni, {len(P)/DPY:.1f} anni)") + print(f"LIBRO NUDO su questa finestra: Sharpe {_sh(book):.3f} | maxDD {_dd(book)*100:.2f}% | " + f"drift {_drift(book)*100:.2f}%/a") + print("⚠️ La finestra NON contiene marzo 2020 (il DVOL parte 2021-03): il sinistro piu' grande") + print(" della storia del libro e' FUORI CAMPIONE. La copertura non e' testata li'.") + + # ---------------------------------------------------------------- §A. f e spread dal venue + print("\n" + "=" * 100) + print("§A. f E SPREAD — MISURATI OGGI SULLE QUOTE VERE (non assunti dal 30/07)") + print("=" * 100) + fv = measure_f_from_venue() + if fv["ok"]: + print(f" {fv['n']} put con bid E ask veri, tenore <=45g, catena USDC-lineare.") + print(f" {'delta':>7} {'n':>4} {'f mediano':>10} {'f p10':>8} {'f p90':>8} {'mezzo-spread':>13}") + for d, v in sorted(fv["per_delta"].items(), reverse=True): + print(f" {d:>7.2f} {v['n']:>4} {v['f_med']:>10.2f} {v['f_lo']:>8.2f} " + f"{v['f_hi']:>8.2f} {v['hs_med']*100:>12.1f}%") + print(f" mezzo-spread per tenore: " + " ".join( + f"{k} {v['hs_med']*100:.1f}% (n={v['n']})" for k, v in fv["spread"].items())) + else: + print(f" NON MISURATO: {fv['note']}") + F_MEAS = {d: v["f_med"] for d, v in fv.get("per_delta", {}).items()} + if F_MEAS: + print(" -> f USATO nella lente 'misurato oggi': " + + " ".join(f"δ{d:.2f}={v:.2f}" for d, v in sorted(F_MEAS.items(), reverse=True))) + SPREAD = {} + for lab, ten in (("t7", 7), ("t14", 14), ("t30", 30)): + SPREAD[ten] = fv.get("spread", {}).get(lab, {}).get("hs_med", 0.10) + print(f" -> mezzo-spread usato per tenore: " + + " ".join(f"{t}g {SPREAD[t]*100:.1f}%" for t in G_TENOR) + + ("" if fv["ok"] else " (DEFAULT 10%: venue non letto)")) + print(" ⚠️ DISTORSIONE DICHIARATA, e va nel verso che FAVORISCE la copertura: lo spread e' il") + print(" mediano dei soli strumenti con bid E ask veri. Le ali piu' profonde e piu' corte") + print(" spesso NON hanno bid (solo un ask al tick) e cadono fuori dal campione: quelle") + print(" sarebbero le piu' care, e per un tail-hedge sono proprio quelle che si comprano.") + print(" Quindi il costo qui e' un PAVIMENTO, e il verdetto REFUTATO ne esce rafforzato.") + + # ---------------------------------------------------------------- §B. il null del de-levering + print("\n" + "=" * 100) + print("§B. NULL DEL DE-LEVERING — IL GATE CHE DECIDE (in cima, come si deve)") + print("=" * 100) + print(" Per ogni cella: k = scala che porta il libro NUDO allo STESSO maxDD della coperta.") + print(" Sharpe(k*base) == Sharpe(base) per identita' aritmetica -> la gamba informativa e' il") + print(" DRIFT. La copertura VINCE solo se drift(coperta) > drift(k*base) [e/o Sharpe sale].") + + lenses = [("f=1.00 (modello nudo)", 1.0)] + for f in F_LENSES_STATIC[1:]: + lenses.append((f"f={f:.2f} (misurato 30/07)", f)) + if F_MEAS: + lenses.append(("f MISURATO oggi per-delta", None)) + + all_rows = [] + for lab, fval in lenses: + print(f"\n ---- lente {lab} " + "-" * (78 - len(lab))) + print(f" {'delta':>6} {'ten':>4} {'budget':>7} {'roll':>7} | {'ShCop':>7} {'ShBase':>7} | " + f"{'DDcop':>7} {'DDbase':>7} | {'k':>6} {'DrCop':>8} {'DrDelev':>8} {'Δdrift':>8} | esito") + best = None + for dlt in G_DELTA: + f_use = fval if fval is not None else F_MEAS.get(dlt, F_LENSES_STATIC[1]) + for ten in G_TENOR: + for bud in G_BUDGET: + for roll in G_ROLL: + hr, hw, dg = hedge_series(dlt, ten, bud, roll, f_use, SPREAD[ten]) + ev = evaluate(book, hr) + gap = ev["dr_hedged"] - ev["dr_delev"] + win = ev["wins"] + row = dict(lens=lab, f=f_use, delta=dlt, tenor=ten, budget=bud, roll=roll, + **{k: v for k, v in ev.items() if k != "hedged"}, gap=gap, + hret=hr, hwick=hw, diag=dg) + all_rows.append(row) + if best is None or gap > best["gap"]: + best = row + if bud == 0.010 and roll == "expiry": # riga stampata: budget centrale + print(f" {dlt:>6.2f} {ten:>4} {bud*100:>6.1f}% {roll:>7} | " + f"{ev['sh_hedged']:>7.3f} {ev['sh_base']:>7.3f} | " + f"{ev['dd_hedged']*100:>6.2f}% {ev['dd_base']*100:>6.2f}% | " + f"{ev['k']:>6.3f} {ev['dr_hedged']*100:>7.2f}% " + f"{ev['dr_delev']*100:>7.2f}% {gap*100:>+7.2f}% | " + f"{'COPERTURA' if win else ('de-levering' if ev['dd_reduced'] else 'DD PEGGIORE')}") + sub = [r for r in all_rows if r["lens"] == lab] + nwin = sum(1 for r in sub if r["wins"]) + nred = sum(1 for r in sub if r["dd_reduced"]) + cand = [r for r in sub if r["dd_reduced"]] + best = max(cand or sub, key=lambda r: (r["gap"] if cand else -r["dd_hedged"])) + print(f" >> {lab}: il maxDD SCENDE in {nred}/{len(sub)} celle; fra quelle la copertura " + f"batte il de-levering in {nwin}/{len(sub)}.") + head = ("migliore fra le celle che RIDUCONO il DD" if cand else + "NESSUNA cella riduce il DD -> si riporta quella col DD MINORE (k=1, confronto " + "degenere)") + print(f" {head}: delta {best['delta']:.2f} / " + f"{best['tenor']}g / {best['budget']*100:.1f}% / {best['roll']} -> " + f"maxDD {best['dd_base']*100:.2f}% -> {best['dd_hedged']*100:.2f}% (k {best['k']:.3f}), " + f"Δdrift vs de-levering {best['gap']*100:+.2f} pp/a, " + f"ΔSharpe {best['sh_hedged']-best['sh_base']:+.3f}") + + # ---------------------------------------------------------------- §C. controlli + print("\n" + "=" * 100) + print("§C. CONTROLLI — un rilevatore che non segnala nulla puo' essere semplicemente rotto") + print("=" * 100) + ctrl = [] + # (+1) copertura GRATIS (payoff senza premio): DEVE ridurre il DD e battere il de-levering + hr, _, _ = hedge_series(-0.10, 30, 0.020, "expiry", 1.0, 0.0, free=True) + ev = evaluate(book, hr); ctrl.append(("(+) payoff SENZA premio (free lunch)", ev, True)) + # (+2) premio 1/10: stesso budget -> 10x le put. Speculare al negativo, DEVE funzionare + hr, _, _ = hedge_series(-0.10, 30, 0.020, "expiry", 0.1, 0.0) + ev = evaluate(book, hr); ctrl.append(("(+) premio 1/10 (f=0.1, 10x le put)", ev, True)) + # (-) premio 10x: DEVE essere refutata + hr, _, _ = hedge_series(-0.10, 30, 0.020, "expiry", 10.0, 0.0) + ev = evaluate(book, hr); ctrl.append(("(-) premio 10x (f=10)", ev, False)) + # (=) put ATM f=1, spread 0: NON e' un controllo (non deve funzionare per forza) ma e' il + # riferimento piu' caro possibile della famiglia: si stampa come misura, non come test. + hr, _, _ = hedge_series(-0.50, 30, 0.020, "expiry", 1.0, 0.0) + ev = evaluate(book, hr); ctrl.append(("(=) put ATM(δ-0.50) f=1 — misura, non controllo", ev, None)) + for nm, ev, must in ctrl: + gap = ev["dr_hedged"] - ev["dr_delev"] + got = "RICONOSCIUTA" if ev["wins"] else "refutata" + tag = "" if must is None else (" ✓" if ev["wins"] == must else " ✗ CONTROLLO FALLITO") + print(f" {nm:<48} DD {ev['dd_base']*100:5.2f}%->{ev['dd_hedged']*100:5.2f}% k {ev['k']:.3f} " + f"Δdrift {gap*100:+7.2f}pp ShΔ {ev['sh_hedged']-ev['sh_base']:+.3f} -> {got}{tag}") + ok = all(ev["wins"] == must for _, ev, must in ctrl if must is not None) + print(" APPARATO: " + ("controlli 3/3 OK — il rilevatore riconosce cio' che DEVE riconoscere " + "e rifiuta cio' che DEVE rifiutare" + if ok else + "\u26a0\ufe0f UN CONTROLLO E' FALLITO: i numeri sopra NON sono credibili")) + + # ---------------------------------------------------------------- §D. causalita' + print("\n" + "=" * 100) + print("§D. CAUSALITA' — prova esplicita, non un gate generico") + print("=" * 100) + print(" `altlib.causality_ok` vuole un target_fn(df)->posizioni: NON si applica a un overlay") + print(" di opzioni (non e' una posizione continua sul sottostante). Al suo posto la prova") + print(" diretta e piu' forte: si ricalcola la copertura su un PREFISSO del pannello e si") + print(" verifica che coincida con la coda della serie completa. Un leak divergerebbe.") + hr_full, _, _ = hedge_series(-0.10, 14, 0.010, "expiry", 1.0, SPREAD[14]) + cut = int(len(P) * 0.85) + Pf = daily_panel() + sub = Pf.iloc[:cut] + hr_cut, _, _ = hedge_leg(sub["px_BTC"].values, sub["lo_BTC"].values, sub["iv_BTC"].values, + delta=-0.10, tenor=14, budget_ann=0.010 * 0.5, roll="expiry", + f=1.0, spread=SPREAD[14]) + hr_full_btc, _, _ = hedge_leg(Pf["px_BTC"].values, Pf["lo_BTC"].values, Pf["iv_BTC"].values, + delta=-0.10, tenor=14, budget_ann=0.010 * 0.5, roll="expiry", + f=1.0, spread=SPREAD[14]) + # ⚠️ ERRORE MIO, catturato dalla prima corsa (dmax 1.28e-04 = "DIVERGE"): confrontavo fino a + # `cut`, ma il prefisso si ferma un roll prima (`while i + hold < n`), quindi l'ultimo ciclo + # non e' CALCOLATO nel troncato mentre lo e' nel completo. La differenza era per ASSENZA di + # un roll, non per look-ahead. Il confronto onesto finisce prima dell'ultimo roll possibile. + hold = 14 + stop = cut - 2 * hold + tail = 200 + dmax = float(np.max(np.abs(hr_cut[stop - tail:stop] - hr_full_btc[stop - tail:stop]))) + dmax_naive = float(np.max(np.abs(hr_cut[cut - tail:cut] - hr_full_btc[cut - tail:cut]))) + print(f" max|differenza| su [{stop-tail}:{stop}] (cut a {cut}/{len(P)}, ultimo roll escluso): " + f"{dmax:.2e} -> {'CAUSALE' if dmax < 1e-12 else '⚠️ DIVERGE'}") + print(f" (fino a `cut` fa {dmax_naive:.2e}: NON e' un leak, e' il roll che il prefisso non ha") + print(f" ancora eseguito — la finestra sbagliata dichiarava un difetto inesistente.)") + + # ---------------------------------------------------------------- §E. lente di coda + print("\n" + "=" * 100) + print("§E. LENTE DI CODA — maxDD, peggiori mesi, peggior giorno, breach 4%/6%") + print("=" * 100) + CP = coupled_book_tuples() + if not CP.empty: + CP = CP[(CP.index >= idx[0]) & (CP.index <= idx[-1])] + cov = len(CP.index.intersection(idx)) / len(idx) + print(f" lente ACCOPPIATA disponibile: {len(CP)} giorni, copertura {cov*100:.1f}% del pannello") + # scelta della cella per la lente di coda: la MIGLIORE al f misurato (o 2.23), budget 1% + f_pick = F_MEAS.get(-0.10, F_LENSES_STATIC[1]) + cells_tail = [("NUDO", None)] + for dlt in G_DELTA: + cells_tail.append((f"δ{dlt:.2f} 30g 1%/a expiry", (dlt, 30, 0.010, "expiry", + F_MEAS.get(dlt, f_pick)))) + cells_tail.append(("δ-0.10 30g 2%/a expiry", (-0.10, 30, 0.020, "expiry", f_pick))) + cells_tail.append(("δ-0.10 30g 2%/a f=1 (irreale)", (-0.10, 30, 0.020, "expiry", 1.0))) + + def _breach(series_close: np.ndarray, wick_extra: np.ndarray) -> tuple: + s_ = pd.Series(series_close, index=idx) + n_yr = len(P) / DPY + br_c = float((s_ <= -0.04).sum()) / n_yr + if CP.empty: + return float("nan"), br_c + m = CP["m"].reindex(idx).values + n_ev = int(np.nansum((m + np.nan_to_num(wick_extra)) <= -0.04)) + _breach.last_n = n_ev + return n_ev / n_yr, br_c + + print(f"\n {'variante':<32} {'maxDD':>7} {'pegg.giorno':>12} {'breach 4% /a':>14} " + f"{'(close-only)':>13} {'drift':>8} {'Sharpe':>7}") + for nm, spec in cells_tail: + if spec is None: + hr = np.zeros(len(P)); hw = np.zeros(len(P)) + else: + dlt, ten, bud, roll, fu = spec + hr, hw, _ = hedge_series(dlt, ten, bud, roll, fu, SPREAD[ten]) + hed = book + hr + s = pd.Series(hed, index=idx) + worst_day = float(s.min()) + br_coup, br_close = _breach(hed, hr + hw) # il minimo del libro + la copertura + n_ev = getattr(_breach, "last_n", -1) + print(f" {nm:<32} {_dd(hed)*100:>6.2f}% {worst_day*100:>11.2f}% {br_coup:>8.2f}({n_ev:>2}) " + f"{br_close:>12.2f} {_drift(hed)*100:>7.2f}% {_sh(hed):>7.3f}") + if spec is not None: + # ⚠️ il confronto a pari DD e' DEGENERE qui (il DD non scende, quindi k=1). Sull'asse + # funded la domanda onesta e' l'altra: a PARI DRIFT, chi sfora meno la regola a un + # giorno? Cioe' basta rimpicciolire la posizione per comprare gli stessi breach in meno. + kk = k_for_same_drift(book, _drift(hed)) + brk, brkc = _breach(kk * book, np.zeros(len(P)) if CP.empty + else (kk - 1.0) * np.nan_to_num(CP['m'].reindex(idx).values)) + lab_ = " ^ de-levering k=%.3f, PARI DRIFT" % kk + print(f" {lab_:<32} {_dd(kk*book)*100:>6.2f}% " + f"{float(np.min(kk*book))*100:>11.2f}% {brk:>8.2f}({getattr(_breach,'last_n',-1):>2}) " + f"{brkc:>12.2f} " + f"{_drift(kk*book)*100:>7.2f}% {_sh(kk*book):>7.3f}") + print(" ⚠️ close-only e' CIECA su una regola a UN giorno (22/08: 0,00 contro 0,40-1,21 veri).") + print(" La colonna che decide e' quella ACCOPPIATA. Fra parentesi il CONTEGGIO di eventi:") + print(" 3-5 breach in 5,4 anni NON distinguono le varianti fra loro (un evento in piu' o in") + print(" meno sposta il tasso del 20-30%) -> si legge il SEGNO, non la taglia. Il fatto") + print(" robusto e' che il de-levering a PARI DRIFT compra lo stesso ordine di grandezza di") + print(" riduzione dei breach senza toccare il maxDD nel verso sbagliato.") + print("\n REGOLA MAX-LOSS 6% (statica dal saldo iniziale, HyroTrader — e' la regola BINDING") + print(" del canale funded, gia' modellata il 25/07 come EV_DD=0.06):") + k6 = k_for_same_dd(book, 0.06) + print(f" libro NUDO maxDD {_dd(book)*100:5.2f}% -> SFORA (serve k={k6:.3f} per " + f"stare a 6%, drift {_drift(k6*book)*100:.2f}%/a)") + for nm, spec in cells_tail[1:]: + dlt, ten, bud, roll, fu = spec + hr2, _, _ = hedge_series(dlt, ten, bud, roll, fu, SPREAD[ten]) + hd = book + hr2 + k6h = k_for_same_dd(hd, 0.06) + print(f" {nm:<20} maxDD {_dd(hd)*100:5.2f}% -> SFORA (serve k={k6h:.3f}, " + f"drift {_drift(k6h*hd)*100:.2f}%/a)") + print(" >> Sull'UNICA regola che decide un conto funded la copertura va nel verso SBAGLIATO:") + print(" alza il maxDD e quindi ABBASSA la leva ammessa. Il beneficio sui breach a un") + print(" giorno (0,92 -> 0,55/anno) e' reale ma si compra piu' a buon mercato de-levando,") + print(" che sulla regola a 6 punti e' l'unica leva che funziona.") + + print("\n 5 PEGGIORI MESI (nudo vs coperto δ-0.10 30g 1%/a, f misurato):") + hr, _, _ = hedge_series(-0.10, 30, 0.010, "expiry", F_MEAS.get(-0.10, f_pick), SPREAD[30]) + wm_b = worst_months(bs_book) + wm_h = worst_months(pd.Series(book + hr, index=idx)) + mh = dict(wm_h) + hs_m = (1.0 + pd.Series(book + hr, index=idx)).resample("ME").prod() - 1.0 + hs_m.index = [str(i.date())[:7] for i in hs_m.index] + for mth, v in wm_b: + hv = float(hs_m.get(mth, float("nan"))) + print(f" {mth} nudo {v*100:>7.2f}% coperto {hv*100:>7.2f}% " + f"delta {(hv - v)*100:>+6.2f}pp") + + # ---------------------------------------------------------------- §F. il rischio sbagliato + print("\n" + "=" * 100) + print("§F. L'ATTESA (c) MISURATA — la put protegge dal rischio che il libro ha davvero?") + print("=" * 100) + und = 0.5 * (P["px_BTC"].pct_change() + P["px_ETH"].pct_change()).fillna(0.0).values + ordr = np.argsort(book)[:20] + n_crash = int((und[ordr] <= -0.05).sum()) + print(f" Sui 20 PEGGIORI giorni del libro: {n_crash}/20 sono giorni di crollo del sottostante") + print(f" (BTC/ETH medio <= -5%). Negli altri {20-n_crash} il libro perde mentre il mercato NON") + print(f" crolla -> una put non puo' proteggere per costruzione.") + print(f" {'giorno':>12} {'libro':>8} {'BTC/ETH':>9} {'TP01':>8} {'SKH01':>8}") + for j in ordr[:8]: + print(f" {str(idx[j].date()):>12} {book[j]*100:>7.2f}% {und[j]*100:>8.2f}% " + f"{P['tp'].values[j]*100:>7.2f}% {P['skh'].values[j]*100:>7.2f}%") + beta = float(np.polyfit(und, book, 1)[0]) + corr = float(np.corrcoef(book, und)[0, 1]) + print(f" corr(libro, sottostante) = {corr:+.3f} | beta del libro al sottostante = {beta:+.3f}") + print() + print(" --- VERDETTO SULL'ATTESA A PRIORI (si riporta anche dove e' REFUTATA) ---") + if n_crash >= 12: + print(f" (c) REFUTATA nella forma in cui l'ho scritta: {n_crash}/20 dei peggiori giorni del") + print( " libro SONO giorni di crollo. Non e' vero che si compra assicurazione contro il") + print( " rischio sbagliato in senso di DIREZIONE. ⚠️ E il caso che avevo in mente (lo") + print( " short-squeeze di SKH01) sta FUORI FINESTRA: il DVOL parte 2021-03.") + else: + print(f" (c) CONFERMATA: solo {n_crash}/20 dei peggiori giorni sono crolli.") + print(f" (b) CONFERMATA, ed e' il meccanismo operativo: beta del libro al sottostante {beta:+.3f}.") + print( " Il 2021-05-19 il sottostante ha fatto -21,1% e il libro -2,0%: la strategia aveva") + print( " GIA' ridotto la perdita di ~10x. La put paga in proporzione al crollo del") + print( " SOTTOSTANTE, ma protegge una perdita gia' ridotta di un ordine di grandezza ->") + print( " per coprire un DD del libro serve un nozionale di put ~1/beta volte l'equity, e") + print( " quel nozionale costa ~1/beta volte il budget.") + print( " (a) CONFERMATA: vedi il decadimento del premio nel §B (il costo cresce col budget e") + print( " col f mentre il beneficio satura).") + + # ---------------------------------------------------------------- §G. eseguibilita' + print("\n" + "=" * 100) + print("§G. ESEGUIBILITA' — misurata DAL VENUE. Comprare un'opzione costa il PREMIO, non il") + print(" NOZIONALE: il muro dei 'lotti da migliaia di dollari' e' il collaterale per VENDERE.") + print("=" * 100) + vm = venue_min_ticket() + if vm["ok"]: + print(f" {'asset':>6} {'famiglia':>14} {'min_trade_amount':>18} {'tick':>10} {'settle':>8}") + for (a, fam), v in sorted(vm["fam"].items()): + print(f" {a:>6} {fam:>14} {v['min_amt']:>18} {v['tick']:>10} {v['settle']:>8}") + else: + print(f" NON MISURATO: {vm['note']}") + # costo del lotto minimo con i premi modellati di fine campione + print(f"\n Costo del LOTTO MINIMO e capitale a cui la copertura diventa comprabile a ogni roll:") + print(f" {'delta':>6} {'ten':>4} {'premio/unita':>13} {'lotto min':>10} " + f"{'$ lotto':>9} {'budget 1%/roll a $635':>22} {'capitale minimo':>16}") + for dlt in G_DELTA: + for ten in (7, 30): + S = float(P["px_BTC"].values[-1]); sig = float(P["iv_BTC"].values[-1]) + T = ten / DPY + K = _strike_from_delta(S, T, sig, dlt) + pm = _bs_put(S, K, T, sig) + fu = F_MEAS.get(dlt, F_LENSES_STATIC[1]) + price = fu * pm + minamt = vm["fam"].get(("BTC", "USDC-lineare"), {}).get("min_amt", 0.01) + lot = minamt * price + per_roll_635 = 0.010 * (ten / DPY) * 0.5 * 635.0 # 1%/a, meta' su BTC + cap_min = lot / (0.010 * (ten / DPY) * 0.5) + print(f" {dlt:>6.2f} {ten:>4} {price:>13.2f} {minamt:>10} {lot:>9.2f} " + f"{per_roll_635:>21.2f} {cap_min:>15,.0f}") + print(" ⚠️ E c'e' un secondo muro, STRUTTURALE e non di capitale: il TICK. Su BTC_USDC il tick") + print(" e' 5 USDC -> una put che vale meno di $5 NON e' quotabile sotto $5. Piu' e' deep-OTM") + print(" (cioe' piu' e' economica la copertura) piu' il tick domina il prezzo.") + # eval_weights_smallcap sulla gamba LINEARE, come riferimento del pavimento + try: + from src.strategies.trend_portfolio import TrendPortfolio, CANONICAL + d1 = resample_1d(load_data("BTC", "1h")) + tpm = TrendPortfolio(**CANONICAL) + sc = al.eval_weights_smallcap(d1, tpm.target_series(d1), capital=635.0, min_order=5.0) + print(f"\n Riferimento (gamba LINEARE dello stesso libro): TP01/BTC a $635 haircut Sharpe " + f"{sc['sharpe_haircut']:+.3f} -> il pavimento $5 non morde sul perpetual.") + print(" La differenza NON e' il livello del costo ma la sua FORMA (regola 25/07): sul") + print(" perpetual il minimo e' in DOLLARI di nozionale, sull'opzione e' in UNITA' di") + print(" sottostante MOLTIPLICATE per un premio che tende a zero -> due muri diversi.") + except Exception as e: + print(f"\n (riferimento smallcap non calcolato: {type(e).__name__}: {e})") + + # ---------------------------------------------------------------- §H. gate residui + print("\n" + "=" * 100) + print("§H. GATE RESIDUI") + print("=" * 100) + # cella migliore in assoluto al f misurato/realistico + real = [r for r in all_rows if r["lens"] != "f=1.00 (modello nudo)"] + pool = real if real else all_rows + bestr = max(pool, key=lambda r: r["gap"]) + hedged_best = pd.Series(book + bestr["hret"], index=idx) + imp = al.implausible_sharpe(hedged_best) + print(f" implausible_sharpe(libro coperto, cella migliore al f realistico): " + f"implausible={imp['implausible']} Sharpe {imp.get('sharpe', float('nan')):.2f} " + f"maxDD {imp.get('maxdd', float('nan'))*100:.2f}%") + for r in imp["reasons"]: + print(f" - {r}") + sr_all = [r["sh_hedged"] for r in pool] + dsr, nullmax = al.deflated_sharpe(bestr["sh_hedged"], sr_all, hedged_best) + print(f" deflated_sharpe della cella MIGLIORE su {len(pool)} trial: DSR {dsr:.3f} " + f"(max atteso dal nullo {nullmax:.2f}) -> {'PASS' if dsr >= 0.95 else 'FAIL'}") + print(" ⚠️ Il DSR qui e' informativo ma NON e' il gate che decide: la domanda non e' 'questa") + print(" cella ha alpha' (non ne ha per costruzione) ma 'batte il de-levering' — ed e' il §B.") + + # sensitivita': f asimmetrico (compri caro, vendi a modello) + print("\n SENSITIVITA' f ASIMMETRICO (si compra a f, si vende a modello — caso pessimo per il") + print(" roll anticipato, che nel modello base non paga f perche' non arriva a intrinseco):") + fu = F_MEAS.get(-0.10, F_LENSES_STATIC[1]) + for roll in G_ROLL: + hs, _, _ = hedge_series(-0.10, 30, 0.010, roll, fu, SPREAD[30]) + ha, _, _ = hedge_series(-0.10, 30, 0.010, roll, fu, SPREAD[30], f_sell=1.0) + es, ea = evaluate(book, hs), evaluate(book, ha) + print(f" roll={roll:<7} f simmetrico Δdrift {(es['dr_hedged']-es['dr_delev'])*100:+6.2f}pp" + f" f asimmetrico Δdrift {(ea['dr_hedged']-ea['dr_delev'])*100:+6.2f}pp") + + # ---------------------------------------------------------------- §I. verdetto + print("\n" + "=" * 100) + print("§I. VERDETTO") + print("=" * 100) + for lab, _ in lenses: + sub = [r for r in all_rows if r["lens"] == lab] + nw = sum(1 for r in sub if r["wins"]) + nred = sum(1 for r in sub if r["dd_reduced"]) + cand = [r for r in sub if r["dd_reduced"]] or sub + bb = max(cand, key=lambda r: r["gap"]) + print(f" {lab:<32}: DD scende {nred:>2}/{len(sub)} | batte il de-levering {nw:>2}/{len(sub)}" + f" | miglior Δdrift {bb['gap']*100:+6.2f} pp/a | ΔSharpe {bb['sh_hedged']-bb['sh_base']:+.3f}") + nw_real = sum(1 for r in pool if r["wins"]) + nred_real = sum(1 for r in pool if r["dd_reduced"]) + verdict = "REFUTATO" if nw_real == 0 else ("PARZIALE" if nw_real < len(pool) * 0.5 else "REGGE") + print(f"\n >>> {verdict}: al f realistico il maxDD scende in {nred_real}/{len(pool)} celle e la") + print(f" copertura batte il de-levering in {nw_real}/{len(pool)}.") + print( " ⚠️ Il fatto che il maxDD SALGA nella maggior parte delle celle e' il risultato piu'") + print( " forte del filone: prima ancora del prezzo, la premessa 'la copertura riduce il") + print( " drawdown' e' FALSA su questo libro — il bleed del premio cade DENTRO i drawdown") + print( " (che sono lunghi e poco profondi) mentre il payoff cade su singoli giorni di") + print( " crollo che NON sono il fondo del drawdown.") + print(f"\n Tempo: {time.time()-t0:.0f}s") + + +if __name__ == "__main__": + main()