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>
This commit is contained in:
Adriano Dal Pastro
@@ -0,0 +1,507 @@
|
||||
"""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
Block a user