Adriano/PythagorasGoal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

research(crypto): 4 filoni 2026-06-29 — ERM lead sub-daily (forward), 3 scartati/deboli

Browse Source 
Ricerca onesta su BTC/ETH + universo HL, branch separato (nessun impatto live).
Harness condiviso altlib (causale, fee 0.10% RT, marginal vs TP01, day-boundary,
haircut $600). Test 19/19 verdi.

- A DVOL direzionale  -> LEAD hedge/DD-dampener, NON sleeve (buy-the-fear; is_hedge).
- B Intraday ERM 8h   -> LEAD forte / forward-monitor: earns_slot=True, ADDS oltre
                         SKH01 (TP01+SKH+ERM 60/25/15 FULL 1.88/HOLD 1.46/DD 8.9%).
                         Caveat: plateau hold-out single-row, multiple-testing non
                         deflazionato, exec 8h. Controllo TOD = FAIL atteso.
- C Cross-sectional non-mom (low-vol HL) -> DEBOLE/forward-monitor (deflated-Sh 0.13,
                         storia 2.5a, non eseguibile $600) STAT-MODE.
- D Macro regime-gate -> RIDONDANTE col trend (corr->TP01 0.989), SCARTATO.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Adriano Dal Pastro
2026-06-29 19:48:36 +00:00
parent a158d0e2ae
commit aad69f9790
12 changed files with 2283 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/research/macro_regime_gate.py
+458
View File
@@ -0,0 +1,458 @@
"""macro_regime_gate.py — Filone D: MACRO REGIME-GATE sul book crypto (eseguibile).
TESI
----
Usare segnali macro/cross-market — equity (SPY/QQQ/IWM), credito (HYG/LQD), oro (GLD/SLV),
tassi (TLT/IEF) — come GATE risk-on/risk-off applicato al book crypto (BTC/ETH) per migliorare
il TIMING del drawdown di TP01. Quando il regime macro e' risk-off (credito che cede, equity
sotto trend, fuga sui bond) -> riduci/azzera l'esposizione crypto; risk-on -> lascia agire TP01.
E' ESEGUIBILE perche' GATA solo BTC/ETH perp (non aggiunge gambe).
NON e' un lead-lag direzionale (gia' esplorato e morto: vedi 2026-06-22/-23 leadlag diaries).
L'angolo nuovo = un OVERLAY binario/continuo di DE-RISK sul book.
IL RISCHIO (da CLAUDE.md): il gate di de-risk rischia di essere RIDONDANTE col trend — TP01 e'
gia' long-flat e va a 0 nei crash (lezione DVOL-spike "ridondante col trend, Delta 0.00"). Questo
script DEVE dimostrare che il gate aggiunge OLTRE quel che TP01 fa da solo, altrimenti SCARTATO.
CAUSALITA' (fusi orari, regola di prim'ordine)
----------------------------------------------
- Barre equity: open-labeled a 00:00 del giorno di trading; il CLOSE e' ~20:00-21:00 UTC dello
STESSO giorno (NYSE 16:00 ET).
- Barre crypto 1d: open-labeled a 00:00; il CLOSE e' a 00:00 UTC del giorno DOPO. TP01 decide la
posizione a close[i] e la TIENE durante la barra i+1 (eval_weights shift-a per te).
- Quindi: gate[i] allineato (merge_asof backward, equity-label <= crypto-label day i) usa il
close equity del giorno i (noto ~20:00 day i) per la posizione tenuta durante la barra i+1
(giorno i+1). Margine causale >= 4h. Leak-free. Variante STRICT (equity-label < crypto-label)
come margine extra. Verifica con al.causality_ok + day_boundary_robust.
USO: uv run python scripts/research/macro_regime_gate.py
"""
from __future__ import annotations
import sys
from pathlib import Path
import numpy as np
import pandas as pd
_ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(_ROOT))
sys.path.insert(0, str(_ROOT / "scripts" / "research" / "alt"))
import altlib as al # noqa: E402
from src.strategies.trend_portfolio import CANONICAL, TrendPortfolio # noqa: E402
DATA = _ROOT / "data" / "raw"
HOLDOUT = al.HOLDOUT
ASSETS = ("BTC", "ETH")
# ===========================================================================
# MACRO FRAME — ETF daily, allineati causalmente sul calendario SPY (master).
# ===========================================================================
def _load_eq(sym: str) -> pd.DataFrame:
p = DATA / f"eq_{sym}_1d.parquet"
df = pd.read_parquet(p).sort_values("timestamp").reset_index(drop=True)
df["dt"] = pd.to_datetime(df["timestamp"], unit="ms", utc=True)
return df[["timestamp", "dt", "close"]].rename(columns={"close": sym})
def macro_frame(syms=("spy", "qqq", "iwm", "hyg", "lqd", "gld", "slv", "tlt", "ief")) -> pd.DataFrame:
"""Frame macro sul calendario NYSE (master = SPY). Ogni colonna e' il close equity,
allineato causale (merge_asof backward) -> nessun valore futuro per riga."""
base = _load_eq("spy")[["timestamp", "dt"]].copy()
out = base
for s in syms:
e = _load_eq(s)[["timestamp", s]]
out = pd.merge_asof(out, e, on="timestamp", direction="backward")
return out
# ===========================================================================
# GATE BUILDERS — ognuno ritorna (timestamp_ms, gate in [0,1]) sul calendario equity.
# Tutti CAUSALI: la riga i usa solo close <= riga i (rolling/SMA, niente expanding-future).
# gate=1 => risk-on (TP01 pieno); gate=g_off (0 o 0.5) => risk-off (de-risk).
# ===========================================================================
def _sma(x: np.ndarray, n: int) -> np.ndarray:
return pd.Series(x).rolling(n, min_periods=n).mean().values
def _ret(x: np.ndarray, n: int) -> np.ndarray:
r = np.full(len(x), np.nan)
r[n:] = x[n:] / x[:-n] - 1.0
return r
def gate_trend(mf: pd.DataFrame, col: str, n: int, g_off: float) -> pd.DataFrame:
"""Risk-on se col_close > SMA(col, n). Filtro di trend classico (SPY200, HYG, ...)."""
c = mf[col].values.astype(float)
on = c > _sma(c, n)
g = np.where(on, 1.0, g_off)
g[~np.isfinite(c)] = np.nan
g[np.isnan(_sma(c, n))] = np.nan
return pd.DataFrame({"timestamp": mf["timestamp"].values, "gate": g})
def gate_ratio(mf: pd.DataFrame, num: str, den: str, n: int, g_off: float) -> pd.DataFrame:
"""Risk-on se ratio num/den (proxy spread di credito) > la sua SMA(n).
HYG/LQD o HYG/IEF in calo = spread che si allarga = risk-off."""
ratio = (mf[num].values.astype(float) / mf[den].values.astype(float))
on = ratio > _sma(ratio, n)
g = np.where(on, 1.0, g_off)
g[~np.isfinite(ratio)] = np.nan
g[np.isnan(_sma(ratio, n))] = np.nan
return pd.DataFrame({"timestamp": mf["timestamp"].values, "gate": g})
def gate_combo(mf: pd.DataFrame, n: int, g_off: float, thr: float = 0.5,
continuous: bool = False) -> pd.DataFrame:
"""Score di regime = media di 3 voti risk-on: SPY>SMA, HYG>SMA, HYG/LQD>SMA.
continuous=True -> gate = g_off + (1-g_off)*score (de-risk graduale).
continuous=False -> gate = 1 se score>=thr else g_off (binario su maggioranza)."""
spy = mf["spy"].values.astype(float)
hyg = mf["hyg"].values.astype(float)
ratio = hyg / mf["lqd"].values.astype(float)
votes = np.vstack([spy > _sma(spy, n), hyg > _sma(hyg, n), ratio > _sma(ratio, n)]).astype(float)
warm = np.isnan(_sma(spy, n)) | np.isnan(_sma(hyg, n)) | np.isnan(_sma(ratio, n))
score = votes.mean(axis=0)
if continuous:
g = g_off + (1.0 - g_off) * score
else:
g = np.where(score >= thr, 1.0, g_off)
g[warm] = np.nan
return pd.DataFrame({"timestamp": mf["timestamp"].values, "gate": g})
def gate_flight(mf: pd.DataFrame, safe: str, risk: str, n: int, g_off: float) -> pd.DataFrame:
"""Fuga verso la sicurezza: risk-off quando il safe-asset (TLT/GLD) sale MENTRE il risk
(SPY) scende sull'orizzonte n. Divergenza risk-off classica (flight-to-quality)."""
s = mf[safe].values.astype(float)
rk = mf[risk].values.astype(float)
off = (_ret(s, n) > 0) & (_ret(rk, n) < 0)
g = np.where(off, g_off, 1.0)
warm = np.isnan(_ret(s, n)) | np.isnan(_ret(rk, n))
g[warm] = np.nan
return pd.DataFrame({"timestamp": mf["timestamp"].values, "gate": g})
def gate_eqvol(mf: pd.DataFrame, n: int, win: int, z: float, g_off: float) -> pd.DataFrame:
"""Regime di vol equity: de-risk quando la vol realizzata SPY (win g) e' alta vs la sua
storia espandente-causale (z-score > z). Proxy 'VIX spike' senza VIX."""
spy = mf["spy"].values.astype(float)
r = np.zeros(len(spy)); r[1:] = spy[1:] / spy[:-1] - 1.0
rv = pd.Series(r).rolling(win, min_periods=win).std().values
zsc = al.zscore(rv, n)
off = zsc > z
g = np.where(off, g_off, 1.0)
g[~np.isfinite(zsc)] = np.nan
return pd.DataFrame({"timestamp": mf["timestamp"].values, "gate": g})
# ===========================================================================
# ALIGN GATE -> CRYPTO (causale) + GATED TARGET
# ===========================================================================
def align_gate(gate_df: pd.DataFrame, crypto_df: pd.DataFrame, strict: bool = False) -> np.ndarray:
"""Allinea il gate (calendario equity) alle barre crypto. merge_asof backward:
crypto-label day i -> ultimo gate equity con label <= day i (strict: < day i).
NaN pre-storia -> gate=1 (nessun de-risk quando non c'e' info)."""
left = pd.DataFrame({"timestamp": crypto_df["timestamp"].astype("int64").values})
g = gate_df.dropna(subset=["gate"]).sort_values("timestamp")
m = pd.merge_asof(left, g, on="timestamp", direction="backward",
allow_exact_matches=not strict)
return pd.Series(m["gate"].values).ffill().fillna(1.0).values
def tp01_pos(df: pd.DataFrame) -> np.ndarray:
return TrendPortfolio(**CANONICAL).target_series(df)
def make_target_fn(gate_builder, strict: bool = False):
"""Ritorna target_fn(df, asset) = posizione TP01 * gate macro allineato (causale).
gate_builder() costruisce il gate sul calendario equity una volta (cache esterna)."""
_MF = macro_frame()
gate_df = gate_builder(_MF)
def target_fn(df: pd.DataFrame, asset: str = "") -> np.ndarray:
pos = tp01_pos(df)
g = align_gate(gate_df, df, strict=strict)
return pos * g
return target_fn, gate_df
# ===========================================================================
# EVALUATION — solo vs gated, combo 50/50 + per-asset, FULL/HOLD/DD/CAGR.
# ===========================================================================
def _combo_daily(target_fn) -> pd.Series:
series = {}
for a in ASSETS:
df = al.get(a, "1d")
ev = al.eval_weights(df, target_fn(df, a))
series[a] = pd.Series(ev["net"], index=ev["idx"])
J = pd.concat(series, axis=1, join="inner").fillna(0.0)
return al._to_daily(0.5 * J[ASSETS[0]] + 0.5 * J[ASSETS[1]])
def _sh(s):
return al._sh(s)
def _metrics(s: pd.Series) -> dict:
sh = _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
yrs = max((s.index[-1] - s.index[0]).days / 365.25, 1e-6)
cagr = eq[-1] ** (1 / yrs) - 1 if len(eq) and eq[-1] > 0 else -1.0
return dict(sharpe=round(sh, 3), dd=round(dd, 4), cagr=round(cagr, 4))
def eval_solo_vs_gated(target_fn):
"""Combo 50/50: solo (gate=1) vs gated. Ritorna dict con FULL e HOLD per entrambi."""
solo_fn = lambda df, a="": tp01_pos(df)
solo = _combo_daily(solo_fn)
gated = _combo_daily(target_fn)
J = pd.concat({"solo": solo, "gated": gated}, axis=1, join="inner").dropna()
JH = J[J.index >= HOLDOUT]
return dict(
full_solo=_metrics(J["solo"]), full_gated=_metrics(J["gated"]),
hold_solo=_metrics(JH["solo"]), hold_gated=_metrics(JH["gated"]),
n=len(J), nh=len(JH), solo_series=J["solo"], gated_series=J["gated"])
def per_asset_table(target_fn) -> dict:
out = {}
for a in ASSETS:
df = al.get(a, "1d")
solo = al.eval_weights(df, tp01_pos(df))
gated = al.eval_weights(df, target_fn(df, a))
out[a] = dict(
full_solo=dict(sharpe=solo["full"]["sharpe"], dd=solo["full"]["maxdd"], cagr=solo["full"]["cagr"]),
full_gated=dict(sharpe=gated["full"]["sharpe"], dd=gated["full"]["maxdd"], cagr=gated["full"]["cagr"]),
hold_solo=dict(sharpe=solo["holdout"].get("sharpe", 0.0)),
hold_gated=dict(sharpe=gated["holdout"].get("sharpe", 0.0)))
return out
# ===========================================================================
# REDUNDANCY DIAGNOSTIC — il controllo DECISIVO contro il trend.
# ===========================================================================
def redundancy_diag(gate_df: pd.DataFrame) -> dict:
"""Quanto FA davvero il gate, dato che TP01 e' gia' flat nei crash?
- exposure TP01 nei giorni risk-off (gate<1) vs risk-on: se gia' ~0 -> ridondante.
- quota di giorni in cui il gate riduce una posizione NON gia' flat (lavoro effettivo).
- corr fra (1-gate) e (1-exposure_norm)."""
lev = CANONICAL["leverage"]
rows = []
for a in ASSETS:
df = al.get(a, "1d")
pos = tp01_pos(df)
expo = np.clip(np.abs(pos) / lev, 0, 1) # esposizione normalizzata 0..1
g = align_gate(gate_df, df)
dt = pd.to_datetime(df["datetime"], utc=True)
m = pd.DataFrame({"dt": dt, "expo": expo, "g": g})
m = m[m["dt"] >= dt.iloc[0]] # tutto
roff = m["g"] < 0.999
ron = ~roff
flat_thr = 0.05
# lavoro effettivo: gate<1 E TP01 non gia' flat
work = roff & (m["expo"] > flat_thr)
rows.append(dict(
asset=a,
expo_riskoff=round(float(m.loc[roff, "expo"].mean()) if roff.any() else 0.0, 3),
expo_riskon=round(float(m.loc[ron, "expo"].mean()) if ron.any() else 0.0, 3),
pct_days_riskoff=round(float(roff.mean()), 3),
pct_days_gate_works=round(float(work.mean()), 3),
corr_1mg_1mexpo=round(float(np.corrcoef(1 - m["g"], 1 - m["expo"])[0, 1])
if m["g"].std() > 0 else float("nan"), 3),
))
return {r["asset"]: r for r in rows}
# ===========================================================================
# RUNNER
# ===========================================================================
GATES = {
# binari classici
"SPY>MA200": lambda mf: gate_trend(mf, "spy", 200, 0.0),
"SPY>MA100": lambda mf: gate_trend(mf, "spy", 100, 0.0),
"SPY>MA50": lambda mf: gate_trend(mf, "spy", 50, 0.0),
"SPY>MA200_half": lambda mf: gate_trend(mf, "spy", 200, 0.5),
"HYG>MA100": lambda mf: gate_trend(mf, "hyg", 100, 0.0),
"HYG/LQD>MA50": lambda mf: gate_ratio(mf, "hyg", "lqd", 50, 0.0),
"HYG/LQD>MA100": lambda mf: gate_ratio(mf, "hyg", "lqd", 100, 0.0),
# combinati / score
"COMBO_maj100": lambda mf: gate_combo(mf, 100, 0.0, thr=0.5),
"COMBO_all100": lambda mf: gate_combo(mf, 100, 0.0, thr=0.99),
"COMBO_cont100": lambda mf: gate_combo(mf, 100, 0.0, continuous=True),
"COMBO_cont100h": lambda mf: gate_combo(mf, 100, 0.5, continuous=True),
# flight-to-quality / vol
"TLTup&SPYdn20": lambda mf: gate_flight(mf, "tlt", "spy", 20, 0.0),
"GLDup&SPYdn20": lambda mf: gate_flight(mf, "gld", "spy", 20, 0.0),
"SPYvol_z1": lambda mf: gate_eqvol(mf, 250, 20, 1.0, 0.0),
}
def delever_frontier(target_vols=(0.10, 0.12, 0.14, 0.16, 0.18, 0.20)) -> dict:
"""CONTROLLO DECISIVO (lezione DVOL-overlay): per meno DD la leva e' target_vol, non un
overlay. Frontiera DD/Sharpe di TP01 puro a target_vol decrescente. Se il miglior gate
sta SOPRA (DD piu' alto a parita' di Sharpe, o Sharpe piu' basso a parita' di DD) di questa
frontiera, il suo taglio di DD e' solo de-levering replicabile meglio senza macro."""
out = {}
for tv in target_vols:
cfg = {**CANONICAL, "target_vol": tv}
fn = lambda df, a="", cfg=cfg: TrendPortfolio(**cfg).target_series(df)
c = _combo_daily(fn)
out[tv] = _metrics(c)
return out
def fmt_cmp(label, m_solo, m_gated) -> str:
ds = m_gated["sharpe"] - m_solo["sharpe"]
dd = m_gated["dd"] - m_solo["dd"]
dc = m_gated.get("cagr", 0) - m_solo.get("cagr", 0)
return (f" {label:5s} Sh {m_solo['sharpe']:+.2f}->{m_gated['sharpe']:+.2f} (d{ds:+.2f}) "
f"DD {m_solo['dd']*100:4.1f}%->{m_gated['dd']*100:4.1f}% (d{dd*100:+.1f}pp) "
f"CAGR {m_solo.get('cagr',0)*100:+5.1f}%->{m_gated.get('cagr',0)*100:+5.1f}% (d{dc*100:+.1f}pp)")
def main():
pd.set_option("display.width", 160)
print("=" * 92)
print("MACRO REGIME-GATE sul book crypto (TP01 BTC/ETH) — Filone D")
print(f" TP01 CANONICAL = {CANONICAL}")
print(f" HOLD-OUT >= {HOLDOUT.date()} fee {al.FEE_SIDE*2*100:.2f}%RT")
mf = macro_frame()
print(f" Macro frame: {len(mf)} barre {mf['dt'].iloc[0].date()} -> {mf['dt'].iloc[-1].date()} "
f"cols={[c for c in mf.columns if c not in ('timestamp','dt')]}")
print("=" * 92)
# ---- 1) SWEEP DI TUTTI I GATE (combo 50/50) -------------------------------------
print("\n[1] SWEEP GATE — combo 50/50 BTC+ETH, FULL & HOLD-OUT, vs TP01-solo\n")
results = {}
for name, builder in GATES.items():
tf, gate_df = make_target_fn(builder)
cmp = eval_solo_vs_gated(tf)
results[name] = (cmp, gate_df, tf)
print(f"GATE {name}")
print(fmt_cmp("FULL", cmp["full_solo"], cmp["full_gated"]))
print(fmt_cmp("HOLD", cmp["hold_solo"], cmp["hold_gated"]))
# baseline (solo) numbers come from any cmp
any_cmp = next(iter(results.values()))[0]
print(f"\n [baseline TP01-solo] FULL Sh {any_cmp['full_solo']['sharpe']} DD {any_cmp['full_solo']['dd']*100:.1f}% "
f"CAGR {any_cmp['full_solo']['cagr']*100:.1f}% | HOLD Sh {any_cmp['hold_solo']['sharpe']} "
f"DD {any_cmp['hold_solo']['dd']*100:.1f}% CAGR {any_cmp['hold_solo']['cagr']*100:.1f}%")
# ---- 2) SELEZIONE: miglior gate per HOLD-OUT Sharpe, poi per riduzione DD --------
def score(name):
c = results[name][0]
return (c["hold_gated"]["sharpe"], -c["full_gated"]["dd"])
best = max(results, key=score)
# anche il "miglior DD-cutter" che non peggiora troppo lo Sharpe FULL
dd_best = min(results, key=lambda n: results[n][0]["full_gated"]["dd"])
print(f"\n[2] Miglior gate per HOLD-OUT Sharpe: {best}")
print(f" Miglior gate per DD FULL ridotto : {dd_best}")
# ---- CONTROLLO DECISIVO: de-lever frontier (target_vol) -------------------------
print("\n[2b] CONTROLLO DECISIVO — TP01 puro a target_vol piu' basso (de-lever) vs gate:")
fr = delever_frontier()
for tv, m in fr.items():
print(f" target_vol {tv:.2f}: FULL Sh {m['sharpe']:+.2f} DD {m['dd']*100:4.1f}% CAGR {m['cagr']*100:+5.1f}%")
print(" -> i gate de-leveranti (COMBO_cont, SPYvol) vanno confrontati con QUESTA frontiera:")
for n in ("COMBO_cont100", "COMBO_cont100h", "SPYvol_z1", "SPY>MA200_half"):
if n in results:
g = results[n][0]["full_gated"]
print(f" {n:16s}: FULL Sh {g['sharpe']:+.2f} DD {g['dd']*100:4.1f}% CAGR {g['cagr']*100:+5.1f}%")
for tag, name in [("BEST-HOLD", best), ("BEST-DD", dd_best)]:
if tag == "BEST-DD" and dd_best == best:
continue
deep_dive(tag, name, results)
def deep_dive(tag, name, results):
cmp, gate_df, tf = results[name]
print("\n" + "=" * 92)
print(f"[DEEP DIVE {tag}] GATE = {name}")
print("=" * 92)
# per-asset
print("\n Per-asset (TP01-solo -> TP01+gate):")
pa = per_asset_table(tf)
for a in ASSETS:
d = pa[a]
print(f" {a}: FULL Sh {d['full_solo']['sharpe']:+.2f}->{d['full_gated']['sharpe']:+.2f} "
f"DD {d['full_solo']['dd']*100:.0f}%->{d['full_gated']['dd']*100:.0f}% "
f"CAGR {d['full_solo']['cagr']*100:+.0f}%->{d['full_gated']['cagr']*100:+.0f}% | "
f"HOLD Sh {d['hold_solo']['sharpe']:+.2f}->{d['hold_gated']['sharpe']:+.2f}")
# ---- 3) CONTROLLO RIDONDANZA COL TREND ------------------------------------------
print("\n [3] CONTROLLO RIDONDANZA COL TREND (il test decisivo):")
rd = redundancy_diag(gate_df)
for a in ASSETS:
r = rd[a]
print(f" {a}: exposure TP01 nei giorni risk-off={r['expo_riskoff']} vs risk-on={r['expo_riskon']} "
f"| giorni risk-off {r['pct_days_riskoff']*100:.0f}% "
f"giorni in cui il gate LAVORA (riduce pos non-flat) {r['pct_days_gate_works']*100:.0f}% "
f"| corr(1-gate, 1-expo)={r['corr_1mg_1mexpo']}")
print(" -> se exposure-risk-off ~ exposure-risk-on e 'gate-lavora' e' basso => RIDONDANTE col trend.")
# ---- 4) MARGINAL SCORER -----------------------------------------------------------
print("\n [4] MARGINAL SCORER vs TP01 (gate come candidato-sleeve):")
rep = al.study_marginal(f"GATE[{name}]", tf, tf="1d")
print(al.fmt_marginal(rep))
# overlay-delta: lo STREAM incrementale del gate = gated - solo (e' alpha o hedge?)
delta = (cmp["gated_series"] - cmp["solo_series"]).dropna()
print("\n overlay-delta (gated - solo) come stream a se':")
md = al.marginal_vs_tp01(delta)
print(f" verdict={md.get('marginal_verdict')} corr->TP01 {md.get('corr_full')} "
f"is_hedge={md.get('is_hedge')} uplift TP01-up {md.get('uplift_tp01_up')} / "
f"TP01-down {md.get('uplift_tp01_down')} cand-Sh full {md.get('cand_full_sharpe')}")
# ---- 5) FEE SWEEP -----------------------------------------------------------------
print("\n [5] FEE SWEEP (combo 50/50 gated, FULL Sharpe):")
for f in al.FEE_SWEEP:
series = {}
for a in ASSETS:
df = al.get(a, "1d")
ev = al.eval_weights(df, tf(df, a), fee_side=f)
series[a] = pd.Series(ev["net"], index=ev["idx"])
J = pd.concat(series, axis=1, join="inner").fillna(0.0)
d = al._to_daily(0.5 * J[ASSETS[0]] + 0.5 * J[ASSETS[1]])
print(f" {2*f*100:.2f}%RT -> Sh {_sh(d):+.2f}")
# ---- 6) ESEGUIBILITA' a $600 ------------------------------------------------------
print("\n [6] ESEGUIBILITA' a $600 (eval_weights_smallcap, haircut reale):")
for a in ASSETS:
df = al.get(a, "1d")
sc = al.eval_weights_smallcap(df, tf(df, a), capital=600, min_order=5)
print(f" {a}: modeled Sh {sc['modeled']['sharpe']:+.2f} -> real Sh {sc['realistic']['sharpe']:+.2f} "
f"haircut {sc['sharpe_haircut']:+.2f} trade eseguiti {sc['n_executed_trades']}")
# ---- 7) LEAK / BOUNDARY -----------------------------------------------------------
print("\n [7] LEAK-FREE & BOUNDARY:")
cz = al.causality_ok(tf, tf="1d")
print(f" causality_ok={cz['ok']} (max_tail_diff {cz['max_tail_diff']}, checked {cz['checked']})")
# variante STRICT (equity-label < crypto-label): margine causale extra
tf_strict, _ = make_target_fn(GATES[name], strict=True)
cmp_s = eval_solo_vs_gated(tf_strict)
print(f" STRICT align (1 barra equity extra di lag): FULL Sh {cmp_s['full_gated']['sharpe']:+.2f} "
f"(vs {cmp['full_gated']['sharpe']:+.2f}) HOLD Sh {cmp_s['hold_gated']['sharpe']:+.2f} "
f"(vs {cmp['hold_gated']['sharpe']:+.2f}) -> robusto se ~uguale")
db = al.day_boundary_robust(tf, tf="1d")
print(f" day_boundary_robust={db['verdict']} (spread {db.get('spread')}, per-offset {db.get('per_offset')})")
# ---- 8) PLATEAU (solo per i trend MA) ---------------------------------------------
if name.startswith("SPY>MA") or name.startswith("HYG"):
print("\n [8] PLATEAU su finestra MA (SPY trend), g_off=0:")
for n in (50, 100, 150, 200, 250):
tfn, _ = make_target_fn(lambda mf, n=n: gate_trend(mf, "spy", n, 0.0))
cc = eval_solo_vs_gated(tfn)
print(f" SPY>MA{n:3d}: FULL Sh {cc['full_gated']['sharpe']:+.2f} DD {cc['full_gated']['dd']*100:.1f}% "
f"HOLD Sh {cc['hold_gated']['sharpe']:+.2f} DD {cc['hold_gated']['dd']*100:.1f}%")
if __name__ == "__main__":
main()