Adriano/PythagorasGoal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
PythagorasGoal/scripts/research/r0702_crt_base.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

508 lines
24 KiB
Python
Raw Permalink Blame History

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