#!/usr/bin/env python """r0822_adaptive_horizon.py — ADAPTIVE-HORIZON TSMOM: stimare *quanto indietro guardare* dai dati (change-point / persistenza) invece di fissarlo a 30/90/180 giorni. DOMANDA ------- Un TSMOM a lookback ADATTIVO batte il TSMOM a orizzonti FISSI 30/90/180 (TP01), in modo MARGINALE e ONESTO? COSA NON E' (i due vicini gia' bocciati, per non rifarli) --------------------------------------------------------- * 02/07 "velocita' trend regime-condizionata": pesava i TRE orizzonti fissi in funzione del percentile di vol -> risulto' un tilt-30d statico travestito (pctl 0.71 vs il null dei pesi statici casuali). Li' gli orizzonti restavano 30/90/180: cambiavano solo i PESI. * 29/06 "meta-allocazione dinamica" fra sleeve: peggiore dei pesi fissi. Li' l'oggetto adattato erano i pesi di PORTAFOGLIO, non un parametro della strategia. Qui l'oggetto adattato e' **l'orizzonte stesso**: L_t stimato da un rilevatore di change-point (CUSUM) o di persistenza (variance-ratio, Hurst). Mai provato nel progetto. FORMA FUNZIONALE (scelta perche' rende il confronto strutturalmente pulito) -------------------------------------------------------------------------- TP01 canonico e' il blend di sign(c[i]/c[i-h]-1) su h = (30, 90, 180) giorni, long-flat, vol-target 20%, leva cap 2x. Ma (30, 90, 180) == (L/3, L, 2L) con **L = 90 giorni**. Quindi il candidato e' LA STESSA FUNZIONE con L che varia nel tempo: blend='triple' -> horizons_t = (L_t/3, L_t, 2*L_t) (TP01 = L_t costante a 90g) blend='single' -> horizons_t = (L_t,) (un solo orizzonte adattivo) Tutto il resto (long-flat, target_vol 20%, vol_win 30g, leva 2x, fee 0.05%/lato, cadenza, 50/50 BTC+ETH) e' IDENTICO al baseline: **l'unico grado di liberta' e' l'orizzonte**. => il null non e' "meglio di zero", e' "meglio del TSMOM fisso a PARI GEOMETRIA". FAMIGLIA DICHIARATA **PRIMA** DI GUARDARE (conteggio al RIALZO) -------------------------------------------------------------- rilevatore (3): cusum | vratio | hurst parametro (3): cusum k in {2,4,8} sigma-cumulate; vratio/hurst finestra W in {60,120,250}g mappatura (2): direct (piu' persistenza / piu' tempo dall'ultimo change-point -> L piu' LUNGO) inverse (specchiata) [il verso NON e' ovvio a priori => e' un asse, non una scelta] blend (2): single | triple timeframe (2): 1d | 12h [>= 12h: sotto, costi+overfit dominano (19/06)] ------------------------------------------------------------------ = 3 x 3 x 2 x 2 = 36 celle x 2 TF = **72 valutazioni**. Costanti FISSATE A PRIORI e non cercate (dichiarate perche' sono gradi di liberta' non contati nella griglia): Lmin=20g, Lmax=200g (contengono 30..180 del canonico); q=5g del variance-ratio; scale del Hurst k in {1,2,4,8}g; finestra della percentile causale M=250g. Per questo il deflated-Sharpe e' riportato ANCHE a N=144 e N=216 trial (sensibilita' del verdetto al conteggio, regola del 30/07): un verdetto che si ribalta col conteggio si cita come tale. GATE (tutti dell'harness del progetto, nessuno riscritto) --------------------------------------------------------- A.study_family_honest -> cella scelta IN-SAMPLE-ONLY + deflated-Sharpe (NO selezione su hold-out) A.marginal_vs_tp01 -> ADDS/HEDGE/NOISE/REDUNDANT/DILUTES + corr al baseline A.causality_ok -> ricalcolo su prefisso (nessun look-ahead nella costruzione del segnale) A.anchor_luck_band -> stima onesta = MEDIANA delle 24 ancore, non l'ancora 00:00 A.anchor_luck_delta -> confronto adattivo-vs-fisso: MEDIANA delle DIFFERENZE APPAIATE A.implausible_sharpe -> Sharpe troppo bello = rischio fuori dal campione A.eval_weights_smallcap-> haircut reale a $600 (min-order $5) null del de-levering -> esiste k<1 che da' al FISSO lo stesso maxDD con Sharpe >=? differenza APPAIATA PER ANNO -> un vantaggio concentrato in un anno e' un evento, non un meccanismo Causalita': ogni rolling ha min_periods e finestra all'indietro; la CUSUM e' un loop in avanti con reset (prefix-stable); la percentile e' rolling causale; nessuna statistica full-sample. Runtime atteso ~3-6 min su 2 core. Nessun file toccato fuori da questo. """ from __future__ import annotations import sys import warnings 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 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 adattivo, in giorni FIXED_H = (30, 90, 180) # == (L/3, L, 2L) con L=90 -> il baseline TGT_VOL, LEV, VOLWIN = 0.20, 2.0, 30 VR_Q_DAYS = 5 # passo del variance-ratio HURST_K_DAYS = (1, 2, 4, 8) # scale del Hurst PCTL_M_DAYS = 250 # finestra della percentile causale TFS = ("1d", "12h") DETECTORS = ("cusum", "vratio", "hurst") PARAMS = {"cusum": (2.0, 4.0, 8.0), "vratio": (60, 120, 250), "hurst": (60, 120, 250)} MAPPINGS = ("direct", "inverse") BLENDS = ("single", "triple") # =========================================================================== # RILEVATORI -> lookback adattivo L_t (in barre). Tutti causali. # =========================================================================== def _logret(c: np.ndarray) -> np.ndarray: r = np.zeros(len(c)) r[1:] = np.log(c[1:] / c[:-1]) return r def _roll_pctl(x: np.ndarray, m: int) -> np.ndarray: """Percentile CAUSALE del valore corrente nella sua finestra all'indietro (rolling rank). Warmup -> 0.5 (neutro): non si inventa un regime dove non c'e' campione.""" s = pd.Series(x) u = s.rolling(m, min_periods=max(20, m // 4)).rank(pct=True).values return np.where(np.isfinite(u), u, 0.5) def _cusum_age(c: np.ndarray, k: float, bpd: int) -> np.ndarray: """Eta' (in barre) dall'ultimo change-point CUSUM a due code su rendimenti standardizzati. Loop in avanti con reset: dipende solo dal passato -> stabile su prefisso.""" lr = _logret(c) sig = pd.Series(lr).rolling(30 * bpd, min_periods=max(5, 15 * bpd)).std().values s = np.where((sig > 0) & np.isfinite(sig), lr / np.where(sig > 0, sig, 1.0), 0.0) s = np.nan_to_num(s) n = len(c) age = np.zeros(n) sp = sm = 0.0 last = 0 for i in range(n): sp = max(0.0, sp + s[i]) sm = min(0.0, sm + s[i]) if max(sp, -sm) > k: sp = sm = 0.0 last = i age[i] = i - last return age def _vratio(c: np.ndarray, w_days: int, bpd: int) -> np.ndarray: """Variance ratio VR = Var(r_q) / (q * Var(r_1)) su finestra w_days. >1 = persistente.""" lr = _logret(c) qb = max(2, VR_Q_DAYS * bpd) lg = np.log(c) lrq = np.zeros(len(c)) lrq[qb:] = lg[qb:] - lg[:-qb] lrq[:qb] = np.nan wb = max(20, w_days * bpd) v1 = pd.Series(lr).rolling(wb, min_periods=wb // 2).var().values vq = pd.Series(lrq).rolling(wb, min_periods=wb // 2).var().values with np.errstate(invalid="ignore", divide="ignore"): vr = vq / (qb * v1) return vr def _hurst(c: np.ndarray, w_days: int, bpd: int) -> np.ndarray: """Hurst da scaling della volatilita': sd(r_k) ~ k^H. Regressione di log sd su log k (x fisso -> forma chiusa). Tutto rolling causale.""" lg = np.log(c) wb = max(20, w_days * bpd) xs, ys = [], [] for kd in HURST_K_DAYS: kb = max(1, kd * bpd) d = np.full(len(c), np.nan) d[kb:] = lg[kb:] - lg[:-kb] sd = pd.Series(d).rolling(wb, min_periods=wb // 2).std().values with np.errstate(invalid="ignore", divide="ignore"): ys.append(np.log(np.where(sd > 0, sd, np.nan))) xs.append(np.log(kb)) X = np.asarray(xs, float) Y = np.vstack(ys) # (n_scale, n_bar) xm = X.mean() denom = float(((X - xm) ** 2).sum()) with warnings.catch_warnings(): # warmup = colonne tutte-NaN warnings.simplefilter("ignore", RuntimeWarning) ym = np.nanmean(Y, axis=0) H = ((X - xm)[:, None] * (Y - ym[None, :])).sum(axis=0) / denom return H _STAT_CACHE: dict = {} def _stat_key(df: pd.DataFrame, det: str, par) -> tuple: ts = df["timestamp"].values return (det, par, len(df), int(ts[0]), int(ts[-1])) def lookback_bars(df: pd.DataFrame, det: str, par, mapping: str) -> np.ndarray: """L_t in BARRE, dentro [Lmin, Lmax] giorni. Cache chiusa sul CONTENUTO del frame (lunghezza + primo/ultimo timestamp) -> un prefisso NON prende la cache del pieno, altrimenti causality_ok verrebbe ingannata dalla cache stessa.""" bpd = A.bars_per_day(df) lo, hi = max(2, LMIN_D * bpd), max(3, LMAX_D * bpd) c = df["close"].values.astype(float) key = _stat_key(df, det, par) if key not in _STAT_CACHE: if det == "cusum": _STAT_CACHE[key] = _cusum_age(c, float(par), bpd) elif det == "vratio": _STAT_CACHE[key] = _roll_pctl(_vratio(c, int(par), bpd), PCTL_M_DAYS * bpd) elif det == "hurst": _STAT_CACHE[key] = _roll_pctl(_hurst(c, int(par), bpd), PCTL_M_DAYS * bpd) else: raise ValueError(det) stat = _STAT_CACHE[key] if det == "cusum": base = np.clip(stat, lo, hi) L = base if mapping == "direct" else (lo + hi - base) else: u = np.clip(stat, 0.0, 1.0) L = lo + u * (hi - lo) if mapping == "direct" else hi - u * (hi - lo) return np.clip(np.round(L), 2, hi).astype(int) # =========================================================================== # DIREZIONE TSMOM (orizzonte costante o variabile) + vol-target # =========================================================================== 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): """TSMOM a orizzonte COSTANTE L giorni, STESSA geometria del candidato adattivo. blend='triple' -> (L/3, L, 2L): con L=90 e' ESATTAMENTE TP01 canonico 30/90/180.""" 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) d = np.clip(_dir_from_L(c, Lb, blend), 0, None) return A.vol_target(d, df, TGT_VOL, VOLWIN, LEV) fn.__name__ = f"const_L{L}_{blend}" return fn def factory(tf: str, det: str, par, mapping: str, blend: str): """target_fn(df) -> posizione per barra, decisa con dati <= close[i].""" def fn(df: pd.DataFrame) -> np.ndarray: c = df["close"].values.astype(float) L = lookback_bars(df, det, par, mapping) d = np.clip(_dir_from_L(c, L, blend), 0, None) return A.vol_target(d, df, TGT_VOL, VOLWIN, LEV) fn.__name__ = f"adapt_{det}{par}_{mapping}_{blend}" return fn GRID = [dict(det=d, par=p, mapping=m, blend=b) for d in DETECTORS for p in PARAMS[d] for m in MAPPINGS for b in BLENDS] # CONTROLLO dichiarato DOPO aver visto che il vincitore ha L_t incollato al bordo (sezione 3): # il null "TSMOM fisso 30/90/180" NON basta piu', perche' confonde due cose diverse — # "adattare l'orizzonte paga" e "un orizzonte piu' CORTO paga su questo campione". # Il controllo giusto e' il migliore fra gli orizzonti COSTANTI della stessa banda, scelto # in-sample-only. Trial aggiuntivi, contati al rialzo. CONST_L = (20, 30, 45, 60, 90, 120, 150, 200) CONST_GRID = [dict(L=L, blend=b) for L in CONST_L for b in BLENDS] # ESTENSIONE dichiarata DOPO aver visto che l'ottimo costante cade sul BORDO BASSO della banda # (L=20g = LMIN_D): un ottimo al bordo non e' un ottimo, e' il punto in cui si e' smesso di # guardare. Si estende SOTTO e si guarda se e' un plateau o una scivolata verso il rumore. # +4 valori x 1 blend x 2 TF = 8 trial in piu' (contati al rialzo). CONST_L_EXT = (5, 7, 10, 15) # =========================================================================== # UTILITA' DI MISURA # =========================================================================== 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 def resample_offset(df_1h: pd.DataFrame, hours: int, off: int) -> pd.DataFrame: """Stessa convenzione di trend_portfolio.resample_tf, con ancora spostata di `off` ore (r0702_tp01_offset.py). NB: si usa `offset=`, non `origin=` (pandas ignora origin su 7D).""" 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 daily_from_target(fn, tf: str) -> pd.Series: return A.candidate_daily(fn, tf=tf) 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), ret_cand=round(float(np.prod(1 + g["C"].values) - 1) * 100, 1), ret_base=round(float(np.prod(1 + g["B"].values) - 1) * 100, 1))) return rows def delever_null(cand: pd.Series, base: pd.Series) -> dict: """Il fisso, DE-LEVERATO a pari maxDD del candidato, che Sharpe fa? Scalare le POSIZIONI per k scala gross e fee insieme -> il netto scala per k e lo Sharpe e' INVARIANTE: percio' un candidato che compra solo meno DD non e' un miglioramento.""" 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), dd_base_delev=round(_dd(k * J["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: """Serie giornaliera 50/50 BTC+ETH della strategia `fn` ribilanciata sull'ancora `off` (resample del 1h certificato). Cache per (name, hours, off).""" 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, mapping: str) -> dict: """Quanto si MUOVE davvero L_t (diagnostica di MECCANISMO, non di rendimento). Una cella con L_t incollato a un bordo NON e' adattiva: e' un TSMOM a orizzonte COSTANTE travestito, e non appartiene alla famiglia che si sta testando.""" fr_lo, fr_hi, iqr, med = [], [], [], [] for a in ASSETS: df = A.get(a, tf) bpd = A.bars_per_day(df) L = lookback_bars(df, det, par, mapping) / bpd fr_lo.append(float(np.mean(L <= LMIN_D + 1))) fr_hi.append(float(np.mean(L >= LMAX_D - 1))) iqr.append(float(np.percentile(L, 75) - np.percentile(L, 25))) med.append(float(np.median(L))) return dict(frac_lo=float(np.mean(fr_lo)), frac_hi=float(np.mean(fr_hi)), iqr_days=float(np.mean(iqr)), med_days=float(np.mean(med)), adattiva=bool(np.mean(fr_lo) < 0.5 and np.mean(fr_hi) < 0.5 and np.mean(iqr) >= 10.0)) def hr(t: str = "") -> None: print("\n" + "=" * 78) if t: print(t) print("=" * 78) # =========================================================================== def main() -> None: print(__doc__.split("Runtime atteso")[0]) # -- 0. SANITY: il mio baseline DEVE essere TP01 canonico, bit-per-bit ---------------- hr("0. SANITY — il baseline 'fisso' e' TP01 canonico (se non lo e', ogni Δ 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 = daily_from_target(fixed_target, "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 - al.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" # -- 1. GRIGLIA ONESTA: cella scelta IN-SAMPLE + deflated-Sharpe ---------------------- hr(f"1. study_family_honest — {len(GRID)} celle x {len(TFS)} TF = {len(GRID)*len(TFS)} trial") rep = A.study_family_honest("ADAPT-H", factory, GRID, TFS) ch = rep["chosen"] print(f" celle valutate : {rep['n_cells']}") print(f" cella scelta IN-SAMPLE : tf={ch['tf']} {ch['params']} " 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-8 IN-SAMPLE (come si sceglie onestamente):") for r in rep["rows"][:8]: print(f" IS {r['insample_sharpe']:+.3f} FULL {r['full_sharpe']:+.3f} " f"{r['tf']:>3} {r['params']}") # sensibilita' del verdetto al CONTEGGIO dei trial (regola 30/07) fn_ch = factory(tf=ch["tf"], **ch["params"]) d_ch = daily_from_target(fn_ch, ch["tf"]) allsr = [r["full_sharpe"] for r in rep["rows"]] print("\n sensibilita' del DSR al numero di trial dichiarati:") for n_decl in (len(GRID) * len(TFS), 144, 216): pad = allsr + [float(np.mean(allsr))] * max(0, n_decl - len(allsr)) dsr, _ = A.deflated_sharpe(A._sh(d_ch), pad, d_ch) print(f" N={n_decl:>4} DSR={dsr:.3f} {'PASS' if dsr >= 0.95 else 'FAIL'}") # cosa avrebbe scelto un disonesto (diagnostica, NON una decisione) diag = [] for r in rep["rows"][:12]: s_all = daily_from_target(factory(tf=r["tf"], **r["params"]), r["tf"]) diag.append({**r, "hold": round(A._sh(s_all[s_all.index >= HOLDOUT]), 3)}) diag.sort(key=lambda r: -r["hold"]) print("\n (diagnostica) le stesse 12 celle riordinate per HOLD-OUT — cio' che NON si fa:") for r in diag[:4]: print(f" HOLD {r['hold']:+.3f} IS {r['insample_sharpe']:+.3f} {r['tf']:>3} {r['params']}") # -- 2. NUMERI del candidato vs il FISSO a pari geometria ----------------------------- hr("2. CANDIDATO vs TSMOM FISSO 30/90/180 — stessa geometria, stesso TF, stesse fee") base_ch = daily_from_target(fixed_target, ch["tf"]) J = pd.concat({"C": d_ch, "B": base_ch}, axis=1, join="inner").dropna() for lab, sl in (("FULL", J), ("HOLD-OUT", J[J.index >= HOLDOUT])): if len(sl) < 30: continue print(f" {lab:<9} adattivo Sh {A._sh(sl['C']):+.3f} DD {_dd(sl['C'])*100:5.2f}% " f"CAGR {_cagr(sl['C'])*100:6.2f}% | fisso Sh {A._sh(sl['B']):+.3f} " f"DD {_dd(sl['B'])*100:5.2f}% CAGR {_cagr(sl['B'])*100:6.2f}% " f"| dSh {A._sh(sl['C'])-A._sh(sl['B']):+.3f}") print(f" corr(adattivo, fisso) = {J['C'].corr(J['B']):.3f}") print("\n DIFFERENZA APPAIATA PER ANNO (un vantaggio in un anno solo e' un evento):") rows = paired_by_year(d_ch, base_ch) for r in rows: print(f" {r['year']} adatt Sh {r['sh_cand']:+.2f} ({r['ret_cand']:+7.1f}%) " f"fisso Sh {r['sh_base']:+.2f} ({r['ret_base']:+7.1f}%) dSh {r['d_sh']:+.2f}") ds = [r["d_sh"] for r in rows] print(f" -> anni con dSh>0: {sum(1 for x in ds if x > 0)}/{len(ds)} " f"mediana dSh {np.median(ds):+.2f} media {np.mean(ds):+.2f}") # -- 3. LOOKBACK: si muove davvero? --------------------------------------------------- hr("3. IL LOOKBACK ADATTIVO SI MUOVE? (se sta al bordo e' un TSMOM COSTANTE travestito)") for a in ASSETS: df = A.get(a, ch["tf"]) bpd = A.bars_per_day(df) L = lookback_bars(df, ch["params"]["det"], ch["params"]["par"], ch["params"]["mapping"]) / bpd lo, hi = LMIN_D, LMAX_D print(f" {a}: L_t giorni min {L.min():.0f} p25 {np.percentile(L,25):.0f} " f"mediana {np.median(L):.0f} p75 {np.percentile(L,75):.0f} max {L.max():.0f} " f"| al bordo basso {np.mean(L<=lo+1)*100:.1f}% al bordo alto {np.mean(L>=hi-1)*100:.1f}%" f" | |dL|/giorno mediano {np.median(np.abs(np.diff(L))):.1f}g") # -- 4. GATE --------------------------------------------------------------------------- hr("4. GATE") 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}") dl = delever_null(d_ch, base_ch) print(f" null del de-levering : {dl}") # small-cap a $600 print(" haircut a $600 (min-order $5):") for a in ASSETS: df = A.get(a, ch["tf"]) sc_c = A.eval_weights_smallcap(df, A._call_target(fn_ch, df, a), capital=600.0) sc_b = A.eval_weights_smallcap(df, fixed_target(df), capital=600.0) print(f" {a}: adattivo modellato {sc_c['modeled']['sharpe']:+.3f} -> reale " f"{sc_c['realistic']['sharpe']:+.3f} (haircut {sc_c['sharpe_haircut']:+.3f}, " f"{sc_c['n_executed_trades']} trade) | fisso {sc_b['modeled']['sharpe']:+.3f} -> " f"{sc_b['realistic']['sharpe']:+.3f} (haircut {sc_b['sharpe_haircut']:+.3f}, " f"{sc_b['n_executed_trades']} trade)") # -- 5. BANDA D'ANCORA ----------------------------------------------------------------- hours = 24 if ch["tf"] == "1d" else 12 offs = list(range(hours)) hr(f"5. BANDA D'ANCORA — {len(offs)} ancore, stima onesta = MEDIANA (mai l'ancora 00:00)") def _series_at(off: int, which: str) -> pd.Series: return series_at_anchor(fn_ch if which == "cand" else fixed_target, which, hours, off) # sanity: l'ancora 0 riproduce la serie canonica s0 = _series_at(0, "fixed") 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(o, "cand"), offs, metric=mtr) b_f = A.anchor_luck_band(lambda o: _series_at(o, "fixed"), offs, metric=mtr) print(f" {lab:<12} adattivo: canonica {b_c['canonical']:+.3f} (pctl {b_c['canonical_pctl']:.2f}) " f"MEDIANA {b_c['median']:+.3f} banda [{b_c['lo']:+.3f},{b_c['hi']:+.3f}]") print(f" {'':<12} fisso : canonica {b_f['canonical']:+.3f} (pctl {b_f['canonical_pctl']:.2f}) " f"MEDIANA {b_f['median']:+.3f} banda [{b_f['lo']:+.3f},{b_f['hi']:+.3f}]") dlt = A.anchor_luck_delta(lambda o: _series_at(o, "cand"), lambda o: _series_at(o, "fixed"), offs, metric=mtr) print(f" {'':<12} DELTA APPAIATO: mediana {dlt['median_paired']:+.3f} " f"positivo in {dlt['n_positive']}/{dlt['n_anchors']} ancore " f"banda [{dlt['lo']:+.3f},{dlt['hi']:+.3f}] gate_pass={dlt['gate_pass']}") # -- 6. ADATTIVITA' EFFETTIVA DI TUTTA LA FAMIGLIA ------------------------------------ hr("6. QUANTE DELLE 72 CELLE SONO DAVVERO ADATTIVE? (filtro di MECCANISMO, non di resa)") print(" criterio: L_t sta a un bordo <50% del tempo E l'IQR di L_t e' >= 10 giorni") seen, adatt_rows = {}, [] for r in rep["rows"]: k = (r["tf"], r["params"]["det"], r["params"]["par"], r["params"]["mapping"]) if k not in seen: seen[k] = adaptivity(*k) r["_adapt"] = seen[k] if seen[k]["adattiva"]: adatt_rows.append(r) n_ad_cells = sum(1 for r in rep["rows"] if r["_adapt"]["adattiva"]) print(f" celle DAVVERO adattive: {n_ad_cells}/{len(rep['rows'])} " f"(le altre sono TSMOM a orizzonte COSTANTE travestito)") for k, v in sorted(seen.items()): print(f" {k[0]:>3} {k[1]:>6} {str(k[2]):>5} {k[3]:<7} bordo-basso {v['frac_lo']*100:5.1f}% " f"bordo-alto {v['frac_hi']*100:5.1f}% IQR {v['iqr_days']:5.1f}g " f"mediana {v['med_days']:5.1f}g -> {'ADATTIVA' if v['adattiva'] else 'costante'}") # -- 7. IL CONTROLLO CHE MANCAVA: il MIGLIOR ORIZZONTE COSTANTE ------------------------ hr(f"7. CONTROLLO L COSTANTE — {len(CONST_GRID)} celle x {len(TFS)} TF = " f"{len(CONST_GRID)*len(TFS)} trial IN PIU'") print(" la domanda diventa: il guadagno viene dall'ADATTARE, o dal fatto che su questo") print(" campione un orizzonte piu' CORTO batte 30/90/180 anche se lo si tiene fisso?") csel = A.select_cell_insample(lambda tf, L, blend: const_factory(tf, L, blend), CONST_GRID, TFS) cch = csel["chosen"] print(f"\n miglior COSTANTE scelto in-sample: tf={cch['tf']} {cch['params']} " f"Sharpe IS {cch['insample_sharpe']} FULL {cch['full_sharpe']}") print(" griglia costante, in-sample (L=90 triple == TP01 canonico):") for r in csel["rows"]: tag = " <== TP01 canonico" 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(**cch["params"], tf=cch["tf"]) d_const = daily_from_target(fn_const, cch["tf"]) JC = pd.concat({"C": d_ch, "K": d_const}, axis=1, join="inner").dropna() print(f"\n ADATTIVO vs MIGLIOR COSTANTE (entrambi scelti in-sample):") for lab, sl in (("FULL", JC), ("HOLD-OUT", JC[JC.index >= HOLDOUT])): if len(sl) < 30: continue print(f" {lab:<9} adattivo Sh {A._sh(sl['C']):+.3f} DD {_dd(sl['C'])*100:5.2f}% | " f"costante Sh {A._sh(sl['K']):+.3f} DD {_dd(sl['K'])*100:5.2f}% | " f"dSh {A._sh(sl['C'])-A._sh(sl['K']):+.3f}") print(f" corr(adattivo, miglior costante) = {JC['C'].corr(JC['K']):.3f}") print("\n differenza APPAIATA per anno vs il miglior COSTANTE:") rows2 = paired_by_year(d_ch, d_const) for r in rows2: print(f" {r['year']} adatt {r['sh_cand']:+.2f} costante {r['sh_base']:+.2f} " f"dSh {r['d_sh']:+.2f}") d2 = [r["d_sh"] for r in rows2] print(f" -> anni con dSh>0: {sum(1 for x in d2 if x > 0)}/{len(d2)} " f"mediana {np.median(d2):+.2f}") dl2 = delever_null(d_ch, d_const) print(f" null del de-levering vs costante: {dl2}") # ancora appaiata adattivo vs miglior costante (se stesso TF, altrimenti dichiarato) if cch["tf"] == ch["tf"]: for lab, mtr in (("Sharpe FULL", A._sh), ("Sharpe HOLD", lambda s: A._sh(s[s.index >= HOLDOUT]))): dd2 = A.anchor_luck_delta(lambda o: series_at_anchor(fn_ch, "cand", hours, o), lambda o: series_at_anchor(fn_const, "const", hours, o), offs, metric=mtr) print(f" ancora {lab}: DELTA APPAIATO mediana {dd2['median_paired']:+.3f} " f"positivo in {dd2['n_positive']}/{dd2['n_anchors']} " f"banda [{dd2['lo']:+.3f},{dd2['hi']:+.3f}] gate_pass={dd2['gate_pass']}") else: print(f" (ancora appaiata non girata: TF diversi, adattivo {ch['tf']} vs " f"costante {cch['tf']} -> le ancore non sono coppie)") # fee sweep: l'adattivo tradà di piu', il conto va fatto print("\n FEE SWEEP (il candidato ha piu' turnover: e' li' che muore o no)") for nm, f_ in (("adattivo", fn_ch), ("costante", fn_const), ("TP01 30/90/180", fixed_target)): tfx = ch["tf"] if nm != "costante" else cch["tf"] line = [] for fee in (0.0, 0.0005, 0.001, 0.0015): dser = A.candidate_daily(f_, tf=tfx, fee_side=fee) line.append(f"{2*fee*100:.2f}%RT {A._sh(dser):+.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:<16} " + " ".join(line) + f" | turnover {tt:.0f}x/anno") # -- 8. LA FAMIGLIA RISTRETTA ALLE CELLE DAVVERO ADATTIVE ----------------------------- hr("8. GATE ONESTO SULLA SOLA FAMIGLIA ADATTIVA (la domanda della missione, isolata)") if not adatt_rows: print(" NESSUNA cella della famiglia e' davvero adattiva -> la domanda non e'") print(" misurabile su questa famiglia: ogni 'vincitore' e' un orizzonte costante.") else: best_ad = max(adatt_rows, key=lambda r: r["insample_sharpe"]) print(f" celle adattive: {len(adatt_rows)} miglior IN-SAMPLE: tf={best_ad['tf']} " f"{best_ad['params']} IS {best_ad['insample_sharpe']} FULL {best_ad['full_sharpe']}") fn_ad = factory(tf=best_ad["tf"], **best_ad["params"]) d_ad = daily_from_target(fn_ad, best_ad["tf"]) sm_ad = A.study_marginal("ADAPT-H(solo adattive)", fn_ad, tf=best_ad["tf"]) dsr_ad, sr0_ad = A.deflated_sharpe(A._sh(d_ad), [r["full_sharpe"] for r in adatt_rows], d_ad) print(f" marginal={sm_ad['marginal_verdict']} earns_slot={sm_ad['earns_slot']} " f"corr->TP01 {sm_ad['marginal'].get('corr_full')}") print(f" deflated-Sharpe (N={len(adatt_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() for lab, sl in (("FULL", JA), ("HOLD-OUT", JA[JA.index >= HOLDOUT])): if len(sl) < 30: continue print(f" {lab:<9} adattiva-vera Sh {A._sh(sl['C']):+.3f} | miglior costante " f"{A._sh(sl['K']):+.3f} | TP01 {A._sh(sl['B']):+.3f}") rows3 = paired_by_year(d_ad, d_const) d3 = [r["d_sh"] for r in rows3] print(f" vs miglior costante, per anno: dSh>0 in {sum(1 for x in d3 if x>0)}/{len(d3)}" f" mediana {np.median(d3):+.2f}") # -- 9. L'OTTIMO COSTANTE E' AL BORDO: si estende la banda ---------------------------- hr("9. L'OTTIMO COSTANTE CADE SUL BORDO — plateau o scivolata verso il rumore?") print(" un argmax al bordo della griglia non e' un ottimo: e' il punto in cui si e' smesso") print(" di guardare. Si estende SOTTO i 20 giorni (8 trial in piu', contati al rialzo).") ext = sorted(set(CONST_L) | set(CONST_L_EXT)) tab: dict = {} for tfx in TFS: print(f"\n --- TF {tfx}, blend triple (L/3, L, 2L) ---") print(f" {'L(g)':>5} {'orizzonti':>16} {'IS':>7} {'FULL':>7} {'HOLD':>7} " f"{'maxDD':>7} {'CAGR':>7} {'turn/a':>7} {'Sh@0.30%RT':>11}") for L in ext: fnx = const_factory(tfx, L, "triple") dx = daily_from_target(fnx, tfx) dx_h = dx[dx.index >= HOLDOUT] dx_i = dx[dx.index < HOLDOUT] d030 = A.candidate_daily(fnx, tf=tfx, fee_side=0.0015) tt = np.mean([A.eval_weights(A.get(a, tfx), A._call_target(fnx, A.get(a, tfx), a)) ["turnover_per_year"] for a in ASSETS]) tag = " <== TP01" if L == 90 else "" tab[(tfx, L)] = dict(IS=A._sh(dx_i), FULL=A._sh(dx), HOLD=A._sh(dx_h)) print(f" {L:>5} {f'{L//3}/{L}/{2*L}':>16} {A._sh(dx_i):>7.3f} {A._sh(dx):>7.3f} " f"{A._sh(dx_h):>7.3f} {_dd(dx)*100:>6.2f}% {_cagr(dx)*100:>6.2f}% " f"{tt:>7.0f} {A._sh(d030):>11.3f}{tag}") # --- e' una CURVA con un ottimo, o un crinale di RUMORE? -------------------------- print("\n RISOLUZIONE della griglia (regola 07/08: un criterio che decide su una frazione") print(" dello spread della griglia e' rumore anche quando 'passa'):") for tfx in TFS: v = [tab[(tfx, L)]["IS"] for L in ext] jump = float(np.median(np.abs(np.diff(v)))) arg = ext[int(np.argmax(v))] arg_orig = max(CONST_L, key=lambda L: tab[(tfx, L)]["IS"]) print(f" {tfx:>3}: argmax IS con la griglia ORIGINALE L={arg_orig}g -> con la griglia " f"ESTESA L={arg}g ({'SPOSTATO' if arg != arg_orig else 'stabile'})") print(f" salto IS mediano fra L ADIACENTI {jump:+.3f} contro il vantaggio del " f"vincitore su TP01 {max(v) - tab[(tfx, 90)]['IS']:+.3f} " f"(rapporto {jump / max(1e-9, max(v) - tab[(tfx, 90)]['IS']):.2f})") allc = [(tf, L) for tf in TFS for L in ext] for met in ("IS", "FULL", "HOLD"): vals = sorted((tab[k][met] for k in allc), reverse=True) r1d = 1 + sum(1 for x in vals if x > tab[("1d", 90)][met]) print(f" rango di TP01 (1d, L=90g) fra le {len(allc)} celle su {met}: " f"{r1d}/{len(allc)} (suo {tab[('1d',90)][met]:+.3f}, migliore {vals[0]:+.3f})") hr("FINE") if __name__ == "__main__": main()