"""r0702_eventclock.py — EVENT-CLOCK BARS (campionamento a tempo-informazione), 2026-07-02. IPOTESI: campionare il tempo per INFORMAZIONE (volume bars, vol bars = cum|logret|, range bars) normalizza i regimi e migliora trend/breakout A PARITA' di strategia e frequenza media rispetto alle barre wall-clock. Mai testato nel progetto (tutte le 104 famiglie girano su barre wall-clock). DISEGNO ONESTO: * Barre-evento costruite dal 5m certificato Deribit (al.get). Soglia CAUSALE: EWMA-90g dell'incremento per barra 5m, SHIFTATA di 1 (solo passato), x N_target barre 5m per la durata nominale (4h/12h/24h). Nessuna calibrazione full-sample. Parametri fissati a priori (span 90g, warm-up 14g, durate 4/12/24h). * Decisione a close della barra-evento k -> posizione tenuta DALLA prima barra 5m dopo la chiusura (shift +1 barra-evento). Mark-to-market sul 5m, compounding a griglia daily UTC (stessa convenzione di al.candidate_daily). Fee 0.0005/lato su |Δpos|. * Selezione cella SOLO IN-SAMPLE (pre-2025) sul Sharpe 50/50; hold-out riportato per QUELLA cella. deflated_sharpe su TUTTI i trial (event + wall). * CONTROLLO DECISIVO: stessa strategia, stessi parametri (in unita' di barre, convertiti per durata nominale) su barre WALL-CLOCK 4h/12h/1d (al.get, path resample leak-free). * Guardia causalita': ricostruzione barre+target su prefisso (80%/92%) -> i confini e i target devono coincidere con la run full troncata. * NIENTE ffill mixed-timeframe; niente DatetimeIndex.view('int64') (uso la colonna timestamp in ms). Run: uv run python scripts/research/r0702_eventclock.py """ from __future__ import annotations import sys import time import numpy as np import pandas as pd sys.path.insert(0, "/opt/docker/PythagorasGoal/scripts/research/alt") import altlib as al # noqa: E402 FEE = al.FEE_SIDE HOLDOUT = al.HOLDOUT ASSETS = ("BTC", "ETH") # ---- parametri FISSATI A PRIORI (nessun tuning) ----------------------------------------- DUR_HOURS = (4.0, 12.0, 24.0) # durata media nominale delle barre-evento BAR_TYPES = ("volume", "volbar", "range") WALL_TF = {4.0: "4h", 12.0: "12h", 24.0: "1d"} EWM_SPAN_5M = 90 * 288 # soglia adattiva: EWMA 90 giorni di barre 5m WARMUP_5M = 14 * 288 # min_periods 14 giorni prima della prima barra BARS_5M_PER_H = 12 # strategie (parametri in GIORNI-equivalenti, convertiti in barre per durata nominale) STRATS = [ ("TSMOM-30/90/180", "tsmom", dict(days=(30, 90, 180))), ("DONCH-10d", "donch", dict(days=10)), ("DONCH-30d", "donch", dict(days=30)), ("EWMA-5/30", "ewma", dict(days=(5, 30))), ("EWMA-15/75", "ewma", dict(days=(15, 75))), ] def bars_for_days(days: float, dur_h: float) -> int: return max(2, int(round(days * 24.0 / dur_h))) # ========================================================================================== # COSTRUZIONE BARRE-EVENTO (causale) # ========================================================================================== def _increments(df5: pd.DataFrame, kind: str) -> np.ndarray: c = df5["close"].values.astype(float) if kind == "volume": return df5["volume"].values.astype(float) if kind == "volbar": # cum |logret| r = np.zeros(len(c)); r[1:] = np.abs(np.log(c[1:] / c[:-1])) return r if kind == "range": # cum range relativo (high-low)/close h = df5["high"].values.astype(float); l = df5["low"].values.astype(float) return (h - l) / np.where(c > 0, c, np.nan) raise ValueError(kind) def _bar_close_indices(x: np.ndarray, thr: np.ndarray) -> np.ndarray: """Loop di formazione barre: chiude una barra quando il cum degli incrementi dal close della barra precedente raggiunge la soglia CAUSALE thr[i] (gia' shiftata).""" e = [] cum = 0.0 ap = e.append for i in range(len(x)): t = thr[i] if not (t > 0.0): # NaN o <=0 (warm-up): non accumulare cum = 0.0 continue cum += x[i] if cum >= t: ap(i) cum = 0.0 return np.asarray(e, dtype=np.int64) class EventBars: """Barre-evento per (asset, tipo, durata): OHLC + indici 5m di chiusura.""" def __init__(self, df5: pd.DataFrame, kind: str, dur_h: float): x = np.nan_to_num(_increments(df5, kind), nan=0.0) # soglia causale: EWMA(span 90g) dell'incremento per 5m, shift(1), x N barre target m = pd.Series(x).ewm(span=EWM_SPAN_5M, min_periods=WARMUP_5M, adjust=False).mean().shift(1).values n_target = dur_h * BARS_5M_PER_H thr = m * n_target self.e = _bar_close_indices(x, thr) # indici 5m dei close di barra if len(self.e) < 300: raise RuntimeError(f"troppo poche barre-evento ({len(self.e)}) per {kind}/{dur_h}h") c5 = df5["close"].values.astype(float) h5 = df5["high"].values.astype(float) l5 = df5["low"].values.astype(float) i0 = int(np.argmax(thr > 0)) # primo indice utilizzabile starts = np.concatenate([[i0], self.e[:-1] + 1]) sl = slice(0, self.e[-1] + 1) self.close = c5[self.e] self.high = np.maximum.reduceat(h5[sl], starts) self.low = np.minimum.reduceat(l5[sl], starts) # close-time in ms (fine barra 5m = open label + 5m); NIENTE .view su tz-aware ts5 = df5["timestamp"].values.astype(np.int64) self.ts_close_ms = ts5[self.e] + 300_000 self.n5 = len(df5) # statistiche durata d_h = np.diff(self.ts_close_ms) / 3.6e6 self.dur_median_h = float(np.median(d_h)) self.dur_p5_h = float(np.percentile(d_h, 5)) span_days = (self.ts_close_ms[-1] - self.ts_close_ms[0]) / 86.4e6 self.bars_per_day = len(self.e) / max(span_days, 1.0) ho_ms = int(HOLDOUT.value // 1_000_000) mask_h = self.ts_close_ms >= ho_ms span_h = (self.ts_close_ms[-1] - ho_ms) / 86.4e6 self.bars_per_day_holdout = float(mask_h.sum() / max(span_h, 1.0)) # ========================================================================================== # STRATEGIE (target causale su barre-evento O wall-clock: array close/high/low) # ========================================================================================== def strat_target(close: np.ndarray, high: np.ndarray, low: np.ndarray, fn: str, params: dict, dur_h: float) -> np.ndarray: n = len(close) if fn == "tsmom": hs = [bars_for_days(d, dur_h) for d in params["days"]] d = np.zeros(n) for k in hs: s = np.zeros(n); s[k:] = np.sign(close[k:] - close[:-k]) d += s t = (d > 0).astype(float) t[:max(hs)] = 0.0 # tutte le finestre disponibili return t if fn == "donch": N = bars_for_days(params["days"], dur_h) hi = pd.Series(high).rolling(N, min_periods=N).max().shift(1).values lo = pd.Series(low).rolling(N, min_periods=N).min().shift(1).values pos = np.where(close > hi, 1.0, np.nan) pos = np.where(close < lo, 0.0, pos) return pd.Series(pos).ffill().fillna(0.0).values if fn == "ewma": f_d, s_d = params["days"] fs, ss = bars_for_days(f_d, dur_h), bars_for_days(s_d, dur_h) f = pd.Series(close).ewm(span=fs, adjust=False).mean().values s = pd.Series(close).ewm(span=ss, adjust=False).mean().values t = (f > s).astype(float) t[:ss] = 0.0 return t raise ValueError(fn) # ========================================================================================== # VALUTAZIONE — barre-evento marked-to-market sul 5m, compounding daily # ========================================================================================== def pos5_from_event(n5: int, e: np.ndarray, tgt: np.ndarray) -> np.ndarray: """Espande i target di barra-evento a posizione per-barra-5m. Il target deciso al close della barra-evento k (indice 5m e[k]) e' tenuto DURANTE le barre 5m (e[k], e[k+1]] -> shift +1 barra-evento by construction.""" tgt = np.nan_to_num(np.asarray(tgt, float), nan=0.0) pos = np.zeros(n5) if len(e) >= 2: pos[e[0] + 1:e[-1] + 1] = np.repeat(tgt[:-1], np.diff(e)) if len(e) >= 1: pos[e[-1] + 1:] = tgt[-1] return pos def daily_from_pos5(df5: pd.DataFrame, pos5: np.ndarray, fee_side: float = FEE) -> pd.Series: c = df5["close"].values.astype(float) r = np.zeros(len(c)); r[1:] = c[1:] / c[:-1] - 1.0 turn = np.abs(np.diff(pos5, prepend=0.0)) net = pos5 * r - fee_side * turn net[0] = 0.0 idx = pd.DatetimeIndex(pd.to_datetime(df5["datetime"], utc=True)) return al._to_daily(pd.Series(net, index=idx)) def daily_wall(asset: str, tf: str, fn: str, params: dict, dur_h: float, fee_side: float = FEE) -> pd.Series: df = al.get(asset, tf) tgt = strat_target(df["close"].values.astype(float), df["high"].values.astype(float), df["low"].values.astype(float), fn, params, dur_h) ev = al.eval_weights(df, tgt, fee_side=fee_side) # shift +1 fatto dall'harness return al._to_daily(pd.Series(ev["net"], index=ev["idx"])) def combo5050(dA: pd.Series, dB: pd.Series) -> pd.Series: J = pd.concat({"A": dA, "B": dB}, axis=1, join="inner").fillna(0.0) return 0.5 * J["A"] + 0.5 * J["B"] def met(d: pd.Series) -> dict: """Sharpe/CAGR/maxDD full + hold + in-sample da una serie daily.""" di = d[d.index < HOLDOUT]; dh = d[d.index >= HOLDOUT] def _cagr(s): if len(s) < 10: return float("nan") tot = float(np.prod(1.0 + s.values)) return tot ** (365.25 / len(s)) - 1.0 if tot > 0 else -1.0 return dict(is_sh=round(al._sh(di), 3), full_sh=round(al._sh(d), 3), hold_sh=round(al._sh(dh), 3), full_dd=round(al._dd_ret(d), 4), hold_dd=round(al._dd_ret(dh), 4), full_cagr=round(_cagr(d), 4), hold_cagr=round(_cagr(dh), 4)) def yearly(d: pd.Series) -> dict: out = {} for y, g in d.groupby(d.index.year): eq = np.cumprod(1 + g.values); pk = np.maximum.accumulate(eq) out[int(y)] = (round(float(eq[-1] - 1), 4), round(float(np.max((pk - eq) / pk)), 4)) return out # ========================================================================================== # GUARDIA CAUSALITA' — ricostruzione su prefisso # ========================================================================================== def causality_prefix_check(asset: str, kind: str, dur_h: float, fn: str, params: dict) -> dict: """Ricostruisce barre+target sul prefisso 80%/92% del 5m: i confini di barra devono essere un prefisso esatto di quelli full (tranne l'ultima barra incompleta) e i target delle barre condivise identici. Qualunque dipendenza dal futuro diverge.""" df5 = al.get(asset, "5m") full = EventBars(df5, kind, dur_h) t_full = strat_target(full.close, full.high, full.low, fn, params, dur_h) worst = 0.0; ok = True; checked = 0 for frac in (0.80, 0.92): cut = int(len(df5) * frac) sub = df5.iloc[:cut].reset_index(drop=True) eb = EventBars(sub, kind, dur_h) m = len(eb.e) if not np.array_equal(eb.e, full.e[:m]): ok = False continue t_sub = strat_target(eb.close, eb.high, eb.low, fn, params, dur_h) d = float(np.max(np.abs(t_sub - t_full[:m]))) if m else 0.0 worst = max(worst, d) checked += 1 return dict(ok=bool(ok and worst <= 1e-9), max_diff=worst, checked=checked) # ========================================================================================== # SMALL-CAP a $600 sulle transizioni della cella scelta # ========================================================================================== def smallcap_event(df5: pd.DataFrame, pos5: np.ndarray, capital=600.0, min_order=5.0) -> dict: tgt = np.nan_to_num(pos5, nan=0.0) held = np.empty(len(tgt)); cur = 0.0; n_tr = 0 for i in range(len(tgt)): if abs(tgt[i] - cur) * capital >= min_order: cur = tgt[i]; n_tr += 1 held[i] = cur d_real = daily_from_pos5(df5, held) d_mod = daily_from_pos5(df5, tgt) return dict(realistic_sh=round(al._sh(d_real), 3), modeled_sh=round(al._sh(d_mod), 3), haircut=round(al._sh(d_mod) - al._sh(d_real), 3), n_executed=n_tr) # ========================================================================================== # MAIN # ========================================================================================== def main(): t0 = time.time() print("=" * 100) print("R0702 EVENT-CLOCK BARS — volume/volbar/range vs wall-clock, selezione in-sample") print("=" * 100) df5 = {a: al.get(a, "5m") for a in ASSETS} for a in ASSETS: print(f"{a} 5m: {len(df5[a])} barre, {df5[a]['datetime'].iloc[0]} -> " f"{df5[a]['datetime'].iloc[-1]}") # ---- 1. costruzione barre-evento (cache) -------------------------------------------- print("\n--- CALIBRAZIONE CLOCK (barre/giorno; target 4h=6, 12h=2, 24h=1) ---") bars = {} print(f"{'asset':5s} {'tipo':7s} {'dur':>5s} {'n_bars':>7s} {'bars/g':>7s} " f"{'bars/g HOLD':>11s} {'med(h)':>7s} {'p5(h)':>6s}") for a in ASSETS: for k in BAR_TYPES: for dh in DUR_HOURS: eb = EventBars(df5[a], k, dh) bars[(a, k, dh)] = eb print(f"{a:5s} {k:7s} {dh:4.0f}h {len(eb.e):7d} {eb.bars_per_day:7.2f} " f"{eb.bars_per_day_holdout:11.2f} {eb.dur_median_h:7.2f} {eb.dur_p5_h:6.2f}") # ---- 2. tutte le celle: event (3 tipi x 3 durate x 5 strategie) + wall (3 tf x 5) --- cells = [] # dict(kind, bar_type, dur_h, strat, daily {asset}, daily5050, met) for sname, fn, params in STRATS: for dh in DUR_HOURS: # wall-clock control dw = {a: daily_wall(a, WALL_TF[dh], fn, params, dh) for a in ASSETS} c5050 = combo5050(dw["BTC"], dw["ETH"]) cells.append(dict(kind="wall", bar_type=WALL_TF[dh], dur_h=dh, strat=sname, fn=fn, params=params, daily=dw, d5050=c5050, met=met(c5050))) # event-clock for k in BAR_TYPES: de = {} for a in ASSETS: eb = bars[(a, k, dh)] tgt = strat_target(eb.close, eb.high, eb.low, fn, params, dh) de[a] = daily_from_pos5(df5[a], pos5_from_event(eb.n5, eb.e, tgt)) c5050 = combo5050(de["BTC"], de["ETH"]) cells.append(dict(kind="event", bar_type=k, dur_h=dh, strat=sname, fn=fn, params=params, daily=de, d5050=c5050, met=met(c5050))) print(f"\n--- TUTTE LE CELLE (Sharpe 50/50: IN-SAMPLE pre-2025 | FULL | HOLD 2025-26) ---") print(f"{'clock':6s} {'barre':7s} {'dur':>4s} {'strategia':16s} {'IS':>6s} {'FULL':>6s} {'HOLD':>6s}") for c in sorted(cells, key=lambda x: (x["strat"], x["dur_h"], x["kind"], x["bar_type"])): m = c["met"] print(f"{c['kind']:6s} {c['bar_type']:7s} {c['dur_h']:3.0f}h {c['strat']:16s} " f"{m['is_sh']:6.2f} {m['full_sh']:6.2f} {m['hold_sh']:6.2f}") # ---- 3. CONTROLLO DECISIVO: paired event vs wall a parita' di strategia+durata ------ print("\n--- PAIRED: event vs wall (Δ Sharpe = event − wall, per cella accoppiata) ---") print(f"{'strategia':16s} {'dur':>4s} {'tipo':7s} {'ΔIS':>7s} {'ΔHOLD':>7s}") n_pairs = n_is_win = n_hold_win = n_both_win = 0 for sname, fn, params in STRATS: for dh in DUR_HOURS: w = next(c for c in cells if c["kind"] == "wall" and c["strat"] == sname and c["dur_h"] == dh) for k in BAR_TYPES: e = next(c for c in cells if c["kind"] == "event" and c["strat"] == sname and c["dur_h"] == dh and c["bar_type"] == k) d_is = e["met"]["is_sh"] - w["met"]["is_sh"] d_h = e["met"]["hold_sh"] - w["met"]["hold_sh"] n_pairs += 1 n_is_win += d_is > 0 n_hold_win += d_h > 0 n_both_win += (d_is > 0 and d_h > 0) print(f"{sname:16s} {dh:3.0f}h {k:7s} {d_is:+7.2f} {d_h:+7.2f}") print(f"\nevent batte wall: IS {n_is_win}/{n_pairs}, HOLD {n_hold_win}/{n_pairs}, " f"ENTRAMBI {n_both_win}/{n_pairs}") # ---- 4. selezione IN-SAMPLE della cella event migliore ------------------------------ ev_cells = [c for c in cells if c["kind"] == "event"] wall_cells = [c for c in cells if c["kind"] == "wall"] chosen = max(ev_cells, key=lambda c: c["met"]["is_sh"]) paired = next(c for c in wall_cells if c["strat"] == chosen["strat"] and c["dur_h"] == chosen["dur_h"]) best_wall_is = max(wall_cells, key=lambda c: c["met"]["is_sh"]) print("\n" + "=" * 100) print(f"CELLA SCELTA (max Sharpe IN-SAMPLE 50/50 tra le {len(ev_cells)} event): " f"{chosen['bar_type']} {chosen['dur_h']:.0f}h {chosen['strat']}") print("=" * 100) for label, d in (("BTC", chosen["daily"]["BTC"]), ("ETH", chosen["daily"]["ETH"]), ("50/50", chosen["d5050"])): m = met(d) print(f" {label:6s} FULL Sh {m['full_sh']:+.2f} DD {m['full_dd']*100:5.1f}% " f"CAGR {m['full_cagr']*100:+6.1f}% | HOLD Sh {m['hold_sh']:+.2f} " f"DD {m['hold_dd']*100:5.1f}% CAGR {m['hold_cagr']*100:+6.1f}% | IS {m['is_sh']:+.2f}") print(f"\n paired wall ({paired['bar_type']}, stessa strategia):") for label, d in (("BTC", paired["daily"]["BTC"]), ("ETH", paired["daily"]["ETH"]), ("50/50", paired["d5050"])): m = met(d) print(f" {label:6s} FULL Sh {m['full_sh']:+.2f} DD {m['full_dd']*100:5.1f}% " f"CAGR {m['full_cagr']*100:+6.1f}% | HOLD Sh {m['hold_sh']:+.2f} " f"DD {m['hold_dd']*100:5.1f}% CAGR {m['hold_cagr']*100:+6.1f}% | IS {m['is_sh']:+.2f}") bw = best_wall_is["met"] print(f"\n best WALL in-sample: {best_wall_is['bar_type']} {best_wall_is['strat']} " f"IS {bw['is_sh']:+.2f} FULL {bw['full_sh']:+.2f} HOLD {bw['hold_sh']:+.2f}") print("\n per-anno 50/50 cella scelta (ret, maxDD):") for y, (r, dd) in yearly(chosen["d5050"]).items(): print(f" {y}: {r*100:+6.1f}% dd {dd*100:5.1f}%") # decisive control per-asset print("\n CONTROLLO DECISIVO per-asset (event − wall):") dec_ok = True for a in ASSETS: me, mw = met(chosen["daily"][a]), met(paired["daily"][a]) d_is, d_h = me["is_sh"] - mw["is_sh"], me["hold_sh"] - mw["hold_sh"] dec_ok = dec_ok and (d_is > 0 and d_h > 0) print(f" {a}: ΔIS {d_is:+.2f} ΔHOLD {d_h:+.2f}") print(f" event batte wall IS E HOLD su entrambi gli asset: {dec_ok}") # ---- 5. fee sweep sulla cella scelta ------------------------------------------------- print("\n FEE SWEEP (Sharpe FULL 50/50):") fee_sh = {} for f in al.FEE_SWEEP: dd_ = {} for a in ASSETS: eb = bars[(a, chosen["bar_type"], chosen["dur_h"])] tgt = strat_target(eb.close, eb.high, eb.low, chosen["fn"], chosen["params"], chosen["dur_h"]) dd_[a] = daily_from_pos5(df5[a], pos5_from_event(eb.n5, eb.e, tgt), fee_side=f) fee_sh[f] = round(al._sh(combo5050(dd_["BTC"], dd_["ETH"])), 3) print(f" {2*f*100:.2f}%RT: {fee_sh[f]:+.2f}") fee_ok = fee_sh[0.0015] > 0 # ---- 6. deflated Sharpe su TUTTI i trial --------------------------------------------- all_sr = [c["met"]["full_sh"] for c in cells] dsr, sr0 = al.deflated_sharpe(al._sh(chosen["d5050"]), all_sr, chosen["d5050"].values) print(f"\n DEFLATED SHARPE: DSR={dsr:.3f} (soglia 0.95) | expected null max Sh {sr0:.2f} " f"| trial totali {len(all_sr)} (event {len(ev_cells)} + wall {len(wall_cells)})") # ---- 7. marginal vs TP01 ------------------------------------------------------------- print("\n MARGINAL vs TP01 (cella scelta in-sample):") marg = al.marginal_vs_tp01(chosen["d5050"]) for kk in ("marginal_verdict", "corr_full", "corr_hold", "cand_insample_sharpe", "has_insample_edge", "is_hedge", "robust_oos", "multicut_uplift", "multicut_persistent", "clean_year_uplift", "jackknife_min_uplift", "beta_to_tp01", "resid_sharpe_full", "hedge_yearly_corr", "uplift_tp01_up", "uplift_tp01_down"): print(f" {kk}: {marg.get(kk)}") for w, b in marg.get("blends", {}).items(): print(f" blend {w}: full {b['full']} (uplift {b['uplift_full']:+.3f}) " f"hold {b['hold']} (uplift {b['uplift_hold']:+.3f})") earns = (marg.get("marginal_verdict") == "ADDS" and marg.get("robust_oos", False) and marg.get("has_insample_edge", False) and not marg.get("is_hedge", False)) dsr_pass = np.isfinite(dsr) and dsr >= 0.95 print(f" earns_slot(marginale)={earns} dsr_pass={dsr_pass} " f"earns_slot_honest={earns and dsr_pass and fee_ok}") # ---- 8. causalita' + executability ---------------------------------------------------- print("\n GUARDIA CAUSALITA' (prefisso 80%/92%, entrambi gli asset):") for a in ASSETS: cz = causality_prefix_check(a, chosen["bar_type"], chosen["dur_h"], chosen["fn"], chosen["params"]) print(f" {a}: ok={cz['ok']} max_diff={cz['max_diff']:.2e} checked={cz['checked']}") print("\n EXECUTABILITY:") for a in ASSETS: eb = bars[(a, chosen["bar_type"], chosen["dur_h"])] tgt = strat_target(eb.close, eb.high, eb.low, chosen["fn"], chosen["params"], chosen["dur_h"]) sc = smallcap_event(df5[a], pos5_from_event(eb.n5, eb.e, tgt)) print(f" {a}: {eb.bars_per_day:.2f} barre/g (hold-out {eb.bars_per_day_holdout:.2f}), " f"durata mediana {eb.dur_median_h:.1f}h p5 {eb.dur_p5_h:.1f}h | " f"smallcap $600: modeled {sc['modeled_sh']:+.2f} realistic {sc['realistic_sh']:+.2f} " f"haircut {sc['haircut']:+.3f} ({sc['n_executed']} trade)") print(f"\n[done in {time.time()-t0:.0f}s]") if __name__ == "__main__": main()