"""r0717_wk_intra — WK-INTRA: strategie INTRADAY attive SOLO nella finestra weekend (ven 20:00 -> lun 12:00 UTC), BTC/ETH certificati, 1h e 15m. Domanda: il microclima weekend (liquidita' bassa, TradFi chiuso, riapertura CME dom ~22:00 UTC) crea pattern intraday sfruttabili NETTI fee (0.10% RT)? SOTTO-FILONI (ogni cella = filone x parametri x direzione x asset x TF; tutte le celle contano come trial per il deflated-Sharpe di famiglia): a. Donchian weekend-only: canale N-ore calcolato SOLO su barre sab/dom (strettamente precedenti, shiftate), breakout FOLLOW vs FADE, N in {12,24,48} ore. Entrate solo sab/dom, posizione chiusa forzatamente entro lun 12:00. b. Transizione dom->lun / riapertura CME: gap = close dom 22:00 vs close ven 21:00 UTC (chiusura CME equity). |gap| >= soglia {0.5,1,2%} -> REVERT (gap-fill) o CONT (continuazione), uscita lun alle {0,8,12} UTC. + cella riferimento always-long dom22->lun12 (drift incondizionato). NB: per costruzione il filone e' quasi TF-invariante (stessi prezzi di decisione a 1h e 15m). c. Livelli del venerdi' come magnete: H/L/C del venerdi' COMPLETO (noti da sab 00:00); sab/dom tocco/rottura del livello -> FOLLOW vs FADE; configurazioni HL (canale, hold fino a segnale opposto), C (sign(close-friC) per barra), H-only / L-only (posizione finche' oltre il livello). Chiusura forzata lun 12:00. PRIOR ART vincolante (NON rifatto qui): CRT/sweep-reclaim triplo-refutato (2026-07-02 / 07-07); "il ritest e' informazione negativa"; win-rate e' un knob -> qui tutto in Sharpe/ ritorni netti, nessun claim su WR; anchor timing-luck -> shift-test orario +/-2/4h. METODO (obbligatorio, CLAUDE.md): * causalita': segnale deciso con dati <= close[i], tenuto nella barra i+1 (eval_weights shifta; il gating di finestra usa il calendario della barra SUCCESSIVA, deterministico); * selezione SOLO in-sample (pre-2025, min-asset Sharpe netto) -> hold-out; banda min/med/max su tutte le celle, mai solo la best; * netto fee 0.10% RT + sweep {0, 0.05, 0.10, 0.15}%/side; lordo riportato per distinguere "muore di fee" da "non esiste"; * deflated_sharpe su TUTTI i trial della famiglia; shift-test +/-2/4h del calendario visto dal segnale (al._shift_calendar) = detector di artefatti di etichettatura; * per-anno sulla cella in-sample; eseguibilita' a $600 (ordini/settimana, quota <$5, eval_weights_smallcap). Esecuzione: uv run python scripts/research/r0717_wk_intra.py """ from __future__ import annotations import sys from functools import partial import numpy as np import pandas as pd sys.path.insert(0, "/opt/docker/PythagorasGoal/scripts/research/alt") import altlib as al # noqa: E402 ASSETS = ("BTC", "ETH") TFS = ("1h", "15m") FEE_SWEEP = (0.0, 0.0005, 0.001, 0.0015) # per-side OFFSETS = (-4, -2, 0, 2, 4) # shift-test ore # =========================================================================== # CALENDARIO + GATING (tutto causale: la finestra e' calendario deterministico) # =========================================================================== def _cal(df: pd.DataFrame) -> dict: dt = pd.to_datetime(df["datetime"], utc=True) step = float(dt.diff().dt.total_seconds().median()) ct = dt + pd.Timedelta(seconds=step) # istante di CHIUSURA della barra return dict(step_h=step / 3600.0, dt=dt, wd=dt.dt.dayofweek.values, hr=dt.dt.hour.values, mi=dt.dt.minute.values, cwd=ct.dt.dayofweek.values, chh=ct.dt.hour.values, cmm=ct.dt.minute.values) def _next(mask: np.ndarray) -> np.ndarray: """Gating sulla barra TENUTA: eval_weights tiene target[i] nella barra i+1, quindi target[i] = segnale(<=close[i]) * finestra(barra i+1). Calendario noto in anticipo.""" out = np.zeros(len(mask), dtype=float) out[:-1] = mask[1:].astype(float) return out # =========================================================================== # BUILDER (target continui, unita' +/-1; decisione a close[i]) # =========================================================================== def build_a(df: pd.DataFrame, n_hours: int, mode: str) -> np.ndarray: """Donchian su SOLE barre weekend (sab/dom), canale su N ore weekend strettamente precedenti. follow: close>hi -> +1, close -1 (fade: opposto). Hold fino a segnale opposto; flat forzato da lun 12:00. Reset a inizio weekend.""" C = _cal(df) c = df["close"].values.astype(float) h = df["high"].values.astype(float) l = df["low"].values.astype(float) bars = max(2, int(round(n_hours / C["step_h"]))) wk = (C["wd"] == 5) | (C["wd"] == 6) iw = np.where(wk)[0] hi = np.full(len(c), np.nan) lo = np.full(len(c), np.nan) if len(iw) > bars: hi[iw] = pd.Series(h[iw]).rolling(bars, min_periods=bars).max().shift(1).values lo[iw] = pd.Series(l[iw]).rolling(bars, min_periods=bars).min().shift(1).values sgn = 1.0 if mode == "follow" else -1.0 raw = np.full(len(c), np.nan) start = wk & ~np.roll(wk, 1) start[0] = wk[0] raw[start] = 0.0 # reset: niente carry tra weekend with np.errstate(invalid="ignore"): up = wk & (c > hi) dn = wk & (c < lo) raw[up] = sgn raw[dn] = -sgn sig = pd.Series(raw).ffill().fillna(0.0).values w = wk | ((C["wd"] == 0) & (C["hr"] < 12)) return sig * _next(w) def build_b(df: pd.DataFrame, thr: float, direction: str, h_exit: int) -> np.ndarray: """Gap dom 22:00 vs close ven 21:00 UTC. |gap|>=thr -> revert (-sign) o cont (+sign). Posizione dom 22:00 -> lun h_exit. direction='always' = riferimento long incondizionato.""" C = _cal(df) c = df["close"].values.astype(float) n = len(c) ref_mask = (C["cwd"] == 4) & (C["chh"] == 21) & (C["cmm"] == 0) ref = pd.Series(np.where(ref_mask, c, np.nan)).ffill().values dec = (C["cwd"] == 6) & (C["chh"] == 22) & (C["cmm"] == 0) raw = np.full(n, np.nan) di = np.where(dec & np.isfinite(ref))[0] if direction == "always": raw[di] = 1.0 else: g = c[di] / ref[di] - 1.0 d = np.where(np.abs(g) >= thr, np.sign(g), 0.0) if direction == "revert": d = -d raw[di] = d reset = (C["wd"] == 0) & (C["hr"] == 12) & (C["mi"] == 0) raw[reset & ~dec] = 0.0 sig = pd.Series(raw).ffill().fillna(0.0).values w = ((C["wd"] == 6) & (C["hr"] >= 22)) | ((C["wd"] == 0) & (C["hr"] < h_exit)) return sig * _next(w) def build_c(df: pd.DataFrame, level: str, mode: str) -> np.ndarray: """Livelli del venerdi' COMPLETO (H/L/C noti da sab 00:00), segnali su sab/dom: HL: rottura high -> +1 / low -> -1 (follow; fade opposto), hold fino a opposto. C : sign(close - friC) per barra. H: +/-1 finche' close>friH. L: -/+1 finche' close fH)] = sgn raw[wk & (c < fL)] = -sgn elif level == "C": m = wk & np.isfinite(fC) raw[m] = sgn * np.sign(c[m] - fC[m]) elif level == "H": m = wk & np.isfinite(fH) raw[m] = np.where(c[m] > fH[m], sgn, 0.0) elif level == "L": m = wk & np.isfinite(fL) raw[m] = np.where(c[m] < fL[m], -sgn, 0.0) sig = pd.Series(raw).ffill().fillna(0.0).values w = wk | ((C["wd"] == 0) & (C["hr"] < 12)) return sig * _next(w) # =========================================================================== # GRIGLIE (builder monoargomento -> compatibili con shift-test e causality_ok) # =========================================================================== def _one(fn, **kw): """Builder monoargomento fn(df): signature a 1 parametro, cosi' altlib._call_target (che ispeziona la signature) non prova a passare l'asset come 2o posizionale.""" def g(df): return fn(df, **kw) return g GRID = { "a": {f"N{n}h-{m}": _one(build_a, n_hours=n, mode=m) for n in (12, 24, 48) for m in ("follow", "fade")}, "b": {**{f"g{thr * 100:.1f}%-{d}-ex{hx:02d}": _one(build_b, thr=thr, direction=d, h_exit=hx) for thr in (0.005, 0.01, 0.02) for d in ("revert", "cont") for hx in (0, 8, 12)}, "alwayslong-ex12": _one(build_b, thr=0.0, direction="always", h_exit=12)}, "c": {f"{lvl}-{m}": _one(build_c, level=lvl, mode=m) for lvl in ("HL", "C", "H", "L") for m in ("follow", "fade")}, } # =========================================================================== # VALUTAZIONE (netto E lordo, split IS/HOLD, serie daily per DSR) # =========================================================================== def eval_cell(df: pd.DataFrame, tgt: np.ndarray, fee_side: float = al.FEE_SIDE) -> dict: c = df["close"].values.astype(float) r = al.simple_returns(c) t = np.nan_to_num(np.asarray(tgt, float)) pos = np.zeros(len(t)) pos[1:] = t[:-1] # tenuta in barra i+1 (stessa convenzione di eval_weights) gross_r = pos * r turn = np.abs(np.diff(pos, prepend=0.0)) idx = pd.DatetimeIndex(pd.to_datetime(df["datetime"], utc=True)) im = idx < al.HOLDOUT hm = ~im out = {} for tag, f in (("net", fee_side), ("gross", 0.0)): net = gross_r - f * turn net[0] = 0.0 out[tag] = dict(full=al._metrics_from_net(net, idx), ins=al._metrics_from_net(net[im], idx[im]) if im.sum() > 10 else None, hold=al._metrics_from_net(net[hm], idx[hm]) if hm.sum() > 10 else None) if tag == "net": out["yearly"] = al._yearly(net, idx) out["dser"] = al._to_daily(pd.Series(net, index=idx)) span_y = max((idx[-1] - idx[0]).days / 365.25, 1e-9) n_ord = int((turn > 1e-12).sum()) nz = turn[turn > 1e-12] out["orders_per_week"] = n_ord / (span_y * 52.18) out["frac_sub5_at600"] = float(np.mean(nz * 600.0 < 5.0)) if len(nz) else 0.0 out["turnover_py"] = float(turn.sum() / span_y) out["tim"] = float(np.mean(pos != 0)) return out def causality_tail_ok(builder, tf: str, tail: int = 80) -> dict: """Come al.causality_ok ma ESCLUDE l'ultima barra del prefisso troncato: il gating di finestra usa il calendario della barra SUCCESSIVA (deterministico, noto in anticipo in deploy), che sul prefisso non esiste -> al.causality_ok segna diff=1 sull'ultima barra anche senza alcun look-ahead sui PREZZI. Qui verifichiamo che TUTTE le altre barre del tail coincidano (vero test di leak sui dati di mercato) e riportiamo a parte la diff dell'ultima barra (attesa non-zero quando la finestra e' attiva).""" worst = 0.0 last = 0.0 checked = 0 for a in ASSETS: df = al.get(a, tf) full = np.nan_to_num(np.asarray(builder(df), float)) n = len(df) for cut in (int(n * 0.80), int(n * 0.92)): sub = df.iloc[:cut].reset_index(drop=True) s = np.nan_to_num(np.asarray(builder(sub), float)) if len(s) != cut: return dict(ok=False, reason="length-mismatch") d = np.abs(s[cut - tail:cut - 1] - full[cut - tail:cut - 1]) worst = max(worst, float(d.max()) if len(d) else 0.0) last = max(last, float(abs(s[cut - 1] - full[cut - 1]))) checked += 1 return dict(ok=bool(worst <= 1e-9), max_tail_diff=round(worst, 9), lastbar_gating_diff=round(last, 4), checked=checked) def _fmt(x, nd=2): if x is None or (isinstance(x, float) and not np.isfinite(x)): return " nan" return f"{x:+.{nd}f}" def _seg_sh(ev, tag, seg): d = ev[tag][seg] return d["sharpe"] if d else float("nan") def main() -> None: print("=" * 100) print(" WK-INTRA — strategie intraday nella finestra weekend (ven 20:00 -> lun 12:00 UTC)") print(" BTC/ETH certificati, 1h + 15m | fee 0.10% RT | selezione in-sample (pre-2025) -> hold-out") print("=" * 100) for a in ASSETS: for tf in TFS: df = al.get(a, tf) print(f" dati {a} {tf}: {len(df)} barre {df['datetime'].iloc[0]} .. {df['datetime'].iloc[-1]}") # ---- valuta TUTTE le celle -------------------------------------------------- rows = [] EVS = {} for f, grid in GRID.items(): for tf in TFS: for a in ASSETS: df = al.get(a, tf) for cfg, builder in grid.items(): ev = eval_cell(df, builder(df)) EVS[(f, tf, cfg, a)] = ev rows.append(dict( filone=f, tf=tf, cfg=cfg, asset=a, is_sh=_seg_sh(ev, "net", "ins"), hold_sh=_seg_sh(ev, "net", "hold"), full_sh=_seg_sh(ev, "net", "full"), is_shg=_seg_sh(ev, "gross", "ins"), hold_shg=_seg_sh(ev, "gross", "hold"), full_shg=_seg_sh(ev, "gross", "full"), dd=ev["net"]["full"]["maxdd"], opw=ev["orders_per_week"], tim=ev["tim"], dsh_full=al._sh(ev["dser"]))) R = pd.DataFrame(rows) n_cells = len(R) # combo 50/50 daily per config (trials like-for-like per il DSR) combo = {} for (f, tf, cfg), _ in R.groupby(["filone", "tf", "cfg"]).size().items(): b = EVS[(f, tf, cfg, "BTC")]["dser"] e = EVS[(f, tf, cfg, "ETH")]["dser"] J = pd.concat([b, e], axis=1, join="inner").fillna(0.0) combo[(f, tf, cfg)] = 0.5 * (J.iloc[:, 0] + J.iloc[:, 1]) family_combo_sh = [al._sh(s) for s in combo.values()] print(f"\n TRIAL DI FAMIGLIA: {n_cells} celle per-asset " f"({len(combo)} configurazioni x 2 asset) su 3 sotto-filoni") # ---- report per filone ------------------------------------------------------ FIL_DESC = {"a": "Donchian weekend-only (follow/fade, N=12/24/48h)", "b": "gap dom22 vs ven21 UTC (CME) revert/cont, exit lun {0,8,12}", "c": "livelli ven H/L/C, follow/fade dentro il weekend"} for f in ("a", "b", "c"): Rf = R[R.filone == f] print("\n" + "=" * 100) print(f" FILONE {f.upper()} — {FIL_DESC[f]} [{len(Rf)} celle]") print("=" * 100) # banda su tutte le celle per-asset for tag, k_is, k_h in (("NETTO", "is_sh", "hold_sh"), ("LORDO", "is_shg", "hold_shg")): print(f" banda {tag}: IS Sharpe min/med/max = " f"{Rf[k_is].min():+.2f} / {Rf[k_is].median():+.2f} / {Rf[k_is].max():+.2f}" f" (celle IS>0: {(Rf[k_is] > 0).sum()}/{len(Rf)})") print(f" HOLD Sharpe min/med/max = " f"{Rf[k_h].min():+.2f} / {Rf[k_h].median():+.2f} / {Rf[k_h].max():+.2f}" f" (celle HOLD>0: {(Rf[k_h] > 0).sum()}/{len(Rf)})") # selezione IN-SAMPLE-ONLY: max del min-asset IS Sharpe netto agg = Rf.groupby(["tf", "cfg"]).agg(min_is=("is_sh", "min"), min_hold=("hold_sh", "min"), min_isg=("is_shg", "min")).reset_index() agg = agg.sort_values("min_is", ascending=False) print("\n top-5 config per min-asset IS Sharpe NETTO (selezione solo in-sample):") for _, rr in agg.head(5).iterrows(): print(f" {rr['tf']:>3s} {rr['cfg']:<22s} minIS {rr['min_is']:+.2f} " f"(lordo {rr['min_isg']:+.2f}) -> minHOLD {rr['min_hold']:+.2f}") ch = agg.iloc[0] ctf, ccfg = ch["tf"], ch["cfg"] builder = GRID[f][ccfg] if f == "b": # riferimento: drift incondizionato dom22->lun12 print("\n riferimento always-long dom22->lun12 (drift incondizionato, netto):") for a in ASSETS: evr = EVS[(f, "1h", "alwayslong-ex12", a)] print(f" {a} 1h: IS {_fmt(_seg_sh(evr, 'net', 'ins'))} " f"FULL {_fmt(_seg_sh(evr, 'net', 'full'))} " f"HOLD {_fmt(_seg_sh(evr, 'net', 'hold'))} " f"(lordo IS {_fmt(_seg_sh(evr, 'gross', 'ins'))})") print(f"\n CELLA SCELTA (in-sample): {ctf} {ccfg}") for a in ASSETS: ev = EVS[(f, ctf, ccfg, a)] print(f" {a}: NETTO IS {_fmt(_seg_sh(ev,'net','ins'))} FULL {_fmt(_seg_sh(ev,'net','full'))} " f"(ret {ev['net']['full']['ret']*100:+.1f}%, DD {ev['net']['full']['maxdd']*100:.1f}%) " f"HOLD {_fmt(_seg_sh(ev,'net','hold'))} (ret {ev['net']['hold']['ret']*100:+.1f}%)" if ev['net']['hold'] else "") print(f" LORDO IS {_fmt(_seg_sh(ev,'gross','ins'))} FULL {_fmt(_seg_sh(ev,'gross','full'))} " f"HOLD {_fmt(_seg_sh(ev,'gross','hold'))} | TiM {ev['tim']*100:.1f}% " f"ordini/sett {ev['orders_per_week']:.2f} quota<$5@600$ {ev['frac_sub5_at600']*100:.0f}%") yr = " ".join(f"{y}:{d['ret']*100:+.1f}%" for y, d in ev["yearly"].items()) print(f" per-anno (netto): {yr}") # combo 50/50 + DSR (filone e famiglia) cd = combo[(f, ctf, ccfg)] cd_is = cd[cd.index < al.HOLDOUT] cd_h = cd[cd.index >= al.HOLDOUT] sh_cd = al._sh(cd) fil_sh = [al._sh(s) for (kf, ktf, kc), s in combo.items() if kf == f] dsr_fam, sr0_fam = al.deflated_sharpe(sh_cd, family_combo_sh, cd) dsr_fil, sr0_fil = al.deflated_sharpe(sh_cd, fil_sh, cd) print(f"\n combo 50/50 daily: FULL {sh_cd:+.2f} IS {al._sh(cd_is):+.2f} HOLD {al._sh(cd_h):+.2f}") print(f" deflated-Sharpe: vs filone ({len(fil_sh)} trial) DSR={dsr_fil:.3f} (null-max {sr0_fil:+.2f})" f" | vs FAMIGLIA ({len(family_combo_sh)} trial) DSR={dsr_fam:.3f} (null-max {sr0_fam:+.2f})") # fee sweep sulla cella scelta print(" fee sweep (Sharpe FULL/HOLD per fee/side):") for a in ASSETS: df = al.get(a, ctf) tgt = builder(df) parts = [] for fe in FEE_SWEEP: evw = al.eval_weights(df, tgt, fee_side=fe) parts.append(f"{fe*100:.2f}%: {evw['full']['sharpe']:+.2f}/{evw['holdout'].get('sharpe', float('nan')):+.2f}") print(f" {a}: " + " ".join(parts)) # shift-test +/-2/4h (il segnale vede il calendario shiftato, il backtest no) print(" shift-test orario (Sharpe FULL netto per offset del calendario visto dal segnale):") flip = False for a in ASSETS: df0 = al.get(a, ctf) vals = {} for off in OFFSETS: tgt = builder(al._shift_calendar(df0, off)) vals[off] = al.eval_weights(df0, tgt)["full"]["sharpe"] base = vals[0] if any(np.sign(v) != np.sign(base) and abs(v) > 0.15 and abs(base) > 0.15 for o, v in vals.items() if o != 0): flip = True print(f" {a}: " + " ".join(f"{o:+d}h: {v:+.2f}" for o, v in vals.items())) print(f" -> sign-flip a |offset|<=4h: {flip} " f"({'ARTIFACT-RISK di etichettatura' if flip else 'nessun flip evidente'})") # causalita' + small-cap $600 cz = causality_tail_ok(builder, ctf) czr = al.causality_ok(builder, tf=ctf) print(f" causalita' (tail escl. ultima barra del prefisso): {cz['ok']} " f"(max diff {cz['max_tail_diff']}; diff ultima barra da gating-calendario " f"{cz['lastbar_gating_diff']}; altlib.causality_ok grezzo: {czr['ok']})") for a in ASSETS: df = al.get(a, ctf) sc = al.eval_weights_smallcap(df, builder(df), capital=600.0, min_order=5.0) print(f" small-cap $600 {a}: Sharpe modellato {sc['modeled']['sharpe']:+.2f} -> " f"realistico {sc['realistic']['sharpe']:+.2f} (haircut {sc['sharpe_haircut']:+.2f}), " f"trade eseguiti {sc['n_executed_trades']}") # verdetto suggerito (finale nel diario/report) if ch["min_is"] > 0.3 and ch["min_hold"] > 0 and (np.isfinite(dsr_fam) and dsr_fam >= 0.95): v = "CANDIDATO (passa IS+HOLD+DSR: servono marginal scorer e scettico)" elif ch["min_is"] > 0.3 and ch["min_hold"] > 0: v = "LEAD DEBOLE (IS+HOLD>0 ma non sopravvive al deflated-Sharpe)" elif ch["min_is"] > 0.3: v = "SCARTATO (IS ok ma hold-out negativo)" elif ch["min_isg"] <= 0: v = "SCARTATO (nessun edge nemmeno LORDO in-sample)" elif ch["min_isg"] - ch["min_is"] > 0.15 and ch["min_is"] <= 0.1: v = (f"SCARTATO (lordo IS {ch['min_isg']:+.2f} debole e sotto soglia; " f"le fee lo azzerano: netto {ch['min_is']:+.2f})") else: v = f"SCARTATO (edge in-sample netto {ch['min_is']:+.2f} sotto soglia = rumore)" print(f"\n VERDETTO SUGGERITO filone {f.upper()}: {v}") # ---- riepilogo famiglia ----------------------------------------------------- print("\n" + "=" * 100) print(" RIEPILOGO FAMIGLIA WK-INTRA") print("=" * 100) print(f" trial totali: {n_cells} celle per-asset / {len(combo)} configurazioni combo") best_key = max(combo, key=lambda k: al._sh(combo[k])) print(f" miglior combo FULL (senno' di poi, NON selezione): {best_key} Sh {al._sh(combo[best_key]):+.2f}") pos_is = int((R.is_sh > 0).sum()) pos_h = int((R.hold_sh > 0).sum()) posg_is = int((R.is_shg > 0).sum()) print(f" celle con IS netto>0: {pos_is}/{n_cells} (lordo {posg_is}/{n_cells}); HOLD netto>0: {pos_h}/{n_cells}") print(" nota b: la riapertura CME reale oscilla 22:00/23:00 UTC (DST) -> lo shift-test copre la banda.") if __name__ == "__main__": main()