#!/usr/bin/env python """r0702_crt_mtf.py — CRT (Candle Range Theory) MULTI-TIMEFRAME — filone 2026-07-02. TESI DA TESTARE (scuola CRT): il pattern a 3 candele C1 = candela-range forte; C2 = sweep di UN estremo di C1 con close back-inside (presa di liquidita'); C3 = ingresso CONTRO il breakout renderebbe molto di piu' eseguito MULTI-TF: struttura su TF alto (4h/1h), ingresso sul TF basso (15m/5m) al RITEST della zona violata -> stop dietro lo swing del TF basso (piu' stretto del "dietro l'estremo di C2" single-TF) -> R:R da ~1.3 a ~3+. DISEGNO SPERIMENTALE: confronto CONTROLLATO/PAIRED sugli STESSI pattern C1-C2, tre esecuzioni: (i) BASE single-TF: entry a open di C3, stop dietro l'estremo di C2, target estremo opposto C1 (ii) MTF ritest della zona + trigger di conferma sul TF basso, stop dietro lo swing basso (iii) NOTRIG ritest puro (entry al primo tocco della zona), senza conferma bassa Tutte e tre simulate sulla STESSA griglia di barre del TF basso (fill intrabar identici, conservativi: SL prima di TP nella stessa barra bassa). Fee 0.10% RT + sweep 0/0.10/0.20. DEFINIZIONI FISSATE A PRIORI (dichiarate prima di guardare i risultati, nessuna sensibilita' qui; la sensibilita' della detection e' del filone base single-TF): - C1 forte: range >= 1.2 * ATR14 del TF alto (UNA definizione; body/range NON usato). - C2: rompe UN SOLO estremo di C1 (doppio sweep = skip) e chiude DENTRO il range di C1. - Finestra: 1 barra del TF alto dopo la chiusura di C2 (la "C3"). - Max hold: 20 barre high-TF dall'apertura della finestra, poi exit a market al close (identico per tutte le varianti -> confronto pulito). - Invalidation (solo MTF/NOTRIG): se PRIMA del trigger il prezzo supera l'estremo di C2 (>=, conservativo), setup invalidato -> no trade (la BASE nello stesso caso viene semplicemente stoppata: e' la differenza strutturale fra le esecuzioni). - R:R >= 1.3 all'entry per MTF/NOTRIG (parte della tesi CRT-MTF). La BASE non e' filtrata (e' l'esecuzione classica single-TF). - Sizing: 1.0x nozionale per trade; book SEQUENZIALE per asset (1 trade aperto alla volta) per la serie daily (Sharpe/DD); expectancy per-trade su TUTTI i pattern (indip.). GRIGLIA (unica, chiusa a priori): d in {0.10, 0.25} x trigger in {closeback, sweeprec} per MTF; d in {0.10, 0.25} per NOTRIG. Selezione cella SOLO in-sample (<2025-01-01). Trials per DSR = (1 base + 4 MTF + 2 NOTRIG) x 2 coppie TF = 14. Trigger meccanici sul TF basso (per short, simmetrico per long); L = estremo C1 violato, zona = [L - d*ATR14_alto, L]: - closeback: dopo che una barra bassa ha TOCCATO la zona (high >= L - d*ATR), la prima barra bassa che CHIUDE sotto L -> entry al suo close. - sweeprec: barra bassa j che tocca la zona E sweep del massimo della barra bassa precedente (high[j] > high[j-1]) E chiude sotto high[j-1] E sotto L -> entry al close di j. Stop MTF/NOTRIG = estremo dello swing basso (max high dall'apertura della finestra alla barra del trigger inclusa). Target (tutte): estremo opposto di C1. Esecuzione: uv run python scripts/research/r0702_crt_mtf.py NON tocca src/, config/, scripts/live/. Nessun file scritto. """ from __future__ import annotations import sys import time from collections import Counter sys.path.insert(0, "/opt/docker/PythagorasGoal/scripts/research/alt") import numpy as np import pandas as pd import altlib as al # ---------------------------------------------------------------- config (a priori) PAIRS = (("4h", "15m"), ("1h", "5m")) ASSETS = ("BTC", "ETH") D_GRID = (0.10, 0.25) TRIGGERS = ("closeback", "sweeprec") MAXHOLD_HTF = 20 # barre high-TF di holding max, dalla apertura della finestra C3 RR_MIN = 1.3 # filtro R:R all'entry (solo MTF/NOTRIG) ATR_MULT = 1.2 # C1 forte: range >= 1.2*ATR14 (definizione unica) FEE_RT = 0.001 # 0.10% round-trip FEE_SWEEP_RT = (0.0, 0.001, 0.002) TF_MS = {"5m": 300_000, "15m": 900_000, "1h": 3_600_000, "4h": 14_400_000} HOLDOUT = al.HOLDOUT HOLDOUT_MS = int(HOLDOUT.value // 10**6) CAPITAL = 600.0 LEV_CAP = 2.0 MIN_ORDER = 5.0 # ---------------------------------------------------------------- detection (vettoriale) def detect_patterns(dfh: pd.DataFrame, tf_hi: str) -> list[dict]: """CRT C1-C2 sul TF alto. Pattern noto alla CHIUSURA di C2 (causale: usa solo barre <= C2).""" ts = dfh["timestamp"].astype("int64").values o = dfh["open"].values.astype(float) h = dfh["high"].values.astype(float) l = dfh["low"].values.astype(float) c = dfh["close"].values.astype(float) a = al.atr(dfh, 14) rng = h - l strong = rng >= ATR_MULT * a h1 = np.roll(h, 1); l1 = np.roll(l, 1); s1 = np.roll(strong, 1) up = s1 & (h > h1) & ~(l < l1) & (c <= h1) & (c >= l1) # sweep del massimo di C1 -> SHORT dn = s1 & (l < l1) & ~(h > h1) & (c >= l1) & (c <= h1) # sweep del minimo di C1 -> LONG idx = np.where(up | dn)[0] tf_ms = TF_MS[tf_hi] pats = [] for i in idx: if i < 20: # warm-up ATR continue if up[i]: d, level, target, c2ext = -1, h1[i], l1[i], h[i] else: d, level, target, c2ext = +1, l1[i], h1[i], l[i] pats.append(dict(i=int(i), dir=d, level=float(level), target=float(target), c2ext=float(c2ext), atr=float(a[i]), win_open=int(ts[i] + tf_ms), win_close=int(ts[i] + 2 * tf_ms), hold_end=int(ts[i] + (1 + MAXHOLD_HTF) * tf_ms))) return pats # ---------------------------------------------------------------- low-TF arrays class Low: __slots__ = ("ts", "o", "h", "l", "c", "tsclose", "n") def __init__(self, df: pd.DataFrame, tf_lo: str): self.ts = df["timestamp"].astype("int64").values self.o = df["open"].values.astype(float) self.h = df["high"].values.astype(float) self.l = df["low"].values.astype(float) self.c = df["close"].values.astype(float) self.tsclose = self.ts + TF_MS[tf_lo] self.n = len(self.ts) def scan_exit(L: Low, j0: int, j1: int, dr: int, entry_ts: int, stop: float, target: float): """Barre j0..j1-1; conservativo: SL prima di TP nella stessa barra. Ritorna (px, ts, kind).""" Lh, Ll, Lc, Ltsc = L.h, L.l, L.c, L.tsclose for j in range(j0, j1): if dr < 0: if Lh[j] >= stop: return stop, int(Ltsc[j]), "SL" if Ll[j] <= target: return target, int(Ltsc[j]), "TP" else: if Ll[j] <= stop: return stop, int(Ltsc[j]), "SL" if Lh[j] >= target: return target, int(Ltsc[j]), "TP" if j1 - 1 < j0: return None, entry_ts, "NOBARS" return float(Lc[j1 - 1]), int(Ltsc[j1 - 1]), "TIME" def _mk_trade(p, entry, entry_ts, stop, exitp, exit_ts, kind, jt=None, j1=None): dr = p["dir"] risk = abs(stop - entry) / entry gross = dr * (exitp / entry - 1.0) rr = (abs(entry - p["target"]) / abs(stop - entry)) if stop != entry else np.inf return dict(ok=True, dir=dr, entry=entry, stop=stop, target=p["target"], risk=risk, rr=rr, gross=gross, entry_ts=int(entry_ts), exit_ts=int(exit_ts), kind=kind, jt=jt, j1=j1) def trade_base(p: dict, L: Low): """(i) BASE: entry a open C3 (= prima barra bassa della finestra), stop dietro estremo C2.""" j0 = int(np.searchsorted(L.ts, p["win_open"])) if j0 >= L.n or L.ts[j0] >= p["win_close"]: return dict(ok=False, reason="nodata") entry = float(L.o[j0]) dr, stop, target = p["dir"], p["c2ext"], p["target"] if (dr < 0 and not (target < entry < stop)) or (dr > 0 and not (stop < entry < target)): return dict(ok=False, reason="degenerate") j1 = int(np.searchsorted(L.ts, p["hold_end"])) exitp, exit_ts, kind = scan_exit(L, j0, j1, dr, int(L.ts[j0]), stop, target) if exitp is None: return dict(ok=False, reason="nodata") return _mk_trade(p, entry, L.ts[j0], stop, exitp, exit_ts, kind, jt=j0, j1=j1) def trade_mtf(p: dict, L: Low, d_mult: float, trigger: str | None): """(ii) MTF con trigger / (iii) NOTRIG (trigger=None): ritest della zona nella finestra C3.""" j0 = int(np.searchsorted(L.ts, p["win_open"])) if j0 >= L.n or L.ts[j0] >= p["win_close"]: return dict(ok=False, reason="nodata") jw = int(np.searchsorted(L.ts, p["win_close"])) dr, level, c2ext, target = p["dir"], p["level"], p["c2ext"], p["target"] zone = d_mult * p["atr"] Lh, Ll, Lc = L.h, L.l, L.c touched = False jt = -1 if dr < 0: swing = -np.inf for j in range(j0, jw): if Lh[j] > swing: swing = Lh[j] if Lh[j] >= c2ext: # struttura violata prima del trigger return dict(ok=False, reason="invalidated") if Lh[j] >= level - zone: touched = True if touched: if trigger is None: jt = j; break if trigger == "closeback" and Lc[j] < level: jt = j; break if (trigger == "sweeprec" and j >= 1 and Lh[j] >= level - zone and Lh[j] > Lh[j - 1] and Lc[j] < Lh[j - 1] and Lc[j] < level): jt = j; break if jt < 0: return dict(ok=False, reason=("notrigger" if touched else "noretest")) stop = float(max(swing, Lh[jt])) entry = float(Lc[jt]) if not (target < entry < stop): return dict(ok=False, reason="degenerate") else: swing = np.inf for j in range(j0, jw): if Ll[j] < swing: swing = Ll[j] if Ll[j] <= c2ext: return dict(ok=False, reason="invalidated") if Ll[j] <= level + zone: touched = True if touched: if trigger is None: jt = j; break if trigger == "closeback" and Lc[j] > level: jt = j; break if (trigger == "sweeprec" and j >= 1 and Ll[j] <= level + zone and Ll[j] < Ll[j - 1] and Lc[j] > Ll[j - 1] and Lc[j] > level): jt = j; break if jt < 0: return dict(ok=False, reason=("notrigger" if touched else "noretest")) stop = float(min(swing, Ll[jt])) entry = float(Lc[jt]) if not (stop < entry < target): return dict(ok=False, reason="degenerate") rr = abs(entry - target) / abs(stop - entry) if rr < RR_MIN: return dict(ok=False, reason="rrfail") j1 = int(np.searchsorted(L.ts, p["hold_end"])) exitp, exit_ts, kind = scan_exit(L, jt + 1, j1, dr, int(L.tsclose[jt]), stop, target) if exitp is None: # trigger sull'ultima barra: flat, -fee exitp, exit_ts, kind = entry, int(L.tsclose[jt]), "NOBARS" return _mk_trade(p, entry, L.tsclose[jt], stop, exitp, exit_ts, kind, jt=jt, j1=j1) # ---------------------------------------------------------------- stats & book def trade_stats(trades: list[dict], fee_rt: float = FEE_RT) -> dict: tr = [t for t in trades if t and t.get("ok")] if not tr: return dict(n=0, exp_bps=np.nan, wr=np.nan, avgR=np.nan, med_risk=np.nan, avg_rr=np.nan) nets = np.array([t["gross"] - fee_rt for t in tr]) Rs = np.array([(t["gross"] - fee_rt) / t["risk"] for t in tr if t["risk"] > 0]) return dict(n=len(tr), exp_bps=float(nets.mean() * 1e4), wr=float((nets > 0).mean() * 100), avgR=float(Rs.mean()) if len(Rs) else np.nan, med_risk=float(np.median([t["risk"] for t in tr]) * 100), avg_rr=float(np.mean([min(t["rr"], 50.0) for t in tr]))) def seq_filter(trades: list[dict]) -> list[dict]: out, last = [], -1 for t in sorted((t for t in trades if t and t.get("ok")), key=lambda x: x["entry_ts"]): if t["entry_ts"] >= last: out.append(t) last = t["exit_ts"] return out def daily_series(seq_trades: list[dict], span: tuple[int, int], fee_rt: float = FEE_RT) -> pd.Series: idx = pd.date_range(pd.Timestamp(span[0], unit="ms", tz="UTC").normalize(), pd.Timestamp(span[1], unit="ms", tz="UTC").normalize(), freq="D") s = pd.Series(0.0, index=idx) for t in seq_trades: d = pd.Timestamp(t["exit_ts"], unit="ms", tz="UTC").normalize() if d in s.index: s[d] += t["gross"] - fee_rt return s def sh_dd(s: pd.Series) -> tuple[float, float]: sharpe = al._sh(s) eq = np.cumprod(1.0 + s.values) pk = np.maximum.accumulate(eq) dd = float(np.max((pk - eq) / pk)) if len(eq) else 0.0 return sharpe, dd def portfolio_daily(res_pair: dict, key, spans: dict, fee_rt: float = FEE_RT) -> pd.Series: per = [] for a in ASSETS: seq = seq_filter(res_pair[a][key]) per.append(daily_series(seq, spans[a], fee_rt)) J = pd.concat(per, axis=1).fillna(0.0) return 0.5 * J.iloc[:, 0] + 0.5 * J.iloc[:, 1] def split_hold(trades: list[dict]) -> tuple[list, list]: ins = [t for t in trades if t and t.get("ok") and t["entry_ts"] < HOLDOUT_MS] hold = [t for t in trades if t and t.get("ok") and t["entry_ts"] >= HOLDOUT_MS] return ins, hold # ---------------------------------------------------------------- delayed execution (cron orario) def delayed_eval(trades: list[dict], L: Low, fee_rt: float = FEE_RT) -> dict: """Il book live gira ogni ora: il segnale (close barra bassa) viene eseguito alla PRIMA chiusura di barra bassa sulla griglia oraria successiva. Se nel frattempo SL/TP e' gia' stato attraversato -> nessun ingresso (skip). Ritorna expectancy originale vs ritardata.""" orig, dela, delays = [], [], [] n_skip_sl = n_skip_tp = n_missed_window = 0 for t in trades: if not (t and t.get("ok")): continue ts_e = t["entry_ts"] boundary = ((ts_e + 3_599_999) // 3_600_000) * 3_600_000 delays.append((boundary - ts_e) / 60_000.0) if boundary == ts_e: orig.append(t["gross"] - fee_rt) dela.append(t["gross"] - fee_rt) continue jb = int(np.searchsorted(L.tsclose, boundary)) j1 = t["j1"] if jb >= L.n or jb >= j1: n_missed_window += 1 orig.append(t["gross"] - fee_rt) continue dr, stop, target = t["dir"], t["stop"], t["target"] crossed = None for j in range(t["jt"] + 1, jb + 1): if dr < 0: if L.h[j] >= stop: crossed = "SL"; break if L.l[j] <= target: crossed = "TP"; break else: if L.l[j] <= stop: crossed = "SL"; break if L.h[j] >= target: crossed = "TP"; break orig.append(t["gross"] - fee_rt) if crossed == "SL": n_skip_sl += 1 continue if crossed == "TP": n_skip_tp += 1 continue entry2 = float(L.c[jb]) if (dr < 0 and not (target < entry2 < stop)) or (dr > 0 and not (stop < entry2 < target)): n_skip_sl += 1 continue exitp, _, _ = scan_exit(L, jb + 1, j1, dr, int(L.tsclose[jb]), stop, target) if exitp is None: exitp = entry2 dela.append(dr * (exitp / entry2 - 1.0) - fee_rt) n_sig = len(orig) return dict(n_signals=n_sig, mean_delay_min=float(np.mean(delays)) if delays else np.nan, n_entered=len(dela), n_skip_sl=n_skip_sl, n_skip_tp=n_skip_tp, n_missed_window=n_missed_window, exp_orig_bps=float(np.mean(orig) * 1e4) if orig else np.nan, exp_delayed_entered_bps=float(np.mean(dela) * 1e4) if dela else np.nan, exp_delayed_per_signal_bps=float(np.sum(dela) / n_sig * 1e4) if n_sig else np.nan) # ---------------------------------------------------------------- main def key_label(key) -> str: if key == ("base",): return "BASE single-TF " if key[0] == "mtf": return f"MTF d={key[1]:.2f} {key[2]:<9s}" return f"NOTRIG d={key[1]:.2f} " def main(): t0 = time.time() print("=" * 100) print("r0702 CRT MULTI-TIMEFRAME — struttura su TF alto, ingresso su TF basso (paired vs base)") print(f"C1 forte: range>={ATR_MULT}*ATR14 | maxhold {MAXHOLD_HTF} barre HTF | RR>={RR_MIN} (MTF) " f"| fee {FEE_RT*1e4:.0f}bps RT | hold-out >= {HOLDOUT.date()}") print("=" * 100) all_trial_sharpes = [] # per DSR: full Sharpe di OGNI (pair, variant-cell) chosen_summaries = [] # per selezione finale cross-pair for tf_hi, tf_lo in PAIRS: print(f"\n{'#'*100}\n### COPPIA {tf_hi} -> {tf_lo}\n{'#'*100}") res: dict[str, dict] = {} spans: dict[str, tuple[int, int]] = {} reasons: dict[str, dict] = {} n_pats: dict[str, int] = {} variant_keys = [("base",)] + [("mtf", d, tr) for d in D_GRID for tr in TRIGGERS] \ + [("notrig", d) for d in D_GRID] for a in ASSETS: dfh = al.get(a, tf_hi) L = Low(al.get(a, tf_lo), tf_lo) spans[a] = (int(L.ts[0]), int(L.tsclose[-1])) pats = detect_patterns(dfh, tf_hi) n_pats[a] = len(pats) res[a] = {} reasons[a] = {} for key in variant_keys: outs = [] for p in pats: if key[0] == "base": outs.append(trade_base(p, L)) elif key[0] == "mtf": outs.append(trade_mtf(p, L, key[1], key[2])) else: outs.append(trade_mtf(p, L, key[1], None)) res[a][key] = outs reasons[a][key] = Counter(t.get("reason") for t in outs if not t.get("ok")) print(f" {a}: {len(pats)} pattern C1-C2 su {tf_hi} " f"(short={sum(1 for p in pats if p['dir'] < 0)}, long={sum(1 for p in pats if p['dir'] > 0)})") # ------- tabella varianti: per-trade (tutti i pattern, indip.) + book sequenziale 50/50 print(f"\n --- VARIANTI (pooled BTC+ETH; per-trade su tutti i pattern; Sharpe/DD su book " f"sequenziale 50/50, daily) ---") hdr = (f" {'variante':<24s} | {'n_FULL':>6s} {'exp(bps)':>8s} {'WR%':>5s} {'avgR':>6s} " f"{'RRm':>5s} {'Sh_F':>6s} {'DD_F%':>6s} | {'n_H':>5s} {'expH':>8s} {'WRH':>5s} " f"{'Sh_H':>6s} | {'riskMed%':>8s}") print(hdr) table = {} for key in variant_keys: pooled = res["BTC"][key] + res["ETH"][key] ins, hold = split_hold(pooled) st_f = trade_stats(ins + hold) st_h = trade_stats(hold) port = portfolio_daily(res, key, spans) sh_f, dd_f = sh_dd(port) ph = port[port.index >= HOLDOUT] sh_h, _ = sh_dd(ph) if len(ph) > 30 else (np.nan, np.nan) pi = port[port.index < HOLDOUT] sh_is, _ = sh_dd(pi) if len(pi) > 30 else (np.nan, np.nan) st_is = trade_stats(ins) table[key] = dict(st_f=st_f, st_h=st_h, st_is=st_is, sh_f=sh_f, dd_f=dd_f, sh_h=sh_h, sh_is=sh_is, port=port) all_trial_sharpes.append(sh_f) print(f" {key_label(key)} | {st_f['n']:>6d} {st_f['exp_bps']:>8.1f} {st_f['wr']:>5.1f} " f"{st_f['avgR']:>6.2f} {st_f['avg_rr']:>5.1f} {sh_f:>6.2f} {dd_f*100:>6.1f} | " f"{st_h['n']:>5d} {st_h['exp_bps']:>8.1f} {st_h['wr']:>5.1f} {sh_h:>6.2f} | " f"{st_f['med_risk']:>8.3f}") # ------- quota pattern senza ritest / invalidati / rr-fail (per cella MTF) print("\n --- FUNNEL pattern -> trade (pooled, % dei pattern) ---") for key in variant_keys[1:]: cnt = reasons["BTC"][key] + reasons["ETH"][key] tot = n_pats["BTC"] + n_pats["ETH"] n_tr = table[key]["st_f"]["n"] print(f" {key_label(key)} | trade {n_tr:>5d} ({n_tr/tot*100:4.1f}%) | " f"no-ritest {cnt.get('noretest', 0)/tot*100:4.1f}% | " f"no-trigger {cnt.get('notrigger', 0)/tot*100:4.1f}% | " f"invalidato {cnt.get('invalidated', 0)/tot*100:4.1f}% | " f"RR<{RR_MIN} {cnt.get('rrfail', 0)/tot*100:4.1f}% | " f"altro {sum(v for k, v in cnt.items() if k in ('nodata', 'degenerate'))/tot*100:4.1f}%") # ------- selezione cella SOLO in-sample (<2025) mtf_keys = [k for k in variant_keys if k[0] == "mtf"] ntg_keys = [k for k in variant_keys if k[0] == "notrig"] def is_score(k): v = table[k]["sh_is"] return v if np.isfinite(v) else -9 best_mtf = max(mtf_keys, key=is_score) best_ntg = max(ntg_keys, key=is_score) print(f"\n --- SELEZIONE IN-SAMPLE (<2025, Sharpe book 50/50) ---") for k in mtf_keys + ntg_keys: mark = " <== scelta" if k in (best_mtf, best_ntg) else "" print(f" {key_label(k)} | Sh_IS={table[k]['sh_is']:>6.2f} exp_IS={table[k]['st_is']['exp_bps']:>7.1f}bps " f"(n_IS={table[k]['st_is']['n']}){mark}") print(f" BASE | Sh_IS={table[('base',)]['sh_is']:>6.2f} " f"exp_IS={table[('base',)]['st_is']['exp_bps']:>7.1f}bps (n_IS={table[('base',)]['st_is']['n']})") # ------- confronto PAIRED sugli stessi pattern (subset dove TUTTE e 3 hanno tradato) print(f"\n --- PAIRED sugli stessi pattern (BASE vs MTF{best_mtf[1:]} vs NOTRIG d={best_ntg[1]}) ---") for label, mask_hold in (("FULL", None), ("HOLD", True)): diffs_mb, diffs_nb = [], [] rows = {k: [] for k in (("base",), best_mtf, best_ntg)} for a in ASSETS: for tb, tm, tn in zip(res[a][("base",)], res[a][best_mtf], res[a][best_ntg]): if not (tb.get("ok") and tm.get("ok") and tn.get("ok")): continue if mask_hold and tm["entry_ts"] < HOLDOUT_MS: continue if mask_hold is None and False: continue rows[("base",)].append(tb) rows[best_mtf].append(tm) rows[best_ntg].append(tn) diffs_mb.append((tm["gross"] - FEE_RT) - (tb["gross"] - FEE_RT)) diffs_nb.append((tn["gross"] - FEE_RT) - (tb["gross"] - FEE_RT)) n = len(diffs_mb) if n < 5: print(f" [{label}] n={n} — potenza statistica insufficiente per il paired") continue d_mb = np.array(diffs_mb); d_nb = np.array(diffs_nb) t_mb = d_mb.mean() / (d_mb.std(ddof=1) / np.sqrt(n)) if d_mb.std() > 0 else np.nan t_nb = d_nb.mean() / (d_nb.std(ddof=1) / np.sqrt(n)) if d_nb.std() > 0 else np.nan print(f" [{label}] n_paired={n}") for k in (("base",), best_mtf, best_ntg): st = trade_stats(rows[k]) print(f" {key_label(k)} | exp={st['exp_bps']:>7.1f}bps WR={st['wr']:>5.1f}% " f"avgR={st['avgR']:>6.2f} RRmedio={st['avg_rr']:>4.1f} riskMed={st['med_risk']:.3f}%") print(f" Δ(MTF-BASE) = {d_mb.mean()*1e4:>+7.1f}bps/trade t={t_mb:+.2f}") print(f" Δ(NOTRIG-BASE)= {d_nb.mean()*1e4:>+7.1f}bps/trade t={t_nb:+.2f}") # ------- fee sweep (celle scelte + base) print(f"\n --- FEE SWEEP (exp bps/trade FULL | Sharpe book) ---") for k in (("base",), best_mtf, best_ntg): parts = [] for f in FEE_SWEEP_RT: pooled = [t for t in res["BTC"][k] + res["ETH"][k] if t.get("ok")] e = np.mean([t["gross"] - f for t in pooled]) * 1e4 if pooled else np.nan shf, _ = sh_dd(portfolio_daily(res, k, spans, fee_rt=f)) parts.append(f"{f*1e4:3.0f}bps: {e:+7.1f}bps/Sh {shf:+5.2f}") print(f" {key_label(k)} | " + " | ".join(parts)) # ------- esecuzione ritardata alla griglia oraria (celle MTF scelte) print(f"\n --- ESECUZIONE RITARDATA (cron orario) ---") for k in (best_mtf, best_ntg): agg = dict(n_signals=0, n_entered=0, n_skip_sl=0, n_skip_tp=0, n_missed_window=0) wsum_o = wsum_d = wsum_ps = 0.0 dsum = 0.0 for a in ASSETS: L = Low(al.get(a, tf_lo), tf_lo) d = delayed_eval(res[a][k], L) for kk in agg: agg[kk] += d[kk] wsum_o += d["exp_orig_bps"] * d["n_signals"] if d["n_signals"] else 0 wsum_d += d["exp_delayed_entered_bps"] * d["n_entered"] if d["n_entered"] else 0 wsum_ps += d["exp_delayed_per_signal_bps"] * d["n_signals"] if d["n_signals"] else 0 dsum += d["mean_delay_min"] * d["n_signals"] if d["n_signals"] else 0 ns, ne = agg["n_signals"], agg["n_entered"] print(f" {key_label(k)} | segnali {ns} | gap medio {dsum/ns if ns else np.nan:.1f}min | " f"entrati {ne} ({ne/ns*100 if ns else 0:.0f}%) skipSL {agg['n_skip_sl']} " f"skipTP {agg['n_skip_tp']} persi-finestra {agg['n_missed_window']}") print(f" exp originale {wsum_o/ns if ns else np.nan:+.1f}bps/trade -> ritardata " f"{wsum_d/ne if ne else np.nan:+.1f}bps/trade (entrati) | per-SEGNALE " f"{wsum_ps/ns if ns else np.nan:+.1f}bps") # ------- executability a $600 print(f"\n --- EXECUTABILITY $600 (cap leva {LEV_CAP}x, min order ${MIN_ORDER}) ---") for k in (("base",), best_mtf, best_ntg): tr = [t for t in res["BTC"][k] + res["ETH"][k] if t.get("ok")] if not tr: continue risks = np.array([t["risk"] for t in tr]) * 100 med = float(np.median(risks)) lev_1pct = 1.0 / med if med > 0 else np.inf yrs = (spans["BTC"][1] - spans["BTC"][0]) / (365.25 * 86400e3) tpy = len(seq_filter(res["BTC"][k])) / yrs + len(seq_filter(res["ETH"][k])) / yrs print(f" {key_label(k)} | stopMed {med:.3f}% (p25 {np.percentile(risks, 25):.3f} / " f"p75 {np.percentile(risks, 75):.3f}) | leva per rischio-1% = {lev_1pct:.1f}x " f"-> CAP {LEV_CAP}x: rischio/trade {LEV_CAP*med:.2f}% (${CAPITAL*LEV_CAP*med/100:.1f}) " f"| nozionale ${CAPITAL*LEV_CAP:.0f} > min ${MIN_ORDER} OK | ~{tpy:.0f} trade/anno (seq)") chosen_summaries.append(dict(pair=f"{tf_hi}->{tf_lo}", key=best_mtf, table=table, res=res, spans=spans, tf_lo=tf_lo)) # ---------------- DSR sul candidato scelto in-sample fra TUTTI i trial print(f"\n{'='*100}\n### GATE STATISTICI GLOBALI\n{'='*100}") best = max(chosen_summaries, key=lambda cs: cs["table"][cs["key"]]["sh_is"] if np.isfinite(cs["table"][cs["key"]]["sh_is"]) else -9) bt = best["table"][best["key"]] print(f"Candidato scelto (best in-sample fra le celle MTF): {best['pair']} {key_label(best['key'])} " f"| Sh_IS={bt['sh_is']:.2f} Sh_FULL={bt['sh_f']:.2f} Sh_HOLD={bt['sh_h']:.2f}") valid_trials = [s for s in all_trial_sharpes if np.isfinite(s)] dsr, sr0 = al.deflated_sharpe(bt["sh_f"], valid_trials, bt["port"].values) print(f"Deflated Sharpe (n_trials={len(valid_trials)}): DSR={dsr:.3f} " f"(expected null max Sharpe={sr0:.2f}) -> {'PASS' if dsr >= 0.95 else 'FAIL'} (soglia 0.95)") if np.isfinite(bt["sh_f"]) and bt["sh_f"] >= 0.5: print("\nSharpe >= 0.5 -> marginal_vs_tp01:") m = al.marginal_vs_tp01(bt["port"]) print(f" verdict={m.get('marginal_verdict')} corr_full={m.get('corr_full')} " f"uplift w25 full={m['blends']['w25']['uplift_full']} hold={m['blends']['w25']['uplift_hold']} " f"has_insample_edge={m.get('has_insample_edge')} is_hedge={m.get('is_hedge')} " f"robust_oos={m.get('robust_oos')}") else: print(f"\nSharpe full {bt['sh_f']:.2f} < 0.5 -> marginal_vs_tp01 NON eseguito (sotto soglia).") print(f"\n[runtime {time.time()-t0:.0f}s]") if __name__ == "__main__": main()