feat(XS01): + gate di dispersione (p30) — portafoglio FULL 1.48->1.55, HOLD 1.06->1.55
Momentum cross-sectional vive nella dispersione; gate: entra solo se la dispersione cross-section del momentum supera il percentile ESPANDENTE causale (altrimenti flat). Plateau robusto p15-p35 (non knife-edge: il crollo a p40+ e' over-gating); scelto p30. XS01 standalone FULL 1.10->1.50, HOLD 1.03->1.71, DD 14%->10.8%. Portafoglio TP01 70+XS 30: FULL 1.48->1.55, HOLD 1.06->1.55, DD 4.6%->4.4%. Il gate alza SIA FULL SIA hold-out (tiene XS attivo nei regimi dispersi, flat nei bull compatti; causale). E' il concetto del vecchio XS01. sleeves.XS_CFG disp_pct=30; engine _xsec_returns gatea su dispersione. 12 test ok. Diario 2026-06-19-xsec-dispgate.md. Affinamenti del segnale (blend+gate) > espansione universo. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,101 @@
|
||||
"""AFFINAMENTO XS01 — GATE DI DISPERSIONE.
|
||||
|
||||
Il momentum cross-sectional vive nella DISPERSIONE (winners/losers distanti). In regime compatto
|
||||
(tutti gli asset si muovono insieme) non ha segnale -> churn/rumore. Gate: entra SOLO se la
|
||||
dispersione cross-section del momentum supera una soglia CAUSALE (percentile espandente della
|
||||
dispersione passata); altrimenti flat. Sul blend [30,90] dei 19 major. Sweep soglia + contributo.
|
||||
|
||||
uv run python scripts/portfolio/xsec_dispgate.py
|
||||
"""
|
||||
from __future__ import annotations
|
||||
import sys
|
||||
from pathlib import Path
|
||||
PROJECT_ROOT = Path(__file__).resolve().parents[2]
|
||||
sys.path.insert(0, str(PROJECT_ROOT))
|
||||
import numpy as np, pandas as pd
|
||||
from src.portfolio.portfolio import to_daily, metrics, HOLDOUT
|
||||
from src.portfolio.sleeves import tp01_sleeve, XS_UNIVERSE
|
||||
|
||||
RAW = PROJECT_ROOT / "data" / "raw"
|
||||
FEE = 0.001
|
||||
LOOKBACKS = (30, 90); H = 10; K = 5; TV = 0.20
|
||||
|
||||
|
||||
def load_majors():
|
||||
cols = {}
|
||||
for sym in XS_UNIVERSE:
|
||||
p = RAW / f"hl_{sym.lower()}_1d.parquet"
|
||||
if p.exists():
|
||||
d = pd.read_parquet(p)
|
||||
cols[sym] = pd.Series(d["close"].values.astype(float), index=pd.to_datetime(d["timestamp"], unit="ms", utc=True))
|
||||
return pd.concat(cols, axis=1, join="inner").sort_index().dropna()
|
||||
|
||||
|
||||
def xs_gated(C, disp_pct=0, min_hist=20):
|
||||
px = C.values; n, A = px.shape
|
||||
dret = np.vstack([np.zeros(A), px[1:] / px[:-1] - 1.0])
|
||||
mlb = max(LOOKBACKS)
|
||||
# dispersione del momentum a ogni barra: media (su lookback) della std cross-section di ret_L
|
||||
disp = np.full(n, np.nan)
|
||||
for i in range(mlb, n):
|
||||
acc = 0.0; c = 0
|
||||
for L in LOOKBACKS:
|
||||
acc += (px[i] / px[i - L] - 1.0).std(); c += 1
|
||||
disp[i] = acc / c
|
||||
W = np.zeros((n, A)); w = np.zeros(A)
|
||||
hist = []
|
||||
gated_flat = 0; total = 0
|
||||
for i in range(n):
|
||||
if i >= mlb and i % H == 0:
|
||||
thr = np.percentile(hist, disp_pct) if (disp_pct > 0 and len(hist) >= min_hist) else -np.inf
|
||||
total += 1
|
||||
if disp[i] >= thr:
|
||||
score = np.zeros(A)
|
||||
for L in LOOKBACKS:
|
||||
rL = px[i] / px[i - L] - 1.0; sd = rL.std()
|
||||
if sd > 0:
|
||||
score += (rL - rL.mean()) / sd
|
||||
order = np.argsort(score); w = np.zeros(A); lo, hi = order[:K], order[-K:]
|
||||
w[hi] = 0.5 / K; w[lo] = -0.5 / K
|
||||
else:
|
||||
w = np.zeros(A); gated_flat += 1
|
||||
hist.append(disp[i])
|
||||
W[i] = w
|
||||
gross = np.zeros(n); gross[1:] = np.sum(W[:-1] * dret[1:], axis=1)
|
||||
turn = np.zeros(n); turn[0] = np.abs(W[0]).sum(); turn[1:] = np.abs(np.diff(W, axis=0)).sum(axis=1)
|
||||
s = pd.Series(gross - turn * (FEE / 2.0), index=C.index)
|
||||
rv = s.rolling(30, min_periods=15).std().shift(1) * np.sqrt(365.25)
|
||||
scale = np.clip(np.nan_to_num(TV / rv.replace(0, np.nan).values, nan=0.0), 0, 3.0)
|
||||
return to_daily(pd.Series(s.values * scale, index=C.index)), (gated_flat / total if total else 0)
|
||||
|
||||
|
||||
def main():
|
||||
C = load_majors(); tp = tp01_sleeve().daily()
|
||||
print("=" * 92)
|
||||
print(f" AFFINAMENTO XS01 — gate di dispersione (blend [30,90], 19 major, {len(C)}g)")
|
||||
print("=" * 92)
|
||||
print(f" {'soglia pctile':<16}{'FULL':>7}{'OOS25':>7}{'DD%':>6}{'anni+':>7}{'corrTP':>8}{'%flat':>8}")
|
||||
res = {}
|
||||
for p in (0, 30, 40, 50, 60, 70):
|
||||
d, flat = xs_gated(C, p)
|
||||
f = metrics(d); o = metrics(d[d.index >= HOLDOUT])
|
||||
yr = [float((1 + g).prod() - 1) for _, g in d.groupby(d.index.year)]
|
||||
pct = sum(v > 0 for v in yr) / len(yr) if yr else 0
|
||||
corr = float(pd.concat({"a": tp, "b": d}, axis=1, join="inner").dropna().corr().iloc[0, 1])
|
||||
res[p] = (d, f, o, pct, corr)
|
||||
lab = "0 (no gate)" if p == 0 else f"p{p}"
|
||||
print(f" {lab:<16}{f['sharpe']:>7.2f}{o['sharpe']:>7.2f}{f['maxdd']*100:>6.0f}{pct*100:>6.0f}%{corr:>+8.2f}{flat*100:>7.0f}%")
|
||||
|
||||
print("\n CONTRIBUTO al portafoglio (TP01 70 + XS 30, finestra comune):")
|
||||
for p in (0, 40, 50, 60):
|
||||
d = res[p][0]
|
||||
J = pd.concat({"tp": tp, "xs": d}, axis=1, join="inner").dropna()
|
||||
comb = 0.7 * J["tp"] + 0.3 * J["xs"]
|
||||
cf, ch = metrics(comb), metrics(comb[comb.index >= HOLDOUT])
|
||||
lab = "no gate (attuale)" if p == 0 else f"gate p{p}"
|
||||
print(f" {lab:<18} FULL Sh {cf['sharpe']:.2f} DD {cf['maxdd']*100:.0f}% | HOLD Sh {ch['sharpe']:.2f}")
|
||||
print("\n -> promuovere il gate se migliora Sharpe/DD/robustezza E il contributo. Sennò no-gate resta.")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user