Adriano Dal Pastro
4c1961d77b
3 agenti web (~50 fonti) + 3 ondate empiriche in una notte. Convergenza indipendente ricerca<->empirica: il factor zoo crypto si riduce a liquidity/momentum (gia' nel book). XS su HL: lead-lag all-others e' matematicamente l'inverso del momentum proprio (check di degenerazione da riusare); MAX segno instabile L7<->L30. HLP probe: crash-long reale (corr book -0.19, +10.4% nel crash ott-25) ma decay 79->12%/anno + coda JELLY fuori serie -> ALLOCATION-WATCH. Pista nuova prossima ondata: Polymarket<->Deribit (gap 11pp, arXiv 2606.19517, spec-mismatch da riconciliare prima di credere). Book INVARIATO. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
284 lines
13 KiB
Python
284 lines
13 KiB
Python
"""r0724_xs — LEAD-LAG cross-crypto (Guo/Sang/Tu/Wang JEDC 2024) + SALIENCE/MAX (Cai&Zhao
|
||
JBF 2024) sui 19 major Hyperliquid certificati (1d, 2024-2026).
|
||
|
||
TESI (2026-07-24). Due famiglie cross-sectional dalla letteratura accademica, mai meccanizzate
|
||
qui, sullo stesso universo/harness di XS01:
|
||
|
||
FAMIGLIA A — LEAD-LAG "cross-cryptocurrency return predictability" (diffusione lenta
|
||
dell'informazione: i ritardi degli ALTRI coin predicono il focale). Segnale
|
||
S_i(t) = media equal-weight dei ritorni a k giorni degli ALTRI asset (mai il proprio);
|
||
variante predictor-set = solo large-cap (BTC/ETH/SOL/BNB/XRP, leave-one-out se il focale
|
||
e' large). Book L/S top-5 / bottom-5 per S_i, hold H, come XS01.
|
||
Griglia: k in {3,7} x predictor in {all-others, large5} x H in {3,7} = 8 celle.
|
||
⚠️ IDENTITA' MATEMATICA (dichiarata PRIMA di guardare i numeri): con predictor
|
||
all-others equal-weight, S_i = (SommaTotale − r_i)/(A−1) e' AFFINE DECRESCENTE nel
|
||
ritorno proprio r_i → il rank cross-section di S e' ESATTAMENTE il rank inverso del
|
||
momentum proprio a k giorni: la cella "all-others" E' la short-term reversal (gia'
|
||
negativa nella griglia rev di xsec_research). La verifico empiricamente (rank-corr −1).
|
||
⚠️ DEGENERAZIONE large5: i 14 asset non-large hanno segnale IDENTICO (media dei 5
|
||
large) → la composizione del book fra i pari e' decisa dal tie-break dell'argsort
|
||
(stabile = ordine colonne), economicamente arbitraria. Dichiarata, non nascosta.
|
||
|
||
FAMIGLIA B — SALIENCE/MAX (lottery demand: i coin col massimo ritorno giornaliero
|
||
recente vengono sovracomprati → overpriced). MAX_i(t) = max ritorno 1d negli ultimi L
|
||
giorni; book LONG bottom-quintile MAX (k=4) / SHORT top-quintile, hold H.
|
||
Griglia: L in {7,30} x H in {7,10} = 4 celle.
|
||
|
||
GATE (CLAUDE.md, metodologia obbligatoria):
|
||
1. CAUSALE: segnale da close<=t, peso tenuto da t+1 (l'engine shifta W[i-1]*dret[i]).
|
||
2. NETTO fee 0.10% RT per gamba (FEE=0.001, addebitata FEE/2 per lato sul turnover,
|
||
identica a xsec_research/sleeves) + probe fee-zero sulla cella scelta (morte-per-fee
|
||
vs assenza di edge lordo).
|
||
3. Selezione cella IN-SAMPLE-ONLY (pre-2025-01-01), poi hold-out bloccato.
|
||
4. DEFLATED Sharpe (Bailey & Lopez de Prado, altlib) su TUTTE le celle della famiglia.
|
||
5. RIDONDANZA: corr giornaliera vs XS01 (sleeves._xsec_returns) e TP01
|
||
(altlib.tp01_baseline_daily) + uplift del blend 0.75*XS01+0.25*candidato (full e
|
||
hold-out). corr>0.5 a XS01 o uplift negativo -> REDUNDANT.
|
||
6. BANDA DI FASE del ciclo H sulla cella scelta (lezione anchor-luck 2026-07-02: i
|
||
numeri di strategie a ribilanciamento ancorato si citano con la banda).
|
||
|
||
CAVEAT IMMUTABILI: storia HL nativa ~2.5 anni (join 19 major = 2024-01-01+, 0 barre
|
||
sintetiche/vol=0 — verificato a runtime); IS = solo il 2024 (~330 giorni utili, ~48
|
||
ribilanci a H=7 → small-n); book L/S a 8-10 gambe NON eseguibile a $600 → STAT-MODE
|
||
come XS01 (eventuale esito positivo = forward-monitor, non deploy).
|
||
|
||
uv run python scripts/research/r0724_xs_leadlag_max.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 / "scripts" / "research" / "alt"))
|
||
sys.path.insert(0, str(PROJECT_ROOT))
|
||
|
||
import numpy as np
|
||
import pandas as pd
|
||
|
||
from altlib import deflated_sharpe, tp01_baseline_daily # noqa: E402
|
||
from src.portfolio.portfolio import HOLDOUT, metrics, to_daily # noqa: E402
|
||
from src.portfolio.sleeves import XS_UNIVERSE, _xsec_returns # noqa: E402
|
||
|
||
RAW = PROJECT_ROOT / "data" / "raw"
|
||
FEE = 0.001 # 0.10% RT per gamba (come xsec_research: FEE/2 per lato sul turnover)
|
||
TV = 0.20 # vol-target come XS01
|
||
WARMUP = 30 # warmup comune a tutte le celle (max lookback della griglia) -> stessi giorni
|
||
DPY = 365.25
|
||
LARGE5 = ["BTC", "ETH", "SOL", "BNB", "XRP"]
|
||
|
||
|
||
def load_universe():
|
||
"""Chiusure (e guardia volume) dei 19 major XS01. Inner-join = 2024+ nativo (0 backfill)."""
|
||
cols, vols = {}, {}
|
||
for s in XS_UNIVERSE:
|
||
d = pd.read_parquet(RAW / f"hl_{s.lower()}_1d.parquet")
|
||
idx = pd.to_datetime(d["timestamp"], unit="ms", utc=True)
|
||
cols[s] = pd.Series(d["close"].values.astype(float), index=idx)
|
||
vols[s] = pd.Series(d["volume"].values.astype(float), index=idx)
|
||
C = pd.concat(cols, axis=1, join="inner").sort_index().dropna()
|
||
V = pd.concat(vols, axis=1, join="inner").sort_index().reindex(C.index)
|
||
nz = int((V.values == 0).sum())
|
||
assert nz == 0, f"barre vol=0 nella finestra joined ({nz}): possibile backfill sintetico"
|
||
return C
|
||
|
||
|
||
def kret_matrix(px: np.ndarray, k: int) -> np.ndarray:
|
||
"""Ritorno a k giorni fino a close[t] (causale). NaN nelle prime k righe."""
|
||
R = np.full_like(px, np.nan, dtype=float)
|
||
R[k:] = px[k:] / px[:-k] - 1.0
|
||
return R
|
||
|
||
|
||
def sig_leadlag(px: np.ndarray, cols: list[str], k: int, pred: str) -> np.ndarray:
|
||
"""S_i(t) = media dei ritorni k-giorni degli ALTRI asset (mai il proprio).
|
||
pred='all' : tutti gli altri 18 (⚠️ affine decrescente nel proprio ritorno).
|
||
pred='large5': solo BTC/ETH/SOL/BNB/XRP (leave-one-out se il focale e' large)."""
|
||
R = kret_matrix(px, k)
|
||
n, A = R.shape
|
||
S = np.full((n, A), np.nan)
|
||
if pred == "all":
|
||
tot = R.sum(axis=1)
|
||
for i in range(A):
|
||
S[:, i] = (tot - R[:, i]) / (A - 1)
|
||
else:
|
||
li = [cols.index(s) for s in LARGE5]
|
||
totL = R[:, li].sum(axis=1)
|
||
for i in range(A):
|
||
if i in li:
|
||
S[:, i] = (totL - R[:, i]) / (len(li) - 1)
|
||
else:
|
||
S[:, i] = totL / len(li)
|
||
return S
|
||
|
||
|
||
def sig_neg_max(px: np.ndarray, L: int) -> np.ndarray:
|
||
"""score = −MAX_i(t) (max ritorno 1d negli ultimi L giorni): rank alto = low-MAX = LONG."""
|
||
n, A = px.shape
|
||
dret = np.vstack([np.full((1, A), np.nan), px[1:] / px[:-1] - 1.0])
|
||
mx = pd.DataFrame(dret).rolling(L, min_periods=L).max().values
|
||
return -mx
|
||
|
||
|
||
def xs_book(C: pd.DataFrame, S: np.ndarray, H: int, k: int,
|
||
phase: int = 0, fee: float = FEE) -> tuple[pd.Series, dict]:
|
||
"""Book L/S market-neutral da matrice di segnale (rank a close[i], tenuto da i+1),
|
||
fee sul turnover, vol-target 20% causale — convenzioni identiche a XS01."""
|
||
px = C.values
|
||
n, A = px.shape
|
||
dret = np.vstack([np.zeros((1, A)), px[1:] / px[:-1] - 1.0])
|
||
W = np.zeros((n, A))
|
||
w = np.zeros(A)
|
||
nreb = nreb_is = 0
|
||
is_mask = C.index < HOLDOUT
|
||
for i in range(n):
|
||
if i >= WARMUP and (i - WARMUP - phase) % H == 0 and np.isfinite(S[i]).all():
|
||
order = np.argsort(S[i], kind="stable")
|
||
w = np.zeros(A)
|
||
w[order[-k:]] = 0.5 / k
|
||
w[order[:k]] = -0.5 / k
|
||
nreb += 1
|
||
nreb_is += int(is_mask[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)
|
||
net = gross - turn * (fee / 2.0)
|
||
s = pd.Series(net, index=C.index)
|
||
rv = s.rolling(30, min_periods=15).std().shift(1) * np.sqrt(DPY)
|
||
scale = np.clip(np.nan_to_num(TV / rv.replace(0, np.nan).values, nan=0.0), 0, 3.0)
|
||
out = pd.Series(s.values * scale, index=C.index)
|
||
diag = dict(time_in_market=float((W != 0).any(axis=1).mean()),
|
||
turnover_yr=float(turn.sum() / (n / DPY)),
|
||
n_reb=nreb, n_reb_is=nreb_is)
|
||
return out, diag
|
||
|
||
|
||
def cell_row(d: pd.Series) -> dict:
|
||
f = metrics(d)
|
||
i = metrics(d[d.index < HOLDOUT])
|
||
h = metrics(d[d.index >= HOLDOUT])
|
||
return dict(full=f, ins=i, hold=h)
|
||
|
||
|
||
def redundancy(cand: pd.Series) -> dict:
|
||
"""Corr vs XS01/TP01 + uplift del blend 0.75*XS01+0.25*cand (finestra comune)."""
|
||
xs = to_daily(_xsec_returns())
|
||
tp = tp01_baseline_daily()
|
||
Jx = pd.concat({"xs": xs, "c": cand}, axis=1, join="inner").dropna()
|
||
Jt = pd.concat({"tp": tp, "c": cand}, axis=1, join="inner").dropna()
|
||
corr_xs = float(Jx["xs"].corr(Jx["c"])) if len(Jx) > 5 else float("nan")
|
||
corr_tp = float(Jt["tp"].corr(Jt["c"])) if len(Jt) > 5 else float("nan")
|
||
blend = 0.75 * Jx["xs"] + 0.25 * Jx["c"]
|
||
up_f = metrics(blend)["sharpe"] - metrics(Jx["xs"])["sharpe"]
|
||
bh = blend[blend.index >= HOLDOUT]
|
||
xh = Jx["xs"][Jx["xs"].index >= HOLDOUT]
|
||
up_h = metrics(bh)["sharpe"] - metrics(xh)["sharpe"]
|
||
return dict(corr_xs=corr_xs, corr_tp=corr_tp, uplift_full=up_f, uplift_hold=up_h,
|
||
xs_full=metrics(Jx["xs"])["sharpe"], xs_hold=metrics(xh)["sharpe"])
|
||
|
||
|
||
def phase_band(C, S, H, k) -> list[float]:
|
||
"""Sharpe FULL a ogni fase del ciclo H (lezione anchor-luck: cita la banda, non la fase 0)."""
|
||
return [metrics(xs_book(C, S, H, k, phase=p)[0])["sharpe"] for p in range(H)]
|
||
|
||
|
||
def run_family(name: str, C: pd.DataFrame, cells: list[tuple[str, np.ndarray, int, int]]):
|
||
"""cells = [(tag, S, H, k)]. Selezione IS-only, DSR su tutte le celle, ridondanza + fasi."""
|
||
print("\n" + "=" * 100)
|
||
print(f" FAMIGLIA {name}")
|
||
print("=" * 100)
|
||
print(f" {'cella':<30}{'IS Sh':>8}{'FULL Sh':>9}{'HOLD Sh':>9}{'DD%':>7}{'TiM':>6}{'to/yr':>7}{'reb':>5}")
|
||
rows = []
|
||
for tag, S, H, k in cells:
|
||
d, diag = xs_book(C, S, H, k)
|
||
m = cell_row(d)
|
||
rows.append((tag, S, H, k, d, diag, m))
|
||
print(f" {tag:<30}{m['ins']['sharpe']:>8.2f}{m['full']['sharpe']:>9.2f}"
|
||
f"{m['hold']['sharpe']:>9.2f}{m['full']['maxdd'] * 100:>7.1f}"
|
||
f"{diag['time_in_market']:>6.2f}{diag['turnover_yr']:>7.1f}{diag['n_reb']:>5}")
|
||
# selezione IN-SAMPLE-ONLY
|
||
best = max(rows, key=lambda r: r[6]["ins"]["sharpe"])
|
||
tag, S, H, k, d, diag, m = best
|
||
all_full = [r[6]["full"]["sharpe"] for r in rows]
|
||
dsr, sr0 = deflated_sharpe(m["full"]["sharpe"], all_full, d)
|
||
red = redundancy(d)
|
||
d0, _ = xs_book(C, S, H, k, fee=0.0)
|
||
m0 = cell_row(d0)
|
||
band = phase_band(C, S, H, k)
|
||
print(f"\n CELLA IS-BEST: {tag} (scelta sul solo IS Sharpe {m['ins']['sharpe']:.2f})")
|
||
print(f" FULL Sh {m['full']['sharpe']:.2f} DD {m['full']['maxdd'] * 100:.1f}% | "
|
||
f"HOLD Sh {m['hold']['sharpe']:.2f} DD {m['hold']['maxdd'] * 100:.1f}%")
|
||
print(f" deflated-Sharpe (N={len(rows)} celle): DSR={dsr:.3f} (null-max ~{sr0:.2f}) "
|
||
f"{'PASS' if dsr >= 0.95 else 'FAIL'}")
|
||
print(f" fee-zero probe: IS {m0['ins']['sharpe']:.2f} / FULL {m0['full']['sharpe']:.2f} / "
|
||
f"HOLD {m0['hold']['sharpe']:.2f} (edge lordo vs morte-per-fee)")
|
||
print(f" corr vs XS01 {red['corr_xs']:+.2f} | corr vs TP01 {red['corr_tp']:+.2f}")
|
||
print(f" blend 0.75*XS01+0.25*cand: uplift FULL {red['uplift_full']:+.2f} "
|
||
f"(XS01 solo {red['xs_full']:.2f}) | HOLD {red['uplift_hold']:+.2f} (XS01 solo {red['xs_hold']:.2f})")
|
||
print(f" banda di fase H={H} (Sh FULL): min {min(band):.2f} / med {np.median(band):.2f} / "
|
||
f"max {max(band):.2f} [fase0 = {band[0]:.2f}]")
|
||
print(f" ribilanci: {diag['n_reb']} totali, {diag['n_reb_is']} in IS"
|
||
+ (" ⚠️ SMALL-N (<50 in IS)" if diag['n_reb_is'] < 50 else ""))
|
||
return dict(tag=tag, m=m, dsr=dsr, red=red, band=band, diag=diag, fee0=m0)
|
||
|
||
|
||
def main():
|
||
C = load_universe()
|
||
cols = list(C.columns)
|
||
px = C.values
|
||
print("=" * 100)
|
||
print(f" r0724 LEAD-LAG + MAX — {len(cols)} major HL, {len(C)} giorni "
|
||
f"[{C.index[0].date()} -> {C.index[-1].date()}], fee {FEE * 100:.2f}% RT/gamba, "
|
||
f"holdout {HOLDOUT.date()}, IS = {int((C.index < HOLDOUT).sum())} giorni")
|
||
print("=" * 100)
|
||
|
||
# verifica empirica dell'identita' (Famiglia A, all-others): rank-corr(S, r_proprio) = -1
|
||
Sa = sig_leadlag(px, cols, 7, "all")
|
||
Ra = kret_matrix(px, 7)
|
||
i = len(px) // 2
|
||
rc = pd.Series(Sa[i]).corr(pd.Series(Ra[i]), method="spearman")
|
||
print(f"\n [check identita' A/all-others] rank-corr(S_i, ritorno proprio 7g) al giorno "
|
||
f"{C.index[i].date()}: {rc:+.3f} (atteso -1: la cella 'all' E' reversal del proprio momentum)")
|
||
|
||
# FAMIGLIA A — lead-lag: k x predictor x H = 8 celle (k lati book = 5 come XS01)
|
||
cells_a = []
|
||
for kk in (3, 7):
|
||
for pred in ("all", "large5"):
|
||
S = sig_leadlag(px, cols, kk, pred)
|
||
for H in (3, 7):
|
||
cells_a.append((f"LL k{kk} {pred:<7} H{H}", S, H, 5))
|
||
res_a = run_family("A — LEAD-LAG cross-crypto (8 celle)", C, cells_a)
|
||
|
||
# FAMIGLIA B — MAX/salience: L x H = 4 celle (quintile: 4 gambe/lato)
|
||
cells_b = []
|
||
for L in (7, 30):
|
||
S = sig_neg_max(px, L)
|
||
for H in (7, 10):
|
||
cells_b.append((f"MAX L{L} H{H}", S, H, 4))
|
||
res_b = run_family("B — SALIENCE/MAX lottery-demand (4 celle)", C, cells_b)
|
||
|
||
print("\n" + "=" * 100)
|
||
print(" VERDETTO (bar: DSR>=0.95, corr XS01<0.5, uplift blend>0, hold-out>0, small-n onesto)")
|
||
print("=" * 100)
|
||
for nm, r in (("A LEAD-LAG", res_a), ("B MAX", res_b)):
|
||
flags = []
|
||
if r["dsr"] < 0.95:
|
||
flags.append(f"DSR {r['dsr']:.2f}<0.95")
|
||
if abs(r["red"]["corr_xs"]) > 0.5:
|
||
flags.append(f"corrXS01 {r['red']['corr_xs']:+.2f}")
|
||
if r["red"]["uplift_full"] < 0 or r["red"]["uplift_hold"] < 0:
|
||
flags.append("uplift blend negativo")
|
||
if r["m"]["hold"]["sharpe"] <= 0:
|
||
flags.append("hold-out <=0")
|
||
if r["diag"]["n_reb_is"] < 50:
|
||
flags.append("small-n IS")
|
||
print(f" {nm}: cella {r['tag']} — " + ("; ".join(flags) if flags else "tutti i gate passati"))
|
||
print("\n NB: STAT-MODE (8-10 gambe, non eseguibile a $600); storia ~2.5 anni; IS = solo 2024.")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|