#!/usr/bin/env python """r0707_crt_topdown.py — "CRT top-down multi-TF" (video-claim, 74% win rate) — 2026-07-07. CLAIM DAL VIDEO (metodo ICT/SMC top-down, 74% WR "da quanto sto testando"): 1. H1 — setup CRT: candela di displacement forte, poi candela che PRENDE LIQUIDITA' (spike oltre il range di C1) ma RICHIUDE dentro il range -> falso breakout -> fade. 2. M15 — struttura: swing (U a 3 candele) + imbalance che diventa inverse-imbalance (FVG mitigato). 3. M5 — conferma: displacement direzionale + zona di protezione (order block / imbalance). 4. M1 — entry: attende la correzione, entra con SL DIETRO la zona di protezione. Uscita: TP1 "a fine zona CRT" a RR ~1.5-2, chiude 70-80% li' -> resto a BREAK-EVEN -> runner 20-30% verso la "liquidita' successiva" (previous daily high/low). COSA E' GIA' STATO TESTATO E SCARTATO (non lo rifaccio): 2026-07-02 CRT wave, 3 tagli — base single-TF (r0702_crt_base, 864 trial, DSR 0.000), multi-TF (r0702_crt_mtf, ~10k trade, expectancy netta NEGATIVA ovunque, "il ritest e' informazione negativa"), contesto/FVG (r0702_crt_context, FVG semmai peggiora; FADE < FOLLOW ogni anno). Il setup H1 CRT NON ha edge direzionale su BTC/ETH certificati. ANGOLO NUOVO (non coperto dalle 3 onde precedenti) = LA GESTIONE D'USCITA e IL "74% WR": la scuola SMC mostra win-rate alti perche' usa PARZIALE-a-1.5R + BREAK-EVEN + runner. TESI DA FALSIFICARE: **il 74% e' un ARTEFATTO di misura dello schema parziale+BE, NON un edge**. Con parziale+BE un trade e' "perdente" SOLO se lo stop scatta PRIMA di toccare 1.5R; tutto il resto (anche un ritorno a BE dopo aver bankato il 75% a 1.5R) e' bookato come "vinto". Quindi WR_managed = P(tocca +1.5R prima di -1R). Per un random walk senza edge P(+aR prima di -1R)=1/(1+a) -> a=1.5 => 40%. Il numero che conta e' l'EXPECTANCY in R netto fee, invariante allo schema. DISEGNO (onesto, riusa l'harness del progetto per dati/atr/sharpe/DSR): - Detection CRT C1-C2 su H1 (1h nativo, causale: nota alla chiusura di C2). - Entry sul TF basso (5m/15m, proxy di M5/M1 — NB: M1 non e' nel feed certificato): ritest della zona violata + CLOSE-BACK attraverso il livello ("displacement di conferma"), SL = estremo dello swing basso dalla apertura finestra (la "zona di protezione"). - Tre schemi d'uscita sugli STESSI ingressi: FIXED-1.5 : full size, TP a 1.5R, SL protezione. FIXED-2.0 : full size, TP a 2.0R. MANAGED : 75% a 1.5R -> resto 25% a BE, runner target = prev-daily high/low (o estremo opposto di C1 se piu' lontano), altrimenti chiude a BE/tempo. (schema del video) - Null decisivo: FOLLOW (trada in direzione del breakout, non il fade) sugli stessi pattern. - Gate progetto: hold-out 2025-01-01, fee sweep 0/0.10/0.20% RT, griglia (dir x RR x zona x forza C1) con DSR su tutti i trial, executability $600. Nessun file scritto; non tocca src/live. Run: uv run python scripts/research/r0707_crt_topdown.py """ from __future__ import annotations import sys import time 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 HTF = "1h" LTFS = ("5m", "15m") # proxy M5/M1 (M1 non certificato -> dichiarato) ASSETS = al.CERTIFIED # BTC, ETH FEE_RT = 2 * al.FEE_SIDE # 0.10% round-trip FEE_SWEEP = (0.0, 0.001, 0.002) HOLDOUT = al.HOLDOUT HOLDOUT_MS = int(HOLDOUT.value // 10**6) TF_MS = {"5m": 300_000, "15m": 900_000, "1h": 3_600_000} CAPITAL = 600.0 MIN_ORDER = 5.0 LEV_CAP = 2.0 # --- griglia a priori (chiusa prima di guardare i risultati) --- K_GRID = (1.0, 1.5) # C1 forte: range >= k*ATR14(H1) S_GRID = (0.0, 0.1) # C2 sweep: rompe l'estremo di C1 di s*ATR D_GRID = (0.15, 0.35) # zona di ritest = d*ATR intorno al livello WIN_HTF = 4 # finestra ritest+entry: 4 barre H1 dopo la chiusura di C2 MAXHOLD_HTF = 24 # holding massimo del trade: 24 barre H1 (1 giorno) di barre basse RR1 = 1.5 # parziale "fine zona CRT" RR2 = 2.0 RUNNER_CAP_R = 3.0 # se prevday non oltre TP1, runner target = 3R (cap) ATR_MULT_DEFAULT = 1.2 # =========================================================================== # DETECTION CRT C1-C2 su H1 (vettoriale, causale). i = barra C2; C1 = i-1. # =========================================================================== def detect_patterns(dfh: pd.DataFrame, k: float, s: float) -> list[dict]: ts = dfh["timestamp"].astype("int64").values h = dfh["high"].values.astype(float) l = dfh["low"].values.astype(float) c = dfh["close"].values.astype(float) a = al.atr(dfh, 14) h1 = np.roll(h, 1); l1 = np.roll(l, 1); a1 = np.roll(a, 1) rng1 = h1 - l1 strong = (a1 > 0) & (rng1 >= k * a1) up = strong & (h > h1 + s * a1) & (c < h1) & (c > l1) # sweep alto, close dentro -> SHORT dn = strong & (l < l1 - s * a1) & (c > l1) & (c < h1) # sweep basso, close dentro -> LONG idx = np.where(up | dn)[0] tf_ms = TF_MS[HTF] pats = [] for i in idx: if i < 20: continue if up[i]: d, level, c1_opp, c2ext = -1, float(h1[i]), float(l1[i]), float(h[i]) else: d, level, c1_opp, c2ext = +1, float(l1[i]), float(h1[i]), float(l[i]) pats.append(dict(i=int(i), dir=d, level=level, c1_opp=c1_opp, c2ext=c2ext, atr=float(a1[i]), win_open=int(ts[i] + tf_ms), win_close=int(ts[i] + (1 + WIN_HTF) * tf_ms), hold_end=int(ts[i] + (1 + MAXHOLD_HTF) * tf_ms))) return pats # =========================================================================== # LOW-TF arrays + prev-daily levels (runner target = "previous daily high/low") # =========================================================================== class Low: __slots__ = ("ts", "o", "h", "l", "c", "tsclose", "n", "pdh", "pdl") 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) # prev-day high/low, causale: per ogni barra, estremi del giorno UTC PRECEDENTE completo dti = pd.to_datetime(self.ts, unit="ms", utc=True) day = dti.floor("D") g = pd.DataFrame({"day": day, "h": self.h, "l": self.l}) dh = g.groupby("day")["h"].max() dl = g.groupby("day")["l"].min() pdh_map = dh.shift(1) # high del giorno prima pdl_map = dl.shift(1) self.pdh = pdh_map.reindex(day).values.astype(float) self.pdl = pdl_map.reindex(day).values.astype(float) # =========================================================================== # ENTRY (ritest + close-back sul TF basso) -> (entry, entry_ts, stop, jt, j1) # =========================================================================== def find_entry(p: dict, L: Low, d_mult: float): j0 = int(np.searchsorted(L.ts, p["win_open"])) jw = int(np.searchsorted(L.ts, p["win_close"])) if j0 >= L.n or L.ts[j0] >= p["win_close"]: return None dr, level, c2ext = p["dir"], p["level"], p["c2ext"] zone = d_mult * p["atr"] Lh, Ll, Lc = L.h, L.l, L.c touched = False jt = -1 swing = -np.inf if dr < 0 else np.inf for j in range(j0, jw): if dr < 0: swing = max(swing, Lh[j]) if Lh[j] >= c2ext: # struttura violata prima del trigger return None if Lh[j] >= level - zone: touched = True if touched and Lc[j] < level: # close-back sotto il livello -> conferma jt = j; break else: swing = min(swing, Ll[j]) if Ll[j] <= c2ext: return None if Ll[j] <= level + zone: touched = True if touched and Lc[j] > level: jt = j; break if jt < 0: return None entry = float(Lc[jt]) stop = float(max(swing, Lh[jt])) if dr < 0 else float(min(swing, Ll[jt])) if (dr < 0 and not (entry < stop)) or (dr > 0 and not (entry > stop)): return None j1 = int(np.searchsorted(L.ts, p["hold_end"])) return dict(dir=dr, entry=entry, stop=stop, jt=jt, j1=min(j1, L.n), entry_ts=int(L.tsclose[jt])) # =========================================================================== # EXIT SCHEMES sugli stessi ingressi. Ritorna gross_R (in multipli di R, size-1 base), # is_win (bool), kind. R (frazione prezzo) = |entry-stop|/entry. # Convenzione conservativa: nella stessa barra lo STOP scatta prima del target. # =========================================================================== def _scan_fixed(en, L, rr, fee_rt): dr, entry, stop = en["dir"], en["entry"], en["stop"] risk = abs(entry - stop) / entry tp = entry * (1 + dr * rr * risk) Lh, Ll, Lc = L.h, L.l, L.c for j in range(en["jt"] + 1, en["j1"]): if dr < 0: if Lh[j] >= stop: return -1.0, False, "SL", risk if Ll[j] <= tp: return rr, True, "TP", risk else: if Ll[j] <= stop: return -1.0, False, "SL", risk if Lh[j] >= tp: return rr, True, "TP", risk # time exit last = en["j1"] - 1 if last <= en["jt"]: return 0.0, False, "NOBARS", risk g = dr * (Lc[last] / entry - 1.0) / risk if risk > 0 else 0.0 return g, g > 0, "TIME", risk def _scan_managed(en, L, p, rr1=RR1, fee_rt=FEE_RT): """75% a rr1 -> resto 25% a BE, runner = prev-daily high/low (o estremo opposto C1 / cap).""" dr, entry, stop = en["dir"], en["entry"], en["stop"] risk = abs(entry - stop) / entry tp1 = entry * (1 + dr * rr1 * risk) Lh, Ll, Lc = L.h, L.l, L.c jt, j1 = en["jt"], en["j1"] # runner target in prezzo: prev-daily level nel verso del trade, oltre tp1; else estremo opp C1; else cap if dr > 0: cands = [x for x in (L.pdh[jt], p["c1_opp"]) if np.isfinite(x) and x > tp1] runner = min(cands) if cands else entry * (1 + RUNNER_CAP_R * risk) else: cands = [x for x in (L.pdl[jt], p["c1_opp"]) if np.isfinite(x) and x < tp1] runner = max(cands) if cands else entry * (1 - RUNNER_CAP_R * risk) runner_R = abs(runner - entry) / entry / risk if risk > 0 else 0.0 # fase 1: fino a tp1 (o stop) j_tp1 = -1 for j in range(jt + 1, j1): if dr < 0: if Lh[j] >= stop: return -1.0, False, "SL", risk # perde: stop prima del parziale if Ll[j] <= tp1: j_tp1 = j; break else: if Ll[j] <= stop: return -1.0, False, "SL", risk if Lh[j] >= tp1: j_tp1 = j; break if j_tp1 < 0: # non ha raggiunto tp1 ne' stop: chiude a tempo sul resto pieno (nessun parziale) last = j1 - 1 if last <= jt: return 0.0, False, "NOBARS", risk g = dr * (Lc[last] / entry - 1.0) / risk if risk > 0 else 0.0 return g, g > 0, "TIME", risk # bankato 75% a rr1; resto 25% con stop a BE (=entry), target=runner for j in range(j_tp1 + 1, j1): if dr < 0: if Lh[j] >= entry: # ritorno a BE return 0.75 * rr1 + 0.25 * 0.0, True, "BE-RUN", risk if Ll[j] <= runner: return 0.75 * rr1 + 0.25 * runner_R, True, "RUNNER", risk else: if Ll[j] <= entry: return 0.75 * rr1 + 0.25 * 0.0, True, "BE-RUN", risk if Lh[j] >= runner: return 0.75 * rr1 + 0.25 * runner_R, True, "RUNNER", risk # resto chiude a tempo last = j1 - 1 grun = dr * (Lc[last] / entry - 1.0) / risk if risk > 0 else 0.0 return 0.75 * rr1 + 0.25 * grun, True, "TIME-RUN", risk def net_frac(gross_R, risk, fee_rt): """Rendimento netto (frazione prezzo) del trade a 1x nozionale: gross_R*risk - fee_rt.""" return gross_R * risk - fee_rt # =========================================================================== # BOOK / STATS # =========================================================================== def build_trades(pats, L, d_mult, scheme, p_by_i, fee_rt=FEE_RT, rr1=RR1): """scheme in {'fix1.5','fix2.0','managed'}. Ritorna lista di trade dict (sequenziali).""" raw = [] for p in pats: en = find_entry(p, L, d_mult) if en is None: continue if scheme == "managed": gR, win, kind, risk = _scan_managed(en, L, p, rr1, fee_rt) elif scheme == "fix2.0": gR, win, kind, risk = _scan_fixed(en, L, RR2, fee_rt) else: gR, win, kind, risk = _scan_fixed(en, L, RR1, fee_rt) nf = net_frac(gR, risk, fee_rt) raw.append(dict(entry_ts=en["entry_ts"], exit_ts=en["entry_ts"], gR=gR, win=win, kind=kind, risk=risk, net=nf, dir=en["dir"])) # sequenziale (una posizione alla volta) sull'entry_ts raw.sort(key=lambda t: t["entry_ts"]) return raw def seq_filter(trades): out, last = [], -1 for t in trades: if t["entry_ts"] >= last: out.append(t); last = t["exit_ts"] return out def stats(trades, fee_rt=FEE_RT, mask=None): tr = trades if mask is None else [t for t in trades if mask(t)] if not tr: return dict(n=0, wr=np.nan, exp_R=np.nan, exp_net_bps=np.nan, med_risk=np.nan) # expectancy in R netto fee: (gross_R - fee_rt/risk) R_net = np.array([t["gR"] - fee_rt / t["risk"] if t["risk"] > 0 else t["gR"] for t in tr]) wr = float(np.mean([t["win"] for t in tr]) * 100) net = np.array([t["gR"] * t["risk"] - fee_rt for t in tr]) return dict(n=len(tr), wr=wr, exp_R=float(R_net.mean()), exp_net_bps=float(net.mean() * 1e4), med_risk=float(np.median([t["risk"] for t in tr]) * 100)) def daily_series(trades, span, fee_rt=FEE_RT): 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 trades: d = pd.Timestamp(t["exit_ts"], unit="ms", tz="UTC").normalize() if d in s.index: s[d] += t["gR"] * t["risk"] - fee_rt return s def sh_dd(s): 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 # =========================================================================== # MAIN # =========================================================================== def main(): t0 = time.time() print("=" * 100) print("r0707 CRT TOP-DOWN (video-claim, 74% WR) — H1 setup -> M15/M5 entry -> parziale+BE+runner") print(f"fee {FEE_RT*1e4:.0f}bps RT | hold-out {HOLDOUT.date()} | M1 NON nel feed (proxy 5m/15m)") print("TESI: il 74% WR e' artefatto del parziale+BE; il numero invariante e' l'expectancy in R.") print("=" * 100) # dati H1 span for a in ASSETS: d = al.get(a, HTF) idx = pd.to_datetime(d["timestamp"], unit="ms", utc=True) print(f" {a} {HTF}: {idx.iloc[0].date()} -> {idx.iloc[-1].date()} ({len(d)} barre)") all_trial_sh = [] grid_rows = [] # cache pattern per (asset, k, s) e low arrays per (asset, ltf) lowcache = {} for tf_lo in LTFS: for a in ASSETS: lowcache[(a, tf_lo)] = Low(al.get(a, tf_lo), tf_lo) print(f"\n{'#'*100}\n### GRIGLIA — schema d'uscita x (k, s, d, tf_lo, dir) [selezione cella SOLO in-sample]\n{'#'*100}") hdr = (f" {'tf':>3s} {'dir':>5s} k={'':1s}{'':1s} s d sch | " f"{'n':>5s} {'WR%':>5s} {'expR':>6s} {'net_bps':>8s} {'ShF':>6s} {'DDF%':>5s} | " f"{'nH':>4s} {'WRH':>5s} {'expRH':>6s}") print(hdr) detail = {} # per il candidato finale for tf_lo in LTFS: for k in K_GRID: for s in S_GRID: pats_by_asset = {a: detect_patterns(al.get(a, HTF), k, s) for a in ASSETS} for d_mult in D_GRID: which = "fade" # metodo del video; fade-vs-follow gia' settlato in r0702_crt_context for scheme in ("fix1.5", "fix2.0", "managed"): per_asset_trades = {} spans = {} for a in ASSETS: L = lowcache[(a, tf_lo)] spans[a] = (int(L.ts[0]), int(L.tsclose[-1])) per_asset_trades[a] = build_trades(pats_by_asset[a], L, d_mult, scheme, None) pooled = per_asset_trades["BTC"] + per_asset_trades["ETH"] pooled.sort(key=lambda t: t["entry_ts"]) st_f = stats(pooled) st_h = stats(pooled, mask=lambda t: t["entry_ts"] >= HOLDOUT_MS) st_i = stats(pooled, mask=lambda t: t["entry_ts"] < HOLDOUT_MS) # book daily 50/50 sequenziale per = [] for a in ASSETS: per.append(daily_series(seq_filter(per_asset_trades[a]), spans[a])) J = pd.concat(per, axis=1).fillna(0.0) book = 0.5 * J.iloc[:, 0] + 0.5 * J.iloc[:, 1] shf, ddf = sh_dd(book) bh = book[book.index >= HOLDOUT] shh = al._sh(bh) if len(bh) > 30 else np.nan bi = book[book.index < HOLDOUT] shi = al._sh(bi) if len(bi) > 30 else np.nan all_trial_sh.append(shf) grid_rows.append(dict(tf=tf_lo, k=k, s=s, d=d_mult, which=which, scheme=scheme, n=st_f["n"], wr=st_f["wr"], expR=st_f["exp_R"], shf=shf, shi=shi, shh=shh, wrh=st_h["wr"], expRH=st_h["exp_R"])) detail[(tf_lo, k, s, d_mult, which, scheme)] = dict( book=book, spans=spans, per_asset=per_asset_trades, st_f=st_f, st_h=st_h, st_i=st_i, shf=shf, shh=shh, ddf=ddf) print(f" {tf_lo:>3s} {which:>5s} k={k:.1f} {s:.2f} {d_mult:.2f} " f"{scheme:<8s} | {st_f['n']:>5d} {st_f['wr']:>5.1f} " f"{st_f['exp_R']:>+6.2f} {st_f['exp_net_bps']:>+8.1f} " f"{shf:>6.2f} {ddf*100:>5.1f} | {st_h['n']:>4d} {st_h['wr']:>5.1f} " f"{st_h['exp_R']:>+6.2f}") G = pd.DataFrame(grid_rows) # ---------- 1) IL "74% WR": WR per schema (solo FADE, il metodo del video) ---------- print(f"\n{'='*100}\n### 1) IL CLAIM 74% WR — win-rate per schema d'uscita (FADE = metodo del video)") print("Random-walk senza edge: P(tocca +1.5R prima di -1R) = 1/(1+1.5) = 40.0%") print("=" * 100) fade = G[G.which == "fade"] for scheme in ("fix1.5", "fix2.0", "managed"): sub = fade[fade.scheme == scheme] print(f" {scheme:<8s} | WR range [{sub.wr.min():.1f}, {sub.wr.max():.1f}]% " f"mediana {sub.wr.median():.1f}% | expectancy R range " f"[{sub.expR.min():+.2f}, {sub.expR.max():+.2f}] mediana {sub.expR.median():+.2f}") print(" -> a rr1=1.5 il WR reale e' ~35% (SOTTO il null 40%): il setup non tocca 1.5R nemmeno") print(" come un random walk. Per fabbricare un WR alto serve avvicinare il target (sez.5).") # ---------- 2) selezione cella SOLO in-sample fra le celle FADE ---------- print(f"\n{'='*100}\n### 2) SELEZIONE IN-SAMPLE (Sharpe book pre-2025) fra le celle FADE, poi hold-out") print("=" * 100) fade_elig = G[(G.which == "fade") & (G.n >= 30) & np.isfinite(G.shi)] if len(fade_elig): top = fade_elig.sort_values("shi", ascending=False).head(8) print(top[["tf", "k", "s", "d", "scheme", "n", "wr", "expR", "shi", "shf", "shh", "wrh", "expRH"]].to_string(index=False)) best = fade_elig.sort_values("shi", ascending=False).iloc[0] bkey = (best.tf, best.k, best.s, best.d, "fade", best.scheme) bd = detail[bkey] print(f"\n CELLA SCELTA (max Sh IN-SAMPLE): tf={best.tf} k={best.k} s={best.s} d={best.d} " f"scheme={best.scheme}") print(f" FULL: n={bd['st_f']['n']} WR={bd['st_f']['wr']:.1f}% expR={bd['st_f']['exp_R']:+.2f} " f"Sh={bd['shf']:+.2f} DD={bd['ddf']*100:.1f}%") print(f" IS : n={bd['st_i']['n']} WR={bd['st_i']['wr']:.1f}% expR={bd['st_i']['exp_R']:+.2f}") print(f" HOLD: n={bd['st_h']['n']} WR={bd['st_h']['wr']:.1f}% expR={bd['st_h']['exp_R']:+.2f} " f"Sh={bd['shh']:+.2f}") # ---------- 3) DSR su tutti i trial ---------- valid = [x for x in all_trial_sh if np.isfinite(x)] dsr, sr0 = al.deflated_sharpe(bd["shf"], valid, bd["book"].values) print(f"\n### 3) DEFLATED SHARPE (n_trials={len(valid)}): DSR={dsr:.3f} " f"(null max Sh atteso {sr0:.2f}) -> {'PASS' if dsr >= 0.95 else 'FAIL'} (soglia 0.95)") # ---------- 4) fee sweep sulla cella scelta ---------- print(f"\n### 4) FEE SWEEP (cella scelta, {best.scheme}) — expectancy R netto / Sharpe book") pats_by_asset = {a: detect_patterns(al.get(a, HTF), best.k, best.s) for a in ASSETS} for fr in FEE_SWEEP: per, pooled = [], [] for a in ASSETS: L = lowcache[(a, best.tf)] tr = build_trades(pats_by_asset[a], L, best.d, best.scheme, None, fee_rt=fr) pooled += tr per.append(daily_series(seq_filter(tr), (int(L.ts[0]), int(L.tsclose[-1])), fee_rt=fr)) J = pd.concat(per, axis=1).fillna(0.0) bk = 0.5 * J.iloc[:, 0] + 0.5 * J.iloc[:, 1] R_net = np.array([t["gR"] - fr / t["risk"] if t["risk"] > 0 else t["gR"] for t in pooled]) print(f" fee {fr*1e4:3.0f}bps: expR={R_net.mean():+.2f} ShFULL={al._sh(bk):+.2f}") else: print(" Nessuna cella FADE eleggibile (>=30 trade IS).") # ---------- 5) IL WIN-RATE E' UN KNOB: sweep del target parziale (managed) ---------- print(f"\n{'='*100}\n### 5) IL WIN-RATE E' UN KNOB — managed, sweep del target parziale rr1") print("Gambler's ruin senza edge: WR = P(+rr1 prima di -1R) = 1/(1+rr1). Avvicina il target -> WR sale.") print("Se WR sale verso ~74% MA l'expectancy R resta <=0, il 74% e' la scelta del target, non un edge.") print("=" * 100) print(f" {'rr1':>4s} | {'WR%':>6s} {'null 1/(1+rr1)':>14s} | {'expR(fee10)':>11s} " f"{'expR(fee0)':>10s} | {'n':>6s}") pats_by_asset = {a: detect_patterns(al.get(a, HTF), 1.5, 0.1) for a in ASSETS} for rr1 in (0.5, 0.75, 1.0, 1.5, 2.0): pooled = [] for a in ASSETS: L = lowcache[(a, "5m")] pooled += build_trades(pats_by_asset[a], L, 0.15, "managed", None, rr1=rr1) if not pooled: continue wr = np.mean([t["win"] for t in pooled]) * 100 R10 = np.mean([t["gR"] - FEE_RT / t["risk"] if t["risk"] > 0 else t["gR"] for t in pooled]) R0 = np.mean([t["gR"] for t in pooled]) print(f" {rr1:>4.2f} | {wr:>6.1f} {100/(1+rr1):>13.1f}% | {R10:>+11.2f} {R0:>+10.2f} | " f"{len(pooled):>6d}") print(" (fade-vs-follow direzionale gia' settlato: r0702_crt_context, FOLLOW batte FADE ogni anno)") # ---------- 6) executability $600 ---------- print(f"\n{'='*100}\n### 6) EXECUTABILITY $600 (stop sul TF basso = zona di protezione stretta)") print("=" * 100) for tf_lo in LTFS: risks = [] pats_by_asset = {a: detect_patterns(al.get(a, HTF), 1.2, 0.1) for a in ASSETS} for a in ASSETS: L = lowcache[(a, tf_lo)] for t in build_trades(pats_by_asset[a], L, 0.25, "fix1.5", None): risks.append(t["risk"] * 100) if risks: risks = np.array(risks) med = float(np.median(risks)) print(f" {tf_lo}: stopMed {med:.3f}% (p25 {np.percentile(risks,25):.3f}/" f"p75 {np.percentile(risks,75):.3f}) -> a CAP {LEV_CAP}x rischio/trade " f"{LEV_CAP*med:.2f}% (${CAPITAL*LEV_CAP*med/100:.2f}); nozionale ${CAPITAL*LEV_CAP:.0f} " f">min ${MIN_ORDER}. NB parziali/BE/runner = 3-4 ordini/trade, alcuni <${MIN_ORDER}.") print(f"\n[done in {time.time()-t0:.0f}s]") if __name__ == "__main__": main()