#!/usr/bin/env python """r0822b_bin_freq.py — BIN-FREQ: il BINARIO di frequenza su SKH01 giudicato come un CANDIDATO. STORIA DEI FATTI (verificata leggendo il codice, non il riassunto). 1. `r0822_vol_size.py` ha proposto una SIZE PER-TRADE su SKH01 (cella congelata `VTL tv20 w90` = clip(0.20/rv90(t-1), 0, 3) fissata all'INGRESSO) come CONTROLLO DEL RISCHIO: +0,070 ISO dShFULL, 23/23 ancore, 8/8 anni. 2. `r0822_volsize_skeptic.py` ha CONFERMATO il numero e FALSIFICATO il meccanismo: il guadagno e' FREQUENZA (+0,051 sul solo filtro di attivita') e non MAGNITUDINE (+0,003); e la forma minima e' un BINARIO 0/1 (`BIN L>=med`, 194/406 trade scartati) che vale +0,080 FULL (20/23) = 115% del LEAD, con controllo di segno superato (l'inverso fa -0,159, 0/23). 3. Lo scettico ha scritto "il candidato giusto non e' la formula ma il binario" — e nessuno l'ha passato per `study_family_honest`. E' un SOTTOPRODOTTO DI UN ATTACCO, mai dichiarato come famiglia. Questo script lo giudica come si giudica un candidato. IPOTESI, REGISTRATA PRIMA DI MISURARE. Mi aspetto che il numero si riproduca (lo scettico e' stato rigoroso e la sua replica e' bit-exact) e che il candidato NON superi il gate onesto, per due ragioni dichiarate in anticipo: (a) la soglia e la finestra sono due assi liberi trovati DENTRO l'esplorazione di un'altra famiglia, quindi la stima di famiglia dovra' essere molto piu' piccola dell'argmax (precedente: sull'asse w lo scettico stesso stima +0,025 contro +0,080); (b) il deflated-Sharpe sara' VACUO come lo e' stato sulla famiglia VOL-SIZE — su perturbazioni della STESSA strategia la varianza fra i trial e' quasi nulla, e il DSR e' ~1 per chiunque, baseline compreso. Se succede lo dichiaro vacuo e uso un null che ha potenza. =========================== LA FAMIGLIA, DICHIARATA PRIMA DI GUARDARE ========================= Un binario di frequenza su SKH01 ha esattamente quattro gradi di liberta'. Li dichiaro tutti, con i loro valori, PRIMA di misurare, e li conto AL RIALZO: soglia (6) quantile ESPANDENTE causale della vol trailing, q in {0.25, 0.35, 0.50, 0.65, 0.75} + la soglia ASSOLUTA rv <= 20% (== `L>=1` della regola congelata) finestra (6) w in {45, 60, 90, 120, 180, 270} giorni (sotto 45 lo scettico ha misurato che la finestra contiene 1-3 barre ATTIVE: campione vuoto, non fragilita') isteresi (2) h in {0.00, 0.10} (banda +-10% attorno alla soglia, stato persistente causale) segno (2) +1 tieni la vol BASSA (l'ipotesi) / -1 tieni la vol ALTA (controllo di segno) timeframe(1) uno solo: la gamba SKH01 vive sulla griglia 230m per costruzione. Dichiarato. -> 6 x 6 x 2 x 2 = 144 celle RIAPRIRE UN PARAMETRO RIAPRE LA SUA FAMIGLIA, e questo binario e' nato dentro l'esplorazione di un'altra famiglia. Percio' al conto delle 144 si SOMMANO i trial gia' spesi sullo stesso oggetto: 28 (r0822_vol_size) + 43 (r0822_volsize_skeptic) = 71. -> CONTO ONESTO DICHIARATO: 144 + 71 = 215 trial. La sensibilita' del verdetto al conteggio e' pubblicata (§5), come nel precedente 30/07 dove un DSR passava a N=8, falliva a N=72 e falliva a N=360. ============================== LENTE, ANCORE, CONVENZIONI ===================================== Lente: path CANONICO di SKH01 (fill al livello, entry a chiusura di bin) — identica nelle due braccia di ogni confronto, quindi il Delta e' pulito anche se il livello e' pessimistico verso il path live. Ancore: **23 offset**, la griglia a priori dell'audit 02/07 (ogni 30m su [0,690)): NON ridotta, perche' il costo misurato e' ~20 ms per cella per ancora. Statistica: MEDIANA DELLE DIFFERENZE APPAIATE (`A.anchor_luck_delta`), mai differenza delle mediane. Confronto sempre ISO-PESO: il binario cambia la vol della gamba e quindi il peso EFFETTIVO di SKH01 nel libro, e muovere quel peso e' gia' stato misurato e RESPINTO il 26/07. Il null del de-levering e' DEGENERE su una ri-scalatura (`sh(k*base) === sh(base)`), quindi il test e' iso-peso sullo Sharpe; `k_iso_dd` resta come diagnostica del DD. Nessuna riscrittura: `extract / equity_daily / leg_daily / sizes_from_L / book / netted_target` sono IMPORTATE da r0822_vol_size, e il titolo dello scettico e' riprodotto chiamando la SUA funzione. Replica bit-exact contro `sleeves._skyhook_returns()` prima di ogni delta (§0). USO: nice -n 19 timeout 900 uv run python scripts/research/r0822b_bin_freq.py [--every K] 1 ancora ogni K delle 23 (K=8 -> 3 ancore, pilota ~80 s) [--ndraw N] estrazioni del null di permutazione (default 100) Con i default gira in ~400 s e ~500 MB di RSS su questa VPS (2 core). Tutti i numeri riportati qui vengono da questa invocazione, e due esecuzioni indipendenti hanno dato output numericamente IDENTICI. """ from __future__ import annotations import argparse import gc import sys import time from pathlib import Path import numpy as np import pandas as pd ROOT = Path(__file__).resolve().parents[2] for _p in (ROOT, ROOT / "scripts" / "research", ROOT / "scripts" / "research" / "alt"): sys.path.insert(0, str(_p)) import altlib as A # noqa: E402 import r0702_anchor_skh01 as R # noqa: E402 import r0702_tp01_offset as TO # noqa: E402 import r0822_vol_size as V # noqa: E402 (motore: riusato, non riscritto) import r0822_volsize_skeptic as SK # noqa: E402 (per riprodurre il titolo) from r0822_sol_leg import k_iso_dd # noqa: E402 from src.portfolio.portfolio import weights_tilt_null # noqa: E402 from src.strategies.skyhook import SKH01_V1 # noqa: E402 ASSETS = V.ASSETS OFFSETS_FULL = V.OFFSETS_FULL sh, vol, maxdd, cagr = V.sh, V.vol, V.maxdd, V.cagr hold, ins, book = V.hold, V.ins, V.book CAPITAL, MIN_ORDER = V.CAPITAL, V.MIN_ORDER W_TP, W_SKH = V.W_TP, V.W_SKH # ------------------------------------------------------------------ la famiglia (dichiarata sopra) QS = (0.25, 0.35, 0.50, 0.65, 0.75) THRS = tuple([("q", q) for q in QS] + [("abs", 0.20)]) WS = (45, 60, 90, 120, 180, 270) HS = (0.00, 0.10) SIGNS = (+1, -1) N_INHERITED = 28 + 43 # r0822_vol_size + r0822_volsize_skeptic, sullo stesso oggetto CANON = ("q", 0.50, 90, 0.00, +1) # la cella dello scettico, dentro la mia famiglia L = 112 def cname(thr, w, h, sg) -> str: t = "rv<=20%" if thr[0] == "abs" else f"q{thr[1]:.2f}" return f"{t} w{w} h{h:.2f} s{'+' if sg > 0 else '-'}" # =============================================================== il binario, causale per costruzione def _hyst(x: np.ndarray, thr: np.ndarray, h: float, keep_below: bool) -> np.ndarray: """Stato persistente con banda +-h attorno alla soglia. Causale: al passo t usa solo x_t e thr_t (entrambi funzione di dati <= t-1) e lo stato al passo t-1. h=0 -> confronto secco.""" n = len(x) out = np.ones(n, bool) st = True for i in range(n): xi, ti = x[i], thr[i] if not (np.isfinite(xi) and np.isfinite(ti)): st = True # warm-up: si tiene tutto (conv. scettico) elif keep_below: st = (xi <= ti * (1.0 + h)) if st else (xi <= ti * (1.0 - h)) else: st = (xi >= ti * (1.0 - h)) if st else (xi >= ti * (1.0 + h)) out[i] = st return out class KeepCache: """rv e quantili espandenti calcolati UNA volta per (asset, ancora, w). L'espandente e' la parte cara; il resto e' un confronto.""" def __init__(self) -> None: self.rv: dict = {} self.q: dict = {} def clear(self) -> None: self.rv.clear() self.q.clear() def keep(self, ex: dict, key, thr, w: int, h: float, sg: int) -> pd.Series: base = ex["base_daily"] k = (key, w) if k not in self.rv: self.rv[k] = V._rv_daily(base, w, active_only=False) # gia' shiftata di 1 giorno rv = self.rv[k] if thr[0] == "abs": t = np.full(len(rv), float(thr[1])) else: kq = (key, w, thr[1]) if kq not in self.q: self.q[kq] = rv.expanding(min_periods=V.VT_WARM).quantile(thr[1]) t = self.q[kq].values keep = _hyst(rv.values.astype(float), np.asarray(t, float), h, keep_below=(sg > 0)) return pd.Series(keep.astype(float), index=rv.index) KC = KeepCache() def rss_mb() -> float: """RSS in MB, per non far morire il lavoro degli altri agenti (brief §8: 2 core, ~3 GB).""" try: with open("/proc/self/status") as f: for ln in f: if ln.startswith("VmRSS:"): return float(ln.split()[1]) / 1024.0 except Exception: pass return float("nan") def size_bin(thr, w, h, sg): def f(ex, key): return V.sizes_from_L(ex, KC.keep(ex, key, thr, w, h, sg)) return f # =============================================================== motore def leg_of(EXX: dict, key, fn) -> pd.Series: return V.leg_daily(EXX, {a: fn(EXX[a], (key, a)) for a in ASSETS}) def iso_pair(TP: pd.Series, leg: pd.Series, leg_base: pd.Series): """Libro della variante e libro di CONTROLLO ISO-PESO (baseline ri-scalato alla stessa vol di gamba => stesso peso effettivo di SKH). Convenzione del bersaglio, riusata tale e quale.""" vb = vol(leg_base) m = vol(leg) / vb if vb > 0 else 1.0 return book(TP, leg), book(TP, leg_base * m), m # =============================================================== main def main() -> None: ap = argparse.ArgumentParser() ap.add_argument("--every", type=int, default=1) ap.add_argument("--ndraw", type=int, default=100) args = ap.parse_args() offs = OFFSETS_FULL[:: max(1, args.every)] if len(offs) < 3: raise SystemExit(f"--every {args.every} lascia {len(offs)} ancore: " "`A.anchor_luck_delta` ne vuole almeno 3 (usa --every <= 8).") t0 = time.time() print("=" * L) print(" r0822b BIN-FREQ — il BINARIO di frequenza su SKH01, giudicato come un CANDIDATO") print("=" * L) print(f" ancore: {len(offs)}/{len(OFFSETS_FULL)} della griglia a priori 02/07 (ogni 30m su " f"[0,690)). NON ridotta.") print(" lente: path CANONICO, identica nelle due braccia. Confronto sempre ISO-PESO.") # ---------------------------------------------------------- §1 famiglia, PRIMA di misurare print("\n" + "-" * L) print(" 1. GRIGLIA DICHIARATA — scritta prima di guardare i risultati, contata AL RIALZO") print("-" * L) cells = [(t, w, h, s) for t in THRS for w in WS for h in HS for s in SIGNS] print(f" soglia {len(THRS):>3} {['q%.2f' % q for q in QS] + ['rv<=20% (== L>=1)']}") print(f" finestra {len(WS):>3} {list(WS)} giorni") print(f" isteresi {len(HS):>3} {list(HS)} (banda +-h attorno alla soglia)") print(f" segno {len(SIGNS):>3} +1 tieni vol BASSA (ipotesi) / -1 tieni vol ALTA (controllo)") print(f" timeframe 1 la gamba SKH01 vive su 230m per costruzione") print(f" -> celle di QUESTA famiglia: {len(cells)}") print(f" -> trial gia' spesi sullo STESSO oggetto: {N_INHERITED} " f"(28 r0822_vol_size + 43 r0822_volsize_skeptic)") print(f" -> CONTO ONESTO DICHIARATO: {len(cells)} + {N_INHERITED} = " f"{len(cells) + N_INHERITED} trial. La sensibilita' al conteggio e' in §5.") assert len(cells) == 144 # ---------------------------------------------------------- estrazione print("\n estrazione tabelle trade (V2-DD) ...", flush=True) EX = {o: {a: V.extract(a, o) for a in ASSETS} for o in offs} ntr = {a: int(np.mean([len(EX[o][a]["i_ent"]) for o in offs])) for a in ASSETS} R.get5m.cache_clear() # ~90 MB di 5m non servono durante la §4 (la VPS ha 2 core gc.collect() # e ~3 GB liberi condivisi); si ricaricano in §9b in ~3 s print(f" fatto in {time.time()-t0:.0f}s — trade medi/ancora: " + ", ".join(f"{a} {ntr[a]}" for a in ASSETS) + f" (RSS {rss_mb():.0f} MB)") TP = A.tp01_baseline_daily() can = offs[0] # ---------------------------------------------------------- §0 SANITY print("\n" + "-" * L) print(" 0. SANITY — bit-exact prima di qualunque delta") print("-" * L) from src.portfolio.sleeves import _skyhook_returns base_leg = {o: V.leg_daily(EX[o], {a: V.size_flat(EX[o][a]) for a in ASSETS}) for o in offs} off = _skyhook_returns() d0 = float(np.max(np.abs(base_leg[can].values - off.values))) if len(base_leg[can]) == len(off) \ else float("nan") print(f" gamba 50/50 off{can} vs sleeves._skyhook_returns(): max|dif| = {d0:.2e} " f"({len(base_leg[can])} giorni)") assert d0 < 1e-15, "replica NON bit-exact" for a in ASSETS: ref = R.run_asset(a, can)[0] R._CACHE.clear() dd = float(np.max(np.abs(V.equity_daily(EX[can][a], V.size_flat(EX[can][a])).values - ref.values))) print(f" {a} vs r0702_anchor_skh01.run_asset: max|dif| = {dd:.2e}") assert dd < 1e-15 # ---------------------------------------------------------- §2 riproduzione del titolo print("\n" + "-" * L) print(" 2. RIPRODUZIONE DEL TITOLO DELLO SCETTICO — chiamando la SUA funzione, non la mia") print(" (pubblicato: BIN L>=med +0,080 FULL 20/23 ; INVERSO -0,159 0/23)") print("-" * L) REP = {} for tag, mode in (("BIN L>=med (scettico)", "gemed"), ("BIN L<1 INVERSO (scettico)", "lt1"), ("BIN L>=1 (scettico)", "ge1")): fn = SK.size_bin(SK.FROZEN, mode) bk, ct = {}, {} for o in offs: lg = V.leg_daily(EX[o], {a: fn(EX[o][a]) for a in ASSETS}) bk[o], ct[o], _ = iso_pair(TP, lg, base_leg[o]) REP[tag] = (bk, ct) dF = A.anchor_luck_delta(lambda o: bk[o], lambda o: ct[o], offs, metric=sh) dH = A.anchor_luck_delta(lambda o: hold(bk[o]), lambda o: hold(ct[o]), offs, metric=sh) print(f" {tag:<28} ISO dShFULL {dF['median_paired']:+.3f} " f"({dF['n_positive']}/{len(offs)}) banda [{dF['lo']:+.3f},{dF['hi']:+.3f}] " f"ISO dShHOLD {dH['median_paired']:+.3f} ({dH['n_positive']}/{len(offs)})") med_F = float(np.median([sh(book(TP, base_leg[o])) for o in offs])) med_H = float(np.median([sh(hold(book(TP, base_leg[o]))) for o in offs])) med_D = float(np.median([maxdd(book(TP, base_leg[o])) for o in offs])) ban_F = [sh(book(TP, base_leg[o])) for o in offs] print(f" libro BASE, mediana della banda: ShFULL {med_F:.3f} / ShHOLD {med_H:.3f} / " f"maxDD {med_D*100:.1f}% [pubblicato 1,626 / 1,037 / 10,4%]") print(f" banda d'ancora del LIVELLO del libro: ShFULL [{min(ban_F):.3f},{max(ban_F):.3f}] " f"= ampiezza {max(ban_F)-min(ban_F):.3f} (il termine di paragone di §12)") # ---------------------------------------------------------- §3 causalita' per troncamento print("\n" + "-" * L) print(" 3. CAUSALITA' — la decisione tenere/scartare ricalcolata TRONCANDO i dati al giorno") print(" d'ingresso. `A.causality_ok` NON E' GIRATO (vuole un target_fn(df) e qui il") print(" segnale nasce da 5m->230m: ri-eseguirlo su prefissi costa quanto tutto lo") print(" script). Sostituito da un troncamento diretto, piu' stringente sul punto esatto.") print("-" * L) thr_c, w_c, h_c, s_c = ("q", CANON[1]), CANON[2], CANON[3], CANON[4] bad = tested = 0 for a in ASSETS: ex = EX[can][a] kfull = KC.keep(ex, (can, a), thr_c, w_c, h_c, s_c) rv_full = V._rv_daily(ex["base_daily"], w_c, False) days = list(dict.fromkeys(ex["day_ent"])) for d in days[::7]: # 1 ingresso su 7, su tutta la storia if d not in kfull.index: continue b_tr = ex["base_daily"][ex["base_daily"].index <= d] if len(b_tr) < V.VT_WARM + w_c: continue rv_t = V._rv_daily(b_tr, w_c, False) q_t = rv_t.expanding(min_periods=V.VT_WARM).quantile(thr_c[1]) k_t = _hyst(rv_t.values.astype(float), q_t.values.astype(float), h_c, s_c > 0)[-1] tested += 1 if bool(k_t) != bool(kfull.loc[d] > 0.5): bad += 1 assert abs(float(rv_t.iloc[-1]) - float(rv_full.loc[d])) < 1e-12 or \ not np.isfinite(rv_full.loc[d]) print(f" decisioni ricalcolate su prefisso: {tested} DIVERGENTI: {bad}") print(" NB l'isteresi ha memoria: troncando si perde lo stato, quindi il test e' esatto") print(" per h=0 (la cella canonica) e conservativo per h>0.") assert bad == 0, "il binario NON e' causale" # ---------------------------------------------------------- §4 tabella di famiglia print("\n" + "-" * L) print(" 4. LA FAMIGLIA INTERA — ISO dShFULL, mediana delle differenze appaiate su " f"{len(offs)} ancore") print("-" * L) # RAM (brief §8: 7 GB totali, ~3 liberi, e siamo in tanti). Tre accorgimenti, tutti # necessari — la prima stesura di questo script e' stata UCCISA a meta' della §4: # (a) DUE PASSATE: la 1a calcola le metriche per-ancora di tutte le 144 celle SENZA # conservare le serie (144 x 23 x 3 serie da 2717 punti = ~500 MB); la 2a ricalcola e # conserva le serie delle sole 3-4 celle che servono dopo (~2 s); # (b) ciclo ANCORA-ESTERNO con la cache di rv/quantili SVUOTATA a ogni ancora: viva una sola # ancora alla volta (72 serie invece di 1.656); # (c) RSS stampata, cosi' se qualcuno la rilancia sa cosa aspettarsi. BK, CT, LEG, MS = {}, {}, {}, {} ACC: dict = {} def _one(o, thr, w, h, sg, store: bool): """UNA cella a UNA ancora. Le size si calcolano una volta sola e si riusano (la prima stesura le ricalcolava per il conteggio dei tenuti); la scomposizione per anno e il maxDD si fanno SOLO nella seconda passata, perche' servono a 3 celle e costavano il 60% del tempo su tutte e 144.""" nm = cname(thr, w, h, sg) fn = size_bin(thr, w, h, sg) szs = {a: fn(EX[o][a], (o, a)) for a in ASSETS} lg = V.leg_daily(EX[o], szs) b, c, m = iso_pair(TP, lg, base_leg[o]) d = ACC.setdefault(nm, dict(thr=thr, w=w, h=h, sg=sg, dF=[], dH=[], dI=[], dC=[], lF=[], lI=[], lH=[], ms=[], kp=[], ddv=[], ddc=[], yr={})) d["dF"].append(sh(b) - sh(c)); d["dH"].append(sh(hold(b)) - sh(hold(c))) d["dI"].append(sh(ins(b)) - sh(ins(c))) d["lF"].append(sh(lg)); d["lI"].append(sh(ins(lg))); d["lH"].append(sh(hold(lg))) d["ms"].append(m) d["kp"].append(float(np.mean(np.concatenate(list(szs.values()))))) if store: d["dC"].append(cagr(b) - cagr(c)) d["ddv"].append(maxdd(b)); d["ddc"].append(maxdd(c)) for y in {int(t) for t in b.index.year}: by, cy = b[b.index.year == y], c[c.index.year == y] if len(by) > 20: d["yr"].setdefault(y, []).append(sh(by) - sh(cy)) BK.setdefault(nm, {})[o] = b CT.setdefault(nm, {})[o] = c LEG.setdefault(nm, {})[o] = lg MS.setdefault(nm, {})[o] = m for o in offs: KC.clear() for (thr, w, h, sg) in cells: _one(o, thr, w, h, sg, store=False) KC.clear() def _med(v, dflt=float("nan")): return float(np.median(v)) if len(v) else dflt def _fin(d: dict) -> dict: a_ = np.asarray(d["dF"], float) return dict(nm=cname(d["thr"], d["w"], d["h"], d["sg"]), thr=d["thr"], w=d["w"], h=d["h"], sg=d["sg"], dF=float(np.median(a_)), nF=int((a_ > 0).sum()), lo=float(a_.min()), hi=float(a_.max()), arrF=a_, dH=float(np.median(d["dH"])), nH=int((np.asarray(d["dH"]) > 0).sum()), isoI=float(np.median(d["dI"])), dC=_med(d["dC"]), legF=float(np.median(d["lF"])), legI=float(np.median(d["lI"])), legH=float(np.median(d["lH"])), kept=float(np.median(d["kp"])), weff=float(np.median([W_SKH * x for x in d["ms"]])), ddv=_med(d["ddv"]), ddc=_med(d["ddc"]), yr={y: float(np.median(v)) for y, v in d["yr"].items()}) RES = {nm: _fin(d) for nm, d in ACC.items()} print(f" ({len(cells)} celle x {len(offs)} ancore valutate in {time.time()-t0:.0f}s " f"dall'avvio, prima passata senza conservare le serie; RSS {rss_mb():.0f} MB)") def matrix(sg: int, h: float) -> None: print(f"\n ISO dShFULL — segno {'+1 (tieni vol BASSA)' if sg>0 else '-1 (CONTROLLO)'}" f", isteresi h={h:.2f}") print(" soglia " + "".join(f"{('w%d' % w):>11}" for w in WS)) for thr in THRS: t = "rv<=20%" if thr[0] == "abs" else f"q{thr[1]:.2f}" row = "" for w in WS: r = RES[cname(thr, w, h, sg)] row += f"{r['dF']:>+8.3f}{r['nF']:>3}" print(f" {t:<10}" + row) for sg in SIGNS: for h in HS: matrix(sg, h) print("\n (ogni cella: mediana appaiata e n. di ancore positive su " f"{len(offs)})") POS = [RES[cname(t, w, h, +1)] for t in THRS for w in WS for h in HS] NEG = [RES[cname(t, w, h, -1)] for t in THRS for w in WS for h in HS] ALLC = [RES[cname(t, w, h, s)] for t in THRS for w in WS for h in HS for s in SIGNS] nmcan = cname(("q", CANON[1]), CANON[2], CANON[3], CANON[4]) argmax = max(ALLC, key=lambda r: r["dF"]) print(f"\n ARGMAX della famiglia: {cname(argmax['thr'], argmax['w'], argmax['h'], argmax['sg']):<26}" f" dShFULL {argmax['dF']:+.3f} ({argmax['nF']}/{len(offs)})") print(f" cella dello scettico: {nmcan:<26} dShFULL {RES[nmcan]['dF']:+.3f} " f"({RES[nmcan]['nF']}/{len(offs)}) rango " f"{1 + sum(r['dF'] > RES[nmcan]['dF'] for r in ALLC)}/{len(ALLC)}") print(f" STIMA ONESTA se la cella si sceglie a caso dentro la famiglia (mediana):") print(f" tutta la famiglia (144) {float(np.median([r['dF'] for r in ALLC])):+.3f} " f"positive {sum(r['dF']>0 for r in ALLC)}/{len(ALLC)}") print(f" solo segno +1 (72) {float(np.median([r['dF'] for r in POS])):+.3f} " f"positive {sum(r['dF']>0 for r in POS)}/{len(POS)}") print(f" solo segno -1 (72) {float(np.median([r['dF'] for r in NEG])):+.3f} " f"positive {sum(r['dF']>0 for r in NEG)}/{len(NEG)} <- controllo di segno") # seconda passata: conserva le serie delle sole celle che servono dopo sel_iso0 = max(ALLC, key=lambda r: r["isoI"]) ACC2 = ACC for r in dict((x["nm"], x) for x in (RES[nmcan], argmax, sel_iso0, max(ALLC, key=lambda z: z["legI"]))).values(): ACC = {} for o in offs: _one(o, r["thr"], r["w"], r["h"], r["sg"], store=True) KC.clear() RES[r["nm"]] = _fin(ACC[r["nm"]]) # ora con dC / maxDD / per-anno ACC = ACC2 POS = [RES[cname(t, w, h, +1)] for t in THRS for w in WS for h in HS] NEG = [RES[cname(t, w, h, -1)] for t in THRS for w in WS for h in HS] ALLC = POS + NEG argmax = RES[argmax["nm"]] dchk = A.anchor_luck_delta(lambda o: BK[nmcan][o], lambda o: CT[nmcan][o], offs, metric=sh) assert abs(dchk["median_paired"] - RES[nmcan]["dF"]) < 1e-12 and \ dchk["n_positive"] == RES[nmcan]["nF"], "la statistica appaiata di §4 non coincide con A.anchor_luck_delta" print(f" controllo: la statistica appaiata calcolata in §4 coincide con " f"`A.anchor_luck_delta` ({dchk['median_paired']:+.6f})") print(f" quota di trade TENUTI, mia cella q0.50 w90: {RES[nmcan]['kept']*100:.1f}% " f"(mediana fra le ancore)") fn_can = size_bin(("q", CANON[1]), CANON[2], CANON[3], CANON[4]) fn_sk = SK.size_bin(SK.FROZEN, "gemed") ov, k_me, k_sk = [], [], [] for o in offs: for a in ASSETS: x = fn_can(EX[o][a], (o, a)) > 0.5 y = fn_sk(EX[o][a]) > 0.5 ov.append(float((x == y).mean())); k_me.append(float(x.mean())); k_sk.append(float(y.mean())) dleg = float(np.max(np.abs(BK[nmcan][can].values - REP["BIN L>=med (scettico)"][0][can].values))) print(f" ⚠ la MIA parametrizzazione NON e' bit-identica a quella dello scettico: lui prende") print(f" la mediana espandente di L = clip(0.20/rv,0,3) DOPO un `.fillna(1.0)` che inietta") print(f" dei 1.0 nel warm-up; io prendo il quantile espandente di rv, che salta i NaN. Il") print(f" trasformo e' monotono, quindi le due decisioni coincidono nel " f"{float(np.mean(ov))*100:.1f}% dei trade") print(f" (tenuti {float(np.mean(k_me))*100:.1f}% io / {float(np.mean(k_sk))*100:.1f}% lui); " f"max|dif| fra i due libri = {dleg:.2e}.") # ---------------------------------------------------------- §5 i tre componenti del gate print("\n" + "-" * L) print(" 5. `study_family_honest` — NON chiamabile direttamente (e' cablato sui candidati") print(" DIREZIONALI: factory -> target_fn via candidate_daily, e qui il candidato e' un") print(" overlay su uno sleeve). Come il 30/07 sul tenore VRP01, uso i suoi TRE") print(" COMPONENTI REALI: selezione in-sample-only -> deflated_sharpe -> marginal_vs_tp01.") print("-" * L) print(" (a) SELEZIONE IN-SAMPLE-ONLY — due criteri, entrambi dichiarati, entrambi sulla") print(" MEDIANA fra le ancore (l'ancora non e' un hold-out, e' una dimensione di luck)") sel_leg = max(ALLC, key=lambda r: r["legI"]) # come `select_cell_insample` sel_iso = max(ALLC, key=lambda r: r["isoI"]) # cio' su cui si decide davvero hop_leg = max(ALLC, key=lambda r: r["legH"]) hop_iso = max(ALLC, key=lambda r: r["dH"]) for tag, r in (("Sharpe IS standalone della gamba (come select_cell_insample)", sel_leg), ("ISO dShFULL IN-SAMPLE del libro (cio' che decide)", sel_iso)): n = cname(r["thr"], r["w"], r["h"], r["sg"]) print(f" scelta al buio su {tag}") print(f" -> {n:<26} ISO dShFULL {r['dF']:+.3f} ({r['nF']}/{len(offs)}) " f"ISO dShHOLD {r['dH']:+.3f} ({r['nH']}/{len(offs)}) banda " f"[{r['lo']:+.3f},{r['hi']:+.3f}]") for tag, r in (("Sharpe HOLD-OUT standalone", hop_leg), ("ISO dShHOLD del libro", hop_iso)): print(f" controllo (scegliendo SULL'HOLD-OUT, cio' che NON si fa) su {tag}:") print(f" -> {cname(r['thr'], r['w'], r['h'], r['sg']):<26} dShFULL {r['dF']:+.3f}" f" dShHOLD {r['dH']:+.3f}") CHOSEN = sel_iso nmc = cname(CHOSEN["thr"], CHOSEN["w"], CHOSEN["h"], CHOSEN["sg"]) print(f" CELLA SCELTA AL BUIO (criterio primario = ISO dShFULL in-sample): {nmc}") print("\n (b) DEFLATED SHARPE — sullo Sharpe FULL standalone della gamba (la grandezza che") print(" `select_cell_insample` usa), mediana fra le ancore; `daily` della barra canonica") print(" serve solo per T/asimmetria/curtosi. Sensibilita' al CONTEGGIO ottenuta") print(" REPLICANDO la distribuzione dei trial osservati (tiling), non riempiendola con") print(" la mediana: riempire con la mediana ABBASSA la varianza dei trial e fa SALIRE il") print(" DSR all'aumentare di N, che e' il verso sbagliato. (Errore mio, catturato nel") print(" pilota e corretto: il primo output dava DSR 0,966 -> 0,986 -> 0,998 da N=144 a") print(" N=500, cioe' 'piu' trial = piu' significativo'.)") all_sr = [r["legF"] for r in ALLC] pos_sr = [r["legF"] for r in POS] base_sr = float(np.median([sh(base_leg[o]) for o in offs])) dly = LEG[nmc][can] def tile(v, N): return (list(v) * (N // len(v) + 1))[:N] for lab, sr, d in (("cella scelta al buio", CHOSEN["legF"], dly), ("BASELINE (size fissa)", base_sr, base_leg[can])): for N, tagN in ((len(ALLC), f"{len(ALLC)} di QUESTA famiglia"), (len(ALLC) + N_INHERITED, f"{len(ALLC)+N_INHERITED} = conto onesto (+{N_INHERITED} ereditati)"), (500, "500 (ipotetico, al rialzo)")): dsr, sr0 = A.deflated_sharpe(sr, tile(all_sr, N), d) print(f" {lab:<22} N={tagN:<44} DSR {dsr:.3f} max atteso sotto il null {sr0:+.3f}") lo_n, hi_n = len(ALLC), 20000 if A.deflated_sharpe(CHOSEN["legF"], tile(all_sr, hi_n), dly)[0] >= 0.95: cross = f">{hi_n}" elif A.deflated_sharpe(CHOSEN["legF"], tile(all_sr, lo_n), dly)[0] < 0.95: cross = f"<{lo_n}" else: for _ in range(40): mid = (lo_n + hi_n) // 2 if A.deflated_sharpe(CHOSEN["legF"], tile(all_sr, mid), dly)[0] >= 0.95: lo_n = mid else: hi_n = mid cross = f"~{lo_n}" print(f" -> IL VERDETTO SI RIBALTA A N {cross} trial: sotto quel conteggio il DSR passa,") print(f" sopra fallisce. Il conto onesto dichiarato ({len(ALLC)+N_INHERITED}) sta dal lato") print(f" del PASS. Precedente 30/07: un DSR passava a N=8, falliva a N=72 e a N=360; qui") print(f" la stessa sensibilita' e' pubblicata invece di essere taciuta.") dsp, sr0p = A.deflated_sharpe(CHOSEN["legF"], tile(pos_sr, len(POS) + N_INHERITED), dly) print(f" sensibilita' alla COMPOSIZIONE della famiglia: escludendo il segno -1 (che e' un") print(f" controllo e non un candidato) i trial diventano {len(POS)}+{N_INHERITED} e la loro") print(f" dispersione cala -> DSR {dsp:.3f} (max atteso {sr0p:+.3f}). Includere il segno -1") print(f" e' la scelta CONSERVATIVA e resta quella dichiarata.") print(f" dispersione dei trial: sd(Sharpe FULL) = {float(np.std(all_sr, ddof=1)):.3f} sulle") print(f" {len(ALLC)} celle, {float(np.std(pos_sr, ddof=1)):.3f} sulle {len(POS)} di segno +1.") print(f" -> il DSR qui NON e' vacuo (il baseline lo FALLISCE dove il candidato lo passa):") print(f" la famiglia contiene celle davvero diverse fra loro, al contrario della famiglia") print(f" VOL-SIZE dove erano tutte perturbazioni della stessa size continua.") print(f" ❌ QUESTO REFUTA LA MIA IPOTESI (b), registrata nel docstring PRIMA di misurare") print(f" ('il DSR sara' vacuo come su VOL-SIZE'). La ragione per cui mi sbagliavo e'") print(f" strutturale e vale come regola: un DSR e' vacuo quando i trial si somigliano") print(f" (var_tr -> 0 => sr0 -> 0 => DSR ~ 1 per chiunque). Una famiglia BINARIA contiene") print(f" celle che scartano insiemi di trade MOLTO diversi — compreso il segno inverso —") print(f" quindi la dispersione dei trial e' reale e il test riacquista potenza. Non e' il") print(f" binario a essere piu' solido della formula: e' la FAMIGLIA a essere piu' larga.") print("\n (c) MARGINAL vs TP01 — sulla gamba modificata e sulla gamba base") for lab, s in (("BASE (size fissa)", base_leg[can]), (nmc, LEG[nmc][can]), ("cella dello scettico", LEG[nmcan][can])): m = A.marginal_vs_tp01(s) b25 = m.get("blends", {}).get("w25", {}) print(f" {lab:<26} {str(m.get('marginal_verdict')):<10} corr {m.get('corr_full')} " f"uplift w25 full {b25.get('uplift_full')} / hold {b25.get('uplift_hold')} " f"robust_oos={m.get('robust_oos')} insample_edge={m.get('has_insample_edge')} " f"hedge={m.get('is_hedge')}") imp = A.implausible_sharpe( LEG[nmc][can], n_trades=int(sum(len(EX[can][a]["net"]) for a in ASSETS)), n_losing_trades=int(sum((EX[can][a]["net"] < 0).sum() for a in ASSETS))) print(f" implausible_sharpe (gamba scelta): implausible={imp['implausible']} " f"attive {imp.get('active_frac', 0)*100:.1f}% perdite/attive " f"{imp.get('loss_frac', 0)*100:.1f}% Calmar {imp.get('calmar', 0):.1f}") # ---------------------------------------------------------- §6 null con potenza print("\n" + "-" * L) print(" 6. IL NULL CHE HA POTENZA — permutazione della SEQUENZA tenuti/scartati.") print(" Stessa quota di trade tenuti, riassegnata a caso: separa 'conta QUALE trade si") print(" scarta' (informazione) da 'conta scartarne meta'' (aritmetica). Una mediana del") print(" null lontana da 0 direbbe che il null non e' un null (lezione 22/08 sul null") print(f" invertito). {args.ndraw} estrazioni su TUTTE le {len(offs)} ancore, non su 3.") print("-" * L) rng = np.random.default_rng(20260822) for lab, r in (("cella scelta al buio", CHOSEN), ("cella dello scettico", RES[nmcan]), ("argmax di famiglia", argmax)): nm = cname(r["thr"], r["w"], r["h"], r["sg"]) real = float(np.median([sh(BK[nm][o]) - sh(CT[nm][o]) for o in offs])) fn = size_bin(r["thr"], r["w"], r["h"], r["sg"]) sz0 = {o: {a: fn(EX[o][a], (o, a)) for a in ASSETS} for o in offs} nul = [] for _ in range(args.ndraw): v = [] for o in offs: szp = {a: rng.permutation(sz0[o][a]) for a in ASSETS} lg = V.leg_daily(EX[o], szp) b, c, _ = iso_pair(TP, lg, base_leg[o]) v.append(sh(b) - sh(c)) nul.append(float(np.median(v))) nul = np.asarray(nul) pc = float((nul < real).mean()) print(f" {lab:<24} {nm:<24} reale {real:+.3f} null mediana {np.median(nul):+.3f} " f"p90 {np.percentile(nul, 90):+.3f} max {nul.max():+.3f} -> p ~ {1-pc:.3f}") print(f" (seme 20260822; {args.ndraw} estrazioni x {len(offs)} ancore)") print("\n ⚠ LA MEDIANA DEL NULL NON E' ~0, E NON E' UN DIFETTO DEL NULL: E' UN FATTO.") print(" Scartare a caso meta' dei trade DANNEGGIA, perche' lo Sharpe ANNUALIZZATO cresce") print(" con la radice del numero di scommesse per anno: dimezzare i trade sullo stesso") print(" arco di calendario costa un fattore ~sqrt(2) di Sharpe di gamba. Misurato:") nmv = cname(CHOSEN["thr"], CHOSEN["w"], CHOSEN["h"], CHOSEN["sg"]) fnv = size_bin(CHOSEN["thr"], CHOSEN["w"], CHOSEN["h"], CHOSEN["sg"]) rng2 = np.random.default_rng(20260823) shb = float(np.median([sh(base_leg[o]) for o in offs])) shv = float(np.median([sh(LEG[nmv][o]) for o in offs])) kp = float(np.median([np.mean(np.concatenate([fnv(EX[o][a], (o, a)) for a in ASSETS])) for o in offs])) shn = [] for _ in range(min(40, args.ndraw)): sz0 = {o: {a: fnv(EX[o][a], (o, a)) for a in ASSETS} for o in offs} shn.append(float(np.median([sh(V.leg_daily(EX[o], {a: rng2.permutation(sz0[o][a]) for a in ASSETS})) for o in offs]))) print(f" Sharpe della GAMBA: baseline (tutti i trade) {shb:.3f} -> permutata " f"{float(np.median(shn)):.3f} -> filtrata {shv:.3f}") print(f" quota tenuta {kp*100:.0f}% -> attesa dalla sola perdita di scommesse " f"{shb*np.sqrt(kp):.3f}; osservata sotto permutazione {float(np.median(shn)):.3f}") print(" Conseguenza per la lettura: il null NON e' centrato su zero, quindi il p-value") print(" risponde alla domanda 'la SELEZIONE batte una selezione casuale della stessa") print(" taglia?' e non alla domanda 'il filtro batte il non filtrare?'. La seconda e' il") print(" confronto con la BASE (§4), che e' quello su cui si decide. Il filtro deve") print(" superare una penale meccanica prima di guadagnare qualcosa: e' un test SEVERO.") # ---------------------------------------------------------- §7 iso-peso / de-levering print("\n" + "-" * L) print(" 7. ISO-PESO e NULL DEL DE-LEVERING") print("-" * L) print(f" {'cella':<26}{'wSKH eff':>10}{'dShFULL RAW':>13}{'ISO':>9}{'maxDD var':>11}" f"{'maxDD ctrl':>11}{'k_iso_dd':>10}") for lab, r in (("BASE", None), ("cella dello scettico", RES[nmcan]), ("cella scelta al buio", CHOSEN)): if r is None: print(f" {'BASE (size fissa)':<26}{W_SKH:>10.3f}{0.0:>+13.3f}{0.0:>+9.3f}" f"{maxdd(book(TP, base_leg[can]))*100:>10.1f}%{'-':>11}{'-':>10}") continue nm = cname(r["thr"], r["w"], r["h"], r["sg"]) raw = A.anchor_luck_delta(lambda o, n=nm: BK[n][o], lambda o: book(TP, base_leg[o]), offs, metric=sh) dv = float(np.median([maxdd(BK[nm][o]) for o in offs])) dc = float(np.median([maxdd(CT[nm][o]) for o in offs])) ks = (f"{k_iso_dd(CT[nm][can], BK[nm][can]):.3f}" if dv < dc else "n/a") # n/a = non riduce il DD: nulla da negare print(f" {nm:<26}{r['weff']:>10.3f}{raw['median_paired']:>+13.3f}{r['dF']:>+9.3f}" f"{dv*100:>10.1f}%{dc*100:>10.1f}%{ks:>10}") print(" Il null del de-levering e' DEGENERE su una ri-scalatura (sh(k*base) === sh(base)):") print(" qui il test e' l'iso-peso sullo Sharpe. `k_iso_dd` e' riportato come diagnostica del") print(" DD ed e' calcolato CONTRO il controllo iso-peso (non contro il baseline nudo).") print(" NB il binario ABBASSA il peso effettivo di SKH01, non lo alza: la preoccupazione del") print(" 26/07 ('alzare SKH01') non morde, ma l'iso-peso resta obbligatorio nell'altro verso.") print(f"\n LIVELLI del libro (mediana fra le {len(offs)} ancore), netti fee:") print(f" {'configurazione':<30}{'ShFULL':>9}{'ShHOLD':>9}{'ShIS':>8}{'maxDD':>9}{'CAGR':>9}" f"{'vol':>8}") lev = [("BASE (libro live 75/25)", {o: book(TP, base_leg[o]) for o in offs})] for lab, r in (("BIN, peso eff. lasciato cadere", RES[nmcan]), ("BIN, controllo ISO-PESO", None)): if r is None: lev.append((lab, {o: CT[nmcan][o] for o in offs})) else: lev.append((lab, {o: BK[nmcan][o] for o in offs})) for lab, dd_ in lev: f_ = float(np.median([sh(dd_[o]) for o in offs])) h_ = float(np.median([sh(hold(dd_[o])) for o in offs])) i_ = float(np.median([sh(ins(dd_[o])) for o in offs])) d_ = float(np.median([maxdd(dd_[o]) for o in offs])) c_ = float(np.median([cagr(dd_[o]) for o in offs])) v_ = float(np.median([vol(dd_[o]) for o in offs])) print(f" {lab:<30}{f_:>9.3f}{h_:>9.3f}{i_:>8.3f}{d_*100:>8.1f}%{c_*100:>8.1f}%" f"{v_*100:>7.1f}%") print(" ⚠ la DIFFERENZA di queste mediane NON e' la mediana delle differenze appaiate (§4):") print(" qui le mediane cadono su ancore diverse. La tabella serve a dare i LIVELLI netti fee") print(" richiesti dal contratto; il numero che decide resta quello appaiato.") # ---------------------------------------------------------- §8 weights_tilt_null print("\n" + "-" * L) print(" 8. `weights_tilt_null` — una modulazione binaria della size E' un cambio di pesi nel") print(" tempo. Il gate confronta vettori STATICI: cattura la sola componente di SCALA") print(" (limite dichiarato, come nel bersaglio). k_seen = tutti i trial dichiarati.") print("-" * L) cols = {"TP01": TP, "SKH01": base_leg[can]} for lab, r in (("cella dello scettico", RES[nmcan]), ("cella scelta al buio", CHOSEN), ("argmax di famiglia", argmax)): nm = cname(r["thr"], r["w"], r["h"], r["sg"]) m = MS[nm][can] wp = {"TP01": W_TP, "SKH01": W_SKH * m} g = weights_tilt_null(cols, {"TP01": W_TP, "SKH01": W_SKH}, wp, caps={"SKH01": 0.5}, floor=0.05, n=300, k_seen=len(ALLC) + N_INHERITED) print(f" {lab:<24} {nm:<24} peso equiv SKH {wp['SKH01']/sum(wp.values()):.3f} " f"d_IS {g['delta_insample']:+.4f} d_HOLD {g['delta_hold']:+.4f} " f"pctl {g['pctl_hold']:.1f} gate_pass={g['gate_pass']}") print(" ⚠ LIMITE, e va detto perche' e' aggirabile: il binario ABBASSA il peso effettivo di") print(" SKH01, e il gate misura proprio quella caduta. Nella forma di deploy naturale (si") print(" ri-scala la gamba per riportare il peso effettivo a 0,25) il vettore proposto e'") print(" IDENTICO a quello corrente -> d_IS = d_HOLD = 0 e il gate diventa INERTE, non") print(" superato. Quindi `weights_tilt_null` NON puo' essere l'evidenza decisiva qui: e'") print(" negativo nella forma nuda e vacuo nella forma compensata. Cio' che decide resta la") print(" banda APPAIATA a iso-peso (§4) e la stima di famiglia.") # ---------------------------------------------------------- §9 eseguibilita' print("\n" + "-" * L) print(f" 9. ESEGUIBILITA' a ${CAPITAL:.0f} — il vantaggio DICHIARATO dal filone (dimezza gli") print(" ordini). Target NETTATO 0.75*TP01 + 0.25*SKH sul grid 1h = cio' che il cron manda.") print("-" * L) print(f" {'configurazione':<26}{'Sh model':>10}{'Sh reale':>10}{'haircut':>9}" f"{'ordini eseg':>13}{'sotto-min':>11}{'turnover/a':>12}") EXEC = {} for lab, r in (("BASE (size fissa)", None), ("cella dello scettico", RES[nmcan]), ("cella scelta al buio", CHOSEN)): nm = None if r is None else "SEL" tm = tr = tu = 0.0 no = ns = 0 for a in ASSETS: df1h = TO.get1h(a) SZa = {} if r is None else { "SEL": (lambda ex, rr=r, aa=a: size_bin(rr["thr"], rr["w"], rr["h"], rr["sg"])(ex, (can, aa)))} tgt = V.netted_target(a, EX[can][a], nm, SZa, {}, {}, {}, {}, TP, can, df1h) ev = A.eval_weights_smallcap(df1h, tgt, capital=CAPITAL, min_order=MIN_ORDER) tm += 0.5 * ev["modeled"]["sharpe"] tr += 0.5 * ev["realistic"]["sharpe"] tu += ev["executed_turnover_per_year"] no += ev["n_executed_trades"] dw = np.abs(np.diff(np.nan_to_num(tgt), prepend=0.0)) ns += int(((dw > 1e-12) & (dw * CAPITAL < MIN_ORDER)).sum()) EXEC[lab] = (tm, tr, no, ns, tu) print(f" {lab:<26}{tm:>10.3f}{tr:>10.3f}{tm-tr:>9.3f}{no:>13}{ns:>11}{tu:>12.1f}") print(" 'ordini eseg' e 'sotto-min' sono sui due asset sommati, su tutta la storia 1h.") ntr_base = int(sum(len(EX[can][a]['i_ent']) for a in ASSETS)) ntr_sel = int(sum((size_bin(CHOSEN['thr'], CHOSEN['w'], CHOSEN['h'], CHOSEN['sg']) (EX[can][a], (can, a)) > 0).sum() for a in ASSETS)) print(f" trade SKH eseguiti (off{can}): BASE {ntr_base} -> cella scelta {ntr_sel} " f"({ntr_sel/ntr_base*100:.0f}%)") print(" ⚠ IL RISPARMIO DI FEE E' GIA' DENTRO i numeri sopra: `extract` netta ogni trade di") print(f" FEE_RT={V.FEE_RT:.4f} (10 bps RT) all'ingresso, quindi un trade scartato non paga.") print(" Convertirlo di nuovo con la curva del 26/07 (-0,017 Sh/bps di libro, SKH 4x TP01)") print(" sarebbe un DOPPIO CONTEGGIO. Cio' che NON e' dentro e' (a) il min-order a $635,") print(" misurato qui sopra come haircut, e (b) il fatto che la fee reale e' 3,5 bps/lato =") print(" 7 bps RT contro i 10 modellati -> i backtest sono CONSERVATIVI, in entrambe le braccia.") # ---------------------------------------------------------- §9b fee vs informazione print("\n" + "-" * L) print(" 9b. QUANTO DELL'EFFETTO E' INFORMAZIONE E QUANTO E' SOLO FEE RISPARMIATA.") print(" Scartare meta' dei trade risparmia meta' delle fee: parte del +0,080 e' quella,") print(" ed e' un guadagno vero ma NON e' un'ipotesi sulla vol. Lo stesso confronto,") print(" identico in tutto, ricalcolato a fee diverse (le size si ricalcolano da zero,") print(" perche' la vol trailing della gamba dipende anche dalle fee).") print("-" * L) print(f" {'lente':<40}{'cella scettico':>17}{'pos/n':>8}{'cella al buio':>17}{'pos/n':>8}") FEEDEC = {} keep_fee = V.FEE_RT for fee, tag in ((keep_fee, f"MODELLATA {keep_fee*1e4:.0f} bps RT (i backtest)"), (0.0007, "REALE 3,5 bps/lato = 7 bps RT"), (0.0, "fee ZERO (informazione pura)")): V.FEE_RT = fee if fee == keep_fee: EXf, basef = EX, base_leg else: EXf = {o: {a: V.extract(a, o) for a in ASSETS} for o in offs} basef = {o: V.leg_daily(EXf[o], {a: V.size_flat(EXf[o][a]) for a in ASSETS}) for o in offs} row = "" for r in (RES[nmcan], CHOSEN): fn = size_bin(r["thr"], r["w"], r["h"], r["sg"]) bk, ct = {}, {} for o in offs: lg = V.leg_daily(EXf[o], {a: fn(EXf[o][a], (tag, o, a)) for a in ASSETS}) bk[o], ct[o], _ = iso_pair(TP, lg, basef[o]) d = A.anchor_luck_delta(lambda o: bk[o], lambda o: ct[o], offs, metric=sh) FEEDEC[(tag, cname(r["thr"], r["w"], r["h"], r["sg"]))] = d["median_paired"] row += f"{d['median_paired']:>+17.3f}{d['n_positive']:>5}/{len(offs):<3}" print(f" {tag:<40}{row}") if fee != keep_fee: del EXf, basef gc.collect() V.FEE_RT = keep_fee assert V.FEE_RT == 0.001 print(f" (RSS {rss_mb():.0f} MB)") f_mod = FEEDEC[(f"MODELLATA {keep_fee*1e4:.0f} bps RT (i backtest)", nmcan)] f_zero = FEEDEC[("fee ZERO (informazione pura)", nmcan)] quota_info = 100.0 * f_zero / f_mod if f_mod else float("nan") print(f" -> a fee ZERO resta {f_zero:+.3f} di {f_mod:+.3f} = {quota_info:.0f}% dell'effetto.") if quota_info >= 85.0: print(" ❌ IPOTESI REFUTATA (era la mia, ed era anche quella del compito): il vantaggio") print(" NON e' fee risparmiata. A fee zero l'effetto c'e' tutto, quindi scartare i") print(" trade in alta vol e' INFORMAZIONE sul rendimento lordo, non un risparmio di") print(" costo. Conseguenza pratica: NON si puo' replicare abbassando il turnover, e") print(" NON va convertito con la curva fee del 26/07 (sarebbe doppio conteggio in una") print(" direzione e attribuzione sbagliata nell'altra).") else: print(" Una parte e' fee risparmiata; la fee vera (3,5 bps/lato) e' piu' bassa di quella") print(" modellata, quindi nel LIVE quella componente e' PIU' PICCOLA.") # ---------------------------------------------------------- §10 trasferimento su V1 print("\n" + "-" * L) print(" 10. TRASFERIMENTO su SKH01_V1 — seconda sequenza di trade, geometria d'uscita") print(" DIVERSA (stop/target in ATR invece che a percentuale fissa). Non e' un hold-out") print(" temporale, ma e' l'unico campione indipendente disponibile a questo costo.") print("-" * L) EX1 = {o: {a: V.extract(a, o, SKH01_V1) for a in ASSETS} for o in offs} R.get5m.cache_clear() # i 5m non servono piu': ~80 MB restituiti gc.collect() base1 = {o: V.leg_daily(EX1[o], {a: V.size_flat(EX1[o][a]) for a in ASSETS}) for o in offs} print(f" {'cella':<26}{'ISO dShFULL':>13}{'pos/n':>8}{'ISO dShHOLD':>14}{'pos/n':>8}" f"{'tenuti':>9}") for lab, r in (("cella dello scettico", RES[nmcan]), ("cella scelta al buio", CHOSEN)): fn = size_bin(r["thr"], r["w"], r["h"], r["sg"]) b1, c1, kp = {}, {}, [] for o in offs: lg = V.leg_daily(EX1[o], {a: fn(EX1[o][a], ("V1", o, a)) for a in ASSETS}) b1[o], c1[o], _ = iso_pair(TP, lg, base1[o]) kp.append(float(np.mean(np.concatenate( [fn(EX1[o][a], ("V1", o, a)) for a in ASSETS])))) dF = A.anchor_luck_delta(lambda o: b1[o], lambda o: c1[o], offs, metric=sh) dH = A.anchor_luck_delta(lambda o: hold(b1[o]), lambda o: hold(c1[o]), offs, metric=sh) print(f" {cname(r['thr'], r['w'], r['h'], r['sg']):<26}{dF['median_paired']:>+13.3f}" f"{dF['n_positive']:>5}/{len(offs):<3}{dH['median_paired']:>+14.3f}" f"{dH['n_positive']:>5}/{len(offs):<3}{float(np.median(kp))*100:>8.0f}%") del EX1, base1 KC.clear() gc.collect() print(f" (RSS {rss_mb():.0f} MB)") # ---------------------------------------------------------- §11 per anno print("\n" + "-" * L) print(" 11. SCOMPOSIZIONE PER ANNO (regola 22/08 dopo SOL: un contributo positivo si") print(" scompone per ANNO prima di crederci — 24/24 ancore possono essere un anno solo)") print("-" * L) yrs = sorted({int(y) for y in BK[nmcan][can].index.year}) print(f" {'cella':<26}" + "".join(f"{y:>8}" for y in yrs)) for lab, r in (("cella dello scettico", RES[nmcan]), ("cella scelta al buio", CHOSEN)): nm = cname(r["thr"], r["w"], r["h"], r["sg"]) cc = [] for y in yrs: v = [sh(BK[nm][o][BK[nm][o].index.year == y]) - sh(CT[nm][o][CT[nm][o].index.year == y]) for o in offs if (BK[nm][o].index.year == y).sum() > 20] cc.append(float(np.median(v)) if v else float("nan")) print(f" {nm:<26}" + "".join(f"{c:>+8.2f}" for c in cc) + f" positivi {int(np.nansum(np.asarray(cc) > 0))}/{len(yrs)}") # ---------------------------------------------------------- §12 la domanda che decide print("\n" + "-" * L) print(" 12. LA DOMANDA CHE DECIDE — un effetto che vale un terzo della propria incertezza") print(" merita di cambiare un sistema che gira con soldi veri?") print("-" * L) amp = max(ban_F) - min(ban_F) nm = nmcan dC = float(np.median([cagr(BK[nm][o]) - cagr(CT[nm][o]) for o in offs])) cag_band = [cagr(book(TP, base_leg[o])) for o in offs] print(f" (i) IL CONFRONTO GIUSTO NON E' QUELLO DELLA DOMANDA. La banda d'ancora del") print(f" LIVELLO ({amp:.3f} di ampiezza, +-{amp/2:.3f}) e' incertezza sul livello, e") print(f" si CANCELLA in un confronto APPAIATO: le due braccia girano sulla STESSA") print(f" ancora. Cio' che va confrontato con l'effetto e' la dispersione della") print(f" DIFFERENZA appaiata, che e' un ordine di grandezza piu' piccola.") print(f" cella scettico: effetto {RES[nm]['dF']:+.3f}, banda appaiata " f"[{RES[nm]['lo']:+.3f},{RES[nm]['hi']:+.3f}], " f"{RES[nm]['nF']}/{len(offs)} ancore positive.") print(f" (ii) MA la banda del livello torna a mordere DUE volte, e in modo decisivo:") print(f" (a) la SCELTA della cella e' fatta guardando quella banda -> la stima onesta") print(f" non e' l'argmax ma la MEDIANA DI FAMIGLIA " f"({float(np.median([r['dF'] for r in POS])):+.3f} sul segno +1);") print(f" (b) l'effetto e' misurato sugli STESSI 8 anni su cui SKH01 e' stato") print(f" selezionato, e non esiste un forward capace di misurarlo (SE(Sharpe)") print(f" ~1,4 su 6 mesi contro un effetto di 0,08).") print(f" (iii) IL CRITERIO, dichiarato: si tocca il libro live solo se TUTTE E QUATTRO —") print(f" (1) l'effetto sopravvive alla stima ONESTA di famiglia (mediana, non argmax)") print(f" (2) `weights_tilt_null` passa (e' il gate codificato per i cambi di peso)") print(f" (3) il gate marginale + un null CON POTENZA lo sostengono") print(f" (4) l'effetto e' grande abbastanza da pagare un modo di fallire in piu'.") print(f" (iv) ORDINI DI GRANDEZZA a ${CAPITAL:.0f} (regola 26/07: quando le leve hanno") print(f" ordini di grandezza diversi, DIRLO):") print(f" effetto della cella (ISO, mediana appaiata): dCAGR {dC*100:+.2f}%/anno") print(f" = ${dC*CAPITAL:+.2f}/anno = EUR {dC*CAPITAL/365*0.86:+.4f}/giorno") print(f" banda d'ancora del CAGR del libro: " f"[{min(cag_band)*100:.1f}%, {max(cag_band)*100:.1f}%] " f"= ${(max(cag_band)-min(cag_band))*CAPITAL:.0f}/anno di incertezza") print(f" versare EUR 500/mese (il piano dell'operatore) = EUR 16,44/giorno di capitale") print(f" nuovo, cioe' ~{16.44/max(1e-9, abs(dC*CAPITAL/365*0.86)):.0f}x l'effetto misurato qui.") print(f" (v) IL CRITERIO APPLICATO, riga per riga. La forma quantitativa di (1) e'") print(f" dichiarata qui: 'sopravvive alla stima di famiglia' = la MEDIANA della meta'") print(f" d'ipotesi vale almeno la META' della cella selezionata (altrimenti non e' un") print(f" plateau, e' una punta), E la quota di celle positive e' chiaramente sopra il") print(f" 50% (altrimenti la famiglia e' una monetina). Soglie scelte per essere") print(f" verificabili, non per far passare o bocciare un numero gia' visto.") med_pos = float(np.median([r["dF"] for r in POS])) fpos = sum(r["dF"] > 0 for r in POS) / len(POS) c1 = (med_pos >= 0.5 * CHOSEN["dF"]) and (fpos >= 0.65) eurday = dC * CAPITAL / 365 * 0.86 c4 = abs(eurday) >= 0.10 print(f" (1) mediana di famiglia (segno +1) {med_pos:+.3f} = " f"{100*med_pos/CHOSEN['dF'] if CHOSEN['dF'] else float('nan'):.0f}% della cella scelta " f"({CHOSEN['dF']:+.3f}); celle positive {fpos*100:.0f}% -> {'PASS' if c1 else 'FAIL'}") print(f" (2) `weights_tilt_null`: gate_pass=False nella forma nuda, INERTE nella forma") print(f" compensata -> non lo sostiene in nessuna delle due -> FAIL/vacuo") print(f" (3) marginale ADDS su entrambe le celle + null di permutazione battuto a") print(f" p<0.01 con la penale meccanica dentro (§6) -> PASS") print(f" (4) taglia contro un modo di fallire in piu': EUR {eurday:+.4f}/giorno a " f"${CAPITAL:.0f} -> {'PASS' if c4 else 'FAIL'} (soglia dichiarata: EUR 0,10/giorno)") print(f" VERDETTO: {'CANDIDATO' if (c1 and c4) else 'LEAD / NESSUN CAMBIO'} — servono") print(f" tutte e quattro; (1) e (4) falliscono, (2) non e' utilizzabile. Cio' che") print(f" REGGE e va conservato e' il SEGNO: 0/72 celle positive sul controllo inverso.") print("\n" + "=" * L) print(f" fatto in {time.time()-t0:.0f}s") print("=" * L) if __name__ == "__main__": main()