PythagorasGoal/Old/scripts/analysis/dispersion_lab.py

"""Harness CONDIVISO per la ricerca dispersion/correlation index (crypto).

Feature CAUSALI (dalle sole close, nessun feed opzioni — la dispersion IMPLICITA
non e' backtestabile, muro ARGO/GEX documentato). Calcolate sull'universo comune
e allineabili a ogni singolo asset. Tutte note a close[i] (nessun look-ahead):

  - avg_corr[W] : correlazione media a coppie dei log-rendimenti, rolling W (causale)
  - disp[W]     : dispersione cross-sectional (std cross-asset del rendimento di barra),
                  media rolling W
  - idx_ret     : rendimento dell'"indice" equal-weight (proxy mercato)
  - beta_<A>[W] : beta rolling dell'asset A vs indice
  - rel_<A>     : rendimento di A meno rendimento indice (componente idiosincratica)

Uso dagli agenti di ricerca:
    from scripts.analysis.dispersion_lab import features, align_to, UNIVERSE, COMMON_START
    from scripts.analysis.explore_lab import get_df, evaluate, robust
    F = features()                       # DataFrame indicizzato per timestamp(ms)
    df = get_df("ETH", "1h")
    fa = align_to(F, df)                 # feature riallineate alle barre di df (ffill causale)
    # ... costruisci entries causali (entry decisa con dati <= close[i]) ...
    res = evaluate("nome", entries, df); robust(res)

Check no-look-ahead: `python -m scripts.analysis.dispersion_lab` (perturba il futuro
e verifica che le feature fino a T non cambino).
"""
from __future__ import annotations

import sys
from pathlib import Path

import numpy as np
import pandas as pd

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

from scripts.analysis.explore_lab import get_df  # noqa: E402

UNIVERSE = ["ADA", "BNB", "BTC", "DOGE", "ETH", "LTC", "SOL", "XRP"]
COMMON_START = "2022-07-22"      # ultimo asset a entrare (LTC) -> universo completo
WINDOWS = [24, 72, 168, 336]     # 1g, 3g, 1sett, 2sett in barre 1h
_CACHE: dict | None = None


def _panel():
    """{asset: close-series} allineato sui timestamp comuni dell'universo (1h)."""
    frames = {}
    for a in UNIVERSE:
        d = get_df(a, "1h")
        frames[a] = pd.Series(d["close"].values, index=d["timestamp"].values, name=a)
    P = pd.concat(frames, axis=1).dropna()
    P = P[P.index >= int(pd.Timestamp(COMMON_START, tz="UTC").timestamp() * 1000)]
    return P


def _avg_pairwise_corr(R: np.ndarray, win: int) -> np.ndarray:
    """Media delle correlazioni a coppie dei log-rendimenti su finestra rolling.
    CAUSALE: la riga i usa R[i-win+1 .. i]. Vettoriale via media/var rolling delle
    scommatorie (corr di Pearson per coppia, poi media off-diagonale)."""
    n, m = R.shape
    out = np.full(n, np.nan)
    # somme rolling per asset
    df = pd.DataFrame(R)
    s = df.rolling(win).sum().values                  # Σx
    ss = (df * df).rolling(win).sum().values          # Σx²
    for i in range(win - 1, n):
        w = R[i - win + 1:i + 1]                       # (win, m)
        mean = s[i] / win
        var = ss[i] / win - mean * mean
        sd = np.sqrt(np.clip(var, 1e-18, None))
        # matrice di covarianza della finestra
        cov = (w.T @ w) / win - np.outer(mean, mean)
        corr = cov / np.outer(sd, sd)
        iu = np.triu_indices(m, k=1)
        vals = corr[iu]
        vals = vals[np.isfinite(vals)]
        if vals.size:
            out[i] = float(np.mean(vals))
    return out


_CACHE_FILE = PROJECT_ROOT / "data" / "regime" / "dispersion_features.parquet"


