Adriano Dal Pastro
aad69f9790
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>
91 lines
3.9 KiB
Python
91 lines
3.9 KiB
Python
"""Test del filone D — MACRO REGIME-GATE (scripts/research/macro_regime_gate.py).
|
|
|
|
Verifica leggera che:
|
|
* il frame macro (ETF daily) carichi e sia su un calendario monotono;
|
|
* i gate builder producano un gate in [g_off, 1] e CAUSALE (SMA/ratio rolling, no future);
|
|
* align_gate sia backward-only (la barra crypto i usa solo gate equity con label <= i);
|
|
* il VERDETTO del filone regga: il gate macro e' RIDONDANTE col trend di TP01 — "lavora"
|
|
(riduce una posizione TP01 NON gia' flat) solo in una piccola quota di giorni.
|
|
NB: un solo gate per i test lenti (costruisce TP01) -> velocita'.
|
|
"""
|
|
import sys
|
|
import importlib.util
|
|
|
|
import numpy as np
|
|
|
|
sys.path.insert(0, "/opt/docker/PythagorasGoal")
|
|
sys.path.insert(0, "/opt/docker/PythagorasGoal/scripts/research/alt")
|
|
import altlib as al # noqa: E402
|
|
|
|
_spec = importlib.util.spec_from_file_location(
|
|
"macro_regime_gate", "/opt/docker/PythagorasGoal/scripts/research/macro_regime_gate.py")
|
|
mg = importlib.util.module_from_spec(_spec)
|
|
_spec.loader.exec_module(mg)
|
|
|
|
|
|
def test_macro_frame_loads_and_monotone():
|
|
mf = mg.macro_frame()
|
|
for col in ("spy", "qqq", "hyg", "lqd", "gld", "tlt", "ief"):
|
|
assert col in mf.columns, f"colonna macro mancante: {col}"
|
|
ts = mf["timestamp"].values
|
|
assert (np.diff(ts) > 0).all(), "calendario macro non strettamente crescente"
|
|
# spy non-null dopo l'inizio (master del calendario)
|
|
assert np.isfinite(mf["spy"].values[-1])
|
|
|
|
|
|
def test_gate_in_range_and_binary():
|
|
mf = mg.macro_frame()
|
|
for gdf, g_off in ((mg.gate_trend(mf, "spy", 200, 0.0), 0.0),
|
|
(mg.gate_combo(mf, 200, 0.5), 0.5)):
|
|
g = gdf["gate"].values
|
|
fin = g[np.isfinite(g)]
|
|
assert len(fin) > 0
|
|
assert (fin >= g_off - 1e-9).all() and (fin <= 1.0 + 1e-9).all(), \
|
|
"gate fuori da [g_off, 1]"
|
|
assert len(gdf) == len(mf)
|
|
|
|
|
|
def test_gate_trend_causal_prefix():
|
|
# gate_trend usa solo SMA rolling -> ricalcolato su un prefisso, la coda combacia
|
|
mf = mg.macro_frame()
|
|
full = mg.gate_trend(mf, "spy", 200, 0.0)["gate"].values
|
|
k = len(mf) - 300
|
|
pref = mg.gate_trend(mf.iloc[:k].reset_index(drop=True), "spy", 200, 0.0)["gate"].values
|
|
a, b = full[:k], pref
|
|
both = np.isfinite(a) & np.isfinite(b)
|
|
assert both.any()
|
|
assert np.allclose(a[both], b[both]), "look-ahead nel gate_trend (prefix != coda)"
|
|
|
|
|
|
def test_align_gate_backward_no_future():
|
|
# la barra crypto i deve mappare a un gate equity con timestamp <= timestamp crypto i
|
|
mf = mg.macro_frame()
|
|
gate_df = mg.gate_trend(mf, "spy", 200, 0.0)
|
|
df = al.get("BTC", "1d")
|
|
g = mg.align_gate(gate_df, df)
|
|
assert len(g) == len(df)
|
|
assert np.isfinite(g).all()
|
|
# nessun valore di gate puo' provenire dal futuro: per una manciata di barre crypto,
|
|
# il gate allineato deve coincidere con l'ultimo gate equity NON-NaN con ts <= ts_crypto
|
|
gd = gate_df.dropna(subset=["gate"]).sort_values("timestamp")
|
|
gt, gv = gd["timestamp"].values, gd["gate"].values
|
|
cts = df["timestamp"].astype("int64").values
|
|
for i in range(len(cts) - 1, max(len(cts) - 50, 0), -1):
|
|
prior = np.searchsorted(gt, cts[i], side="right") - 1
|
|
if prior < 0:
|
|
continue
|
|
assert abs(g[i] - gv[prior]) < 1e-9, f"align_gate usa gate futuro a i={i}"
|
|
|
|
|
|
def test_verdict_gate_is_redundant_with_trend():
|
|
# IL VERDETTO del filone D: TP01 e' gia' flat nei crash -> il gate macro "lavora"
|
|
# (riduce una posizione NON gia' flat) solo in una piccola quota di giorni.
|
|
mf = mg.macro_frame()
|
|
gate_df = mg.gate_combo(mf, 200, 0.0) # combo SPY+HYG+HYG/LQD, de-risk a 0
|
|
diag = mg.redundancy_diag(gate_df)
|
|
for a in mg.ASSETS:
|
|
r = diag[a]
|
|
# esposizione TP01 nei giorni risk-off gia' bassa, e gate "lavora" raramente
|
|
assert r["pct_days_gate_works"] < 0.20, \
|
|
f"{a}: gate lavora {r['pct_days_gate_works']} (atteso piccolo = ridondante col trend)"
|