From 822aa1307e02d7261c4185455295b4080f12c8bb Mon Sep 17 00:00:00 2001 From: Adriano Dal Pastro Date: Wed, 8 Jul 2026 06:02:38 +0000 Subject: [PATCH] =?UTF-8?q?research(crt):=20video-claim=20"CRT=20top-down?= =?UTF-8?q?=2074%=20win=20rate"=20SCARTATO=20=E2=80=94=20il=2074%=20e'=20u?= =?UTF-8?q?n=20knob,=20non=20un=20edge?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Meccanizzato e falsificato il claim ICT/SMC top-down (H1 setup CRT -> M15/M5 conferma -> entry, uscita parziale+BE+runner) su BTC/ETH certificati. - Setup H1 CRT gia' triplo-refutato (onda 2026-07-02); qui testato l'angolo nuovo: la gestione d'uscita e il "74% WR". - WR reale 30-37% a RR 1.5-2, SOTTO il null gambler's-ruin 40% -> il ritest tocca il target meno di una moneta (conferma "il ritest e' informazione negativa"). - Il 74% si fabbrica avvicinando il target: sweep rr1 0.5->2.0 = WR sale, expectancy R INVARIANTE e negativa (-1.2..-3.3R netto). DSR 0.000; a fee 0 ancora negativa -> non morte-per-fee, l'edge lordo non c'e'. - Non eseguibile pulito a $600 (3-4 ordini/trade, alcuni sub-min-order). - Regola candidata: convertire ogni claim "WR X%" da schema parziale+BE in expectancy R netto fee (WR ~= 1/(1+rr1), non un merito). Nessuna modifica a src/config/live/tests. Book e pesi INVARIATI. Diario: docs/diary/2026-07-07-crt-topdown-74winrate.md Co-Authored-By: Claude Opus 4.8 (1M context) --- CLAUDE.md | 15 + .../diary/2026-07-07-crt-topdown-74winrate.md | 109 ++++ scripts/research/r0707_crt_topdown.py | 506 ++++++++++++++++++ 3 files changed, 630 insertions(+) create mode 100644 docs/diary/2026-07-07-crt-topdown-74winrate.md create mode 100644 scripts/research/r0707_crt_topdown.py diff --git a/CLAUDE.md b/CLAUDE.md index 652e020..a3c28f4 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -292,6 +292,21 @@ Prima ondata di ricerca onesta su BTC/ETH certificati (5 track, harness condivis ⚠️ Lezione pandas: `resample("7D", origin=...)` IGNORA origin (pandas 2.x, solo RuntimeWarning) → bande d'ancora weekly finte; usare `"168h"`. Diario `2026-07-02-elliott-albimarini-capital.md`; script `scripts/research/r0702_{ell_*,alb_*,capital_scaling}.py` (6 file). +- **Video-claim "CRT top-down multi-TF, 74% win rate" — SCARTATO (2026-07-07): il 74% è un KNOB, non + un edge.** `scripts/research/r0707_crt_topdown.py`. Metodo ICT/SMC top-down (H1 setup CRT → M15 + struttura/FVG → M5 conferma → M1 entry) con uscita parziale-70/80%-a-1.5R + break-even + runner verso + prev-daily high/low. Il setup H1 CRT è **già triplo-refutato** (onda 2026-07-02: base DSR 0.000, MTF + expectancy neg. ovunque "il ritest è informazione negativa", contesto/FVG peggiora, fade 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()