PythagorasGoal/Old/scripts/analysis/xex_divergence_research.py

"""XEX — Discordanze cross-exchange Deribit (testnet) vs Hyperliquid.

Ricerca 2026-06-12. Domanda: il prezzo Deribit testnet si discosta da quello
Hyperliquid (proxy della realta'); lo scostamento e' tradabile dal nostro conto?

Esito (vedi diario docs/diary/2026-06-12-xex-divergence.md):
- Lo spread log(D/H) e' enorme per standard reali (std 0.9-4.5%) e MEAN-REVERTING
  (AR1 rho 0.77-0.94, half-life 2.7-12 barre 1h).
- Il gap viene chiuso da ENTRAMBI i lati: beta del ritorno futuro Deribit sullo
  spread e' negativo e cresce con l'orizzonte (ETH -0.36, BTC -0.23 a 24h)
  -> tradabile dal lato Deribit (il nostro conto).
- TRAPPOLA SMASCHERATA: su DOGE/SOL (lineari USDC illiquidi, 87%/35% barre flat)
  l'edge del backtest (Sharpe 6.7/2.7) e' FINZIONE da print stantii: il BOOK
  live sta attaccato a HL (+0.16%/-0.05%) mentre i print restano vecchi.
  Su BTC/ETH inverse invece il BOOK STESSO e' dislocato (-0.94%/-2.16% misurati
  live con depth >$1M) -> la' la discordanza e' reale ed eseguibile.
- Candidati: solo BTC-PERPETUAL / ETH-PERPETUAL. Edge netto (fee 0.10% RT)
  moderato e sensibile al timing (half-life corta: lag 1h di entry lo erode).
  NON deployare senza: segnale dal BOOK (non dal close), poll fitto, gate PORT06.

NB: e' un edge da TESTNET (la dislocazione e' l'artefatto del feed testnet che
rientra verso la realta'): non trasferibile a mainnet, dove lo spread D/H reale
e' <0.05%. Utile per il paper/shadow corrente, non per capitale vero.
"""
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 src.live.cerbero_client import CerberoClient

FEE_RT = 0.001
START, END = "2026-03-01", "2026-06-12"
SPLIT = pd.Timestamp("2026-05-10", tz="UTC")
PAIRS = [
    ("BTC", "BTC-PERPETUAL"),
    ("ETH", "ETH-PERPETUAL"),
    ("SOL", "SOL_USDC-PERPETUAL"),
    ("DOGE", "DOGE_USDC-PERPETUAL"),
]


def fetch(c: CerberoClient, coin: str, d_inst: str) -> pd.DataFrame:
    def hist(ex: str, inst: str) -> pd.Series:
        rows = c.get_historical_v2(inst, START, END, interval="1h", exchange=ex)
        df = pd.DataFrame(rows)
        df["ts"] = pd.to_datetime(df["timestamp"], unit="ms", utc=True)
        df = df.set_index("ts")
        return df.loc[~df.index.duplicated(), "close"]

    return pd.DataFrame({"d": hist("deribit", d_inst), "h": hist("hyperliquid", coin)}).dropna()


def convergence_table(j: pd.DataFrame) -> None:
    """Chi chiude il gap: regressione spread[i] -> ritorno futuro per venue."""
    s = np.log(j["d"] / j["h"])
    for hz in (1, 6, 12, 24):
        rd = np.log(j["d"].shift(-hz) / j["d"])
        rh = np.log(j["h"].shift(-hz) / j["h"])
        m = s.notna() & rd.notna() & rh.notna()
        bd = np.polyfit(s[m], rd[m], 1)[0]
        bh = np.polyfit(s[m], rh[m], 1)[0]
        print(f"  h={hz:>2}: beta_D={bd:+.2f} (lato tradabile) beta_H={bh:+.2f}")


