"""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()