"""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_[W] : beta rolling dell'asset A vs indice - rel_ : 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()