PythagorasGoal/scripts/research/r0822_sol_leg.py

#!/usr/bin/env python
"""r0822_sol_leg.py — SOL come TERZA GAMBA direzionale del book Deribit: aggiunge o diluisce?

DOMANDA (operatore, 2026-08-22): XRP/SOL/UNI sono gia' stati analizzati, ma sempre dentro panieri
cross-sectional su Hyperliquid. SOL e' l'UNICO dei tre eseguibile su Deribit (SOL_USDC-PERPETUAL,
min 0.001 SOL ~ $0.08, molto sotto il pavimento $5 del book) e TP01/SKH01 non sono mai stati
misurati su di lui. Questo script chiude il buco.

DISEGNO — cio' che NON si fa qui:
  * NESSUNA ri-ottimizzazione su SOL. I meccanismi sono CONGELATI (TP01 `CANONICAL`,
    SKH01 `SKH01_V2_DD`). Ritarare i parametri su un asset nuovo sarebbe una nuova famiglia e
    richiederebbe `study_family_honest` + deflated-Sharpe; qui la domanda e' se il meccanismo
    ESISTENTE trasferisce, che e' una domanda diversa e piu' onesta.
  * NESSUNA scelta di finestra a posteriori: le due lenti sono dichiarate PRIMA (sotto).

IPOTESI A PRIORI (registrata prima di guardare): DILUISCE. Il test trend-multi-asset del
2026-06-19 trovo' corr 0.74 con TP01 -> ridondante; SOL e' un beta alto di BTC. Se il risultato
fosse "aggiunge", va guardato con piu' sospetto, non con meno.

DUE LENTI, dichiarate prima (la certificazione le impone, non le sceglie il risultato):
  L-FULL   finestra comune 2022-03+  -> usa 2022-23, che la certificazione segnala come rumorosi
                                        (>1% da Coinbase nell'1.3% / 0.6% delle barre)
  L-PULITA finestra 2024+            -> SOL a med 4-6 bps da Coinbase, ma e' quasi tutta hold-out

ANCORA: ogni numero daily e' calcolato su TUTTE E 24 le ancore e riportato come MEDIANA DELLE
DIFFERENZE APPAIATE (regola codificata 2026-07-26: un Delta fra due varianti eredita la fortuna
d'ancora come ogni altro numero, e la mediana delle differenze != differenza delle mediane).

    uv run python scripts/research/r0822_sol_leg.py [--anchors N]
"""
from __future__ import annotations

import sys
from functools import lru_cache
from pathlib import Path

import numpy as np
import pandas as pd

ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(ROOT))
sys.path.insert(0, str(ROOT / "scripts" / "research"))

from r0702_tp01_offset import resample_offset  # noqa: E402  (importato, non riscritto)
from src.backtest.harness import backtest_signals  # noqa: E402
from src.data.downloader import load_data  # noqa: E402  (solo BTC/ETH: vedi _alt sotto)
from src.strategies.skyhook import SKH01_V2_DD, build_frames, skyhook_entries  # noqa: E402
from src.strategies.trend_portfolio import CANONICAL, TrendPortfolio  # noqa: E402

# ⚠️ SOL vive in un NAMESPACE DEDICATO `alt_sol_*.parquet`, non in `sol_*.parquet`.
# Motivo (trovato in questa sessione): il "guardrail solo dati certi" documentato in CLAUDE.md
# — `load_data("SOL", ...) -> FileNotFoundError` — non e' codice, e' l'ASSENZA del file:
# `load_data` non ha alcuna whitelist. Ricostruendo lo storico di SOL in `data/raw/sol_*.parquet`
# il guardrail si disattiva IN SILENZIO, e per giunta quel file NON viene rinfrescato dal cron
# (`cron_daily.sh` fa `--asset BTC ETH`) -> diventerebbe dato stantio che sembra attivo.
# Il namespace dedicato e' la convenzione che il progetto usa gia' per `hl_` e `eq_`.
# Per rigenerarlo:  uv run python scripts/analysis/rebuild_history.py --asset SOL
#                   poi rinominare sol_*.parquet -> alt_sol_*.parquet
ALT_NS = "alt_"

def load_alt(asset: str, tf: str) -> pd.DataFrame:
    """Legge un asset del namespace di ricerca (fuori dal feed attivo BTC/ETH)."""
    path = ROOT / "data" / "raw" / f"{ALT_NS}{asset.lower()}_{tf}.parquet"
    if not path.exists():
        raise FileNotFoundError(
            f"{path} assente. Rigenera con:\n"
            f"  uv run python scripts/analysis/rebuild_history.py --asset {asset}\n"
            f"  poi rinomina data/raw/{asset.lower()}_*.parquet -> {ALT_NS}{asset.lower()}_*.parquet")
    df = pd.read_parquet(path)
    df["datetime"] = pd.to_datetime(df["timestamp"], unit="ms", utc=True)
    return df


def dati(asset: str, tf: str) -> pd.DataFrame:
    """BTC/ETH dal feed certificato attivo; tutto il resto dal namespace di ricerca."""
    return load_data(asset, tf) if asset in ("BTC", "ETH") else load_alt(asset, tf)


DUE = ("BTC", "ETH")
TRE = ("BTC", "ETH", "SOL")
W_TP, W_SKH = 0.75, 0.25                       # pesi del book live Deribit
PULITA = pd.Timestamp("2024-01-01", tz="UTC")  # dalla certificazione, non dal risultato
HOLDOUT = pd.Timestamp("2025-01-01", tz="UTC")
ANNO = 365.0


def sh(s: pd.Series) -> float:
    s = s.dropna()
    return float(s.mean() / s.std() * np.sqrt(ANNO)) if len(s) > 30 and s.std() > 0 else float("nan")


def maxdd(s: pd.Series) -> float:
    e = (1 + s.fillna(0)).cumprod()
    return float((e / e.cummax() - 1).min())


def cagr(s: pd.Series) -> float:
    s = s.dropna()
    if len(s) < 30:
        return float("nan")
    return float((1 + s).prod() ** (ANNO / len(s)) - 1)


@lru_cache(maxsize=8)
def _h1(asset: str) -> pd.DataFrame:
    return dati(asset, "1h")


@lru_cache(maxsize=128)
def tp01_leg(asset: str, h: int) -> pd.Series:
    """TP01 CONGELATO su un asset, ancora h. Rendimenti netti giornalieri, causale."""
    df = resample_offset(_h1(asset), h)
    tp = TrendPortfolio(**CANONICAL)
    px = df["close"].values.astype(float)
    r = np.zeros(len(px)); r[1:] = px[1:] / px[:-1] - 1.0
    tgt = tp.target_series(df)
    held = np.zeros(len(tgt)); held[1:] = tgt[:-1]
    net = held * r - tp.fee_side * np.abs(np.diff(held, prepend=0.0)); net[0] = 0.0
    return pd.Series(np.clip(net, -0.99, None),
                     index=pd.DatetimeIndex(pd.to_datetime(df["datetime"], utc=True)))


@lru_cache(maxsize=8)
def skh_leg(asset: str) -> pd.Series:
    """SKH01-V2-DD CONGELATO su un asset, riportato su griglia giornaliera (come lo sleeve)."""
    ltf, htf = build_frames(dati(asset, "5m"))
    ent = skyhook_entries(ltf, htf, SKH01_V2_DD)
    m = backtest_signals(ltf, ent, fee_rt=0.001, leverage=1.0, asset=asset, tf="230m")
    s = pd.Series(m.equity, index=pd.DatetimeIndex(pd.to_datetime(m.eq_index, utc=True)))
    return s.resample("1D").last().ffill().pct_change().dropna()


def ew(series: dict) -> pd.Series:
    """Equal-weight sulle date COMUNI (inner): un book a N gambe si valuta dove ci sono tutte."""
    J = pd.concat(series, axis=1, join="inner").dropna()
    return J.mean(axis=1)


def book(assets: tuple, h: int) -> pd.Series:
    tp = ew({a: tp01_leg(a, h) for a in assets})
    sk = ew({a: skh_leg(a) for a in assets})
    J = pd.concat({"tp": tp, "sk": sk}, axis=1, join="inner").fillna(0.0)
    return W_TP * J["tp"] + W_SKH * J["sk"]


def k_iso_dd(base: pd.Series, target: pd.Series, lo: float = 0.01, hi: float = 1.0) -> float:
    """Fattore k per cui `k * base` ha lo STESSO maxDD di `target`. PURA.

    E' il cuore del NULL DEL DE-LEVERING, che in questo progetto e' il primo test dovuto a ogni
    claim "meno drawdown" (5a occorrenza: VRP-DD 03/07, TP01xDVOL 26/06, MAT01 25/07, azioni
    intere UCITS 26/07, e SOL qui). maxDD e' monotono in k, quindi la bisezione converge.
    """
    obiettivo = maxdd(target)
    for _ in range(60):
        k = 0.5 * (lo + hi)
        if maxdd(k * base) < obiettivo:          # piu' profondo del bersaglio -> ridurre la size
            hi = k
        else:
            lo = k
    return 0.5 * (lo + hi)


