Adriano/PythagorasGoal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

research(intraday): de-bias del lead ERM (filone B) — falso positivo + gate selection-on-holdout

Browse Source 
L'analisi di robustezza affonda lo "earns_slot=True" di ERM: era prodotto da
selezione-sull'hold-out + coda 2026 + multiple-testing non corretto.
  A) deflated-Sharpe FAIL: 0.00 (tutti 122 trial) / 0.16 (no-TOD) / 0.24 (solo-ERM) << 0.95
  B) selezione in-sample-only -> ALTRA cella (long-flat, corr->TP01 0.53) = NEUTRAL, no slot
  C) ensemble del plateau (no cherry-pick) -> ADDS ma robust_oos=False -> no slot
  D) uplift FULL solo +0.10, negativo 2021/2022; uplift HOLD +0.30 concentrato nel 2026
=> ERM SCARTATO come sleeve. Conferma ennesima del soffitto BTC/ETH-direzionale ~1.3.

Lezione CODIFICATA in altlib (LESSON 4, test in tests/test_harness_realism.py):
  - deflated_sharpe()       Bailey & Lopez de Prado, PASS >= 0.95
  - select_cell_insample()  scelta cella col solo Sharpe pre-HOLDOUT (no peeking)
  - study_family_honest()   gate combinato: earns_slot[cella in-sample] AND DSR>=0.95
Regola: una strategia direzionale grid-searched si giudica con study_family_honest,
non chiamando study_marginal sulla cella a max hold-out. Verificato end-to-end su ERM
(earns_slot_honest=False). Chiude il punto cieco gemello di CC01.

Diario aggiornato (verdetto downgrade), CLAUDE.md aggiornato. Test 119/119 verdi.
Nessun impatto live (branch separato).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Adriano Dal Pastro
2026-06-29 20:04:36 +00:00
parent aad69f9790
commit b4ec92734c
5 changed files with 471 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/research/alt/altlib.py
+90
View File
@@ -29,12 +29,14 @@ from __future__ import annotations
import inspect
import json
import math
import sys
from functools import lru_cache
from pathlib import Path
import numpy as np
import pandas as pd
from scipy.stats import norm
# --- make `from src...` work no matter where the agent's script lives -------
_ROOT = Path(__file__).resolve().parents[3]
@@ -670,6 +672,94 @@ def causality_ok(target_fn, tf: str = "1h", assets=CERTIFIED,
reason=("length-mismatch on prefix" if bad else None))
# ===========================================================================
# SELECTION-ON-HOLDOUT GATE — codified 2026-06-29 from filone B (intraday ERM).
#
# LESSON 3: intraday_regime.py picked its "winner" cell by MAX hold-out Sharpe over a ~60-cell
# grid, then ran study_marginal on THAT cell -> earns_slot=True. But the slot was an artifact of
# SELECTING THE CELL ON THE HOLD-OUT: picking the cell IN-SAMPLE-ONLY (no peeking) lands on a
# DIFFERENT, TP01-correlated cell that scores NEUTRAL, and the standalone Sharpe deflates to
# DSR~0.0-0.24 over the trials searched. study_marginal alone can't catch this — it judges ONE
# stream and never sees how the cell was chosen. The fix is two-fold and lives here:
# (1) choose the cell IN-SAMPLE-ONLY (or walk-forward) BEFORE scoring the marginal, and
# (2) DEFLATE the standalone Sharpe for the number of cells/families searched.
# Twin of the CC01 ("implausible Sharpe -> hidden risk") and alt-sweep ("hold-out-fitting") blind
# spots, in its "selection-on-holdout" form.
# ===========================================================================
def deflated_sharpe(sr_ann, all_sr_ann, daily_ret, dpy: float = 365.25):
"""Deflated Sharpe Ratio (Bailey & Lopez de Prado): P(true Sharpe > the MAX Sharpe expected
under the null of N independent trials). Penalizes multiple-testing — a standalone Sharpe ~1
over a 100+ cell grid is routinely NOT significant once deflated. sr_ann = annualized Sharpe
of the CHOSEN config; all_sr_ann = the Sharpe of EVERY cell searched; daily_ret = the chosen
config's daily returns (for skew/kurt/T). Returns (DSR, expected_null_max_sharpe_ann);
PASS if DSR >= 0.95."""
r = np.asarray(pd.Series(daily_ret).dropna().values, float)
T = len(r)
if T < 30 or np.std(r) == 0:
return float("nan"), float("nan")
sr = sr_ann / math.sqrt(dpy)
trials = np.asarray([s / math.sqrt(dpy) for s in all_sr_ann if np.isfinite(s)], float)
N = max(len(trials), 2)
var_tr = float(np.var(trials, ddof=1)) if N > 1 else 0.0
emc = 0.5772156649
z1 = norm.ppf(1 - 1.0 / N)
z2 = norm.ppf(1 - 1.0 / (N * math.e))
sr0 = math.sqrt(var_tr) * ((1 - emc) * z1 + emc * z2)
sk = float(pd.Series(r).skew())
ku = float(pd.Series(r).kurt()) + 3.0 # pandas kurt = excess
den = math.sqrt(max(1e-9, 1 - sk * sr + (ku - 1) / 4.0 * sr ** 2))
dsr = float(norm.cdf((sr - sr0) * math.sqrt(T - 1) / den))
return dsr, sr0 * math.sqrt(dpy)
def select_cell_insample(factory, grid, tfs, fee_side: float = FEE_SIDE) -> dict:
"""Pick a config WITHOUT looking at the hold-out: rank grid cells by IN-SAMPLE (pre-HOLDOUT)
standalone Sharpe of the 50/50 candidate. `factory(tf=..., **params)` -> target_fn; each grid
item is a dict of factory kwargs (besides tf). Returns the in-sample-best cell, all rows
(sorted), and EVERY cell's FULL Sharpe (for deflated_sharpe). This is the honest replacement
for picking the max-hold-out cell."""
rows = []
for tf in tfs:
for params in grid:
try:
daily = candidate_daily(factory(tf=tf, **params), tf=tf, fee_side=fee_side)
except Exception:
continue
ins = daily[daily.index < HOLDOUT]
is_sh = _sh(ins) if len(ins) > 60 else float("nan")
rows.append(dict(tf=tf, params=params, insample_sharpe=round(is_sh, 3),
full_sharpe=round(_sh(daily), 3)))
valid = [r for r in rows if np.isfinite(r["insample_sharpe"])]
chosen = max(valid, key=lambda r: r["insample_sharpe"]) if valid else None
return dict(chosen=chosen,
rows=sorted(valid, key=lambda r: r["insample_sharpe"], reverse=True),
all_full_sharpe=[r["full_sharpe"] for r in rows])
def study_family_honest(name: str, factory, grid, tfs, fee_side: float = FEE_SIDE,
dsr_min: float = 0.95) -> dict:
"""HARDENED family gate. A grid-searched directional candidate earns a slot ONLY if, picking
the cell IN-SAMPLE-ONLY (no hold-out peeking), it STILL earns_slot via study_marginal AND its
standalone Sharpe survives deflation for the WHOLE grid searched. Use this INSTEAD of
cherry-picking the max-hold cell and calling study_marginal on it."""
sel = select_cell_insample(factory, grid, tfs, fee_side=fee_side)
ch = sel["chosen"]
if ch is None:
return dict(name=name, chosen=None, earns_slot_honest=False,
reason="no valid in-sample cell")
fn = factory(tf=ch["tf"], **ch["params"])
sm = study_marginal(f"{name} ISpick {ch['params']}", fn, tf=ch["tf"], fee_side=fee_side)
daily = candidate_daily(fn, tf=ch["tf"], fee_side=fee_side)
dsr, sr0 = deflated_sharpe(_sh(daily), sel["all_full_sharpe"], daily)
dsr_pass = bool(np.isfinite(dsr) and dsr >= dsr_min)
return dict(name=name, n_cells=len(sel["all_full_sharpe"]), chosen=ch, rows=sel["rows"],
marginal=sm, earns_slot_marginal=bool(sm["earns_slot"]),
deflated_sharpe=round(dsr, 3) if np.isfinite(dsr) else None,
expected_null_max=round(sr0, 3) if np.isfinite(sr0) else None,
dsr_pass=dsr_pass,
earns_slot_honest=bool(sm["earns_slot"] and dsr_pass))
# ===========================================================================
# DRIVERS — run a hypothesis across both assets, several TFs, with a fee sweep.
# ===========================================================================
scripts/research/intraday_regime_analysis.py
+267
View File
@@ -0,0 +1,267 @@
"""intraday_regime_analysis.py — ANALISI DI ROBUSTEZZA del LEAD ERM (filone B) — 2026-06-29.
Il lead di B (ERM 8h L=2.0 thr=0.35 L/S) fa earns_slot=True, ma con 2 caveat NON quantificati
dallo script di scoperta `intraday_regime.py`:
(1) il VINCITORE e' selezionato per min_hold MASSIMO su ~60 celle -> selezione-sull'hold-out;
(2) il plateau hold-out e' a UNA SOLA RIGA (positivo solo a L~2.0; L>=2.5 va negativo sull'hold).
Insieme = rischio multiple-testing / overfit della finestra recente, mai deflazionato (a
differenza del filone C che ha il deflated-Sharpe).
Questo script attacca esattamente quei nodi, SENZA toccare il live (read-only, branch separato):
A) DEFLATED SHARPE (Bailey & Lopez de Prado) del vincitore vs TUTTI i trial realmente
cercati (ERM+VEM+VBR+TOD, tutte le celle/TF). Se DSR << 0.95 lo Sharpe non e' significativo
dopo la correzione per multiple-testing.
B) SELEZIONE IN-SAMPLE-ONLY: ri-scelgo la cella ERM usando SOLO lo Sharpe PRE-2025 (mai
l'hold-out), poi ne valuto earns_slot sull'intera storia. Se una cella scelta SENZA vedere
l'hold-out continua ad ADDS, l'edge non e' hold-out-mined.
C) ENSEMBLE DEL PLATEAU: invece della singola cella migliore, media i pesi su tutto il
plateau ERM 8h (L x thr) -> un candidato unico "non-cherry-picked" -> earns_slot. Se la
famiglia regge senza scegliere L, il caveat (1)+(2) si attenua.
D) DOVE VIVE L'EDGE: Sharpe per-anno standalone + uplift per-anno del blend 3-way
(TP01+SKH+ERM) vs 2-way (TP01+SKH), e corr(ERM,SKH) per-anno (e' un hedge-di-SKH?).
Esecuzione: uv run python scripts/research/intraday_regime_analysis.py
Idempotente, niente scritture su disco (solo report a stdout).
"""
from __future__ import annotations
import sys
import numpy as np
import pandas as pd
sys.path.insert(0, "/opt/docker/PythagorasGoal/scripts/research/alt")
import altlib as al # noqa: E402
from intraday_regime import make_erm, make_vem, make_vbr, make_tod, _skh_daily # noqa: E402
DPY = 365.25
HOLD = al.HOLDOUT
ASSETS = ("BTC", "ETH")
SCREEN_TFS = ("1h", "4h", "6h", "8h", "12h")
WIN = dict(tf="8h", L_days=2.0, thr=0.35, long_flat=False) # il lead di B
deflated_sharpe = al.deflated_sharpe # gate canonico (codificato in altlib da questo filone)
# ===========================================================================
def all_trials():
"""Ricostruisce la griglia COMPLETA realmente cercata in intraday_regime.py, ritorna una
lista di (tag, factory, tf, params) per pesare il multiple-testing onestamente."""
out = []
erm_grid = [dict(L_days=L, thr=t, long_flat=lf)
for L in (1.0, 2.0, 3.0) for t in (0.35, 0.50) for lf in (False, True)]
for tf in SCREEN_TFS:
for p in erm_grid:
out.append(("ERM", make_erm, tf, p))
# + il plateau fine a 8h (L x thr, lf=False) — anche quelle sono celle testate
for L in (1.5, 2.0, 2.5, 3.0):
for t in (0.30, 0.35, 0.40, 0.45, 0.50):
out.append(("ERM-plat", make_erm, "8h", dict(L_days=L, thr=t, long_flat=False)))
vem_grid = [dict(Lmom_days=lm, Lshort_days=2.0, Llong_days=10.0, long_flat=lf)
for lm in (1.0, 3.0) for lf in (False, True)]
for tf in ("4h", "6h", "8h", "12h"):
for p in vem_grid:
out.append(("VEM", make_vem, tf, p))
vbr_grid = [dict(k=k, atr_win=14, long_flat=lf) for k in (0.5, 1.0, 1.5) for lf in (False, True)]
for tf in ("4h", "6h", "8h", "12h"):
for p in vbr_grid:
out.append(("VBR", make_vbr, tf, p))
for lf in (False, True):
out.append(("TOD", make_tod, "1h", dict(long_flat=lf)))
return out
def cand_full_is_sharpe(factory, tf, params):
"""(daily, full Sharpe, in-sample<2025 Sharpe) del candidato 50/50 di quella cella."""
fn = factory(tf=tf, **params)
daily = al.candidate_daily(fn, tf=tf)
full = al._sh(daily)
ins = al._sh(daily[daily.index < HOLD]) if (daily.index < HOLD).sum() > 60 else float("nan")
return daily, full, ins
# ===========================================================================
def part_A_deflated():
print("=" * 78)
print("A) DEFLATED SHARPE del vincitore vs TUTTI i trial cercati (multiple-testing)")
print("=" * 78)
trials = all_trials()
sr_all, sr_no_tod, sr_erm = [], [], []
win_daily = win_full = None
for tag, factory, tf, params in trials:
try:
daily, full, _ = cand_full_is_sharpe(factory, tf, params)
except Exception:
continue
sr_all.append(full)
if tag != "TOD":
sr_no_tod.append(full)
if tag.startswith("ERM"):
sr_erm.append(full)
if tag == "ERM" and tf == WIN["tf"] and params.get("L_days") == WIN["L_days"] \
and params.get("thr") == WIN["thr"] and params.get("long_flat") is False:
win_daily, win_full = daily, full
sr_arr = np.array([s for s in sr_all if np.isfinite(s)])
print(f" N trial finiti : {len(sr_arr)}")
print(f" Sharpe winner (50/50) : {win_full:+.3f}")
print(f" Sharpe trial: mean {sr_arr.mean():+.2f} std {sr_arr.std():.2f} "
f"max {sr_arr.max():+.2f} >0: {int((sr_arr > 0).sum())}/{len(sr_arr)}")
dsr = None
for label, pool in (("TUTTI 122", sr_all), ("no-TOD", sr_no_tod), ("solo-ERM", sr_erm)):
d, sr0 = deflated_sharpe(win_full, pool, win_daily)
n = int(np.isfinite(np.array(pool)).sum())
print(f" DSR [{label:>9} N={n:>3}]: {d:.3f} (Sh-max null {sr0:+.2f}) -> "
f"{'PASS' if d >= 0.95 else 'FAIL'}")
if label == "TUTTI 122":
dsr = d
return dsr, win_full, None
# ===========================================================================
def part_B_insample_pick():
print("\n" + "=" * 78)
print("B) SELEZIONE IN-SAMPLE-ONLY (scelgo la cella ERM solo con Sharpe PRE-2025)")
print("=" * 78)
erm_grid = [dict(L_days=L, thr=t, long_flat=lf)
for L in (1.0, 1.5, 2.0, 2.5, 3.0) for t in (0.30, 0.35, 0.40, 0.50)
for lf in (False, True)]
rows = []
for tf in SCREEN_TFS:
for p in erm_grid:
try:
_, full, ins = cand_full_is_sharpe(make_erm, tf, p)
except Exception:
continue
rows.append((ins, full, tf, p))
rows = [r for r in rows if np.isfinite(r[0])]
rows.sort(key=lambda r: r[0], reverse=True) # ordina per Sharpe IN-SAMPLE (no hold-out)
print(f" Top 5 celle per Sharpe IN-SAMPLE (<2025):")
for ins, full, tf, p in rows[:5]:
print(f" IS {ins:+.2f} FULL {full:+.2f} tf={tf:>3} {p}")
ins, full, tf, p = rows[0]
print(f"\n -> cella scelta SENZA vedere l'hold-out: tf={tf} {p}")
sm = al.study_marginal(f"ERM-ISpick {p}", make_erm(tf=tf, **p), tf=tf)
m = sm["marginal"]
print(f" earns_slot={sm['earns_slot']} marginal={m['marginal_verdict']} "
f"abs={sm['absolute']['verdict']['grade']}")
print(f" corr->TP01 {m['corr_full']} has_insample_edge={m['has_insample_edge']} "
f"is_hedge={m['is_hedge']} robust_oos={m['robust_oos']}")
print(f" blend w25: full uplift {m['blends']['w25']['uplift_full']:+.3f} "
f"hold uplift {m['blends']['w25']['uplift_hold']:+.3f}")
same = (tf == WIN["tf"] and abs(p["L_days"] - WIN["L_days"]) < 1e-9
and abs(p["thr"] - WIN["thr"]) < 1e-9 and p["long_flat"] is False)
print(f" coincide col vincitore max-hold? {same}")
return sm["earns_slot"], (tf, p)
# ===========================================================================
def ensemble_target(tf, cells):
"""Media (equal-weight) dei pesi vol-targeted su piu' celle ERM -> un unico stream per asset.
Ritorna un target_fn(df) che ricostruisce l'ensemble per quel df."""
def fn(df):
ws = [make_erm(tf=tf, **c)(df) for c in cells]
return np.nanmean(np.vstack(ws), axis=0)
return fn
def part_C_plateau_ensemble():
print("\n" + "=" * 78)
print("C) ENSEMBLE DEL PLATEAU ERM 8h (media celle L x thr, NIENTE cherry-pick)")
print("=" * 78)
cells = [dict(L_days=L, thr=t, long_flat=False)
for L in (1.5, 2.0, 2.5, 3.0) for t in (0.30, 0.35, 0.40, 0.45, 0.50)]
fn = ensemble_target("8h", cells)
print(f" celle nell'ensemble: {len(cells)} (L 1.5-3.0 x thr 0.30-0.50, lf=False)")
caus = al.causality_ok(fn, tf="8h")
print(f" causale: ok={caus['ok']} max_tail_diff={caus['max_tail_diff']}")
sm = al.study_marginal("ERM-plateau-ens", fn, tf="8h")
m = sm["marginal"]
print(f" earns_slot={sm['earns_slot']} marginal={m['marginal_verdict']} "
f"abs={sm['absolute']['verdict']['grade']}")
print(f" standalone full {m['cand_full_sharpe']} hold {m['cand_hold_sharpe']} "
f"in-sample {m.get('cand_insample_sharpe')}")
print(f" corr->TP01 {m['corr_full']} has_insample_edge={m['has_insample_edge']} "
f"is_hedge={m['is_hedge']} robust_oos={m['robust_oos']}")
print(f" blend w25: full {m['blends']['w25']['full']} (uplift "
f"{m['blends']['w25']['uplift_full']:+.3f}) hold {m['blends']['w25']['hold']} "
f"(uplift {m['blends']['w25']['uplift_hold']:+.3f})")
print(f" multicut: {m['multicut_uplift']}")
return sm["earns_slot"]
# ===========================================================================
def part_D_where_edge():
print("\n" + "=" * 78)
print("D) DOVE VIVE L'EDGE — per-anno standalone + uplift 3-way vs 2-way + corr(ERM,SKH)")
print("=" * 78)
fn = make_erm(**WIN)
cand = al.candidate_daily(fn, tf=WIN["tf"])
tp = al.tp01_baseline_daily()
skh = _skh_daily()
J = pd.concat({"T": tp, "S": skh, "C": cand}, axis=1, join="inner").dropna()
two = 0.75 * J["T"] + 0.25 * J["S"]
three = 0.60 * J["T"] + 0.25 * J["S"] + 0.15 * J["C"]
print(f" {'anno':>5} {'ERM Sh':>7} {'TP+SKH':>7} {'+ERM':>7} {'Δuplift':>8} "
f"{'corr(ERM,SKH)':>14} {'corr(ERM,TP)':>13}")
for y in sorted(set(J.index.year)):
sub = J[J.index.year == y]
if len(sub) < 40:
continue
s2 = 0.75 * sub["T"] + 0.25 * sub["S"]
s3 = 0.60 * sub["T"] + 0.25 * sub["S"] + 0.15 * sub["C"]
print(f" {y:>5} {al._sh(sub['C']):>+7.2f} {al._sh(s2):>+7.2f} {al._sh(s3):>+7.2f} "
f"{al._sh(s3) - al._sh(s2):>+8.2f} {sub['C'].corr(sub['S']):>+14.2f} "
f"{sub['C'].corr(sub['T']):>+13.2f}")
print(f" {'FULL':>5} {al._sh(J['C']):>+7.2f} {al._sh(two):>+7.2f} {al._sh(three):>+7.2f} "
f"{al._sh(three) - al._sh(two):>+8.2f} {J['C'].corr(J['S']):>+14.2f} "
f"{J['C'].corr(J['T']):>+13.2f}")
JH = J[J.index >= HOLD]
h2 = 0.75 * JH["T"] + 0.25 * JH["S"]
h3 = 0.60 * JH["T"] + 0.25 * JH["S"] + 0.15 * JH["C"]
print(f" {'HOLD':>5} {al._sh(JH['C']):>+7.2f} {al._sh(h2):>+7.2f} {al._sh(h3):>+7.2f} "
f"{al._sh(h3) - al._sh(h2):>+8.2f} {JH['C'].corr(JH['S']):>+14.2f} "
f"{JH['C'].corr(JH['T']):>+13.2f}")
def part_E_codified_gate():
"""Validazione END-TO-END del gate appena codificato in altlib: study_family_honest sulla
famiglia ERM deve dare earns_slot_honest=False (sceglie in-sample-only + deflaziona)."""
print("\n" + "=" * 78)
print("E) GATE CODIFICATO (al.study_family_honest) sulla famiglia ERM — deve bocciare")
print("=" * 78)
grid = [dict(L_days=L, thr=t, long_flat=lf)
for L in (1.0, 1.5, 2.0, 2.5, 3.0) for t in (0.30, 0.35, 0.40, 0.50)
for lf in (False, True)]
rep = al.study_family_honest("ERM", make_erm, grid, SCREEN_TFS)
ch = rep["chosen"]
print(f" n_celle={rep['n_cells']} cella in-sample-best: tf={ch['tf']} {ch['params']}")
print(f" earns_slot_marginal={rep['earns_slot_marginal']} "
f"deflated_sharpe={rep['deflated_sharpe']} (dsr_pass={rep['dsr_pass']})")
print(f" => earns_slot_HONEST = {rep['earns_slot_honest']} "
f"(atteso False: slot bocciato dal gate)")
return rep["earns_slot_honest"]
def main():
print("ANALISI ROBUSTEZZA LEAD ERM (filone B) — read-only, nessun impatto live\n")
dsr, win_full, sr0 = part_A_deflated()
es_is, is_cell = part_B_insample_pick()
es_ens = part_C_plateau_ensemble()
part_D_where_edge()
es_honest = part_E_codified_gate()
print("\n" + "=" * 78)
print("SINTESI ANALISI B")
print("=" * 78)
print(f" A) deflated-Sharpe winner = {dsr:.3f} ({'PASS' if dsr >= 0.95 else 'FAIL'} vs 0.95)")
print(f" B) cella scelta in-sample-only earns_slot = {es_is} (cella {is_cell})")
print(f" C) ensemble del plateau earns_slot = {es_ens}")
print(f" E) gate codificato earns_slot_honest = {es_honest}")
if __name__ == "__main__":
main()