def features(use_disk: bool = True) -> pd.DataFrame:
    """DataFrame indicizzato per timestamp(ms) con le feature causali. Cache di
    processo + cache su disco (i molti agenti di ricerca la caricano invece di
    ricalcolarla; la corr rolling e' costosa)."""
    global _CACHE
    if _CACHE is not None:
        return _CACHE
    if use_disk and _CACHE_FILE.exists():
        _CACHE = pd.read_parquet(_CACHE_FILE)
        return _CACHE
    P = _panel()
    logp = np.log(P.values)
    R = np.vstack([np.zeros((1, P.shape[1])), np.diff(logp, axis=0)])   # log-ret per barra
    R[0] = 0.0
    idx_ret = R.mean(axis=1)                                            # indice EW
    out = pd.DataFrame(index=P.index)
    out["idx_ret"] = idx_ret
    # dispersione cross-sectional (std cross-asset del rendimento di barra) + medie rolling
    xs = R.std(axis=1)
    out["disp_bar"] = xs
    for w in WINDOWS:
        out[f"avg_corr_{w}"] = _avg_pairwise_corr(R, w)
        out[f"disp_{w}"] = pd.Series(xs, index=P.index).rolling(w).mean().values
    # componente idiosincratica e beta rolling vs indice (per ogni asset)
    ir = pd.Series(idx_ret, index=P.index)
    for k, a in enumerate(UNIVERSE):
        ra = pd.Series(R[:, k], index=P.index)
        out[f"rel_{a}"] = (ra - ir).values
        for w in (72, 168):
            cov = ra.rolling(w).cov(ir)
            var = ir.rolling(w).var()
            out[f"beta_{a}_{w}"] = (cov / var.replace(0, np.nan)).values
    if use_disk:
        _CACHE_FILE.parent.mkdir(parents=True, exist_ok=True)
        out.to_parquet(_CACHE_FILE)
    _CACHE = out
    return out


def align_to(F: pd.DataFrame, df: pd.DataFrame) -> pd.DataFrame:
    """Riallinea le feature (indicizzate per ts comuni) alle barre di `df` (un asset),
    con ffill CAUSALE (riempie in avanti: la feature a i usa l'ultima nota <= ts[i])."""
    f = F.reindex(F.index.union(df["timestamp"].values)).sort_index().ffill()
    return f.reindex(df["timestamp"].values).reset_index(drop=True)


def _check_no_lookahead() -> bool:
    """Perturba il FUTURO dei prezzi e verifica che le feature fino a T non cambino."""
    global _CACHE
    _CACHE = None
    F0 = features().copy()
    P = _panel()
    T = int(len(P) * 0.6)
    # perturbo le close DOPO T per tutti gli asset
    P2 = P.copy()
    P2.iloc[T + 1:] = P2.iloc[T + 1:] * 1.5
    # ricostruisco le feature da P2 inline (stessa logica)
    _CACHE = None
    saved = globals()["_panel"]
    globals()["_panel"] = lambda: P2
    _CACHE = None
    F1 = features()
    globals()["_panel"] = saved
    _CACHE = None
    cols = [c for c in F0.columns if c.startswith(("avg_corr", "disp", "beta"))]
    a = F0[cols].iloc[:T - max(WINDOWS)].values
    b = F1[cols].iloc[:T - max(WINDOWS)].values
    ok = np.allclose(np.nan_to_num(a), np.nan_to_num(b), atol=1e-9)
    print(f"[no-look-ahead] feature fino a T={T} invarianti al futuro: {'OK' if ok else 'VIOLATO'}")
    return ok


if __name__ == "__main__":
    F = features()
    P = _panel()
    print(f"universo {UNIVERSE}")
    print(f"finestra comune: {pd.to_datetime(P.index[0], unit='ms', utc=True).date()} "
          f"-> {pd.to_datetime(P.index[-1], unit='ms', utc=True).date()} ({len(P)} barre)")
    print(f"feature: {list(F.columns)}")
    print(F[[f'avg_corr_{w}' for w in WINDOWS]].describe().round(3).to_string())
    _check_no_lookahead()