PythagorasGoal/scripts/analysis/dispersion_edges/index_comp_disp.py

"""FAMIGLIA index_comp_disp (W=24) — dispersion trading REALIZZATO.

Idea: l'indice EW vs le componenti. Quando la dispersione cross-sectional rolling
(disp_24) e' ALTA, le componenti idiosincratiche estreme (rel_A = ret_A - ret_idx)
tendono a RIENTRARE verso l'indice -> fade della componente idiosincratica estrema:
  - se rel_A e' molto positivo (A ha sovraperformato l'indice oltre soglia) -> SHORT A
  - se rel_A e' molto negativo (A ha sottoperformato l'indice oltre soglia) -> LONG A
condizionato a disp_24 sopra una soglia (regime di alta dispersione).

CAUSALE: la decisione a close[i] usa SOLO feature note a i:
  - rel_A[i]   = log-ret di A meno log-ret indice nella barra [i-1->i] (nota a close[i])
  - disp_24[i] = media rolling 24 della disp cross-sectional fino a i (nota a close[i])
Ingresso eseguibile a close[i]. Niente uso di i+1 nella decisione.

Per rendere le soglie comparabili fra asset/tempo, rel_A si normalizza con la sua
deviazione standard rolling CAUSALE (rolling 168h, shiftata di 1 per non includere i).
disp_24 si normalizza col suo quantile rolling causale (percentile rolling).

Exit: time-stop max_bars (con TP/SL ATR opzionali). Il fade verso l'indice e' un
ritorno alla media -> orizzonte breve.
"""
from __future__ import annotations

import sys
from pathlib import Path

import numpy as np
import pandas as pd

PROJECT_ROOT = Path(__file__).resolve().parents[3]
sys.path.insert(0, str(PROJECT_ROOT))

from scripts.analysis.dispersion_lab import features, align_to, UNIVERSE  # noqa: E402
from scripts.analysis.explore_lab import get_df, evaluate, robust, atr     # noqa: E402

W = 24  # finestra famiglia


def _last_rank(x: np.ndarray) -> float:
    """Frazione dei valori (esclusi l'ultimo) strettamente < dell'ultimo. Causale:
    l'ultimo elemento e' la barra i, confrontata coi 719 valori precedenti."""
    if x.shape[0] < 2:
        return np.nan
    return float((x[:-1] < x[-1]).mean())


def _causal_signals(asset: str, df: pd.DataFrame, fa: pd.DataFrame):
    """Precalcola (una volta per asset) rel_z, disp_pctl, atr — feature CAUSALI."""
    a14 = atr(df, 14)
    rel = fa[f"rel_{asset}"].values.astype(float)
    disp = fa["disp_24"].values.astype(float)

    # z-score CAUSALE di rel: media/std rolling 168h SHIFTATA di 1 (solo barre <= i-1)
    rel_s = pd.Series(rel)
    rmean = rel_s.rolling(168).mean().shift(1)
    rstd = rel_s.rolling(168).std().shift(1)
    rel_z = ((rel_s - rmean) / rstd.replace(0, np.nan)).values

    # percentile rolling CAUSALE di disp_24 (rank di disp[i] vs i 720 valori fino a i).
    # vettoriale via rank rolling: pos dell'ultimo elemento / (win-1).
    win = 720
    dr = pd.Series(disp).rolling(win).apply(_last_rank, raw=True)
    disp_pctl = dr.values
    return rel_z, disp_pctl, a14


def build_entries(asset: str, df: pd.DataFrame, fa: pd.DataFrame,
                  rel_z_thr: float, disp_pctl_thr: float,
                  max_bars: int, tp_atr: float | None, sl_atr: float | None,
                  precomp=None) -> list[dict]:
    """Costruisce entries CAUSALI per il fade della componente idiosincratica."""
    n = len(df)
    c = df["close"].values
    if precomp is None:
        rel_z, disp_pctl, a14 = _causal_signals(asset, df, fa)
    else:
        rel_z, disp_pctl, a14 = precomp

    entries: list[dict] = []
    for i in range(n - 1):
        z = rel_z[i]
        dp = disp_pctl[i]
        if not np.isfinite(z) or not np.isfinite(dp):
            continue
        if dp < disp_pctl_thr:           # solo regime di alta dispersione
            continue
        if abs(z) < rel_z_thr:           # solo componente idio estrema
            continue
        d = -1 if z > 0 else +1          # FADE: rel alto -> short A; rel basso -> long A
        a = a14[i]
        if not np.isfinite(a) or a <= 0:
            tp = sl = None
        else:
            tp = c[i] + d * tp_atr * a if tp_atr else None
            sl = c[i] - d * sl_atr * a if sl_atr else None
        entries.append({"i": i, "d": d, "max_bars": max_bars, "tp": tp, "sl": sl})
    return entries


