PythagorasGoal/scripts/research/r0726_skh_onbook.py

#!/usr/bin/env python
"""r0726_skh_onbook — T1 dell'ondata 2026-07-26: il -0.35 Sharpe del path live di SKH01
e' recuperabile con ORDINI RESTING ON-BOOK invece dell'exit software del cron?

CONTESTO. Il book live Deribit (TP01 0.75 + SKH01 0.25) gira SKH01 col cron ORARIO e uscite
SOFTWARE: il codice guarda il prezzo una volta all'ora e, se il livello SL/TP e' stato superato,
chiude a mercato. Costo misurato (audit 2026-07-02, riconfermato il 24/07): alla lente 50/50 il
book passa FULL 1.46 -> 1.19, HOLD 1.64 -> 1.15, DD 18% -> 25%. E' il **buco singolo piu' grande
del progetto** (~ -0.35 di Sharpe), e il 24/07 ha gia' stabilito che **un cron piu' veloce non lo
recupera** ("nessun cron riporta l'exit software sul livello SL dentro una barra che gappa";
allineare la cadenza a 230m vale ~ +0.01/+0.02 Sharpe = rumore).

L'ANGOLO MAI TESTATO: l'unico meccanismo che non e' un cron — gli **ordini resting on-book**.
E li' c'e' un'ASIMMETRIA che nessuno ha sfruttato:

  * il **TP e' un LIMIT order** appoggiato sul book al livello. Se il prezzo ci arriva, il fill
    e' AL LIVELLO **per costruzione** (sei tu il maker: il prezzo non puo' attraversare il tuo
    limite senza eseguirlo). -> il lato TP e' **integralmente recuperabile**, e per giunta a
    fee MAKER invece che taker.
  * lo **SL e' uno STOP-MARKET**: al trigger diventa un ordine a mercato. In condizioni normali
    riempie a ~pochi bps dal livello; **in un gap riempie molto peggio**. -> il lato SL e'
    recuperabile solo IN PARTE, e il gap resta (com'e' giusto: e' un rischio vero, non un
    artefatto di modello).

Da cui l'ipotesi da falsificare: **il grosso del -0.35 e' sul lato TP, quindi e' recuperabile
senza prendersi nessun rischio nuovo.** Se invece il grosso e' sul lato SL, la conclusione del
24/07 ("il degrado e' strutturale") si estende anche agli ordini on-book e non c'e' niente da fare.

MODALITA' SIMULATE (stesse trade boundaries, cambia SOLO il prezzo d'esecuzione)
  canonical : entry a close[i], exit AL LIVELLO      -> il BACKTEST (riferimento ottimista)
  hourly    : entry/exit al close 5m del prossimo multiplo orario -> il PATH LIVE DI OGGI
  onbook    : entry al prossimo multiplo orario (l'ingresso resta una decisione del cron:
              e' un segnale a chiusura di barra, non si puo' appoggiare in anticipo), ma:
                - TP  -> fill AL LIVELLO, fee MAKER (limit resting)
                - SL  -> stop-market: fill al livello se la barra 5m del trigger lo ATTRAVERSA;
                         fill all'OPEN della barra 5m se la barra ci GAPPA dentro (open gia'
                         oltre il livello), in entrambi i casi meno SLIPPAGE dichiarato
                - time-> resta al prossimo multiplo orario (e' il cron che decide)
  onbook_tp : contro-fattuale che ISOLA il lato TP (TP on-book, SL e time come oggi) -> dice
              quanta parte del recupero viene dal lato che NON aggiunge rischio.

ONESTA'
  * la rilevazione del gap e' a risoluzione **5m** (il gap vero e' piu' fine) -> il fill SL
    modellato e' un OTTIMISMO residuo su quel lato; il lato TP invece non ha questo problema
    (un limit resting non ha bisogno di risoluzione: o il prezzo tocca il livello o no).
  * si ignora la POSIZIONE IN CODA sul limit TP: a $300 di nozionale su BTC perp e' una
    semplificazione benigna (size irrilevante rispetto al book), ma e' dichiarata.
  * risultati riportati sull'off 0 E sulla **banda dei 23 offset** a priori: il progetto ha gia'
    stabilito che l'off 0 e' al 93-98 pctl (best-of-23) -> la mediana di banda e' la lente onesta.
  * niente file di produzione toccato. Se il risultato regge, la modifica al live e' una
    RACCOMANDAZIONE, non un fatto compiuto.

Uso: `uv run python scripts/research/r0726_skh_onbook.py`   (~5-10 min: 23 offset x 2 asset)
"""
from __future__ import annotations

