PythagorasGoal/Old/scripts/analysis/dispersion_edges/rel_idio_fade.py

"""rel_idio_fade (W=24): fade della componente idiosincratica rel_A vs indice.

Idea: rel_A = ret(A) - ret(indice EW) e' il rendimento idiosincratico (residuo di
mercato). Quando l'asset diverge troppo dall'indice (z-score di rel_A su finestra
W=24 elevato), si fada il residuo verso l'indice: se A ha sovraperformato troppo
(z alto) -> SHORT A; se ha sottoperformato (z basso) -> LONG A. Mean-reversion del
residuo.

ENTRY CAUSALE: la decisione a close[i] usa SOLO rel_A fino a i incluso. Lo z-score
e' costruito con media/std rolling su [i-W+1 .. i] (causale). Ingresso eseguibile a
close[i]; exit a tempo (max_bars), opzionale TP/SL ad ATR.

Esegui: cd /opt/docker/PythagorasGoal && PYTHONPATH=. uv run python \
    scripts/analysis/_disp_scratch/rel_idio_fade_24.py
"""
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 correlazione/dispersione (richiesta dalla famiglia)
ASSETS = ["BTC", "ETH", "ADA", "BNB", "DOGE", "LTC", "SOL", "XRP"]
Z_GRID = [1.5, 2.0, 2.5, 3.0]
MB_GRID = [12, 24, 48]
TP_ATR = None               # exit a tempo puro per il primo screening
SL_ATR = None


def build_entries(asset: str, df: pd.DataFrame, fa: pd.DataFrame,
                  z_thr: float, max_bars: int,
                  tp_atr=None, sl_atr=None) -> list[dict]:
    """Entries CAUSALI per il fade del residuo idiosincratico.

    z[i] = (rel[i] - mean(rel[i-W+1..i])) / std(rel[i-W+1..i])  -> usa solo dati <= i.
    rel[i] e' gia' causale (deriva da log-ret fino a close[i]). Quando |z[i]|>=thr:
      z>0 (A ha sovraperformato l'indice) -> SHORT (d=-1), fade verso l'indice
      z<0 (A ha sottoperformato)          -> LONG  (d=+1)
    """
    rel = fa[f"rel_{asset}"].values.astype(float)
    s = pd.Series(rel)
    mu = s.rolling(W).mean().values
    sd = s.rolling(W).std(ddof=0).values
    z = (rel - mu) / np.where(sd > 0, sd, np.nan)

    a = atr(df, 14)
    c = df["close"].values
    n = len(c)
    entries: list[dict] = []
    for i in range(W, n - 1):
        zi = z[i]
        if not np.isfinite(zi) or abs(zi) < z_thr:
            continue
        d = -1 if zi > 0 else 1          # fade del residuo
        e = {"i": i, "d": d, "max_bars": max_bars}
        if tp_atr is not None and np.isfinite(a[i]):
            e["tp"] = c[i] + d * tp_atr * a[i]     # TP nella direzione del fade
        if sl_atr is not None and np.isfinite(a[i]):
            e["sl"] = c[i] - d * sl_atr * a[i]
        entries.append(e)
    return entries


def check_no_lookahead(asset: str, df: pd.DataFrame, fa: pd.DataFrame,
                       z_thr: float, max_bars: int) -> bool:
    """Perturba i prezzi DOPO un indice T e verifica che le entries con i<=T non
    cambino (la entry-rule usa solo dati <= close[i])."""
    rel = fa[f"rel_{asset}"].values.astype(float)
    n = len(rel)
    T = int(n * 0.6)

    def z_of(relv):
        s = pd.Series(relv)
        mu = s.rolling(W).mean().values
        sd = s.rolling(W).std(ddof=0).values
        return (relv - mu) / np.where(sd > 0, sd, np.nan)

    z0 = z_of(rel)
    rel2 = rel.copy()
    rel2[T + 1:] = rel2[T + 1:] + 0.05            # shock del futuro
    z2 = z_of(rel2)

    def ents_from(z):
        out = []
        for i in range(W, n - 1):
            if i > T:
                break
            zi = z[i]
            if np.isfinite(zi) and abs(zi) >= z_thr:
                out.append((i, -1 if zi > 0 else 1))
        return out

    ok = ents_from(z0) == ents_from(z2)
    print(f"  [no-look-ahead {asset}] entries i<=T={T} invarianti al futuro: "
          f"{'OK' if ok else 'VIOLATO'}")
    return ok


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

    best = None
    look_ok_all = True
    for asset in ASSETS:
        df = get_df(asset, "1h")
        fa = align_to(F, df)
        # un check no-look-ahead per asset (config centrale)
        look_ok_all &= check_no_lookahead(asset, df, fa, z_thr=2.0, max_bars=24)
        print(f"--- {asset} ---")
        for z_thr in Z_GRID:
            for mb in MB_GRID:
                ents = build_entries(asset, df, fa, z_thr, mb, TP_ATR, SL_ATR)
                if len(ents) < 30:
                    continue
                name = f"{asset} z{z_thr} mb{mb}"
                res = evaluate(name, ents, df)
                rb = robust(res)
                score = res["oos"]["ret"] + res["full"]["ret"]
                cand = {
                    "asset": asset, "z": z_thr, "mb": mb,
                    "full": res["full"]["ret"], "oos": res["oos"]["ret"],
                    "fee02_oos": res["sweep_oos"][0.002],
                    "dd": res["full"]["dd"], "sharpe": res["full"]["sharpe"],
                    "pos_yrs": res["pos_yrs"], "n_yrs": res["n_yrs"],
                    "robust": rb, "score": score, "trades": res["full"]["trades"],
                }
                # preferisci robuste; a parita' di robustezza, score piu' alto
                if best is None or (cand["robust"], cand["score"]) > (best["robust"], best["score"]):
                    best = cand

    print("\n=== CELLA MIGLIORE ===")
    if best:
        print(f"  asset={best['asset']} z={best['z']} mb={best['mb']} trades={best['trades']}")
        print(f"  FULL={best['full']:+.0f}% OOS={best['oos']:+.0f}% "
              f"fee0.2%OOS={best['fee02_oos']:+.0f}% DD={best['dd']:.0f}% "
              f"Sharpe={best['sharpe']:.2f} anniPos={best['pos_yrs']}/{best['n_yrs']} "
              f"robust={best['robust']}")
    print(f"  no-look-ahead tutti gli asset: {'OK' if look_ok_all else 'VIOLATO'}")
    return best


if __name__ == "__main__":
    main()