Adriano/PythagorasGoal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
PythagorasGoal/scripts/research/r0707_crt_topdown.py
T
Adriano Dal Pastro 822aa1307e research(crt): video-claim "CRT top-down 74% win rate" SCARTATO — il 74% e' un knob, non un edge 
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) <noreply@anthropic.com>
2026-07-08 06:02:38 +00:00

507 lines
25 KiB
Python
Raw Permalink Blame History

#!/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()
Reference in New Issue View Git Blame Copy Permalink