#!/usr/bin/env python """r0822b_bocpd.py — IL FALSIFICATORE NOMINATO E MAI ESEGUITO del filone ADAPTIVE-HORIZON. CONTESTO (perche' questo file esiste) ------------------------------------- Il 22/08 `r0822_adaptive_horizon.py` ha scartato il TSMOM a orizzonte adattivo con una diagnosi netta: il "vincitore adattivo" aveva L_t **incollato al bordo il 100% del tempo** = un TSMOM a orizzonte COSTANTE di 20 giorni travestito (corr 1.000, dSh 0.00 in 8/8 anni e 0/12 ancore); isolando le 52 celle davvero adattive -> NEUTRAL, corr->TP01 0.90 = TP01 travestito. Ma quell'agente ha dichiarato **cosa lo smentirebbe** e non ha potuto eseguirlo: "Un rilevatore il cui L_t si muove DAVVERO e batte il miglior lookback COSTANTE (non TP01) in >=6/8 anni appaiati con corr->TP01 < 0.6. La mia famiglia copre 3 rilevatori a soglia/finestra fissa — un run-length posterior bayesiano (BOCPD/PELT vero) o un L scelto per walk-forward sul PnL realizzato non sono stati provati." Questo file esegue **quel** test. Oggi il filone e' chiuso su UNA famiglia (3 proxy a soglia fissa), non sul MECCANISMO. Una seconda chiusura, con rilevatori veri, rende il "no" definitivo. PERCHE' BOCPD E' STRUTTURALMENTE DIVERSO DAI 3 PROXY DEL PRIMO TENTATIVO ----------------------------------------------------------------------- I 3 proxy (CUSUM a soglia k, variance-ratio, Hurst) producevano una STATISTICA che veniva poi MAPPATA linearmente dentro [Lmin, Lmax]: la mappatura e' arbitraria, e infatti la statistica saturava contro il clip (bordo 100%). Un run-length posterior NON ha bisogno di mappatura: la lunghezza di run **e' gia' un lookback**, nelle stesse unita' (barre). "Quanto indietro posso guardare" = "da quanto tempo non cambia il regime" e' un'identita', non una scelta di scala. * BOCPD (Adams & MacKay 2007): posterior online su r_t con hazard costante e verosimiglianza Normale-InverseGamma coniugata (predittiva Student-t). Causale PER COSTRUZIONE: e' un filtro in avanti, ogni P(r_t | x_{1:t}) usa solo x_{1:t}. * OP/PELT causale: PELT e' un'accelerazione (pruning) dell'**optimal partitioning** F(t) = min_{s e' un asse] blend 2 (single L | triple L/3,L,2L — con L=90 il triple E' TP01) timeframe 2 (1d | 12h) [>=12h: sotto, costi+overfit dominano, 19/06] = 2*3*2*2*2*2 = **96 trial** B) controllo L COSTANTE 12 L x 2 blend x 2 TF = **48 trial** C) WF-PnL (L da walk-forward) 3 finestre x 2 blend x 2 TF = **12 trial** D) taratura della SCALA 8 penalita' OP + 5 hazard BOCPD = **13** (sezione 1-pre) ---------------------------------------------------------------- TOTALE DICHIARATO **169** ⚠️ (D) e' una scansione fatta guardando SOLO il numero di change-point rilevati — una proprieta' del RILEVATORE — e mai uno Sharpe: nessuna di quelle 13 configurazioni produce un rendimento. Si dichiara lo stesso, al rialzo, perche' ha ristretto la griglia di (A). Il suo esito e' gia' un risultato: a penalita' OP >= 5 l'eta' resta incollata al tetto della finestra il 44% del tempo, cioe' **l'artefatto del 22/08 riprodotto su un rilevatore vero** — e la griglia (A) usa solo la regione in cui il rilevatore rileva davvero (0.5/1.0/2.0 = 18-40 change-point l'anno). Costanti FISSATE A PRIORI (gradi di liberta' non cercati, dichiarati perche' esistono): Lmin=20g, Lmax=200g (contengono 30..180 del canonico e sono gli stessi del primo tentativo); BOCPD su rendimenti log STANDARDIZZATI da vol rolling causale 30g (il bersaglio e' il DRIFT, non la vol); prior NIG mu0=0, kappa0=1, alpha0=1, beta0=1; troncamento del posterior a Lmax barre; costo OP = mean-shift gaussiano su dati standardizzati, penalita' beta = k*log(t) causale. Il deflated-Sharpe e' riportato a N=169 (mia famiglia), N=281 (mia + le 112 del primo tentativo: **stessa domanda**, i trial si sommano) e N=449 (di SCREEN — lezione 22/08 del filone ORTHO: su BTC/ETH direzionale il massimo atteso dal rumore su 168 trial e' Sharpe 1.572, SOPRA il soffitto misurato ~1.3, quindi il conteggio di screen e' quello che decide). ORDINE DI ESECUZIONE (imposto dalla missione, ed e' l'ordine giusto) -------------------------------------------------------------------- La diagnostica che ha smascherato il primo tentativo va PER PRIMA: quanto si muove davvero L_t? Se sta al bordo, l'artefatto e' riprodotto e il filone si chiude di nuovo senza guardare i rendimenti. Solo se L_t si muove ha senso misurare se il movimento PAGA. Runtime atteso ~6-10 min su 2 core. Nessun file toccato fuori da questo. """ from __future__ import annotations import sys import time from pathlib import Path import numpy as np import pandas as pd from scipy.special import gammaln from scipy.stats import norm ROOT = Path("/opt/docker/PythagorasGoal") sys.path.insert(0, str(ROOT / "scripts" / "research" / "alt")) sys.path.insert(0, str(ROOT)) import altlib as A # noqa: E402 from src.strategies.trend_portfolio import CANONICAL, TrendPortfolio # noqa: E402 ASSETS = A.CERTIFIED HOLDOUT = A.HOLDOUT # ---- costanti FISSATE A PRIORI (dichiarate sopra, non cercate) ------------------------- LMIN_D, LMAX_D = 20, 200 # banda dell'orizzonte, IDENTICA al primo tentativo FIXED_H = (30, 90, 180) # == (L/3, L, 2L) con L=90 -> TP01 canonico TGT_VOL, LEV, VOLWIN = 0.20, 2.0, 30 NIG_MU0, NIG_KAP0, NIG_AL0, NIG_BE0 = 0.0, 1.0, 1.0, 1.0 STD_WIN_D = 30 # finestra della standardizzazione causale dei rendimenti TFS = ("1d", "12h") DETECTORS = ("bocpd", "op") PARAMS = {"bocpd": (50, 100, 250), "op": (0.5, 1.0, 2.0)} # TARATURA DELLA SCALA (sezione 1-pre): scansione ESPLORATIVA fatta guardando SOLO il numero di # change-point rilevati — una proprieta' del RILEVATORE — e mai lo Sharpe. Un rilevatore che non # rileva nulla non e' un rilevatore: a pen_k>=5 l'eta' resta incollata al tetto della finestra # (44% del tempo) = ESATTAMENTE l'artefatto del 22/08, riprodotto qui su un rilevatore vero. CAL_PEN = (0.02, 0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 5.0) CAL_LAM = (25, 50, 100, 250, 500) STATS = {"bocpd": ("mean", "map"), "op": ("age", "segmed")} MAPPINGS = ("direct", "inverse") BLENDS = ("single", "triple") CONST_L = (5, 7, 10, 15, 20, 30, 45, 60, 90, 120, 150, 200) WF_WIN_D = (180, 365, 730) N_DECLARED_A = len(DETECTORS) * 3 * 2 * len(MAPPINGS) * len(BLENDS) * len(TFS) # 96 N_DECLARED_B = len(CONST_L) * len(BLENDS) * len(TFS) # 48 N_DECLARED_C = len(WF_WIN_D) * len(BLENDS) * len(TFS) # 12 N_DECLARED_CAL = len(CAL_PEN) + len(CAL_LAM) # 13 (taratura) N_DECLARED = N_DECLARED_A + N_DECLARED_B + N_DECLARED_C + N_DECLARED_CAL # 169 N_PRIOR_FAMILY = 112 # trial dichiarati da r0822_adaptive_horizon (stessa domanda) N_SCREEN = 168 # pool di screen direzionale BTC/ETH dell'ondata (filone 12) # ======================================================================================= # RILEVATORI VERI # ======================================================================================= def _logret(c: np.ndarray) -> np.ndarray: r = np.zeros(len(c)) r[1:] = np.log(c[1:] / c[:-1]) return r def _std_returns(c: np.ndarray, bpd: int) -> np.ndarray: """Rendimenti log standardizzati da vol rolling CAUSALE (min_periods -> warmup a 0). Il bersaglio del rilevatore e' il cambio di DRIFT: senza standardizzare, un cambio di regime di VOLATILITA' dominerebbe la verosimiglianza e il rilevatore misurerebbe la vol.""" lr = _logret(c) w = max(5, STD_WIN_D * bpd) sig = pd.Series(lr).rolling(w, min_periods=max(5, w // 4)).std().values y = np.where((sig > 0) & np.isfinite(sig), lr / np.where(sig > 0, sig, 1.0), 0.0) return np.nan_to_num(np.clip(y, -8.0, 8.0)) def _student_logpdf(x: float, mu: np.ndarray, kap: np.ndarray, al: np.ndarray, be: np.ndarray) -> np.ndarray: """Predittiva Student-t della Normale-InverseGamma coniugata (Murphy 2007, eq. 99-100): x | mu,kappa,alpha,beta ~ t_{2 alpha}(mu, beta(kappa+1)/(alpha kappa)).""" nu = 2.0 * al s2 = be * (kap + 1.0) / (al * kap) z = (x - mu) ** 2 / (nu * s2) return (gammaln(0.5 * (nu + 1.0)) - gammaln(0.5 * nu) - 0.5 * np.log(nu * np.pi * s2) - 0.5 * (nu + 1.0) * np.log1p(z)) def _bocpd(y: np.ndarray, lam_bars: float, rmax: int) -> tuple[np.ndarray, np.ndarray]: """BOCPD (Adams & MacKay 2007) con hazard COSTANTE H = 1/lam_bars e verosimiglianza NIG. Ritorna (E[r_t], argmax_r P(r_t)) in BARRE. Filtro in avanti: ogni valore usa solo il passato -> causale per costruzione, e stabile su prefisso (nessuna statistica full-sample). Il posterior e' troncato a rmax barre e rinormalizzato (il troncamento e' un limite di memoria, NON un clip sul valore: se la massa si accumula a rmax lo si vede in diagnostica).""" n = len(y) haz = 1.0 / max(2.0, float(lam_bars)) R = np.array([1.0]) mu = np.array([NIG_MU0]); kap = np.array([NIG_KAP0]) al = np.array([NIG_AL0]); be = np.array([NIG_BE0]) rl = np.arange(1) mean_rl = np.zeros(n); map_rl = np.zeros(n) for t in range(n): x = float(y[t]) lp = _student_logpdf(x, mu, kap, al, be) lp -= lp.max() pred = np.exp(lp) gr = R * pred * (1.0 - haz) cp = float(np.sum(R * pred * haz)) m_new = min(len(R) + 1, rmax + 1) newR = np.empty(m_new) newR[0] = cp newR[1:] = gr[:m_new - 1] s = newR.sum() if not np.isfinite(s) or s <= 0: newR = np.zeros(m_new); newR[0] = 1.0; s = 1.0 newR /= s # aggiornamento delle statistiche sufficienti (run r -> run r+1), prior in testa mu_f = np.concatenate(([NIG_MU0], (kap * mu + x) / (kap + 1.0)))[:m_new] kap_f = np.concatenate(([NIG_KAP0], kap + 1.0))[:m_new] al_f = np.concatenate(([NIG_AL0], al + 0.5))[:m_new] be_f = np.concatenate(([NIG_BE0], be + kap * (x - mu) ** 2 / (2.0 * (kap + 1.0))))[:m_new] R, mu, kap, al, be = newR, mu_f, kap_f, al_f, be_f if len(rl) != m_new: rl = np.arange(m_new, dtype=float) mean_rl[t] = float(np.dot(R, rl)) map_rl[t] = float(rl[int(np.argmax(R))]) return mean_rl, map_rl def _op_causal(y: np.ndarray, pen_k: float, smax: int) -> tuple[np.ndarray, np.ndarray]: """OPTIMAL PARTITIONING **online** (l'obiettivo esatto che PELT accelera con il pruning): F(t) = min_{s in [t-smax, t-1]} [ F(s) + C(y[s:t]) + beta(t) ] con C = costo gaussiano di mean-shift su dati standardizzati e beta(t) = pen_k*log(t). ⚠️ Il termine sum(y^2) del costo gaussiano e' INVARIANTE alla partizione (ogni punto sta in esattamente un segmento) -> si omette dall'argmin. C(s:t) = -(S[t]-S[s])^2/(t-s). Ritorna (eta' dall'ultimo change-point, mediana delle lunghezze dei segmenti CONCLUSI), entrambe in barre. Per ogni t si usa SOLO y[0:t]: e' la differenza fra questo e un `ruptures.Pelt().fit_predict(y)` retrospettivo, che guarda tutto il campione.""" n = len(y) S = np.concatenate(([0.0], np.cumsum(y))) F = np.full(n + 1, np.inf) F[0] = 0.0 prev = np.zeros(n + 1, dtype=int) age = np.zeros(n); segmed = np.zeros(n) for t in range(1, n + 1): a = max(0, t - smax) s_idx = np.arange(a, t) seg = (S[t] - S[a:t]) ln = (t - s_idx).astype(float) cand = F[a:t] - seg * seg / ln j = int(np.argmin(cand)) beta = pen_k * np.log(max(2.0, float(t))) F[t] = cand[j] + beta prev[t] = int(s_idx[j]) # backtracking della partizione VALIDA A t (cap a 40 segmenti: ci serve la mediana) last = prev[t] age[t - 1] = t - last lens = [] u = last while u > 0 and len(lens) < 40: p = prev[u] lens.append(u - p) u = p segmed[t - 1] = float(np.median(lens)) if lens else float(t) return age, segmed _STAT_CACHE: dict = {} def _stat_key(df: pd.DataFrame, det: str, par) -> tuple: """Chiave chiusa sul CONTENUTO del frame (lunghezza + primo/ultimo timestamp): un PREFISSO non deve prendere la cache del campione pieno, altrimenti `causality_ok` verrebbe ingannata dalla cache invece che dal codice.""" ts = df["timestamp"].values return (det, par, len(df), int(ts[0]), int(ts[-1])) def detector_stats(df: pd.DataFrame, det: str, par) -> dict: key = _stat_key(df, det, par) if key in _STAT_CACHE: return _STAT_CACHE[key] bpd = A.bars_per_day(df) c = df["close"].values.astype(float) y = _std_returns(c, bpd) rmax = max(4, LMAX_D * bpd) if det == "bocpd": m, mp = _bocpd(y, float(par) * bpd, rmax) out = {"mean": m, "map": mp} elif det == "op": ag, sm = _op_causal(y, float(par), rmax) out = {"age": ag, "segmed": sm} else: raise ValueError(det) _STAT_CACHE[key] = out return out def lookback_bars(df: pd.DataFrame, det: str, par, stat: str, mapping: str) -> np.ndarray: """L_t in BARRE dentro [Lmin, Lmax] giorni. `direct` = la lunghezza di run E' il lookback (dopo un change-point si puo' guardare indietro solo fin la'); `inverse` e' lo specchio, dichiarato come asse perche' il verso non e' ovvio a priori.""" bpd = A.bars_per_day(df) lo, hi = max(2, LMIN_D * bpd), max(3, LMAX_D * bpd) v = detector_stats(df, det, par)[stat] base = np.clip(np.nan_to_num(v, nan=float(lo)), lo, hi) L = base if mapping == "direct" else (lo + hi - base) return np.clip(np.round(L), 2, hi).astype(int) # ======================================================================================= # DIREZIONE TSMOM (orizzonte costante o variabile) + vol-target — IDENTICA al primo tentativo # ======================================================================================= def _sign_at(c: np.ndarray, h: np.ndarray) -> tuple[np.ndarray, np.ndarray]: n = len(c) i = np.arange(n) j = i - h ok = j >= 0 s = np.zeros(n) jj = np.where(ok, j, 0) s[ok] = np.sign(c[i[ok]] / c[jj[ok]] - 1.0) return s, ok def _dir_from_L(c: np.ndarray, L: np.ndarray, blend: str) -> np.ndarray: hs = [L] if blend == "single" else [ np.maximum(2, np.round(L / 3.0).astype(int)), L, np.round(2.0 * L).astype(int)] acc = np.zeros(len(c)); cnt = np.zeros(len(c)) for h in hs: s, ok = _sign_at(c, np.asarray(h, int)) acc[ok] += s[ok]; cnt[ok] += 1 out = np.zeros(len(c)); nz = cnt > 0 out[nz] = acc[nz] / cnt[nz] return out def fixed_target(df: pd.DataFrame) -> np.ndarray: """BASELINE: TSMOM fisso 30/90/180, long-flat, vol-target 20%, leva 2x = TP01 canonico.""" c = df["close"].values.astype(float) bpd = A.bars_per_day(df) acc = np.zeros(len(c)); cnt = np.zeros(len(c)) for hd in FIXED_H: s, ok = _sign_at(c, np.full(len(c), hd * bpd, int)) acc[ok] += s[ok]; cnt[ok] += 1 d = np.zeros(len(c)); nz = cnt > 0 d[nz] = acc[nz] / cnt[nz] return A.vol_target(np.clip(d, 0, None), df, TGT_VOL, VOLWIN, LEV) def const_factory(tf: str, L: int, blend: str): def fn(df: pd.DataFrame) -> np.ndarray: c = df["close"].values.astype(float) bpd = A.bars_per_day(df) Lb = np.full(len(c), max(2, int(round(L * bpd))), int) return A.vol_target(np.clip(_dir_from_L(c, Lb, blend), 0, None), df, TGT_VOL, VOLWIN, LEV) fn.__name__ = f"const_L{L}_{blend}" return fn def factory(tf: str, det: str, par, stat: str, mapping: str, blend: str): def fn(df: pd.DataFrame) -> np.ndarray: c = df["close"].values.astype(float) L = lookback_bars(df, det, par, stat, mapping) return A.vol_target(np.clip(_dir_from_L(c, L, blend), 0, None), df, TGT_VOL, VOLWIN, LEV) fn.__name__ = f"adapt_{det}{par}_{stat}_{mapping}_{blend}" return fn _WF_CACHE: dict = {} def wf_picks(df: pd.DataFrame, win_d: int, blend: str): """(C) L scelto per WALK-FORWARD sul PnL NETTO realizzato: a ogni barra si prende l'L che, fra i candidati COSTANTI, ha lo Sharpe netto migliore nella finestra all'indietro `win_d`. Nessun modello di regime: e' il "cosa avrebbe funzionato finora". Causale: il PnL della barra i si conosce solo a fine barra i, quindi per decidere a i si usa la finestra che finisce a i-1; `eval_weights` shifta poi la posizione di un'altra barra.""" cands = [L for L in CONST_L if LMIN_D <= L <= LMAX_D] key = (win_d, blend) + _stat_key(df, "wf", 0)[2:] if key in _WF_CACHE: return _WF_CACHE[key] c = df["close"].values.astype(float) bpd = A.bars_per_day(df) n = len(c) r = A.simple_returns(c) w = max(20, win_d * bpd) tgts, sh = [], [] for L in cands: Lb = np.full(n, max(2, int(round(L * bpd))), int) t_ = A.vol_target(np.clip(_dir_from_L(c, Lb, blend), 0, None), df, TGT_VOL, VOLWIN, LEV) pos = np.zeros(n); pos[1:] = t_[:-1] net = pos * r - A.FEE_SIDE * np.abs(np.diff(pos, prepend=0.0)) ss = pd.Series(net) mu = ss.rolling(w, min_periods=w // 2).mean() sd = ss.rolling(w, min_periods=w // 2).std() with np.errstate(invalid="ignore", divide="ignore"): sr = (mu / sd).values sr = np.concatenate(([np.nan], sr[:-1])) # PnL noto solo a fine barra tgts.append(t_); sh.append(np.where(np.isfinite(sr), sr, -np.inf)) SH = np.vstack(sh); TG = np.vstack(tgts) idx = np.argmax(SH, axis=0) idx[~np.isfinite(SH).any(axis=0)] = cands.index(90) # warmup -> il canonico, non il migliore out = (np.asarray([cands[i] for i in idx]), TG[idx, np.arange(n)], cands) _WF_CACHE[key] = out return out def wf_factory(tf: str, win_d: int, blend: str): def fn(df: pd.DataFrame) -> np.ndarray: return wf_picks(df, win_d, blend)[1] fn.__name__ = f"wfpnl_{win_d}_{blend}" return fn GRID_A = [dict(det=d, par=p, stat=s, mapping=m, blend=b) for d in DETECTORS for p in PARAMS[d] for s in STATS[d] for m in MAPPINGS for b in BLENDS] GRID_B = [dict(L=L, blend=b) for L in CONST_L for b in BLENDS] GRID_C = [dict(win_d=w, blend=b) for w in WF_WIN_D for b in BLENDS] # ======================================================================================= # UTILITA' # ======================================================================================= def _dd(s: pd.Series) -> float: eq = np.cumprod(1.0 + np.asarray(s.dropna().values, float)) pk = np.maximum.accumulate(eq) return float(np.max((pk - eq) / pk)) if len(eq) else 0.0 def _cagr(s: pd.Series) -> float: s = s.dropna() if len(s) < 5: return float("nan") eq = float(np.prod(1.0 + s.values)) yrs = (s.index[-1] - s.index[0]).total_seconds() / (86400 * 365.25) return eq ** (1 / max(yrs, 1e-6)) - 1.0 if eq > 0 else -1.0 def resample_offset(df_1h: pd.DataFrame, hours: int, off: int) -> pd.DataFrame: """Stessa convenzione di trend_portfolio.resample_tf, ancora spostata di `off` ore. NB `offset=`, non `origin=` (pandas 2.x ignora origin in silenzio — lezione 02/07).""" g = df_1h.copy() idx = pd.to_datetime(g["timestamp"], unit="ms", utc=True) idx.name = "dt" g.index = idx out = g.resample(f"{hours}h", offset=pd.Timedelta(hours=off), 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"]] def paired_by_year(cand: pd.Series, base: pd.Series) -> list[dict]: J = pd.concat({"C": cand, "B": base}, axis=1, join="inner").dropna() rows = [] for y, g in J.groupby(J.index.year): if len(g) < 40: continue rows.append(dict(year=int(y), n=len(g), sh_cand=round(A._sh(g["C"]), 2), sh_base=round(A._sh(g["B"]), 2), d_sh=round(A._sh(g["C"]) - A._sh(g["B"]), 2))) return rows def delever_null(cand: pd.Series, base: pd.Series) -> dict: """Esiste k<1 che da' al BASELINE lo stesso maxDD del candidato con Sharpe >= ? Se si', il candidato non ha comprato protezione: ha solo tolto leva.""" J = pd.concat({"C": cand, "B": base}, axis=1, join="inner").dropna() dd_c, dd_b = _dd(J["C"]), _dd(J["B"]) if dd_c >= dd_b or dd_b <= 0: return dict(applicabile=False, dd_cand=round(dd_c, 4), dd_base=round(dd_b, 4), nota="il candidato NON riduce il maxDD: il null non si applica") lo, hi = 0.0, 1.0 for _ in range(60): k = 0.5 * (lo + hi) if _dd(k * J["B"]) < dd_c: lo = k else: hi = k k = 0.5 * (lo + hi) return dict(applicabile=True, k=round(k, 4), dd_cand=round(dd_c, 4), dd_base=round(dd_b, 4), sharpe_cand=round(A._sh(J["C"]), 3), sharpe_base_delev=round(A._sh(k * J["B"]), 3), superato=bool(A._sh(J["C"]) > A._sh(k * J["B"]))) _ANCHOR_CACHE: dict = {} def series_at_anchor(fn, name: str, hours: int, off: int) -> pd.Series: key = (name, hours, off) if key not in _ANCHOR_CACHE: ser = {} for a in ASSETS: dfo = resample_offset(A.get(a, "1h"), hours, off) ev = A.eval_weights(dfo, A._call_target(fn, dfo, a)) ser[a] = pd.Series(ev["net"], index=ev["idx"]) J = pd.concat(ser, axis=1, join="inner").fillna(0.0) _ANCHOR_CACHE[key] = A._to_daily(0.5 * J[ASSETS[0]] + 0.5 * J[ASSETS[1]]) return _ANCHOR_CACHE[key] def adaptivity(tf: str, det: str, par, stat: str, mapping: str) -> dict: """LA DIAGNOSTICA CHE HA SMASCHERATO IL PRIMO TENTATIVO — quanto si MUOVE davvero L_t. Criterio (identico a quello del 22/08, per confrontabilita'): sta a un bordo <50% del tempo E l'IQR e' >= 10 giorni. In piu': |dL|/giorno e numero di RESET (cadute di L > 5 giorni), che sono la firma di un vero rilevatore di change-point.""" fl, fh, iq, md, dl, nrs, sat = [], [], [], [], [], [], [] for a in ASSETS: df = A.get(a, tf) bpd = A.bars_per_day(df) L = lookback_bars(df, det, par, stat, mapping) / bpd raw = np.nan_to_num(detector_stats(df, det, par)[stat], nan=0.0) / bpd fl.append(float(np.mean(L <= LMIN_D + 1))) fh.append(float(np.mean(L >= LMAX_D - 1))) iq.append(float(np.percentile(L, 75) - np.percentile(L, 25))) md.append(float(np.median(L))) dl.append(float(np.median(np.abs(np.diff(L))))) d = np.diff(L) nrs.append(float(np.sum(d < -5.0) / (len(L) / (365.25)))) # reset all'anno sat.append(float(np.mean(raw >= LMAX_D - 1))) # saturazione del GREZZO return dict(frac_lo=float(np.mean(fl)), frac_hi=float(np.mean(fh)), iqr_days=float(np.mean(iq)), med_days=float(np.mean(md)), dL_med=float(np.mean(dl)), reset_yr=float(np.mean(nrs)), sat_raw=float(np.mean(sat)), adattiva=bool(np.mean(fl) < 0.5 and np.mean(fh) < 0.5 and np.mean(iq) >= 10.0)) def hr(t: str = "") -> None: print("\n" + "=" * 86) if t: print(t) print("=" * 86) def dsr_given_sr0(daily: pd.Series, sr_ann: float, sr0_ann: float, dpy: float = 365.25) -> float: """Deflated-Sharpe con il massimo-atteso-dal-rumore IMPOSTO dall'esterno. Serve perche' `A.deflated_sharpe` stima sr0 dalla VARIANZA degli Sharpe della griglia: su una famiglia OMOGENEA (qui sono tutte varianti di TSMOM, quindi molto correlate) quella varianza e' piccola, sr0 esce basso e il DSR esce alto **per costruzione**. Il conteggio che decide, sulla ricerca direzionale BTC/ETH, e' quello di SCREEN: il filone ORTHO ha misurato che su 168 trial il massimo atteso dal puro rumore e' Sharpe 1.572.""" r = np.asarray(pd.Series(daily).dropna().values, float) T = len(r) sr, sr0 = sr_ann / np.sqrt(dpy), sr0_ann / np.sqrt(dpy) sk = float(pd.Series(r).skew()) ku = float(pd.Series(r).kurt()) + 3.0 den = np.sqrt(max(1e-9, 1 - sk * sr + (ku - 1) / 4.0 * sr ** 2)) return float(norm.cdf((sr - sr0) * np.sqrt(T - 1) / den)) def dsr_line(sr: float, allsr: list, daily: pd.Series, label: str, n_decl: int) -> str: pad = list(allsr) + [float(np.mean(allsr))] * max(0, n_decl - len(allsr)) d, s0 = A.deflated_sharpe(sr, pad, daily) return (f" N={n_decl:>4} ({label:<24}) DSR={d:.3f} {'PASS' if d >= 0.95 else 'FAIL'}" f" max atteso dal rumore {s0:+.3f}") # ======================================================================================= def main() -> None: t0 = time.time() print(__doc__.split("Runtime atteso")[0]) # -- 0. SANITY ----------------------------------------------------------------------- hr("0. SANITY — il baseline 'fisso' e' TP01 canonico (se non lo e', ogni delta e' inventato)") tp = TrendPortfolio(**CANONICAL) worst = 0.0 for a in ASSETS: df = A.get(a, "1d") worst = max(worst, float(np.max(np.abs(fixed_target(df) - tp.target_series(df))))) base_1d = A.candidate_daily(fixed_target, tf="1d") tp01 = A.tp01_baseline_daily() JJ = pd.concat({"a": base_1d, "b": tp01}, axis=1, join="inner").dropna() print(f" max|target_mio - TrendPortfolio.target_series| (1d, BTC+ETH) = {worst:.3e}") print(f" max|serie_giornaliera_mia - A.tp01_baseline_daily()| = " f"{float(np.max(np.abs(JJ['a'] - JJ['b']))):.3e} (n={len(JJ)})") assert worst < 1e-12, "il baseline NON riproduce TP01: fermarsi qui" # -- 0-bis. IL RILEVATORE E' CAUSALE? (troncamento MANUALE, prima dei gate) ----------- hr("0-bis. TRONCAMENTO MANUALE — la trappola del change-point RETROSPETTIVO") print(" Un `Pelt().fit_predict(serie_intera)` restituisce la segmentazione dell'INTERO") print(" campione: al tempo t 'saprebbe' dei cambi futuri. Qui BOCPD e OP sono ricorsioni in") print(" avanti. Prova diretta: ricalcolo su un PREFISSO e confronto la coda con il pieno.") for det, par in (("bocpd", 100), ("op", 5.0)): for stat in STATS[det]: wmax = 0.0 for a in ASSETS: df = A.get(a, "1d") full = lookback_bars(df, det, par, stat, "direct") for frac in (0.70, 0.85): cut = int(len(df) * frac) sub = df.iloc[:cut].reset_index(drop=True) pre = lookback_bars(sub, det, par, stat, "direct") wmax = max(wmax, float(np.max(np.abs(pre[-100:] - full[cut - 100:cut])))) print(f" {det:<6} {str(par):>5} {stat:<7}: max|L(prefisso)-L(pieno)| sulla coda " f"= {wmax:.3e} barre -> {'CAUSALE' if wmax < 1e-9 else 'SOSPETTO'}") # -- 1-pre. TARATURA DELLA SCALA — guardando il RILEVATORE, mai lo Sharpe ----------- hr("1-pre. TARATURA DELLA SCALA DEI PARAMETRI (guardando SOLO quanti change-point rileva)") print(" Un rilevatore che non rileva nulla non e' un rilevatore. Questa scansione guarda il") print(" NUMERO di change-point e la distribuzione di L_t — mai un rendimento. Dichiarata") print(" come 13 trial in piu' (al rialzo), benche' nessuno produca uno Sharpe.") dfc = A.get("BTC", "1d") yc = _std_returns(dfc["close"].values.astype(float), 1) print(f"\n OP (BTC 1d) {'pen_k':>6} {'cp/anno':>8} {'age med':>8} {'IQR':>7} {'@tetto':>8}") for k in CAL_PEN: ag, _ = _op_causal(yc, k, LMAX_D) ncp = int(np.sum(np.diff(ag) < 0)) flag = " <= USATO" if k in PARAMS["op"] else (" <= degenere (artefatto 22/08)" if np.mean(ag >= LMAX_D - 1) > 0.30 else "") print(f" {k:>6} {ncp/(len(yc)/365.25):>8.1f} {np.median(ag):>8.1f} " f"{np.percentile(ag,75)-np.percentile(ag,25):>7.1f} " f"{np.mean(ag>=LMAX_D-1)*100:>7.1f}%{flag}") print(f"\n BOCPD (BTC 1d) {'lambda':>6} {'E[r] med':>9} {'IQR':>7} {'@tetto':>8} " f"{'MAP med':>8} {'@tetto':>8}") for lam in CAL_LAM: mn, mp = _bocpd(yc, float(lam), LMAX_D) flag = " <= USATO" if lam in PARAMS["bocpd"] else "" print(f" {lam:>6} {np.median(mn):>9.1f} " f"{np.percentile(mn,75)-np.percentile(mn,25):>7.1f} " f"{np.mean(mn>=LMAX_D-1)*100:>7.1f}% {np.median(mp):>8.1f} " f"{np.mean(mp>=LMAX_D-1)*100:>7.1f}%{flag}") # -- 1. LA DIAGNOSTICA DI MECCANISMO, PER PRIMA -------------------------------------- hr("1. L_t SI MUOVE DAVVERO? (se sta al bordo l'artefatto e' riprodotto: si chiude e basta)") print(" criterio (identico al 22/08, per confrontabilita'): a un bordo <50% del tempo E") print(" IQR >= 10 giorni. `reset/anno` = cadute di L > 5 giorni = firma di un change-point.") print(f"\n {'TF':>4} {'det':>6} {'par':>5} {'stat':>7} {'map':<7} {'bordo-lo':>9} " f"{'bordo-hi':>9} {'IQR(g)':>7} {'med(g)':>7} {'|dL|(g)':>8} {'reset/a':>8} " f"{'sat.grezzo':>10} esito") diag: dict = {} for tf in TFS: for d in DETECTORS: for p in PARAMS[d]: for st in STATS[d]: for mp in MAPPINGS: v = adaptivity(tf, d, p, st, mp) diag[(tf, d, p, st, mp)] = v print(f" {tf:>4} {d:>6} {str(p):>5} {st:>7} {mp:<7} " f"{v['frac_lo']*100:8.1f}% {v['frac_hi']*100:8.1f}% " f"{v['iqr_days']:7.1f} {v['med_days']:7.1f} {v['dL_med']:8.2f} " f"{v['reset_yr']:8.1f} {v['sat_raw']*100:9.1f}% " f"{'ADATTIVA' if v['adattiva'] else 'costante-travestito'}") n_ad = sum(1 for v in diag.values() if v["adattiva"]) print(f"\n configurazioni di rilevatore DAVVERO adattive: {n_ad}/{len(diag)}") if n_ad == 0: hr("STOP — ARTEFATTO RIPRODOTTO") print(" Nessun rilevatore muove L_t: ogni 'vincitore' sarebbe un orizzonte COSTANTE") print(" travestito, esattamente come il 22/08. Il filone si chiude per la seconda volta") print(" senza bisogno di guardare i rendimenti. FINE.") return # -- 2. GRIGLIA (A): cella scelta IN-SAMPLE + deflated-Sharpe ------------------------- hr(f"2. study_family_honest (A) — {len(GRID_A)} celle x {len(TFS)} TF = " f"{len(GRID_A)*len(TFS)} trial") rep = A.study_family_honest("BOCPD-H", factory, GRID_A, TFS) ch = rep["chosen"] print(f" celle valutate : {rep['n_cells']}") print(f" cella scelta IN-SAMPLE : tf={ch['tf']} {ch['params']}") print(f" Sharpe IS {ch['insample_sharpe']} FULL {ch['full_sharpe']}") print(f" deflated-Sharpe (N={rep['n_cells']}): {rep['deflated_sharpe']} " f"(max atteso sotto il null {rep['expected_null_max']}) PASS={rep['dsr_pass']}") print(f" earns_slot_marginal : {rep['earns_slot_marginal']}") print(f" EARNS_SLOT_HONEST : {rep['earns_slot_honest']}") print("\n top-10 IN-SAMPLE (come si sceglie onestamente):") for r in rep["rows"][:10]: ad = diag[(r["tf"], r["params"]["det"], r["params"]["par"], r["params"]["stat"], r["params"]["mapping"])]["adattiva"] print(f" IS {r['insample_sharpe']:+.3f} FULL {r['full_sharpe']:+.3f} {r['tf']:>3} " f"{r['params']} {'[adattiva]' if ad else '[costante-travestito]'}") fn_ch = factory(tf=ch["tf"], **ch["params"]) d_ch = A.candidate_daily(fn_ch, tf=ch["tf"]) # -- 3. CONTROLLO (B): il MIGLIOR ORIZZONTE COSTANTE --------------------------------- hr(f"3. CONTROLLO L COSTANTE (B) — {len(GRID_B)} celle x {len(TFS)} TF = " f"{len(GRID_B)*len(TFS)} trial [IL NULL GIUSTO, non TP01]") csel = A.select_cell_insample(const_factory, GRID_B, TFS) cch = csel["chosen"] print(f" miglior COSTANTE scelto in-sample: tf={cch['tf']} {cch['params']} " f"IS {cch['insample_sharpe']} FULL {cch['full_sharpe']}") print(" top-8 costanti in-sample (L=90 triple 1d == TP01 canonico):") for r in csel["rows"][:8]: tag = " <== TP01" if (r["params"]["L"] == 90 and r["params"]["blend"] == "triple" and r["tf"] == "1d") else "" print(f" IS {r['insample_sharpe']:+.3f} FULL {r['full_sharpe']:+.3f} {r['tf']:>3} " f"L={r['params']['L']:>3}g {r['params']['blend']}{tag}") fn_const = const_factory(tf=cch["tf"], **cch["params"]) d_const = A.candidate_daily(fn_const, tf=cch["tf"]) # -- 4. WF-PnL (C) -------------------------------------------------------------------- hr(f"4. WALK-FORWARD SUL PnL REALIZZATO (C) — {len(GRID_C)} celle x {len(TFS)} TF = " f"{len(GRID_C)*len(TFS)} trial") wsel = A.select_cell_insample(wf_factory, GRID_C, TFS) wch = wsel["chosen"] print(f" miglior WF scelto in-sample: tf={wch['tf']} {wch['params']} " f"IS {wch['insample_sharpe']} FULL {wch['full_sharpe']}") for r in wsel["rows"]: print(f" IS {r['insample_sharpe']:+.3f} FULL {r['full_sharpe']:+.3f} {r['tf']:>3} " f"{r['params']}") fn_wf = wf_factory(tf=wch["tf"], **wch["params"]) d_wf = A.candidate_daily(fn_wf, tf=wch["tf"]) # quanto si muove L nella variante WF? (stesso metro della sezione 1) for a in ASSETS: df = A.get(a, wch["tf"]); bpd = A.bars_per_day(df) pick, _, cands = wf_picks(df, **wch["params"]) print(f" {a}: L_WF giorni mediana {np.median(pick):.0f} IQR " f"{np.percentile(pick,75)-np.percentile(pick,25):.0f} " f"bordo basso {np.mean(pick<=min(cands))*100:.1f}% " f"bordo alto {np.mean(pick>=max(cands))*100:.1f}% " f"cambi/anno {np.sum(np.diff(pick)!=0)/(len(pick)/(bpd*365.25)):.0f}") # -- 5. IL CONFRONTO CHE DECIDE ------------------------------------------------------ hr("5. ADATTIVO vs MIGLIOR COSTANTE (entrambi scelti IN-SAMPLE) — e vs TP01") base_ch = A.candidate_daily(fixed_target, tf=ch["tf"]) for nome, ser in (("BOCPD/OP (A)", d_ch), ("WF-PnL (C)", d_wf), ("miglior COSTANTE (B)", d_const), ("TP01 30/90/180", base_1d)): h = ser[ser.index >= HOLDOUT] print(f" {nome:<22} FULL Sh {A._sh(ser):+.3f} HOLD Sh {A._sh(h):+.3f} " f"maxDD {_dd(ser)*100:5.2f}% CAGR {_cagr(ser)*100:6.2f}% " f"corr->TP01 {pd.concat({'a':ser,'b':tp01},axis=1,join='inner').dropna().corr().iloc[0,1]:+.3f}") for lab, cand, ref, refname in (("A vs B", d_ch, d_const, "miglior costante"), ("C vs B", d_wf, d_const, "miglior costante"), ("A vs TP01", d_ch, base_1d, "TP01")): J = pd.concat({"C": cand, "B": ref}, axis=1, join="inner").dropna() JH = J[J.index >= HOLDOUT] rows = paired_by_year(cand, ref) ds = [r["d_sh"] for r in rows] print(f"\n --- {lab} ({refname}) --- corr {J['C'].corr(J['B']):+.3f}") print(f" dSh FULL {A._sh(J['C'])-A._sh(J['B']):+.3f} " f"dSh HOLD {A._sh(JH['C'])-A._sh(JH['B']):+.3f}") print(" differenza APPAIATA per anno:") print(" " + " ".join(f"{r['year']}:{r['d_sh']:+.2f}" for r in rows)) print(f" -> anni con dSh>0: {sum(1 for x in ds if x > 0)}/{len(ds)} " f"mediana {np.median(ds):+.2f} media {np.mean(ds):+.2f} " f"[criterio del falsificatore: >=6/8 anni]") # decomposizione: quanto del vantaggio su TP01 e' ADATTIVITA' e quanto e' "L piu' corto"? JT = pd.concat({"C": d_ch, "K": d_const, "B": base_1d}, axis=1, join="inner").dropna() v_tp = A._sh(JT["C"]) - A._sh(JT["B"]) v_ct = A._sh(JT["K"]) - A._sh(JT["B"]) print(f"\n DECOMPOSIZIONE del vantaggio dell'adattivo su TP01 (FULL, finestra comune):") print(f" adattivo - TP01 = {v_tp:+.3f}") print(f" miglior COSTANTE - TP01 = {v_ct:+.3f} <- quota spiegata da 'un L piu' corto'") print(f" adattivo - miglior COSTANTE= {v_tp - v_ct:+.3f} <- quota spiegata dall'ADATTARE") print(f" => l'adattivita' spiega il {100*(v_tp-v_ct)/v_tp if abs(v_tp)>1e-9 else float('nan'):+.0f}% " f"del vantaggio su TP01") # la superficie di L costante e' una CURVA o un CRINALE DI RUMORE? (replica del 22/08) rows12 = {r["params"]["L"]: r["insample_sharpe"] for r in csel["rows"] if r["tf"] == cch["tf"] and r["params"]["blend"] == cch["params"]["blend"]} Ls = sorted(rows12) adj = float(np.median(np.abs(np.diff([rows12[L] for L in Ls])))) print(f"\n RISOLUZIONE della griglia costante ({cch['tf']}, {cch['params']['blend']}): salto IS " f"mediano fra L ADIACENTI {adj:+.3f}") print(f" contro il vantaggio del miglior costante su TP01 ({v_ct:+.3f}) -> rapporto " f"{adj/max(1e-9, abs(v_ct)):.2f} [replica del 'crinale di rumore' del 22/08]") # -- 6. GATE -------------------------------------------------------------------------- hr("6. GATE sul candidato (A) scelto in-sample") print(A.fmt_marginal(rep["marginal"])) cz = A.causality_ok(fn_ch, tf=ch["tf"]) print(f"\n causality_ok : ok={cz['ok']} max_tail_diff={cz['max_tail_diff']} " f"(prefissi controllati {cz['checked']})") imp = A.implausible_sharpe(d_ch) print(f" implausible_sharpe : implausible={imp['implausible']} Sh {imp.get('sharpe')} " f"maxDD {imp.get('maxdd')} perdite/attive {imp.get('loss_frac')} " f"attive {imp.get('active_frac')}") for rr in imp["reasons"]: print(f" - {rr}") print(f" null de-levering vs COSTANTE : {delever_null(d_ch, d_const)}") print(f" null de-levering vs TP01 : {delever_null(d_ch, base_1d)}") print(" haircut a $600 (min-order $5):") for a in ASSETS: df = A.get(a, ch["tf"]) sc = A.eval_weights_smallcap(df, A._call_target(fn_ch, df, a), capital=600.0) print(f" {a}: modellato {sc['modeled']['sharpe']:+.3f} -> reale " f"{sc['realistic']['sharpe']:+.3f} (haircut {sc['sharpe_haircut']:+.3f}, " f"{sc['n_executed_trades']} trade)") # DSR a piu' conteggi: famiglia, famiglia+112 del primo tentativo, di SCREEN all_sr = ([r["full_sharpe"] for r in rep["rows"]] + [r["full_sharpe"] for r in csel["rows"]] + [r["full_sharpe"] for r in wsel["rows"]]) hr("6-bis. DEFLATED-SHARPE — sensibilita' del verdetto al CONTEGGIO (regola 30/07 + 22/08)") print(f" Sharpe FULL del candidato (A) = {A._sh(d_ch):+.3f}") print(dsr_line(A._sh(d_ch), all_sr, d_ch, "mia famiglia A+B+C", N_DECLARED)) print(dsr_line(A._sh(d_ch), all_sr, d_ch, "+ le 112 del 22/08", N_DECLARED + N_PRIOR_FAMILY)) print(dsr_line(A._sh(d_ch), all_sr, d_ch, "di SCREEN (ondata)", N_DECLARED + N_PRIOR_FAMILY + N_SCREEN)) print("\n ⚠️ QUEL 'PASS' NON CERTIFICA NIENTE, e va detto: `deflated_sharpe` stima il massimo") print(" atteso dal rumore dalla VARIANZA degli Sharpe della griglia. Qui la griglia e'") print(" OMOGENEA (tutte varianti di TSMOM, molto correlate fra loro) -> varianza piccola") print(" -> sr0 basso -> DSR alto per COSTRUZIONE. Aumentare N padding con la media alza N") print(" ma non la varianza. Il conteggio che decide e' quello di SCREEN, e il filone ORTHO") print(" lo ha gia' misurato: su 168 trial di ricerca direzionale BTC/ETH il massimo atteso") print(" dal puro rumore e' Sharpe 1.572. Riscritto con QUEL sr0:") for nome, ser in (("adattivo (A)", d_ch), ("miglior costante (B)", d_const), ("TP01 30/90/180", base_1d)): print(f" {nome:<22} Sharpe FULL {A._sh(ser):+.3f} vs sr0_screen 1.572 -> " f"DSR {dsr_given_sr0(ser, A._sh(ser), 1.572):.3f} " f"{'PASS' if dsr_given_sr0(ser, A._sh(ser), 1.572) >= 0.95 else 'FAIL'}") # PROVA DI IDENTITA': la cella "adattiva" al 100% sul bordo E' il costante, bit per bit hr("6-ter. PROVA DI IDENTITA' — la cella al bordo NON somiglia a un costante: LO E'") deg = [r for r in rep["rows"] if diag[(r["tf"], r["params"]["det"], r["params"]["par"], r["params"]["stat"], r["params"]["mapping"])]["frac_lo"] >= 0.999] if deg: dr = max(deg, key=lambda r: r["insample_sharpe"]) s_deg = A.candidate_daily(factory(tf=dr["tf"], **dr["params"]), tf=dr["tf"]) s_con = A.candidate_daily(const_factory(tf=dr["tf"], L=LMIN_D, blend=dr["params"]["blend"]), tf=dr["tf"]) Jd = pd.concat({"a": s_deg, "b": s_con}, axis=1, join="inner").dropna() print(f" cella 100% al bordo basso: {dr['tf']} {dr['params']} IS {dr['insample_sharpe']}" f" FULL {dr['full_sharpe']}") print(f" costante L={LMIN_D}g {dr['params']['blend']} {dr['tf']}: " f"FULL {A._sh(s_con):+.3f}") print(f" max|serie_adattiva - serie_costante| = {float(np.max(np.abs(Jd['a']-Jd['b']))):.3e}" f" (n={len(Jd)}) -> {'SONO LA STESSA STRATEGIA' if float(np.max(np.abs(Jd['a']-Jd['b']))) < 1e-12 else 'diverse'}") else: print(" nessuna cella e' al 100% sul bordo basso (la prova non si applica)") # -- 7. FAMIGLIA RISTRETTA ALLE SOLE CELLE DAVVERO ADATTIVE -------------------------- hr("7. GATE SULLA SOLA FAMIGLIA DAVVERO ADATTIVA (la domanda della missione, isolata)") ad_rows = [r for r in rep["rows"] if diag[(r["tf"], r["params"]["det"], r["params"]["par"], r["params"]["stat"], r["params"]["mapping"])]["adattiva"]] d_ad_keep = fn_ad_keep = best_keep = None if not ad_rows: print(" NESSUNA cella della famiglia e' davvero adattiva -> vedi sezione 1.") else: best = max(ad_rows, key=lambda r: r["insample_sharpe"]) print(f" celle adattive: {len(ad_rows)}/{len(rep['rows'])} miglior IN-SAMPLE: " f"tf={best['tf']} {best['params']} IS {best['insample_sharpe']} " f"FULL {best['full_sharpe']}") fn_ad = factory(tf=best["tf"], **best["params"]) d_ad = A.candidate_daily(fn_ad, tf=best["tf"]) sm = A.study_marginal("BOCPD-H (sole adattive)", fn_ad, tf=best["tf"]) dsr_ad, sr0_ad = A.deflated_sharpe(A._sh(d_ad), [r["full_sharpe"] for r in ad_rows], d_ad) print(f" marginal={sm['marginal_verdict']} earns_slot={sm['earns_slot']} " f"corr->TP01 {sm['marginal'].get('corr_full')}") print(f" deflated-Sharpe (N={len(ad_rows)}) = {dsr_ad:.3f} " f"{'PASS' if dsr_ad >= 0.95 else 'FAIL'} (null max atteso {sr0_ad:+.3f})") JA = pd.concat({"C": d_ad, "K": d_const, "B": base_1d}, axis=1, join="inner").dropna() JH = JA[JA.index >= HOLDOUT] print(f" FULL adattiva-vera {A._sh(JA['C']):+.3f} | costante {A._sh(JA['K']):+.3f} " f"| TP01 {A._sh(JA['B']):+.3f}") print(f" HOLD-OUT adattiva-vera {A._sh(JH['C']):+.3f} | costante {A._sh(JH['K']):+.3f} " f"| TP01 {A._sh(JH['B']):+.3f}") r3 = paired_by_year(d_ad, d_const) d3 = [r["d_sh"] for r in r3] print(" vs miglior costante, differenza APPAIATA per anno:") print(" " + " ".join(f"{r['year']}:{r['d_sh']:+.2f}" for r in r3)) print(f" -> dSh>0 in {sum(1 for x in d3 if x>0)}/{len(d3)} mediana {np.median(d3):+.2f}" f" corr al costante {JA['C'].corr(JA['K']):+.3f}") print(" ⚠️ l'hold-out sono gli ULTIMI DUE anni di questa stessa riga: se il vantaggio") print(" hold-out sta li' e non nel resto, e' un evento, non un meccanismo (regola:") print(" un vantaggio concentrato in un anno non e' un vantaggio).") print(f" null de-levering vs costante: {delever_null(d_ad, d_const)}") d_ad_keep, fn_ad_keep, best_keep = d_ad, fn_ad, best # -- 8. BANDA D'ANCORA (delta APPAIATO adattivo - costante) -------------------------- hours = 24 if ch["tf"] == "1d" else 12 offs = list(range(hours)) hr(f"8. BANDA D'ANCORA — {len(offs)} ancore. Stima onesta = MEDIANA; il confronto fra due " f"varianti\n e' la MEDIANA DELLE DIFFERENZE APPAIATE (mai la differenza delle mediane).") if cch["tf"] != ch["tf"]: print(f" ⚠️ adattivo su {ch['tf']} e miglior costante su {cch['tf']}: per il delta " f"appaiato uso il miglior costante RIVALUTATO su {ch['tf']} (stesse ancore = coppie).") crow = max([r for r in csel["rows"] if r["tf"] == ch["tf"]], key=lambda r: r["insample_sharpe"]) fn_const_anch = const_factory(tf=ch["tf"], **crow["params"]) print(f" miglior costante su {ch['tf']}: {crow['params']} IS {crow['insample_sharpe']}") else: fn_const_anch = fn_const s0 = series_at_anchor(fixed_target, "fixed", hours, 0) K = pd.concat({"a": s0, "b": base_ch}, axis=1, join="inner").dropna() print(f" sanity ancora 0 vs serie canonica del fisso: max|Δ| " f"{float(np.max(np.abs(K['a']-K['b']))):.3e} (n={len(K)})") for lab, mtr in (("Sharpe FULL", A._sh), ("Sharpe HOLD", lambda s: A._sh(s[s.index >= HOLDOUT]))): b_c = A.anchor_luck_band(lambda o: series_at_anchor(fn_ch, "cand", hours, o), offs, metric=mtr) b_k = A.anchor_luck_band(lambda o: series_at_anchor(fn_const_anch, "const", hours, o), offs, metric=mtr) print(f" {lab:<12} adattivo: canonica {b_c['canonical']:+.3f} " f"(pctl {b_c['canonical_pctl']:.2f}) MEDIANA {b_c['median']:+.3f} " f"banda [{b_c['lo']:+.3f},{b_c['hi']:+.3f}]") print(f" {'':<12} costante: canonica {b_k['canonical']:+.3f} " f"(pctl {b_k['canonical_pctl']:.2f}) MEDIANA {b_k['median']:+.3f} " f"banda [{b_k['lo']:+.3f},{b_k['hi']:+.3f}]") dl = A.anchor_luck_delta(lambda o: series_at_anchor(fn_ch, "cand", hours, o), lambda o: series_at_anchor(fn_const_anch, "const", hours, o), offs, metric=mtr) print(f" {'':<12} DELTA APPAIATO (adatt - cost): mediana {dl['median_paired']:+.3f} " f"positivo in {dl['n_positive']}/{dl['n_anchors']} " f"banda [{dl['lo']:+.3f},{dl['hi']:+.3f}] gate_pass={dl['gate_pass']}") # -- 8-bis. ANCORA APPAIATA per la cella DAVVERO ADATTIVA --------------------------- if fn_ad_keep is not None and best_keep["tf"] == ch["tf"]: hr("8-bis. ANCORA APPAIATA — cella DAVVERO ADATTIVA vs miglior costante") for lab, mtr in (("Sharpe FULL", A._sh), ("Sharpe HOLD", lambda s_: A._sh(s_[s_.index >= HOLDOUT]))): dl = A.anchor_luck_delta( lambda o: series_at_anchor(fn_ad_keep, "adatt", hours, o), lambda o: series_at_anchor(fn_const_anch, "const", hours, o), offs, metric=mtr) b = A.anchor_luck_band(lambda o: series_at_anchor(fn_ad_keep, "adatt", hours, o), offs, metric=mtr) print(f" {lab:<12} adattiva-vera: canonica {b['canonical']:+.3f} " f"(pctl {b['canonical_pctl']:.2f}) MEDIANA {b['median']:+.3f} " f"banda [{b['lo']:+.3f},{b['hi']:+.3f}]") print(f" {'':<12} DELTA APPAIATO vs costante: mediana {dl['median_paired']:+.3f} " f"positivo in {dl['n_positive']}/{dl['n_anchors']} " f"banda [{dl['lo']:+.3f},{dl['hi']:+.3f}] gate_pass={dl['gate_pass']}") elif fn_ad_keep is not None: print(f"\n (8-bis non girata: la cella adattiva sta su {best_keep['tf']} e il costante " f"su {ch['tf']} -> le ancore non sarebbero coppie)") # -- 8-ter. IL CRITERIO DEL FALSIFICATORE, SU **TUTTE** LE CELLE ADATTIVE ------------ hr("8-ter. IL CRITERIO DEL FALSIFICATORE APPLICATO A **OGNI** CELLA DAVVERO ADATTIVA") print(" Il 22/08 dichiarava: «un rilevatore il cui L_t si muove DAVVERO e batte il miglior") print(" lookback COSTANTE in >=6/8 anni appaiati con corr->TP01 < 0.6». Non basta che") print(" fallisca il VINCITORE: si applica il criterio a tutte e", len(ad_rows), "le celle adattive.") print(f"\n {'tf':>4} {'cella':<58} {'corr':>6} {'anni>0':>7} {'dShFULL':>8} {'dShHOLD':>8} esito") passed = [] for r in sorted(ad_rows, key=lambda x: -x["insample_sharpe"]): fnr = factory(tf=r["tf"], **r["params"]) dr = A.candidate_daily(fnr, tf=r["tf"]) JJ2 = pd.concat({"a": dr, "b": tp01}, axis=1, join="inner").dropna() cor = float(JJ2["a"].corr(JJ2["b"])) yr = paired_by_year(dr, d_const) nyp = sum(1 for x in yr if x["d_sh"] > 0) Jr = pd.concat({"C": dr, "B": d_const}, axis=1, join="inner").dropna() JrH = Jr[Jr.index >= HOLDOUT] dF = A._sh(Jr["C"]) - A._sh(Jr["B"]) dH = A._sh(JrH["C"]) - A._sh(JrH["B"]) ok = bool(cor < 0.6 and nyp >= 6) if ok: passed.append((r, cor, nyp, dF, dH)) lab = (f"{r['params']['det']} {r['params']['par']} {r['params']['stat']} " f"{r['params']['mapping']} {r['params']['blend']}") print(f" {r['tf']:>4} {lab:<58} {cor:>6.3f} {nyp:>5}/8 {dF:>+8.3f} {dH:>+8.3f} " f"{'PASSA' if ok else ''}") print(f"\n celle che soddisfano ENTRAMBE le condizioni del falsificatore: " f"{len(passed)}/{len(ad_rows)}") cors = [] for r in ad_rows: dr = A.candidate_daily(factory(tf=r["tf"], **r["params"]), tf=r["tf"]) JJ2 = pd.concat({"a": dr, "b": tp01}, axis=1, join="inner").dropna() cors.append(float(JJ2["a"].corr(JJ2["b"]))) print(f" corr->TP01 sulle celle adattive: min {min(cors):.3f} mediana " f"{np.median(cors):.3f} max {max(cors):.3f} (soglia del falsificatore 0.60)") # -- 9. FEE SWEEP --------------------------------------------------------------------- hr("9. FEE SWEEP — l'adattivo trada di piu': e' li' che muore o no") for nm, f_, tfx in (("adattivo (A)", fn_ch, ch["tf"]), ("WF-PnL (C)", fn_wf, wch["tf"]), ("miglior costante (B)", fn_const, cch["tf"]), ("TP01 30/90/180", fixed_target, "1d")): line = [] for fee in (0.0, 0.0005, 0.001, 0.0015): line.append(f"{2*fee*100:.2f}%RT {A._sh(A.candidate_daily(f_, tf=tfx, fee_side=fee)):+.3f}") tt = np.mean([A.eval_weights(A.get(a, tfx), A._call_target(f_, A.get(a, tfx), a)) ["turnover_per_year"] for a in ASSETS]) print(f" {nm:<22} " + " ".join(line) + f" | turnover {tt:.0f}x/anno") hr(f"FINE — {time.time()-t0:.0f}s") if __name__ == "__main__": main()