Adriano/PythagorasGoal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
PythagorasGoal/scripts/research/r0702_crt_mtf.py
T
Adriano Dal Pastro 73d74c5e53 research(wave-0702): ondata timing + CRT — 8 filoni, 0 nuovi sleeve, finding anchor timing-luck TP01 
Goal: "altre strategie su Deribit con timing differenti". 8 filoni multi-agente + scettico:
- event-clock bars, expiry calendar Deribit, clock lenti/bande, regime-speed: SCARTATI
- CRT (Candle Range Theory) base/multi-TF/contesto: SCARTATA 3/3 (DSR~0, ritest =
  informazione negativa; sottoprodotto: FOLLOW>FADE sui livelli prior-day ogni anno,
  conferma il lead prevday)
- FINDING (confermato da scettico indipendente): hold-out 0.31 di TP01 = migliore delle
  24 ancore orarie (mediana 0.04, banda [-0.13,+0.30]) -> narrativa corretta in CLAUDE.md
  e docstring: l'hold-out non risolve l'edge di ritorno, regge il taglio DD a ogni ancora.
  Tranching K=2/4 = solo varianza della stima, no deploy a $600. Audit d'ancora pendente
  su XS01/SKH01. Book live e portafoglio INVARIATI. Test 168/168.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-02 22:12:22 +00:00

580 lines
28 KiB
Python
Raw Permalink Blame History

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