#!/usr/bin/env python """r0822_volsize_skeptic.py — SCETTICO del filone VOL-SIZE (r0822_vol_size.py). BERSAGLIO. Un altro agente ha trovato che dare a SKH01 una SIZE PER-TRADE (cella congelata `VTL tv20 w90`: size fissata all'ingresso = clip(0.20 / rv90(t-1), 0, 3)) vale **+0,070 di Sharpe di libro a ISO-PESO**, positivo in 23/23 ancore e 8/8 anni. Ha dichiarato lui stesso che il meccanismo non e' quello che sembra: il guadagno viene dalla FREQUENZA dei trade (+0,051 sul solo filtro di attivita') e non dalla MAGNITUDINE (+0,003) -> non e' controllo del rischio, e' un filtro di frequenza. E ha PRE-REGISTRATO senza eseguirlo il test che lo falsificherebbe. QUATTRO ATTACCHI, dichiarati prima di misurare, con l'esito che mi aspetto: A1 IL TEST PRE-REGISTRATO — famiglia IV (inverse-vol per-trade sulla vol dell'ASSET) applicata a SKH01_V1. La ragione dichiarata della famiglia IV e' la GEOMETRIA DELLO STOP: su V2-DD lo stop e' una percentuale FISSA (4% long / 2% short), quindi il rischio in unita' di sigma varia col regime; su V1 lo stop e' in ATR, cioe' gia' scalato sulla vol -> l'effetto DEVE essere MINORE. Previsione direzionale: |eff. su V1| < |eff. su V2-DD|, cella per cella. Se e' UGUALE, la spiegazione e' sbagliata anche se il numero regge. ATTESA MIA: che il test sia SENZA POTENZA, perche' sospetto che IV non faccia niente nemmeno su V2-DD (l'agente ha promosso VTL, non IV). Un test pre-registrato su una famiglia inerte non falsifica nulla. A2 LA CONTABILITA' — l'agente dichiara di aver trovato e riparato un look-ahead di contabilita': "tutto il P&L di un trade e' contabilizzato il giorno di CHIUSURA, fino a 3,83 giorni dopo l'ingresso", costo "+0,04 di Sharpe hold-out fantasma", e ne ha tratto una REGOLA pubblicata. Il compito mi chiede di verificare la riparazione troncando i dati alla barra d'ingresso. Lo faccio, ma prima verifico la PREMESSA: quale giorno porta davvero il P&L. ATTESA MIA: la riparazione e' corretta e ridondante. A3 L'ATTACCO ALLA SPIEGAZIONE — se il guadagno e' un filtro di frequenza, deve essere riproducibile SENZA ALCUN vol-target. Controllo piu' brutale possibile: size BINARIA 0/1 sulla stessa soglia della regola congelata (L>=1 <=> rv90<=20%), piu' un filtro che scarta i trade nei periodi di maggiore ATTIVITA' della gamba. Se un binario replica il +0,070, la formula di sizing e' ORNAMENTO. Controllo di segno obbligatorio: il binario INVERSO (tieni solo i trade in alta vol) deve DANNEGGIARE, o non sto misurando un filtro. A4 L'ASSE DELLA FINESTRA — l'agente dichiara che w e' un PICCO, non un plateau (w30 negativo, w180 ~0, w90 ottimo). Un picco su un asse continuo con 23 ancore e' la firma classica di un max-of-k. Misuro l'asse fitto (9 finestre) e rifaccio la scelta di w IN-SAMPLE-ONLY. DISCIPLINA. Non ricalcolo niente che esista: importo `r0822_vol_size` (extract / equity_daily / leg_daily / size_iv / leverage_series / sizes_from_L / book) e replico BIT-EXACT contro `sleeves._skyhook_returns()` prima di ogni delta. Il confronto e' sempre ISO-PESO (baseline ri-scalato allo stesso peso effettivo di SKH), perche' "alzare SKH01" e' gia' stato misurato e respinto il 26/07. Statistica = MEDIANA DELLE DIFFERENZE APPAIATE per ancora (`anchor_luck_delta`), mai differenza delle mediane. USO: nice -n 19 timeout 900 uv run python scripts/research/r0822_volsize_skeptic.py [--every K] --every 8 -> 3 ancore (pilota). default: tutte e 23. """ from __future__ import annotations import argparse 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 r0822_vol_size as V # noqa: E402 (bersaglio: riusato, non riscritto) from src.strategies.skyhook import SKH01_V1 # noqa: E402 ASSETS = V.ASSETS OFFSETS_FULL = V.OFFSETS_FULL FROZEN = dict(tv=0.20, w=90, cap=3.0, active_only=False) # la cella congelata del LEAD sh, vol, maxdd, cagr = V.sh, V.vol, V.maxdd, V.cagr hold, ins, book = V.hold, V.ins, V.book # --------------------------------------------------------------------------- famiglie di size def _keep_to_sizes(ex: dict, keep: pd.Series) -> np.ndarray: return V.sizes_from_L(ex, keep.astype(float)) def size_vtl(sp: dict): def f(ex): return V.sizes_from_L(ex, V.leverage_series(ex["base_daily"], **sp)) return f def size_bin(sp: dict, mode: str): """Size BINARIA 0/1 dalla STESSA L della regola congelata. Nessuna magnitudine: o il trade si fa a size 1, o non si fa. `ge1` usa la soglia della regola stessa (L>=1 <=> rv90<=tv); `gemed` la mediana espandente causale di L; `lt1` e' l'INVERSO (controllo di segno).""" def f(ex): L = V.leverage_series(ex["base_daily"], **sp) if mode == "ge1": keep = pd.Series(L.values >= 1.0, index=L.index) elif mode == "lt1": keep = pd.Series(L.values < 1.0, index=L.index) elif mode == "gemed": med = L.expanding(min_periods=V.VT_WARM).median() keep = pd.Series(np.where(med.notna().values, L.values >= med.values, True), index=L.index) else: raise ValueError(mode) return _keep_to_sizes(ex, keep) return f def size_bin_grad(sp: dict): """BINARIO + GRADUAZIONE: fra i trade tenuti, la size continua di L. Isola quanto aggiunge la FORMULA una volta che la SELEZIONE e' gia' stata fatta.""" def f(ex): L = V.leverage_series(ex["base_daily"], **sp) g = pd.Series(np.where(L.values >= 1.0, L.values, 0.0), index=L.index) return V.sizes_from_L(ex, g) return f def size_chan(mode: str, binary: bool, w: int = 90, cap: float = 3.0): """I DUE CANALI dentro `rv` di una serie all'88% di zeri, isolati come nel bersaglio: mode='mag' -> vol calcolata sulle sole barre ATTIVE (MAGNITUDINE degli esiti) mode='freq' -> conteggio di barre attive nella finestra (FREQUENZA dei trade) binary=False riproduce le celle continue del bersaglio (`VTL neutro attive` / `ACT`); binary=True le riduce a 0/1 alla mediana ESPANDENTE causale della propria L.""" def f(ex): if mode == "mag": Ls = V.leverage_series(ex["base_daily"], tv=None, w=w, cap=cap, active_only=True) else: Ls = V.activity_L(ex["base_daily"], w=w, cap=cap) if not binary: return V.sizes_from_L(ex, Ls) med = Ls.expanding(min_periods=V.VT_WARM).median() keep = pd.Series(np.where(med.notna().values, Ls.values >= med.values, True), index=Ls.index) return _keep_to_sizes(ex, keep) return f def size_dropact(w: int = 90, q: float = 0.75): """Scarta i trade il cui giorno d'ingresso cade nei periodi in cui la gamba e' stata piu' ATTIVA (conteggio di barre attive negli ultimi w giorni sopra il quantile ESPANDENTE causale q). Nessun vol-target, nessuna magnitudine, nessun prezzo: solo la frequenza dei trade.""" def f(ex): base = ex["base_daily"] cnt = (base != 0.0).astype(float).rolling(w, min_periods=max(5, w // 3)).sum().shift(1) thr = cnt.expanding(min_periods=V.VT_WARM).quantile(q) keep = pd.Series(np.where(thr.notna().values & cnt.notna().values, cnt.values <= thr.values, True), index=base.index) return _keep_to_sizes(ex, keep) return f def iv_family() -> dict: d = {} for p in (0.5, 1.0): for w in (30, 90): for cp in (2.0, 4.0): d[f"IV p{p} w{w} cap{cp:.0f}"] = V.size_iv(p, w, cp) return d # --------------------------------------------------------------------------- motore def legs(EX: dict, funcs: dict) -> dict: """name -> serie giornaliera della gamba SKH 50/50, per un'ancora.""" out = {"BASE": V.leg_daily(EX, {a: V.size_flat(EX[a]) for a in ASSETS})} for nm, fn in funcs.items(): out[nm] = V.leg_daily(EX, {a: fn(EX[a]) for a in ASSETS}) return out 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). E' la convenzione del bersaglio, riusata.""" vb = vol(leg_base) m_s = vol(leg) / vb if vb > 0 else 1.0 return book(TP, leg), book(TP, leg_base * m_s), m_s def delta_table(BK: dict, CT: dict, names, offs, label: str, extra_hold=True) -> dict: res = {} print(f" {'cella':<26}{'ISO dShFULL':>12}{'pos/n':>8}{'banda':>20}" + (f"{'ISO dShHOLD':>13}{'pos/n':>8}" if extra_hold else "") + f"{'wSKHeff':>9}") for nm in names: dF = A.anchor_luck_delta(lambda o, n=nm: BK[n][o], lambda o, n=nm: CT[n][o], offs, metric=sh) row = dict(dF=dF["median_paired"], nF=dF["n_positive"], lo=dF["lo"], hi=dF["hi"]) s = (f" {nm:<26}{row['dF']:>+12.3f}{row['nF']:>5}/{len(offs):<3}" f" [{row['lo']:+.3f},{row['hi']:+.3f}]".ljust(20 + 26 + 12 + 8)) s = (f" {nm:<26}{row['dF']:>+12.3f}{row['nF']:>5}/{len(offs):<3}" f"{'[%+.3f,%+.3f]' % (row['lo'], row['hi']):>20}") if extra_hold: dH = A.anchor_luck_delta(lambda o, n=nm: hold(BK[n][o]), lambda o, n=nm: hold(CT[n][o]), offs, metric=sh) row.update(dH=dH["median_paired"], nH=dH["n_positive"]) s += f"{row['dH']:>+13.3f}{row['nH']:>5}/{len(offs):<3}" we = float(np.median([0.25 * m for m in MSCALE[nm]])) if nm in MSCALE else float("nan") row["weff"] = we s += f"{we:>9.3f}" print(s) res[nm] = row return res MSCALE: dict = {} def per_year(BK: dict, CT: dict, nm: str, offs) -> tuple: yrs = sorted({int(y) for y in BK[nm][offs[0]].index.year}) cells = [] 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] cells.append(float(np.median(v)) if v else float("nan")) return yrs, cells # =========================================================================== main def main() -> None: ap = argparse.ArgumentParser() ap.add_argument("--every", type=int, default=1) args = ap.parse_args() offs = OFFSETS_FULL[:: max(1, args.every)] t0 = time.time() L = 112 print("=" * L) print(" r0822 VOL-SIZE — SCETTICO. Bersaglio: `VTL tv20 w90` = +0,070 ISO dShFULL (23/23)") print("=" * L) print(f" ancore: {len(offs)}/{len(OFFSETS_FULL)} della griglia a priori 02/07 {list(offs)}") print(" lente: path CANONICO (identica nelle due braccia di ogni confronto). TP01 canonico.") TP = A.tp01_baseline_daily() # ------------------------------------------------------------------ celle IV = iv_family() FUN_V2 = dict(IV) FUN_V2["VTL tv20 w90 (LEAD)"] = size_vtl(FROZEN) FUN_V2["BIN L>=1 (0/1)"] = size_bin(FROZEN, "ge1") FUN_V2["BIN L>=med (0/1)"] = size_bin(FROZEN, "gemed") FUN_V2["BIN L<1 (INVERSO)"] = size_bin(FROZEN, "lt1") FUN_V2["BIN+GRAD"] = size_bin_grad(FROZEN) FUN_V2["DROP attivita' >p75"] = size_dropact(90, 0.75) FUN_V2["DROP attivita' >p50"] = size_dropact(90, 0.50) FUN_V2["canale MAGNITUDINE cont."] = size_chan("mag", False) FUN_V2["canale FREQUENZA cont."] = size_chan("freq", False) FUN_V2["canale MAGNIT. 0/1"] = size_chan("mag", True) FUN_V2["canale FREQ. 0/1"] = size_chan("freq", True) WGRID = (15, 20, 30, 45, 60, 90, 120, 180, 270) for w in WGRID: if w != 90: FUN_V2[f"VTL tv20 w{w}"] = size_vtl(dict(FROZEN, w=w)) FUN_V1 = dict(IV) FUN_V1["VTL tv20 w90 (LEAD)"] = size_vtl(FROZEN) FUN_V1["BIN L>=1 (0/1)"] = size_bin(FROZEN, "ge1") NTRIAL = len(FUN_V2) + len(FUN_V1) + 2 # +2 baseline print(f"\n GRIGLIA DICHIARATA: {len(FUN_V2)} celle su V2-DD + {len(FUN_V1)} su V1 " f"+ 2 baseline = {NTRIAL} valutazioni x {len(offs)} ancore. Nessuna cella scelta") print(" dopo aver guardato: IV e' la famiglia PRE-REGISTRATA, VTL/w e' l'asse dichiarato") print(" fragile dall'agente, BIN/DROP sono controlli miei dichiarati sopra.") # ------------------------------------------------------------------ ciclo ancore BK2, CT2, BK1, CT1 = {}, {}, {}, {} EX0 = None print("\n calcolo per ancora ...", flush=True) for o in offs: EXV2 = {a: V.extract(a, o) for a in ASSETS} EXV1 = {a: V.extract(a, o, SKH01_V1) for a in ASSETS} if o == offs[0]: EX0 = (EXV2, EXV1) for tag, EXX, FUN, BKd, CTd in (("V2", EXV2, FUN_V2, BK2, CT2), ("V1", EXV1, FUN_V1, BK1, CT1)): lg = legs(EXX, FUN) base = lg["BASE"] if tag == "V2": # controlli di §1(c): L calcolata sulla gamba COMBINATA (cio' che fa il ramo # NAIVE del bersaglio), applicata nei DUE modi. Se coincidono, il "fantasma" # non e' un problema di QUANDO si applica la size. for w in (30, 90): Lc = V.leverage_series(base, **dict(FROZEN, w=w)) lg[f"NAIVEcomb w{w}"] = base * Lc.reindex(base.index).fillna(1.0) lg[f"COMBingr w{w}"] = V.leg_daily( EXX, {a: V.sizes_from_L(EXX[a], Lc) for a in ASSETS}) for nm, s in lg.items(): b, c, m_s = iso_pair(TP, s, base) BKd.setdefault(nm, {})[o] = b CTd.setdefault(nm, {})[o] = c if tag == "V2": MSCALE.setdefault(nm, []).append(m_s) if o != offs[0]: del EXV2, EXV1 print(f" fatto in {time.time()-t0:.0f}s") EXV2, EXV1 = EX0 # ------------------------------------------------------------------ §0 SANITY print("\n" + "-" * L) print(" 0. SANITY — replica BIT-EXACT prima di qualunque delta (se non e' 0, il problema") print(" e' mio e va detto)") print("-" * L) from src.portfolio.sleeves import _skyhook_returns mine = V.leg_daily(EXV2, {a: V.size_flat(EXV2[a]) for a in ASSETS}) off = _skyhook_returns() d0 = float(np.max(np.abs(mine.values - off.values))) if len(mine) == len(off) else float("nan") print(f" gamba 50/50 off{offs[0]} vs sleeves._skyhook_returns(): max|dif| = {d0:.2e} " f"({len(mine)} giorni)") assert d0 < 1e-15, "replica NON bit-exact" for a in ASSETS: ref = R.run_asset(a, offs[0])[0] R._CACHE.clear() dd = float(np.max(np.abs(V.equity_daily(EXV2[a], V.size_flat(EXV2[a])).values - ref.values))) print(f" {a} vs r0702_anchor_skh01.run_asset: max|dif| = {dd:.2e}") assert dd < 1e-15 lead = "VTL tv20 w90 (LEAD)" dLEAD = A.anchor_luck_delta(lambda o: BK2[lead][o], lambda o: CT2[lead][o], offs, metric=sh) dLEADH = A.anchor_luck_delta(lambda o: hold(BK2[lead][o]), lambda o: hold(CT2[lead][o]), offs, metric=sh) print(f" RIPRODUZIONE DEL TITOLO: ISO dShFULL {dLEAD['median_paired']:+.3f} " f"({dLEAD['n_positive']}/{len(offs)}) ISO dShHOLD {dLEADH['median_paired']:+.3f} " f"({dLEADH['n_positive']}/{len(offs)}) [pubblicato +0,070 (23/23) / +0,101 (19/23)]") med_F = float(np.median([sh(BK2["BASE"][o]) for o in offs])) med_H = float(np.median([sh(hold(BK2["BASE"][o])) for o in offs])) med_D = float(np.median([maxdd(BK2["BASE"][o]) for o in offs])) print(f" libro BASE (V2-DD), mediana della banda: ShFULL {med_F:.3f} / ShHOLD {med_H:.3f} " f"/ maxDD {med_D*100:.1f}% [pubblicato 1,626 / 1,037 / 10,4%]") # ------------------------------------------------------------------ §1 CONTABILITA' print("\n" + "-" * L) print(" 1. LA CONTABILITA' — verifica della PREMESSA prima della riparazione") print(" (l'agente: 'tutto il P&L di un trade e' contabilizzato il giorno di CHIUSURA')") print("-" * L) tot_ent = tot_ex = tot_chk = 0 for a in ASSETS: ex = EXV2[a] d_e = ex["idx"][ex["i_ent"]].floor("D") d_x = ex["idx"][ex["i_ex"]].floor("D") dl = ex["base_daily"] multi = np.where(d_e.values != d_x.values)[0] ne = nx = ck = 0 for k in multi: n = float(ex["net"][k]) if abs(n) < 1e-6: continue ck += 1 ne += int(abs(float(dl.get(d_e[k], 0.0)) - n) < 1e-9) nx += int(abs(float(dl.get(d_x[k], 0.0)) - n) < 1e-9) print(f" {a}: {ck} trade multi-giorno -> ritorno giornaliero == net del trade " f"al giorno d'INGRESSO {ne}/{ck}, al giorno di USCITA {nx}/{ck}") tot_ent += ne; tot_ex += nx; tot_chk += ck print(f" TOTALE: INGRESSO {tot_ent}/{tot_chk} ({tot_ent/max(tot_chk,1)*100:.1f}%) " f"USCITA {tot_ex}/{tot_chk} ({tot_ex/max(tot_chk,1)*100:.1f}%)") print(" -> `backtest_signals` fa `equity[i_ent:i_ex+1] = capital`: il gradino comincia") print(" alla barra d'INGRESSO e il resample giornaliero (last) lo legge quel giorno.") print("\n (a) TRONCAMENTO ALLA BARRA D'INGRESSO — le size della regola congelata " "ricalcolate") print(" con i soli dati <= barra d'ingresso devono coincidere ESATTAMENTE") worst = 0.0 ntest = 0 for a in ASSETS: ex = EXV2[a] full = size_vtl(FROZEN)(ex) sel = [k for k in range(len(ex["i_ent"])) if ex["i_ent"][k] >= V.SIG_WARM] sel = sel[:: max(1, len(sel) // 25)][:25] bad = 0 for k in sel: i = int(ex["i_ent"][k]) # ricostruisco la gamba usando SOLO i trade gia' CHIUSI prima dell'ingresso k sub = {kk: ex[kk] for kk in ("n", "idx")} m = ex["i_ex"] < i sub = dict(n=i + 1, idx=ex["idx"][: i + 1], i_ent=ex["i_ent"][m], i_ex=ex["i_ex"][m], net=ex["net"][m]) if not len(sub["i_ent"]): continue bd = V.equity_daily(sub, np.ones(len(sub["i_ent"]))) Lt = V.leverage_series(bd, **FROZEN) day = ex["idx"][i].floor("D") v = Lt.reindex([day]).values[0] v = 1.0 if not np.isfinite(v) else float(v) worst = max(worst, abs(v - float(full[k]))) bad += int(abs(v - float(full[k])) > 1e-9) ntest += 1 print(f" {a}: {len(sel)-bad}/{len(sel)} size identiche a dati troncati " f"(divergenze {bad})") print(f" max|dif| su {ntest} size ricalcolate da zero: {worst:.2e} -> " f"{'CAUSALE' if worst < 1e-9 else 'NON CAUSALE'}") print("\n (b) LA 'RIPARAZIONE' E' UN NO-OP: size-all'INGRESSO vs moltiplicare la SERIE") print(" GIORNALIERA, a parita' di tutto il resto (stessa L, PER ASSET)") for a in ASSETS: ex = EXV2[a] Ls = V.leverage_series(ex["base_daily"], **FROZEN) s_in = V.equity_daily(ex, V.sizes_from_L(ex, Ls)) s_mul = ex["base_daily"] * Ls.reindex(ex["base_daily"].index).fillna(1.0) j = pd.concat({"a": s_in, "b": s_mul}, axis=1, join="inner").dropna() print(f" {a}: max|dif| = {np.max(np.abs(j['a']-j['b'])):.2e} " f"Sh ingresso {sh(s_in):.4f} vs Sh moltiplicazione {sh(s_mul):.4f}") print("\n (c) ALLORA COS'E' IL '+0,04 FANTASMA'? Il ramo NAIVE del bersaglio calcola L") print(" sulla gamba 50/50 COMBINATA e MOLTIPLICA la serie; il ramo 'causale' calcola L") print(" PER ASSET e la applica all'INGRESSO. Sono DUE differenze in una. Le separo:") for w in (30, 90): d = [] for o in offs: j = pd.concat({"a": BK2[f"NAIVEcomb w{w}"][o], "b": BK2[f"COMBingr w{w}"][o]}, axis=1, join="inner").dropna() d.append(float(np.max(np.abs(j["a"] - j["b"])))) print(f" w{w}: [L combinata, MOLTIPLICA] vs [L combinata, size all'INGRESSO] -> " f"max|dif| mediano {np.median(d):.2e} = IDENTICHE") print(f"\n {'coppia':<38}{'dShFULL':>10}{'dShHOLD':>10} (delta GREZZI vs baseline nudo,") print(f" {'':<38}{'':>10}{'':>10} la convenzione del §4 del bersaglio)") for w in (30, 90): for nm in (f"NAIVEcomb w{w}", f"COMBingr w{w}", (lead if w == 90 else f"VTL tv20 w{w}")): gF = A.anchor_luck_delta(lambda o, n=nm: BK2[n][o], lambda o: BK2["BASE"][o], offs, metric=sh) gH = A.anchor_luck_delta(lambda o, n=nm: hold(BK2[n][o]), lambda o: hold(BK2["BASE"][o]), offs, metric=sh) print(f" {nm:<38}{gF['median_paired']:>+10.3f}{gH['median_paired']:>+10.3f}") print(" -> il 'fantasma' e' interamente [L combinata] - [L per asset]: una scelta di") print(" SERIE DI RIFERIMENTO, non un look-ahead. Il ramo NAIVE non e' non-causale.") # ------------------------------------------------------------------ §2 TEST PRE-REGISTRATO print("\n" + "-" * L) print(" 2. IL TEST PRE-REGISTRATO — famiglia IV su V2-DD (stop % FISSO) vs V1 (stop in ATR)") print(" Previsione dichiarata dall'agente: |effetto su V1| < |effetto su V2-DD|.") print("-" * L) n2 = {a: len(EXV2[a]["i_ent"]) for a in ASSETS} n1 = {a: len(EXV1[a]["i_ent"]) for a in ASSETS} print(f" trade all'ancora {offs[0]}: V2-DD {n2} V1 {n1}") mb2 = float(np.median([sh(BK2["BASE"][o]) for o in offs])) mb1 = float(np.median([sh(BK1["BASE"][o]) for o in offs])) print(f" libro BASE mediana di banda: V2-DD ShFULL {mb2:.3f} | V1 ShFULL {mb1:.3f}") names_iv = list(IV) print(f"\n {'cella IV':<22}{'V2-DD ISO dF':>14}{'pos/n':>8}{'V1 ISO dF':>12}{'pos/n':>8}" f"{'|V1|<|V2|?':>12}") ok = tot = 0 e2s, e1s = [], [] for nm in names_iv: a2 = A.anchor_luck_delta(lambda o, n=nm: BK2[n][o], lambda o, n=nm: CT2[n][o], offs, metric=sh) a1 = A.anchor_luck_delta(lambda o, n=nm: BK1[n][o], lambda o, n=nm: CT1[n][o], offs, metric=sh) e2, e1 = a2["median_paired"], a1["median_paired"] e2s.append(e2); e1s.append(e1) good = abs(e1) < abs(e2) ok += int(good); tot += 1 print(f" {nm:<22}{e2:>+14.3f}{a2['n_positive']:>5}/{len(offs):<3}{e1:>+12.3f}" f"{a1['n_positive']:>5}/{len(offs):<3}{('SI' if good else 'NO'):>12}") print(f"\n previsione verificata in {ok}/{tot} celle della famiglia") print(f" |effetto| mediano di famiglia: V2-DD {np.median(np.abs(e2s)):.3f} " f"V1 {np.median(np.abs(e1s)):.3f}") print(f" effetto mediano SEGNATO: V2-DD {np.median(e2s):+.3f} V1 {np.median(e1s):+.3f}") print(" ⚠️ POTENZA DEL TEST: se |effetto su V2-DD| e' gia' ~0, la previsione e' verificabile") print(" solo per caso e il gate pre-registrato NON puo' falsificare niente.") for nm in ("VTL tv20 w90 (LEAD)", "BIN L>=1 (0/1)"): a2 = A.anchor_luck_delta(lambda o, n=nm: BK2[n][o], lambda o, n=nm: CT2[n][o], offs, metric=sh) a1 = A.anchor_luck_delta(lambda o, n=nm: BK1[n][o], lambda o, n=nm: CT1[n][o], offs, metric=sh) print(f" (riferimento) {nm:<24} V2-DD {a2['median_paired']:+.3f} " f"({a2['n_positive']}/{len(offs)}) V1 {a1['median_paired']:+.3f} " f"({a1['n_positive']}/{len(offs)})") # ------------------------------------------------------------------ §3 ATTACCO ALLA SPIEGAZIONE print("\n" + "-" * L) print(" 3. ATTACCO ALLA SPIEGAZIONE — quanto del +0,070 resta senza NESSUN vol-target?") print(" Tutte ISO-PESO vs lo stesso baseline. 'wSKHeff' = peso effettivo di SKH nel libro") print("-" * L) grp = [lead, "BIN L>=1 (0/1)", "BIN+GRAD", "BIN L>=med (0/1)", "BIN L<1 (INVERSO)", "DROP attivita' >p75", "DROP attivita' >p50", "canale MAGNITUDINE cont.", "canale FREQUENZA cont.", "canale MAGNIT. 0/1", "canale FREQ. 0/1"] RES = delta_table(BK2, CT2, grp, offs, "attacco") base_eff = RES[lead]["dF"] print() for nm in grp[1:]: frac = RES[nm]["dF"] / base_eff * 100 if abs(base_eff) > 1e-9 else float("nan") print(f" {nm:<26} recupera {frac:>6.0f}% del vantaggio del LEAD") # quanti trade tocca ciascun filtro print(f"\n {'cella':<26}{'trade tenuti':>14}{'su':>6}{'size mediana':>14}" f"{'size = 0':>10}") for nm in [lead] + grp[1:]: kept = tot_t = zer = 0 med = [] for a in ASSETS: s = FUN_V2[nm](EXV2[a]) tot_t += len(s); kept += int((s > 0).sum()); zer += int((s == 0).sum()) med.append(s) mm = float(np.median(np.concatenate(med))) print(f" {nm:<26}{kept:>14}{tot_t:>6}{mm:>14.2f}{zer:>10}") print(f"\n COSA SELEZIONA il filtro (ancora {offs[0]}, i due asset uniti): net medio dei") print(" trade TENUTI vs SCARTATI, e attivita' trailing media dei due gruppi") print(f" {'cella':<26}{'net TENUTI':>12}{'net SCARTATI':>14}{'att.TEN':>10}{'att.SCA':>10}" f"{'%tenuti':>9}") for nm in ("BIN L>=1 (0/1)", "BIN L>=med (0/1)", "BIN L<1 (INVERSO)", "DROP attivita' >p50", "canale MAGNIT. 0/1", "canale FREQ. 0/1"): nk, nd, ak, ad = [], [], [], [] for a in ASSETS: ex = EXV2[a] szv = FUN_V2[nm](ex) cnt = ((ex["base_daily"] != 0.0).astype(float) .rolling(90, min_periods=30).sum().shift(1)) av = cnt.reindex(ex["day_ent"]).values keep = szv > 0 nk += list(ex["net"][keep]); nd += list(ex["net"][~keep]) ak += list(av[keep]); ad += list(av[~keep]) f_ = 100.0 * len(nk) / max(len(nk) + len(nd), 1) print(f" {nm:<26}{np.mean(nk)*100:>11.2f}%{np.mean(nd)*100 if nd else float('nan'):>13.2f}%" f"{np.nanmean(ak):>10.1f}{np.nanmean(ad) if ad else float('nan'):>10.1f}{f_:>9.0f}") print("\n per anno (ISO dSh, mediana appaiata sulle ancore):") yrs = None for nm in [lead, "BIN L>=1 (0/1)", "DROP attivita' >p75", "BIN L<1 (INVERSO)"]: yrs, cells = per_year(BK2, CT2, nm, offs) if nm == lead: print(" " + f"{'':<26}" + "".join(f"{y:>8}" for y in yrs)) npos = int(np.nansum(np.asarray(cells) > 0)) print(f" {nm:<26}" + "".join(f"{c:>+8.2f}" for c in cells) + f" anni positivi {npos}/{len(yrs)}") # ------------------------------------------------------------------ §4 ASSE DELLA FINESTRA print("\n" + "-" * L) print(" 4. L'ASSE DELLA FINESTRA — picco o plateau? + scelta di w IN-SAMPLE-ONLY") print("-" * L) wnames = [(w, f"VTL tv20 w{w}" if w != 90 else lead) for w in WGRID] print(f" {'w':>6}{'ISO dShFULL':>14}{'pos/n':>8}{'ISO dShIS':>12}{'ISO dShHOLD':>13}" f"{'pos/n':>8}") wf, wi, wh = {}, {}, {} for w, nm in wnames: dF = A.anchor_luck_delta(lambda o, n=nm: BK2[n][o], lambda o, n=nm: CT2[n][o], offs, metric=sh) dI = A.anchor_luck_delta(lambda o, n=nm: ins(BK2[n][o]), lambda o, n=nm: ins(CT2[n][o]), offs, metric=sh) dH = A.anchor_luck_delta(lambda o, n=nm: hold(BK2[n][o]), lambda o, n=nm: hold(CT2[n][o]), offs, metric=sh) wf[w], wi[w], wh[w] = dF["median_paired"], dI["median_paired"], dH["median_paired"] print(f" {w:>6}{wf[w]:>+14.3f}{dF['n_positive']:>5}/{len(offs):<3}{wi[w]:>+12.3f}" f"{wh[w]:>+13.3f}{dH['n_positive']:>5}/{len(offs):<3}") wbest_full = max(WGRID, key=lambda w: wf[w]) wbest_is = max(WGRID, key=lambda w: wi[w]) print(f"\n argmax su FULL (cio' che l'agente ha pubblicato): w = {wbest_full} " f"({wf[wbest_full]:+.3f})") print(f" argmax IN-SAMPLE-ONLY (la sola scelta ammessa): w = {wbest_is} " f"(IS {wi[wbest_is]:+.3f}) -> quella cella vale FULL {wf[wbest_is]:+.3f} / " f"HOLD {wh[wbest_is]:+.3f}") order = sorted(WGRID) k = order.index(wbest_full) nb = [order[k - 1] if k > 0 else None, order[k + 1] if k + 1 < len(order) else None] for w in nb: if w is not None: print(f" vicino w={w:<4} vale {wf[w]:+.3f} = {wf[w]/wf[wbest_full]*100:>5.0f}% " f"del picco") pos = sum(1 for w in WGRID if wf[w] > 0) print(f" celle con ISO dShFULL > 0: {pos}/{len(WGRID)} " f"spread della famiglia {max(wf.values())-min(wf.values()):.3f} " f"sd fra celle {np.std(list(wf.values()), ddof=1):.3f}") print(f" stima ONESTA se w e' SCELTO (mediana della famiglia, come per la banda " f"d'ancora): {np.median(list(wf.values())):+.3f} FULL / " f"{np.median(list(wh.values())):+.3f} HOLD contro il picco " f"{wf[wbest_full]:+.3f} / {wh[wbest_full]:+.3f}") rho = float(np.corrcoef([wi[w] for w in WGRID], [wh[w] for w in WGRID])[0, 1]) print(f" corr(ISO dSh in-sample, ISO dSh hold-out) fra le {len(WGRID)} finestre: {rho:+.2f}") print(" -> se e' ~0 o negativa, la scelta di w in-sample NON predice l'hold-out: l'asse") print(" e' rumore e il picco a w=90 e' un max-of-k.") # ------------------------------------------------------------------ §5 sintesi print("\n" + "-" * L) print(" 5. GATE (per il contratto di consegna)") print("-" * L) can = offs[0] legs_can = {} for nm in ("BASE", lead, "BIN L>=med (0/1)"): legs_can[nm] = (V.leg_daily(EXV2, {a: V.size_flat(EXV2[a]) for a in ASSETS}) if nm == "BASE" else V.leg_daily(EXV2, {a: FUN_V2[nm](EXV2[a]) for a in ASSETS})) nlos = int((EXV2["BTC"]["net"] < 0).sum() + (EXV2["ETH"]["net"] < 0).sum()) ntot = int(len(EXV2["BTC"]["net"]) + len(EXV2["ETH"]["net"])) for nm, s_ in legs_can.items(): m = A.marginal_vs_tp01(s_) imp = A.implausible_sharpe(s_, n_trades=ntot, n_losing_trades=nlos) print(f" {nm:<22} marginal {str(m.get('marginal_verdict')):<9} corr {m.get('corr_full')}" f" implausible={imp['implausible']} (perdite/attive " f"{imp.get('loss_frac', 0)*100:.1f}%)") trials = [float(np.median([sh(BK2[n][o]) for o in offs])) for n in BK2] dsr_c, sr0 = A.deflated_sharpe(float(np.median([sh(BK2[lead][o]) for o in offs])), trials, BK2[lead][can]) dsr_b, _ = A.deflated_sharpe(med_F, trials, BK2["BASE"][can]) print(f" deflated-Sharpe: candidato {dsr_c:.3f} / BASELINE {dsr_b:.3f} su {len(trials)} " f"trial (sd {np.std(trials, ddof=1):.4f}) -> VACUO, lo passa anche il baseline") print(" null de-levering: DEGENERE per costruzione (sh(k*base) == sh(base)); il suo") print(" sostituto e' il confronto ISO-PESO usato in tutte le tabelle sopra.") print(" causalita': §1(a) 50/50 size ricalcolate da dati troncati, max|dif| 0.00e+00.") print("\n" + "-" * L) print(" 6. SINTESI DEI NUMERI CHE CONTANO") print("-" * L) print(f" fortuna d'ancora del LIBRO BASE: ShFULL canonico " f"{sh(BK2['BASE'][0]) if 0 in BK2['BASE'] else float('nan'):.3f} vs mediana " f"{med_F:.3f}") print(f" vantaggio del LEAD (ISO): {dLEAD['median_paired']:+.3f} FULL / " f"{dLEADH['median_paired']:+.3f} HOLD") print(f" riprodotto dal FILTRO BINARIO: {RES['BIN L>=1 (0/1)']['dF']:+.3f} FULL / " f"{RES['BIN L>=1 (0/1)']['dH']:+.3f} HOLD") print(f" riprodotto dal BINARIO alla MEDIANA:{RES['BIN L>=med (0/1)']['dF']:+.3f} FULL / " f"{RES['BIN L>=med (0/1)']['dH']:+.3f} HOLD ({RES['BIN L>=med (0/1)']['nF']}/{len(offs)} ancore)") print(f" controllo di segno (INVERSO): {RES['BIN L<1 (INVERSO)']['dF']:+.3f} FULL " f"({RES['BIN L<1 (INVERSO)']['nF']}/{len(offs)}) -> il filtro e' reale, non un artefatto") print(f"\n fatto in {time.time()-t0:.0f}s") print("=" * L) if __name__ == "__main__": main()