"""r0725_capcurve — LA CURVA CAPITALE -> BOOK ESEGUIBILE (2026-07-25). Perche' questo script esiste. Il modello di crescita del 2026-07-24 (`r0724_goal50_math.py` parte A) usa un **CAGR FISSO 15%** per stimare il tempo da $600 a 50 EUR/g. Ma il CAGR del book NON e' una costante: a $600 girano 2 sleeve su 6 (TP01+SKH01 su Deribit), a $5k si aggiunge XSR01, a ~$20k XS01, e GTAA01 (6 ETF su IB) ha un COSTO FISSO PER ORDINE che a capitale piccolo se lo mangia. Il tasso di crescita e' quindi **funzione dello stato** (equity), e la traiettoria vera e' super-lineare: piu' capitale -> piu' diversificazione -> piu' Sharpe -> piu' CAGR. Tre parti: A) COST-FLOOR PER SLEEVE — Sharpe/CAGR realistici di ogni sleeve in funzione dei dollari ALLOCATI a quello sleeve, con il modello di costo del SUO venue: - Deribit (TP01, SKH01): fee proporzionale 0.05%/lato, min order $5 (skip, non trade). - IB (GTAA01): commissione = min(max($0.35, $0.0035/azione), 1% del controvalore) -> **pavimento FISSO per ordine**. gtaa.py usava 2bps proporzionali + ribilancio continuo: difetto trovato QUI e poi corretto nel modulo (banda + costo IB reale). - Hyperliquid (XSR01): fee 0.045%/lato, min order $5, 50 gambe. Novita' vs tutto il lavoro precedente: la curva GTAA, cioe' a che capitale il 3o sleeve smette di essere un costo. B) FRONTIERA — per ogni capitale totale C, il MIGLIOR sottoinsieme di sleeve eseguibili (pesi canonici rinormalizzati), con Sharpe/CAGR/DD del book risultante. Da qui la funzione CAGR(C) che sostituisce il 15% fisso. C) CRESCITA STATE-DEPENDENT — traiettoria da $600 con CAGR(C), depositi mensili e fisco italiano 2026 (33% sulle plusvalenze), vs il modello a CAGR fisso. Piu' il "premio di sblocco": quanto vale anticipare un deposito che fa scattare una soglia. ONESTA': (1) i rendimenti di SKH01 sono modellati equity-daily-step; il path live e' peggiore (FULL 1.19/HOLD 1.15/DD 25% vs 1.46/1.64/18%) -> parte B gira anche in modalita' de-luck. (2) Le soglie XS01/VRP01 sono quelle gia' documentate (non ricalcolate qui). (3) Il fisco e' applicato come 33% sul realizzato annuo: e' l'ipotesi peggiore (nessuna compensazione). Uso: `uv run python scripts/research/r0725_capcurve.py` """ from __future__ import annotations import sys 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)) from src.data.downloader import load_data # noqa: E402 from src.portfolio.gtaa import EQ_UNIVERSE, _close, _exposure # noqa: E402 from src.portfolio.portfolio import combine_outer, metrics, to_daily # noqa: E402 from src.portfolio.sleeves import _skyhook_returns, _tp01_returns # noqa: E402 from src.strategies.trend_portfolio import CANONICAL, TrendPortfolio, resample_1d, simple_returns # noqa: E402 EURUSD = 1.09 TARGET_EUR_DAY = 50.0 TAX_RATE = 0.33 # L.199/2025: capital gain crypto 33% dal 2026, esenzione abolita ASSETS = ("BTC", "ETH") # --- modelli di costo per VENUE ------------------------------------------------- DERIBIT_FEE_SIDE = 0.0005 # taker 0.05%/lato (0.10% RT) - gia' il default del progetto DERIBIT_MIN_ORDER = 5.0 IB_MIN_COMM = 0.35 # USD per ordine (IBKR Pro tiered, US stocks/ETF) IB_PER_SHARE = 0.0035 # USD per azione IB_MAX_FRAC = 0.01 # cap: 1% del controvalore dell'ordine IB_MIN_TRADE = 1.0 # ordine frazionale minimo sensato in USD HL_FEE_SIDE = 0.00045 HL_MIN_ORDER = 5.0 VOL_BUDGET = 0.15 # budget di rischio del confronto ISO-RISCHIO (vol annua) CAP_GRID = (100.0, 200.0, 400.0, 600.0, 1_000.0, 2_000.0, 5_000.0, 10_000.0, 20_000.0, 50_000.0, 100_000.0, 200_000.0) # ============================================================== A) COST FLOOR @lru_cache(maxsize=64) def _tp01_realistic_cached(alloc: float) -> pd.Series: """TP01 con capitale ALLOCATO `alloc` (meta' per asset): un ribilanciamento il cui nozionale |Dw|*cap_asset e' sotto min order NON si esegue (si tiene la posizione).""" cap_asset = alloc / len(ASSETS) tp = TrendPortfolio(**CANONICAL) series = {} for a in ASSETS: df = resample_1d(load_data(a, "1h")) r = simple_returns(df["close"].values.astype(float)) tgt = np.nan_to_num(np.asarray(tp.target_series(df), float)) held_tgt = np.empty(len(tgt)) cur = 0.0 for i in range(len(tgt)): if abs(tgt[i] - cur) * cap_asset >= DERIBIT_MIN_ORDER: cur = tgt[i] held_tgt[i] = cur pos = np.zeros(len(held_tgt)) pos[1:] = held_tgt[:-1] # causale: deciso a i-1, tenuto in i net = pos * r - DERIBIT_FEE_SIDE * np.abs(np.diff(pos, prepend=0.0)) net[0] = 0.0 series[a] = pd.Series(np.clip(net, -0.99, None), index=pd.to_datetime(df["datetime"])) J = pd.concat(series, axis=1, join="inner").fillna(0.0) out = pd.Series(0.5 * J["BTC"].values + 0.5 * J["ETH"].values, index=J.index) if out.index.tz is None: out.index = out.index.tz_localize("UTC") return out @lru_cache(maxsize=64) def _gtaa_realistic_cached(alloc: float, universe=EQ_UNIVERSE) -> pd.Series: """GTAA con capitale ALLOCATO `alloc`, costo IB REALE (pavimento fisso per ordine). gtaa.py modellava 2bps proporzionali: a $600 su 6 gambe l'ordine tipico e' ~$10-30 e la commissione IB minima ($0.35) e' l'**1-3%** del controvalore, cioe' 15-150x il modello. Qui ogni gamba e' un conto separato: si ribilancia solo se il nozionale supera il minimo frazionale, e si paga min(max(0.35, 0.0035*azioni), 1%*valore) A OGNI ordine. """ n = len(universe) cap_leg = alloc / n legs = {} for sym in universe: close = _close(sym) ex = np.nan_to_num(np.asarray(_exposure(close).values, float)) px = close.values.astype(float) ret = np.zeros(len(px)) ret[1:] = px[1:] / px[:-1] - 1.0 held = np.empty(len(ex)) cur = 0.0 comm = np.zeros(len(ex)) for i in range(len(ex)): notional = abs(ex[i] - cur) * cap_leg if notional >= IB_MIN_TRADE: shares = notional / px[i] c = min(max(IB_MIN_COMM, IB_PER_SHARE * shares), IB_MAX_FRAC * notional) comm[i] = c / cap_leg # commissione in frazione del capitale-gamba cur = ex[i] held[i] = cur pos = np.zeros(len(held)) pos[1:] = held[:-1] net = pos * ret - comm net[0] = 0.0 legs[sym] = pd.Series(net, index=close.index) r = pd.concat(legs, axis=1, sort=True).sort_index().mean(axis=1, skipna=True).dropna() if r.index.tz is None: r.index = r.index.tz_localize("UTC") days = pd.date_range(r.index.min().normalize(), r.index.max().normalize(), freq="1D", tz="UTC") return r.reindex(days).fillna(0.0) @lru_cache(maxsize=256) def _gtaa_banded_cached(alloc: float, band_usd: float, every: int = 1, universe=EQ_UNIVERSE) -> pd.Series: """GTAA con BANDA di ribilanciamento e cadenza: si guarda il target solo ogni `every` barre, e si esegue solo se il nozionale supera `band_usd`. Con un costo FISSO per ordine la banda ottimale non e' il minimo eseguibile (come su Deribit) ma un valore molto piu' alto: ogni ordine costa $0.35 a prescindere dalla taglia.""" n = len(universe) cap_leg = alloc / n legs = {} for sym in universe: close = _close(sym) ex = np.nan_to_num(np.asarray(_exposure(close).values, float)) px = close.values.astype(float) ret = np.zeros(len(px)) ret[1:] = px[1:] / px[:-1] - 1.0 held = np.empty(len(ex)) cur = 0.0 comm = np.zeros(len(ex)) for i in range(len(ex)): if i % every == 0: notional = abs(ex[i] - cur) * cap_leg if notional >= max(band_usd, IB_MIN_TRADE): shares = notional / px[i] c = min(max(IB_MIN_COMM, IB_PER_SHARE * shares), IB_MAX_FRAC * notional) comm[i] = c / cap_leg cur = ex[i] held[i] = cur pos = np.zeros(len(held)) pos[1:] = held[:-1] net = pos * ret - comm net[0] = 0.0 legs[sym] = pd.Series(net, index=close.index) r = pd.concat(legs, axis=1, sort=True).sort_index().mean(axis=1, skipna=True).dropna() if r.index.tz is None: r.index = r.index.tz_localize("UTC") days = pd.date_range(r.index.min().normalize(), r.index.max().normalize(), freq="1D", tz="UTC") return r.reindex(days).fillna(0.0) def part_a2() -> dict: """La banda salva GTAA? Sweep banda x cadenza per capitale. Il parametro e' di COSTO (non di segnale): la scelta si fa sul costo atteso, non sullo Sharpe in-sample.""" print("\n" + "=" * 100) print(" A2) GTAA01 — la BANDA di ribilanciamento puo' salvarlo? (costo fisso IB $0.35/ordine)") print("=" * 100) from src.portfolio.gtaa import gtaa_returns g_model = to_daily(gtaa_returns()) # calendario 365: vedi nota in part_a() print(f" riferimento MODELLATO (costo/banda correnti del modulo): " f"Sh {metrics(g_model)['sharpe']:.2f} CAGR {metrics(g_model)['cagr']:.1%}") best = {} for alloc in (600.0, 2_000.0, 5_000.0, 20_000.0): print(f"\n --- capitale allocato a GTAA: ${alloc:,.0f} (per gamba ${alloc/6:,.0f}) ---") print(f" {'banda$':>8} {'cadenza':>9} {'Sharpe':>8} {'CAGR':>8} {'maxDD':>8} {'ord/anno':>9}") rows = [] for every, lbl in ((1, "daily"), (5, "weekly"), (21, "monthly")): for band in (1.0, 5.0, 25.0, 50.0, 100.0): if band > alloc / 6: # banda piu' grande della gamba = mai trade continue s = gtaa_banded(alloc, band, every) m = metrics(s) # ordini/anno stimati dal drag: ricostruiti come costo/anno / 0.35 rows.append((band, lbl, every, m, s)) for band, lbl, every, m, s in rows: yrs = len(s) / 365.25 print(f" {band:>8.0f} {lbl:>9} {m['sharpe']:>8.2f} {m['cagr']:>8.1%} " f"{m['maxdd']:>8.1%} {'':>9}") top = max(rows, key=lambda r: r[3]["sharpe"]) best[alloc] = dict(band=top[0], every=top[2], label=top[1], metrics=top[3], series=top[4]) print(f" -> migliore: banda ${top[0]:.0f} {top[1]} Sh {top[3]['sharpe']:.2f} " f"CAGR {top[3]['cagr']:.1%}") return best def _sh(s: pd.Series, lo=None) -> dict: x = s if lo is None else s[s.index >= lo] return metrics(x) # --- wrapper pubblici: le implementazioni sopra sono memoizzate e lru_cache ritorna lo STESSO # oggetto a ogni hit -> si consegna sempre una COPIA, cosi' una mutazione in place di un # chiamante non puo' corrompere in silenzio i risultati successivi. def tp01_realistic(alloc: float) -> pd.Series: return _tp01_realistic_cached(float(alloc)).copy() def gtaa_realistic(alloc: float, universe=EQ_UNIVERSE) -> pd.Series: return _gtaa_realistic_cached(float(alloc), universe).copy() def gtaa_banded(alloc: float, band_usd: float, every: int = 1, universe=EQ_UNIVERSE) -> pd.Series: return _gtaa_banded_cached(float(alloc), float(band_usd), int(every), universe).copy() def part_a() -> dict: print("=" * 100) print(" A) COST-FLOOR PER SLEEVE — Sharpe reale vs dollari ALLOCATI a quello sleeve") print("=" * 100) alloc_grid = (50.0, 100.0, 200.0, 500.0, 1_000.0, 2_000.0, 5_000.0, 10_000.0, 50_000.0) out = {"TP01": {}, "GTAA01": {}} print("\n TP01 (Deribit, 2 gambe, fee 0.05%/lato, min order $5)") print(f" {'alloc':>10} {'Sharpe':>8} {'CAGR':>8} {'maxDD':>8} {'vs modellato':>14}") base_tp = _tp01_returns() if base_tp.index.tz is None: base_tp.index = base_tp.index.tz_localize("UTC") m_model = metrics(base_tp) for a in alloc_grid: s = tp01_realistic(a) m = metrics(s) out["TP01"][a] = s print(f" {a:>10,.0f} {m['sharpe']:>8.2f} {m['cagr']:>8.1%} {m['maxdd']:>8.1%}" f" {m['sharpe'] - m_model['sharpe']:>+14.2f}") print(f" {'MODELLATO':>10} {m_model['sharpe']:>8.2f} {m_model['cagr']:>8.1%} {m_model['maxdd']:>8.1%}") print("\n GTAA01 (IB, 6 ETF, commissione min $0.35/ordine — PAVIMENTO FISSO)") print(f" {'alloc':>10} {'Sharpe':>8} {'CAGR':>8} {'maxDD':>8} {'drag comm/anno':>15}") from src.portfolio.gtaa import gtaa_returns # to_daily: la serie GTAA grezza ha solo i giorni di borsa (~252/anno) e metrics() # annualizza a 365 -> senza questa conversione il riferimento e' gonfiato ~1.20x (Sharpe) # e ~1.45x (CAGR). Le righe realistiche sono gia' su calendario 365. g_model = to_daily(gtaa_returns()) mg_model = metrics(g_model) for a in alloc_grid: s = gtaa_realistic(a) m = metrics(s) out["GTAA01"][a] = s drag = mg_model["cagr"] - m["cagr"] print(f" {a:>10,.0f} {m['sharpe']:>8.2f} {m['cagr']:>8.1%} {m['maxdd']:>8.1%}" f" {drag:>15.2%}") print(f" {'MODELLATO':>10} {mg_model['sharpe']:>8.2f} {mg_model['cagr']:>8.1%} {mg_model['maxdd']:>8.1%}" f" {'(modulo, banda)':>15}") out["_models"] = dict(TP01=m_model, GTAA01=mg_model) return out # ============================================================== B) FRONTIERA # Config GTAA scelta sul PLATEAU (non sulla cella max): cadenza settimanale + banda $50. # La banda e' un parametro di COSTO, non di segnale — si sceglie sul plateau robusto # (weekly/monthly x banda 25-100 danno 0.45-0.64 a ogni capitale), non sull'argmax # in-sample (banda 50 daily a $600 = 0.57 e' una cella isolata, crolla a 0.28 a banda 100). GTAA_BAND_USD, GTAA_EVERY = 50.0, 5 def part_b(deluck: float = 1.0) -> dict: """Per ogni capitale totale C: il book eseguibile migliore e il suo CAGR/Sharpe. A $600 il book live e' TP01+SKH01 su Deribit. La domanda nuova: conviene SPEZZARE il capitale (una parte su IB per GTAA01)? Il costo dello split e' il pavimento IB; il beneficio e' un 3o sleeve a corr ~+0.10. deluck: moltiplicatore sul DRIFT (0.6 = convenzione de-luck del progetto), la vol resta. """ tag = "MODELLATO" if deluck == 1.0 else f"DE-LUCK x{deluck}" print("\n" + "=" * 100) print(f" B) FRONTIERA CAPITALE -> BOOK ESEGUIBILE [{tag}]") print("=" * 100) skh = _skyhook_returns() if skh.index.tz is None: skh.index = skh.index.tz_localize("UTC") def _dl(s: pd.Series) -> pd.Series: if deluck == 1.0: return s return s - (1.0 - deluck) * float(s.mean()) print(f"\n {'capitale':>9} {'book':>28} {'ShFULL':>7} {'ShHOLD':>7} {'CAGR':>7} " f"{'maxDD':>7} {'EUR/g':>7}") rows = [] for C in (600.0, 1_000.0, 2_000.0, 5_000.0, 10_000.0, 20_000.0, 50_000.0, 100_000.0, 200_000.0): cands = [] # --- 1) solo Deribit (book live attuale) for wg in (0.0, 0.10, 0.20, 0.30): w_der = 1.0 - wg alloc_tp, alloc_gt = C * w_der * 0.75, C * wg cols = {"TP01": _dl(tp01_realistic(alloc_tp)), "SKH01": _dl(skh)} wts = {"TP01": w_der * 0.75, "SKH01": w_der * 0.25} if wg > 0: cols["GTAA01"] = _dl(gtaa_banded(alloc_gt, GTAA_BAND_USD, GTAA_EVERY)) wts["GTAA01"] = wg combo = combine_outer(cols, wts, lo=pd.Timestamp("2019-03-14", tz="UTC")) m = metrics(combo) mh = metrics(combo[combo.index >= pd.Timestamp("2025-01-01", tz="UTC")]) name = "TP01+SKH01 (Deribit)" if wg == 0 else f"+GTAA01 {wg:.0%} (IB split)" cands.append((name, m, mh, wg)) best = max(cands, key=lambda r: r[1]["sharpe"]) for name, m, mh, wg in cands: mark = " *" if name == best[0] else " " eur = C * m["cagr"] * (1 - TAX_RATE) / EURUSD / 365.0 print(f" {C:>9,.0f} {name:>28} {m['sharpe']:>7.2f} {mh['sharpe']:>7.2f} " f"{m['cagr']:>7.1%} {m['maxdd']:>7.1%} {eur:>7.2f}{mark}") rows.append((C, best)) print() return dict(rows=rows) # ============================================================== C) RENDITA def book_series(lens: str = "hourly", with_gtaa: float = 0.0, alloc: float = 600.0) -> pd.Series: """Serie giornaliera del BOOK ESEGUIBILE. lens='hourly' -> SKH01 sul PATH LIVE (fill al close 5m del prossimo multiplo orario: e' la stima onesta di cio' che il cron esegue davvero). lens='canonical' -> SKH01 modellato (fill AL LIVELLO SL/TP): e' il backtest, non il live. """ sys.path.insert(0, str(ROOT / "scripts" / "research")) import r0724_skh_live_weight as skl # noqa: E402 (riusa la macchineria d'audit gia' validata) skh = skl.skh_book(0, lens) if skh.index.tz is None: skh.index = skh.index.tz_localize("UTC") cols = {"TP01": tp01_realistic(alloc * 0.75), "SKH01": skh} wts = {"TP01": 0.75, "SKH01": 0.25} if with_gtaa > 0: cols = {"TP01": tp01_realistic(alloc * (1 - with_gtaa) * 0.75), "SKH01": skh, "GTAA01": gtaa_banded(alloc * with_gtaa, GTAA_BAND_USD, GTAA_EVERY)} wts = {"TP01": (1 - with_gtaa) * 0.75, "SKH01": (1 - with_gtaa) * 0.25, "GTAA01": with_gtaa} return combine_outer(cols, wts, lo=pd.Timestamp("2019-03-14", tz="UTC")) def _boot_paths(r: np.ndarray, n_paths: int, n_days: int, block: int, rng: np.random.Generator) -> np.ndarray: """Block bootstrap (blocchi di `block` giorni) — preserva la dipendenza seriale (momentum/vol-clustering) che un bootstrap iid distruggerebbe, gonfiando la sopravvivenza.""" n_blocks = int(np.ceil(n_days / block)) starts = rng.integers(0, len(r) - block, size=(n_paths, n_blocks)) idx = (starts[:, :, None] + np.arange(block)[None, None, :]).reshape(n_paths, -1) return r[idx[:, :n_days]] def survival(r: np.ndarray, cap0: float, draw_usd_yr: float, years: int, k: float, n_paths: int = 4000, block: int = 20, index_rate: float = 0.0, seed: int = 725) -> dict: """P(il capitale sopravvive `years` anni prelevando `draw_usd_yr`/anno) con leva `k`. Prelievo giornaliero pro-rata (una rendita si preleva in continuo, non a fine anno). """ rng = np.random.default_rng(seed) n_days = int(years * 365) paths = _boot_paths(r, n_paths, n_days, block, rng) * k cap = np.full(n_paths, cap0, float) alive = np.ones(n_paths, bool) daily_draw = draw_usd_yr / 365.0 ruin_day = np.full(n_paths, n_days + 1, int) for t in range(n_days): d = daily_draw * (1.0 + index_rate) ** (t / 365.0) cap = np.where(alive, cap * (1.0 + paths[:, t]) - d, cap) newly = alive & (cap <= 0) ruin_day[newly] = t alive &= cap > 0 return dict(p_survive=float(alive.mean()), median_end=float(np.median(np.where(alive, cap, 0.0))), p10_end=float(np.percentile(np.where(alive, cap, 0.0), 10)), median_ruin_yr=float(np.median(ruin_day[~alive]) / 365.0) if (~alive).any() else np.nan) def part_c() -> dict: """La RENDITA sostenibile del book. Il muro del 24/07 usava capitale = target/CAGR, che ignora il RISCHIO DI SEQUENZA: prelevare da un book volatile mentre e' in drawdown consuma capitale che non torna. Qui il muro vero, a 95% di sopravvivenza.""" print("\n" + "=" * 100) print(" C) RENDITA SOSTENIBILE — quanto capitale serve DAVVERO per 50 EUR/g") print("=" * 100) out = {} for lens, deluck in (("canonical", 1.0), ("hourly", 1.0), ("hourly", 0.6)): b = book_series(lens) if deluck != 1.0: b = b - (1.0 - deluck) * float(b.mean()) # taglia il DRIFT, lascia la vol m = metrics(b) r = b.values.astype(float) vol = float(r.std() * np.sqrt(365)) lbl = ("MODELLATO (backtest)" if lens == "canonical" else "PATH LIVE (cron orario)") if deluck != 1.0: lbl += f" DE-LUCK drift x{deluck}" print(f"\n --- book TP01 75 / SKH01 25 — {lbl} ---") print(f" Sharpe {m['sharpe']:.2f} CAGR {m['cagr']:.1%} vol {vol:.1%} " f"maxDD {m['maxdd']:.1%} ({m['n']} giorni)") # tasso di prelievo massimo sostenibile al 95% di sopravvivenza, per leva k print(f"\n {'leva':>5} {'vol':>7} {'CAGR':>7} {'SWR20a':>8} {'SWR30a':>8} " f"{'capitale per 50EUR/g netti':>28}") rows = {} for k in (0.5, 0.75, 1.0, 1.5, 2.0): # SWR = massimo tasso di prelievo (% del capitale iniziale) con P(vivo) >= 95% swr = {} for years in (20, 30): lo, hi = 0.0, 0.40 for _ in range(14): # bisezione sul tasso mid = (lo + hi) / 2 p = survival(r, 1_000_000.0, 1_000_000.0 * mid, years, k, n_paths=2000)["p_survive"] if p >= 0.95: lo = mid else: hi = mid swr[years] = lo gross_needed = TARGET_EUR_DAY * 365 * EURUSD / (1 - TAX_RATE) cap_needed = gross_needed / swr[20] if swr[20] > 0.001 else float("inf") print(f" {k:>5.2f} {vol*k:>7.1%} {m['cagr']*k:>7.1%} {swr[20]:>8.2%} " f"{swr[30]:>8.2%} ${cap_needed:>26,.0f}") rows[k] = dict(swr20=swr[20], swr30=swr[30], cap=cap_needed) out[(lens, deluck)] = dict(metrics=m, vol=vol, rows=rows, series=b) best_k = max(rows, key=lambda k: rows[k]["swr20"]) print(f" -> leva che massimizza la rendita: {best_k:.2f}x " f"(SWR {rows[best_k]['swr20']:.2%}, capitale ${rows[best_k]['cap']:,.0f})") return out # ====================================================== C2) RENDITA PERPETUA def part_c2() -> dict: """'Survive 20 anni' NON e' una rendita: e' un piano di consumo del capitale. Una rendita passiva vera vuole il capitale INTATTO (altrimenti a 20 anni il reddito finisce). Qui il criterio giusto: prelievo massimo con P(capitale a 20a >= capitale iniziale) >= 90%.""" print("\n" + "=" * 100) print(" C2) RENDITA PERPETUA — prelievo che PRESERVA il capitale (P(cap_20a >= cap_0) >= 90%)") print("=" * 100) gross_needed = TARGET_EUR_DAY * 365 * EURUSD / (1 - TAX_RATE) print(f" 50 EUR/g netti = EUR {TARGET_EUR_DAY*365:,.0f}/anno netti " f"= ${gross_needed:,.0f}/anno LORDI (fisco {TAX_RATE:.0%})\n") print(f" {'lente':>34} {'leva':>5} {'SWR-20a':>8} {'PERPETUA':>9} " f"{'capitale (perpetua)':>21}") out = {} for lens, deluck in (("canonical", 1.0), ("hourly", 1.0), ("hourly", 0.6)): b = book_series(lens) if deluck != 1.0: b = b - (1.0 - deluck) * float(b.mean()) r = b.values.astype(float) lbl = ("MODELLATO" if lens == "canonical" else "PATH LIVE") if deluck != 1.0: lbl += f" de-luck x{deluck}" for k in (1.0, 1.5, 2.0): lo, hi = 0.0, 0.40 for _ in range(13): mid = (lo + hi) / 2 s = survival(r, 1_000_000.0, 1_000_000.0 * mid, 20, k, n_paths=2000) # criterio PERPETUO: il 90mo percentile dei path deve finire >= capitale iniziale keep = s["p10_end"] >= 1_000_000.0 if keep: lo = mid else: hi = mid swr20 = None lo2, hi2 = 0.0, 0.40 for _ in range(13): mid = (lo2 + hi2) / 2 if survival(r, 1e6, 1e6 * mid, 20, k, n_paths=2000)["p_survive"] >= 0.95: lo2 = mid else: hi2 = mid swr20 = lo2 cap = gross_needed / lo if lo > 0.002 else float("inf") print(f" {lbl:>34} {k:>5.2f} {swr20:>8.2%} {lo:>9.2%} " f"{('$' + format(cap, ',.0f')) if np.isfinite(cap) else 'IRRAGGIUNGIBILE':>21}") out[(lens, deluck, k)] = dict(swr20=swr20, perp=lo, cap=cap) print() return out # ============================================================== D) TRAIETTORIA def part_d(perp_rate: float, cap_needed: float) -> None: """Quanto tempo per accumulare il capitale della rendita — STOCASTICO (il 24/07 era deterministico: un CAGR fisso nasconde che meta' dei path arriva dopo).""" print("\n" + "=" * 100) print(f" D) TRAIETTORIA — da $600 a ${cap_needed:,.0f} (rendita perpetua {perp_rate:.2%})") print("=" * 100) b = book_series("hourly") b = b - 0.4 * float(b.mean()) # de-luck x0.6: la lente della decisione r = b.values.astype(float) rng = np.random.default_rng(725) n_days = 365 * 25 paths = _boot_paths(r, 3000, n_days, 20, rng) print(f"\n {'deposito/mese':>14} {'p10':>8} {'mediana':>9} {'p90':>8} " f"{'P(entro 10a)':>12} {'P(entro 20a)':>12}") for dep_eur in (0, 250, 500, 1_000, 2_000, 5_000): dep = dep_eur * EURUSD cap = np.full(3000, 600.0) hit = np.full(3000, -1, int) for t in range(n_days): cap = cap * (1.0 + paths[:, t]) if t % 30 == 0 and t > 0: cap += dep newly = (hit < 0) & (cap >= cap_needed) hit[newly] = t yrs = np.where(hit >= 0, hit / 365.0, np.nan) finite = yrs[~np.isnan(yrs)] p10 = np.percentile(finite, 10) if len(finite) > 100 else np.nan p50 = np.percentile(finite, 50) if len(finite) > 100 else np.nan p90 = np.percentile(finite, 90) if len(finite) > 100 else np.nan p10y = float((yrs <= 10).sum() / len(yrs)) p20y = float((yrs <= 20).sum() / len(yrs)) f = lambda v: f"{v:.1f}a" if np.isfinite(v) else " —" # noqa: E731 print(f" {dep_eur:>10,.0f}/m {f(p10):>8} {f(p50):>9} {f(p90):>8} " f"{p10y:>12.0%} {p20y:>12.0%}") print("\n (bootstrap a blocchi dei ritorni REALI del book, de-luck x0.6; i depositi") print(" sono in EUR/mese. 'P(entro Na)' = quota di path che tocca il capitale-rendita.)") # ================================================ E) IL MURO SI ABBASSA? def part_e() -> dict: """L'ipotesi di partenza: piu' capitale -> piu' sleeve eseguibili -> piu' Sharpe -> muro piu' basso. Per il book eseguibile-OGGI e' FALSA (parte B: Sharpe piatto da $600 a $200k). Ma attraverso le SOGLIE di sblocco (XSR01 ~$5k, XS01 ~$20k) puo' essere vera. Qui il conto onesto, con GTAA01 al suo costo IB REALE e SKH01 sul path live. Regole di ammissione rispettate: VRP01 ESCLUSO (regola permanente 'niente short-vol da modello in deploy'); XSR01 escluso dal book deployable (gate pre-registrato 2026-10-23, oggi in forward-monitor) ma mostrato come contro-fattuale. """ print("\n" + "=" * 100) print(" E) IL MURO SI ABBASSA COL CAPITALE? — book per tier di sbloccо") print("=" * 100) from src.portfolio.sleeves import _vrp_combo_returns, _xsec_returns sys.path.insert(0, str(ROOT / "scripts" / "research")) import r0724_skh_live_weight as skl skh = skl.skh_book(0, "hourly") if skh.index.tz is None: skh.index = skh.index.tz_localize("UTC") def _d(s): s = s.dropna().sort_index() if s.index.tz is None: s.index = s.index.tz_localize("UTC") return s xs = _d(_xsec_returns()).resample("1D").apply(lambda x: (1 + x).prod() - 1) vrp = _d(_vrp_combo_returns()).resample("1D").apply(lambda x: (1 + x).prod() - 1) gross_needed = TARGET_EUR_DAY * 365 * EURUSD / (1 - TAX_RATE) tiers = [ ("$600 — 2 sleeve (Deribit)", 600.0, {"TP01": 0.75, "SKH01": 0.25}), ("$2k — + GTAA01 (IB, banda)", 2_000.0, {"TP01": 0.60, "SKH01": 0.20, "GTAA01": 0.20}), ("$20k — + XS01 (Hyperliquid)", 20_000.0, {"TP01": 0.45, "SKH01": 0.25, "GTAA01": 0.20, "XS01": 0.10}), ("$20k — + XS01 + VRP01 (contro-fattuale)", 20_000.0, {"TP01": 0.40, "SKH01": 0.22, "GTAA01": 0.18, "XS01": 0.10, "VRP01": 0.10}), ] print(f"\n {'tier':>42} {'Sharpe':>7} {'CAGR':>7} {'vol':>6} {'maxDD':>7} " f"{'rendita':>8} {'capitale 50EUR/g':>18}") out = {} for name, C, w in tiers: cols = {"TP01": tp01_realistic(C * w["TP01"]), "SKH01": skh} if "GTAA01" in w: cols["GTAA01"] = gtaa_banded(C * w["GTAA01"], GTAA_BAND_USD, GTAA_EVERY) if "XS01" in w: cols["XS01"] = xs if "VRP01" in w: cols["VRP01"] = vrp combo = combine_outer(cols, w, lo=pd.Timestamp("2019-03-14", tz="UTC")) combo = combo - 0.4 * float(combo.mean()) # de-luck x0.6 (lente della decisione) m = metrics(combo) r = combo.values.astype(float) lo, hi = 0.0, 0.40 for _ in range(13): mid = (lo + hi) / 2 if survival(r, 1e6, 1e6 * mid, 20, 1.0, n_paths=2000)["p10_end"] >= 1e6: lo = mid else: hi = mid cap = gross_needed / lo if lo > 0.002 else float("inf") vol = float(r.std() * np.sqrt(365)) # ISO-RISCHIO: e' QUI che la diversificazione paga. A pari nozionale un diversificatore # a basso CAGR diluisce il ritorno; il suo valore e' che LIBERA BUDGET DI RISCHIO. Si # confronta portando ogni tier alla STESSA vol (15%), non allo stesso nozionale. k_iso = min(VOL_BUDGET / vol, 3.0) lo_i, hi_i = 0.0, 0.60 for _ in range(13): mid = (lo_i + hi_i) / 2 if survival(r, 1e6, 1e6 * mid, 20, k_iso, n_paths=2000)["p10_end"] >= 1e6: lo_i = mid else: hi_i = mid cap_i = gross_needed / lo_i if lo_i > 0.002 else float("inf") print(f" {name:>42} {m['sharpe']:>7.2f} {m['cagr']:>7.1%} {vol:>6.1%} " f"{m['maxdd']:>7.1%} {lo:>8.2%} " f"{('$' + format(cap, ',.0f')) if np.isfinite(cap) else 'IRRAGG.':>18}" f" | {k_iso:>4.2f}x {lo_i:>7.2%} " f"{('$' + format(cap_i, ',.0f')) if np.isfinite(cap_i) else 'IRRAGG.':>18}") out[name] = dict(perp=lo, cap=cap, metrics=m, k_iso=k_iso, perp_iso=lo_i, cap_iso=cap_i) print("\n (tutti de-luck x0.6, SKH01 su path live, GTAA01 a costo IB reale con banda.") print(" 'rendita' = prelievo perpetuo al 90% di preservazione del capitale.)") return out if __name__ == "__main__": part_a() part_a2() part_b() part_c() res = part_c2() part_e() key = ("hourly", 0.6, 1.0) part_d(res[key]["perp"], res[key]["cap"])