def backtest(j: pd.DataFrame, entry: float = 1.0, exit_: float = 0.25,
             max_bars: int = 24, fee: float = FEE_RT, lag: int = 0):
    """Fade dello spread sul solo lato Deribit. Entry al close (o close+lag per
    stress staleness), skip barre flat, exit a |s|<=exit_ o max_bars."""
    d, h = j["d"].values, j["h"].values
    s = np.log(d / h) * 100
    dret = np.r_[0.0, np.diff(np.log(d))]
    flat = np.r_[True, dret[1:] == 0]
    pos, entry_i, pnl, pend = 0, -1, 0.0, None
    eq, trades = [0.0], []
    for i in range(1, len(j)):
        r = 0.0
        if pos != 0:
            r = pos * dret[i]
            pnl += r
            if abs(s[i]) <= exit_ or (i - entry_i) >= max_bars:
                r -= fee / 2
                trades.append(pnl - fee)
                pos, pnl = 0, 0.0
        if pend is not None and pend[0] == i:
            if pos == 0:
                pos, entry_i, pnl = pend[1], i, 0.0
                r -= fee / 2
            pend = None
        if pos == 0 and pend is None and abs(s[i]) >= entry and not flat[i]:
            if lag == 0:
                pos, entry_i, pnl = -np.sign(s[i]), i, 0.0
                r -= fee / 2
            else:
                pend = (i + lag, -np.sign(s[i]))
        eq.append(r)
    return pd.Series(eq, index=j.index), np.array(trades)


def report(rets: pd.Series, trades: np.ndarray, label: str) -> None:
    ann = np.sqrt(24 * 365)
    sh = rets.mean() / rets.std() * ann if rets.std() > 0 else 0.0
    cum = rets.cumsum()
    dd = (cum - cum.cummax()).min() * 100
    wr = (trades > 0).mean() * 100 if len(trades) else 0.0
    print(f"  {label:10} ret={rets.sum() * 100:+7.1f}% Sh={sh:5.2f} DD={dd:6.2f}% "
          f"n={len(trades):3d} WR={wr:4.1f}%")


def book_reality_check(c: CerberoClient) -> None:
    """Il test che separa edge vero da illusione: il BOOK e' dislocato o solo i print?"""
    print("\n== Book Deribit vs mark Hyperliquid (live) ==")
    for coin, inst in PAIRS:
        try:
            ob = c._post("/mcp-deribit/tools/get_orderbook", {"instrument_name": inst, "depth": 5})
            ht = c._post("/mcp-hyperliquid/tools/get_ticker", {"instrument": coin})
            bb, ba = ob["bids"][0][0], ob["asks"][0][0]
            mid, hm = (bb + ba) / 2, ht["mark_price"]
            print(f"  {inst:22} book {bb}/{ba}  Δbook-HL={100 * (mid / hm - 1):+.2f}%  "
                  f"depth5 bid={sum(b[1] for b in ob['bids']):.3g} ask={sum(a[1] for a in ob['asks']):.3g}")
        except Exception as e:  # endpoint o strumento indisponibile: solo report
            print(f"  {inst:22} ERR {e}")


def run() -> None:
    c = CerberoClient()
    data = {coin: fetch(c, coin, inst) for coin, inst in PAIRS}
    for coin, j in data.items():
        s = np.log(j["d"] / j["h"]) * 100
        rho = (s - s.mean()).autocorr(1)
        hlife = -np.log(2) / np.log(rho) if 0 < rho < 1 else float("inf")
        flat = (j["d"].pct_change() == 0).mean() * 100
        print(f"\n== {coin} ({len(j)} barre 1h) spread mean={s.mean():+.2f}% std={s.std():.2f}% "
              f"half-life={hlife:.1f}h flatD={flat:.0f}%")
        convergence_table(j)
        for lag in (0, 1):
            r, t = backtest(j, lag=lag)
            report(r, t, f"FULL lag{lag}")
            roo, too = backtest(j[j.index >= SPLIT], lag=lag)
            report(roo, too, f"OOS  lag{lag}")
    book_reality_check(c)


if __name__ == "__main__":
    run()