def check_no_lookahead(asset: str, df: pd.DataFrame, fa: pd.DataFrame) -> bool:
    """Perturba i PREZZI dopo un indice T e verifica che le entries con i<=T non cambino.
    Qui ricostruiamo la entry-rule su una copia di df/fa col futuro alterato e confrontiamo
    le entries (i, d) con i<=T."""
    n = len(df)
    T = int(n * 0.6)

    base = build_entries(asset, df, fa, rel_z_thr=2.0, disp_pctl_thr=0.6,
                         max_bars=12, tp_atr=None, sl_atr=None)

    # perturba i prezzi dopo T: alza del 50% close/high/low/open
    df2 = df.copy()
    for col in ("open", "high", "low", "close"):
        df2.loc[df2.index > T, col] = df2.loc[df2.index > T, col] * 1.5
    # perturba anche le feature riferite a barre > T (rel_<asset>, disp_24)
    fa2 = fa.copy()
    for col in (f"rel_{asset}", "disp_24"):
        fa2.loc[fa2.index > T, col] = fa2.loc[fa2.index > T, col] * 1.5

    pert = build_entries(asset, df2, fa2, rel_z_thr=2.0, disp_pctl_thr=0.6,
                         max_bars=12, tp_atr=None, sl_atr=None)

    base_le = {(e["i"], e["d"]) for e in base if e["i"] <= T}
    pert_le = {(e["i"], e["d"]) for e in pert if e["i"] <= T}
    ok = base_le == pert_le
    print(f"[no-look-ahead] entries con i<=T={T} invarianti al futuro: "
          f"{'OK' if ok else 'VIOLATO'} (base={len(base_le)} pert={len(pert_le)})")
    if not ok:
        diff = (base_le ^ pert_le)
        print(f"    differenze: {sorted(diff)[:10]}")
    return ok


def main():
    F = features()
    print(f"feature caricate: {F.shape[0]} barre")

    # asset single-asset da esplorare (i piu' liquidi + qualche alt)
    assets = ["BTC", "ETH", "SOL", "ADA", "BNB", "DOGE", "LTC", "XRP"]

    # griglia piccola di soglie
    rel_z_grid = [1.5, 2.0, 2.5]
    disp_pctl_grid = [0.5, 0.7]
    mb_grid = [6, 12, 24]
    # exit: prima senza tp/sl (puro time-stop), poi con un TP/SL ATR moderato
    exit_grid = [
        (None, None),
        (1.5, 2.0),
    ]

    best = None
    # no-look-ahead check una volta (su ETH)
    df_eth = get_df("ETH", "1h")
    fa_eth = align_to(F, df_eth)
    la_ok = check_no_lookahead("ETH", df_eth, fa_eth)
    print()

    for asset in assets:
        df = get_df(asset, "1h")
        fa = align_to(F, df)
        precomp = _causal_signals(asset, df, fa)   # una volta per asset (costoso)
        for rz in rel_z_grid:
            for dp in disp_pctl_grid:
                for mb in mb_grid:
                    for (tp_a, sl_a) in exit_grid:
                        ents = build_entries(asset, df, fa, rz, dp, mb, tp_a, sl_a,
                                             precomp=precomp)
                        if len(ents) < 30:
                            continue
                        tag = f"{asset} rz{rz} dp{dp} mb{mb} tp{tp_a} sl{sl_a}"
                        res = evaluate(tag, ents, df)
                        rb = robust(res)
                        # criterio "migliore": OOS ret, poi sharpe; preferisci robuste
                        score = (rb, res["oos"]["ret"], res["full"]["sharpe"])
                        if best is None or score > best[0]:
                            best = (score, tag, res, rb,
                                    dict(asset=asset, rz=rz, dp=dp, mb=mb, tp=tp_a, sl=sl_a))

    print("\n=== MIGLIORE ===")
    if best is None:
        print("nessuna cella con abbastanza trade")
        return
    score, tag, res, rb, cfg = best
    print(f"config: {tag}")
    print(f"robust={rb}  lookahead_ok={la_ok}")
    print(f"FULL ret={res['full']['ret']:+.0f}%  OOS ret={res['oos']['ret']:+.0f}%  "
          f"DD={res['full']['dd']:.0f}%  Sharpe={res['full']['sharpe']:.2f}")
    print(f"fee0.2% OOS={res['sweep_oos'][0.002]:+.0f}%  anniPos={res['pos_yrs']}/{res['n_yrs']}")


if __name__ == "__main__":
    main()