import sys
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"))
sys.path.insert(0, str(ROOT / "scripts" / "research" / "alt"))

import r0702_anchor_skh01 as r02  # noqa: E402
from src.portfolio.portfolio import metrics  # noqa: E402

MSH = 3_600_000
MS_LTF = r02.MS_LTF
FEE_TAKER_SIDE = 0.0005          # Deribit taker ~0.05%/lato (il RT 0.10% del progetto)
FEE_MAKER_SIDE = 0.0000          # limit resting: maker (Deribit futures ~0, a volte rebate)
SL_SLIP = 0.0005                 # slippage dichiarata sullo stop-market (5 bps)
HOLDOUT = "2025-01-01"


# --------------------------------------------------------------------------- simulazione

def sim_equity_exec(ltf: pd.DataFrame, ent: list, mode: str,
                    ts5_close: np.ndarray, o5: np.ndarray, h5: np.ndarray,
                    l5: np.ndarray, c5: np.ndarray) -> np.ndarray:
    """Come r0702.sim_equity ma con le modalita' d'esecuzione ON-BOOK.

    Le trade boundaries (barra d'ingresso, barra di trigger, non-overlap) sono IDENTICHE al
    canonico in ogni modalita': cambia solo il prezzo a cui si esegue e la fee del lato.
    """
    c = ltf["close"].values.astype(float)
    h = ltf["high"].values.astype(float)
    lo = ltf["low"].values.astype(float)
    n = len(c)
    close_ts = ltf["timestamp"].values.astype(np.int64) + MS_LTF

    def px_hour(t: int) -> float:
        hb = ((t + MSH - 1) // MSH) * MSH
        j = np.searchsorted(ts5_close, hb, side="left")
        return c5[min(j, len(c5) - 1)]

    def stop_fill(t_start: int, t_end: int, level: float, direction: int) -> float:
        """Prezzo di fill di uno STOP-MARKET on-book fra due timestamp, a risoluzione 5m.

        Trova la prima barra 5m che tocca il livello: se ci GAPPA dentro (open gia' oltre)
        il fill e' l'open (il gap e' un rischio vero, non un artefatto); altrimenti e' il
        livello. In entrambi i casi meno la slippage dichiarata.
        """
        a = int(np.searchsorted(ts5_close, t_start, side="left"))
        b = int(np.searchsorted(ts5_close, t_end, side="right"))
        for j in range(a, min(b + 1, len(c5))):
            touched = (l5[j] <= level) if direction == 1 else (h5[j] >= level)
            if touched:
                gapped = (o5[j] < level) if direction == 1 else (o5[j] > level)
                px = o5[j] if gapped else level
                return px * (1.0 - SL_SLIP * direction)
        return level * (1.0 - SL_SLIP * direction)      # non trovato a 5m: cap al livello

    initial = 1000.0
    capital = initial
    equity = np.full(n, capital, dtype=float)
    busy_until = -1
    stats = dict(sl=0, tp=0, time=0, gap_sl=0)
    for i in range(n):
        e = ent[i] if i < len(ent) else None
        if e is None or e.get("dir", 0) == 0 or i <= busy_until:
            equity[i] = capital
            continue
        direction = int(e["dir"])
        tp = e.get("tp"); sl = e.get("sl")
        max_bars = int(e.get("max_bars") or 24)

        entry = c[i] if mode == "canonical" else px_hour(close_ts[i])
        fee_entry = FEE_TAKER_SIDE

        if mode == "fastdetect":
            # RILEVAZIONE INTRA-BARRA (proposta per il live, 2026-07-26): il livello SL/TP e'
            # fissato all'INGRESSO da barre chiuse, quindi osservare che il prezzo l'ha gia'
            # toccato NON e' look-ahead — e' cio' che fa qualunque stop reale. Oggi invece
            # `_skyhook_positions` guarda solo barre 230m CHIUSE, e il tocco resta invisibile
            # per una mediana di 115 minuti (p90 210).
            # Qui: si trova il tocco a 5m, e si esegue al PROSSIMO multiplo orario dopo il tocco
            # (non al livello: resta un'uscita a mercato del cron).
            t_end_bar = int(ltf["timestamp"].values[min(i + max_bars, n - 1)]) + MS_LTF
            a5 = int(np.searchsorted(ts5_close, close_ts[i], side="left"))
            b5 = int(np.searchsorted(ts5_close, t_end_bar, side="right"))
            hit_j = None
            for j5 in range(a5, min(b5 + 1, len(c5))):
                hs = (l5[j5] <= sl) if direction == 1 else (h5[j5] >= sl)
                ht = (h5[j5] >= tp) if direction == 1 else (l5[j5] <= tp)
                if hs or ht:
                    hit_j = j5
                    break
            # la barra 230m in cui cade il tocco resta quella che chiude il trade (stesse
            # boundaries del canonico: cambia SOLO il momento dell'esecuzione)
            for j in range(i + 1, min(i + max_bars + 1, n)):
                hit_sl = sl is not None and (
                    (direction == 1 and lo[j] <= sl) or (direction == -1 and h[j] >= sl))
                hit_tp = tp is not None and (
                    (direction == 1 and h[j] >= tp) or (direction == -1 and lo[j] <= tp))
                if hit_sl or hit_tp:
                    exit_idx, kind = j, ("sl" if hit_sl else "tp")
                    break
                exit_idx = j
                kind = "time"
            else:
                exit_idx = min(i + max_bars, n - 1)
                kind = "time"
            stats[kind] += 1
            if hit_j is not None and kind in ("sl", "tp"):
                exit_price = px_hour(int(ts5_close[hit_j]))     # primo cron DOPO il tocco
            else:
                exit_price = px_hour(close_ts[exit_idx])
            gross = (exit_price - entry) / entry * direction
            capital = max(capital + capital * (gross - 2 * FEE_TAKER_SIDE), 1.0)
            equity[i:exit_idx + 1] = capital
            busy_until = exit_idx
            continue

        # --- detection identica al canonico (SL prioritario)
        exit_idx = min(i + max_bars, n - 1)
        exit_lvl = c[exit_idx]
        kind = "time"
        for j in range(i + 1, min(i + max_bars + 1, n)):
            hit_sl = sl is not None and (
                (direction == 1 and lo[j] <= sl) or (direction == -1 and h[j] >= sl))
            hit_tp = tp is not None and (
                (direction == 1 and h[j] >= tp) or (direction == -1 and lo[j] <= tp))
            if hit_sl:
                exit_lvl, exit_idx, kind = sl, j, "sl"
                break
            if hit_tp:
                exit_lvl, exit_idx, kind = tp, j, "tp"
                break
            exit_lvl, exit_idx = c[j], j
        stats[kind] += 1

        # --- prezzo d'uscita per modalita'
        if mode == "canonical":
            exit_price, fee_exit = exit_lvl, FEE_TAKER_SIDE
        elif mode == "hourly":
            exit_price, fee_exit = px_hour(close_ts[exit_idx]), FEE_TAKER_SIDE
        elif mode in ("onbook", "onbook_tp"):
            if kind == "tp":
                exit_price, fee_exit = tp, FEE_MAKER_SIDE          # limit resting: al livello
            elif kind == "sl" and mode == "onbook":
                t0 = int(ltf["timestamp"].values[exit_idx])
                exit_price = stop_fill(t0, t0 + MS_LTF, sl, direction)
                fee_exit = FEE_TAKER_SIDE
                if (direction == 1 and exit_price < sl * (1 - SL_SLIP) - 1e-9) or \
                   (direction == -1 and exit_price > sl * (1 + SL_SLIP) + 1e-9):
                    stats["gap_sl"] += 1
            else:                                                   # sl in onbook_tp, o time
                exit_price, fee_exit = px_hour(close_ts[exit_idx]), FEE_TAKER_SIDE
        else:
            raise ValueError(mode)

        gross = (exit_price - entry) / entry * direction
        net = gross - (fee_entry + fee_exit)
        capital = max(capital + capital * net, 1.0)
        equity[i:exit_idx + 1] = capital
        busy_until = exit_idx
    return equity, stats


def skh_series(off: int, mode: str) -> tuple[pd.Series, dict]:
    """Rendimenti daily dello sleeve SKH01 (50/50 BTC+ETH) a un offset e una modalita'."""
    out = {}
    agg = dict(sl=0, tp=0, time=0, gap_sl=0)
    for asset in ("BTC", "ETH"):
        _, _, ltf, ent = r02.run_asset(asset, off)
        df5 = r02.get5m(asset)
        ts5 = df5["timestamp"].values.astype(np.int64) + 300_000
        eq, st = sim_equity_exec(ltf, ent, mode, ts5,
                                 df5["open"].values.astype(float),
                                 df5["high"].values.astype(float),
                                 df5["low"].values.astype(float),
                                 df5["close"].values.astype(float))
        for k in agg:
            agg[k] += st[k]
        idx = pd.to_datetime(ltf["timestamp"].values + MS_LTF, unit="ms", utc=True)
        s = pd.Series(eq, index=idx).resample("1D").last().ffill()
        out[asset] = s.pct_change().fillna(0.0)
    J = pd.concat(out, axis=1, join="inner").fillna(0.0)
    return pd.Series(0.5 * J["BTC"].values + 0.5 * J["ETH"].values, index=J.index), agg


def book(skh: pd.Series, w: float) -> pd.Series:
    tp = r02.al.tp01_baseline_daily()
    tp = tp.dropna()
    if tp.index.tz is None:
        tp.index = tp.index.tz_localize("UTC")
    tp.index = tp.index.floor("D")
    s = skh.copy()
    s.index = s.index.floor("D")
    J = pd.concat({"tp": tp, "skh": s}, axis=1, join="inner").fillna(0.0)
    return pd.Series((1 - w) * J["tp"].values + w * J["skh"].values, index=J.index)


def sh3(s: pd.Series) -> tuple[float, float, float, float]:
    m_all = metrics(s)
    m_is = metrics(s[s.index < HOLDOUT])
    m_ho = metrics(s[s.index >= HOLDOUT])
    return m_all["sharpe"], m_is["sharpe"], m_ho["sharpe"], m_all["maxdd"]


MODES = ("canonical", "hourly", "fastdetect", "onbook_tp", "onbook")
LABEL = {"canonical": "canonical  (backtest, fill-al-livello)",
         "hourly": "hourly     (PATH LIVE DI OGGI)",
         "fastdetect": "fastdetect (rilevazione INTRA-BARRA, exit a mercato)",
         "onbook_tp": "onbook_tp  (solo TP a limite)",
         "onbook": "onbook     (TP limite + SL stop-market)"}


def main() -> None:
    global SL_SLIP
    print("=" * 100)
    print("  T1 — SKH01: il degrado del path live e' recuperabile con ordini RESTING ON-BOOK?")
    print("=" * 100)
    print(f"\n  fee: taker {FEE_TAKER_SIDE:.4%}/lato, maker {FEE_MAKER_SIDE:.4%}/lato | "
          f"slippage stop-market {SL_SLIP:.2%} | hold-out {HOLDOUT}")

    # ------------------------------------------------ offset 0 (la griglia che gira davvero)
    print("\n" + "-" * 100)
    print("  A. OFFSET 0 — la griglia che gira davvero nel live")
    print("-" * 100)
    ser0 = {}
    print(f"  {'modalita':>40} {'SKH standalone':>28}   {'book 75/25':>26}")
    print(f"  {'':>40} {'FULL':>6} {'IS':>6} {'HOLD':>6} {'DD':>6}   "
          f"{'FULL':>6} {'IS':>6} {'HOLD':>6} {'DD':>6}")
    stats0 = {}
    for m in MODES:
        s, st = skh_series(0, m)
        ser0[m] = s
        stats0[m] = st
        a = sh3(s)
        b = sh3(book(s, 0.25))
        print(f"  {LABEL[m]:>40} {a[0]:>6.2f} {a[1]:>6.2f} {a[2]:>6.2f} {a[3]:>5.1%}   "
              f"{b[0]:>6.2f} {b[1]:>6.2f} {b[2]:>6.2f} {b[3]:>5.1%}")
    st = stats0["onbook"]
    tot = st["sl"] + st["tp"] + st["time"]
    print(f"\n  composizione delle uscite (off 0, BTC+ETH): {tot} trade — "
          f"SL {st['sl']} ({st['sl']/tot:.0%}) | TP {st['tp']} ({st['tp']/tot:.0%}) | "
          f"time {st['time']} ({st['time']/tot:.0%})")
    print(f"  di cui SL con GAP dentro la barra 5m (fill peggiore del livello): "
          f"{st['gap_sl']} = {st['gap_sl']/max(st['sl'],1):.0%} degli SL")

    # ------------------------------------------------ scomposizione del recupero
    print("\n" + "-" * 100)
    print("  B. DOVE VIVE IL DEGRADO — scomposizione del gap canonical -> live")
    print("-" * 100)
    b_can, b_hr = sh3(book(ser0["canonical"], 0.25)), sh3(book(ser0["hourly"], 0.25))
    b_tp, b_ob = sh3(book(ser0["onbook_tp"], 0.25)), sh3(book(ser0["onbook"], 0.25))
    gap_tot = b_can[0] - b_hr[0]
    print(f"  book 75/25, Sharpe FULL:  canonical {b_can[0]:.3f}  ->  live {b_hr[0]:.3f}   "
          f"(degrado {gap_tot:+.3f})")
    if abs(gap_tot) > 1e-9:
        print(f"    recupero col solo TP a limite : {b_tp[0]-b_hr[0]:+.3f}  "
              f"= {(b_tp[0]-b_hr[0])/gap_tot:>5.0%} del degrado   [NESSUN rischio nuovo]")
        print(f"    recupero TP + SL on-book      : {b_ob[0]-b_hr[0]:+.3f}  "
              f"= {(b_ob[0]-b_hr[0])/gap_tot:>5.0%} del degrado")
        print(f"    residuo NON recuperabile      : {b_can[0]-b_ob[0]:+.3f}  "
              f"= {(b_can[0]-b_ob[0])/gap_tot:>5.0%} del degrado   [gap + entry del cron]")

    # ------------------------------------------------ banda dei 23 offset (la lente onesta)
    print("\n" + "-" * 100)
    print("  C. BANDA DEI 23 OFFSET — la lente de-luckata (off 0 e' al 93-98 pctl, audit 02/07)")
    print("-" * 100)
    rows = {m: [] for m in MODES}
    for off in r02.OFFSETS:
        for m in MODES:
            s, _ = skh_series(off, m)
            rows[m].append(sh3(book(s, 0.25)))
    print(f"  {'modalita':>40} {'FULL med':>9} {'IS med':>8} {'HOLD med':>9} "
          f"{'DD med':>7} {'HOLD min':>9} {'HOLD max':>9}")
    med = {}
    A = {m: np.array(rows[m]) for m in MODES}
    for m in MODES:
        med[m] = np.median(A[m], axis=0)
        print(f"  {LABEL[m]:>40} {med[m][0]:>9.2f} {med[m][1]:>8.2f} {med[m][2]:>9.2f} "
              f"{med[m][3]:>6.1%} {A[m][:, 2].min():>9.2f} {A[m][:, 2].max():>9.2f}")

    # ⚠️ gli offset sono coppie APPAIATE: la statistica giusta e' la mediana delle DIFFERENZE
    # per-offset, non la differenza delle mediane (che confronta offset diversi fra loro).
    print("\n  differenze APPAIATE per-offset (mediana [min, max] sui 23 offset):")
    print(f"  {'':>40} {'dFULL':>22} {'dHOLD':>22} {'dDD':>20}")
    for m in ("canonical", "fastdetect", "onbook_tp", "onbook"):
        d0 = A[m][:, 0] - A["hourly"][:, 0]
        d2 = A[m][:, 2] - A["hourly"][:, 2]
        d3 = A[m][:, 3] - A["hourly"][:, 3]
        print(f"  {LABEL[m]:>40} "
              f"{np.median(d0):>+7.3f} [{d0.min():+.2f},{d0.max():+.2f}] "
              f"{np.median(d2):>+7.3f} [{d2.min():+.2f},{d2.max():+.2f}] "
              f"{np.median(d3):>+6.2%} [{d3.min():+.1%},{d3.max():+.1%}]")

    dmax = np.median(A["canonical"][:, 0] - A["hourly"][:, 0])
    dob = np.median(A["onbook"][:, 0] - A["hourly"][:, 0])
    dtp = np.median(A["onbook_tp"][:, 0] - A["hourly"][:, 0])
    print(f"\n  degrado recuperabile al massimo (canonical-hourly, mediana appaiata): {dmax:+.3f} FULL")
    if abs(dmax) > 1e-9:
        print(f"    recuperato dal solo TP a limite: {dtp:+.3f} = {dtp/dmax:>5.0%}")
        print(f"    recuperato da TP + SL on-book  : {dob:+.3f} = {dob/dmax:>5.0%}")
    print("\n  in quanti dei 23 offset la modalita' MIGLIORA il path live di oggi?")
    for m in ("fastdetect", "onbook_tp", "onbook"):
        nf = int((A[m][:, 0] - A["hourly"][:, 0] > 0).sum())
        nh = int((A[m][:, 2] - A["hourly"][:, 2] > 0).sum())
        nd = int((A[m][:, 3] - A["hourly"][:, 3] < 0).sum())
        print(f"    {LABEL[m]:>40}:  FULL {nf}/23   HOLD {nh}/23   DD {nd}/23")
    # il lato SL da solo = differenza fra onbook (TP+SL) e onbook_tp (solo TP)
    dsl = A["onbook"][:, 0] - A["onbook_tp"][:, 0]
    print(f"\n  contributo del SOLO lato SL on-book (onbook - onbook_tp), FULL: "
          f"mediana {np.median(dsl):+.3f} [{dsl.min():+.2f},{dsl.max():+.2f}], "
          f"positivo in {int((dsl > 0).sum())}/23 offset")

    # ------------------------------------------------ robustezza alla slippage assunta
    print("\n" + "-" * 100)
    print("  D. SENSIBILITA' ALLA SLIPPAGE ASSUNTA sullo stop-market (off 0, book 75/25)")
    print("-" * 100)
    keep = SL_SLIP
    print(f"  {'slippage':>10} {'FULL':>7} {'HOLD':>7} {'DD':>7}  (live di oggi: "
          f"FULL {b_hr[0]:.2f} / HOLD {b_hr[2]:.2f})")
    for slip in (0.0000, 0.0005, 0.0020, 0.0050, 0.0100):
        SL_SLIP = slip
        s, _ = skh_series(0, "onbook")
        r = sh3(book(s, 0.25))
        flag = "" if r[0] > b_hr[0] else "   <- peggio del live"
        print(f"  {slip:>9.2%} {r[0]:>7.2f} {r[2]:>7.2f} {r[3]:>6.1%}{flag}")
    SL_SLIP = keep
    print("\n" + "=" * 100)


if __name__ == "__main__":
    main()