Adriano/PythagorasGoal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
2751a7efd0871c44941aedbf479000f56db3c2c7
PythagorasGoal/scripts/research/r0822_ortho_screen.py
T

685 lines
34 KiB
Python
Raw Blame History

#!/usr/bin/env python3
"""r0822_ortho_screen.py — ONDATA 2026-08-22, filone ORTHO-SCREEN (la rete larga, onesta).
NON e' un'ipotesi: e' un SETACCIO. Sette famiglie di segnale mai (o solo parzialmente)
coperte dalle ondate precedenti, passate TUTTE dallo stesso protocollo, senza eccezioni:
1. famiglia DICHIARATA prima (parametri + celle), trial contati AL RIALZO;
2. A.study_family_honest -> cella scelta IN-SAMPLE-ONLY + deflated-Sharpe di famiglia;
3. A.marginal_vs_tp01 -> ADDS / NEUTRAL / DILUTES / HEDGE / NOISE / REDUNDANT;
4. A.causality_ok (+ A.day_boundary_robust dove ha senso, vedi NOTA CONFINE);
5. banda d'ancora su 8 offset orari (bar 1d ricostruite da 1h) -> la stima onesta e' la MEDIANA;
6. null del DE-LEVERING su ogni claim di drawdown;
7. A.eval_weights_smallcap($600) + A.implausible_sharpe sulla cella scelta;
8. **DSR di SCREEN**: oltre al deflated-Sharpe di famiglia, il deflated-Sharpe calcolato sul
POOL di TUTTE le celle di TUTTE le famiglie. Se scelgo il migliore fra 7 famiglie, i trial
sono 168, non 24: e' l'unico conto onesto per un setaccio.
PRIOR DICHIARATO PRIMA DI MISURARE: il soffitto direzionale BTC/ETH misurato e' Sharpe ~1.3 e
in 104 ipotesi + 153 agenti (giugno 2026) NESSUNA e' sopravvissuta alla verifica avversariale.
Mi aspetto 0 CANDIDATI. Il valore consegnato e' **quale gate uccide cosa**, perche' questo
restringe lo spazio della prossima ondata.
LE 7 FAMIGLIE (24 celle ciascuna = 168 trial dichiarati)
RSKEW skew realizzata (da 1h) come PREDITTORE direzionale, non come gate
[nuovo vs blind/agent_40_skewgate, che la usava come FILTRO su un trend]
TACC accelerazione del trend (derivata seconda del log-prezzo) come direzione standalone
VPX relazione volume-prezzo su barre 1d (conferma / divergenza a volume alto e basso)
XDISP dispersione cross-sezionale dei 51 alt HL usata come segnale TIME-SERIES sul mercato
[nuovo vs XS01, dove la dispersione e' un gate sul paniere cross-sectional]
XCORR correlazione realizzata BTC-ETH come STATO di mercato direzionale
[nuovo vs ortho/agent_09 e XAS08, dove la corr gata un book relative-value]
XTAIL struttura dei ritorni dopo uno shock a k-sigma (continuazione vs inversione)
VRAT variance-ratio (persistenza) come CONVINZIONE continua sul TSMOM
[nuovo vs blind/agent_37_hurst, che faceva uno switch BINARIO di modo]
NOTA CONFINE (perche' day_boundary_robust non gira su tutte): il test sposta le ETICHETTE del
calendario lasciando i prezzi fermi. Per un segnale di solo prezzo (TACC/VPX/XTAIL/VRAT) e'
esattamente il controllo giusto e deve dare INVARIANT. Per un segnale che fa merge_asof su una
serie AUSILIARIA (RSKEW da 1h, XDISP da HL, XCORR da 1h BTC/ETH) spostare l'etichetta in AVANTI
sposta il punto di merge nel FUTURO: sarebbe un LEAK indotto dallo strumento, non un controllo.
Per quelle famiglie il test d'ancora corretto e' la RICOSTRUZIONE delle barre a offset orario
(banda d'ancora, punto 5), che gira su tutte e sette.
USO: nice -n 19 timeout 900 uv run python scripts/research/r0822_ortho_screen.py
"""
from __future__ import annotations
import os
import sys
import time
from functools import lru_cache
import numpy as np
import pandas as pd
sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "alt"))
import altlib as A # noqa: E402
HL_DIR = A.DATA_DIR
ANCHOR_OFFSETS = (0, 3, 6, 9, 12, 15, 18, 21)
W_BLEND = 0.25 # peso del candidato nel blend con TP01 (convenzione altlib)
# ===========================================================================
# INFRASTRUTTURA — barre a offset, merge causale di serie ausiliarie
# ===========================================================================
@lru_cache(maxsize=64)
def bars_at(asset: str, off: int) -> pd.DataFrame:
"""Barre 1d ancorate all'ora `off`, ricostruite dal feed 1h certificato.
off=0 deve riprodurre esattamente A.get(asset,'1d') (controllo in _sanity)."""
h = A.get(asset, "1h").copy()
idx = pd.to_datetime(h["timestamp"], unit="ms", utc=True) - pd.Timedelta(hours=off)
h.index = idx
o = (h.resample("1D", label="left", closed="left")
.agg({"open": "first", "high": "max", "low": "min", "close": "last", "volume": "sum"})
.dropna(subset=["open"]))
o["datetime"] = o.index + pd.Timedelta(hours=off)
epoch = pd.Timestamp("1970-01-01", tz="UTC")
o["timestamp"] = ((o["datetime"] - epoch) // pd.Timedelta(milliseconds=1)).astype("int64")
return o.reset_index(drop=True)[["timestamp", "open", "high", "low", "close", "volume", "datetime"]]
def _bar_ms(df: pd.DataFrame) -> int:
d = pd.Series(df["timestamp"].astype("int64")).diff().median()
return int(d) if np.isfinite(d) and d > 0 else 86_400_000
def merge_aux(df: pd.DataFrame, aux_ts: np.ndarray, aux_val: np.ndarray) -> np.ndarray:
"""Valore di una serie ausiliaria NOTO alla CHIUSURA di ogni barra di df.
`aux_ts` e' gia' il timestamp in cui il valore diventa noto (fine della sua barra).
merge_asof backward sul close della barra -> mai un dato del futuro: eval_weights
tiene target[i] durante la barra i+1, quindi close[i] e' l'ultimo istante lecito."""
close_ts = df["timestamp"].astype("int64").values + _bar_ms(df)
left = pd.DataFrame({"timestamp": close_ts})
right = (pd.DataFrame({"timestamp": np.asarray(aux_ts, dtype="int64"),
"v": np.asarray(aux_val, dtype=float)})
.dropna().sort_values("timestamp").reset_index(drop=True))
if right.empty:
return np.full(len(df), np.nan)
return pd.merge_asof(left, right, on="timestamp", direction="backward")["v"].values
def roll_z(x: np.ndarray, win: int) -> np.ndarray:
s = pd.Series(np.asarray(x, float))
mp = max(20, win // 4)
m = s.rolling(win, min_periods=mp).mean()
sd = s.rolling(win, min_periods=mp).std()
return ((s - m) / sd.where(sd > 0)).values
def exp_pct(x: np.ndarray, min_obs: int = 180) -> np.ndarray:
"""Percentile ESPANDENTE causale (rank fra tutti i valori visti finora, incluso l'attuale)."""
return pd.Series(np.asarray(x, float)).expanding(min_periods=min_obs).rank(pct=True).values
def _shape(u: np.ndarray, shape: str) -> np.ndarray:
u = np.nan_to_num(np.asarray(u, float), nan=0.0)
return np.sign(u) if shape == "sign" else np.clip(u, -1.0, 1.0)
def _sized(direction: np.ndarray, df: pd.DataFrame) -> np.ndarray:
return A.vol_target(np.nan_to_num(direction, nan=0.0), df,
target_vol=0.20, vol_win_days=30, leverage_cap=2.0)
# ===========================================================================
# SERIE AUSILIARIE (calcolate una volta, riusate da tutte le celle)
# ===========================================================================
@lru_cache(maxsize=16)
def skew_aux(asset: str, w_days: int):
"""Skew realizzata dei ritorni 1h su finestra trascinata di w_days giorni.
Valore al bar 1h i -> noto alla sua chiusura (ts_i + 1h)."""
h = A.get(asset, "1h")
r = A.log_returns(h["close"].values.astype(float))
win = w_days * 24
sk = pd.Series(r).rolling(win, min_periods=win // 2).skew().values
return h["timestamp"].astype("int64").values + 3_600_000, sk
@lru_cache(maxsize=4)
def disp_aux():
"""Dispersione cross-sezionale dei 51 alt Hyperliquid (1d, 2024+): deviazione standard
cross-section dei log-ritorni giornalieri. Valore del giorno d noto alla chiusura di d."""
import glob
files = sorted(glob.glob(str(HL_DIR / "hl_*_1d.parquet")))
cols = {}
for f in files:
x = pd.read_parquet(f, columns=["timestamp", "close"])
s = pd.Series(x["close"].values.astype(float),
index=x["timestamp"].astype("int64").values)
cols[os.path.basename(f)] = np.log(s).diff()
M = pd.DataFrame(cols).sort_index()
n_ok = M.notna().sum(axis=1)
disp = M.std(axis=1, ddof=1).where(n_ok >= 15)
ts = M.index.values.astype("int64") + 86_400_000 # noto a fine giornata
return ts, disp.values, int(len(files))
@lru_cache(maxsize=16)
def xcorr_aux(w_days: int):
"""Correlazione realizzata BTC-ETH sui ritorni 1h, finestra trascinata w_days giorni."""
b = A.get("BTC", "1h")[["timestamp", "close"]].rename(columns={"close": "b"})
e = A.get("ETH", "1h")[["timestamp", "close"]].rename(columns={"close": "e"})
J = pd.merge(b, e, on="timestamp", how="inner").sort_values("timestamp")
rb = pd.Series(A.log_returns(J["b"].values.astype(float)))
re = pd.Series(A.log_returns(J["e"].values.astype(float)))
win = w_days * 24
c = rb.rolling(win, min_periods=win // 2).corr(re).values
return J["timestamp"].astype("int64").values + 3_600_000, c
# ===========================================================================
# LE 7 FAMIGLIE — factory(tf=..., **params) -> target(df, asset)
# ===========================================================================
def mk_rskew(tf="1d", w_days=30, z_days=365, shape="lin", sgn=-1):
"""RSKEW: la skew realizzata recente come PREDITTORE del ritorno successivo.
A-priori (Amaya et al., azioni): skew realizzata alta -> ritorno futuro BASSO (sgn=-1)."""
def target(df, asset):
ts, sk = skew_aux(asset, w_days)
v = merge_aux(df, ts, sk)
u = np.clip(roll_z(v, z_days) / 2.0, -1.0, 1.0)
return _sized(sgn * _shape(u, shape), df)
return target
def mk_tacc(tf="1d", H=60, L=20, shape="lin", sgn=1):
"""TACC: accelerazione = variazione della PENDENZA log-prezzo su H barre, misurata a L barre
di distanza. Direzione standalone (non e' il LIVELLO del trend). A-priori sgn=+1."""
def target(df, asset):
c = np.log(df["close"].values.astype(float))
n = len(c)
slope = np.full(n, np.nan)
slope[H:] = (c[H:] - c[:-H]) / H
acc = np.full(n, np.nan)
acc[L:] = slope[L:] - slope[:-L]
u = np.clip(roll_z(acc, 365) / 2.0, -1.0, 1.0)
return _sized(sgn * _shape(u, shape), df)
return target
def mk_vpx(tf="1d", W=30, k=5, mode="confirm", thr=0.5):
"""VPX: volume-prezzo su barre 1d (a 1d il volume Deribit e' informativo, a 5m no).
confirm -> segui il movimento SOLO se il volume conferma (z >= thr)
fade_low -> fai il contrario del movimento avvenuto a volume BASSO (z <= -thr)
fade_high -> fai il contrario del movimento avvenuto a volume ALTO (climax, z >= thr)"""
def target(df, asset):
c = df["close"].values.astype(float)
vol = df["volume"].values.astype(float)
n = len(c)
vz = roll_z(np.log(np.where(vol > 0, vol, np.nan)), W)
rk = np.full(n, np.nan)
rk[k:] = c[k:] / c[:-k] - 1.0
s = np.sign(np.nan_to_num(rk, nan=0.0))
vzf = np.nan_to_num(vz, nan=0.0)
if mode == "confirm":
d = np.where(vzf >= thr, s, 0.0)
elif mode == "fade_low":
d = np.where(vzf <= -thr, -s, 0.0)
else: # fade_high
d = np.where(vzf >= thr, -s, 0.0)
return _sized(d, df)
return target
def mk_xdisp(tf="1d", S=5, trans="z252", shape="lin", sgn=1):
"""XDISP: dispersione cross-sezionale dei 51 alt HL come STATO del mercato (time-series),
non come gate di un paniere cross-sectional. Nessun prior forte sul segno -> entrambi in griglia."""
def target(df, asset):
ts, dsp, _ = disp_aux()
sm = pd.Series(dsp).rolling(S, min_periods=1).mean().values
u = (np.clip(roll_z(sm, 252) / 2.0, -1, 1) if trans == "z252"
else 2.0 * exp_pct(sm, 180) - 1.0)
v = merge_aux(df, ts, u)
return _sized(sgn * _shape(v, shape), df)
return target
def mk_xcorr(tf="1d", W=60, trans="z252", shape="lin", sgn=-1):
"""XCORR: co-movimento BTC-ETH (corr realizzata) come stato direzionale del mercato.
A-priori debole: corr alta = 'tutto il crypto e' un trade solo' = regime tardo -> sgn=-1."""
def target(df, asset):
ts, cr = xcorr_aux(W)
u_aux = (np.clip(roll_z(cr, 252 * 24) / 2.0, -1, 1) if trans == "z252"
else 2.0 * exp_pct(cr, 180 * 24) - 1.0)
v = merge_aux(df, ts, u_aux)
return _sized(sgn * _shape(v, shape), df)
return target
def mk_xtail(tf="1d", ksig=3.0, N=3, sgn=1, sigwin=30):
"""XTAIL: struttura dei ritorni ESTREMI. Dopo |r| > ksig*sigma trascurata, tieni per N barre
la direzione (sgn=+1 continuazione) o il suo opposto (sgn=-1 inversione).
ATTENZIONE dichiarata: sgn=-1 e' mean-reversion, famiglia gia' MORTA -> se vince li', si chiude."""
def target(df, asset):
c = df["close"].values.astype(float)
r = A.simple_returns(c)
sd = pd.Series(r).rolling(sigwin, min_periods=sigwin // 2).std().shift(1).values
n = len(c)
d = np.zeros(n)
hit = np.where(np.isfinite(sd) & (sd > 0) & (np.abs(r) > ksig * sd))[0]
for i in hit:
d[i:min(n, i + N)] = sgn * np.sign(r[i])
return _sized(d, df)
return target
def mk_vrat(tf="1d", q=10, LB=365, conv="gate", lf=True):
"""VRAT: variance-ratio VR(q)=Var(r_q)/(q*Var(r_1)) su finestra LB come CONVINZIONE CONTINUA
sul TSMOM (non uno switch binario di modo come blind/agent_37_hurst).
gate -> pesa il trend solo dove VR>1 (persistente), altrimenti flat
flip -> inverte il trend dove VR<1 (anti-persistente)"""
HZ = (30, 90, 180)
def target(df, asset):
c = df["close"].values.astype(float)
lc = np.log(c)
n = len(c)
r1 = pd.Series(A.log_returns(c))
rq = pd.Series(np.concatenate([np.full(q, np.nan), lc[q:] - lc[:-q]]))
v1 = r1.rolling(LB, min_periods=LB // 2).var()
vq = rq.rolling(LB, min_periods=LB // 2).var()
vr = (vq / (q * v1.where(v1 > 0))).values
dirs = np.zeros(n)
for H in HZ:
m = np.zeros(n)
m[H:] = np.sign(c[H:] / c[:-H] - 1.0)
dirs += m / len(HZ)
if lf:
dirs = np.maximum(dirs, 0.0)
k = np.nan_to_num(vr, nan=1.0) - 1.0
w = np.clip(k * 5.0, 0.0, 1.0) if conv == "gate" else np.clip(k * 5.0, -1.0, 1.0)
return _sized(dirs * w, df)
return target
# ---- griglie DICHIARATE (24 celle ciascuna, 7 famiglie = 168 trial) ----------
def _grid(**axes):
keys = list(axes)
out = [{}]
for k in keys:
out = [dict(o, **{k: v}) for o in out for v in axes[k]]
return out
FAMILIES = [
dict(code="RSKEW", factory=mk_rskew, price_only=False,
grid=_grid(w_days=[14, 30, 60], z_days=[180, 365], shape=["lin", "sign"], sgn=[1, -1])),
dict(code="TACC", factory=mk_tacc, price_only=True,
grid=_grid(H=[20, 60, 120], L=[5, 20], shape=["lin", "sign"], sgn=[1, -1])),
dict(code="VPX", factory=mk_vpx, price_only=True,
grid=_grid(W=[30, 90], k=[5, 20], mode=["confirm", "fade_low", "fade_high"], thr=[0.5, 1.0])),
dict(code="XDISP", factory=mk_xdisp, price_only=False,
grid=_grid(S=[1, 5, 20], trans=["z252", "pct"], shape=["lin", "sign"], sgn=[1, -1])),
dict(code="XCORR", factory=mk_xcorr, price_only=False,
grid=_grid(W=[20, 60, 120], trans=["z252", "pct"], shape=["lin", "sign"], sgn=[1, -1])),
dict(code="XTAIL", factory=mk_xtail, price_only=True,
grid=_grid(ksig=[2.5, 3.0], N=[1, 3, 10], sgn=[1, -1], sigwin=[30, 90])),
dict(code="VRAT", factory=mk_vrat, price_only=True,
grid=_grid(q=[5, 10, 20], LB=[180, 365], conv=["gate", "flip"], lf=[True, False])),
]
# ===========================================================================
# GATE AGGIUNTIVI
# ===========================================================================
def cand_daily_at(fn, off: int) -> pd.Series:
"""Serie giornaliera 50/50 BTC+ETH del candidato con barre ancorate all'ora `off`."""
ser = {}
for a in A.CERTIFIED:
df = bars_at(a, off)
ev = A.eval_weights(df, A._call_target(fn, df, a))
ser[a] = pd.Series(ev["net"], index=ev["idx"])
J = pd.concat(ser, axis=1, join="inner").fillna(0.0)
return A._to_daily(0.5 * J[A.CERTIFIED[0]] + 0.5 * J[A.CERTIFIED[1]])
def anchor_report(fn) -> dict:
"""Banda d'ancora su 8 offset orari: Sharpe standalone E uplift del blend.
La stima ONESTA e' la MEDIANA della banda, non l'ancora 00:00 (lezione 02/07)."""
band = A.anchor_luck_band(lambda o: cand_daily_at(fn, o), ANCHOR_OFFSETS)
B = A.tp01_baseline_daily()
ups = {}
for o in ANCHOR_OFFSETS:
C = cand_daily_at(fn, o)
J = pd.concat({"B": B, "C": C}, axis=1, join="inner").dropna()
if len(J) > 30:
ups[o] = round(A._sh((1 - W_BLEND) * J["B"] + W_BLEND * J["C"]) - A._sh(J["B"]), 3)
vals = np.array(list(ups.values()), float) if ups else np.array([np.nan])
band["uplift_per_offset"] = ups
band["uplift_canonical"] = ups.get(0)
band["uplift_median"] = round(float(np.median(vals)), 3)
band["uplift_frac_positive"] = round(float((vals > 0).mean()), 3)
return band
def delever_null(C: pd.Series, w: float = W_BLEND) -> dict:
"""NULL DEL DE-LEVERING (obbligatorio su ogni claim di meno drawdown, 5 occorrenze nel
progetto): esiste k<1 che, applicato al SOLO TP01, da' lo STESSO maxDD con Sharpe MIGLIORE?
Se si', l'overlay non serve: bastava meno leva."""
B = A.tp01_baseline_daily()
J = pd.concat({"B": B, "C": C}, axis=1, join="inner").dropna()
if len(J) < 60:
return dict(run=False, reason="overlap insufficiente")
blend = (1 - w) * J["B"] + w * J["C"]
dd_bl, sh_bl = A._dd_ret(blend), A._sh(blend)
dd_b, sh_b = A._dd_ret(J["B"]), A._sh(J["B"])
if dd_bl >= dd_b:
return dict(run=True, claims_less_dd=False, dd_blend=round(dd_bl, 4),
dd_tp01=round(dd_b, 4), sharpe_blend=round(sh_bl, 3),
sharpe_tp01=round(sh_b, 3),
note="nessun claim di DD: il blend NON riduce il drawdown -> null non pertinente")
ks = np.linspace(0.05, 1.0, 96)
dds = np.array([A._dd_ret(k * J["B"]) for k in ks])
k = float(ks[int(np.argmin(np.abs(dds - dd_bl)))])
sh_k = A._sh(k * J["B"])
return dict(run=True, claims_less_dd=True, dd_blend=round(dd_bl, 4), dd_tp01=round(dd_b, 4),
sharpe_blend=round(sh_bl, 3), k_equal_dd=round(k, 3),
dd_k=round(A._dd_ret(k * J["B"]), 4), sharpe_k_tp01=round(sh_k, 3),
passes=bool(sh_bl > sh_k),
note=("il blend batte il de-levering" if sh_bl > sh_k
else "REFUTED: k*TP01 da' lo stesso DD con Sharpe migliore"))
def smallcap_report(fn) -> dict:
out = {}
for a in A.CERTIFIED:
df = A.get(a, "1d")
sc = A.eval_weights_smallcap(df, A._call_target(fn, df, a), capital=600.0, min_order=5.0)
out[a] = dict(modeled=sc["modeled"]["sharpe"], realistic=sc["realistic"]["sharpe"],
haircut=sc["sharpe_haircut"], n_trades=sc["n_executed_trades"])
return out
def active_window_metrics(C: pd.Series) -> dict:
"""Trappola #9 (barre attive vs calendario): una serie che e' 0 per meta' della storia
(XDISP parte nel 2024) ha Sharpe DILUITO di sqrt(frazione attiva). Si riporta anche il
numero sulla sola finestra ATTIVA, senza usarlo per decidere."""
nz = C[C != 0.0]
if len(nz) < 30:
return dict(active_days=int(len(nz)))
first = nz.index[0]
sub = C[C.index >= first]
return dict(active_days=int(len(nz)), frac_active=round(float(len(nz) / max(1, len(C))), 3),
first_active=str(first.date()), sharpe_active_window=round(A._sh(sub), 3),
maxdd_active_window=round(A._dd_ret(sub), 4))
# ===========================================================================
# SANITY — un setaccio che non si controlla non sta setacciando niente
# ===========================================================================
def sanity() -> None:
print("### SANITY")
for a in A.CERTIFIED:
b0, g = bars_at(a, 0), A.get(a, "1d")
n = min(len(b0), len(g))
d = float(np.max(np.abs(b0["close"].values[:n] - g["close"].values[:n])))
dts = float(np.max(np.abs(b0["timestamp"].values[:n] - g["timestamp"].values[:n])))
print(f" bars_at({a},0) vs A.get({a},'1d'): n={n}/{len(g)} max|dclose|={d:.6g} max|dts|={dts:.0f}")
assert d < 1e-9 and dts == 0, "ricostruzione barre a offset 0 NON identica al feed certificato"
ts, dsp, nfiles = disp_aux()
ok = np.isfinite(dsp)
print(f" disp_aux: {nfiles} file HL, {int(ok.sum())} giorni con >=15 asset, "
f"da {pd.Timestamp(ts[ok][0], unit='ms', tz='UTC').date()} "
f"a {pd.Timestamp(ts[ok][-1], unit='ms', tz='UTC').date()}")
B = A.tp01_baseline_daily()
print(f" TP01 baseline: n={len(B)} ShFULL={A._sh(B):.3f} ShHOLD={A._sh(B[B.index >= A.HOLDOUT]):.3f}")
print()
# ===========================================================================
# MAIN
# ===========================================================================
def main() -> None:
t0 = time.time()
print(__doc__.split("USO:")[0])
sanity()
pooled_sh: list[float] = []
results = []
for fam in FAMILIES:
code, fac, grid = fam["code"], fam["factory"], fam["grid"]
t1 = time.time()
print(f"### {code} — {len(grid)} celle dichiarate")
sel = A.select_cell_insample(fac, grid, ("1d",))
pooled_sh.extend([s for s in sel["all_full_sharpe"] if np.isfinite(s)])
rep = A.study_family_honest(code, fac, grid, ("1d",))
ch = rep["chosen"]
if ch is None:
print(f" nessuna cella valida ({rep.get('reason')})\n")
results.append(dict(code=code, n_cells=len(grid), chosen=None,
killed_by="nessuna cella valutabile"))
continue
fn = fac(tf=ch["tf"], **ch["params"])
C = A.candidate_daily(fn, tf=ch["tf"])
m = rep["marginal"]["marginal"]
absr = rep["marginal"]["absolute"]["cells"][0]
caus = A.causality_ok(fn, tf="1d")
dayb = A.day_boundary_robust(fn, tf="1d") if fam["price_only"] else \
dict(verdict="NON GIRATO", reason="segnale con merge su serie ausiliaria: "
"spostare l'etichetta in avanti sposta il merge nel FUTURO (leak indotto), "
"non e' un controllo -> vale la banda d'ancora")
anch = anchor_report(fn)
deln = delever_null(C)
smal = smallcap_report(fn)
impl = A.implausible_sharpe(C)
actw = active_window_metrics(C)
full_sh = A._sh(C)
hold_sh = A._sh(C[C.index >= A.HOLDOUT])
mm = A._metrics_from_net(C.values, C.index)
results.append(dict(
code=code, n_cells=len(grid), chosen=ch, fn=fn, C=C,
full_sharpe=round(full_sh, 3), hold_sharpe=round(hold_sh, 3),
maxdd=mm["maxdd"], cagr=mm["cagr"],
corr=m.get("corr_full"), corr_hold=m.get("corr_hold"),
verdict=rep["marginal"]["marginal_verdict"],
earns_slot=rep["earns_slot_marginal"], honest=rep["earns_slot_honest"],
dsr_family=rep["deflated_sharpe"], dsr_pass=rep["dsr_pass"],
insample_sharpe=m.get("cand_insample_sharpe"),
has_insample_edge=m.get("has_insample_edge"),
robust_oos=m.get("robust_oos"), is_hedge=m.get("is_hedge"),
uplift_full=m["blends"]["w25"]["uplift_full"],
uplift_hold=m["blends"]["w25"]["uplift_hold"],
multicut=m.get("multicut_uplift"), abs_grade=rep["marginal"]["abs_grade"], _m=m,
fee_survives=absr.get("fee_survives"),
causality=caus, dayb=dayb, anchor=anch, delever=deln,
smallcap=smal, implausible=impl, active=actw,
))
print(f" cella IN-SAMPLE-ONLY: {ch['params']} (IS Sh {ch['insample_sharpe']}, "
f"rango 1/{len(sel['rows'])} in-sample; FULL Sh di quella cella {ch['full_sharpe']})")
print(f" standalone FULL {full_sh:+.3f} | HOLD {hold_sh:+.3f} | maxDD {mm['maxdd']:.1%} | "
f"CAGR {mm['cagr']:+.2%} | IS {m.get('cand_insample_sharpe')}")
print(f" marginale {rep['marginal']['marginal_verdict']:<9} corr->TP01 {m.get('corr_full')} "
f"(hold {m.get('corr_hold')}) uplift w25 full {m['blends']['w25']['uplift_full']:+.3f} "
f"hold {m['blends']['w25']['uplift_hold']}")
print(f" robust_oos={m.get('robust_oos')} insample_edge={m.get('has_insample_edge')} "
f"is_hedge={m.get('is_hedge')} abs={rep['marginal']['abs_grade']} "
f"fee_survives={absr.get('fee_survives')}")
print(f" DSR famiglia {rep['deflated_sharpe']} (atteso null max {rep['expected_null_max']}) "
f"pass={rep['dsr_pass']} earns_slot_honest={rep['earns_slot_honest']}")
print(f" causality ok={caus['ok']} max_tail_diff={caus['max_tail_diff']}")
print(f" confine {dayb.get('verdict')} " +
(f"spread {dayb.get('spread')} per_offset {dayb.get('per_offset')}"
if fam["price_only"] else f"({dayb.get('reason','')[:60]}...)"))
_nan = float("nan")
print(f" ancora Sh canonica {anch.get('canonical', _nan):+.3f} "
f"(pctl {anch.get('canonical_pctl')}) "
f"MEDIANA {anch.get('median', _nan):+.3f} "
f"banda [{anch.get('lo', _nan):+.3f},{anch.get('hi', _nan):+.3f}] "
f"frac>0 {anch.get('frac_positive')}")
print(f" uplift canonico {anch.get('uplift_canonical')} MEDIANA "
f"{anch.get('uplift_median')} frac>0 {anch.get('uplift_frac_positive')}")
print(f" de-levering {deln.get('note')}")
print(f" $600 " + " ".join(
f"{a}: mod {v['modeled']:+.2f} -> real {v['realistic']:+.2f} (haircut {v['haircut']:+.2f}, "
f"{v['n_trades']} trade)" for a, v in smal.items()))
print(f" implausible {impl.get('implausible')} {impl.get('reasons')}")
if actw.get("frac_active", 1.0) < 0.9:
print(f" attive {actw}")
print(f" [{time.time() - t1:.1f}s]\n")
# ---- DSR DI SCREEN: i trial sono TUTTE le celle di TUTTE le famiglie -------------
print("### DSR DI SCREEN (il conto onesto per un setaccio: scelgo il meglio fra "
f"{len(pooled_sh)} celle, non fra 24)")
for r in results:
if r.get("chosen") is None:
continue
d, s0 = A.deflated_sharpe(r["full_sharpe"], pooled_sh, r["C"])
r["dsr_screen"] = round(d, 3) if np.isfinite(d) else None
print(f" {r['code']:<6} Sh {r['full_sharpe']:+.3f} DSR famiglia {r['dsr_family']} "
f"DSR screen {r['dsr_screen']} (null max atteso {s0:.3f})")
print()
deep_dive(results)
# ---- TABELLA FINALE --------------------------------------------------------------
def killer(r) -> str:
if r.get("chosen") is None:
return "nessuna cella valutabile"
why = []
if not r["causality"]["ok"]:
why.append("CAUSALITY")
if r["verdict"] != "ADDS":
why.append(f"marginal={r['verdict']}")
if not r["has_insample_edge"]:
why.append("no in-sample edge (<0.5)")
if not r["robust_oos"]:
why.append("robust_oos")
if r["is_hedge"]:
why.append("is_hedge")
if not r["dsr_pass"]:
why.append(f"DSR fam {r['dsr_family']}")
if r.get("dsr_screen") is not None and r["dsr_screen"] < 0.95:
why.append(f"DSR screen {r['dsr_screen']}")
if r["dayb"].get("verdict") == "ARTIFACT-RISK":
why.append("confine ARTIFACT-RISK")
um = r["anchor"].get("uplift_median")
if um is not None and np.isfinite(um) and um <= 0:
why.append("ancora: uplift mediano <=0")
if r["delever"].get("claims_less_dd") and not r["delever"].get("passes"):
why.append("null de-levering")
if r["implausible"].get("implausible"):
why.append("implausible_sharpe")
return " + ".join(why) if why else "-- nessun gate lo uccide --"
print("### TABELLA — ORTHO-SCREEN 2026-08-22 (candidato | trial | ShFULL | hold-out | "
"corr->TP01 | marginale | DSR fam / screen | esito)")
hdr = (f"{'cand':<7}{'trial':>6}{'ShFULL':>8}{'hold':>7}{'maxDD':>8}{'CAGR':>8}"
f"{'corr':>7} {'marginale':<10}{'DSRfam':>7}{'DSRscr':>8} esito / gate che l'ha ucciso")
print(hdr); print("-" * len(hdr))
for r in results:
if r.get("chosen") is None:
print(f"{r['code']:<7}{r['n_cells']:>6}{'--':>8}{'--':>7}{'--':>8}{'--':>8}{'--':>7} "
f"{'--':<10}{'--':>7}{'--':>8} SCARTATO — {killer(r)}")
continue
ok = (r["verdict"] == "ADDS" and r["dsr_pass"] and (r.get("dsr_screen") or 0) >= 0.95
and r["causality"]["ok"])
esito = "CANDIDATO" if ok else "SCARTATO"
print(f"{r['code']:<7}{r['n_cells']:>6}{r['full_sharpe']:>+8.3f}{r['hold_sharpe']:>+7.2f}"
f"{r['maxdd']:>8.1%}{r['cagr']:>+8.1%}{r['corr']:>7.2f} {r['verdict']:<10}"
f"{str(r['dsr_family']):>7}{str(r.get('dsr_screen')):>8} {esito} — {killer(r)}")
print()
print("### FAMIGLIE NON TESTATE (l'elenco di cio' che non ho guardato fa parte del risultato)")
for line in NOT_TESTED:
print(" - " + line)
print(f"\n[totale {time.time() - t0:.1f}s]")
def mk_mom(tf="1d", H=5, shape="lin"):
"""CONTROLLO per TACC: il LIVELLO del momentum a orizzonte H, stessa pipeline
(z-score 365g -> clip -> vol_target). Serve a rispondere alla domanda della famiglia:
l'ACCELERAZIONE aggiunge qualcosa al LIVELLO, o e' un momentum corto travestito?"""
def target(df, asset):
c = df["close"].values.astype(float)
n = len(c)
m = np.full(n, np.nan)
m[H:] = c[H:] / c[:-H] - 1.0
u = np.clip(roll_z(m, 365) / 2.0, -1.0, 1.0)
return _sized(_shape(u, shape), df)
return target
MOM_CONTROL_GRID = _grid(H=[5, 20, 60], shape=["lin", "sign"]) # 6 celle di CONTROLLO
def deep_dive(results) -> None:
"""APPROFONDIMENTO sull'unica famiglia arrivata a marginal=ADDS.
Due domande, entrambe misurate:
(a) QUALE sotto-gate di robust_oos ha fallito (clean-year? jackknife? multi-cut?);
(b) l'accelerazione e' distinta dal LIVELLO del momentum corto, o e' quello travestito?
-> 6 celle di CONTROLLO dichiarate (H x shape), che si SOMMANO ai 168 trial."""
adds = [r for r in results if r.get("chosen") is not None and r["verdict"] == "ADDS"]
if not adds:
print("### APPROFONDIMENTO — nessuna famiglia a marginal=ADDS, niente da approfondire\n")
return
r = max(adds, key=lambda x: x["uplift_hold"] or -9)
print(f"### APPROFONDIMENTO — {r['code']} (l'unica a marginal=ADDS)")
print(f" perche' robust_oos=False: clean_year_uplift={r['_m'].get('clean_year_uplift')} "
f"(serve >0.02) jackknife_min_uplift={r['_m'].get('jackknife_min_uplift')} (serve >0)")
print(f" multicut (uplift a ogni taglio d'anno): {r['multicut']}")
print(f" uplift TP01-up {r['_m'].get('uplift_tp01_up')} / TP01-down {r['_m'].get('uplift_tp01_down')} "
f"| hedge_yearly_corr {r['_m'].get('hedge_yearly_corr')} | alpha/anno {r['_m'].get('alpha_ann')}")
print(" --- accelerazione vs LIVELLO del momentum (6 celle di controllo) ---")
B = A.tp01_baseline_daily()
C = r["C"]
best = None
rows_ctl = []
for p in MOM_CONTROL_GRID:
fnm = mk_mom(tf="1d", **p)
Cm = A.candidate_daily(fnm, tf="1d")
J = pd.concat({"A": C, "M": Cm}, axis=1, join="inner").dropna()
JB = pd.concat({"B": B, "M": Cm}, axis=1, join="inner").dropna()
up = A._sh((1 - W_BLEND) * JB["B"] + W_BLEND * JB["M"]) - A._sh(JB["B"])
row = dict(params=p, sharpe=round(A._sh(Cm), 3),
insample=round(A._sh(Cm[Cm.index < A.HOLDOUT]), 3),
hold=round(A._sh(Cm[Cm.index >= A.HOLDOUT]), 3),
corr_to_acc=round(float(J["A"].corr(J["M"])), 3),
corr_to_tp01=round(float(JB["B"].corr(JB["M"])), 3),
uplift_w25_full=round(up, 3))
print(f" MOM {str(p):<28} Sh {row['sharpe']:+.3f} IS {row['insample']:+.3f} "
f"hold {row['hold']:+.3f} corr->{r['code']} {row['corr_to_acc']:+.3f} "
f"corr->TP01 {row['corr_to_tp01']:+.3f} uplift {row['uplift_w25_full']:+.3f}")
rows_ctl.append(row)
if best is None or row["corr_to_acc"] > best["corr_to_acc"]:
best = row
print(f" livello piu' vicino all'accelerazione: {best['params']} corr {best['corr_to_acc']:+.3f}")
print(" lettura: corr alta col LIVELLO => l'accelerazione non e' una dimensione nuova; "
"corr bassa => e' distinta ma resta da spiegare perche' non sopravvive al DSR.")
bh = max(rows_ctl, key=lambda x: x["hold"])
bi = max(rows_ctl, key=lambda x: x["insample"])
print(f" TRAPPOLA DICHIARATA: la cella di controllo migliore SULL'HOLD-OUT e' {bh['params']} "
f"(hold {bh['hold']:+.3f}); quella scelta IN-SAMPLE-ONLY e' {bi['params']} "
f"(IS {bi['insample']:+.3f}, hold {bi['hold']:+.3f}). Se coincidono, il controllo e'"
" selezionabile; se no, guardare la colonna hold e' selezione-sull'hold-out (trappola 3)"
" e NON promuove nulla.\n")
NOT_TESTED = [
"catena opzioni / vol term-structure: 3-5 mesi di storia -> non esiste hold-out, al massimo "
"un LEAD; e altri agenti dell'ondata ci stanno sopra.",
"funding (carry, time-series, cross-sectional): filone dichiarato CHIUSO su 3 lati.",
"calendario in ogni forma (weekend 375 trial, expiry, stagionalita', event-clock, "
"ora-del-giorno): famiglia dichiarata SATURA.",
"mean-reversion pura / fade: morta anche a fee zero. Qui rientra SOLO come branch dichiarato "
"di XTAIL e VPX, per misurarla dentro un conditioner nuovo.",
"microstruttura 5m/15m: morta per fee, e una sweep su 5m sfonda il budget macchina.",
"cross-sectional sui 51 alt HL: territorio XS01, e l'espansione d'universo e' gia' refutata. "
"Qui i 51 alt entrano solo come AGGREGATO time-series (XDISP).",
"equity / GTAA / MAT01: GTAA01 non e' deployabile (PRIIPs) e MAT01 e' refutato.",
"macro regime gate: ridondante col trend (corr 0.989).",
"CRT / ICT / Elliott / Fibonacci / Albimarini: 3 ondate di refutazione.",
"gamma scalping long-vol e overlay DD su VRP01 (5/5 refutati).",
"SOL come terza gamba: decisione dell'operatore del 22/08.",
]
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink