"""HL-EXEC (2026-08-22) — AUDIT DI FATTO: le regole VERE di Hyperliquid contro le soglie ASSUNTE. DOMANDA. Il progetto tiene due edge cross-sectional fuori dal libro **per taglia**: XS01 "serve ~$20.000" (origine: diario 2026-06-19-hyperliquid-xsec, "rumore arrotondamento" — una STIMA A OCCHIO, mai calcolata) XSR01 "diventa reale a ~$5.000" + gate pre-registrato 2026-10-23 con soglia "haircut di eseguibilita' a $5.000 <= 40%, altrimenti RITIRO" Entrambe sono calcolate col pavimento **min_order $5**, che e' il minimo di **DERIBIT**. XS01/XSR01 si eseguirebbero su **HYPERLIQUID**, che ha regole sue. Qui le regole si LEGGONO dal venue invece di assumerle, e si dice quali conclusioni del progetto cambiano. NON manda ordini, non tocca il conto: sole letture PUBBLICHE (nessuna chiave, nessuna firma). GRADO DELLE FONTI (dichiarato per ogni numero, come chiede il brief): [A] letto dall'API pubblica del venue api.hyperliquid.xyz/info (meta, metaAndAssetCtxs, userFees, l2Book) [B] derivato da [A] + verificato empiricamente sui book live [C] documentazione ufficiale (hyperliquid.gitbook.io) [D] assunto / ereditato dal progetto La verifica MAINNET non e' un atto di fede: i mark price dell'API si incrociano con l'ultima chiusura del feed CERTIFICATO su disco (data/raw/hl_*_1d.parquet). Il testnet non puo' superarlo. (Il progetto ha gia' pagato il prezzo di un feed testnet creduto vero: e' la causa del reset v2.0.0.) nice -n 19 timeout 900 uv run python scripts/research/r0822_hl_exec.py ... --no-net usa solo la cache su disco (nessuna chiamata di rete) ... --snaps N quanti snapshot del book L2 (default 3, ~45s di distanza) """ from __future__ import annotations import argparse import json import os import sys import time import urllib.request from pathlib import Path import numpy as np import pandas as pd PROJECT_ROOT = Path(__file__).resolve().parents[2] sys.path.insert(0, str(PROJECT_ROOT)) sys.path.insert(0, str(PROJECT_ROOT / "scripts" / "research")) RAW = PROJECT_ROOT / "data" / "raw" CACHE = Path(os.environ.get("HLEXEC_CACHE", "/tmp/claude-1001/-opt-docker-PythagorasGoal/" "b6cc75e7-14f8-4c32-bd07-ab8a0d2aaee6/scratchpad/hlexec")) INFO_URL = "https://api.hyperliquid.xyz/info" # MAINNET (il testnet e' api.hyperliquid-testnet.xyz) # --- parametri sotto esame ---------------------------------------------------------------------- MIN_ORDER_ASSUNTO = 5.0 # [D] pavimento DERIBIT usato da eval_weights_smallcap e da paper_xsr MIN_ORDER_HL = 10.0 # [C] "Order must have minimum value of $10." (docs/error-responses) FEE_LEG_MODELLO = 0.0005 # [D] 0.05%/gamba, config CONGELATA di XSR01 XSR_SOGLIA_HAIRCUT = 0.40 # [D] soglia pre-registrata del gate 2026-10-23 CAPITALI = (600.0, 1000.0, 1500.0, 2000.0, 3000.0, 4000.0, 5000.0, 7500.0, 10000.0, 20000.0, 50000.0) TICKETS = (50.0, 100.0, 300.0) ANN = np.sqrt(365.0) # ================================================================================================== # 0. RETE — letture pubbliche, con cache su disco (una corsa non deve ri-martellare il venue) # ================================================================================================== def _post(payload: dict, timeout: int = 30) -> object: req = urllib.request.Request(INFO_URL, data=json.dumps(payload).encode(), headers={"Content-Type": "application/json"}) with urllib.request.urlopen(req, timeout=timeout) as r: return json.loads(r.read().decode()) def cached(name: str, payload: dict, no_net: bool, ttl_s: float = 6 * 3600) -> object: CACHE.mkdir(parents=True, exist_ok=True) f = CACHE / f"{name}.json" if f.exists() and (no_net or (time.time() - f.stat().st_mtime) < ttl_s): return json.loads(f.read_text()) if no_net: raise FileNotFoundError(f"cache assente per {name} e --no-net attivo") d = _post(payload) f.write_text(json.dumps(d)) return d def universo_certificato() -> list[str]: """I 51 alt+BTC gia' CERTIFICATI su disco. Non si inventa un universo: si legge quello vero.""" return sorted(p.stem.replace("hl_", "").replace("_1d", "").upper() for p in RAW.glob("hl_*_1d.parquet")) # ================================================================================================== # 1. REGOLE DEL VENUE # ================================================================================================== def sig_figs_ok(px_str: str, sz_dec: int, max_dec: int = 6) -> bool: """[C] regola tick perp: <=5 cifre significative E <= (MAX_DECIMALS - szDecimals) decimali; i prezzi INTERI sono sempre ammessi. Qui serve per VERIFICARLA sui book veri -> [B].""" s = px_str.lstrip("-") dec = len(s.split(".")[1]) if "." in s else 0 if dec == 0: return True digits = s.replace(".", "").lstrip("0") return len(digits.rstrip("0")) <= 5 and dec <= (max_dec - sz_dec) def leggi_regole(no_net: bool) -> tuple[pd.DataFrame, dict]: meta_ctx = cached("metaAndAssetCtxs", {"type": "metaAndAssetCtxs"}, no_net) fees = cached("userFees", {"type": "userFees", "user": "0x0000000000000000000000000000000000000001"}, no_net) uni, ctxs = meta_ctx[0]["universe"], meta_ctx[1] rows = [] for a, c in zip(uni, ctxs): mark = float(c["markPx"]) if c.get("markPx") else np.nan rows.append(dict(sym=a["name"], szDecimals=int(a["szDecimals"]), maxLeverage=int(a["maxLeverage"]), delisted=bool(a.get("isDelisted", False)), markPx=mark, dayNtlVlm=float(c.get("dayNtlVlm") or np.nan), openInterest=float(c.get("openInterest") or np.nan))) df = pd.DataFrame(rows).set_index("sym") df["lot_units"] = 10.0 ** (-df["szDecimals"]) # [A] passo di size in unita' df["lot_usd"] = df["lot_units"] * df["markPx"] # [B] passo di size in dollari df["px_decimals_max"] = (6 - df["szDecimals"]).clip(lower=0) # [C] return df, fees["feeSchedule"] def verifica_mainnet(df: pd.DataFrame, syms: list[str]) -> pd.DataFrame: """Incrocia i mark dell'API con l'ULTIMA chiusura del feed certificato su disco. Un feed testnet non puo' superare questo controllo: i suoi prezzi sono fantasia.""" out = [] for s in syms: p = RAW / f"hl_{s.lower()}_1d.parquet" if not p.exists() or s not in df.index: continue d = pd.read_parquet(p, columns=["timestamp", "close"]) last_close = float(d["close"].iloc[-1]) last_ts = pd.Timestamp(int(d["timestamp"].iloc[-1]), unit="ms", tz="UTC") mk = float(df.loc[s, "markPx"]) out.append(dict(sym=s, disco=last_close, api=mk, dev_pct=100.0 * (mk - last_close) / last_close, eta_h=(pd.Timestamp.now("UTC") - last_ts) / pd.Timedelta("1h"))) return pd.DataFrame(out).set_index("sym") # ================================================================================================== # 2. BOOK L2 -> spread, profondita', costo di un ticket # ================================================================================================== def scarica_book(syms: list[str], snaps: int, no_net: bool, pause: float = 0.25, gap_s: float = 45.0) -> dict[str, list[dict]]: books: dict[str, list[dict]] = {s: [] for s in syms} for k in range(snaps): if k and not no_net: time.sleep(gap_s) for s in syms: try: b = cached(f"l2_{s}_{k}", {"type": "l2Book", "coin": s}, no_net) except Exception as e: # un book mancante e' un dato, non un crash print(f" [book] {s} snap{k}: {type(e).__name__}") continue books[s].append(b) if not no_net: time.sleep(pause) # pacing: il rate limit e' PER-IP e condiviso return books def costo_ticket(levels: list[dict], mid: float, notional: float, side: str) -> tuple[float, bool]: """VWAP di un ordine marketable da `notional` dollari contro il book, in bps DA MID. Ritorna (bps, book_esaurito). Per ticket piccoli converge al MEZZO SPREAD: e' giusto cosi', a $14 non si 'cammina' il book, si paga il livello top.""" resid, cost, filled = notional, 0.0, 0.0 for lv in levels: px, sz = float(lv["px"]), float(lv["sz"]) cap = px * sz take = min(resid, cap) cost += take filled += take / px resid -= take if resid <= 1e-9: break if filled <= 0: return float("nan"), True vwap = cost / filled bps = (vwap - mid) / mid * 1e4 * (1.0 if side == "buy" else -1.0) return bps, resid > 1e-6 def misura_liquidita(books: dict[str, list[dict]], meta: pd.DataFrame) -> pd.DataFrame: rows = [] for s, snaps in books.items(): if not snaps: continue rec: dict[str, list[float]] = {"spread": [], "d10": [], "viol": []} for t in TICKETS: rec[f"t{int(t)}"] = [] for b in snaps: bids, asks = b["levels"][0], b["levels"][1] if not bids or not asks: continue bb, ba = float(bids[0]["px"]), float(asks[0]["px"]) mid = 0.5 * (bb + ba) rec["spread"].append((ba - bb) / mid * 1e4) near = sum(float(l["px"]) * float(l["sz"]) for l in asks if (float(l["px"]) - mid) / mid <= 0.0010) near += sum(float(l["px"]) * float(l["sz"]) for l in bids if (mid - float(l["px"])) / mid <= 0.0010) rec["d10"].append(near) sd = int(meta.loc[s, "szDecimals"]) if s in meta.index else 0 rec["viol"].append(sum(0 if sig_figs_ok(l["px"], sd) else 1 for l in bids + asks)) for t in TICKETS: bb_, _ = costo_ticket(asks, mid, t, "buy") sb_, _ = costo_ticket(bids, mid, t, "sell") if np.isfinite(bb_) and np.isfinite(sb_): rec[f"t{int(t)}"].append(0.5 * (bb_ + sb_)) if not rec["spread"]: continue row = dict(sym=s, n_snap=len(rec["spread"]), spread_bps=float(np.median(rec["spread"])), half_spread_bps=float(np.median(rec["spread"])) / 2.0, depth10bps_usd=float(np.median(rec["d10"])), tick_viol=int(sum(rec["viol"]))) for t in TICKETS: v = rec[f"t{int(t)}"] row[f"slip{int(t)}_bps"] = float(np.median(v)) if v else np.nan rows.append(row) return pd.DataFrame(rows).set_index("sym").sort_values("spread_bps") def min_notional_osservato(books: dict[str, list[dict]]) -> dict: """Controllo EMPIRICO del pavimento $10: i livelli con n==1 sono UN ordine solo. Un ordine singolo sotto $10 non falsifica la regola (i fill parziali erodono un resto), ma il PAVIMENTO della distribuzione dice dove il venue taglia.""" vals = [] for snaps in books.values(): for b in snaps: for side in b["levels"]: for l in side: if int(l.get("n", 0)) == 1: vals.append(float(l["px"]) * float(l["sz"])) v = np.array(vals, float) if not len(v): return {} return dict(n=len(v), minimo=float(v.min()), p01=float(np.percentile(v, 1)), p05=float(np.percentile(v, 5)), mediana=float(np.median(v)), sotto10_pct=100.0 * float((v < 10.0).mean()), sotto5_pct=100.0 * float((v < 5.0).mean())) # ================================================================================================== # 3. SIMULATORE DI LIBRO A N GAMBE (la contabilita' e' quella di paper_xsr._step, generalizzata) # ================================================================================================== def sim_libro(Weff: np.ndarray, R: np.ndarray, r_hedge: np.ndarray | None, cap0: float, min_order: float | None, lot_usd: np.ndarray | None, px: np.ndarray | None, fee_leg: float | np.ndarray) -> dict: """Un passo per barra. Una gamba il cui |dw|*capitale sta sotto min_order NON si muove (stessa convenzione di altlib.eval_weights_smallcap e di scripts/live/paper_xsr._step). In piu': la size si arrotonda al LOTTO del venue -> se arrotonda a zero, non si esegue. `fee_leg` puo' essere uno scalare o un vettore per-gamba (costo asset-specifico).""" n, k = Weff.shape w = np.zeros(k) cap = cap0 eq = np.empty(n) nets = np.zeros(n) n_fill = n_skip = n_lot0 = 0 fee = np.broadcast_to(np.asarray(fee_leg, float).ravel(), (k,)) if np.ndim(fee_leg) else None for i in range(n): ret = float(np.dot(w, R[i])) if r_hedge is not None: ret -= float(w.sum()) * r_hedge[i] w_t = Weff[i] if min_order is None: w_new = w_t.copy() else: move = np.abs(w_t - w) * cap >= min_order if lot_usd is not None and px is not None: # arrotondamento al lotto: la size eseguita e' un multiplo di lot_units dn = np.abs(w_t - w) * cap lots = np.floor(dn / np.maximum(lot_usd, 1e-12)) zero = move & (lots < 1) n_lot0 += int(zero.sum()) move = move & (lots >= 1) need = np.abs(w_t - w) > 1e-12 # una gamba gia' a posto NON e' un ordine saltato w_new = np.where(move, w_t, w) n_fill += int((move & need).sum()) n_skip += int(((~move) & need).sum()) d = np.abs(w_new - w) c = float((d * fee).sum()) if fee is not None else float(fee_leg) * float(d.sum()) if r_hedge is not None: # la gamba di copertura paga anch'essa dh = abs(float(w_new.sum()) - float(w.sum())) c += (float(fee.mean()) if fee is not None else float(fee_leg)) * dh net = ret - c nets[i] = net cap *= (1.0 + max(net, -0.99)) eq[i] = cap w = w_new tot = n_fill + n_skip return dict(net=nets, eq=eq, n_fill=n_fill, n_skip=n_skip, n_lot0=n_lot0, pct_eseguite=(100.0 * n_fill / tot) if tot else np.nan) def metriche(net: np.ndarray, idx: pd.DatetimeIndex) -> dict: x = np.asarray(net, float) sd = x.std(ddof=1) sh = float(x.mean() / sd * ANN) if sd > 0 else 0.0 eq = np.cumprod(1.0 + np.clip(x, -0.99, None)) dd = float((1.0 - eq / np.maximum.accumulate(eq)).max()) yrs = max((idx[-1] - idx[0]) / pd.Timedelta("365.25D"), 1e-9) cagr = float(eq[-1] ** (1.0 / yrs) - 1.0) return dict(sharpe=round(sh, 3), maxdd=round(100 * dd, 1), cagr=round(100 * cagr, 1)) # ================================================================================================== # 4. XS01 — ricostruzione dei PESI (il sleeve espone solo i ritorni) + prova d'identita' # ================================================================================================== def xs01_pesi(): from src.portfolio.sleeves import XS_CFG, XS_UNIVERSE, _xsec_returns cols = {} for sym in XS_UNIVERSE: p = RAW / 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 lb, H, k, mode, tv = (XS_CFG["lookbacks"], XS_CFG["H"], XS_CFG["k"], XS_CFG["mode"], XS_CFG["target_vol"]) dp, mh = XS_CFG.get("disp_pct", 0), XS_CFG.get("disp_minhist", 20) mlb = max(lb) dret = np.vstack([np.zeros(A), px[1:] / px[:-1] - 1.0]) W = np.zeros((n, A)) w = np.zeros(A) hist: list[float] = [] for i in range(n): if i >= mlb and i % H == 0: rLs = [px[i] / px[i - L] - 1.0 for L in lb] di = float(np.mean([r.std() for r in rLs])) thr = np.percentile(hist, dp) if (dp > 0 and len(hist) >= mh) else -np.inf if di >= thr: sc = np.zeros(A); cnt = 0 for rL in rLs: sd = rL.std() if sd > 0: sc += (rL - rL.mean()) / sd; cnt += 1 if cnt: sc /= cnt o = np.argsort(sc); w = np.zeros(A); lo, hi = o[:k], o[-k:] if mode == "mom": w[hi] = 0.5 / k; w[lo] = -0.5 / k else: w[lo] = 0.5 / k; w[hi] = -0.5 / k else: w = np.zeros(A) hist.append(di) 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) # prova d'identita' col sleeve UFFICIALE (max|diff| deve essere 0.0) off = _xsec_returns() dif = float(np.abs(pd.Series(s.values * scale, index=C.index).reindex(off.index).values - off.values).max()) # posizione TENUTA durante la barra i = W[i-1]*scale[i] => posta a fine barra i: W[i]*scale[i+1] pos_scale = np.concatenate([scale[1:], scale[-1:]]) return W * pos_scale[:, None], dret, C, list(C.columns), dif # ================================================================================================== # MAIN # ================================================================================================== def main() -> None: ap = argparse.ArgumentParser() ap.add_argument("--no-net", action="store_true") ap.add_argument("--snaps", type=int, default=3) a = ap.parse_args() pd.set_option("display.width", 200) print("=" * 104) print(" HL-EXEC — le regole VERE di Hyperliquid contro le soglie ASSUNTE di XS01 / XSR01") print("=" * 104) # ---------------------------------------------------------------- 1. regole meta, fee_sched = leggi_regole(a.no_net) syms = universo_certificato() print(f"\n[1] REGOLE DEL VENUE (fonte: {INFO_URL} = MAINNET)") print(f" universo perp quotato: {len(meta)} strumenti; certificati da noi: {len(syms)}") mancanti = [s for s in syms if s not in meta.index] delist = [s for s in syms if s in meta.index and bool(meta.loc[s, "delisted"])] print(f" certificati assenti dal venue: {mancanti or 'nessuno'} delistati: {delist or 'nessuno'}") ck = verifica_mainnet(meta, syms) print(f"\n [A] verifica MAINNET (mark API vs ultima chiusura del feed certificato, " f"eta' feed {ck['eta_h'].median():.0f}h):") print(f" deviazione |%| mediana {ck['dev_pct'].abs().median():.2f}% " f"max {ck['dev_pct'].abs().max():.2f}% ({ck['dev_pct'].abs().idxmax()}) " f"asset controllati {len(ck)}") print(" -> un feed TESTNET non supera questo controllo: i suoi prezzi sono scollegati.") tk = fee_sched print(f"\n [A] FEE perp lette dal venue (tier BASE, nessuna chiave):") print(f" taker (cross) {float(tk['cross'])*1e4:.2f} bps/lato " f"maker (add) {float(tk['add'])*1e4:.2f} bps/lato") print(f" primo scaglione VIP a ${float(tk['tiers']['vip'][0]['ntlCutoff']):,.0f} di " f"volume 14g -> a $600-20k si sta SEMPRE al tier base.") print(f" [C] la doc (gitbook/trading/fees) dichiara 0.045% / 0.015%: **le due fonti " f"coincidono**.") print(f" [C] pavimento d'ordine perp: \"Order must have minimum value of $10.\" " f"(docs/for-developers/api/error-responses); nessuna esenzione documentata per reduce-only.") print(f" [C] tick: <=5 cifre significative E <= (6 - szDecimals) decimali; size arrotondata " f"a szDecimals.") sub = meta.loc[[s for s in syms if s in meta.index]].copy() print(f"\n [A/B] LOTTO per asset (passo di size) — i 6 piu' grossolani e i 6 piu' fini:") o = sub.sort_values("lot_usd", ascending=False) for who, part in (("piu' grossolano", o.head(6)), ("piu' fine", o.tail(6))): for s, r in part.iterrows(): print(f" {s:<6} szDec={int(r.szDecimals)} lotto {r.lot_units:>10.5f} unita' = " f"${r.lot_usd:>8.4f} mark ${r.markPx:<12.6g} ({who})") print(f" lotto in $: mediana ${sub['lot_usd'].median():.4f}, max " f"${sub['lot_usd'].max():.2f} ({sub['lot_usd'].idxmax()})") print(f" -> il LOTTO non e' mai il vincolo: max ${sub['lot_usd'].max():.2f} << $10 " f"di pavimento. Il vincolo e' il MIN NOTIONAL.") print(f" [A] leva massima: BTC {int(meta.loc['BTC','maxLeverage'])}x, " f"ETH {int(meta.loc['ETH','maxLeverage'])}x, mediana alt certificati " f"{int(sub['maxLeverage'].median())}x, minimo {int(sub['maxLeverage'].min())}x") # ---------------------------------------------------------------- 2. book print(f"\n[2] BOOK L2 — spread, profondita', costo di un ticket ({a.snaps} snapshot per asset)") books = scarica_book(syms, a.snaps, a.no_net) liq = misura_liquidita(books, meta) mn = min_notional_osservato(books) if mn: print(f" [B] controllo empirico del pavimento $10 su {mn['n']} livelli a UN SOLO ordine " f"(n==1):") print(f" minimo ${mn['minimo']:.2f} p01 ${mn['p01']:.2f} p05 ${mn['p05']:.2f} " f"mediana ${mn['mediana']:.0f} sotto $10: {mn['sotto10_pct']:.1f}% " f"sotto $5: {mn['sotto5_pct']:.1f}%") print(f" (un residuo sotto soglia NON falsifica la regola — i fill parziali erodono " f"un ordine gia' piazzato — ma il pavimento della distribuzione dice dove taglia.)") viol = int(liq["tick_viol"].sum()) print(f" [B] regola tick verificata sui book veri: {viol} violazioni su " f"{int(liq['n_snap'].sum())*40} livelli letti -> la regola [C] e' CONFERMATA dal venue.") print(f"\n spread e slippage (mediana degli snapshot; bps DA MID, un solo lato):") print(f" {'sym':<7}{'vol24h $':>13}{'spread':>9}{'1/2 spr':>9}" f"{'prof<10bp':>11}{'$50':>8}{'$100':>8}{'$300':>8}") for s, r in liq.iterrows(): print(f" {s:<7}{r.dayNtlVlm if 'dayNtlVlm' in r else meta.loc[s,'dayNtlVlm']:>13,.0f}" f"{r.spread_bps:>8.1f}{r.half_spread_bps:>9.1f}{r.depth10bps_usd:>11,.0f}" f"{r.slip50_bps:>8.1f}{r.slip100_bps:>8.1f}{r.slip300_bps:>8.1f}") XS19 = ["BTC", "ETH", "SOL", "BNB", "XRP", "DOGE", "AVAX", "LINK", "LTC", "ADA", "ARB", "OP", "SUI", "APT", "INJ", "TIA", "SEI", "NEAR", "AAVE"] maj = liq.reindex([s for s in XS19 if s in liq.index]) coda = liq.drop(index=maj.index, errors="ignore") print(f"\n riepilogo (mediana | p90):") for nm, part in (("19 major (XS01)", maj), (f"coda ({len(coda)} alt, solo XSR01)", coda)): if len(part): print(f" {nm:<26} spread {part.spread_bps.median():>5.1f} | " f"{part.spread_bps.quantile(.9):>5.1f} bps " f"slip$100 {part.slip100_bps.median():>5.1f} | " f"{part.slip100_bps.quantile(.9):>5.1f} " f"slip$300 {part.slip300_bps.median():>5.1f} | " f"{part.slip300_bps.quantile(.9):>5.1f}") # ---------------------------------------------------------------- 3. XS01 print(f"\n[3] XS01 — soglia PUBBLICATA ~$20.000 (origine: stima a occhio, 'rumore arrotondamento')") Wxs, dret, C, cols, dif = xs01_pesi() print(f" prova d'identita' col sleeve ufficiale `_xsec_returns()`: max|diff| = {dif:.3e}") from src.portfolio.sleeves import XS_CFG, _xsec_returns off_xs = _xsec_returns() m_off = metriche(off_xs.values, pd.DatetimeIndex(off_xs.index)) gross_xs = np.abs(Wxs).sum(axis=1) att = gross_xs > 1e-9 dW = np.abs(np.diff(Wxs, axis=0, prepend=np.zeros((1, Wxs.shape[1])))) tk_nz = dW[dW > 1e-9] # DUE popolazioni di ordini, e vanno separate: il RIBILANCIAMENTO del segnale (ogni H=10 giorni, # cambia il paniere 5+5) e la DERIVA del vol-target (ogni giorno, micro-aggiustamenti). H = XS_CFG["H"] is_reb = np.zeros(len(Wxs), bool); is_reb[::H] = True reb = dW[is_reb]; reb = reb[reb > 1e-9] drift = dW[~is_reb]; drift = drift[drift > 1e-9] print(f" sleeve ufficiale `_xsec_returns()`: Sharpe {m_off['sharpe']:.2f} " f"maxDD {m_off['maxdd']:.1f}% CAGR {m_off['cagr']:.1f}%") print(f" gambe simultanee: {int((np.abs(Wxs) > 1e-9).sum(axis=1)[att].max())} " f"(k=5 long + 5 short); lordo mediano {np.median(gross_xs[att]):.2f}x il capitale " f"(vol-target, cap 3x)") print(f"\n DUE popolazioni di ordini (separarle e' il punto: non hanno la stessa taglia):") print(f" RIBILANCIAMENTO (ogni {H}g, cambia il paniere): n={len(reb)} " f"peso mosso mediano {np.median(reb)*100:.2f}% p10 {np.percentile(reb,10)*100:.2f}%") print(f" DERIVA vol-target (giornaliera) : n={len(drift)} " f"peso mosso mediano {np.median(drift)*100:.3f}% p10 {np.percentile(drift,10)*100:.3f}%") print(f" -> {100*len(drift)/max(len(reb)+len(drift),1):.0f}% degli ordini sono DERIVA, " f"ed e' la deriva che il pavimento taglia.") print(f"\n ticket per gamba e % di ordini RICHIESTI eseguibili, per capitale allocato a XS01") print(f" (conto = capitale/0.15, perche' XS01 pesa 15% del book di ricerca):") print(f" {'cap XS01':>10}{'conto @15%':>12}{'tick reb':>10}{'tick drift':>12}" f"{'>=$5 [D]':>10}{'>=$10 [C]':>11}{'reb>=$10':>10}") R_xs = dret idx = C.index base = None for cap in CAPITALI: pct5 = 100.0 * float((tk_nz * cap >= MIN_ORDER_ASSUNTO).mean()) pct10 = 100.0 * float((tk_nz * cap >= MIN_ORDER_HL).mean()) pr10 = 100.0 * float((reb * cap >= MIN_ORDER_HL).mean()) print(f" {cap:>10,.0f}{cap/0.15:>12,.0f}{np.median(reb)*cap:>10.2f}" f"{np.median(drift)*cap:>12.2f}{pct5:>9.0f}%{pct10:>10.0f}%{pr10:>9.0f}%") lot_xs = np.array([float(meta.loc[s, "lot_usd"]) if s in meta.index else 0.0 for s in cols]) px_xs = np.array([float(meta.loc[s, "markPx"]) if s in meta.index else np.nan for s in cols]) fee_xs_mod = 0.001 / 2.0 # [D] cio' che il sleeve modella: 5 bps/lato print(f"\n haircut di eseguibilita' (Sharpe modellato - Sharpe realistico), fee {fee_xs_mod*1e4:.1f} bps/lato.") print(f" NB il 'Sh mod' qui INCLUDE la fee sulla deriva del vol-target, che lo sleeve " f"ufficiale NON addebita") print(f" (`turn=|dW|` prima dello scaling): e' il motivo per cui sta sotto il " f"{m_off['sharpe']:.2f} pubblicato.") print(f" {'cap XS01':>10}{'Sh mod':>9}{'Sh $5 [D]':>11}{'Sh $10 [C]':>12}" f"{'haircut $10':>13}{'ordini fatti':>14}") for cap in CAPITALI: m0 = sim_libro(Wxs, R_xs, None, cap, None, None, None, fee_xs_mod) m5 = sim_libro(Wxs, R_xs, None, cap, MIN_ORDER_ASSUNTO, None, None, fee_xs_mod) m10 = sim_libro(Wxs, R_xs, None, cap, MIN_ORDER_HL, lot_xs, px_xs, fee_xs_mod) a0, a5, a10 = (metriche(m["net"], idx) for m in (m0, m5, m10)) hc = (a0["sharpe"] - a10["sharpe"]) / abs(a0["sharpe"]) * 100 if a0["sharpe"] else np.nan print(f" {cap:>10,.0f}{a0['sharpe']:>9.2f}{a5['sharpe']:>11.2f}{a10['sharpe']:>12.2f}" f"{hc:>12.0f}%{m10['pct_eseguite']:>9.0f}%") # ---------------------------------------------------------------- 4. XSR01 print(f"\n[4] XSR01 — soglia PUBBLICATA ~$5.000, gate 2026-10-23 con haircut a $5.000 <= 40%") from scripts.live.paper_xsr import FEE_LEG, MIN_ORDER, _book, _step, build_panel ts, dtx, Wx, Rx, rbx, sx = build_panel() idxx = pd.DatetimeIndex(dtx) # PROVA D'IDENTITA' col monitor IN PRODUZIONE: sim_libro deve riprodurre paper_xsr._step # barra per barra. Senza questa, i numeri qui sotto sarebbero di un'altra macchina. for mo in (None, MIN_ORDER): bk = _book(5000.0, len(sx)) ref = np.array([_step(bk, Wx[i], Rx[i], float(rbx[i]), mo) for i in range(len(Wx))]) mine = sim_libro(Wx, Rx, rbx, 5000.0, mo, None, None, FEE_LEG)["net"] print(f" identita' vs scripts/live/paper_xsr._step (min_order={mo}): " f"max|diff| = {np.abs(ref - mine).max():.3e}") assert abs(FEE_LEG - FEE_LEG_MODELLO) < 1e-12, "la config congelata di XSR01 e' cambiata" assert abs(MIN_ORDER - MIN_ORDER_ASSUNTO) < 1e-12, "il min-order del monitor e' cambiato" # RICONCILIAZIONE col numero PUBBLICATO (1.82): la scoperta (r0725_statarb_basket_gate, # basket_from_positions) addebita "2 gambe per coppia" = 50 alt + 50 BTC. Ma il demeaning # ANNULLA la gamba BTC (sum dei pesi = 0): non si tradano 50 BTC, se ne trada UNA di taglia # zero. Il monitor in produzione lo sa (`d_btc = |sum(w_new) - sum(w_held)|` ~ 0) -> il # pubblicato SOVRASTIMA le fee di ~2x. net_x2 = sim_libro(Wx, Rx, rbx, 5000.0, None, None, None, 2 * FEE_LEG_MODELLO)["net"] print(f" riconciliazione col pubblicato 1.82: a fee RADDOPPIATA (= convenzione della " f"scoperta, 2 gambe/coppia) Sharpe {metriche(net_x2, idxx)['sharpe']:.2f}") lot_x = np.array([float(meta.loc[s, "lot_usd"]) if s in meta.index else 0.0 for s in sx]) px_x = np.array([float(meta.loc[s, "markPx"]) if s in meta.index else np.nan for s in sx]) dWx = np.abs(np.diff(Wx, axis=0, prepend=np.zeros((1, Wx.shape[1])))) tkx = dWx[dWx > 1e-9] print(f" gambe {len(sx)} lordo mediano {np.median(np.abs(Wx).sum(axis=1)):.2f}x " f"peso mosso mediano/gamba {np.median(tkx)*100:.3f}% p10 {np.percentile(tkx,10)*100:.3f}%") print(f"\n {'cap':>9}{'ticket med':>12}{'Sh mod':>9}{'Sh $5 [D]':>11}{'Sh $10 [C]':>12}" f"{'haircut $5':>12}{'haircut $10':>13}{'ordini fatti':>14}") hc10_5000 = np.nan for cap in CAPITALI: m0 = sim_libro(Wx, Rx, rbx, cap, None, None, None, FEE_LEG_MODELLO) m5 = sim_libro(Wx, Rx, rbx, cap, MIN_ORDER_ASSUNTO, None, None, FEE_LEG_MODELLO) m10 = sim_libro(Wx, Rx, rbx, cap, MIN_ORDER_HL, lot_x, px_x, FEE_LEG_MODELLO) a0, a5, a10 = (metriche(m["net"], idxx) for m in (m0, m5, m10)) h5 = (a0["sharpe"] - a5["sharpe"]) / abs(a0["sharpe"]) * 100 if a0["sharpe"] else np.nan h10 = (a0["sharpe"] - a10["sharpe"]) / abs(a0["sharpe"]) * 100 if a0["sharpe"] else np.nan if abs(cap - 5000.0) < 1e-9: hc10_5000 = h10 print(f" {cap:>9,.0f}{np.median(tkx)*cap:>12.2f}{a0['sharpe']:>9.2f}" f"{a5['sharpe']:>11.2f}{a10['sharpe']:>12.2f}{h5:>11.0f}%{h10:>12.0f}%" f"{m10['pct_eseguite']:>9.0f}%") print(f" -> haircut a $5.000 col pavimento VERO ($10): {hc10_5000:.0f}% " f"(soglia pre-registrata {XSR_SOGLIA_HAIRCUT*100:.0f}%)") print(f" NB: e' la lettura del PARAMETRO sul backtest, NON il gate del 23/10 — " f"quello si decide sulla finestra FORWARD e anticiparlo sarebbe selezione.") # -------- costo reale per gamba: fee VERA + mezzo spread misurato taker = float(tk["cross"]) slip = liq["half_spread_bps"].reindex(sx).astype(float) / 1e4 slip_med = float(np.nanmedian(slip.values)) fee_real = taker + np.nan_to_num(slip.values, nan=slip_med) print(f"\n [B] COSTO REALE PER GAMBA = taker {taker*1e4:.2f} bps + mezzo spread misurato:") print(f" mediana {np.median(fee_real)*1e4:.1f} bps p90 {np.percentile(fee_real,90)*1e4:.1f} bps" f" max {fee_real.max()*1e4:.1f} bps ({sx[int(np.argmax(fee_real))]})") print(f" contro i {FEE_LEG_MODELLO*1e4:.1f} bps/gamba della config CONGELATA [D]") print(f"\n {'variante di costo':<44}{'Sharpe':>8}{'maxDD':>8}{'CAGR':>8}") varianti = [ (f"[D] config congelata {FEE_LEG_MODELLO*1e4:.1f} bps piatti", FEE_LEG_MODELLO), (f"[A] solo taker vero {taker*1e4:.2f} bps piatti", taker), (f"[B] taker + 1/2 spread PER-ASSET (mediana {np.median(fee_real)*1e4:.1f})", fee_real), (f"[B] taker + spread PIENO per-asset (pessimista)", taker + 2 * np.nan_to_num(slip.values, nan=slip_med)), ] for nm, f in varianti: m = sim_libro(Wx, Rx, rbx, 5000.0, None, None, None, f) a_ = metriche(m["net"], idxx) print(f" {nm:<44}{a_['sharpe']:>8.2f}{a_['maxdd']:>7.1f}%{a_['cagr']:>7.1f}%") print(f" (a capitale infinito = nessun vincolo di min-order: isola il COSTO dal PAVIMENTO)") print("\n" + "=" * 104) print(" LIMITI DICHIARATI: gli spread sono UNO/POCHI snapshot di OGGI su un venue la cui") print(" liquidita' e' cresciuta -> sono la stima piu' FAVOREVOLE per i 2.6 anni di backtest.") print(" Il pavimento $10 e' [C] (doc + messaggio d'errore), non provato da un ordine: provarlo") print(" richiederebbe di mandarne uno, e questo audit non tocca il conto.") print("=" * 104) if __name__ == "__main__": main()