#!/usr/bin/env python """r0724_skh_live_weight.py — FOLLOW-UP dell'audit anchor 2026-07-02 su SKH01: "cadenza 230m, peso live da rivedere" (diario 2026-07-02-anchor-audit-xs01-skh01.md). CONTESTO. Il book live Deribit e' TP01 0.75 + SKH01 0.25 (sleeves.deribit_book_sleeves). L'audit r0702 ha mostrato che il PATH LIVE reale (cron ORARIO `0 * * * *` + exit software, non on-book al livello) degrada SKH01: alla lente 50/50 il book passa FULL 1.46->1.19, HOLD 1.64->1.15, DD 18->25%; nei crash gap-through-stop (sl 2% modellato -> -11/-23% realizzato). In piu' l'offset 0 della griglia 230m/690m e' il 93-98 pctl dei 23 offset a priori (best-of-23). DOMANDA: dato il path live vero, qual e' il peso SKH ottimale del book a 2 sleeve, e quanto vale allineare il cron alla cadenza 230m? COSA FA (riusa la macchineria di r0702_anchor_skh01 — import diretto, NON reinventata): 1. SANITY: r02.sanity() (off0 == _skyhook_returns bit-exact) + riconciliazione con i numeri dell'audit (50/50 canonical off0 FULL ~1.46 / HOLD ~1.64 / DD ~18.1%; hourly ~1.19 / ~1.15 / ~24.6%; blend 75/25 canonical HOLD ~1.16). NB: il feed live aggiunge barre ogni giorno -> attesa una piccola deriva-dati vs i numeri del 2026-07-02. 2. Serie SKH live-path: per TUTTI i 23 offset a priori x {canonical, barclose, hourly} (sim_equity di r0702; 'hourly' = fill al close 5m del prossimo multiplo orario = path del cron reale; 'barclose' = fill al close della barra 230m = cio' che otterrebbe un cron allineato ai confini 230m; 'canonical' = fill-al-livello del backtest, riferimento). 3. SWEEP PESO: book = (1-w)*TP01 + w*SKH_live, w in {0,.10,.15,.20,.25,.30,.40,.50} (8 valori = multiple-testing leggero, DICHIARATO; nel gate k_seen=8). TP01 = al.tp01_baseline_daily() (canonico, come nell'audit). Per ogni w: Sharpe FULL / IS (pre-2025) / HOLD (2025+) / maxDD, su off0 E min/med/max sui 23 offset. REGOLA DI SELEZIONE (dichiarata A PRIORI, prima di vedere i numeri): w* = argmax della MEDIANA sugli offset dello Sharpe IN-SAMPLE (pre-holdout, lente hourly = path live, de-luckata). A parita' (<0.03 di Sharpe) si preferisce il w piu' VICINO al corrente 0.25 (conservativo). L'hold-out serve SOLO a verificare, non a scegliere. 4. GATE weights_tilt_null (src/portfolio/portfolio.py) adattato al book a 2 sleeve: null = tilt casuali cap-respecting (floor 0.05, cap SKH 0.50 = estremo dello sweep) attorno al 75/25 corrente, sui daily del PATH LIVE (hourly, off0 = griglia che gira davvero). k_seen=8 (i w guardati sull'hold-out in questo studio). Se il w* non passa il gate -> verdetto onesto "INVARIATO". 5. CADENZA 230m: valore di allineare l'esecuzione ai confini 230m = barclose vs hourly al peso corrente 0.25 (dSharpe FULL/HOLD, dDD), off0 + banda sugli offset. Fattibilita' SOLO STIMATA (non cablata): 230 = 23x10 -> i confini 230m cadono SEMPRE su multipli di 10 minuti dall'epoch => un cron a passo 10m che triggera solo quando (now % 230min) == 0 colpisce ogni confine con ritardo ~0 (vs 0-50 min, media 25, del cron orario) -> 'barclose' e' esattamente il suo modello (al netto di runtime/slippage). CAVEAT DICHIARATI (onesta'): - l'exit software nei crash prende il GAP (gia' misurato dall'audit: sl2% -> -11/-23% realizzato); la sim 'hourly' e' gia' la stima pessimistica giusta del path live e la cadenza 230m NON recupera il gap canonical->barclose (fill-al-livello: nessun cron riporta l'exit software sul livello SL dentro una barra che gappa); - equity daily-step (lens Sharpe), stessa convenzione del canonico e dell'audit; - nessuna selezione sull'hold-out: w* scelto sull'IS mediano-di-banda, hold-out solo verifica; 8 valori di w provati e dichiarati (k_seen=8 nel gate); - la banda sugli offset e' la lente de-luckata; live gira UNA sola griglia (off 0); - deriva-dati vs audit (il feed cresce ogni giorno): riconciliazione "circa", riportata. VINCOLI: nessun file di produzione toccato (config/live.json, sleeves.py, scripts/live, cron INVARIATI); niente commit; dati SOLO dal feed certificato (get5m di r0702 -> load_data). Runtime ~4-8 min (46 run skyhook cachati + 138 sim_equity leggere + sweep + gate). """ from __future__ import annotations import sys import time from pathlib import Path import numpy as np import pandas as pd ROOT = Path("/opt/docker/PythagorasGoal") sys.path.insert(0, str(ROOT / "scripts" / "research")) sys.path.insert(0, str(ROOT / "scripts" / "research" / "alt")) sys.path.insert(0, str(ROOT)) import r0702_anchor_skh01 as r02 # noqa: E402 (macchineria audit: run_asset/sim_equity/get5m/sanity) import altlib as al # noqa: E402 from src.portfolio.portfolio import combine_outer, weights_tilt_null # noqa: E402 HOLDOUT = r02.HOLDOUT ASSETS = r02.ASSETS OFFSETS = r02.OFFSETS # 23 offset a priori (ogni 30m su [0,690)), 0 = canonico MODES = ("canonical", "barclose", "hourly") W_GRID = (0.0, 0.10, 0.15, 0.20, 0.25, 0.30, 0.40, 0.50) # 8 valori — dichiarati (k_seen=8) W_CURRENT = 0.25 TIE_TOL = 0.03 # regola a priori: parita' se |dSh IS| < 0.03 -> w più vicino a 0.25 # --------------------------------------------------------------------------- # Serie daily per (asset, offset, mode) — riusa run_asset + sim_equity di r0702 # --------------------------------------------------------------------------- _D: dict = {} def skh_daily_asset(a: str, off: int, mode: str) -> pd.Series: key = (a, off, mode) if key in _D: return _D[key] _, m0, ltf, ent = r02.run_asset(a, off) df5 = r02.get5m(a) ts5 = df5["timestamp"].values.astype(np.int64) + r02.MS5 c5 = df5["close"].values.astype(float) eq = r02.sim_equity(ltf, ent, mode, ts5_close=ts5, c5=c5) if mode == "canonical": # sanity strutturale: re-sim == harness bit-exact dmax = float(np.max(np.abs(eq - m0.equity))) assert dmax < 1e-6, f"re-sim canonica != harness ({a}, off={off}): max|dif|={dmax:.2e}" s = pd.Series(eq, index=pd.DatetimeIndex(pd.to_datetime(ltf["datetime"], utc=True))) _D[key] = s.resample("1D").last().ffill().pct_change().dropna() return _D[key] def skh_book(off: int, mode: str) -> pd.Series: """Book 50/50 BTC+ETH daily-step (convenzione di _skyhook_returns / audit).""" J = pd.concat({a: skh_daily_asset(a, off, mode) for a in ASSETS}, axis=1, join="inner").fillna(0.0) return pd.Series(0.5 * J["BTC"].values + 0.5 * J["ETH"].values, index=J.index) def stats4(s: pd.Series) -> tuple[float, float, float, float]: f, i_, h = r02.sh3(s) return f, i_, h, al._dd_ret(s) # --------------------------------------------------------------------------- # Main # --------------------------------------------------------------------------- def main() -> None: t0 = time.time() print("=" * 100) print("r0724 — SKH01 sul PATH LIVE (cron orario): peso ottimale del book Deribit a 2 sleeve") print(f"+ valore della cadenza 230m. 23 offset a priori, fee 0.10% RT, HOLD-OUT >= {HOLDOUT.date()}") print("=" * 100) # ---- (1) SANITY + riconciliazione con l'audit r0702 -------------------- r02.sanity() b0 = {m: skh_book(0, m) for m in MODES} print("\n[SANITY] riconciliazione col r0702 (2026-07-02; attesa deriva-dati: il feed cresce):") print(f"{'modo':<10} {'ShFULL':>7} {'ShIS':>7} {'ShHOLD':>7} {'maxDD':>7} audit r0702") ref = {"canonical": "FULL 1.46 / HOLD 1.64 / DD 18.1%", "barclose": "(intermedio, isola il fill-al-livello)", "hourly": "FULL 1.19 / HOLD 1.15 / DD 24.6%"} for m in MODES: f, i_, h, dd = stats4(b0[m]) print(f"{m:<10} {f:>7.3f} {i_:>7.3f} {h:>7.3f} {dd:>6.1%} {ref[m]}") B = al.tp01_baseline_daily() fB, iB, hB, ddB = stats4(B) print(f"\nTP01 baseline canonico (al.tp01_baseline_daily): FULL {fB:.3f} / IS {iB:.3f} / " f"HOLD {hB:.3f} / DD {ddB:.1%} (audit: FULL ~1.30, HOLD ~0.31)") def blend(skh: pd.Series, w: float) -> pd.Series: if w == 0.0: return B bl = combine_outer({"TP": B, "SKH": skh}, {"TP": 1.0 - w, "SKH": w}) return bl[bl.index >= B.index.min()] f, i_, h, dd = stats4(blend(b0["canonical"], 0.25)) print(f"blend 75/25 canonical off0: FULL {f:.3f} / HOLD {h:.3f} (audit: HOLD ~1.16)") # ---- (2) tutte le serie: 23 offset x 3 modi ---------------------------- print(f"\n[{time.time()-t0:5.0f}s] genero i 23 offset x 3 modi (run skyhook cachati)...") SKB: dict[str, dict[int, pd.Series]] = {m: {} for m in MODES} for off in OFFSETS: for m in MODES: SKB[m][off] = b0[m] if off == 0 else skh_book(off, m) fh, ih, hh, ddh = stats4(SKB["hourly"][off]) print(f" [{time.time()-t0:5.0f}s] off {off:>3}m: hourly FULL {fh:+.2f} IS {ih:+.2f} " f"HOLD {hh:+.2f} DD {ddh:.0%}") # ---- (3) SWEEP PESO ---------------------------------------------------- print("\n" + "=" * 100) print(f"(3) SWEEP PESO — book (1-w)*TP01 + w*SKH_live, w in {list(W_GRID)}") print(" (8 valori = multiple-testing leggero, dichiarato; selezione su IS mediano, lente hourly)") print("=" * 100) ST: dict[str, dict[float, dict[int, tuple]]] = {m: {w: {} for w in W_GRID} for m in MODES} for m in MODES: for w in W_GRID: for off in OFFSETS: ST[m][w][off] = stats4(blend(SKB[m][off], w)) def band(m: str, w: float, k: int) -> tuple[float, float, float]: v = np.array([ST[m][w][off][k] for off in OFFSETS]) return float(v.min()), float(np.median(v)), float(v.max()) for m in ("hourly", "canonical"): lens = "PATH LIVE (lente della decisione)" if m == "hourly" else "riferimento backtest" print(f"\n--- modo {m.upper()} — {lens} ---") print(f"{'w':>5} | {'off0:':<6}{'FULL':>6} {'IS':>6} {'HOLD':>6} {'DD':>6} | " f"{'med:':<5}{'FULL':>6} {'IS':>6} {'HOLD':>6} {'DD':>6} | " f"{'HOLD min..max':>14} | {'DD min..max':>13}") for w in W_GRID: f0, i0, h0, d0 = ST[m][w][0] fm = band(m, w, 0)[1]; im = band(m, w, 1)[1] hmn, hmd, hmx = band(m, w, 2) dmn, dmd, dmx = band(m, w, 3) cur = " <- corrente" if w == W_CURRENT else "" print(f"{w:>5.2f} | {'':<6}{f0:>6.2f} {i0:>6.2f} {h0:>6.2f} {d0:>5.1%} | " f"{'':<5}{fm:>6.2f} {im:>6.2f} {hmd:>6.2f} {dmd:>5.1%} | " f"{hmn:>6.2f}..{hmx:<6.2f} | {dmn:>5.1%}..{dmx:<5.1%}{cur}") # ---- (4) selezione (regola a priori) ----------------------------------- print("\n--- (4) SELEZIONE del w* (regola dichiarata a priori) ---") med_is = {w: band("hourly", w, 1)[1] for w in W_GRID} best_val = max(med_is.values()) ties = [w for w in W_GRID if best_val - med_is[w] < TIE_TOL] wstar = min(ties, key=lambda w: (abs(w - W_CURRENT), w)) print("mediana-IS (hourly) per w: " + ", ".join(f"{w:.2f}:{med_is[w]:.3f}" for w in W_GRID)) print(f"argmax = {max(med_is, key=med_is.get):.2f} ({best_val:.3f}); " f"in parita' (<{TIE_TOL}): {[f'{w:.2f}' for w in ties]} -> " f"w* = {wstar:.2f} (il piu' vicino al corrente {W_CURRENT:.2f})") h_star_med = band("hourly", wstar, 2)[1]; h_cur_med = band("hourly", W_CURRENT, 2)[1] d_star_med = band("hourly", wstar, 3)[1]; d_cur_med = band("hourly", W_CURRENT, 3)[1] print(f"verifica hold-out (mediana banda): w*={wstar:.2f} HOLD {h_star_med:.2f} / DD {d_star_med:.1%}" f" vs corrente 0.25: HOLD {h_cur_med:.2f} / DD {d_cur_med:.1%}") # ---- (5) GATE weights_tilt_null --------------------------------------- print("\n--- (5) GATE weights_tilt_null (book a 2 sleeve, daily del PATH LIVE hourly) ---") print("null = tilt casuali floor 0.05 / cap SKH 0.50 attorno al 75/25 corrente; k_seen=8") if wstar == W_CURRENT: print(f"w* == corrente ({W_CURRENT:.2f}) -> nessun cambio proposto: verdetto INVARIATO " f"(gate non necessario). Lo eseguo comunque sul miglior w != corrente, a titolo informativo.") alt = [w for w in W_GRID if w != W_CURRENT and w >= 0.05] w_gate = max(alt, key=lambda w: med_is[w]) else: w_gate = wstar # primario: off0 (la griglia che gira davvero live); robustezza: offset con IS mediano a w corrente is_cur = {off: ST["hourly"][W_CURRENT][off][1] for off in OFFSETS} off_med = sorted(OFFSETS, key=lambda o: is_cur[o])[len(OFFSETS) // 2] for tag, off in (("off0 (griglia live)", 0), (f"off{off_med} (offset IS-mediano)", off_med)): cols = {"TP01": B, "SKH01": SKB["hourly"][off]} res = weights_tilt_null(cols, {"TP01": 1 - W_CURRENT, "SKH01": W_CURRENT}, {"TP01": 1 - w_gate, "SKH01": w_gate}, caps={"SKH01": 0.50}, floor=0.05, n=500, k_seen=8) res.pop("samples", None) print(f" [{tag}] proposto SKH {w_gate:.2f}: " + ", ".join(f"{k}={v}" for k, v in res.items())) # ---- (6) CADENZA 230m -------------------------------------------------- print("\n--- (6) CADENZA 230m — barclose (cron allineato ai confini) vs hourly (cron 0 * * * *) ---") print(f"al peso corrente w={W_CURRENT:.2f}:") rows = [] for off in OFFSETS: fb, ib, hb, db = ST["barclose"][W_CURRENT][off] fh, ih, hh, dh = ST["hourly"][W_CURRENT][off] rows.append((fb - fh, hb - hh, db - dh)) A = np.array(rows) print(f" off0: dShFULL {A[0,0]:+.3f}, dShHOLD {A[0,1]:+.3f}, dDD {A[0,2]:+.1%}") print(f" banda: dShFULL med {np.median(A[:,0]):+.3f} [{A[:,0].min():+.3f},{A[:,0].max():+.3f}], " f"dShHOLD med {np.median(A[:,1]):+.3f} [{A[:,1].min():+.3f},{A[:,1].max():+.3f}], " f"dDD med {np.median(A[:,2]):+.1%} [{A[:,2].min():+.1%},{A[:,2].max():+.1%}]") # riferimento: SKH standalone 50/50 e quota NON recuperabile (canonical - barclose) rows_s = [] for off in OFFSETS: fb, _, hb, db = stats4(SKB["barclose"][off]) fh, _, hh, dh = stats4(SKB["hourly"][off]) fc, _, hc, dc = stats4(SKB["canonical"][off]) rows_s.append((fb - fh, hb - hh, db - dh, fc - fb, hc - hb, dc - db)) S = np.array(rows_s) print(f" SKH standalone 50/50: barclose-hourly dShFULL off0 {S[0,0]:+.3f} " f"(med {np.median(S[:,0]):+.3f}), dShHOLD off0 {S[0,1]:+.3f} (med {np.median(S[:,1]):+.3f}), " f"dDD off0 {S[0,2]:+.1%} (med {np.median(S[:,2]):+.1%})") print(f" quota NON recuperabile da nessun cron (canonical-barclose = fill-al-livello): " f"dShFULL off0 {S[0,3]:+.3f} (med {np.median(S[:,3]):+.3f}), " f"dShHOLD off0 {S[0,4]:+.3f} (med {np.median(S[:,4]):+.3f}), " f"dDD off0 {S[0,5]:+.1%} (med {np.median(S[:,5]):+.1%})") print(" fattibilita' (SOLO STIMA, non cablata): 230 = 23x10 -> i confini 230m cadono sempre su") print(" multipli di 10 min dall'epoch => un cron `*/10 * * * *` che esegue solo quando") print(" (now % 230min)==0 colpisce ogni confine con ritardo ~0 (vs 0-50 min, media 25, del cron") print(" orario): 'barclose' e' esattamente il suo modello (al netto di runtime job e slippage).") print(" NB: la cadenza NON recupera il gap fill-al-livello (exit software in barre che gappano).") # ---- (7) caveat -------------------------------------------------------- print("\n--- (7) CAVEAT (onesta') ---") print("- exit software nei crash = GAP reale (audit: sl2% -> -11/-23% realizzato); 'hourly' e'") print(" la stima pessimistica giusta del path live; niente slippage/parziali modellati;") print("- equity daily-step (lens Sharpe), convenzione identica a canonico e audit r0702;") print("- selezione: w* su IS mediano-di-banda (mai su hold-out); 8 w provati, k_seen=8 nel gate;") print("- banda 23-offset = lente de-luckata; live gira UNA griglia sola (off 0);") print("- deriva-dati vs audit del 2026-07-02 (il feed cresce ogni giorno): riconciliazione 'circa';") print("- a $600 la gamba SKH a w=0.10 vale ~$60/trade (>> min-order $5): la granularita' non vincola.") print(f"\nFatto in {time.time()-t0:.0f}s.") if __name__ == "__main__": main()