#!/usr/bin/env python """r0823_book_3rd.py — §50 BOOK-3RD: qual e' il MIGLIOR TERZO SLEEVE del LIBRO LIVE, e vale la pena? IL PROBLEMA (e perche' non e' una domanda gia' fatta). Il progetto ha DUE oggetti che vengono continuamente confusi: * il BOOK DI RICERCA a 5 sleeve (TP01 33 / XS01 15 / VRP01 12 / SKH01 20 / GTAA01 20) — `sleeves.active_sleeves()` — su cui e' calcolato quasi ogni numero pubblicato; * il LIBRO LIVE a 2 sleeve (`sleeves.deribit_book_sleeves()`, TP01 75 / SKH01 25) — l'unica cosa che gira con soldi veri, e che non ha MAI avuto un terzo sleeve. Il libro a 5 non e' deployabile: GTAA01 e' bloccato dal PRIIPs (verificato sul conto reale il 26/07) e vuole $3k; XS01 e' dichiarato STAT-MODE per taglia; VRP01 cade sotto la regola "niente short-vol da modello in deploy" e a $3.000 sarebbe ETH-only con 0 lotti al peso di book. DOMANDA DECIDIBILE: oggi, con ~$635 su Deribit, quale e' il miglior terzo sleeve REALMENTE eseguibile — e vale la pena? Se la risposta e' "nessuno": a quale capitale cambia, e chi entra? COSA FA 0. ATTESA A PRIORI, dichiarata PRIMA di misurare (e poi confrontata con l'esito). 1. CENSIMENTO dello stato REALE dei candidati — non quello dichiarato in memoria: strumenti richiesti, disponibilita' su Deribit (GET pubblica), LOTTO MINIMO letto dal venue (non stimato), data di quotazione, gate pre-registrato in corso. 2. IL VINCOLO DIMENTICATO — il NETTING. Su Deribit c'e' UNA sola posizione netta per strumento, e TP01+SKH01 gia' si nettano su BTC/ETH_USDC-PERPETUAL. Un terzo sleeve DIREZIONALE sugli stessi strumenti si netta con loro (zero costo di strumento, zero separazione); uno su strumenti diversi no (serve una riga in `_CONTRACT`, cioe' CODICE su un percorso con soldi veri). Verificato per introspezione sul sorgente di produzione. 3. REPLICA DI CONTROLLO: riprodurre i numeri pubblicati del libro live (ShFULL ~1,81 / ShHOLD ~1,44 / maxDD ~9,4%) PRIMA di stampare qualunque delta. Due lenti, dichiarate. 4. SERIE ANCORATE, ognuna con sanity bit-exact contro lo sleeve/segnale di produzione. 5. LA MISURA: dSharpe / d drift / d maxDD del libro live aggiungendo ogni candidato a pesi {5,10,15,20}%, col FUNDING dentro, su ESTRAZIONI CONGIUNTE dello spazio d'ancora (mediana delle DIFFERENZE APPAIATE — mai la differenza delle mediane). 6. `weights_tilt_null` — il gate del progetto per ogni cambio di pesi. 7. ESEGUIBILITA' a $635 / $2k / $5k / $20k: ticket per gamba, min-order, haircut. 8. LA SCALA: a quale capitale entra il primo candidato, e quale. ONESTA'. Nessuna promozione. L'output e' una TABELLA DECIDIBILE. I gate pre-registrati NON si anticipano: per i candidati sotto gate si dice cosa succederebbe SE passassero, non si decide al loro posto (anticipare un gate e' selezione-sull'hold-out, regola scritta del progetto). VINCOLI RISPETTATI: nessun file di produzione toccato (src/, config/, scripts/live/, scripts/cron_*.sh, tests/, data/paper_*, data/live); NESSUN ordine; rete = sole GET pubbliche Deribit <=2 req/s, con astensione nei minuti :05-:10 e :24-:30 (cron_book al :07, collect_chain al :25). Cache in scratchpad, fuori dal repo. USO nice -n 19 timeout 900 uv run python scripts/research/r0823_book_3rd.py ... --no-net solo cache (nessuna richiesta di rete) ... --draws N estrazioni d'ancora (default 200) ... --force ignora la cache delle serie """ from __future__ import annotations import argparse import datetime as dt import inspect import json import pickle import sys import time import urllib.request from pathlib import Path import numpy as np import pandas as pd ROOT = Path("/opt/docker/PythagorasGoal") for _p in (str(ROOT), str(ROOT / "scripts" / "research"), str(ROOT / "scripts" / "research" / "alt"), str(ROOT / "scripts" / "research" / "ortho")): if _p not in sys.path: sys.path.insert(0, _p) # --- produzione (importata, MAI riscritta) --------------------------------------------------- from src.portfolio.portfolio import (HOLDOUT, combine_outer, metrics, # noqa: E402 to_daily, weights_tilt_null) from src.portfolio import sleeves as SL # noqa: E402 from src.live import book as LB # noqa: E402 from src.live import deribit as DB # noqa: E402 from src.strategies.trend_portfolio import (CANONICAL, TrendPortfolio, # noqa: E402 simple_returns) from src.strategies.prevday_breakout import target as prevday_target # noqa: E402 # --- macchineria di ricerca gia' validata (riusata, non riscritta) --------------------------- import altlib as al # noqa: E402 import ortholib as ol # noqa: E402 import orthogonal_signals as OS # noqa: E402 (STATARB-RESID: segnale ESATTO) import r0702_tp01_offset as A_TP # noqa: E402 (daily_off: ancora oraria di TP01) import r0703_vrpimp_anchor as A_VRP # noqa: E402 (fasi settimanali di VRP01) import r0822d_funding as FU # noqa: E402 (funding: fetch/rate/cumF/skh_daily) from r0726_dvolspread_gate import make_book as dvs_make_book # noqa: E402 # ============================================================================================= # COSTANTI DICHIARATE # ============================================================================================= SCRATCH = Path("/tmp/claude-1001/-opt-docker-PythagorasGoal/" "b6cc75e7-14f8-4c32-bd07-ab8a0d2aaee6/scratchpad/b3rd") if not SCRATCH.parent.exists(): SCRATCH = Path("/tmp/pythagoras_book3rd") CACHE = SCRATCH / "series_v3.pkl" CAPITAL_TODAY = 635.0 # il conto reale (CLAUDE.md 2026-08) CAP_LADDER = (635.0, 2000.0, 5000.0, 20000.0) MIN_ORDER = 5.0 # config/live.json -> min_order_usd FEE_SIDE = al.FEE_SIDE # 0.05%/lato (0.10% RT), il default del progetto W_TP, W_SKH = LB.W_TP01, LB.W_SKH # 0.75 / 0.25, LETTI dalla produzione BOOK_START = pd.Timestamp("2019-03-14", tz="UTC") WEIGHTS_TESTED = (0.05, 0.10, 0.15, 0.20) TP_HOURS = tuple(range(24)) # ancora giornaliera di TP01 SKH_OFFSETS = tuple(range(0, 690, 30)) # 23 offset della griglia 230m/690m XS_PHASES = tuple(range(10)) # ciclo H=10 di XS01 VRP_PHASES = tuple(range(7)) # ciclo settimanale di VRP01 SEED = 20260823 # XS01 paga funding perche' il momentum compra i perp col funding piu' caro: +1,45%/anno # MISURATO nell'ondata 2026-08-22 §37 (banda onesta +0,69/+1,45 se HL e' ~2,1x il venue vero). # Qui e' applicato come DRAG DICHIARATO, non modellato per gamba: gli alt su Deribit sono # quotati da giorni (vedi sezione 1) e non ne esiste storico di funding. XS_FUNDING_DRAG_YR = 0.0145 API = "https://www.deribit.com/api/v2/public" CACHE_FRESH_S = 6 * 3600 T0 = time.time() _ARGS = None def sez(t: str) -> None: print("\n" + "=" * 104) print(f" {t}") print("=" * 104) def sub(t: str) -> None: print("\n" + "-" * 104) print(f" {t}") print("-" * 104) def clock() -> str: return f"[{time.time() - T0:5.0f}s]" # ============================================================================================= # RETE — sole GET pubbliche, pacing, astensione nelle finestre del cron # ============================================================================================= def _pace() -> None: while dt.datetime.now(dt.timezone.utc).minute in (5, 6, 7, 8, 9, 10, 24, 25, 26, 27, 28, 29, 30): time.sleep(15) time.sleep(0.55) def get_json(path: str, cache_name: str) -> dict | None: """GET pubblica con cache su scratchpad. Con --no-net usa solo la cache.""" SCRATCH.mkdir(parents=True, exist_ok=True) fp = SCRATCH / cache_name if _ARGS and _ARGS.no_net: return json.loads(fp.read_text()) if fp.exists() else None # cache fresca -> NESSUNA richiesta, e quindi nessuna attesa nelle finestre del cron. # (`_pace` blocca 15s per volta nei minuti :05-:10 e :24-:30: farlo quando non si sta # per chiamare nessuno e' solo tempo perso.) if fp.exists() and (time.time() - fp.stat().st_mtime) < CACHE_FRESH_S: return json.loads(fp.read_text()) try: _pace() with urllib.request.urlopen(f"{API}/{path}", timeout=30) as r: d = json.loads(r.read()) fp.write_text(json.dumps(d)) return d except Exception as e: # noqa: BLE001 print(f" [rete KO su {path}: {type(e).__name__}] -> cache") return json.loads(fp.read_text()) if fp.exists() else None # ============================================================================================= # 0) ATTESA A PRIORI — dichiarata PRIMA di misurare # ============================================================================================= ATTESA = """ Registrata prima di guardare qualunque numero di questo script: (A1) NESSUN candidato e' deployabile oggi, e il muro NON sara' il lotto minimo del venue. Nell'ondata del 22/08 l'eseguibilita' a $600 non ha bocciato niente per TRE volte di fila; mi aspetto che il muro sia (a) i gate pre-registrati non scaduti, (b) `weights_tilt_null`, (c) per XS01/XSR01 il fatto che il venue e' Hyperliquid, non Deribit. (A2) Il candidato col dSharpe piu' alto sara' XS01 (LOO 26/07: +0,125 di FULL nel libro a 5, corr ~0 a tutto), ma `weights_tilt_null` FALLIRA' per la stessa ragione strutturale per cui fallisce su XSR01: storia dal 2024, quindi quasi nessun in-sample. (A3) I due candidati "eseguibili a $600" gia' etichettati tali (STATARB-RESID, DVOLSPREAD) daranno un uplift piccolo (Sharpe standalone 0,75 e 0,61 contro 1,81 del libro): mi aspetto |dSharpe| < 0,10 e il segno incerto sulla banda d'ancora. (A4) La scala: il primo a entrare sara' XS01 fra $2k e $5k, NON ai $20k della memoria — ma per una ragione di VENUE, non di taglia (serve che Deribit quoti abbastanza gambe). (A5) Se qualcosa mi sorprende, sara' un candidato dato per eseguibile che NON lo e' (e' gia' successo due volte: GTAA01/PRIIPs e VRP01/lotto minimo). """ # ============================================================================================= # 1) CENSIMENTO — lo stato REALE dei candidati # ============================================================================================= # I 19 major di XS01, letti dalla PRODUZIONE (non riscritti: se cambiano li', cambiano qui). XS_UNIVERSE = list(SL.XS_UNIVERSE) # gate pre-registrati: (nome, data, cosa decide). Letti da CLAUDE.md / docstring dei monitor. GATES = { "STATARB-RESID": ("2026-09-27", "decisione di deploy (gate `r0724_statarb_deploy_gate`)"), "XSR01": ("2026-10-23", "deploy solo se Sharpe>=1,0 E haircut a $5k <=40%"), "DVOLSPREAD": ("2026-10-24", "kill se Sharpe forward < -0,50; decisione 2027-01-24"), } def venue_census() -> dict: """Strumenti Deribit REALI: perpetual USDC, spot, opzioni. Lotto minimo in DOLLARI.""" out = {"perp": {}, "spot": {}, "opt": {}, "read_ok": True} ins = {} for cur in ("BTC", "ETH", "USDC"): d = get_json(f"get_instruments?currency={cur}&expired=false", f"instr_{cur}.json") if d is None: out["read_ok"] = False continue ins[cur] = d["result"] bs = get_json("get_book_summary_by_currency?currency=USDC&kind=future", "bs_usdc_fut.json") marks = {} if bs: for x in bs["result"]: p = x.get("mark_price") or x.get("last") if p: marks[x["instrument_name"]] = (float(p), float(x.get("volume_usd") or 0.0)) for cur, lst in ins.items(): for x in lst: n = x["instrument_name"] if x["kind"] == "future" and x.get("settlement_period") == "perpetual": mk, vol = marks.get(n, (None, None)) out["perp"][x["base_currency"]] = dict( instr=n, min_amt=float(x["min_trade_amount"]), created=dt.datetime.fromtimestamp(x["creation_timestamp"] / 1000, dt.timezone.utc).date().isoformat(), mark=mk, min_usd=(mk * float(x["min_trade_amount"])) if mk else None, vol24=vol) elif x["kind"] == "spot": out["spot"][n] = dict(min_amt=float(x["min_trade_amount"])) elif x["kind"] == "option": key = (cur, x["base_currency"]) out["opt"].setdefault(key, set()).add(float(x["min_trade_amount"])) return out def sezione1(V: dict) -> dict: sez("1) CENSIMENTO — lo stato REALE dei candidati (venue letto ORA, non memoria)") if not V["read_ok"]: print(" ⚠️ lista strumenti NON letta dal venue in questa corsa: si usa la cache.") sub("1a) I 19 major di XS01 contro cio' che Deribit quota DAVVERO (perp USDC lineari)") print(f" {'asset':<7}{'su Deribit':<12}{'strumento':<26}{'quotato dal':<13}" f"{'lotto min':>11}{'vol 24h $':>14}") xs_ok, xs_stable = [], [] oggi = dt.date.today() for s in XS_UNIVERSE: p = V["perp"].get(s) if not p: print(f" {s:<7}{'NO':<12}{'—':<26}{'—':<13}{'—':>11}{'—':>14}") continue xs_ok.append(s) eta = (oggi - dt.date.fromisoformat(p["created"])).days if eta >= 365: xs_stable.append(s) mu = f"${p['min_usd']:.2f}" if p["min_usd"] else "n/d" v = f"{p['vol24']:,.0f}" if p["vol24"] is not None else "NON QUOTATO" print(f" {s:<7}{'SI':<12}{p['instr']:<26}{p['created']:<13}{mu:>11}{v:>14}") print(f"\n -> gambe su Deribit: {len(xs_ok)}/19 ({', '.join(xs_ok)})") print(f" -> di cui quotate da >=1 anno: {len(xs_stable)}/19 ({', '.join(xs_stable)})") print(" ⚠️ Deribit sta ampliando la lista PROPRIO ORA: le gambe piu' recenti hanno giorni") print(" di vita e volume ~0. Una gamba quotata NON e' una gamba negoziabile.") sub("1b) LOTTO MINIMO DELLE OPZIONI — la famiglia conta (5a occorrenza dello schema fee_watch)") for (cur, base), mins in sorted(V["opt"].items()): if base not in ("BTC", "ETH"): continue m = min(mins) mk = (V["perp"].get(base) or {}).get("mark") fam = "USDC-LINEARE" if cur == "USDC" else "INVERSE" usd = f"${m * mk:,.0f}" if mk else "n/d" print(f" {fam:<14} {base:<4} min {m:<8} = {usd:>9} di sottostante per lotto") print("\n 📌 Il numero pubblicato il 30/07 (\"BTC min 0.1 = $6.210 -> FUORI; ETH 1 contratto") print(" = $1.832\") e' della famiglia INVERSE, che un conto USDC non puo' marginare.") print(" Sulla famiglia che il conto PUO' usare il lotto e' ~10x piu' piccolo.") return dict(xs_deribit=xs_ok, xs_stable=xs_stable) def sezione1c(V: dict, cens: dict) -> None: sub("1c) LA TABELLA — stato REALE di ogni candidato a terzo sleeve") eth = (V["perp"].get("ETH") or {}).get("mark") or 0.0 btc = (V["perp"].get("BTC") or {}).get("mark") or 0.0 opt_eth = min(V["opt"].get(("USDC", "ETH"), {0.1})) * eth opt_btc = min(V["opt"].get(("USDC", "BTC"), {0.01})) * btc n_d, n_s = len(cens["xs_deribit"]), len(cens["xs_stable"]) rows = [ ("STATARB-RESID", "BTC+ETH perp USDC", "SI", f"${MIN_ORDER:.0f} (soglia NOSTRA)", "SI — netta", "27/09 (aperto)"), ("DVOLSPREAD", "BTC+ETH perp USDC", "SI", f"${MIN_ORDER:.0f} (soglia NOSTRA)", "SI — netta", "24/10 kill · 24/01 dec."), ("PREVDAY", "BTC+ETH perp USDC", "SI", f"${MIN_ORDER:.0f} (soglia NOSTRA)", "SI — netta", "nessuno (mai registrato)"), ("XS01 (19 gambe)", "19 alt", "NO — 5 mancano", f"{n_d}/19 quotate, {n_s} da >=1a", "PARZIALE (BTC/ETH)", "nessuno"), (f"XS01-D ({n_s} gambe)", f"{n_s} alt su Deribit", "SI (ridotta)", "$0,09-$7,72 dal venue", "PARZIALE (BTC/ETH)", "nessuno"), ("XSR01", "50 alt Hyperliquid", "NO — venue diverso", "n/d su Deribit", "NO", "23/10 (aperto)"), ("VRP01", "opzioni BTC/ETH USDC", "SI (ETH)", f"lotto ETH ${opt_eth:,.0f} · BTC ${opt_btc:,.0f}", "NO — altro strumento", "nessuno (regola short-vol)"), ("GTAA01", "6 ETF USA (IB)", "NO — altro venue", "PRIIPs: ordine RIFIUTATO", "NO", "nessuno"), ] print(f" {'candidato':<19}{'strumenti':<24}{'su Deribit?':<19}{'capitale/lotto min':<28}" f"{'netta col libro?':<22}{'gate'}") print(" " + "-" * 100) for r in rows: print(f" {r[0]:<19}{r[1]:<24}{r[2]:<19}{r[3]:<28}{r[4]:<22}{r[5]}") print("\n Finestra FORWARD dei monitor (SOLA LETTURA di data/paper_*/, nessuna scrittura):") print(f" {'monitor':<12}{'barre':>7}{'ultima':>13}{'Sh fwd MODELED':>17} stato §32") for nm, d, cad, sano in (("STATARB", "paper_statarb", 365.25, False), ("XSR01", "paper_xsr", 365.25, False), ("DVOLSPREAD", "paper_dvolspread", 365.25, False), ("PREVDAY", "paper_prevday", 365.25 * 24, True)): try: st = json.loads((ROOT / "data" / d / "state.json").read_text()) ts = pd.Timestamp(int(st["last_ts"]), unit="ms", tz="UTC") rr = [json.loads(l)["net_modeled"] for l in (ROOT / "data" / d / "returns.jsonl").read_text().splitlines() if l.strip()] v = np.asarray(rr, float) sh = (float(v.mean() / v.std() * np.sqrt(cad)) if len(v) > 3 and v.std() > 0 else float("nan")) flag = "SANO (1259/1319)" if sano else "ROTTO: frazione di giorno" print(f" {nm:<12}{st.get('n_bars', len(v)):>7}{ts.strftime('%Y-%m-%d'):>13}" f"{sh:>17.2f} {flag}") except Exception as e: # noqa: BLE001 print(f" {nm:<12} stato illeggibile ({type(e).__name__})") print(" ⚠️ §32 (22/08): `advance()` consuma la barra del giorno IN CORSO -> 4 monitor su 6") print(" registrano una FRAZIONE del giorno (STATARB ~4 min/giorno): quei tre Sharpe") print(" NON sono un risultato. `paper_prevday` e' l'unico misurato SANO (orario), ed e'") print(" l'unica lettura forward indipendente disponibile in questo filone.") print(" ⚠️ Anche quello sano ha ~63 giorni: SE(Sharpe) ~ 2,3 -> non decide niente da solo.") print("\n ⚠️ 'gate aperto' = pre-registrato e NON scaduto: in questo script si dice cosa") print(" succederebbe SE passasse, non si decide al suo posto (anticipare un gate") print(" pre-registrato E' selezione-sull'hold-out, ed e' una regola scritta).") # ============================================================================================= # 2) IL NETTING — misurato sul CODICE di produzione, non ricordato # ============================================================================================= def sezione2() -> None: sez("2) IL VINCOLO DIMENTICATO — il NETTING su strumento unico (letto dal sorgente)") src_net = inspect.getsource(LB.book_net_target) src_ord = inspect.getsource(LB.build_book_order) print(f" src/live/deribit.INSTRUMENT = {DB.INSTRUMENT}") print(f" src/live/deribit._CONTRACT = {len(DB._CONTRACT)} strumenti dichiarati: " f"{list(DB._CONTRACT)}") print(f" src/live/book.W_TP01 / W_SKH = {W_TP} / {W_SKH}") uno = ("W_TP01 * max(tp_frac" in src_net) and ("W_SKH * float(skh_sign)" in src_net) print(f" `book_net_target` somma i due sleeve in UN solo numero per asset : {uno}" f" (riga: `raw = weight * equity * (W_TP01*tp_frac + W_SKH*skh_sign)`)") print(f" `build_book_order` manda UN ordine per asset, soglia min_usd : " f"{'min_usd' in src_ord}") print("\n CONSEGUENZA, e nessuno l'aveva messa in tabella:") print(" * un terzo sleeve DIREZIONALE su BTC/ETH e' un TERZO ADDENDO della stessa formula:") print(" non aggiunge un ordine, non aggiunge un min-order, non aggiunge margine lordo —") print(" e non aggiunge nemmeno SEPARAZIONE (nessuna gamba puo' avere il proprio stop).") print(" E' l'esatto motivo per cui la raccomandazione T1 del 26/07 fu abbandonata.") print(" * un terzo sleeve su ALTRI strumenti (opzioni, spot, alt-perp) richiede una riga in") print(f" `_CONTRACT` (oggi {len(DB._CONTRACT)} voci, CABLATE A MANO) piu' il ramo di") print(" sizing corrispondente = CODICE su un percorso con soldi veri, piu' un lotto") print(" minimo e un margine propri.") print(" * il min-order si applica al DELTA NETTO: due sleeve che si contraddicono possono") print(" produrre zero ordini. E' un vantaggio di costo E una perdita di controllo.") # ============================================================================================= # 3) IL FUNDING (importato) + 4) LE SERIE ANCORATE # ============================================================================================= def get_rate() -> dict: """Tassi di funding orari per asset (LINEARE dove esiste, INVERSE come proxy prima).""" F = {} for instr, (start, a, kind) in FU.INSTR.items(): d = FU.fetch_funding(instr, start).copy() d["dt"] = pd.to_datetime(d["ts"], unit="ms", utc=True) F[instr] = d.drop_duplicates(subset="ts").sort_values("dt").set_index("dt") return FU.build_rate(F) def tp01_at(h: int, rate: dict | None) -> pd.Series: """TP01 all'ancora oraria h (replica di sleeves._tp01_returns), funding opzionale.""" tp = TrendPortfolio(**CANONICAL) series = {} for a in ("BTC", "ETH"): df = A_TP.daily_off(a, h) idx = pd.DatetimeIndex(pd.to_datetime(df["datetime"], utc=True)) r = simple_returns(df["close"].values.astype(float)) tgt = tp.target_series(df) held = np.zeros(len(tgt)); held[1:] = tgt[:-1] net = held * r - tp.fee_side * np.abs(np.diff(held, prepend=0.0)) if rate is not None: net = net - held * FU.daily_funding(rate[a], idx) net[0] = 0.0 series[a] = pd.Series(np.clip(net, -0.99, None), index=idx) J = pd.concat(series, axis=1, join="inner").fillna(0.0) return to_daily(pd.Series(0.5 * J["BTC"].values + 0.5 * J["ETH"].values, index=J.index)) def skh_at(off: int, rate: dict | None) -> pd.Series: """SKH01 all'offset `off` della griglia 230m, lente CANONICA (quella dello sleeve).""" return to_daily(FU.skh_book(off, "canonical", rate)) def _joint_off(h: int) -> pd.DataFrame: """Frame BTC/ETH allineato all'ancora oraria h, colonne di OS.build_joint.""" b = A_TP.daily_off("BTC", h)[["timestamp", "datetime", "close"]].rename(columns={"close": "cb"}) e = A_TP.daily_off("ETH", h)[["timestamp", "close"]].rename(columns={"close": "ce"}) j = b.merge(e, on="timestamp", how="inner").sort_values("timestamp").reset_index(drop=True) j["r_btc"] = al.simple_returns(j["cb"].values) j["r_eth"] = al.simple_returns(j["ce"].values) j["log_ratio"] = np.log(j["ce"].values / j["cb"].values) return j def statarb_at(h: int, rate: dict | None): """STATARB-RESID (W=45, sgn=+1) all'ancora oraria h. Ritorna (serie daily, pos).""" j = _joint_off(h) pos = np.nan_to_num(np.asarray(OS.f_statarb_resid(W=45, sgn=+1)(j), float)) idx = pd.DatetimeIndex(pd.to_datetime(j["datetime"], utc=True)) s = OS.spread_ret(j) held = np.zeros(len(pos)); held[1:] = pos[:-1] net = held * s - FEE_SIDE * 2.0 * np.abs(np.diff(held, prepend=0.0)) if rate is not None: # +1 = long ETH / short BTC -> paga f_eth - f_btc net = net - held * (FU.daily_funding(rate["ETH"], idx) - FU.daily_funding(rate["BTC"], idx)) net[0] = 0.0 return to_daily(pd.Series(net, index=idx)), pos # pos NON shiftata (7b shifta da se') DVS_FROZEN = dict(zwin=180, tanh_k=2.0, lw=0.6, zw=1.1, tgt=0.17, svw=60) def dvolspread(rate: dict | None): """DVOLSPREAD coi parametri CONGELATI del monitor. Ancora non variabile: il DVOL e' 1d.""" btc, eth = ol.aligned() wb, we = dvs_make_book(**DVS_FROZEN)(btc, eth) wb = np.clip(np.nan_to_num(np.asarray(wb, float)), -1, 1) we = np.clip(np.nan_to_num(np.asarray(we, float)), -1, 1) n = len(btc) rb = al.simple_returns(btc["close"].values.astype(float)) re = al.simple_returns(eth["close"].values.astype(float)) pb = np.zeros(n); pb[1:] = wb[:-1] pe = np.zeros(n); pe[1:] = we[:-1] net = pb * rb + pe * re - FEE_SIDE * (np.abs(np.diff(pb, prepend=0.0)) + np.abs(np.diff(pe, prepend=0.0))) idx = pd.DatetimeIndex(btc["dt"]) if rate is not None: net = net - (pb * FU.daily_funding(rate["BTC"], idx) + pe * FU.daily_funding(rate["ETH"], idx)) net[0] = 0.0 assert float(np.max(np.abs(wb + we))) < 1e-12, "il book DVOLSPREAD non e' we == -wb" return to_daily(pd.Series(net, index=idx)), wb # gamba BTC, NON shiftata def prevday(rate: dict | None): """PREVDAY range-breakout 1h, libro 50/50. Funding ORARIO (il segnale e' a 1h).""" series, tk = {}, [] for a in ("BTC", "ETH"): df = al.get(a, "1h") idx = pd.DatetimeIndex(pd.to_datetime(df["datetime"], utc=True)) tgt = np.nan_to_num(np.asarray(prevday_target(df), float)) r = al.simple_returns(df["close"].values.astype(float)) pos = np.zeros(len(tgt)); pos[1:] = tgt[:-1] net = pos * r - FEE_SIDE * np.abs(np.diff(pos, prepend=0.0)) if rate is not None: t = FU.to_ms(idx) fh = FU.cumF(rate[a], t + FU.MSH) - FU.cumF(rate[a], t) net = net - pos * fh net[0] = 0.0 series[a] = pd.Series(net, index=idx) tk.append(np.abs(np.diff(pos, prepend=0.0))) J = pd.concat(series, axis=1, join="inner").fillna(0.0) d = to_daily(pd.Series(0.5 * J["BTC"].values + 0.5 * J["ETH"].values, index=J.index)) return d, np.concatenate(tk) def xs_at(universe: list[str], phase: int, funding_drag: float = 0.0): """XS01 su un universo arbitrario, alla fase `phase` del ciclo H. Replica ESATTA della formula di sleeves._xsec_returns (verificata bit-exact a universo pieno / fase 0).""" cols = {} for sym in universe: p = SL._HL_DIR / f"hl_{sym.lower()}_1d.parquet" if not p.exists(): continue d = pd.read_parquet(p) cols[sym] = pd.Series(d["close"].values.astype(float), index=pd.to_datetime(d["timestamp"], unit="ms", utc=True)) C = pd.concat(cols, axis=1, join="inner").sort_index().dropna() px = C.values; n, A = px.shape cfg = SL.XS_CFG lookbacks, H, k, mode, tv = cfg["lookbacks"], cfg["H"], cfg["k"], cfg["mode"], cfg["target_vol"] disp_pct, minhist = cfg.get("disp_pct", 0), cfg.get("disp_minhist", 20) kk = min(k, A // 2) mlb = max(lookbacks) dret = np.vstack([np.zeros(A), px[1:] / px[:-1] - 1.0]) W = np.zeros((n, A)); w = np.zeros(A); disp_hist = [] for i in range(n): if i >= mlb and i % H == phase: rLs = [px[i] / px[i - L] - 1.0 for L in lookbacks] disp_i = float(np.mean([r.std() for r in rLs])) thr = (np.percentile(disp_hist, disp_pct) if (disp_pct > 0 and len(disp_hist) >= minhist) else -np.inf) if disp_i >= thr: score = np.zeros(A); cnt = 0 for rL in rLs: sd = rL.std() if sd > 0: score += (rL - rL.mean()) / sd; cnt += 1 if cnt: score /= cnt order = np.argsort(score) w = np.zeros(A); lo, hi = order[:kk], order[-kk:] if mode == "mom": w[hi] = 0.5 / kk; w[lo] = -0.5 / kk else: w[lo] = 0.5 / kk; w[hi] = -0.5 / kk else: w = np.zeros(A) disp_hist.append(disp_i) W[i] = w gross = np.zeros(n); gross[1:] = np.sum(W[:-1] * dret[1:], axis=1) turn = np.zeros(n); turn[0] = np.abs(W[0]).sum() turn[1:] = np.abs(np.diff(W, axis=0)).sum(axis=1) net = gross - turn * (0.001 / 2.0) s = pd.Series(net, index=C.index) rv = s.rolling(30, min_periods=15).std().shift(1) * np.sqrt(365.25) scale = np.clip(np.nan_to_num(tv / rv.replace(0, np.nan).values, nan=0.0), 0, 3.0) out = pd.Series(s.values * scale, index=C.index) if funding_drag: out = out - funding_drag / 365.25 # ticket per gamba: |ΔW| a ogni ribilanciamento, scalato dal vol-target del giorno tick = np.abs(np.diff(W, axis=0, prepend=0.0)) * scale[:, None] return to_daily(out), tick, list(C.columns) def vrp_at(phase: int) -> pd.Series: return to_daily(A_VRP.to_daily_lumped(A_VRP.combo_weekly(phase, f=SL.VRP_CFG["f"]))) # ============================================================================================= # 3) REPLICA DI CONTROLLO # ============================================================================================= def sezione3(rate: dict) -> dict: sez("3) REPLICA DI CONTROLLO — riprodurre i numeri PUBBLICATI prima di stampare un delta") from src.portfolio.portfolio import StrategyPortfolio P = StrategyPortfolio(SL.deribit_book_sleeves(), capital=CAPITAL_TODAY) ref = P.combined_daily() m, mh = metrics(ref), metrics(ref[ref.index >= HOLDOUT]) print(" LENTE A — 'sleeve' (`deribit_book_sleeves` via StrategyPortfolio): e' quella dei") print(" numeri pubblicati. SKH01 con fill AL LIVELLO, TP01 senza min-order,") print(" NESSUN funding.") print(f" ShFULL {m['sharpe']:.3f} ShHOLD {mh['sharpe']:.3f} maxDD {100*m['maxdd']:.2f}%" f" CAGR {100*m['cagr']:.2f}% n={m['n']}") print(" atteso dal briefing: ShFULL ~1,81 / ShHOLD ~1,44 / maxDD ~9,4% -> " f"{'RIPRODOTTO' if (abs(m['sharpe']-1.81)<0.02 and abs(mh['sharpe']-1.44)<0.02 and abs(100*m['maxdd']-9.4)<0.15) else '⚠️ NON riproduce'}") mine0 = combine_outer({"TP01": tp01_at(0, None), "SKH01": skh_at(0, None)}, {"TP01": W_TP, "SKH01": W_SKH}, lo=BOOK_START) j = pd.concat({"a": mine0, "b": ref}, axis=1, join="outer") dmax = float(np.nanmax(np.abs(j["a"].values - j["b"].values))) print(f"\n [SANITY] la mia replica ancorata a (h=0, off=0) vs lo sleeve di produzione: " f"max|dif| = {dmax:.3e} {'OK' if dmax < 1e-12 else 'FAIL'}") assert dmax < 1e-12, "replica ancorata NON bit-exact: tutto il resto misurerebbe altro" withf = combine_outer({"TP01": tp01_at(0, rate), "SKH01": skh_at(0, rate)}, {"TP01": W_TP, "SKH01": W_SKH}, lo=BOOK_START) mf, mfh = metrics(withf), metrics(withf[withf.index >= HOLDOUT]) print(f"\n LENTE A + FUNDING (misurato §35, non modellato prima del 22/08):") print(f" ShFULL {mf['sharpe']:.3f} ShHOLD {mfh['sharpe']:.3f} " f"maxDD {100*mf['maxdd']:.2f}% CAGR {100*mf['cagr']:.2f}%") print(f" costo del funding: dSharpe {mf['sharpe']-m['sharpe']:+.3f}, " f"dCAGR {100*(mf['cagr']-m['cagr']):+.2f} pp") lp = FU.book_75_25(rate, 0, "hourly", 600.0) mlp, mlph = metrics(lp), metrics(lp[lp.index >= HOLDOUT]) print("\n LENTE B — 'path live' (`CC.book_series('hourly')` + funding): min-order $5, fill al") print(" close 5m del prossimo confine orario. E' cio' che il cron esegue.") print(f" ShFULL {mlp['sharpe']:.3f} ShHOLD {mlph['sharpe']:.3f} " f"maxDD {100*mlp['maxdd']:.2f}% CAGR {100*mlp['cagr']:.2f}%") print("\n 📌 Le due lenti NON danno lo stesso libro (1,81 contro 1,50 di Sharpe FULL). La") print(" misura di questo filone gira sulla LENTE A col funding, perche' e' quella in cui") print(" sono espressi TUTTI i candidati (serie di sleeve) e quella dei numeri pubblicati.") print(" Il delta e' comunque un confronto APPAIATO: la lente si cancella in gran parte.") return dict(ref=ref, m=m, mh=mh, base_f=withf) # ============================================================================================= # 4) COSTRUZIONE DELLE SERIE ANCORATE (con cache) # ============================================================================================= def build_series(rate: dict, cens: dict, force: bool = False) -> dict: if CACHE.exists() and not force: try: S = pickle.loads(CACHE.read_bytes()) if S.get("_universe") == (cens["xs_stable"], cens["xs_deribit"]): print(f" {clock()} serie da cache ({CACHE.name})") return S except Exception: # noqa: BLE001 pass S: dict = {"_universe": (cens["xs_stable"], cens["xs_deribit"])} t = time.time() S["TP01"] = {h: tp01_at(h, rate) for h in TP_HOURS} print(f" {clock()} TP01 x{len(TP_HOURS)} ancore orarie ({time.time()-t:.0f}s)") t = time.time() S["SKH01"] = {o: skh_at(o, rate) for o in SKH_OFFSETS} print(f" {clock()} SKH01 x{len(SKH_OFFSETS)} offset di griglia ({time.time()-t:.0f}s)") t = time.time() sa = {h: statarb_at(h, rate) for h in TP_HOURS} S["STATARB"] = {h: v[0] for h, v in sa.items()} S["_statarb_pos"] = sa[0][1] print(f" {clock()} STATARB x{len(TP_HOURS)} ancore ({time.time()-t:.0f}s)") t = time.time() d, dpos = dvolspread(rate) S["DVOLSPREAD"] = {0: d}; S["_dvs_pos"] = dpos p, ptk = prevday(rate) S["PREVDAY"] = {0: p}; S["_prevday_turn"] = ptk print(f" {clock()} DVOLSPREAD + PREVDAY ({time.time()-t:.0f}s)") t = time.time() xs19 = {ph: xs_at(XS_UNIVERSE, ph, XS_FUNDING_DRAG_YR) for ph in XS_PHASES} S["XS01"] = {ph: v[0] for ph, v in xs19.items()} S["_xs19_tick"] = xs19[0][1]; S["_xs19_cols"] = xs19[0][2] xsd = {ph: xs_at(cens["xs_stable"], ph, XS_FUNDING_DRAG_YR) for ph in XS_PHASES} S["XS01-D"] = {ph: v[0] for ph, v in xsd.items()} S["_xsd_tick"] = xsd[0][1]; S["_xsd_cols"] = xsd[0][2] xsd14 = {ph: xs_at(cens["xs_deribit"], ph, XS_FUNDING_DRAG_YR) for ph in XS_PHASES} S["XS01-D14"] = {ph: v[0] for ph, v in xsd14.items()} S["_xsd14_cols"] = xsd14[0][2] print(f" {clock()} XS01 (19 gambe) + XS01-D ({len(cens['xs_stable'])} gambe) x10 fasi " f"({time.time()-t:.0f}s)") t = time.time() S["VRP01"] = {ph: vrp_at(ph) for ph in VRP_PHASES} print(f" {clock()} VRP01 x{len(VRP_PHASES)} fasi ({time.time()-t:.0f}s)") CACHE.parent.mkdir(parents=True, exist_ok=True) CACHE.write_bytes(pickle.dumps(S)) return S def sezione4(S: dict, cens: dict) -> None: sez("4) LE SERIE ANCORATE — sanity obbligatorio, poi il profilo standalone") x0, _, cols = xs_at(XS_UNIVERSE, 0, 0.0) ref = to_daily(SL._xsec_returns()) jj = pd.concat({"a": x0, "b": ref}, axis=1, join="outer") dm = float(np.nanmax(np.abs(jj["a"].values - jj["b"].values))) print(f" [SANITY] xs_at(19 gambe, fase 0, drag 0) vs sleeves._xsec_returns : " f"max|dif| = {dm:.3e} {'OK' if dm < 1e-12 else 'FAIL'}") assert dm < 1e-12, "la replica XS01 non e' bit-exact" v0 = vrp_at(0); vr = to_daily(SL._vrp_combo_returns()) jj = pd.concat({"a": v0, "b": vr}, axis=1, join="outer") dm2 = float(np.nanmax(np.abs(np.nan_to_num(jj["a"].values) - np.nan_to_num(jj["b"].values)))) print(f" [SANITY] vrp_at(fase 0) vs sleeves._vrp_combo_returns : " f"max|dif| = {dm2:.3e} {'OK' if dm2 < 1e-12 else 'ATTENZIONE'}") jj = OS.build_joint("1d") ev = OS.eval_spread(jj, OS.f_statarb_resid(W=45, sgn=+1)(jj)) mine = metrics(statarb_at(0, None)[0])["sharpe"] print(f" [SANITY] STATARB senza funding, mio percorso {mine:.3f} vs il motore originale " f"`OS.eval_spread` {ev['full']['sharpe']:.3f} (scarto {mine-ev['full']['sharpe']:+.3f})") print(" -> `daily_off(h=0)` e `al.get('1d')` sono lo STESSO resample: due percorsi") print(" indipendenti danno la stessa serie, quindi l'ancora h=0 e' la canonica.") print(f"\n {'candidato':<16}{'da':>12}{'n':>7}{'ShFULL':>9}{'ShHOLD':>9}{'maxDD':>9}" f"{'CAGR':>9}{'corr libro':>12}") base = combine_outer({"TP01": S["TP01"][0], "SKH01": S["SKH01"][0]}, {"TP01": W_TP, "SKH01": W_SKH}, lo=BOOK_START) for nm in CANDIDATI: s = S[nm][sorted(S[nm])[0]] s = s[s.index >= BOOK_START] if nm != "XS01" else s m = metrics(s); mh = metrics(s[s.index >= HOLDOUT]) j = pd.concat({"a": s, "b": base}, axis=1, join="inner").dropna() c = float(j["a"].corr(j["b"])) if len(j) > 60 else float("nan") print(f" {nm:<16}{str(s.index.min().date()):>12}{m['n']:>7}{m['sharpe']:>9.3f}" f"{mh['sharpe']:>9.3f}{100*m['maxdd']:>8.1f}%{100*m['cagr']:>8.1f}%{c:>12.3f}") print(f"\n NB XS01/XS01-D includono il drag di funding dichiarato " f"({100*XS_FUNDING_DRAG_YR:.2f}%/anno, §37); gli altri hanno il funding modellato " "gamba per gamba.") print(f" XS01-D gira su {len(cens['xs_stable'])} gambe: {', '.join(cens['xs_stable'])}") def sezione4b(S: dict) -> None: sub("4b) `marginal_vs_tp01` — il verdetto STANDARD del progetto su ogni candidato") print(" Un candidato direzionale/relative-value si giudica su ADDS (corr, uplift OOS,") print(" alpha residua, robust_oos multicut, has_insample_edge, is_hedge), non sullo Sharpe.") print(f"\n {'candidato':<12}{'verdetto':>10}{'corr':>8}{'robust_oos':>12}" f"{'insample_edge':>15}{'is_hedge':>10}{'beats_noise':>13}") for c in CANDIDATI: d = S[c][sorted(S[c])[0]] try: m = al.marginal_vs_tp01(d) except Exception as e: # noqa: BLE001 print(f" {c:<12} non calcolabile: {type(e).__name__}: {e}") continue cf = m.get("corr_full") print(f" {c:<12}{str(m.get('marginal_verdict')):>10}" f"{(cf if cf is not None else float('nan')):>8.3f}" f"{str(m.get('robust_oos')):>12}{str(m.get('has_insample_edge')):>15}" f"{str(m.get('is_hedge')):>10}{str(m.get('beats_noise_null')):>13}") print("\n ⚠️ Su una serie che comincia nel 2024 (XS01*) `marginal_vs_tp01` ha poco campione") print(" in-sample: il suo verdetto li' e' debole per costruzione, non informativo.") def sezione4c(cens: dict, n_sub: int = 100) -> None: sub("4c) IL COLLASSO DI XS01-D: e' l'AMPIEZZA o e' QUEL sottoinsieme? (null dei subset)") k = len(cens["xs_stable"]) print(f" XS01 a 19 gambe fa Sharpe ~1,27; ristretto alle {k} quotate su Deribit da >=1 anno") print(" crolla. Due spiegazioni possibili, e vanno separate: (a) con poche gambe il") print(f" long-5/short-5 prende 10 posizioni su {k} e non seleziona piu' niente (AMPIEZZA);") print(" (b) le gambe che Deribit non quota sono proprio quelle che portavano l'edge.") print(f" Test: {n_sub} sottoinsiemi CASUALI di {k} gambe fra le 19, stesso meccanismo,") print(" e si guarda dove cade quello di Deribit. (Stessa forma del null dei 6-subset che") print(" il 25/07 mise GTAA6 al 95° pctl.)") rng = np.random.default_rng(SEED) sh = [] for _ in range(n_sub): u = list(rng.choice(XS_UNIVERSE, size=k, replace=False)) sh.append(metrics(xs_at(u, 0, XS_FUNDING_DRAG_YR)[0])["sharpe"]) sh = np.array(sh) mine = metrics(xs_at(cens["xs_stable"], 0, XS_FUNDING_DRAG_YR)[0])["sharpe"] pct = float((sh < mine).mean() * 100.0) print(f"\n {n_sub} sottoinsiemi casuali da {k}: mediana {np.median(sh):.3f} " f"[p10 {np.percentile(sh,10):.3f}, p90 {np.percentile(sh,90):.3f}] " f"min {sh.min():.3f} max {sh.max():.3f}") print(f" il sottoinsieme DERIBIT: {mine:.3f} -> {pct:.0f}° percentile") print(f" riferimento a 19 gambe: {metrics(xs_at(XS_UNIVERSE, 0, XS_FUNDING_DRAG_YR)[0])['sharpe']:.3f}") full = metrics(xs_at(XS_UNIVERSE, 0, XS_FUNDING_DRAG_YR)[0])["sharpe"] med = float(np.median(sh)) q_ampiezza = (full - med) / (full - mine) if abs(full - mine) > 1e-9 else float("nan") print(f"\n -> DIAGNOSI, e sono DUE effetti, non uno (la mia prima stesura ne dichiarava uno") print(" solo e sarebbe stata una semplificazione comoda):") print(f" (1) AMPIEZZA: gia' il sottoinsieme MEDIANO da {k} gambe scende {full:.3f} -> " f"{med:.3f}.") print(f" Vale il {100*q_ampiezza:.0f}% della caduta totale, e nessun capitale la ripara:") print(f" con {k} gambe il long-5/short-5 tiene 10 posizioni su {k}.") print(f" (2) SFORTUNA DI LISTINO: dentro quella distribuzione il sottoinsieme che Deribit") print(f" quota davvero sta al {pct:.0f}° percentile ({mine:.3f} contro mediana {med:.3f}).") print(f" Le 8 gambe mancanti non sono un campione casuale: sono gli alt piu' recenti") print(" e piu' dispersi, cioe' quelli da cui un cross-sectional prende il segnale.") print(" Conseguenza operativa identica in entrambi i casi: aspettare che Deribit quoti") print(f" 2-3 alt in piu' non basta — il sottoinsieme mediano a {k} gambe e' gia' meta'") print(" del 19-gambe, e la ricostruzione dell'universo pieno passa dal SECONDO VENUE.") print(" ⚠️ Questo test gira sui prezzi HYPERLIQUID anche per il sottoinsieme 'Deribit': e'") print(" il segnale, non il fill. La divergenza fra venue su un ranking cross-sezionale e'") print(" gia' stata misurata trascurabile (22/08: corr 0,9991 fra i due venue).") # ============================================================================================= # 5) LA MISURA — delta APPAIATI sulla banda d'ancora congiunta # ============================================================================================= PASSED_TXT = "?" CANDIDATI = ["XS01", "XS01-D", "XS01-D14", "STATARB", "DVOLSPREAD", "PREVDAY", "VRP01"] ANCHOR_OF = {"XS01": XS_PHASES, "XS01-D": XS_PHASES, "XS01-D14": XS_PHASES, "STATARB": TP_HOURS, "DVOLSPREAD": (0,), "PREVDAY": (0,), "VRP01": VRP_PHASES} def _stats(s: pd.Series) -> tuple[float, float, float, float]: m = metrics(s) mh = metrics(s[s.index >= HOLDOUT]) return m["sharpe"], mh["sharpe"], m["cagr"], m["maxdd"] def sezione5(S: dict, draws: int) -> dict: sez(f"5) LA MISURA — dSharpe / d drift / d maxDD del LIBRO LIVE con un terzo sleeve") print(f" Spazio d'ancora congiunto: TP01 x{len(TP_HOURS)} · SKH01 x{len(SKH_OFFSETS)} · " f"candidato x(1..{len(XS_PHASES)}).") print(f" {draws} estrazioni UNIFORMI INDIPENDENTI (seed {SEED}); la statistica e' la") print(" MEDIANA DELLE DIFFERENZE APPAIATE (mai la differenza delle mediane: cadono su") print(" ancore diverse — errore ribaltante catturato il 26/07).") rng = np.random.default_rng(SEED) hs = rng.choice(TP_HOURS, size=draws) os_ = rng.choice(SKH_OFFSETS, size=draws) ph = {c: rng.choice(ANCHOR_OF[c], size=draws) for c in CANDIDATI} out: dict = {c: {w: {"sh": [], "shh": [], "cagr": [], "dd": []} for w in WEIGHTS_TESTED} for c in CANDIDATI} base_stats = [] t = time.time() for i in range(draws): cols2 = {"TP01": S["TP01"][hs[i]], "SKH01": S["SKH01"][os_[i]]} b = combine_outer(cols2, {"TP01": W_TP, "SKH01": W_SKH}, lo=BOOK_START) bs = _stats(b) base_stats.append(bs) for c in CANDIDATI: cols3 = dict(cols2); cols3["CAND"] = S[c][ph[c][i]] for w in WEIGHTS_TESTED: a = combine_outer(cols3, {"TP01": W_TP * (1 - w), "SKH01": W_SKH * (1 - w), "CAND": w}, lo=BOOK_START) st = _stats(a) out[c][w]["sh"].append(st[0] - bs[0]) out[c][w]["shh"].append(st[1] - bs[1]) out[c][w]["cagr"].append(st[2] - bs[2]) out[c][w]["dd"].append(st[3] - bs[3]) if i == 4: print(f" {clock()} ritmo: {(time.time()-t)/5:.2f}s/estrazione -> stimati " f"{(time.time()-t)/5*draws:.0f}s") B = np.array(base_stats) print(f"\n LIBRO LIVE DE-LUCKATO (2 sleeve, funding dentro), mediana delle {draws} estrazioni:") print(f" ShFULL {np.median(B[:,0]):.3f} [p10 {np.percentile(B[:,0],10):.3f}, " f"p90 {np.percentile(B[:,0],90):.3f}] ShHOLD {np.median(B[:,1]):.3f} " f"CAGR {100*np.median(B[:,2]):.2f}% maxDD {100*np.median(B[:,3]):.2f}%") can = _stats(combine_outer({"TP01": S["TP01"][0], "SKH01": S["SKH01"][0]}, {"TP01": W_TP, "SKH01": W_SKH}, lo=BOOK_START)) pc = float((B[:, 0] < can[0]).mean() * 100.0) print(f" ancora CANONICA (h=0, off=0): ShFULL {can[0]:.3f} = {pc:.1f}° pctl della banda " f"-> fortuna d'ancora {can[0]-np.median(B[:,0]):+.3f}") print(" ⚠️ `d maxDD` esce ~0,00 per XS01*: il maxDD FULL del libro e' fissato dal 2022, e") print(" quei candidati cominciano nel 2024 -> non possono toccarlo. Non e' 'non protegge',") print(" e' 'non c'era'. Per loro la colonna che informa e' d maxDD sull'hold-out, non FULL.") for c in CANDIDATI: sub(f"5.{CANDIDATI.index(c)+1}) {c}") print(f" {'peso':>6}{'dShFULL':>12}{'>0':>7}{'[p10,p90]':>20}{'dShHOLD':>11}{'>0':>7}" f"{'d CAGR pp':>12}{'d maxDD pp':>13}{'>0(peggio)':>12}") for w in WEIGHTS_TESTED: r = out[c][w] a = np.array(r["sh"]); ah = np.array(r["shh"]) ac = np.array(r["cagr"]); ad = np.array(r["dd"]) print(f" {w:>5.0%}{np.median(a):>+12.4f}{100*(a>0).mean():>6.0f}%" f" [{np.percentile(a,10):+.3f},{np.percentile(a,90):+.3f}]" f"{np.median(ah):>+11.4f}{100*(ah>0).mean():>6.0f}%" f"{100*np.median(ac):>+12.2f}{100*np.median(ad):>+13.2f}" f"{100*(ad>0).mean():>11.0f}%") return dict(out=out, base=B) def sezione5b(S: dict) -> None: sez("5b) A COSA SERVE UN dSharpe — il guadagno e' RITORNO o solo RISCHIO IN MENO?") print(" In sezione 5 quasi tutti i candidati alzano lo Sharpe e ABBASSANO il drift: a") print(" nozionale invariato il libro guadagna MENO. E' la regola gia' scritta il 25/07") print(" (\"un diversificatore a basso CAGR si giudica a ISO-RISCHIO, mai a iso-nozionale\").") print(" ⚠️ Il null del de-levering qui sarebbe DEGENERE (regola 22/08 §8d: sh(k·base) ≡") print(" sh(base)) -> non e' un test, e' un cambio di unita'. La domanda giusta e' l'altra:") print(" quanto vale quel dSharpe SE lo si riconverte in ritorno, e la leva ESISTE?") print(" Qui: si leva il libro a 2 sleeve del fattore k che ne eguaglia la vol al libro a 3,") print(" e si confrontano i DRIFT. k<1 = il terzo sleeve compra rischio in meno.") print(" ⚠️ E la leva NON esiste come knob: `config/live.json` non ha una chiave di scala") print(" (§38, verificato: servirebbe toccare W_TP01/W_SKH in src/live/book.py = CODICE") print(" su un percorso con soldi veri). Finche' non esiste, la colonna 'd drift' di") print(" sezione 5 e' quella che si incassa, e questa e' quella che si potrebbe incassare.") base = combine_outer({"TP01": S["TP01"][0], "SKH01": S["SKH01"][0]}, {"TP01": W_TP, "SKH01": W_SKH}, lo=BOOK_START) rb = np.asarray(base.values, float) v2 = float(rb.std() * np.sqrt(365.25)) d2 = float(np.mean(np.log1p(rb)) * 365.25) print(f"\n libro a 2 sleeve (ancora canonica): vol {100*v2:.2f}% drift log {100*d2:.2f}%/anno") print(f"\n {'candidato':<12}{'peso':>6}{'vol 3sl':>10}{'k iso-vol':>11}" f"{'drift 2sl levato':>18}{'drift 3sl':>12}{'d drift pp':>12}{'verdetto':>14}") for c in CANDIDATI: for w in (0.15,): cols = {"TP01": S["TP01"][0], "SKH01": S["SKH01"][0], "CAND": S[c][sorted(S[c])[0]]} a = combine_outer(cols, {"TP01": W_TP * (1 - w), "SKH01": W_SKH * (1 - w), "CAND": w}, lo=BOOK_START) # confronto sulla FINESTRA COMUNE: uno sleeve che parte nel 2024 non si giudica # contro un libro che comincia nel 2019 (sarebbero due campioni diversi). j = pd.concat({"a": a, "b": base}, axis=1, join="inner").dropna() ra = np.asarray(j["a"].values, float); rbb = np.asarray(j["b"].values, float) v3 = float(ra.std() * np.sqrt(365.25)); vb = float(rbb.std() * np.sqrt(365.25)) k = v3 / vb if vb > 0 else float("nan") dr3 = float(np.mean(np.log1p(ra)) * 365.25) dr2k = float(np.mean(np.log1p(np.clip(k * rbb, -0.99, None))) * 365.25) dd = 100 * (dr3 - dr2k) verdict = "AGGIUNGE" if dd > 0.05 else ("PEGGIORA" if dd < -0.05 else "neutro") print(f" {c:<12}{w:>5.0%}{100*v3:>9.2f}%{k:>11.3f}{100*dr2k:>17.2f}%" f"{100*dr3:>11.2f}%{dd:>+12.2f}{verdict:>14}") print("\n (finestra COMUNE ai due libri: per XS01* e VRP01 e' la loro storia, non il 2019+.)") # ============================================================================================= # 6) weights_tilt_null # ============================================================================================= def sezione6(S: dict) -> dict: sez("6) `weights_tilt_null` — IL gate del progetto per ogni cambio di pesi") print(" Ogni proposta di cambio pesi si giudica contro il null dei tilt casuali") print(" cap-respecting: gate_pass solo se delta_insample >= 0 E percentile < firma") print(f" best-of-k. k_seen = {len(CANDIDATI)}x{len(WEIGHTS_TESTED)} = " f"{len(CANDIDATI)*len(WEIGHTS_TESTED)} configurazioni guardate sull'hold-out qui.") k_seen = len(CANDIDATI) * len(WEIGHTS_TESTED) res = {} print(f"\n {'candidato':<14}{'peso':>6}{'d_insample':>13}{'d_hold':>10}{'d_full':>10}" f"{'pctl_hold':>11}{'best-of-k':>11}{'rnd>cur':>9}{'GATE':>9}") for c in CANDIDATI: cols = {"TP01": S["TP01"][0], "SKH01": S["SKH01"][0], c: S[c][sorted(S[c])[0]]} w_cur = {"TP01": W_TP, "SKH01": W_SKH, c: 0.0} for w in WEIGHTS_TESTED: w_prop = {"TP01": W_TP * (1 - w), "SKH01": W_SKH * (1 - w), c: w} try: r = weights_tilt_null(cols, w_cur, w_prop, floor=0.03, n=300, k_seen=k_seen) except Exception as e: # noqa: BLE001 print(f" {c:<14}{w:>5.0%} non calcolabile: {type(e).__name__}") continue res[(c, w)] = r print(f" {c:<14}{w:>5.0%}{r['delta_insample']:>+13.4f}{r['delta_hold']:>+10.4f}" f"{r['delta_full']:>+10.4f}{r['pctl_hold']:>10.1f}°{r['bestofk_pctl']:>10.1f}°" f"{r['frac_random_beat_hold']:>9.2f}" f"{('PASS' if r['gate_pass'] else 'FAIL'):>9}") npass = sum(1 for v in res.values() if v["gate_pass"]) print(f"\n PASS {npass}/{len(res)}. ⚠️ VA LETTO BENE, e la colonna che conta e' `rnd>cur`:") print(" e' la frazione di tilt CASUALI a 3 sleeve che battono il libro attuale sull'hold-out.") print(" Dove vale 0,9 il claim \"questo candidato migliora l'hold-out\" e' GENERICO: lo fa") print(" quasi qualunque cosa gli si metta accanto, e il gate passa proprio perche' la") print(" proposta NON e' eccezionale fra i tilt casuali (pctl basso = non e' un best-of-k).") print(" `weights_tilt_null` e' NECESSARIO, non sufficiente — e' scritto nel suo docstring.") print(" ✅ Controllo positivo: il gate HA potenza — fallisce su XS01-D a 10/15/20% (dove il") print(" delta in-sample e' negativo). Un gate che passa tutto sarebbe indistinguibile") print(" da uno rotto.") return res # ============================================================================================= # 7) ESEGUIBILITA' PER CAPITALE # ============================================================================================= def _gated_sharpe(pos: np.ndarray, ret: np.ndarray, idx, alloc: float, n_legs: int, fee_side: float = FEE_SIDE) -> tuple[float, int, float]: """Sharpe con min-order per GAMBA: un Δ di nozionale sotto $5 non si esegue.""" per_leg = alloc / max(n_legs, 1) held = np.empty(len(pos)); cur = 0.0; n_tr = 0; n_skip = 0 for i in range(len(pos)): if abs(pos[i] - cur) * per_leg >= MIN_ORDER: cur = pos[i]; n_tr += 1 elif abs(pos[i] - cur) > 1e-12: n_skip += 1 held[i] = cur h = np.zeros(len(held)); h[1:] = held[:-1] net = h * ret - fee_side * n_legs * np.abs(np.diff(h, prepend=0.0)); net[0] = 0.0 s = pd.Series(net, index=idx) return metrics(to_daily(s))["sharpe"], n_tr, (n_skip / max(n_tr + n_skip, 1)) def sezione7(S: dict, V: dict, cens: dict) -> None: sez("7) ESEGUIBILITA' PER CAPITALE — ticket per gamba, min-order, haircut") print(f" min_order_usd = ${MIN_ORDER:.0f} (config/live.json, letto). E' la soglia NOSTRA:") print(" il lotto del VENUE e' 1-2 ordini di grandezza sotto (sezione 1).") sub("7a) TICKET MEDIO PER GAMBA al peso 15% del libro") rows = [] j = _joint_off(0) idx_sa = pd.DatetimeIndex(pd.to_datetime(j["datetime"], utc=True)) sa_pos = S["_statarb_pos"] sa_dtk = np.abs(np.diff(sa_pos, prepend=0.0)) btc, eth = ol.aligned() dvs_tk = np.abs(np.diff(S["_dvs_pos"], prepend=0.0)) xs19_tk = S["_xs19_tick"]; xsd_tk = S["_xsd_tick"] pv_tk = S["_prevday_turn"] spec = [ ("STATARB", sa_dtk[sa_dtk > 1e-9], 2, "BTC+ETH perp (NETTA col libro)"), ("DVOLSPREAD", dvs_tk[dvs_tk > 1e-9], 2, "BTC+ETH perp (NETTA col libro)"), ("PREVDAY", pv_tk[pv_tk > 1e-9], 2, "BTC+ETH perp (NETTA col libro)"), ("XS01", xs19_tk[xs19_tk > 1e-9], 2 * SL.XS_CFG["k"], "19 gambe HL (venue diverso)"), (f"XS01-D", xsd_tk[xsd_tk > 1e-9], 2 * min(SL.XS_CFG["k"], len(cens["xs_stable"]) // 2), f"{len(cens['xs_stable'])} gambe Deribit (>=1a)"), ] print(f" {'candidato':<12}{'gambe':>7}{'Δpeso/gamba med':>18}" + "".join(f"{('$'+str(int(c))):>12}" for c in CAP_LADDER) + " nota") for nm, tk, legs, nota in spec: med = float(np.median(tk)) if len(tk) else 0.0 cells = "" for C in CAP_LADDER: tick = med * C * 0.15 # Δpeso di UNA gamba x nozionale allocato cells += f"{('$%.2f' % tick):>12}" print(f" {nm:<12}{legs:>7}{med:>18.4f}{cells} {nota}") print(f" (ticket = Δpeso mediano di UNA gamba x (capitale x 15%). Sotto ${MIN_ORDER:.0f} " "l'ordine NON parte.)") sub("7b) HAIRCUT DA MIN-ORDER — Sharpe modellato vs Sharpe eseguibile") print(f" {'candidato':<12}" + "".join(f"{('$'+str(int(c))):>26}" for c in CAP_LADDER)) print(f" {'':<12}" + "".join(f"{'Sh (haircut, %salt)':>26}" for _ in CAP_LADDER)) sa_ret = OS.spread_ret(j) # il book DVOLSPREAD e' (wb su BTC, -wb su ETH) -> il ritorno per unita' di wb e' rb - re. dvs_ret = (al.simple_returns(btc["close"].values.astype(float)) - al.simple_returns(eth["close"].values.astype(float))) for nm, pos, ret, idx, legs in ( ("STATARB", sa_pos, sa_ret, idx_sa, 2), ("DVOLSPREAD", S["_dvs_pos"], dvs_ret, pd.DatetimeIndex(btc["dt"]), 2)): line = f" {nm:<12}" base_sh = _gated_sharpe(pos, ret, idx, 1e12, legs)[0] for C in CAP_LADDER: sh, ntr, skip = _gated_sharpe(pos, ret, idx, C * 0.15, legs) line += f"{('%.3f (%+.3f, %.0f%%)' % (sh, sh-base_sh, 100*skip)):>26}" print(line) print(" (Sh a capitale infinito = modellato. Il segno del haircut puo' essere POSITIVO: il") print(" pavimento salta ribilanci piccoli e risparmia fee — e' de-levering, non merito.)") sub("7c) VRP01 — l'aritmetica dei LOTTI (non e' min-order, e' granularita' intera)") e_mk = (V["perp"].get("ETH") or {}).get("mark") or 0.0 b_mk = (V["perp"].get("BTC") or {}).get("mark") or 0.0 lot_e = min(V["opt"].get(("USDC", "ETH"), {0.1})) * e_mk lot_b = min(V["opt"].get(("USDC", "BTC"), {0.01})) * b_mk print(f" Il collaterale dello sleeve e' il PUT VENDUTO (cash-secured sullo strike): un lotto") print(f" vale ~1 unita' di sottostante x min_trade_amount.") print(f" ETH_USDC opzioni : lotto minimo = ${lot_e:,.0f} di nozionale") print(f" BTC_USDC opzioni : lotto minimo = ${lot_b:,.0f} di nozionale") print(f"\n {'capitale':>10}{'peso 12%':>11}{'lotti ETH':>11}{'lotti BTC':>11}" f"{'% del conto per 1 lotto ETH':>30}") for C in CAP_LADDER: a = C * 0.12 print(f" {('$%.0f' % C):>10}{('$%.0f' % a):>11}{int(a // lot_e) if lot_e else 0:>11}" f"{int(a // lot_b) if lot_b else 0:>11}{(100*lot_e/C):>29.1f}%") print(f"\n -> 1 lotto ETH = peso di book del 12% a capitale ~${lot_e/0.12:,.0f}; il BTC entra") print(f" a ~${lot_b/0.12:,.0f}. Il 30/07 questi numeri erano ${1832/0.12:,.0f} e " f"${6210/0.12:,.0f} perche' misurati sulla famiglia INVERSE.") print(" ⚠️ NON promuove VRP01: resta fuori per la REGOLA 'niente short-vol da modello in") print(" deploy' e per f=0,73 misurato sulle quote vere. Cade il muro del LOTTO, non la regola.") def sezione7d(rate: dict) -> None: sub("7d) IL CONTO GIUSTO PER UN CANDIDATO CHE NETTA — min-order sul TARGET NETTO") print(" 7a/7b trattano il candidato come un libro a se': per uno sleeve che NETTA e' il") print(" conto SBAGLIATO. Il cron manda UN ordine per asset sul target NETTO, e il") print(" pavimento di $5 morde li'. Qui il target netto e' ricostruito come in produzione:") print(" net_a = 0.5 · E · [0.75(1-w)·tp_frac_a + 0.25(1-w)·skh_sign_a + w·cand_pos_a]") tp = TrendPortfolio(**CANONICAL) base_pos, cand_pos = {}, {"STATARB": {}, "DVOLSPREAD": {}} for a in ("BTC", "ETH"): df = A_TP.daily_off(a, 0) idx = pd.DatetimeIndex(pd.to_datetime(df["datetime"], utc=True)) frac = np.nan_to_num(np.asarray(tp.target_series(df), float)) _, tr = FU.skh_daily(a, 0, "canonical", rate) sign = np.zeros(len(idx)) tms = FU.to_ms(idx) for t in tr: # segno SKH01 sui giorni in cui e' aperto m = (tms >= t["t_in"]) & (tms < t["t_out"]) sign[m] = t["dir"] base_pos[a] = pd.DataFrame({"frac": frac, "sign": sign}, index=idx) j = _joint_off(0) ix = pd.DatetimeIndex(pd.to_datetime(j["datetime"], utc=True)) sp = np.nan_to_num(np.asarray(OS.f_statarb_resid(W=45, sgn=+1)(j), float)) cand_pos["STATARB"]["ETH"] = pd.Series(sp, index=ix) cand_pos["STATARB"]["BTC"] = pd.Series(-sp, index=ix) btc, eth = ol.aligned() wb, _we = dvs_make_book(**DVS_FROZEN)(btc, eth) wb = np.clip(np.nan_to_num(np.asarray(wb, float)), -1, 1) ixd = pd.DatetimeIndex(btc["dt"]) cand_pos["DVOLSPREAD"]["BTC"] = pd.Series(wb, index=ixd) cand_pos["DVOLSPREAD"]["ETH"] = pd.Series(-wb, index=ixd) def net_orders(cand: str | None, w: float, E: float) -> tuple[int, float, float]: n_ord, tot = 0, 0.0 yrs = max(len(base_pos[a]) for a in ("BTC", "ETH")) / 365.25 for a in ("BTC", "ETH"): b = base_pos[a] raw = 0.5 * E * ((W_TP * (1 - w)) * np.maximum(b["frac"].values, 0.0) + (W_SKH * (1 - w)) * b["sign"].values) ser = pd.Series(raw, index=b.index) if cand: cp = cand_pos[cand][a].reindex(ser.index).fillna(0.0) ser = ser + 0.5 * E * w * cp.values cur = 0.0 for v in ser.values: if abs(v - cur) >= MIN_ORDER: n_ord += 1; tot += abs(v - cur); cur = v return n_ord, tot / max(n_ord, 1), yrs print(f"\n {'configurazione':<26}" + "".join(f"{('$'+str(int(c))):>30}" for c in CAP_LADDER)) print(f" {'':<26}" + "".join(f"{'ord/a ticket turnover/a':>30}" for _ in CAP_LADDER)) for cand, w in ((None, 0.0), ("STATARB", 0.15), ("DVOLSPREAD", 0.15)): lab = "2 sleeve (com'e' oggi)" if cand is None else f"+ {cand} @15%" line = f" {lab:<26}" for C in CAP_LADDER: n, tk, yrs = net_orders(cand, w, C) line += f"{('%.0f $%.0f $%.0f' % (n / yrs, tk, n * tk / yrs)):>30}" print(line) print("\n 📌 Un terzo sleeve che netta NON paga un min-order proprio: cambia il NUMERO di") print(" ordini del libro, non il loro pavimento. Il conteggio ordini RADDOPPIA, il") print(" TURNOVER (cioe' la fee) sale molto meno — e' il segno che il pavimento sta") print(" tagliando ribilanci piccoli, non che il libro trada di piu' in nozionale.") print(" ✅ E c'e' un vantaggio strutturale mai scritto: le fee modellate negli sleeve sono") print(" un LIMITE SUPERIORE. Ogni sleeve paga sul PROPRIO |Δpos|, ma il conto paga sul") print(" |Δ(somma)|, e per disuguaglianza triangolare |Δ(Σ w·p)| <= Σ w·|Δp|.") print(" ⚠️ Il rovescio, misurato in sezione 2: il candidato non puo' avere un proprio stop,") print(" e i suoi ribilanci si CANCELLANO contro quelli di TP01/SKH01 quando discordano.") # ============================================================================================= # 8) LA SCALA # ============================================================================================= def sezione8(S: dict, M: dict, G: dict, V: dict, cens: dict) -> None: sez("8) LA SCALA — a quale capitale entra il primo, chi e', e per quale muro") global PASSED_TXT byc: dict = {} for (c, w), r in G.items(): byc.setdefault(c, []).append((w, r["gate_pass"])) PASSED_TXT = " · ".join(f"{c}: {sum(1 for _, g in v if g)}/{len(v)}" for c, v in byc.items()) lot_e = min(V["opt"].get(("USDC", "ETH"), {0.1})) * ((V["perp"].get("ETH") or {}).get("mark") or 0) lot_b = min(V["opt"].get(("USDC", "BTC"), {0.01})) * ((V["perp"].get("BTC") or {}).get("mark") or 0) W = 0.15 print(f" `weights_tilt_null` superato (pesi su 4): {PASSED_TXT}\n") sub("8a) IL MURO DI CIASCUNO — quale vincolo morde per primo, e se e' una soglia di CAPITALE") muri = [ ("XS01 (19 gambe)", "VENUE", "5 gambe su 19 non esistono su Deribit; su Hyperliquid serve un SECONDO CONTO,", "e la decisione d'operatore del 26/07 tiene tutto su Deribit fino a $20.000.", "NON e' una soglia di eseguibilita' (falsificata il 22/08): e' la decisione di venue."), (f"XS01-D ({len(cens['xs_stable'])} gambe)", "AMPIEZZA + SFORTUNA DI LISTINO", f"con {len(cens['xs_stable'])} gambe il long-5/short-5 tiene 10 posizioni su " f"{len(cens['xs_stable'])}: gia' il sottoinsieme MEDIANO", "da 11 crolla (sez. 4c), e quello che Deribit quota sta al 9° pctl di quella banda.", f"dShFULL {np.median(M['out']['XS01-D'][W]['sh']):+.3f} a peso 15% " f"(negativo a {100-100*(np.array(M['out']['XS01-D'][W]['sh'])>0).mean():.0f}% delle ancore).", "Nessun capitale lo ripara, e nemmeno 2-3 listing in piu': serve l'universo pieno."), ("XS01-D14", "DATO", "3 delle 14 gambe sono quotate da 2-8 GIORNI e una (APT) non ha nemmeno una quota;", "e `marginal_vs_tp01` la classifica HEDGE, non ADDS.", "Riapribile fra ~1 anno di quotazione, se il volume arriva. Non e' una soglia di $."), ("STATARB-RESID", "GATE PRE-REGISTRATO 2026-09-27", f"eseguibile e netta gia' a $635; dShFULL {np.median(M['out']['STATARB'][W]['sh']):+.3f} " f"a 15%, ma d drift {100*np.median(M['out']['STATARB'][W]['cagr']):+.2f}pp", "(tutto rischio in meno, zero ritorno in piu' a nozionale invariato) e", "`marginal_vs_tp01` = NEUTRAL, robust_oos FALSE. Il gate scade fra 35 giorni."), ("DVOLSPREAD", "GATE PRE-REGISTRATO 2026-10-24 / 2027-01-24", f"eseguibile e netta a $635; dShFULL {np.median(M['out']['DVOLSPREAD'][W]['sh']):+.3f}, ma", f"d drift {100*np.median(M['out']['DVOLSPREAD'][W]['cagr']):+.2f}pp e d maxDD che PEGGIORA " "sopra il 15%.", "robust_oos FALSE. Il gate ha una data e non si anticipa."), ("PREVDAY", "NESSUN GATE — ed e' il problema", f"e' il piu' forte misurato (dShFULL {np.median(M['out']['PREVDAY'][W]['sh']):+.3f}, " f"dShHOLD {np.median(M['out']['PREVDAY'][W]['shh']):+.3f}, d drift " f"{100*np.median(M['out']['PREVDAY'][W]['cagr']):+.2f}pp),", "eseguibile, netta, ADDS, robust_oos TRUE — e sta in forward-monitor dal giugno", "senza gate pre-registrato, senza deflated-Sharpe, e con corr al libro 0.150 (la piu' alta)."), ("VRP01", "REGOLA + PREZZO", f"il lotto NON e' piu' il muro (ETH ${lot_e:,.0f} = peso 12% da ~${lot_e/0.12:,.0f}),", "ma restano 'niente short-vol da modello in deploy' e f=0,73 misurato sulle quote vere;", "e non netta: serve una riga in `_CONTRACT` = codice su un percorso con soldi veri."), ("GTAA01", "PRIIPs", "l'ordine e' RIFIUTATO dal broker retail UE. La via UCITS esiste (27/07) ma e' un", "SECONDO VENUE, quindi cade anch'essa sotto la decisione dei $20.000.", "Non e' una soglia di capitale: e' un divieto."), ] for nm, muro, *righe in muri: print(f"\n {nm} -> MURO: {muro}") for r in righe: print(f" {r}") sub("8b) LA SCALA IN DOLLARI — cosa cambia, e quando") righe = [ (635, "OGGI", "STATARB · DVOLSPREAD · PREVDAY eseguibili e nettanti (ticket $2,8-4,9,", "assorbiti nel netto: 7d). XS01-D eseguibile ma PEGGIORA. Nessuno ammissibile."), (2024, f"1 lotto ETH di VRP01 = peso 12% (${lot_e:,.0f}/0,12)", "cade il muro del lotto per la gamba ETH. NON cade la regola short-vol.", ""), (3000, "GTAA_MIN_CAPITAL", "GTAA01 avrebbe il capitale ma resta PRIIPs-bloccato (verificato sul conto reale).", ""), (6440, f"1 lotto BTC di VRP01 = peso 12% (${lot_b:,.0f}/0,12)", "VRP01 tornerebbe 50/50 BTC+ETH come lo sleeve modellato.", ""), (20000, "DECISIONE DI VENUE (operatore, 26/07)", "si riapre lo split. E' QUI che XS01 (Hyperliquid, 19 gambe) diventa possibile —", "per la decisione di venue, NON per una soglia di eseguibilita'."), ] for C, evento, *righe2 in righe: print(f"\n ${C:,}".replace(",", ".") + f" — {evento}") for r in righe2: if r: print(f" {r}") print("\n 📌 LA RISPOSTA ALLA DOMANDA 'a quale capitale cambia': NON c'e' una soglia di") print(" capitale che ammette un terzo sleeve. Le due date che contano sono un GATE") print(" (27/09, fra 35 giorni, gratis) e una DECISIONE DI VENUE ($20k). Il capitale") print(" entra solo per VRP01, che e' fermo per una regola che il capitale non tocca.") def sezione9(S: dict, M: dict, G: dict, cens: dict) -> None: sez("9) VERDETTO — l'attesa a priori contro l'esito") W = 0.15 print(f" Tabella decidibile, peso {W:.0%}, mediana delle differenze APPAIATE sulla banda") print(" d'ancora congiunta, funding dentro:\n") print(f" {'candidato':<12}{'dShFULL':>10}{'>0':>6}{'dShHOLD':>10}{'d drift':>10} " f"{'eseguibile a $635':<24}{'gate':<12}{'tilt-null'}") for c in CANDIDATI: r = M["out"][c][W] ex = {"XS01": "NO (5 gambe assenti)", "XS01-D": "SI (11 gambe)", "XS01-D14": "solo 3gg di listino", "STATARB": "SI (netta)", "DVOLSPREAD": "SI (netta)", "PREVDAY": "SI (netta)", "VRP01": f"NO (1 lotto=38% conto)"}[c] gt = {"XS01": "—", "XS01-D": "—", "XS01-D14": "—", "STATARB": "27/09", "DVOLSPREAD": "24/10", "PREVDAY": "—", "VRP01": "regola"}[c] gp = "PASS" if G.get((c, W), {}).get("gate_pass") else "FAIL" print(f" {c:<12}{np.median(r['sh']):>+10.4f}{100*(np.array(r['sh'])>0).mean():>5.0f}%" f"{np.median(r['shh']):>+10.4f}{100*np.median(r['cagr']):>+8.2f}pp " f"{ex:<24}{gt:<12}{gp}") dsa = float(np.median(M["out"]["STATARB"][W]["sh"])) ddv = float(np.median(M["out"]["DVOLSPREAD"][W]["sh"])) dxs = float(np.median(M["out"]["XS01"][W]["sh"])) dxd = float(np.median(M["out"]["XS01-D"][W]["sh"])) xs_gate = G.get(("XS01", W), {}).get("gate_pass") print("\n ATTESA A PRIORI vs ESITO (dichiarata in sezione 0, prima di misurare):") print(" (A1) 'nessuno deployabile oggi, e il muro NON e' il lotto del venue'") print(" -> ✅ CONFERMATA. Nessun candidato muore sul lotto del venue (i lotti stanno") print(" fra $0,01 e $7,72). Il muro e' altrove, ed e' diverso per ciascuno.") print(f" (A2) 'XS01 il piu' forte ma tilt-null FALLIRA'' -> ⚠️ META' GIUSTA: XS01 e' fra i") print(f" piu' forti (dShFULL {dxs:+.3f}) ma `weights_tilt_null` " f"{'PASSA' if xs_gate else 'FALLISCE'}. Il gate non e'") print(" il muro di XS01: il muro e' che 5 delle 19 gambe Deribit non le quota.") print(f" (A3) '|dSharpe| < 0,10 per STATARB e DVOLSPREAD' -> ❌ REFUTATA su STATARB") print(f" ({dsa:+.3f} a peso 15%, positivo a 100% delle ancore), confermata su") print(f" DVOLSPREAD ({ddv:+.3f}). E il segno NON e' incerto: e' stabile sulla banda.") print(" (A4) 'il primo a entrare sara' XS01 fra $2k e $5k, per ragione di venue'") print(f" -> ❌ REFUTATA nella sostanza: la versione di XS01 eseguibile su Deribit oggi") print(f" ({len(cens['xs_stable'])} gambe) fa dShFULL {dxd:+.3f} — PEGGIORA il libro. Non e' una") print(" soglia di capitale: e' che l'universo eseguibile non regge il meccanismo") print(" (sez. 4c: la maggior parte e' AMPIEZZA, il resto e' QUALI gambe mancano).") print(" (A5) 'mi sorprendera' un candidato dato per eseguibile che non lo e''") print(" -> ✅ CONFERMATA, ma al ROVESCIO di come me l'aspettavo: la sorpresa e' un") print(" candidato dato per NON eseguibile che lo E' (VRP01: il lotto vero e' 7,5x") print(" piu' piccolo del pubblicato), e uno dato per eseguibile-a-$20k che a") print(" $20k resta comunque rotto (XS01: e\' il VENUE, non la taglia).") print("\n ⚠️ CAVEAT CHE VALE PER TUTTA LA TABELLA — questo e' uno SCREEN, non una scoperta:") print(" 7 candidati x 4 pesi = 28 configurazioni guardate sullo stesso libro e sullo") print(" stesso hold-out. Nessuno di questi delta e' passato per un deflated-Sharpe DI") print(" SCREEN (lezione 22/08 §10: non basta contare i trial, serve la VARIANZA di") print(" screen), e il candidato che esce primo esce da una selezione che io ho fatto.") print(" E' esattamente per questo che l'output e' una TABELLA e non una raccomandazione:") print(" il modo giusto di usarla e' scegliere UN candidato per una ragione dichiarata") print(" PRIMA, e poi fargli un gate pre-registrato — non prendere il massimo di 28.") # ============================================================================================= def main() -> None: global _ARGS ap = argparse.ArgumentParser() ap.add_argument("--no-net", action="store_true") ap.add_argument("--draws", type=int, default=1000) ap.add_argument("--force", action="store_true") _ARGS = ap.parse_args() sys.argv = [sys.argv[0]] + (["--no-net"] if _ARGS.no_net else []) sez("r0823_book_3rd.py — §50 BOOK-3RD | quale terzo sleeve per il LIBRO LIVE?") print(f" avvio {dt.datetime.now(dt.timezone.utc):%Y-%m-%d %H:%M UTC} · " f"capitale reale assunto ${CAPITAL_TODAY:.0f} · estrazioni {_ARGS.draws}") sez("0) ATTESA A PRIORI (dichiarata PRIMA di misurare)") print(ATTESA) V = venue_census() cens = sezione1(V) sezione1c(V, cens) sezione2() print(f"\n {clock()} carico i tassi di funding (cache/rete)...") rate = get_rate() R3 = sezione3(rate) print(f"\n {clock()} costruisco le serie ancorate...") S = build_series(rate, cens, force=_ARGS.force) sezione4(S, cens) sezione4b(S) sezione4c(cens) M = sezione5(S, _ARGS.draws) sezione5b(S) G = sezione6(S) sezione7(S, V, cens) sezione7d(rate) sezione8(S, M, G, V, cens) sezione9(S, M, G, cens) sez(f"FINE {clock()}") if __name__ == "__main__": main()