def taglia(s: pd.Series, start=None) -> pd.Series:
    return s if start is None else s[s.index >= start]


def riga(nome: str, s: pd.Series) -> str:
    return (f"  {nome:<26}{len(s):>7}{sh(s):>9.2f}{maxdd(s)*100:>9.1f}%{cagr(s)*100:>9.1f}%"
            f"{sh(taglia(s, HOLDOUT)):>10.2f}")


def main() -> None:
    n_anc = 24
    if "--anchors" in sys.argv:
        n_anc = int(sys.argv[sys.argv.index("--anchors") + 1])
    anchors = list(range(0, 24, max(1, 24 // n_anc)))[:n_anc]

    print("=" * 104)
    print("  r0822 — SOL COME TERZA GAMBA DEL BOOK DERIBIT (meccanismi CONGELATI)")
    print("=" * 104)
    print("  ipotesi a priori dichiarata: DILUISCE (trend multi-asset 19/06: corr 0.74 con TP01)")
    print(f"  ancore valutate: {len(anchors)}   pesi book TP01 {W_TP} / SKH01 {W_SKH}")

    # ---------------------------------------------------------------- 1. le gambe da sole
    print("\n  [1/5] LE GAMBE DA SOLE (ancora canonica h=0, finestra propria di ogni asset)")
    print(f"\n  {'gamba':<26}{'barre':>7}{'Sharpe':>9}{'maxDD':>10}{'CAGR':>9}{'Sh hold':>10}")
    for a in TRE:
        print(riga(f"TP01 {a}", tp01_leg(a, 0)))
    for a in TRE:
        print(riga(f"SKH01 {a}", skh_leg(a)))

    # ---------------------------------------------------------------- 2. SOL nelle due lenti
    print("\n  [2/5] LA GAMBA SOL NELLE DUE LENTI DICHIARATE")
    solb = W_TP * tp01_leg("SOL", 0).reindex(skh_leg("SOL").index).fillna(0.0) \
        + W_SKH * skh_leg("SOL")
    print(f"\n  {'lente':<26}{'barre':>7}{'Sharpe':>9}{'maxDD':>10}{'CAGR':>9}{'Sh hold':>10}")
    print(riga("SOL book-leg L-FULL", solb))
    print(riga("SOL book-leg L-PULITA", taglia(solb, PULITA)))
    b2 = book(DUE, 0)
    comune = solb.index.intersection(b2.index)
    if len(comune) > 100:
        c = float(np.corrcoef(solb.loc[comune], b2.loc[comune])[0, 1])
        cp = comune[comune >= PULITA]
        c2 = float(np.corrcoef(solb.loc[cp], b2.loc[cp])[0, 1]) if len(cp) > 100 else float("nan")
        print(f"\n    corr(gamba SOL, book attuale BTC/ETH): L-FULL {c:+.3f}   L-PULITA {c2:+.3f}")
        print("    (riferimento: il trend multi-asset del 19/06 fu scartato a corr 0.74)")

    # ---------------------------------------------------------------- 3. book 2 vs 3 gambe
    print("\n  [3/5] BOOK A 2 GAMBE vs 3 GAMBE — differenze APPAIATE per ancora")
    for nome, start in (("L-FULL (2022-03+)", None), ("L-PULITA (2024+)", PULITA)):
        d_sh, d_dd, d_cagr, d_hold = [], [], [], []
        for h in anchors:
            A, B = book(DUE, h), book(TRE, h)
            idx = A.index.intersection(B.index)          # stessa finestra: SOL parte dopo
            A, B = taglia(A.loc[idx], start), taglia(B.loc[idx], start)
            if len(A) < 200:
                continue
            d_sh.append(sh(B) - sh(A)); d_dd.append((maxdd(B) - maxdd(A)) * 100)
            d_cagr.append((cagr(B) - cagr(A)) * 100)
            d_hold.append(sh(taglia(B, HOLDOUT)) - sh(taglia(A, HOLDOUT)))
        if not d_sh:
            print(f"\n    {nome}: campione insufficiente")
            continue
        print(f"\n    {nome}   ({len(d_sh)} ancore)")
        for lab, v, u in (("dSharpe FULL", d_sh, ""), ("dSharpe hold-out", d_hold, ""),
                          ("DD piu' BASSO di", d_dd, " pp"), ("d CAGR", d_cagr, " pp")):
            v = np.array(v, float)
            pos = np.mean(v > 0) * 100
            print(f"      {lab:<20} mediana {np.median(v):+7.3f}{u}   "
                  f"banda [{np.percentile(v, 10):+.3f}, {np.percentile(v, 90):+.3f}]   "
                  f">0 nel {pos:.0f}% delle ancore")

    # ---------------------------------------------------------------- 3-bis. per anno
    print("\n  [3bis] DOVE NASCE IL GUADAGNO — book 2 vs 3 gambe, anno per anno (ancora h=0)")
    A0, B0 = book(DUE, 0), book(TRE, 0)
    i0 = A0.index.intersection(B0.index); A0, B0 = A0.loc[i0], B0.loc[i0]
    print(f"\n    {'anno':<7}{'2g Sh':>8}{'3g Sh':>8}{'dSh':>8}{'2g CAGR':>10}{'3g CAGR':>10}{'dCAGR':>9}")
    for y in sorted(set(A0.index.year)):
        a, b = A0[A0.index.year == y], B0[B0.index.year == y]
        if len(a) < 60:
            continue
        print(f"    {y:<7}{sh(a):>8.2f}{sh(b):>8.2f}{sh(b)-sh(a):>+8.2f}"
              f"{cagr(a)*100:>9.1f}%{cagr(b)*100:>9.1f}%{(cagr(b)-cagr(a))*100:>+8.1f}pp")
    solleg = W_TP * tp01_leg("SOL", 0).reindex(skh_leg("SOL").index).fillna(0.0) + W_SKH * skh_leg("SOL")
    print(f"\n    la sola gamba SOL:  " + "  ".join(
        f"{y} Sh {sh(solleg[solleg.index.year == y]):+.2f}"
        for y in sorted(set(solleg.index.year)) if len(solleg[solleg.index.year == y]) >= 60))

    # ---------------------------------------------------------------- 4. null de-levering
    print("\n  [4/5] NULL DEL DE-LEVERING — il primo test dovuto a ogni claim 'meno drawdown'")
    print("    Aggiungere SOL abbassa DD *e* CAGR insieme. La domanda non e' 'ha meno DD?' ma")
    print("    'lo stesso DD si compra piu' a buon mercato semplicemente riducendo la size?'.")
    print("    Si cerca k tale che maxDD(k * book_2gambe) == maxDD(book_3gambe), poi si confronta.")
    print(f"\n    {'lente':<20}{'k':>7}{'Sh 3gambe':>11}{'Sh 2g x k':>11}{'CAGR 3g':>10}"
          f"{'CAGR 2g x k':>13}{'verdetto':>14}")
    for nome, start in (("L-FULL", None), ("L-PULITA", PULITA)):
        A, B = book(DUE, 0), book(TRE, 0)
        idx = A.index.intersection(B.index)
        A, B = taglia(A.loc[idx], start), taglia(B.loc[idx], start)
        k = k_iso_dd(A, B)
        sh3, sh2 = sh(B), sh(k * A)
        v = "SOL AGGIUNGE" if sh3 > sh2 + 0.02 else ("de-levering" if sh2 > sh3 + 0.02 else "pari")
        print(f"    {nome:<20}{k:>7.3f}{sh3:>11.2f}{sh2:>11.2f}{cagr(B)*100:>9.1f}%"
              f"{cagr(k*A)*100:>12.1f}%{v:>14}")

    # ---------------------------------------------------------------- 5. eseguibilita'
    print("\n  [5/5] ESEGUIBILITA' a $644 (equity reale del conto)")
    px = float(dati("SOL", "1h")["close"].iloc[-1])
    print(f"    SOL_USDC-PERPETUAL  min 0.001 SOL = ${0.001*px:.3f}  <<  pavimento min_order $5")
    print(f"    a 3 gambe il target per asset scende da ~1/2 a ~1/3 dell'esposizione:")
    for eq in (644.0, 2000.0, 6050.0):
        print(f"      equity ${eq:>7,.0f}:  2 gambe ${eq*0.5*0.5:>7,.0f}/asset   "
              f"3 gambe ${eq*0.5/3:>7,.0f}/asset")
    print("\n" + "=" * 104)


if __name__ == "__main__":
    main()