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

506 lines
24 KiB
Python
Raw Blame History

"""r0702_crt_context.py — CRT CON CONTESTO (2026-07-02).
FILONE: la scuola "Candle Range Theory" dice che lo sweep-and-reclaim vale SOLO su zone
importanti (liquidita' sopra massimi/minimi rilevanti, FVG, sessione giusta). Qui testiamo se
i FILTRI DI CONTESTO trasformano un fade (gia' morto in versione generica sul feed certificato)
in un edge. ONESTA' PRIMA DI TUTTO.
SETUP BASE FISSO (identico per tutte le celle, deciso A PRIORI prima di guardare i numeri):
* TF in {1h, 4h} (4h = resample leak-free da 1h via altlib.get).
* C2 = barra che SUPERA un livello di riferimento e CHIUDE dal lato opposto:
SHORT: high[i] > lvl_hi AND close[i] < lvl_hi (sweep del massimo + reclaim)
LONG : low[i] < lvl_lo AND close[i] > lvl_lo (sweep del minimo + reclaim)
(barra che sweppa ENTRAMBI i livelli e chiude in mezzo = ambigua -> scartata, dichiarato)
* Entry a close[i] (decisione con dati <= close[i], eseguibile).
* Stop DIETRO l'estremo di C2: estremo +/- 0.10*ATR14 (causale).
* Target = R FISSO 1.5:1 (scelto a priori; NON centro-range: il centro di un Donchian in
trend e' asimmetrico/ambiguo, R fisso e' uniforme su tutti i level-type).
* max_hold 20 barre; fill conservativi (SL prioritario se TP e SL nella stessa barra,
identico all'harness src/backtest/harness.backtest_signals); fee 0.10% RT.
LIVELLI (tutti causali, shift(1) su aggregati di periodi COMPLETI precedenti):
prevday = high/low del giorno UTC precedente (barre open-labeled, groupby giorno -> shift)
don20/55 = max/min delle N barre STRETTAMENTE precedenti (al.donchian, shift(1) built-in)
prevweek = high/low della settimana ISO precedente (lunedi' 00 UTC)
⚠️ OVERLAP DICHIARATO col lead PREVDAY-BREAKOUT in forward-monitor
(src/strategies/prevday_breakout.py + scripts/live/paper_prevday.py): STESSI livelli prior-day,
condizionamento OPPOSTO — il lead SEGUE il break decisivo (close > lvl + 0.30*range), qui si
FADEA il reclaim (close torna dentro). Se entrambi avessero edge sugli stessi livelli, uno dei
due e' rumore -> confronto esplicito fade-vs-follow (corr daily + chi vince dove) in fondo.
FILTRI DI CONTESTO (la parte "discrezionale" della CRT, meccanizzata):
EQ = equal highs/lows: il livello e' stato toccato >=2 volte entro 0.10*ATR14 nelle ultime
N barre (N = lookback del livello stesso; prevday=2 giorni, prevweek=7 giorni).
FVG = esiste un fair-value-gap a 3 candele NON ancora riempito nelle ultime 20 barre, nella
direzione del trade (short: FVG bullish sotto il prezzo non riempito = magnete giu';
long: FVG bearish sopra non riempito). Meccanizzazione SEMPLICE di un concetto fuzzy
discrezionale — limiti dichiarati: k=20 fisso, zona "non riempita" = mai traversata
interamente, nessuna nozione di "displacement" o "premium/discount".
SES = sessione dello sweep (ora UTC di apertura barra): Asia 00-08 / Europa 08-14 / US 14-22.
Solo a 1h (a 4h la label di sessione e' troppo grossolana). ⚠️ OGNI cella sessione
passa un anchor-shift +/-2/4h (analogo di al.day_boundary_robust a livello trade)
prima di essere creduta.
GATES: selezione SOLO in-sample pre-2025 (HOLDOUT altlib = 2025-01-01); hold-out riportato mai
usato per scegliere; al.deflated_sharpe su TUTTI i 22 trial; fee sweep 0.00-0.20% RT; se il
best-IS regge (Sharpe >= 0.5) -> al.marginal_vs_tp01. Causalita': livelli ricalcolati su
prefisso troncato e confrontati (check esplicito in fondo). Niente .view("int64"), niente
ffill mixed-TF.
Uso: uv run python scripts/research/r0702_crt_context.py
"""
from __future__ import annotations
import sys
from pathlib import Path
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
ROOT = Path("/opt/docker/PythagorasGoal")
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
from src.backtest.harness import backtest_signals # noqa: E402
from src.strategies.prevday_breakout import target as prevday_follow_target # noqa: E402
HOLDOUT = al.HOLDOUT
FEE_RT = 0.001 # 0.10% round-trip (Deribit taker)
MAX_HOLD = 20 # barre
R_MULT = 1.5 # target R fisso 1.5:1 (a priori)
SL_ATR_BUF = 0.10 # stop = estremo C2 +/- 0.10*ATR14 (a priori)
EQ_TOL_ATR = 0.10 # tolleranza equal highs/lows
EQ_MIN_TOUCH = 2
FVG_K = 20 # lookback barre per FVG non riempito
ASSETS = ("BTC", "ETH")
SESSIONS = {"asia": (0, 8), "eu": (8, 14), "us": (14, 22)}
LEVELS = ("prevday", "don20", "don55", "prevweek")
# ===========================================================================
# LIVELLI (causali)
# ===========================================================================
def prior_day_levels(df: pd.DataFrame, shift_h: int = 0):
"""High/low del giorno UTC PRECEDENTE (shift(1) sul groupby giorno -> strettamente
prima di oggi). shift_h sposta il confine del giorno (per l'anchor-shift test)."""
dt = pd.to_datetime(df["datetime"], utc=True) + pd.Timedelta(hours=shift_h)
day = dt.dt.floor("1D")
g = pd.DataFrame({"day": day.values,
"high": df["high"].values.astype(float),
"low": df["low"].values.astype(float)})
per = g.groupby("day").agg(dh=("high", "max"), dl=("low", "min"))
m = pd.DataFrame({"dh": per["dh"].shift(1), "dl": per["dl"].shift(1)}).reindex(g["day"].values)
return m["dh"].values, m["dl"].values
def prior_week_levels(df: pd.DataFrame):
"""High/low della settimana ISO PRECEDENTE (lunedi' 00 UTC, shift(1))."""
dt = pd.to_datetime(df["datetime"], utc=True)
day = dt.dt.floor("1D")
week = (day - pd.to_timedelta(dt.dt.dayofweek, unit="D"))
g = pd.DataFrame({"wk": week.values,
"high": df["high"].values.astype(float),
"low": df["low"].values.astype(float)})
per = g.groupby("wk").agg(wh=("high", "max"), wl=("low", "min"))
m = pd.DataFrame({"wh": per["wh"].shift(1), "wl": per["wl"].shift(1)}).reindex(g["wk"].values)
return m["wh"].values, m["wl"].values
def get_levels(df: pd.DataFrame, level: str, shift_h: int = 0):
if level == "prevday":
return prior_day_levels(df, shift_h)
if level == "prevweek":
return prior_week_levels(df)
if level == "don20":
return al.donchian(df, 20)
if level == "don55":
return al.donchian(df, 55)
raise ValueError(level)
def level_lookback_bars(level: str, bpd: int) -> int:
"""Lookback per il conteggio equal-touch = finestra del livello stesso."""
return {"prevday": 2 * bpd, "prevweek": 7 * bpd, "don20": 20, "don55": 55}[level]
# ===========================================================================
# EVENTI (sweep-and-reclaim) + outcome trade-level (overlap PERMESSO -> paired analysis)
# ===========================================================================
def _unfilled_fvg(h: np.ndarray, l: np.ndarray, i: int, d: int, price: float) -> bool:
"""SHORT (d=-1): esiste FVG BULLISH (low[j] > high[j-2]) nelle ultime FVG_K barre con zona
(high[j-2], low[j]) interamente SOTTO il prezzo e mai riempita (nessuna barra dopo j e'
scesa fino al bordo inferiore). LONG (d=+1): simmetrico con FVG bearish sopra."""
j0 = max(2, i - FVG_K)
for j in range(i - 1, j0 - 1, -1):
if d == -1 and l[j] > h[j - 2]:
zone_lo, zone_hi = h[j - 2], l[j]
if zone_hi < price and np.min(l[j + 1:i + 1]) > zone_lo:
return True
if d == 1 and h[j] < l[j - 2]:
zone_lo, zone_hi = h[j], l[j - 2]
if zone_lo > price and np.max(h[j + 1:i + 1]) < zone_hi:
return True
return False
def build_events(df: pd.DataFrame, level: str, shift_h: int = 0,
with_context: bool = True) -> pd.DataFrame:
"""Tabella eventi sweep-and-reclaim con outcome trade-level (entry close[i], exit da i+1,
SL prioritario, fee 0.10% RT) + feature di contesto (eq/fvg/session). Overlap permesso:
ogni evento valutato indipendentemente -> confronto PAIRED filtro-vs-tutti pulito."""
h = df["high"].values.astype(float)
l = df["low"].values.astype(float)
c = df["close"].values.astype(float)
n = len(c)
lvl_hi, lvl_lo = get_levels(df, level, shift_h)
a14 = al.atr(df, 14)
dt = pd.to_datetime(df["datetime"], utc=True)
hours = dt.dt.hour.values
bpd = al.bars_per_day(df)
lb = level_lookback_bars(level, bpd)
sw_hi = np.isfinite(lvl_hi) & (h > lvl_hi) & (c < lvl_hi)
sw_lo = np.isfinite(lvl_lo) & (l < lvl_lo) & (c > lvl_lo)
both = sw_hi & sw_lo
sw_hi &= ~both
sw_lo &= ~both
rows = []
for i in np.where(sw_hi | sw_lo)[0]:
if i >= n - 1 or i < 60:
continue
d = -1 if sw_hi[i] else 1
entry = c[i]
atr_i = a14[i]
if not np.isfinite(atr_i) or atr_i <= 0:
continue
if d == -1:
sl = h[i] + SL_ATR_BUF * atr_i
risk = sl - entry
tp = entry - R_MULT * risk
else:
sl = l[i] - SL_ATR_BUF * atr_i
risk = entry - sl
tp = entry + R_MULT * risk
if risk <= 0 or tp <= 0:
continue
jend = min(i + MAX_HOLD, n - 1)
exit_price = c[jend]
for j in range(i + 1, jend + 1):
if d == 1:
if l[j] <= sl:
exit_price = sl
break
if h[j] >= tp:
exit_price = tp
break
else:
if h[j] >= sl:
exit_price = sl
break
if l[j] <= tp:
exit_price = tp
break
exit_price = c[j]
gross = d * (exit_price - entry) / entry
L = lvl_hi[i] if d == -1 else lvl_lo[i]
row = dict(i=int(i), dir=int(d), entry=entry, sl=sl, tp=tp, level_px=float(L),
atr=float(atr_i), gross=gross, net=gross - FEE_RT)
if with_context:
j0 = max(0, i - lb)
tol = EQ_TOL_ATR * atr_i
touches = int(np.sum(np.abs((h if d == -1 else l)[j0:i] - L) <= tol))
row["eq"] = touches >= EQ_MIN_TOUCH
row["fvg"] = _unfilled_fvg(h, l, int(i), int(d), float(entry))
hr = int(hours[i])
row["ses"] = next((s for s, (a, b) in SESSIONS.items() if a <= hr < b), "none")
rows.append(row)
ev = pd.DataFrame(rows)
if len(ev):
ev["dt"] = dt.values[ev["i"].values]
ev["hold"] = pd.to_datetime(ev["dt"], utc=True) >= HOLDOUT
ev["year"] = pd.to_datetime(ev["dt"], utc=True).dt.year
return ev
# ===========================================================================
# STRATEGIA (non-overlap, harness ufficiale) — metriche daily-step
# ===========================================================================
def entries_from(df: pd.DataFrame, sub: pd.DataFrame) -> list:
ent: list = [None] * len(df)
for row in sub.itertuples():
ent[row.i] = dict(dir=int(row.dir), tp=float(row.tp), sl=float(row.sl),
max_bars=MAX_HOLD)
return ent
def strat_eval(df: pd.DataFrame, entries: list, fee_rt: float = FEE_RT) -> dict:
m = backtest_signals(df, entries, fee_rt=fee_rt, leverage=1.0)
idx = pd.DatetimeIndex(pd.to_datetime(df["datetime"], utc=True))
eq = pd.Series(m.equity, index=idx)
d = eq.resample("1D").last().dropna().pct_change().dropna()
di, dh = d[d.index < HOLDOUT], d[d.index >= HOLDOUT]
return dict(n_trades=int(m.n_trades), wr=round(m.win_rate, 1), dd=round(m.max_dd, 4),
sh_full=round(al._sh(d), 3), sh_is=round(al._sh(di), 3),
sh_hold=round(al._sh(dh), 3), daily=d)
def apply_filter(ev: pd.DataFrame, filt: str | None) -> pd.DataFrame:
if not len(ev) or filt is None:
return ev
if filt == "eq":
return ev[ev["eq"]]
if filt == "fvg":
return ev[ev["fvg"]]
if filt.startswith("ses_"):
return ev[ev["ses"] == filt[4:]]
raise ValueError(filt)
def eval_trial(events: dict, tf: str, level: str, filt: str | None) -> dict:
per_asset, dailies = {}, {}
for a in ASSETS:
df = al.get(a, tf)
ev = events[(a, tf, level)]
sub = apply_filter(ev, filt)
r = strat_eval(df, entries_from(df, sub))
yrs = (pd.to_datetime(df["datetime"].iloc[-1], utc=True)
- pd.to_datetime(df["datetime"].iloc[0], utc=True)).days / 365.25
per_asset[a] = dict(r, n_ev=len(sub), ev_per_yr=round(len(sub) / yrs, 1),
exp_is=_exp(sub, False), exp_hold=_exp(sub, True))
dailies[a] = r["daily"]
J = pd.concat(dailies, axis=1, join="inner").fillna(0.0)
comb = J.mean(axis=1)
ci, ch = comb[comb.index < HOLDOUT], comb[comb.index >= HOLDOUT]
return dict(tf=tf, level=level, filt=filt or "-", per_asset=per_asset, comb_daily=comb,
sh_is=round(al._sh(ci), 3), sh_hold=round(al._sh(ch), 3),
sh_full=round(al._sh(comb), 3),
min_sh_is=round(min(per_asset[a]["sh_is"] for a in ASSETS), 3),
min_sh_hold=round(min(per_asset[a]["sh_hold"] for a in ASSETS), 3))
def _exp(sub: pd.DataFrame, hold: bool):
"""Expectancy netta per trade (%) sullo slice IS/HOLD."""
if not len(sub):
return None
s = sub[sub["hold"] == hold]["net"]
return round(float(s.mean()) * 100, 3) if len(s) else None
# ===========================================================================
# MAIN
# ===========================================================================
def main():
print("=" * 100)
print("r0702 CRT CON CONTESTO — sweep-and-reclaim su livelli + filtri (EQ/FVG/sessione)")
print(f"setup fisso: entry close C2, SL estremo+/-{SL_ATR_BUF}*ATR14, TP R={R_MULT}:1, "
f"max_hold {MAX_HOLD} barre, fee {FEE_RT*100:.2f}% RT, SL prioritario same-bar")
print("=" * 100)
# ---------- 0. CAUSALITY CHECK sui livelli (prefisso troncato) ----------
print("\n[0] CAUSALITY CHECK livelli (ricalcolo su prefisso troncato, tail 200 barre)")
for level in LEVELS:
worst = 0.0
for a in ASSETS:
df = al.get(a, "1h")
cut = int(len(df) * 0.9)
hi_f, lo_f = get_levels(df, level)
sub = df.iloc[:cut].reset_index(drop=True)
hi_s, lo_s = get_levels(sub, level)
for full, part in ((hi_f, hi_s), (lo_f, lo_s)):
x, y = np.nan_to_num(full[cut - 200:cut]), np.nan_to_num(part[cut - 200:cut])
worst = max(worst, float(np.max(np.abs(x - y))))
print(f" {level:<9s} max_tail_diff={worst:.10f} {'OK' if worst < 1e-9 else 'FAIL'}")
# ---------- 1. EVENTI (cache) ----------
events = {}
for a in ASSETS:
for tf in ("1h", "4h"):
df = al.get(a, tf)
for level in LEVELS:
events[(a, tf, level)] = build_events(df, level)
# ---------- 2. TRIALS (22, definiti a priori) ----------
trials = []
for tf in ("1h", "4h"):
for level in LEVELS:
trials.append((tf, level, None))
for tf in ("1h", "4h"):
for level in ("don20", "prevday"):
trials.append((tf, level, "eq"))
trials.append((tf, level, "fvg"))
for level in ("don20", "prevday"):
for ses in SESSIONS:
trials.append(("1h", level, f"ses_{ses}"))
assert len(trials) == 22
results = {}
for tf, level, filt in trials:
results[(tf, level, filt or "-")] = eval_trial(events, tf, level, filt)
# ---------- 3. BASELINE INCONDIZIONATA (don20, nessun filtro) ----------
print("\n[1] BASELINE INCONDIZIONATA — sweep-and-reclaim Donchian20, nessun contesto")
for tf in ("1h", "4h"):
r = results[(tf, "don20", "-")]
print(f" TF {tf}: comb Sharpe IS={r['sh_is']:+.2f} HOLD={r['sh_hold']:+.2f} "
f"FULL={r['sh_full']:+.2f}")
for a in ASSETS:
p = r["per_asset"][a]
print(f" {a}: ev/yr={p['ev_per_yr']:>6.1f} trades(no-overlap)={p['n_trades']:>5d} "
f"wr={p['wr']:>4.1f}% expIS={p['exp_is']}% expHOLD={p['exp_hold']}% "
f"Sh IS={p['sh_is']:+.2f} HOLD={p['sh_hold']:+.2f} DD={p['dd']*100:.1f}%")
# ---------- 4. TUTTE LE CELLE (tabella) ----------
print("\n[2] TUTTE LE 22 CELLE (comb 50/50, daily-step Sharpe; exp = %/trade netto)")
print(f" {'tf':<3s} {'level':<9s} {'filt':<9s} {'ShIS':>6s} {'ShHOLD':>7s} {'ShFULL':>7s} "
f"{'minShIS':>8s} {'minShHOLD':>9s} {'BTCexpIS':>9s} {'ETHexpIS':>9s} "
f"{'BTCexpH':>8s} {'ETHexpH':>8s} {'nBTC':>5s} {'nETH':>5s}")
for (tf, level, filt), r in sorted(results.items(), key=lambda kv: -kv[1]["sh_is"]):
pb, pe = r["per_asset"]["BTC"], r["per_asset"]["ETH"]
print(f" {tf:<3s} {level:<9s} {filt:<9s} {r['sh_is']:>+6.2f} {r['sh_hold']:>+7.2f} "
f"{r['sh_full']:>+7.2f} {r['min_sh_is']:>+8.2f} {r['min_sh_hold']:>+9.2f} "
f"{str(pb['exp_is']):>9s} {str(pe['exp_is']):>9s} "
f"{str(pb['exp_hold']):>8s} {str(pe['exp_hold']):>8s} "
f"{pb['n_ev']:>5d} {pe['n_ev']:>5d}")
# ---------- 5. UPLIFT PAIRED dei filtri (stessi eventi, sottoinsieme vs tutti) ----------
print("\n[3] UPLIFT PAIRED per filtro (expectancy %/trade: filtrato - tutti; stessi eventi)")
print(f" {'base':<16s} {'filtro':<9s} {'asset':<4s} {'slice':<5s} {'n_all':>6s} {'n_f':>5s} "
f"{'exp_all':>8s} {'exp_f':>8s} {'uplift':>8s}")
filt_names = ["eq", "fvg"] + [f"ses_{s}" for s in SESSIONS]
uplift_summary = {}
for tf in ("1h", "4h"):
for level in ("don20", "prevday"):
for filt in filt_names:
if filt.startswith("ses_") and tf != "1h":
continue
key = (tf, level, filt)
for a in ASSETS:
ev = events[(a, tf, level)]
sub = apply_filter(ev, filt)
for hold, lab in ((False, "IS"), (True, "HOLD")):
ea, ef = _exp(ev, hold), _exp(sub, hold)
na = int((ev["hold"] == hold).sum()) if len(ev) else 0
nf = int((sub["hold"] == hold).sum()) if len(sub) else 0
up = round(ef - ea, 3) if (ea is not None and ef is not None) else None
uplift_summary.setdefault(key, []).append((a, lab, up))
print(f" {tf+'/'+level:<16s} {filt:<9s} {a:<4s} {lab:<5s} {na:>6d} "
f"{nf:>5d} {str(ea):>8s} {str(ef):>8s} {str(up):>8s}")
print("\n Consistenza uplift per filtro (positivo su TUTTE le 4 slice asset x IS/HOLD?):")
for key, ups in uplift_summary.items():
vals = [u for (_, _, u) in ups if u is not None]
n_pos = sum(1 for u in vals if u > 0)
print(f" {key[0]}/{key[1]}+{key[2]:<9s}: {n_pos}/{len(vals)} slice positive "
f"{'<-- consistente' if vals and n_pos == len(vals) else ''}")
# ---------- 6. SELEZIONE IN-SAMPLE + DSR ----------
all_sr = [r["sh_full"] for r in results.values()]
chosen_key = max(results, key=lambda k: results[k]["sh_is"])
ch = results[chosen_key]
dsr, sr0 = al.deflated_sharpe(ch["sh_full"], all_sr, ch["comb_daily"])
print(f"\n[4] SELEZIONE IN-SAMPLE-ONLY (pre-2025) su {len(trials)} trial")
print(f" best-IS: {chosen_key} ShIS={ch['sh_is']:+.2f} ShHOLD={ch['sh_hold']:+.2f} "
f"ShFULL={ch['sh_full']:+.2f}")
print(f" deflated Sharpe (n_trials={len(all_sr)}): DSR={dsr:.3f} "
f"(PASS>=0.95) expected-null-max Sharpe={sr0:.2f}")
# fee sweep sul best-IS e sulla baseline don20/1h
print("\n[5] FEE SWEEP (Sharpe FULL comb per fee RT)")
for key in {chosen_key, ("1h", "don20", "-")}:
tf, level, filt = key
row = []
for fee in (0.0, 0.0005, 0.001, 0.0015, 0.002):
dailies = {}
for a in ASSETS:
df = al.get(a, tf)
sub = apply_filter(events[(a, tf, level)], None if filt == "-" else filt)
# ricalcola net eventi con fee diversa e' lineare; per la strategia rifacciamo il bt
dailies[a] = strat_eval(df, entries_from(df, sub), fee_rt=fee)["daily"]
J = pd.concat(dailies, axis=1, join="inner").fillna(0.0)
row.append(f"{fee*100:.2f}%RT:{al._sh(J.mean(axis=1)):+.2f}")
print(f" {key}: " + " ".join(row))
# ---------- 7. ANCHOR-SHIFT sulle celle sessione (+/-2/4h) ----------
print("\n[6] ANCHOR-SHIFT celle sessione (label ora spostata; uplift expectancy IS per shift)")
for level in ("don20", "prevday"):
for ses in SESSIONS:
per_shift = {}
for sh in (-4, -2, 0, 2, 4):
ups = []
for a in ASSETS:
ev = events[(a, "1h", level)]
if not len(ev):
continue
hrs = (pd.to_datetime(ev["dt"], utc=True).dt.hour + sh) % 24
a_, b_ = SESSIONS[ses]
sub = ev[(hrs >= a_) & (hrs < b_)]
ea, ef = _exp(ev, False), _exp(sub, False)
if ea is not None and ef is not None:
ups.append(ef - ea)
per_shift[sh] = round(float(np.mean(ups)), 3) if ups else None
vals = [v for v in per_shift.values() if v is not None]
flip = vals and (max(vals) > 0 > min(vals)) and (max(vals) - min(vals)) > 0.05
verd = "ARTIFACT-RISK(flip)" if flip else \
("stabile-pos" if vals and min(vals) > 0 else
"stabile-neg/nullo" if vals and max(vals) <= 0 else "misto-debole")
print(f" {level}+{ses:<5s}: " +
" ".join(f"{k:+d}h:{v}" for k, v in per_shift.items()) + f" -> {verd}")
# ---------- 8. DAY-BOUNDARY SHIFT sul fade prevday base (1h) ----------
print("\n[7] DAY-BOUNDARY SHIFT su fade prevday base 1h (livelli ricostruiti col giorno spostato)")
for sh in (0, 2, 4, 8, 12):
dailies = {}
for a in ASSETS:
df = al.get(a, "1h")
ev = build_events(df, "prevday", shift_h=sh, with_context=False)
dailies[a] = strat_eval(df, entries_from(df, ev))["daily"]
J = pd.concat(dailies, axis=1, join="inner").fillna(0.0)
comb = J.mean(axis=1)
ci = comb[comb.index < HOLDOUT]
print(f" shift +{sh:>2d}h: Sh IS={al._sh(ci):+.2f} FULL={al._sh(comb):+.2f}")
# ---------- 9. FADE vs FOLLOW sui livelli prior-day (lead esistente) ----------
print("\n[8] FADE (questo filone, prevday base 1h) vs FOLLOW (lead prevday_breakout congelato)")
fol = {}
for a in ASSETS:
df = al.get(a, "1h")
evw = al.eval_weights(df, prevday_follow_target(df))
fol[a] = pd.Series(evw["net"], index=evw["idx"])
Jf = pd.concat(fol, axis=1, join="inner").fillna(0.0)
follow_d = al._to_daily(0.5 * Jf["BTC"] + 0.5 * Jf["ETH"])
fade_d = results[("1h", "prevday", "-")]["comb_daily"]
JJ = pd.concat({"fade": fade_d, "follow": follow_d}, axis=1, join="inner").dropna()
JH = JJ[JJ.index >= HOLDOUT]
JI = JJ[JJ.index < HOLDOUT]
print(f" corr daily fade-follow: FULL={JJ['fade'].corr(JJ['follow']):+.3f} "
f"HOLD={JH['fade'].corr(JH['follow']):+.3f}")
print(f" Sharpe IS : fade={al._sh(JI['fade']):+.2f} follow={al._sh(JI['follow']):+.2f}")
print(f" Sharpe HOLD: fade={al._sh(JH['fade']):+.2f} follow={al._sh(JH['follow']):+.2f}")
print(" per anno (Sharpe fade | follow):")
for y in sorted(set(JJ.index.year)):
sub = JJ[JJ.index.year == y]
if len(sub) > 40:
print(f" {y}: {al._sh(sub['fade']):+.2f} | {al._sh(sub['follow']):+.2f}")
# ---------- 10. MARGINAL vs TP01 (solo se il best-IS regge) ----------
if ch["sh_full"] >= 0.5 and ch["sh_is"] >= 0.5:
print("\n[9] MARGINAL vs TP01 (best-IS regge >=0.5 -> gate)")
m = al.marginal_vs_tp01(ch["comb_daily"])
print(f" verdict={m.get('marginal_verdict')} corr_full={m.get('corr_full')} "
f"uplift w25 full={m['blends']['w25']['uplift_full']:+.3f} "
f"hold={m['blends']['w25']['uplift_hold']}")
print(f" has_insample_edge={m.get('has_insample_edge')} is_hedge={m.get('is_hedge')} "
f"robust_oos={m.get('robust_oos')} multicut={m.get('multicut_uplift')}")
else:
print(f"\n[9] MARGINAL vs TP01: SALTATO — best-IS Sharpe FULL={ch['sh_full']:+.2f} / "
f"IS={ch['sh_is']:+.2f} sotto la soglia 0.5 standalone")
print("\nFine. Nessun file scritto fuori da questo script; selezione solo in-sample.")
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink