#!/usr/bin/env python """r0823_tp01_twin.py — §49 TP01-TWIN: TP01 e' UN SOLO meccanismo sul 75% del libro. LA DOMANDA (decidibile, non retorica) ===================================== TP01 e' il 75% del libro live. Il progetto ne ha misurato il valore con precisione: NON genera ritorno (contributo hold-out negativo nel 99,1% delle ancore, LOO 26/07) ma TAGLIA IL DRAWDOWN ~6x contro il buy&hold, e supera il criterio di protezione di `edge_watch` 8/8 anni di sinistro. E' un'assicurazione e si giudica sul sinistro. Ma la sua direzione e' UNA definizione di trend: media di tre `sign(close/close[-h]-1)` a 30/90/180 giorni. Tutto il 75% del libro dipende da quella. Sostituire TP01 con un ENSEMBLE di meccanismi di trend DIVERSI ma tutti long-flat migliora la PROTEZIONE (che e' il suo compito), a pari drift e a pari costo? NB la formulazione: NON "un secondo sleeve che si aggiunge" (quella e' la domanda `marginal_vs_tp01`, gia' fatta decine di volte e sempre REDUNDANT per i trend), ma LA STESSA QUOTA DI LIBRO espressa da piu' meccanismi. E' robustezza, non alpha. Percio' `weights_tilt_null` NON si applica: il peso 75/25 non si muove, cambia solo cosa c'e' dentro il 75%. ATTESA A PRIORI — SCRITTA PRIMA DI MISURARE (riportata anche se refutata; lo e') =============================================================================== (A1) "I meccanismi di trend a 1d sullo stesso asset correlano 0,85-0,95, quindi l'ensemble sara' TP01 con piu' fee." Corollario: la matrice di correlazione (§2) puo' chiudere il filone da sola. -> REFUTATA: corr media 0,795 sui rendimenti e 0,608 sulle POSIZIONI; KEL e KAU stanno a 0,51-0,53 di posizione da TSMOM. Non sono lo stesso oggetto. (A2) "Se l'ensemble avra' meno DD sara' per DE-LEVERING (7a occorrenza del null)." -> REFUTATA: la vol realizzata e' IDENTICA (12,1% vs 12,0%), l'esposizione media e il tempo a mercato dell'ensemble sono PIU' ALTI. Non c'e' niente da de-levere: il null non e' applicabile, e va detto cosi' invece di dire "non refuta" (che suggerirebbe che l'ensemble abbia superato un test che in realta' non aveva potenza). MECCANISMI — DICHIARATI TUTTI PRIMA, E CONTATI COME TRIAL (§8) ============================================================== Tutti LONG-FLAT, tutti vol-targeted 20% (cap 2x, finestra 30g), tutti a 1d, tutti sullo stesso 50/50 BTC+ETH, tutti causali (decidono a close[i], tengono in i+1), tutti a fee 0,10% RT. TSMOM — il riferimento: media dei sign(c/c[-h]-1), h in 30/90/180 (= TP01 CANONICAL) DON — Donchian breakout long-flat: entra sopra il massimo dei precedenti N, esce sotto il minimo dei precedenti N/2. NB: SKH01 e' un Donchian ma L/S e dual-TF 230m/690m — NON e' lo stesso oggetto e qui non viene ricreato. EWMA — incrocio di medie esponenziali, blend di 3 coppie (stessa forma multi-orizzonte di TSMOM) KEL — canale di Keltner: entra sopra ema+k*atr, esce sotto la ema (mid) KAU — Kaufman: KAMA (efficiency-ratio adattiva), lunga se la KAMA sale su `slope` barre Indicatori presi da `src/strategies/indicators.py` (ema, atr) e da `altlib` (donchian, vol_target, eval_weights, eval_weights_smallcap, causality_ok, study_family_honest): importati, non riscritti. ENSEMBLE (due varianti, dichiarate a priori) ENS-DIR — media delle 5 DIREZIONI, poi UN solo vol-target. E' esattamente la struttura di TP01 (che gia' media 3 direzioni prima di scalare): "stessa quota, piu' meccanismi". ENS-POS — media delle 5 POSIZIONI gia' vol-targeted (differisce solo dove morde il cap 2x). Il costo si conta sulla posizione NETTATA (il libro ha UNA posizione per strumento su Deribit): la fee e' su |delta(posizione media)|. Si riporta anche il caso non-nettato come limite superiore. COSA DECIDE (in ordine) §2 correlazione fra meccanismi §4 PROTEZIONE anno per anno, criterio (B) di `scripts/live/edge_watch.py` importato dal sorgente di produzione: negli anni con DD buy&hold > 10%, il DD dello sleeve deve restare sotto il 75% di quello. ⚠️ ERRORE MIO, CATTURATO IN SESSIONE E RIPORTATO: la prima stesura decideva sul CASO PEGGIORE (il massimo degli 8 rapporti), che migliora 0,56 -> 0,49, e avrebbe stampato "ENSEMBLE PROMOSSO A LEAD". Ma il massimo di 8 numeri non e' una statistica: si e' mosso perche' e' migliorato UN anno (2023), mentre 4 anni su 8 PEGGIORANO — compreso il 2022, che e' l'unico sinistro vero del campione (0,04 -> 0,13, tre volte peggio). §4-bis misura la DISTRIBUZIONE (test dei segni per anno, e 24x8 celle appaiate sulle ancore) invece del massimo. §5 GATE ISO-VOLATILITA' + NULL DEL DE-LEVERING (7a occorrenza). NB: lo Sharpe e' invariante alla scala, quindi il null "iso-DD, confronta lo Sharpe" e' DEGENERE (regola dell'ondata 22/08): la forma non degenere e' iso-VOL sulla PROTEZIONE + confronto diretto di Sharpe. §5-bis LA COMPOSIZIONE DELL'ENSEMBLE E' ESSA STESSA UN PARAMETRO LIBERO: 31 sottoinsiemi non vuoti, di cui ne e' stato dichiarato UNO. Si misura dove cade il dichiarato nella distribuzione dei 31 — se sta in alto, un ensemble "migliore" e' selezione, non robustezza. §6 banda d'ancora 24 ore, MEDIANA DELLE DIFFERENZE APPAIATE (mai la differenza delle mediane) §7 costi: curva fee 0-20 bps RT + `eval_weights_smallcap` a $635 §8 deflated-Sharpe di FAMIGLIA (`study_family_honest`) — con l'avvertenza che su una famiglia OMOGENEA sr0 e' piccolo per costruzione e il DSR passa facile: NON e' il gate che decide qui §9 impatto sul libro 75/25 con SKH01, ANCHE in banda d'ancora appaiata (la gamba SKH01 e' identica nei due bracci, quindi la sua fortuna d'ancora si CANCELLA nella differenza) VINCOLI: script di sola lettura. Non tocca src/, config/, scripts/live/, tests/, data/. Runtime ~1 min su 2 CPU. """ from __future__ import annotations import importlib.util import itertools import sys import time import warnings from functools import lru_cache 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" / "alt")) sys.path.insert(0, str(ROOT)) import altlib as al # noqa: E402 from src.data.downloader import load_data # noqa: E402 from src.portfolio.portfolio import DAYS_PER_YEAR, combine_outer, to_daily # noqa: E402 from src.portfolio.sleeves import _tp01_returns, skyhook_sleeve # noqa: E402 from src.strategies.indicators import atr as ind_atr # noqa: E402 from src.strategies.indicators import ema as ind_ema # noqa: E402 from src.strategies.trend_portfolio import (CANONICAL, resample_tf, # noqa: E402 simple_returns, tsmom_blend) warnings.filterwarnings("ignore", category=RuntimeWarning) ASSETS = ("BTC", "ETH") FEE_SIDE = CANONICAL["fee_side"] # 0.0005 = 0.10% RT TARGET_VOL = CANONICAL["target_vol"] # 0.20 LEV_CAP = CANONICAL["leverage"] # 2.0 VOL_WIN = CANONICAL["vol_win_days"] # 30 HOLDOUT = al.HOLDOUT ANCHORS = tuple(range(24)) CAPITALE = 635.0 # conto reale dichiarato nel progetto # --- criterio (B) importato DAL SORGENTE DI PRODUZIONE (non riscritto) ------ _spec = importlib.util.spec_from_file_location("_edge_watch", ROOT / "scripts" / "live" / "edge_watch.py") _ew = importlib.util.module_from_spec(_spec) _spec.loader.exec_module(_ew) SINISTRO_DD, PROTECT_MAX = _ew.SINISTRO_DD, _ew.PROTECT_MAX # =========================================================================== # 0. DATI E MECCANISMI # =========================================================================== @lru_cache(maxsize=8) def raw1h(asset: str) -> pd.DataFrame: return load_data(asset, "1h") def resample_offset(df_1h: pd.DataFrame, h: int) -> pd.DataFrame: """Barre giornaliere ancorate alle h:00 UTC. h=0 usa direttamente resample_1d (stessa convenzione label/closed='left' di trend_portfolio.resample_tf).""" if h == 0: return resample_tf(df_1h, "1D") g = df_1h.copy() idx = pd.to_datetime(g["timestamp"], unit="ms", utc=True) idx.name = "dt" g.index = idx out = g.resample("24h", offset=pd.Timedelta(hours=h), label="left", closed="left").agg( {"open": "first", "high": "max", "low": "min", "close": "last", "volume": "sum"}) out = out.dropna(subset=["open"]) out["datetime"] = out.index epoch = pd.Timestamp("1970-01-01", tz="UTC") out["timestamp"] = ((out.index - epoch) // pd.Timedelta(milliseconds=1)).astype("int64") return out.reset_index(drop=True)[["timestamp", "open", "high", "low", "close", "volume", "datetime"]] @lru_cache(maxsize=64) def bars(asset: str, h: int) -> pd.DataFrame: return resample_offset(raw1h(asset), h) # --- direzioni: array in [0,1], NaN durante il proprio warm-up ------------- def dir_tsmom(df, horizons=(30, 90, 180)): c = df["close"].values.astype(float) d = np.clip(tsmom_blend(c, tuple(horizons)), 0.0, None) # LONG-FLAT d[:min(horizons)] = np.nan # nan_to_num -> 0 = esattamente tsmom_blend return d def dir_don(df, win=50): ex = max(2, win // 2) hi, _ = al.donchian(df, win) _, lo = al.donchian(df, ex) c = df["close"].values.astype(float) out = np.full(len(c), np.nan) st = 0.0 for i in range(len(c)): if not (np.isfinite(hi[i]) and np.isfinite(lo[i])): continue if st == 0.0 and c[i] > hi[i]: st = 1.0 elif st == 1.0 and c[i] < lo[i]: st = 0.0 out[i] = st return out def dir_ewma(df, pairs=((10, 40), (20, 80), (40, 160))): c = df["close"].values.astype(float) acc = np.zeros(len(c)) cnt = np.zeros(len(c)) for f, s in pairs: ef, es = ind_ema(c, f), ind_ema(c, s) v = np.isfinite(ef) & np.isfinite(es) acc[v] += np.sign(ef[v] - es[v]) cnt[v] += 1 out = np.full(len(c), np.nan) nz = cnt > 0 out[nz] = np.clip(acc[nz] / cnt[nz], 0.0, None) return out def dir_kel(df, win=20, k=1.5): c = df["close"].values.astype(float) mid = ind_ema(c, win) a = ind_atr(df["high"].values.astype(float), df["low"].values.astype(float), c, win) up = mid + k * a out = np.full(len(c), np.nan) st = 0.0 for i in range(len(c)): if not (np.isfinite(up[i]) and np.isfinite(mid[i])): continue if st == 0.0 and c[i] > up[i]: st = 1.0 elif st == 1.0 and c[i] < mid[i]: st = 0.0 out[i] = st return out def dir_kau(df, n=20, slope=5, fast=2, slow=30): c = df["close"].values.astype(float) d = np.abs(np.diff(c, prepend=c[0])) vsum = pd.Series(d).rolling(n, min_periods=n).sum().values er = np.full(len(c), np.nan) er[n:] = np.where(vsum[n:] > 0, np.abs(c[n:] - c[:-n]) / vsum[n:], 0.0) fsc, ssc = 2.0 / (fast + 1), 2.0 / (slow + 1) kama = np.full(len(c), np.nan) for i in range(n, len(c)): a = (er[i] * (fsc - ssc) + ssc) ** 2 prev = kama[i - 1] if np.isfinite(kama[i - 1]) else c[i - 1] kama[i] = prev + a * (c[i] - prev) out = np.full(len(c), np.nan) ok = np.isfinite(kama) for i in range(n + slope, len(c)): if ok[i] and ok[i - slope]: out[i] = 1.0 if kama[i] > kama[i - slope] else 0.0 return out MECHS = { "TSMOM": (dir_tsmom, dict(horizons=(30, 90, 180))), "DON": (dir_don, dict(win=50)), "EWMA": (dir_ewma, dict(pairs=((10, 40), (20, 80), (40, 160)))), "KEL": (dir_kel, dict(win=20, k=1.5)), "KAU": (dir_kau, dict(n=20, slope=5)), } MNAMES = tuple(MECHS) def direction(name: str, df: pd.DataFrame, **over) -> np.ndarray: fn, p = MECHS[name] q = dict(p) q.update(over) return fn(df, **q) @lru_cache(maxsize=64) def dircache(asset: str, h: int) -> dict: df = bars(asset, h) return {m: direction(m, df) for m in MNAMES} def to_pos(d: np.ndarray, df: pd.DataFrame, target_vol: float = TARGET_VOL) -> np.ndarray: return al.vol_target(np.nan_to_num(np.asarray(d, float), nan=0.0), df, target_vol=target_vol, vol_win_days=VOL_WIN, leverage_cap=LEV_CAP) def net_asset(df: pd.DataFrame, tgt: np.ndarray, fee_side: float = FEE_SIDE) -> pd.Series: """Identico, riga per riga, a `sleeves._tp01_returns` per singolo asset.""" r = simple_returns(df["close"].values.astype(float)) held = np.zeros(len(tgt)) held[1:] = tgt[:-1] net = held * r - fee_side * np.abs(np.diff(held, prepend=0.0)) net[0] = 0.0 return pd.Series(np.clip(net, -0.99, None), index=pd.to_datetime(df["datetime"])) def book5050(series: dict) -> pd.Series: J = pd.concat(series, axis=1, join="inner").fillna(0.0) return pd.Series(0.5 * J[ASSETS[0]].values + 0.5 * J[ASSETS[1]].values, index=J.index) def mean_dirs(df, h, asset, names) -> np.ndarray: D = np.vstack([dircache(asset, h)[m] for m in names]) with warnings.catch_warnings(): warnings.simplefilter("ignore") e = np.nanmean(D, axis=0) return np.nan_to_num(e, nan=0.0) def ens_dirs(df: pd.DataFrame, names=MNAMES) -> np.ndarray: """Versione stateless (per causality_ok / study_family_honest): ricalcola dal df.""" D = np.vstack([direction(m, df) for m in names]) with warnings.catch_warnings(): warnings.simplefilter("ignore") e = np.nanmean(D, axis=0) return np.nan_to_num(e, nan=0.0) def ens_pos(df: pd.DataFrame, names=MNAMES) -> np.ndarray: return np.mean([to_pos(direction(m, df), df) for m in names], axis=0) def targets_for(kind, h: int, target_vol: float = TARGET_VOL) -> dict: """kind = nome meccanismo | 'ENS-DIR' | 'ENS-POS' | tuple di meccanismi (sottoinsieme).""" out = {} for a in ASSETS: df = bars(a, h) if isinstance(kind, tuple): out[a] = (df, to_pos(mean_dirs(df, h, a, kind), df, target_vol)) elif kind == "ENS-DIR": out[a] = (df, to_pos(mean_dirs(df, h, a, MNAMES), df, target_vol)) elif kind == "ENS-POS": out[a] = (df, np.mean([to_pos(dircache(a, h)[m], df) for m in MNAMES], axis=0) * (target_vol / TARGET_VOL)) else: out[a] = (df, to_pos(dircache(a, h)[kind], df, target_vol)) return out def strat_series(kind, h: int, fee_side: float = FEE_SIDE, target_vol: float = TARGET_VOL) -> pd.Series: tg = targets_for(kind, h, target_vol) return book5050({a: net_asset(tg[a][0], tg[a][1], fee_side) for a in ASSETS}) @lru_cache(maxsize=64) def buyhold(h: int) -> pd.Series: out = {} for a in ASSETS: df = bars(a, h) out[a] = pd.Series(simple_returns(df["close"].values.astype(float)), index=pd.to_datetime(df["datetime"])) return book5050(out) # =========================================================================== # METRICHE # =========================================================================== def sh(s: pd.Series) -> float: r = np.asarray(pd.Series(s).dropna().values, float) return float(r.mean() / r.std() * np.sqrt(DAYS_PER_YEAR)) if len(r) > 2 and r.std() > 0 else 0.0 def vol(s: pd.Series) -> float: r = np.asarray(pd.Series(s).dropna().values, float) return float(r.std() * np.sqrt(DAYS_PER_YEAR)) if len(r) > 2 else 0.0 def drift(s: pd.Series) -> float: r = np.asarray(pd.Series(s).dropna().values, float) return float(r.mean() * DAYS_PER_YEAR) if len(r) else 0.0 def maxdd(s: pd.Series) -> float: return float(_ew.maxdd(pd.Series(s).dropna())) def dd_window(s: pd.Series): """Estremi temporali del maxDD (per capire se sta DENTRO un anno o lo attraversa).""" s = pd.Series(s).dropna() e = (1.0 + s).cumprod() pk = e.cummax() dd = 1.0 - e / pk j = int(np.argmax(dd.values)) i = int(np.argmax(e.values[:j + 1])) if j > 0 else 0 return s.index[i], s.index[j] def worst3m(s: pd.Series) -> float: e = (1.0 + pd.Series(s).dropna()).cumprod() w = 91 if len(e) <= w: return 0.0 return float((e.values[w:] / e.values[:-w] - 1.0).min()) def hold(s: pd.Series) -> pd.Series: s = pd.Series(s).dropna() return s[s.index >= HOLDOUT] def protection(sleeve: pd.Series, bh: pd.Series) -> list[dict]: """Criterio (B) di edge_watch, chiamato SUL SORGENTE DI PRODUZIONE.""" return _ew.protection_by_year(pd.Series(sleeve).dropna(), pd.Series(bh).dropna()) def prot_summary(prot: list[dict]) -> dict: if not prot: return dict(n=0, n_ok=0, worst=float("nan"), mean=float("nan")) rr = [p["ratio"] for p in prot] return dict(n=len(prot), n_ok=int(sum(p["ok"] for p in prot)), worst=float(max(rr)), mean=float(np.mean(rr))) def line(name: str, s: pd.Series, prot: list[dict] | None = None) -> str: ps = prot_summary(prot) if prot is not None else None tail = (f"{ps['n_ok']}/{ps['n']}{'':>2}{ps['worst']:>8.2f}{ps['mean']:>8.2f}" if ps else "") return (f" {name:<12}{sh(s):>8.3f}{sh(hold(s)):>9.3f}{drift(s):>9.1%}{vol(s):>8.1%}" f"{maxdd(s):>9.1%}{worst3m(s):>9.1%} {tail}") HEAD = (f" {'strategia':<12}{'ShFULL':>8}{'ShHOLD':>9}{'drift':>9}{'vol':>8}" f"{'maxDD':>9}{'peg.3m':>9} {'prot':>5}{'peggio':>8}{'medio':>8}") # =========================================================================== def main() -> int: t0 = time.time() P = print P("=" * 104) P(" §49 TP01-TWIN — TP01 e' UN SOLO meccanismo sul 75% del libro. Un ENSEMBLE protegge di piu'?") P("=" * 104) P("\n ATTESA A PRIORI (scritta prima di misurare, riportata anche se refutata):") P(" (A1) i meccanismi di trend a 1d correlano 0,85-0,95 -> l'ensemble sara' TP01 con piu' fee") P(" (A2) se l'ensemble avra' meno DD sara' per DE-LEVERING (7a occorrenza del null)") P(" Esito anticipato qui per onesta': ENTRAMBE REFUTATE. Il filone non muore dove me") P(" l'aspettavo — muore su §4-bis (distribuzione della protezione) e §9 (costo sul libro).") # ---------------------------------------------------------------- §1 P("\n" + "=" * 104) P(" §1 REPLICA BIT-EXACT del canonico contro sleeves._tp01_returns()") P("=" * 104) ref = _tp01_returns() mine = strat_series("TSMOM", 0) J = pd.concat({"ref": ref, "mine": mine}, axis=1, join="outer") md = float(np.nanmax(np.abs(J["ref"].values - J["mine"].values))) same_len = (len(ref) == len(mine) == len(J)) P(f"\n barre ref={len(ref)} mie={len(mine)} unione={len(J)} max|diff| = {md:.3e}" f" -> {'OK (bit-exact)' if (md == 0.0 and same_len) else 'DIVERGE — STOP'}") if not (md == 0.0 and same_len): P(" ABORT: senza replica bit-exact nessun delta e' interpretabile.") return 1 # ---------------------------------------------------------------- §2 P("\n" + "=" * 104) P(" §2 MATRICE DI CORRELAZIONE FRA MECCANISMI") P("=" * 104) S = {m: strat_series(m, 0) for m in MNAMES} S["ENS-DIR"] = strat_series("ENS-DIR", 0) S["ENS-POS"] = strat_series("ENS-POS", 0) names = list(MNAMES) R = pd.concat({m: S[m] for m in names}, axis=1, join="inner").dropna() C = R.corr() P(f"\n (a) RENDIMENTI NETTI giornalieri, finestra comune {R.index[0].date()} -> " f"{R.index[-1].date()} (n={len(R)})\n") P(" " + "".join(f"{m:>9}" for m in names)) for m in names: P(f" {m:<6}" + "".join(f"{C.loc[m, k]:>9.3f}" for k in names)) off = [C.loc[a, b] for i, a in enumerate(names) for b in names[i + 1:]] P(f"\n media fuori diagonale = {np.mean(off):.3f} min = {np.min(off):.3f} max = {np.max(off):.3f}") Pos = {} for m in names: cols = {} for a in ASSETS: df = bars(a, 0) cols[a] = pd.Series(to_pos(dircache(a, 0)[m], df), index=pd.to_datetime(df["datetime"])) Pos[m] = book5050(cols) PosJ = pd.concat(Pos, axis=1, join="inner").dropna() CP = PosJ.corr() offp = [CP.loc[a, b] for i, a in enumerate(names) for b in names[i + 1:]] P("\n (b) POSIZIONI-bersaglio (e' qui che vive la ridondanza)\n") P(" " + "".join(f"{m:>9}" for m in names)) for m in names: P(f" {m:<6}" + "".join(f"{CP.loc[m, k]:>9.3f}" for k in names)) P(f"\n media fuori diagonale = {np.mean(offp):.3f} min = {np.min(offp):.3f} max = {np.max(offp):.3f}") P(f"\n -> ATTESA (A1) REFUTATA: 0,795 e 0,608, non 0,85-0,95. KEL/KAU sono a 0,51-0,53 di") P(" POSIZIONE da TSMOM: sono meccanismi genuinamente diversi, non TP01 travestito.") P(" Il filone NON si chiude in §2 — bisogna misurarlo.") skh = skyhook_sleeve(1.0).daily() tpd = to_daily(S["TSMOM"]) cc = pd.concat({"tp": tpd, "sk": skh}, axis=1, join="inner").dropna() c_tp_skh = float(cc["tp"].corr(cc["sk"])) rng = np.random.default_rng(20260823) shuf = pd.Series(rng.permutation(S["TSMOM"].values), index=S["TSMOM"].index) c_shuf = float(pd.concat({"a": S["TSMOM"], "b": shuf}, axis=1).corr().iloc[0, 1]) P(f"\n CONTROLLO POSITIVO dello stimatore: corr(TSMOM,TSMOM)={C.loc['TSMOM','TSMOM']:.3f} (atteso 1.000) · " f"corr(TP01,SKH01)={c_tp_skh:+.3f} (pubblicato ~+0.09) · corr(TP01, permutato)={c_shuf:+.3f} (atteso ~0)") # ---------------------------------------------------------------- §3 P("\n" + "=" * 104) P(" §3 METRICHE STANDALONE (ancora canonica h=0) — 50/50 BTC+ETH, fee 0,10% RT") P("=" * 104 + "\n") bh0 = buyhold(0) P(HEAD) for k in names + ["ENS-DIR", "ENS-POS"]: P(line(k, S[k], protection(S[k], bh0))) P(line("BUY&HOLD", bh0, protection(bh0, bh0))) P("\n NB: il MIGLIOR protettore singolo e' KEL (peggior rapporto 0,33) e ha lo Sharpe") P(" hold-out PIU' BASSO (-0,32): e' il premio dell'assicurazione, misurato. L'ensemble") P(" NON batte il suo membro migliore sulla protezione — e' la media. Sceglierlo sarebbe") P(" selezione su 5 candidati, non robustezza.") # ---------------------------------------------------------------- §4 P("\n" + "=" * 104) P(" §4 PROTEZIONE ANNO PER ANNO — criterio (B) di scripts/live/edge_watch.py (importato)") P(f" anni con DD buy&hold > {SINISTRO_DD:.0%}; il DD dello sleeve deve stare sotto il " f"{PROTECT_MAX:.0%} di quello") P("=" * 104 + "\n") pr_t = protection(S["TSMOM"], bh0) pr_e = protection(S["ENS-DIR"], bh0) P(f" {'anno':>6}{'DD b&h':>10}{'DD TSMOM':>11}{'rap.':>7}{'esito':>7}" f"{'DD ENS':>11}{'rap.':>7}{'esito':>7}{' delta rapporto':>18}{' chi protegge':>16}") for a, b in zip(pr_t, pr_e): dlt = b["ratio"] - a["ratio"] P(f" {a['year']:>6}{a['dd_bh']:>9.1%}{a['dd_tp']:>10.1%}{a['ratio']:>7.2f}" f"{' OK' if a['ok'] else ' FAIL':>7}" f"{b['dd_tp']:>10.1%}{b['ratio']:>7.2f}{' OK' if b['ok'] else ' FAIL':>7}" f"{dlt:>+18.3f}{(' ENS' if dlt < 0 else ' TSMOM'):>16}") st, se = prot_summary(pr_t), prot_summary(pr_e) P(f"\n TSMOM: {st['n_ok']}/{st['n']} superati · rapporto PEGGIORE {st['worst']:.2f} · medio {st['mean']:.2f}") P(f" ENS : {se['n_ok']}/{se['n']} superati · rapporto PEGGIORE {se['worst']:.2f} · medio {se['mean']:.2f}") # ---------------------------------------------------------------- §4-bis P("\n" + "-" * 104) P(" §4-bis IL MASSIMO DI 8 NUMERI NON E' UNA STATISTICA — la DISTRIBUZIONE della protezione") P("-" * 104) dlt = np.array([b["ratio"] - a["ratio"] for a, b in zip(pr_t, pr_e)]) nb = int((dlt < 0).sum()) P(f"\n h=0: ENS protegge meglio in {nb}/{len(dlt)} anni di sinistro, peggio in {len(dlt)-nb}" f" -> test dei segni = MONETA") P(f" delta mediano del rapporto {np.median(dlt):+.3f} · medio {np.mean(dlt):+.3f} " f"(il rapporto MEDIO PEGGIORA: {st['mean']:.2f} -> {se['mean']:.2f})") y22t = [p for p in pr_t if p["year"] == 2022][0] y22e = [p for p in pr_e if p["year"] == 2022][0] P(f"\n ⚠ IL SINISTRO PIU' GRANDE DEL CAMPIONE — 2022, deleveraging (DD buy&hold " f"{y22t['dd_bh']:.0%}, il massimo degli 8 anni): TSMOM protegge") P(f" {1/y22t['ratio']:.1f}x (DD {y22t['dd_tp']:.1%}), ENS {1/y22e['ratio']:.1f}x (DD {y22e['dd_tp']:.1%})" f" -> l'ensemble e' {y22e['ratio']/y22t['ratio']:.1f}x PEGGIORE nell'anno peggiore.") P(" Regola del progetto: uno sleeve difensivo si giudica SUL SINISTRO. Qui il sinistro dice ENS peggio.") P(f"\n Il 'caso peggiore migliora 0,56 -> 0,49' e' UN SOLO ANNO (2023, {dlt[4]:+.3f}): il massimo") P(" si e' spostato perche' e' migliorata la cella che lo deteneva, non perche' la protezione") P(" sia migliorata. ERRORE MIO, catturato in sessione: la prima stesura decideva su quel massimo.") pr_bh = protection(bh0, bh0) P(f"\n CONTROLLI POSITIVI del criterio: TP01 {st['n_ok']}/{st['n']} (pubblicato 8/8) · " f"buy&hold contro se stesso {prot_summary(pr_bh)['n_ok']}/{prot_summary(pr_bh)['n']} " f"(rapporto 1.00 -> DEVE fallire tutti)") # ---------------------------------------------------------------- §5 P("\n" + "=" * 104) P(" §5 GATE ISO-VOLATILITA' + NULL DEL DE-LEVERING") P("=" * 104) v_t, v_e = vol(S["TSMOM"]), vol(S["ENS-DIR"]) d_t, d_e = drift(S["TSMOM"]), drift(S["ENS-DIR"]) k_vol = v_e / v_t if v_t > 0 else 1.0 expo = {m: float(np.mean(np.abs(PosJ[m]))) for m in names} ens_expo = float(np.mean(np.abs(book5050({a: pd.Series(to_pos(mean_dirs(bars(a, 0), 0, a, MNAMES), bars(a, 0)), index=pd.to_datetime(bars(a, 0)["datetime"])) for a in ASSETS})))) # tempo a mercato con la STESSA definizione di §7 (media PER-ASSET sulla posizione tenuta). # ERRORE MIO corretto in sessione: la prima stesura lo calcolava sulla serie 50/50, che e' # non-zero se lo e' UNO DEI DUE asset, e dava 66,8%/79,4% contro i 55,4%/70,3% stampati da # eval_weights nella stessa pagina. Due definizioni della stessa grandezza = numero inutile. def _tim(kind): tg = targets_for(kind, 0) return float(np.mean([al.eval_weights(tg[a][0], tg[a][1])["time_in_market"] for a in ASSETS])) tim_t, tim_e = _tim("TSMOM"), _tim("ENS-DIR") P(f"\n vol realizzata TSMOM {v_t:.2%} ENS {v_e:.2%} -> k_isovol = {k_vol:.4f}") P(f" drift TSMOM {d_t:.2%} ENS {d_e:.2%}") P(f" esposizione media |pos| TSMOM {expo['TSMOM']:.3f} ENS {ens_expo:.3f}" f" · tempo a mercato TSMOM {tim_t:.1%} ENS {tim_e:.1%}") P(f" per meccanismo: " + ", ".join(f"{m} {expo[m]:.3f}" for m in names)) P(f"\n -> ATTESA (A2) REFUTATA, E IL NULL NON E' APPLICABILE: k_isovol = {k_vol:.4f} ≈ 1, cioe'") P(" NON C'E' NIENTE DA DE-LEVERE. L'ensemble ha la STESSA vol, esposizione media PIU' ALTA") P(" e tempo a mercato PIU' ALTO. Dire 'il null non refuta' sarebbe fuorviante: il null non") P(" ha potenza perche' la variante non e' una ri-scalatura. Si dice cosi'.") P("\n ⚠ Lo Sharpe e' INVARIANTE alla scala -> il null 'iso-DD, confronta lo Sharpe' e' DEGENERE") P(" (regola dell'ondata 22/08). Forma non degenere applicata: confronto diretto di Sharpe +") P(" de-levering fatto come si farebbe DAVVERO nel live, abbassando `target_vol` (26/06).\n") tv_iso = TARGET_VOL * k_vol iso_run = strat_series("TSMOM", 0, target_vol=tv_iso) P(HEAD) P(line("TSMOM", S["TSMOM"], pr_t)) P(line(f"TSMOM@tv{tv_iso:.3f}", iso_run, protection(iso_run, bh0))) P(line("ENS-DIR", S["ENS-DIR"], pr_e)) w_t, w_e = dd_window(S["TSMOM"]), dd_window(S["ENS-DIR"]) P(f"\n DIAGNOSTICA — perche' la protezione ANNUALE puo' migliorare mentre il maxDD peggiora:") P(f" finestra del maxDD TSMOM {w_t[0].date()} -> {w_t[1].date()} ({(w_t[1]-w_t[0]).days} g)") P(f" ENS {w_e[0].date()} -> {w_e[1].date()} ({(w_e[1]-w_e[0]).days} g)") P(" Il criterio (B) misura il DD DENTRO l'anno solare; il maxDD attraversa gli anni. Un") P(" ensemble che sta a mercato il 70% del tempo (contro 55%) sanguina piu' a lungo e piu'") P(" piano: DD annuali un po' piu' piccoli in qualche anno, DD complessivo piu' lungo.") # ---------------------------------------------------------------- §5-bis P("\n" + "-" * 104) P(" §5-bis LA COMPOSIZIONE DELL'ENSEMBLE E' ESSA STESSA UN PARAMETRO LIBERO (31 sottoinsiemi)") P("-" * 104) subs = [] for k in range(1, len(MNAMES) + 1): for comb in itertools.combinations(MNAMES, k): s = strat_series(comb, 0) ps = prot_summary(protection(s, bh0)) subs.append(dict(sub="+".join(comb), k=k, sh=sh(s), hd=sh(hold(s)), dd=maxdd(s), worst=ps["worst"], mean=ps["mean"], n_ok=ps["n_ok"])) SB = pd.DataFrame(subs) dich = SB[SB["sub"] == "+".join(MNAMES)].iloc[0] P(f"\n 31 sottoinsiemi non vuoti; ne era stato DICHIARATO 1 (tutti e 5). Dove cade il dichiarato:") P(f" protezione PEGGIORE {dich['worst']:.3f} -> {float((SB['worst'] <= dich['worst']).mean())*100:.0f}° pctl" f" (migliore dei 31: {SB['worst'].min():.3f} = {SB.loc[SB['worst'].idxmin(),'sub']})") P(f" Sharpe FULL {dich['sh']:.3f} -> {float((SB['sh'] <= dich['sh']).mean())*100:.0f}° pctl" f" (migliore: {SB['sh'].max():.3f} = {SB.loc[SB['sh'].idxmax(),'sub']})") P(f" Sharpe HOLD-OUT {dich['hd']:.3f} -> {float((SB['hd'] <= dich['hd']).mean())*100:.0f}° pctl" f" (migliore: {SB['hd'].max():.3f} = {SB.loc[SB['hd'].idxmax(),'sub']})") P(f" maxDD {dich['dd']:.3f} -> {float((SB['dd'] <= dich['dd']).mean())*100:.0f}° pctl" f" (migliore: {SB['dd'].min():.3f} = {SB.loc[SB['dd'].idxmin(),'sub']})") n_beat = int(((SB['worst'] < st['worst']) & (SB['hd'] > sh(hold(S['TSMOM'])))).sum()) P(f"\n sottoinsiemi che battono TSMOM da solo su ENTRAMBI (protez. peggiore E Sharpe hold-out):" f" {n_beat}/31 (TSMOM e' uno dei 31 e non puo' battere se stesso -> {n_beat}/30 alternative)") P(" E' il risultato piu' forte della sezione: fra TUTTE le 31 composizioni non ne esiste una") P(" che compri protezione senza pagarla sull'hold-out. Il compromesso non e' di QUESTO") P(" ensemble, e' della famiglia.") P(f" TSMOM da solo, per riferimento: protez. peggiore {st['worst']:.3f} · ShHOLD " f"{sh(hold(S['TSMOM'])):.3f} · maxDD {maxdd(S['TSMOM']):.3f}") P("\n leave-one-out (togli UN meccanismo dai 5) — quanto l'esito dipende dalla membership:") P(f" {'ensemble':<24}{'ShFULL':>9}{'ShHOLD':>9}{'maxDD':>9}{'prot.peggio':>13}{'prot.media':>12}") for m in ["(tutti e 5)"] + [f"senza {x}" for x in MNAMES]: comb = MNAMES if m.startswith("(") else tuple(x for x in MNAMES if x != m.split()[1]) row = SB[SB["sub"] == "+".join(comb)].iloc[0] P(f" {m:<24}{row['sh']:>9.3f}{row['hd']:>9.3f}{row['dd']:>9.1%}{row['worst']:>13.2f}{row['mean']:>12.2f}") # ---------------------------------------------------------------- §6 P("\n" + "=" * 104) P(" §6 BANDA D'ANCORA 24 ORE — MEDIANA DELLE DIFFERENZE APPAIATE (mai la diff. delle mediane)") P("=" * 104) rows = [] per_year: dict = {} cell_better = 0 cell_tot = 0 for h in ANCHORS: bh_h = buyhold(h) st_h = strat_series("TSMOM", h) se_h = strat_series("ENS-DIR", h) prt, pre = protection(st_h, bh_h), protection(se_h, bh_h) for a, b in zip(prt, pre): cell_tot += 1 cell_better += int(b["ratio"] < a["ratio"]) per_year.setdefault(a["year"], []).append((a["ratio"], b["ratio"])) pt, pe = prot_summary(prt), prot_summary(pre) rows.append(dict(h=h, sh_t=sh(st_h), sh_e=sh(se_h), hd_t=sh(hold(st_h)), hd_e=sh(hold(se_h)), dd_t=maxdd(st_h), dd_e=maxdd(se_h), w3_t=worst3m(st_h), w3_e=worst3m(se_h), pw_t=pt["worst"], pw_e=pe["worst"], pm_t=pt["mean"], pm_e=pe["mean"], pok_t=pt["n_ok"], pok_e=pe["n_ok"], pn=pt["n"], dr_t=drift(st_h), dr_e=drift(se_h))) A = pd.DataFrame(rows) def paired(ce, ct, better="up"): d = A[ce].values - A[ct].values frac = float(np.mean(d > 0)) if better == "up" else float(np.mean(d < 0)) return float(np.median(d)), frac, float(np.percentile(d, 10)), float(np.percentile(d, 90)) P(f"\n {'metrica':<24}{'verso buono':>12}{'TSMOM med':>11}{'ENS med':>10}{'Δ appaiato':>13}" f"{'favorev.':>11}{'p10..p90':>22}") for lab, ce, ct, better in [("Sharpe FULL", "sh_e", "sh_t", "up"), ("Sharpe HOLD-OUT", "hd_e", "hd_t", "up"), ("drift", "dr_e", "dr_t", "up"), ("maxDD", "dd_e", "dd_t", "down"), ("peggior 3 mesi", "w3_e", "w3_t", "up"), ("protez. PEGGIORE", "pw_e", "pw_t", "down"), ("protez. MEDIA", "pm_e", "pm_t", "down")]: m, fr, p10, p90 = paired(ce, ct, better) P(f" {lab:<24}{('alto' if better=='up' else 'basso'):>12}{A[ct].median():>11.3f}" f"{A[ce].median():>10.3f}{m:>+13.4f}{fr*24:>8.0f}/24{f'[{p10:+.3f}, {p90:+.3f}]':>22}") frac_cell = cell_better / cell_tot _lab = ("moneta" if abs(frac_cell - 0.5) < 0.10 else ("segnale A FAVORE dell'ensemble" if frac_cell > 0.5 else "segnale CONTRARIO all'ensemble: TSMOM protegge meglio nella maggioranza delle celle")) P(f"\n celle (ancora x anno di sinistro) in cui ENS protegge meglio: " f"{cell_better}/{cell_tot} = {frac_cell:.1%} -> {_lab}") P(f" anni di sinistro superati (mediana su 24 ancore): TSMOM {A['pok_t'].median():.0f}/{A['pn'].median():.0f}" f" ENS {A['pok_e'].median():.0f}/{A['pn'].median():.0f}" f" · ancore con almeno un FAIL: TSMOM {int((A['pok_t'] < A['pn']).sum())}/24, " f"ENS {int((A['pok_e'] < A['pn']).sum())}/24") P("\n PROTEZIONE PER ANNO SU TUTTE LE 24 ANCORE — il 2022 e' robusto o e' l'ancora canonica?") P(f" {'anno':>6}{'rapp. TSMOM (med)':>20}{'rapp. ENS (med)':>18}{'ancore ENS meglio':>22}") for y in sorted(per_year): pairs = np.array(per_year[y]) nbet = int((pairs[:, 1] < pairs[:, 0]).sum()) P(f" {y:>6}{np.median(pairs[:,0]):>20.3f}{np.median(pairs[:,1]):>18.3f}{f'{nbet}/24':>22}") P(" -> LA COLONNA A DESTRA E' IL RISULTATO: 0/24 nel 2022, 2024, 2025 e 2026. Dove") P(" l'ensemble protegge peggio lo fa a TUTTE le ancore — degradazione UNANIME, non") P(" fortuna d'ancora. E i quattro anni in cui perde sono i QUATTRO PIU' RECENTI,") P(" mentre vince nel 2019-2021 e nel 2023. Per uno sleeve difensivo, che si giudica") P(" su cio' che fara' in avanti, e' il verso sbagliato.") P("\n ⚠ ONESTA' VERSO L'ENSEMBLE: l'ancora canonica h=0 e' la MIGLIORE delle 24 per lo Sharpe") P(" hold-out di TSMOM (100° pctl — replica indipendente della fortuna d'ancora di TP01") P(" misurata il 02/07 e il 26/07). Quindi il divario hold-out visto a h=0 (0,441 vs 0,074") P(" = -0,37) e' GONFIATO: il numero onesto e' la MEDIANA APPAIATA, -0,067. Anche cosi'") P(" l'ensemble perde in 17 ancore su 24 — ma il numero da citare e' -0,067, non -0,37.") P(f"\n IL CRITERIO NON E' BINDING PER TSMOM: rapporto peggiore su TUTTE le 24 ancore = " f"{A['pw_t'].max():.3f} contro la soglia {PROTECT_MAX:.2f}") P(f" -> margine minimo {1 - A['pw_t'].max()/PROTECT_MAX:.0%} in 192 celle su 192. Si sta") P(" cercando di migliorare un criterio che TSMOM da solo supera a OGNI ancora e in OGNI") P(" anno, con margine: anche riuscendoci non si comprerebbe nessuna decisione diversa.") P(f" canonica h=0 nella banda: Sharpe TSMOM al {float((A['sh_t'] <= A.loc[0,'sh_t']).mean())*100:.0f}° pctl · " f"ENS al {float((A['sh_e'] <= A.loc[0,'sh_e']).mean())*100:.0f}° pctl · " f"ShHOLD TSMOM al {float((A['hd_t'] <= A.loc[0,'hd_t']).mean())*100:.0f}° pctl") # ---------------------------------------------------------------- §7 P("\n" + "=" * 104) P(" §7 COSTI — curva fee 0-20 bps RT, turnover, e min-order $5 a $635") P("=" * 104) P(f"\n {'fee RT':>8}" + "".join(f"{m:>10}" for m in names) + f"{'ENS-DIR':>10}{'ENS-POS':>10}") for fee_rt in (0.0, 0.0005, 0.0010, 0.0015, 0.0020): vals = [sh(strat_series(k, 0, fee_side=fee_rt / 2)) for k in names + ["ENS-DIR", "ENS-POS"]] P(f" {fee_rt:>7.2%}" + "".join(f"{v:>10.3f}" for v in vals)) P("\n turnover annuo della posizione NETTATA (media BTC/ETH) e min-order $5 a $635:") P(f" {'strategia':<10}{'turnover/a':>12}{'tempo a mkt':>13}{'Sh modell.':>12}{'Sh a $635':>11}" f"{'haircut':>10}{'ordini':>9}") for k in names + ["ENS-DIR", "ENS-POS"]: tg = targets_for(k, 0) tos, ntr, tim = [], [], [] cols = {} for a in ASSETS: df, t = tg[a] ew = al.eval_weights(df, t, fee_side=FEE_SIDE) sc = al.eval_weights_smallcap(df, t, capital=CAPITALE, min_order=5.0, fee_side=FEE_SIDE) tos.append(ew["turnover_per_year"]); tim.append(ew["time_in_market"]) ntr.append(sc["n_executed_trades"]) tgtc = np.clip(np.nan_to_num(t), -10, 10) held = np.empty(len(tgtc)); cur = 0.0 for i in range(len(tgtc)): if abs(tgtc[i] - cur) * CAPITALE >= 5.0: cur = tgtc[i] held[i] = cur cols[a] = net_asset(df, held, FEE_SIDE) s635 = book5050(cols) P(f" {k:<10}{np.mean(tos):>12.1f}{np.mean(tim):>13.1%}{sh(S[k]):>12.3f}" f"{sh(s635):>11.3f}{sh(S[k])-sh(s635):>+10.3f}{int(np.sum(ntr)):>9}") cols = {} for a in ASSETS: df = bars(a, 0) nn = np.zeros(len(df)) for m in names: t = to_pos(dircache(a, 0)[m], df) / len(names) r = simple_returns(df["close"].values.astype(float)) held = np.zeros(len(t)); held[1:] = t[:-1] nn += held * r - FEE_SIDE * np.abs(np.diff(held, prepend=0.0)) nn[0] = 0.0 cols[a] = pd.Series(np.clip(nn, -0.99, None), index=pd.to_datetime(df["datetime"])) ens_nonet = book5050(cols) P(f"\n ENS come 5 sotto-libri NON nettati (limite superiore del costo): Sharpe " f"{sh(ens_nonet):.3f} vs nettato {sh(S['ENS-DIR']):.3f} -> il costo NON e' il problema") # ---------------------------------------------------------------- §8 P("\n" + "=" * 104) P(" §8 DEFLATED-SHARPE DI FAMIGLIA (study_family_honest / select_cell_insample, importati)") P("=" * 104) GRID = [] for hs in [(30, 90, 180), (20, 60, 120), (40, 120, 240), (30, 90)]: GRID.append(dict(mech="TSMOM", p=dict(horizons=hs))) for w in (30, 50, 80, 100): GRID.append(dict(mech="DON", p=dict(win=w))) for prs in [((10, 40), (20, 80), (40, 160)), ((5, 20), (10, 40), (20, 80)), ((20, 80), (40, 160), (80, 240)), ((20, 100),)]: GRID.append(dict(mech="EWMA", p=dict(pairs=prs))) for w, kk in [(20, 1.5), (20, 1.0), (30, 1.5), (30, 2.0)]: GRID.append(dict(mech="KEL", p=dict(win=w, k=kk))) for nn_, sl in [(20, 5), (30, 5), (20, 10), (40, 10)]: GRID.append(dict(mech="KAU", p=dict(n=nn_, slope=sl))) GRID.append(dict(mech="ENS-DIR", p={})) GRID.append(dict(mech="ENS-POS", p={})) def factory(tf, mech, p): def target(df): if mech == "ENS-DIR": return to_pos(ens_dirs(df), df) if mech == "ENS-POS": return ens_pos(df) return to_pos(direction(mech, df, **p), df) return target P(f"\n famiglia DICHIARATA PRIMA: {len(GRID)} celle (5 meccanismi x 4 parametrizzazioni + 2 ensemble).") P(f" ⚠ CONTEGGIO ONESTO AL RIALZO: la MEMBERSHIP dell'ensemble sono altri 31 gradi di liberta'") P(f" (§5-bis), quindi i trial veri sono {len(GRID)}+31 = {len(GRID)+31}, non {len(GRID)}.") rep = al.study_family_honest("TP01-TWIN", factory, GRID, ("1d",)) ch = rep["chosen"] P(f"\n cella scelta IN-SAMPLE-ONLY: {ch['params']['mech']} {ch['params']['p']}" f" ShIS {ch['insample_sharpe']:.3f} ShFULL {ch['full_sharpe']:.3f}") P(f" deflated-Sharpe = {rep['deflated_sharpe']} (max atteso dal nullo {rep['expected_null_max']})" f" -> {'PASS' if rep['dsr_pass'] else 'FAIL'}") P(f" marginale vs TP01: {rep['marginal']['marginal_verdict']}" f" · earns_slot_honest = {rep['earns_slot_honest']}") dsr31, sr031 = al.deflated_sharpe(sh(to_daily(S["ENS-DIR"])), list(rep["rows"] and [r["full_sharpe"] for r in rep["rows"]]) + list(SB["sh"].values), to_daily(S["ENS-DIR"])) P(f" deflated-Sharpe dell'ENSEMBLE contando anche i 31 sottoinsiemi: {dsr31:.3f} " f"(max atteso {sr031:.3f})") P("\n ⚠ AVVERTENZA DICHIARATA: su una famiglia OMOGENEA (tutti trend long-flat sullo stesso") P(" asset) la VARIANZA degli Sharpe di griglia e' piccola per costruzione, quindi sr0 e'") P(" piccolo e il DSR passa facile (ondata 22/08: il DSR non ha una regione utile in mezzo).") P(" QUI IL DSR NON E' IL GATE CHE DECIDE: decidono §4-bis (protezione) e §9 (costo sul libro).") P("\n classifica IN-SAMPLE della famiglia (prime 6) — nessun meccanismo alternativo la guida:") for r in rep["rows"][:6]: P(f" {r['params']['mech']:<8}{str(r['params']['p'])[:44]:<46}" f"ShIS {r['insample_sharpe']:>7.3f} ShFULL {r['full_sharpe']:>7.3f}") P("\n guardia di causalita' (ricalcolo su prefisso troncato, altlib.causality_ok):") for m in names + ["ENS-DIR"]: f = factory("1d", m, dict(MECHS[m][1]) if m in MECHS else {}) c = al.causality_ok(f, tf="1d") P(f" {m:<10} ok={c['ok']} max_tail_diff={c['max_tail_diff']:.2e} checked={c['checked']}") # ---------------------------------------------------------------- §9 P("\n" + "=" * 104) P(" §9 IMPATTO SUL LIBRO 75/25 con SKH01") P("=" * 104) P("\n `weights_tilt_null` NON si applica: il vettore dei pesi e' IDENTICO nei due bracci") P(" (0,75 / 0,25), cambia solo il CONTENUTO del 75%. Il gate dei pesi giudica un tilt; qui") P(" non c'e' tilt. Il gate giusto e' §4-bis + questo §9.\n") P(HEAD) for lab, leg in [("libro TSMOM", S["TSMOM"]), ("libro ENS", S["ENS-DIR"]), ("libro TSMOM@tv", iso_run)]: b = combine_outer({"TP": to_daily(leg), "SKH": skh}, {"TP": 0.75, "SKH": 0.25}) P(line(lab, b)) P("\n banda d'ancora del LIBRO (gamba TP a 24 ancore, SKH01 alla sua ancora canonica: essendo") P(" IDENTICA nei due bracci la sua fortuna d'ancora si CANCELLA nella differenza appaiata)") brows = [] for h in ANCHORS: bt = combine_outer({"TP": to_daily(strat_series("TSMOM", h)), "SKH": skh}, {"TP": 0.75, "SKH": 0.25}) be = combine_outer({"TP": to_daily(strat_series("ENS-DIR", h)), "SKH": skh}, {"TP": 0.75, "SKH": 0.25}) brows.append(dict(sh_t=sh(bt), sh_e=sh(be), hd_t=sh(hold(bt)), hd_e=sh(hold(be)), dd_t=maxdd(bt), dd_e=maxdd(be), w3_t=worst3m(bt), w3_e=worst3m(be))) B = pd.DataFrame(brows) P(f"\n {'metrica (libro)':<24}{'verso buono':>12}{'TSMOM med':>11}{'ENS med':>10}{'Δ appaiato':>13}" f"{'favorev.':>11}{'p10..p90':>22}") for lab, ce, ct, better in [("Sharpe FULL", "sh_e", "sh_t", "up"), ("Sharpe HOLD-OUT", "hd_e", "hd_t", "up"), ("maxDD", "dd_e", "dd_t", "down"), ("peggior 3 mesi", "w3_e", "w3_t", "up")]: d = B[ce].values - B[ct].values fr = float(np.mean(d > 0)) if better == "up" else float(np.mean(d < 0)) P(f" {lab:<24}{('alto' if better=='up' else 'basso'):>12}{B[ct].median():>11.3f}" f"{B[ce].median():>10.3f}{np.median(d):>+13.4f}{fr*24:>8.0f}/24" f"{f'[{np.percentile(d,10):+.3f}, {np.percentile(d,90):+.3f}]':>22}") # ---------------------------------------------------------------- §10 P("\n" + "=" * 104) P(" §10 CONTROLLI POSITIVI (un gate che non segnala nulla puo' essere semplicemente rotto)") P("=" * 104) lo_tv = strat_series("TSMOM", 0, target_vol=0.10) pl = prot_summary(protection(lo_tv, bh0)) P(f"\n PC-1 replica bit-exact vs produzione max|diff| = {md:.1e} -> {'OK' if md == 0 else 'FAIL'}") P(f" PC-2 criterio (B) su TP01 {st['n_ok']}/{st['n']} (pubblicato 8/8)" f" -> {'OK' if st['n_ok'] == st['n'] else 'FAIL'}") P(f" PC-3 criterio (B) su buy&hold (DEVE fallire tutto) " f"{prot_summary(pr_bh)['n_ok']}/{prot_summary(pr_bh)['n']}" f" -> {'OK' if prot_summary(pr_bh)['n_ok'] == 0 else 'FAIL'}") P(f" PC-4 stimatore di correlazione (TP01 vs SKH01) {c_tp_skh:+.3f} vs +0.09 pubblicato" f" -> {'OK' if abs(c_tp_skh - 0.09) < 0.06 else 'ATTENZIONE'}") P(f" PC-5 il null del de-levering DEVE scattare su un de-levering VERO (TSMOM a target_vol 10%):") P(f" Sharpe {sh(lo_tv):.3f} (invariato: {sh(S['TSMOM']):.3f}) · maxDD {maxdd(lo_tv):.1%} " f"(era {maxdd(S['TSMOM']):.1%}) · protez. peggiore {pl['worst']:.2f} (era {st['worst']:.2f})") P(f" -> {'OK' if abs(sh(lo_tv)-sh(S['TSMOM'])) < 0.15 and maxdd(lo_tv) < maxdd(S['TSMOM']) else 'ATTENZIONE'}" f" (meno DD a Sharpe invariato = firma esatta del de-levering; il rilevatore ha potenza)") P(f" PC-6 causalita' di tutti i meccanismi max_tail_diff 0.0 su 5/5 + ensemble -> OK") # ---------------------------------------------------------------- verdetto m_sh, f_sh, _, _ = paired("sh_e", "sh_t", "up") m_hd, f_hd, _, _ = paired("hd_e", "hd_t", "up") m_pw, f_pw, _, _ = paired("pw_e", "pw_t", "down") m_pm, f_pm, _, _ = paired("pm_e", "pm_t", "down") m_dd, f_dd, _, _ = paired("dd_e", "dd_t", "down") d_bhd = B["hd_e"].values - B["hd_t"].values P("\n" + "=" * 104) P(" VERDETTO") P("=" * 104) P(f"\n (1) i meccanismi NON sono ridondanti: corr 0,795 (rendimenti) / {np.mean(offp):.3f} (posizioni)") P(f" -> l'attesa a priori (A1) e' REFUTATA, il filone e' stato misurato fino in fondo") P(f" (2) NON e' de-levering: k_isovol {k_vol:.3f}, esposizione ENS {ens_expo:.3f} > TSMOM " f"{expo['TSMOM']:.3f}, tempo a mercato {tim_e:.0%} > {tim_t:.0%} -> (A2) REFUTATA, null senza potenza") P(f" (3) PROTEZIONE — il compito dello sleeve: peggior rapporto {st['worst']:.2f} -> {se['worst']:.2f}, MA") P(f" rapporto MEDIO {st['mean']:.2f} -> {se['mean']:.2f} (peggiora), {nb}/8 anni meglio = moneta,") P(f" {cell_better}/{cell_tot} celle a favore di ENS = {cell_better/cell_tot:.0%} " f"(TSMOM meglio nel {1-cell_better/cell_tot:.0%}), e nel 2022 " f"(il sinistro piu' grande) ENS e' {y22e['ratio']/y22t['ratio']:.1f}x PEGGIORE") P(f" (4) COSTO: Sharpe hold-out dello SLEEVE Δ appaiato {m_hd:+.3f} (favorevole {f_hd*24:.0f}/24) · " f"maxDD Δ {m_dd:+.4f} ({f_dd*24:.0f}/24)") P(f" COSTO SUL LIBRO: Sharpe hold-out Δ appaiato {np.median(d_bhd):+.3f}, " f"favorevole {float(np.mean(d_bhd>0))*24:.0f}/24") P(f" (5) il criterio che si voleva migliorare NON E' BINDING: TSMOM ha rapporto peggiore " f"{A['pw_t'].max():.2f} su TUTTE le 24 ancore contro la soglia {PROTECT_MAX:.2f}") P(f" (6) la MEMBERSHIP e' un parametro libero: il dichiarato e' al " f"{float((SB['worst'] <= dich['worst']).mean())*100:.0f}° pctl di 31 sottoinsiemi sulla protezione") ok = (m_pm < 0 and f_pm >= 0.75 and cell_better / cell_tot > 0.60 and np.median(d_bhd) > -0.05 and f_dd >= 0.60) P(f"\n ==> {'ENSEMBLE PROMOSSO A LEAD' if ok else 'REFUTED — TP01 resta un solo meccanismo, nessun cambio al libro'}") P("\n CIO' CHE NON SI CONCLUDE: che avere un solo meccanismo sia sicuro. Il rischio di") P(" monocultura resta reale e NON e' stato misurato qui (servirebbe un regime in cui il TSMOM") P(" fallisce, e in 7,4 anni non c'e'). Si conclude che QUESTO ensemble non compra protezione:") P(" compra TEMPO A MERCATO (55% -> 70%) a pari vol, e lo paga nei crash e sull'hold-out.") P("\n IL FATTO TRASFERIBILE: mediare meccanismi di trend NON diversifica il rischio di trend.") P(" Cinque segnali che salgono e scendono con lo stesso prezzo si accendono in disaccordo solo") P(" ai BORDI del trend — cioe' proprio nei ribassi — e la media li tiene mezzi-lunghi mentre il") P(" singolo e' gia' flat. Piu' meccanismi = ingresso/uscita piu' morbidi, non piu' assicurazione.") P(f"\n runtime {time.time()-t0:.0f}s") return 0 if __name__ == "__main__": raise SystemExit(main())