PythagorasGoal/scripts/analysis/dispersion_edges/gate_index_comp_disp.py

"""GATE PORT06 del candidato index_comp_disp W=168 (ricerca dispersion 2026-06-08).

Edge confermato avversarialmente: fade della componente idiosincratica di BTC verso
l'indice EW, gated da alta dispersione. Config: rel_len=12, z_win=336, z_thr=1.5,
disp_168 >= quantile rolling 0.7 (win 720), TP=1.0*ATR14, SL=1.5*ATR14, max_bars=24.

Domanda del gate (lezione FR01: robusto != migliora-il-portafoglio):
  1) correlazione daily col MASTER e con le fade BTC esistenti (e' un diversificatore?)
  2) PORT06 BASE (17 sleeve) vs +DISP (18 sleeve) con pesi cap: DeltaSharpe/DeltaDD FULL e OOS.
PROMOSSO solo se decorrela E migliora (o non degrada) l'OOS.

    uv run python scripts/analysis/dispersion_edges/gate_index_comp_disp.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
from scripts.analysis.explore_lab import get_df, atr
from scripts.analysis.combine_portfolio import _norm, IDX, port_returns, metrics, SPLIT, OOS_DATE
from scripts.analysis.honest_improve2 import _daily_equity
from scripts.portfolios._defs import PORTFOLIOS
from src.portfolio import weighting as W

FEE_RT, LEV, POS, INIT = 0.001, 3.0, 0.15, 1000.0
CFG = dict(rel_len=12, z_win=336, z_thr=1.5, disp_q=0.7, disp_q_win=720,
           tp_atr=1.0, sl_atr=1.5, max_bars=24)


def _last_rank(x):
    if x.shape[0] < 2:
        return np.nan
    return float((x[:-1] < x[-1]).mean())


def build_trades(asset="BTC"):
    """Entries CAUSALI + exit intrabar (TP/SL/max_bars) -> [(i, j, ret_netto)]."""
    df = get_df(asset, "1h")
    F = features()
    fa = align_to(F, df)
    c, h, l = df["close"].values, df["high"].values, df["low"].values
    n = len(c)
    a14 = atr(df, 14)
    rel = fa[f"rel_{asset}"].values.astype(float)
    disp = fa["disp_168"].values.astype(float)
    # somma rolling rel su rel_len, z-score causale (mean/std rolling z_win shift 1)
    rs = pd.Series(rel).rolling(CFG["rel_len"]).sum()
    rmean = rs.rolling(CFG["z_win"]).mean().shift(1)
    rstd = rs.rolling(CFG["z_win"]).std().shift(1)
    z = ((rs - rmean) / rstd.replace(0, np.nan)).values
    dpct = pd.Series(disp).rolling(CFG["disp_q_win"]).apply(_last_rank, raw=True).values
    fee = FEE_RT * LEV
    out = []
    last = -1
    for i in range(n - 1):
        if i <= last or not np.isfinite(z[i]) or not np.isfinite(dpct[i]):
            continue
        if dpct[i] < CFG["disp_q"] or abs(z[i]) < CFG["z_thr"]:
            continue
        ai = a14[i]
        if not np.isfinite(ai) or ai <= 0:
            continue
        d = -1 if z[i] > 0 else 1
        tp = c[i] + d * CFG["tp_atr"] * ai
        sl = c[i] - d * CFG["sl_atr"] * ai
        mb = CFG["max_bars"]
        j = min(i + mb, n - 1)
        exit_p = c[j]
        for k in range(1, mb + 1):
            j = i + k
            if j >= n:
                j = n - 1; exit_p = c[j]; break
            if d == 1:
                if l[j] <= sl: exit_p = sl; break
                if h[j] >= tp: exit_p = tp; break
            else:
                if h[j] >= sl: exit_p = sl; break
                if l[j] <= tp: exit_p = tp; break
            if k == mb: exit_p = c[j]
        out.append((i, j, (exit_p - c[i]) / c[i] * d * LEV - fee))
        last = j
    return df, out


def daily_equity(df, trades):
    ts = pd.to_datetime(df["timestamp"], unit="ms", utc=True)
    cap = INIT; eq_ts, eq_v = [], []
    for i, j, ret in sorted(trades, key=lambda t: t[1]):
        cap = max(cap + cap * POS * ret, 10.0)
        eq_ts.append(ts.iloc[j]); eq_v.append(cap)
    return _norm(_daily_equity(eq_ts, eq_v, IDX))


def pmetrics(members, p, extra=None):
    ids = list(p.sleeve_ids) + ([extra] if extra else [])
    dr = pd.DataFrame({i: members[i].pct_change().fillna(0.0) for i in ids})
    if extra:
        caps = dict(p.caps); caps["DISP"] = caps.get("DISP", None)
    w = W.weight_vector(p.weighting, ids, dr, weights=p.weights,
                        caps=p.caps, clusters={**{i:(p.clusters or {}).get(i,i) for i in p.sleeve_ids},
                                               **({extra:"disp"} if extra else {})},
                        lookback=p.vol_lookback)
    drp = port_returns({i: members[i] for i in ids}, w)
    return metrics(drp), metrics(drp, lo=SPLIT)


def main():
    p = PORTFOLIOS["PORT06"]
    print("=" * 100)
    print("  GATE PORT06 — candidato index_comp_disp W=168 (BTC) | famiglia DISP nuova")
    print(f"  config {CFG} | OOS da {OOS_DATE}")
    print("=" * 100)

    from src.portfolio.sleeves import all_sleeve_equities
    eq_base = dict(all_sleeve_equities())

    df, trades = build_trades("BTC")
    disp_eq = daily_equity(df, trades)
    fr = (disp_eq.iloc[-1] / disp_eq.iloc[0] - 1) * 100
    o = disp_eq.iloc[SPLIT:]; ofr = (o.iloc[-1] / o.iloc[0] - 1) * 100
    print(f"\n[1] candidato standalone: {len(trades)} trade | FULL {fr:+.0f}% | OOS {ofr:+.0f}%")

    # correlazione daily col MASTER e con le fade BTC
    dr_cand = disp_eq.pct_change().fillna(0.0)
    print("\n[2] correlazione daily col candidato (decorrela?):")
    for sid in ["MR01_BTC", "MR02_BTC", "MR07_BTC", "DIP01_BTC"]:
        corr = dr_cand.corr(eq_base[sid].pct_change().fillna(0.0))
        print(f"    {sid:<12} corr {corr:+.3f}")
    master_dr = pd.DataFrame({i: eq_base[i].pct_change().fillna(0.0) for i in p.sleeve_ids}).mean(axis=1)
    print(f"    {'MASTER(EW)':<12} corr {dr_cand.corr(master_dr):+.3f}")

    # PORT06 base vs +DISP
    f_b, o_b = pmetrics(eq_base, p)
    members = dict(eq_base); members["DISP_BTC"] = disp_eq
    f_e, o_e = pmetrics(members, p, extra="DISP_BTC")
    print("\n[3] PORT06 BASE (17) vs +DISP (18):")
    print(f"  {'':<10}{'FULL Sh':>9}{'FULL DD%':>10}{'OOS Sh':>9}{'OOS DD%':>9}")
    print(f"  {'BASE':<10}{f_b['sharpe']:>9.2f}{f_b['dd']:>10.2f}{o_b['sharpe']:>9.2f}{o_b['dd']:>9.2f}")
    print(f"  {'+DISP':<10}{f_e['sharpe']:>9.2f}{f_e['dd']:>10.2f}{o_e['sharpe']:>9.2f}{o_e['dd']:>9.2f}")
    print(f"  {'DELTA':<10}{f_e['sharpe']-f_b['sharpe']:>+9.2f}{f_e['dd']-f_b['dd']:>+10.2f}"
          f"{o_e['sharpe']-o_b['sharpe']:>+9.2f}{o_e['dd']-o_b['dd']:>+9.2f}")

    promoted = (o_e['sharpe'] >= o_b['sharpe'] - 0.02 and o_e['dd'] <= o_b['dd'] + 0.20
                and f_e['sharpe'] >= f_b['sharpe'] - 0.02)
    print("\n  VERDETTO: " + (">>> PROMOSSO <<<" if promoted else ">>> BOCCIATO (diluisce, come FR01) <<<"))


if __name__ == "__main__":
    main()