"""r0702_crt_base — CRT "Candle Range Theory", versione BASE single-TF (pattern meccanizzato). FILONE 2026-07-02. Falso breakout codificato in 3 candele (turtle soup / spring-upthrust): C1 (range): candela direzionale forte -> body/range >= b AND range >= k*ATR14 (griglia b,k) C2 (manipolazione): rompe un estremo di C1 di almeno s*ATR14 (griglia s) ma CHIUDE DENTRO il range di C1. Flag colore opzionale (short: C2 rossa che apre sopra il close di C1; long: C2 verde che apre sotto il close di C1). C3 (ingresso): entry a open C3 = close C2 (deciso con dati <= close C2 -> causale). SL = estremo di C2 (punto dello sweep). TP = estremo OPPOSTO del range di C1. Filtro R:R >= 1.3 a entry. Direzioni: short su sweep dell'alto, long su sweep del basso. OVERLAP DICHIARATO con la ricerca esistente (grep dei docstring runs/MRV*.py + MIC07): - MRV01-11 = mean-reversion su INDICATORI (RSI2, BB, z-score, IBS, W%R, consec-down, gap-fill, CCI, stochastic, VWAP-dev, %b) — nessuna testa il pattern 3-candele sweep+close-back-inside con SL/TP strutturali. La famiglia MR generica e' MORTA sul feed certificato: CRT e' una MR *condizionata da un evento di liquidita'*, quindi il prior e' fortemente negativo — serve battere il null del fade incondizionato. - MIC07 (pin-bar rejection al supporto) e' il parente piu' vicino: rejection candle single-bar a un N-bar low. CRT differisce: riferimento = range di C1 forte (1 barra), sweep quantificato in ATR, close-back-inside esplicito, TP strutturale (estremo opposto di C1) e non R-multiple. Overlap concettuale parziale, meccanica diversa. GATES: selezione cella SOLO in-sample pre-2025; deflated Sharpe su TUTTI i trial (cella x tf x direzione); ANCHOR-SHIFT (+1/+2/+4h) sul resample 4h/12h/1d; fee sweep 0.00-0.20% RT; marginal_vs_tp01 se Sharpe standalone >= 0.5. NULL decisivi: (i) fade INCONDIZIONATO dello stesso estremo (senza close-back-inside); (ii) condizione INVERTITA (C2 chiude FUORI = breakout confermato, trade col breakout). Motore trade-level CONSERVATIVO (specchia src/backtest/harness.backtest_signals): entry a close[i]; exit scan da i+1; SL/TP fillati AL LIVELLO su high/low; se nella stessa barra sono toccati entrambi scatta PRIMA lo STOP (worst-case); time-exit a close dopo max_hold barre; nessun overlap (una posizione alla volta per asset). Fee 0.10% RT. Equity mark-to-market per barra (lente Sharpe daily-compounded, convenzione progetto). Run: uv run python scripts/research/r0702_crt_base.py """ from __future__ import annotations import sys import time from itertools import product sys.path.insert(0, "/opt/docker/PythagorasGoal/scripts/research/alt") import altlib as al # noqa: E402 import numpy as np # noqa: E402 import pandas as pd # noqa: E402 HOLDOUT = al.HOLDOUT # 2025-01-01 UTC FEE_RT = 2 * al.FEE_SIDE # 0.10% round-trip RR_MIN = 1.3 TFS = ("1h", "4h", "12h", "1d") RULES = {"4h": "4h", "12h": "12h", "1d": "1D"} ASSETS = al.CERTIFIED # BTC, ETH MIN_IS_TRADES = 25 # trade combinati minimi in-sample per cella eleggibile GRID = [dict(b=b, k=k, s=s, color=col, mh=mh) for b, k, s, col, mh in product((0.5, 0.7), (1.0, 1.5), (0.0, 0.1, 0.25), (False, True), (5, 10, 20))] DIRS = ("long", "short", "both") # =========================================================================== # DATI (anchor 00:00 UTC di default; anchor spostabile per il gate anchor-shift) # =========================================================================== def resample_anchor(df_1h: pd.DataFrame, rule: str, offset_hours: int) -> pd.DataFrame: """Come trend_portfolio.resample_tf (label/closed='left') ma con ancora spostata di +offset_hours. Niente .view('int64'): epoca esplicita via // Timedelta (tz-aware safe).""" g = df_1h.copy() idx = pd.to_datetime(g["timestamp"], unit="ms", utc=True) idx.name = "dt" g.index = idx out = g.resample(rule, label="left", closed="left", offset=pd.Timedelta(hours=offset_hours)).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"]] _PREP_CACHE: dict = {} def prep(asset: str, tf: str, anchor: int = 0) -> dict: key = (asset, tf, anchor) if key in _PREP_CACHE: return _PREP_CACHE[key] if anchor == 0 or tf == "1h": df = al.get(asset, tf) else: df = resample_anchor(al.get(asset, "1h"), RULES[tf], anchor) d = dict( df=df, o=df["open"].values.astype(float), h=df["high"].values.astype(float), l=df["low"].values.astype(float), c=df["close"].values.astype(float), atr=al.atr(df, 14), idx=pd.DatetimeIndex(pd.to_datetime(df["datetime"], utc=True)), ) _PREP_CACHE[key] = d return d # =========================================================================== # DETECTION (vettoriale, causale: tutto deciso con OHLC fino alla barra i = C2; # l'ATR usato e' quello di C1 (i-1) -> ancora piu' conservativo) # =========================================================================== def _shift1(x: np.ndarray) -> np.ndarray: out = np.empty_like(x); out[0] = np.nan; out[1:] = x[:-1] return out def detect(d: dict, b: float, k: float, s: float, color: bool, variant: str) -> dict: """Ritorna {dir: (indici C2, sl, tp)} per variant in {'crt','fade','breakout'}. Indice i = barra C2; C1 = i-1. Entry (gestita dal motore) = close[i].""" o, h, l, c = d["o"], d["h"], d["l"], d["c"] h1, l1, o1, c1 = _shift1(h), _shift1(l), _shift1(o), _shift1(c) atr1 = _shift1(d["atr"]) rng1 = h1 - l1 body1 = np.abs(c1 - o1) with np.errstate(invalid="ignore", divide="ignore"): strong = (np.isfinite(atr1) & (atr1 > 0) & (rng1 > 0) & (body1 / rng1 >= b) & (rng1 >= k * atr1)) sweep_up = strong & (h > h1 + s * atr1) # C2 rompe l'alto di C1 sweep_dn = strong & (l < l1 - s * atr1) # C2 rompe il basso di C1 out = {} if variant == "crt": sh = sweep_up & (c < h1) & (c > l1) # chiude DENTRO il range di C1 lg = sweep_dn & (c > l1) & (c < h1) if color: sh &= (c < o) & (o > c1) # rossa che apre sopra il close di C1 lg &= (c > o) & (o < c1) # verde che apre sotto il close di C1 sh &= (h > c) & ((c - l1) / np.where(h - c > 0, h - c, np.nan) >= RR_MIN) lg &= (c > l) & ((h1 - c) / np.where(c - l > 0, c - l, np.nan) >= RR_MIN) out["short"] = (np.where(sh)[0], h, l1) # SL=high C2, TP=low C1 out["long"] = (np.where(lg)[0], l, h1) # SL=low C2, TP=high C1 elif variant == "fade": # NULL (i): stesso sweep, NESSUNA richiesta di close-back-inside (no colore). # Solo validita' geometrica (TP dal lato giusto dell'entry). sh = sweep_up & (c > l1) lg = sweep_dn & (c < h1) sh &= (h > c) & ((c - l1) / np.where(h - c > 0, h - c, np.nan) >= RR_MIN) lg &= (c > l) & ((h1 - c) / np.where(c - l > 0, c - l, np.nan) >= RR_MIN) out["short"] = (np.where(sh)[0], h, l1) out["long"] = (np.where(lg)[0], l, h1) elif variant == "breakout": # NULL (ii): condizione INVERTITA — C2 chiude FUORI dal range di C1 = # breakout confermato, trade IN DIREZIONE del breakout. # SL = livello rotto (rientro nel range = fallimento), TP = measured move # (range di C1 proiettato oltre il livello). Stesso filtro R:R. lg = sweep_up & (c > h1) # rompe l'alto e chiude sopra -> LONG sh = sweep_dn & (c < l1) # rompe il basso e chiude sotto -> SHORT tp_lg = h1 + rng1 tp_sh = l1 - rng1 lg &= (tp_lg > c) & ((tp_lg - c) / np.where(c - h1 > 0, c - h1, np.nan) >= RR_MIN) sh &= (c > tp_sh) & ((c - tp_sh) / np.where(l1 - c > 0, l1 - c, np.nan) >= RR_MIN) out["long"] = (np.where(lg)[0], h1, tp_lg) # SL=high C1, TP=measured move out["short"] = (np.where(sh)[0], l1, tp_sh) # SL=low C1 else: raise ValueError(variant) return out def merge_dirs(sig: dict, which: str): """Lista ordinata di (i, dir, sl, tp) per direzione 'long'/'short'/'both'.""" rows = [] if which in ("long", "both"): ii, sl, tp = sig["long"] rows += [(int(i), 1, float(sl[i]), float(tp[i])) for i in ii] if which in ("short", "both"): ii, sl, tp = sig["short"] rows += [(int(i), -1, float(sl[i]), float(tp[i])) for i in ii] rows.sort(key=lambda r: r[0]) return rows # =========================================================================== # MOTORE TRADE-LEVEL (conservativo; specchia backtest_signals: SL prioritario) # =========================================================================== def run_trades(d: dict, rows: list, mh: int, fee_rt: float = FEE_RT): """Ritorna (trades, barnet). trades: (i_entry, i_exit, dir, net, R, gross). barnet: rendimento netto per-barra mark-to-market (fee 50/50 su entry/exit bar).""" c, h, l = d["c"], d["h"], d["l"] n = len(c) barnet = np.zeros(n) trades = [] busy_until = -1 for i, dr, sl, tp in rows: if i <= busy_until or i >= n - 1: continue entry = c[i] exit_idx = min(i + mh, n - 1) exit_price = c[exit_idx] for j in range(i + 1, min(i + mh, n - 1) + 1): if dr == 1: if l[j] <= sl: # STOP prima (worst-case) exit_price, exit_idx = sl, j; break if h[j] >= tp: exit_price, exit_idx = tp, j; break else: if h[j] >= sl: exit_price, exit_idx = sl, j; break if l[j] <= tp: exit_price, exit_idx = tp, j; break exit_price, exit_idx = c[j], j gross = dr * (exit_price / entry - 1.0) net = gross - fee_rt risk = abs(sl - entry) / entry R = net / risk if risk > 0 else np.nan trades.append((i, exit_idx, dr, net, R, gross)) pp = entry for j in range(i + 1, exit_idx + 1): pj = exit_price if j == exit_idx else c[j] barnet[j] += dr * (pj / pp - 1.0) pp = pj barnet[i] -= fee_rt / 2 barnet[exit_idx] -= fee_rt / 2 busy_until = exit_idx return trades, barnet def daily_series(d: dict, barnet: np.ndarray) -> pd.Series: return al._to_daily(pd.Series(barnet, index=d["idx"])) def combo_daily(dailies: dict) -> pd.Series: J = pd.concat(dailies, axis=1, join="inner").fillna(0.0) return 0.5 * J[ASSETS[0]] + 0.5 * J[ASSETS[1]] def series_metrics(daily: pd.Series) -> dict: def _dd(s): eq = np.cumprod(1.0 + s.values); pk = np.maximum.accumulate(eq) return float(np.max((pk - eq) / pk)) if len(eq) else 0.0 ins, hold = daily[daily.index < HOLDOUT], daily[daily.index >= HOLDOUT] yearly = {int(y): round(float(np.prod(1 + g.values) - 1), 4) for y, g in daily.groupby(daily.index.year)} return dict(full_sh=round(al._sh(daily), 3), is_sh=round(al._sh(ins), 3), hold_sh=round(al._sh(hold), 3), full_dd=round(_dd(daily), 4), hold_dd=round(_dd(hold), 4), full_ret=round(float(np.prod(1 + daily.values) - 1), 4), hold_ret=round(float(np.prod(1 + hold.values) - 1), 4), yearly=yearly) def trade_stats(trades: list, idx: pd.DatetimeIndex) -> dict: if not trades: return dict(n=0, n_is=0, n_hold=0, wr=None, avg_R=None, exp_net=None) t_entry = idx[[t[0] for t in trades]] net = np.array([t[3] for t in trades]); R = np.array([t[4] for t in trades]) is_m = np.asarray(t_entry < HOLDOUT) def _blk(m): if m.sum() == 0: return dict(n=0, wr=None, avg_R=None, exp_net=None) return dict(n=int(m.sum()), wr=round(float(np.mean(net[m] > 0) * 100), 1), avg_R=round(float(np.nanmean(R[m])), 3), exp_net=round(float(np.mean(net[m]) * 100), 3)) full = _blk(np.ones(len(net), bool)) per_year = {} for y in sorted(set(t_entry.year)): per_year[int(y)] = _blk(np.asarray(t_entry.year == y)) return dict(n=full["n"], n_is=int(is_m.sum()), n_hold=int((~is_m).sum()), wr=full["wr"], avg_R=full["avg_R"], exp_net=full["exp_net"], is_blk=_blk(is_m), hold_blk=_blk(~is_m), per_year=per_year) # =========================================================================== # RUNNER di un trial (cella x tf x direzione) su entrambi gli asset # =========================================================================== def run_trial(tf: str, p: dict, which: str, variant: str = "crt", fee_rt: float = FEE_RT, anchor: int = 0): dailies, all_trades, all_stats = {}, {}, {} for a in ASSETS: d = prep(a, tf, anchor) sig = detect(d, p["b"], p["k"], p["s"], p["color"], variant) rows = merge_dirs(sig, which) trades, barnet = run_trades(d, rows, p["mh"], fee_rt) dailies[a] = daily_series(d, barnet) all_trades[a] = trades all_stats[a] = trade_stats(trades, d["idx"]) daily = combo_daily(dailies) sm = series_metrics(daily) n_is = sum(st["n_is"] for st in all_stats.values()) n_full = sum(st["n"] for st in all_stats.values()) return dict(tf=tf, params=p, dir=which, variant=variant, daily=daily, metrics=sm, per_asset_stats=all_stats, n_is=n_is, n_full=n_full) def pooled_trade_stats(trial: dict) -> dict: """Statistiche trade POOLED sui due asset (per il report della cella scelta).""" trades, idxs = [], [] for a in ASSETS: d = prep(a, trial["tf"]) for t in trial["_raw_trades"][a]: trades.append(t); idxs.append(d["idx"][t[0]]) if not trades: return dict(n=0) order = np.argsort(np.array([i.value for i in idxs])) trades = [trades[k] for k in order] idx = pd.DatetimeIndex([idxs[k] for k in order]) return _pooled(trades, idx) def _pooled(trades, idx): net = np.array([t[3] for t in trades]); R = np.array([t[4] for t in trades]) is_m = np.asarray(idx < HOLDOUT) def _blk(m): if m.sum() == 0: return dict(n=0, wr=None, avg_R=None, exp_net=None) return dict(n=int(m.sum()), wr=round(float(np.mean(net[m] > 0) * 100), 1), avg_R=round(float(np.nanmean(R[m])), 3), exp_net=round(float(np.mean(net[m]) * 100), 3)) out = dict(full=_blk(np.ones(len(net), bool)), is_blk=_blk(is_m), hold_blk=_blk(~is_m), per_year={int(y): _blk(np.asarray(idx.year == y)) for y in sorted(set(idx.year))}) return out # =========================================================================== # MAIN # =========================================================================== def main(): t0 = time.time() print("=" * 96) print("r0702 CRT — Candle Range Theory BASE single-TF | fee 0.10% RT | hold-out 2025-01-01+") print("Griglia: b(0.5,0.7) x k(1.0,1.5) x s(0.0,0.1,0.25) x color(off,on) x max_hold(5,10,20)") print(f"= {len(GRID)} celle x {len(TFS)} TF x {len(DIRS)} direzioni = " f"{len(GRID) * len(TFS) * len(DIRS)} trial (tutti contati nel DSR)") print("=" * 96) for a in ASSETS: d = prep(a, "1d") print(f" dati {a} 1d: {d['idx'][0].date()} -> {d['idx'][-1].date()} ({len(d['c'])} barre)") # ---- 1) griglia completa (righe leggere; il daily si ricalcola per la scelta) ---- rows = [] freq_by_tf = {tf: [] for tf in TFS} for tf in TFS: years = {} for a in ASSETS: d = prep(a, tf) years[a] = (d["idx"][-1] - d["idx"][0]).total_seconds() / 86400 / 365.25 span_y = float(np.mean(list(years.values()))) for p in GRID: # detection condivisa fra direzioni e mh (mh influenza solo il motore) for which in DIRS: tr = run_trial(tf, p, which) m = tr["metrics"] rows.append(dict(tf=tf, **p, dir=which, is_sh=m["is_sh"], full_sh=m["full_sh"], hold_sh=m["hold_sh"], n_is=tr["n_is"], n_full=tr["n_full"])) if which == "both": freq_by_tf[tf].append(tr["n_full"] / (2 * span_y)) # trade/anno per asset print(f" [grid] tf={tf} fatto ({time.time() - t0:.0f}s)") R = pd.DataFrame(rows) print("\n--- FREQUENZA PATTERN (CRT, entrambe le direzioni, trade/anno PER ASSET) ---") for tf in TFS: f = np.array(freq_by_tf[tf]) print(f" {tf:>4s}: mediana {np.median(f):6.1f} min {f.min():6.1f} max {f.max():6.1f} " f"(su {len(f)} celle)") # ---- 2) selezione cella SOLO in-sample (pre-2025) ---- elig = R[(R.n_is >= MIN_IS_TRADES) & np.isfinite(R.is_sh)].copy() print(f"\n--- SELEZIONE IN-SAMPLE: {len(elig)}/{len(R)} trial eleggibili " f"(>= {MIN_IS_TRADES} trade IS combinati) ---") top = elig.sort_values("is_sh", ascending=False).head(12) cols = ["tf", "b", "k", "s", "color", "mh", "dir", "is_sh", "hold_sh", "full_sh", "n_is", "n_full"] print(top[cols].to_string(index=False)) if len(elig) == 0: print("\nVERDETTO: FAIL — nessuna cella con abbastanza trade in-sample.") return best = elig.sort_values("is_sh", ascending=False).iloc[0] p = dict(b=float(best.b), k=float(best.k), s=float(best.s), color=bool(best.color), mh=int(best.mh)) tf, which = str(best.tf), str(best.dir) print(f"\n=== CELLA SCELTA (max Sharpe IN-SAMPLE, hold-out solo riportato) ===") print(f" tf={tf} dir={which} {p}") # ricalcolo completo della cella scelta (con trade grezzi per il pooled report) chosen = run_trial(tf, p, which) chosen["_raw_trades"] = {} for a in ASSETS: d = prep(a, tf) sig = detect(d, p["b"], p["k"], p["s"], p["color"], "crt") trades, _ = run_trades(d, merge_dirs(sig, which), p["mh"]) chosen["_raw_trades"][a] = trades m = chosen["metrics"] print(f" COMBINED 50/50: FULL Sh {m['full_sh']} IS Sh {m['is_sh']} HOLD Sh {m['hold_sh']} " f"| FULL ret {m['full_ret'] * 100:+.1f}% DD {m['full_dd'] * 100:.1f}% " f"| HOLD ret {m['hold_ret'] * 100:+.1f}% DD {m['hold_dd'] * 100:.1f}%") print(f" per-anno (ret combo): " + " ".join(f"{y}:{v * 100:+.1f}%" for y, v in m["yearly"].items())) ps = pooled_trade_stats(chosen) if ps.get("full", {}).get("n", 0) > 0: f_, i_, h_ = ps["full"], ps["is_blk"], ps["hold_blk"] print(f" trade POOLED: n={f_['n']} WR={f_['wr']}% avgR={f_['avg_R']} exp={f_['exp_net']}%" f" | IS n={i_['n']} WR={i_['wr']}% avgR={i_['avg_R']} exp={i_['exp_net']}%" f" | HOLD n={h_['n']} WR={h_['wr']}% avgR={h_['avg_R']} exp={h_['exp_net']}%") print(" trade per anno: " + " ".join( f"{y}:n{b['n']}/wr{b['wr']}/exp{b['exp_net']}%" for y, b in ps["per_year"].items())) for a in ASSETS: st = chosen["per_asset_stats"][a] print(f" {a}: n={st['n']} (IS {st['n_is']}/HOLD {st['n_hold']}) WR={st['wr']}% " f"avgR={st['avg_R']} exp={st['exp_net']}%") # per-direzione della cella scelta (stessi parametri) print("\n--- CELLA SCELTA per DIREZIONE (stessi parametri) ---") for wdir in DIRS: tr = run_trial(tf, p, wdir) mm = tr["metrics"] print(f" {wdir:>5s}: IS {mm['is_sh']:+.2f} HOLD {mm['hold_sh']:+.2f} FULL {mm['full_sh']:+.2f} " f"n={tr['n_full']} (IS {tr['n_is']})") # ---- sanity cross-check vs harness ufficiale (al.eval_signals) ---- print("\n--- CROSS-CHECK vs al.eval_signals (harness ufficiale, stessa convenzione) ---") for a in ASSETS: d = prep(a, tf) sig = detect(d, p["b"], p["k"], p["s"], p["color"], "crt") entries = [None] * len(d["c"]) for i, dr, sl, tp in merge_dirs(sig, which): entries[i] = dict(dir=dr, sl=sl, tp=tp, max_bars=p["mh"]) ev = al.eval_signals(d["df"], entries, fee_rt=FEE_RT, asset=a, tf=tf) mine = chosen["_raw_trades"][a] my_ret = float(np.prod([1 + t[3] for t in mine]) - 1) print(f" {a}: harness n={ev['n_trades']} ret={ev['full']['ret'] * 100:+.1f}% " f"| mio n={len(mine)} ret={my_ret * 100:+.1f}% " f"{'OK' if ev['n_trades'] == len(mine) else 'MISMATCH!'}") # ---- 3) DSR su TUTTI i trial ---- all_sr = [r["full_sh"] for r in rows if np.isfinite(r["full_sh"]) and r["n_full"] >= 1] dsr, sr0 = al.deflated_sharpe(m["full_sh"], all_sr, chosen["daily"]) print(f"\n--- DEFLATED SHARPE: DSR={dsr:.3f} (PASS>=0.95) expected-null-max Sh={sr0:.2f} " f"trial contati={len(all_sr)} (su {len(rows)} totali; esclusi 0-trade) ---") # ---- 4) ANCHOR-SHIFT (+1/+2/+4h) ---- print("\n--- ANCHOR-SHIFT (ancora resample spostata; pattern vero ~invariante) ---") anchor_rows = {} if tf == "1h": print(" tf=1h nativo: nessuna dipendenza dall'ancora del resample (N/A).") alt = elig[elig.tf != "1h"].sort_values("is_sh", ascending=False) if len(alt): b2 = alt.iloc[0] p2 = dict(b=float(b2.b), k=float(b2.k), s=float(b2.s), color=bool(b2.color), mh=int(b2.mh)) print(f" (test eseguito sulla miglior cella IS a tf>=4h: tf={b2.tf} dir={b2.dir} {p2})") for off in (0, 1, 2, 4): tr = run_trial(str(b2.tf), p2, str(b2.dir), anchor=off) mm = tr["metrics"] anchor_rows[off] = mm print(f" anchor +{off}h: IS {mm['is_sh']:+.2f} HOLD {mm['hold_sh']:+.2f} " f"FULL {mm['full_sh']:+.2f} n={tr['n_full']}") else: for off in (0, 1, 2, 4): tr = run_trial(tf, p, which, anchor=off) mm = tr["metrics"] anchor_rows[off] = mm print(f" anchor +{off}h: IS {mm['is_sh']:+.2f} HOLD {mm['hold_sh']:+.2f} " f"FULL {mm['full_sh']:+.2f} n={tr['n_full']}") if anchor_rows: fulls = [v["full_sh"] for v in anchor_rows.values()] flip = (max(fulls) > 0) and (min(fulls) < 0) print(f" spread FULL Sh = {max(fulls) - min(fulls):+.2f} " f"{'SIGN-FLIP -> ARTIFACT-RISK' if flip else 'nessun sign-flip'}") # ---- 5) FEE SWEEP 0.00-0.20% RT ---- print("\n--- FEE SWEEP (cella scelta) ---") for fr in (0.0, 0.0005, 0.001, 0.0015, 0.002): tr = run_trial(tf, p, which, fee_rt=fr) mm = tr["metrics"] print(f" fee {fr * 100:.2f}%RT: IS {mm['is_sh']:+.2f} HOLD {mm['hold_sh']:+.2f} " f"FULL {mm['full_sh']:+.2f}") # ---- 6) NULL DECISIVI ---- print("\n--- NULL (i): FADE INCONDIZIONATO dello stesso estremo (no close-back-inside) ---") p_null = dict(p, color=False) for var, lbl in (("fade", "fade-incond"), ("breakout", "breakout-conf")): tr = run_trial(tf, p_null, which, variant=var) mm = tr["metrics"] print(f" {lbl:>14s}: IS {mm['is_sh']:+.2f} HOLD {mm['hold_sh']:+.2f} FULL {mm['full_sh']:+.2f} " f"n={tr['n_full']} (IS {tr['n_is']}) per-anno " + " ".join(f"{y}:{v * 100:+.0f}%" for y, v in mm["yearly"].items())) tr = run_trial(tf, p, which, variant="crt") mm = tr["metrics"] print(f" {'CRT (rif.)':>14s}: IS {mm['is_sh']:+.2f} HOLD {mm['hold_sh']:+.2f} FULL {mm['full_sh']:+.2f} " f"n={tr['n_full']} (IS {tr['n_is']})") # ---- 7) MARGINAL vs TP01 (solo se standalone >= 0.5) ---- if max(m["full_sh"], m["is_sh"]) >= 0.5: print("\n--- MARGINAL vs TP01 (standalone >= 0.5) ---") marg = al.marginal_vs_tp01(chosen["daily"]) keys = ("marginal_verdict", "corr_full", "corr_hold", "cand_insample_sharpe", "has_insample_edge", "is_hedge", "multicut_uplift", "multicut_persistent", "robust_oos", "beta_to_tp01", "resid_sharpe_full") for kk in keys: print(f" {kk}: {marg.get(kk)}") for w, dd in marg.get("blends", {}).items(): print(f" blend {w}: full {dd['full']} (uplift {dd['uplift_full']:+.3f}) " f"hold {dd['hold']} (uplift {dd['uplift_hold']})") else: print(f"\n--- MARGINAL vs TP01: SALTATO (standalone FULL {m['full_sh']} / IS {m['is_sh']} < 0.5) ---") print(f"\n[done in {time.time() - t0:.0f}s]") if __name__ == "__main__": main()