PythagorasGoal/scripts/research/r0822_slip_audit.py

"""r0822 SLIP-AUDIT — l'attrito VERO del libro live: quanto costa davvero un fill.

DOMANDA
-------
Il progetto modella le fee con cura chirurgica (3,50 bps/lato misurati sui fill reali il 21/08)
ma ASSUME il fill al prezzo di chiusura della barra. Lo **slippage realizzato** del libro live
non e' mai stato misurato: e' dichiarato "rischio #1" per XSR01 e semplicemente ignorato per
TP01/SKH01. Ci sono 18 esecuzioni reali dal 2026-07-14. Questo script le misura.

La domanda decidibile e' UNA: i backtest girano a `fee_rt=0.001` (10 bps RT) mentre la fee vera
e' 3,50 bps/lato = 7 bps RT. C'e' **3 bps RT = 1,50 bps/LATO** di margine gia' pagato dal modello.
**Lo slippage realizzato ci sta dentro?**

TRE DIFETTI DEL DATO TROVATI PRIMA DI MISURARE (e riparati/dichiarati qui)
-------------------------------------------------------------------------
(1) **`ts_utc` nel log NON e' l'ora del fill.** `book_execute.py:213` scrive
    `ts_utc=str(pd.Timestamp(r['last_data']))` = l'etichetta dell'ULTIMA BARRA DEL FEED, non
    l'orologio. Tutti e 18 i record dicono `00:00:00`. L'ora vera si recupera solo dalle
    intestazioni di `logs/cron_book.log` (`===== <ISO> cron_book =====`). Senza questo passo
    ogni confronto "al prezzo dell'ora del fill" misurerebbe la mezzanotte.
    Prova che serviva: il fill #1 e' etichettato `2026-07-08` ed e' stato eseguito il
    **2026-07-14T14:00Z** — sei giorni dopo (feed-freeze del 14/07, gia' noto al progetto).
(2) **Il feed certificato NON e' lo strumento che si trada.** `rebuild_history.DERIBIT_INSTR`
    prende `BTC/USD:BTC` / `ETH/USD:ETH` = i perpetual **INVERSE**; `deribit.INSTRUMENT` esegue su
    `BTC_USDC-PERPETUAL` / `ETH_USDC-PERPETUAL` = i **LINEARI USDC**. Il commento nel sorgente dice
    "inverse ~ lineare entro 3 bps": **3 bps e' esattamente la taglia della domanda**. Misurare lo
    slippage contro il feed certificato significherebbe misurare la BASE, non lo slippage.
    Qui la base viene MISURATA (non assunta) e il riferimento vero e' lo strumento tradato.
    Stessa forma del difetto di `fee_watch` del 21/08.
(3) **Convenzione LINEARE**: `amount` = quantita' BASE, `fee` gia' in USDC ->
    `fee_bps = fee / (filled * price) * 1e4`. La convenzione inverse darebbe ~8 ordini di
    grandezza di errore SENZA sollevare nulla. Il controllo e' stampato in chiaro.

TRE RIFERIMENTI, TENUTI SEPARATI (il punto 1c della missione)
------------------------------------------------------------
Il modello decide sulla barra GIORNALIERA, il cron esegue all'ORA: sono grandezze diverse.
  REF-A  "modello"  = close della barra 1d etichettata `last_data` (= il prezzo a cui il
                      BACKTEST assume di riempire). Misura modello-vs-realta', NON slippage.
  REF-B  "5m/1h"    = close dell'ultima barra CHIUSA prima del fill sullo strumento TRADATO.
                      Questo, e solo questo, e' slippage d'esecuzione.
  REF-C  "VWAP 1h"  = VWAP delle 12 barre 5m dell'ora che contiene il fill (benchmark
                      NON causale, dichiarato tale).
Segno orientato al lato: `costo_bps = lato * (fill - ref)/ref * 1e4`, lato +1 buy / -1 sell.
Positivo = costo, negativo = guadagno.

LA STATISTICA CHE HA POTENZA (e perche' quella in bps non ne ha)
----------------------------------------------------------------
Con n=18 la via in bps e' persa in partenza: il rumore e' la deriva del prezzo fra la chiusura
della barra 5m e il fill, che vale decine di bps. La statistica giusta e' NORMALIZZATA dal range
della barra: il nostro fill e' esso stesso un print della barra 5m che lo contiene, quindi
  q = (fill - low)/(high - low)  per un BUY,   q = 1 - quello  per un SELL.
Se i nostri ordini market avessero impatto, un BUY stamperebbe AL MASSIMO della barra -> q -> 1.
Senza impatto E[q] = 0.5, e il null NON si assume uniforme: si MISURA su ~18.000 barre vere.

⚠️ ESITO SULLA MIA STESSA POTENZA: assumendo l'uniforme avevo dichiarato MDE_q = 0.19 di range
(~5 bps). Il null misurato e' 1.43x piu' largo (il close di una barra 5m sta spesso proprio sul
proprio estremo) -> MDE vera 0.27 = ~7.2 bps, contro 7.8 della via in bps. **La normalizzazione
non compra potenza utile.** Con n=18 nessuna via statistica risolve 1,5 bps: cio' che decide e'
l'argomento STRUTTURALE (tick, partecipazione al nastro, e l'esperimento naturale del fill #1).

Uso: nice -n 19 timeout 900 uv run python scripts/research/r0822_slip_audit.py
     (--no-net salta il fetch dello strumento lineare e lo dichiara non misurato)
"""
from __future__ import annotations

import json
import os
import re
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))

SCRATCH = Path(os.environ.get("SLIP_SCRATCH", "/tmp/claude-1001/-opt-docker-PythagorasGoal/"
                              "b6cc75e7-14f8-4c32-bd07-ab8a0d2aaee6/scratchpad"))

EXEC_LOG = ROOT / "data" / "live" / "book_executions.jsonl"
CRON_LOG = ROOT / "logs" / "cron_book.log"

# Sensibilita' PUBBLICATA del libro (r0726_fee_sensitivity.py, CLAUDE.md 26/07).
# Unita': per bps PER LATO (verificato sulla tabella: 0->15 bps/lato = Sharpe 1.849->1.597).
DSHARPE_PER_BPS_SIDE = -0.017
DCAGR_PER_BPS_SIDE = -0.23          # punti percentuali di CAGR
# Margine gia' pagato dal modello: backtest 10 bps RT vs fee vera 7 bps RT = 3 bps RT.
MARGIN_BPS_SIDE = 1.5
FEE_TAKER_BPS_SIDE = 3.50           # tier base misurato (fee_watch, 21/08)

# Strumento TRADATO (src/live/deribit.INSTRUMENT) vs strumento del FEED certificato
# (scripts/analysis/rebuild_history.DERIBIT_INSTR). NON sono lo stesso: vedi difetto (2).
CCXT_LINEAR = {"BTC": "BTC/USDC:USDC", "ETH": "ETH/USDC:USDC"}
CCXT_INVERSE = {"BTC": "BTC/USD:BTC", "ETH": "ETH/USD:ETH"}

RNG = np.random.default_rng(822)
N_BOOT = 20_000


# ----------------------------------------------------------------------------- fills
def _parse_cron_fills() -> list[dict]:
    """Ora VERA di ogni fill, dalle intestazioni di cron_book.log. Vedi difetto (1)."""
    if not CRON_LOG.exists():
        raise SystemExit(f"manca {CRON_LOG}: l'ora vera del fill non e' recuperabile altrove")
    head = re.compile(r"^===== (\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z) cron_book =====")
    fill = re.compile(r"^\s+-> (BUY|SELL) ([\d.]+) @ \$([\d,.]+) fee ([\d.]+) \((\w+)\)")
    out, cur = [], None
    with open(CRON_LOG, errors="replace") as f:
        for line in f:
            m = head.match(line)
            if m:
                cur = pd.Timestamp(m.group(1))
                continue
            m = fill.match(line)
            if m:
                out.append(dict(fill_ts=cur, side=m.group(1).lower(),
                                amt=float(m.group(2)), px=float(m.group(3).replace(",", "")),
                                fee=float(m.group(4)), status=m.group(5)))
    return out


def load_fills() -> pd.DataFrame:
    """JSONL (verita' sui numeri) + cron log (verita' sull'ora), appaiati e VERIFICATI."""
    recs = [json.loads(l) for l in EXEC_LOG.read_text().splitlines() if l.strip()]
    cron = _parse_cron_fills()
    if len(recs) != len(cron):
        raise SystemExit(f"appaiamento impossibile: {len(recs)} righe jsonl vs {len(cron)} fill "
                         f"nel cron log. Non invento un'ora: mi fermo.")
    rows = []
    for i, (r, c) in enumerate(zip(recs, cron)):
        # entrambi i file sono append-only dallo stesso processo -> stesso ordine. Lo PROVO.
        ok = (r["side"] == c["side"]
              and abs(round(r["filled"], 4) - round(c["amt"], 4)) < 1e-9
              and abs(round(r["price"], 1) - round(c["px"], 1)) < 0.051
              and abs(round(r["fee"], 5) - round(c["fee"], 5)) < 1e-9)
        if not ok:
            raise SystemExit(f"riga {i}: jsonl e cron log NON combaciano ({r} vs {c}) -> mi fermo")
        if c["status"] != "OK":
            continue
        rows.append(dict(
            i=i, fill_ts=c["fill_ts"], model_bar=pd.Timestamp(r["ts_utc"], tz="UTC"),
            asset=r["asset"], side=r["side"], lato=1.0 if r["side"] == "buy" else -1.0,
            filled=float(r["filled"]), price=float(r["price"]), fee=float(r["fee"]),
            notional=float(r["filled"]) * float(r["price"]),
            skh=r.get("skh_sign"), tp_frac=r.get("tp_frac"),
        ))
    return pd.DataFrame(rows)


# ----------------------------------------------------------------------------- prezzi
def _certified(asset: str, tf: str, t0, t1) -> pd.DataFrame:
    """Feed certificato su disco = strumento INVERSE. Letto a colonne e tagliato subito (RAM)."""
    f = ROOT / "data" / "raw" / f"{asset.lower()}_{tf}.parquet"
    d = pd.read_parquet(f, columns=["timestamp", "open", "high", "low", "close", "volume"])
    d["ts"] = pd.to_datetime(d["timestamp"], unit="ms", utc=True)
    d = d[(d["ts"] >= t0) & (d["ts"] <= t1)].drop(columns=["timestamp"]).reset_index(drop=True)
    return d


def _fetch_linear(asset: str, t0, t1) -> pd.DataFrame | None:
    """Strumento TRADATO (lineare USDC) via ccxt pubblico. Cache su scratchpad, pacing gentile.

    Il progetto ha una storia di rate-limit per-IP (29-30/07): qui sono ~13 pagine per asset,
    distanziate. Nessuna scrittura sotto data/."""
    SCRATCH.mkdir(parents=True, exist_ok=True)
    cache = SCRATCH / f"lin_{asset.lower()}_5m.parquet"
    if cache.exists():
        d = pd.read_parquet(cache)
        if d["ts"].min() <= t0 and d["ts"].max() >= t1 - pd.Timedelta(minutes=10):
            return d
    try:
        import ccxt, time
        ex = ccxt.deribit({"enableRateLimit": True})
        sym, since, rows = CCXT_LINEAR[asset], int(t0.timestamp() * 1000), []
        end = int(t1.timestamp() * 1000)
        while since < end:
            o = ex.fetch_ohlcv(sym, "5m", since=since, limit=1000)
            if not o:
                break
            rows += o
            since = o[-1][0] + 300_000
            time.sleep(0.35)
        if not rows:
            return None
        d = pd.DataFrame(rows, columns=["timestamp", "open", "high", "low", "close", "volume"])
        d = d.drop_duplicates("timestamp").sort_values("timestamp")
        d["ts"] = pd.to_datetime(d["timestamp"], unit="ms", utc=True)
        d = d.drop(columns=["timestamp"]).reset_index(drop=True)
        d.to_parquet(cache)
        return d
    except Exception as e:                                   # rete assente = si dichiara
        print(f"    ! fetch lineare {asset} fallito: {type(e).__name__}: {str(e)[:90]}")
        return None


def _last_closed(bars: pd.DataFrame, t) -> pd.Series | None:
    """Ultima barra CHIUSA a `t`: una barra etichettata b chiude a b+durata."""
    if bars is None or bars.empty:
        return None
    step = bars["ts"].diff().median()
    ok = bars[bars["ts"] + step <= t]
    return None if ok.empty else ok.iloc[-1]


def _containing(bars: pd.DataFrame, t) -> pd.Series | None:
    if bars is None or bars.empty:
        return None
    step = bars["ts"].diff().median()
    ok = bars[(bars["ts"] <= t) & (t < bars["ts"] + step)]
    return None if ok.empty else ok.iloc[-1]


# ----------------------------------------------------------------------------- stats
def _equity_per_fill(ts_list) -> "pd.Series":
    """Equity del conto AL MOMENTO di ciascun fill, dal libro di bordo (`data/live/trades.db`).

    ⚠️ PERCHE' SERVE (2026-08-25). La sezione SCALA divideva TUTTE le partecipazioni per UNA
    equity sola. Andava bene finche' il conto stava fermo a ~$636; il giorno del versamento
    ($667 -> $2.067) qualunque scelta di quel singolo numero diventa sbagliata, perche' i fill
    del campione sono stati eseguiti a taglie di conto DIVERSE. Il caso peggiore osservato
    (22% di una barra 5m) e' un fill da $74 fatto a ~$650: normalizzarlo sull'equity di oggi lo
    sottostima di 3,2x, sull'equity di allora sovrastima gli altri.
    La grandezza invariante e' **partecipazione per dollaro di equity**, e si ottiene solo
    appaiando ogni fill alla SUA equity.
    """
    import sqlite3
    import pandas as _pd
    f = ROOT / "data" / "live" / "trades.db"
    ts = _pd.DatetimeIndex(_pd.to_datetime(list(ts_list), utc=True))
    try:
        with sqlite3.connect(f"file:{f}?mode=ro", uri=True) as con:
            e = _pd.read_sql("SELECT ts_utc, equity FROM equity ORDER BY ts_utc", con)
        e["ts_utc"] = _pd.to_datetime(e["ts_utc"], utc=True)
        e = e.dropna().set_index("ts_utc")["equity"].sort_index()
        # BACKWARD: l'equity in vigore quando il fill e' avvenuto, mai una lettura futura.
        out = _pd.Series(e.reindex(ts, method="ffill").values, index=ts)
        if out.notna().any():
            return out
    except Exception as ex:  # noqa: BLE001
        print(f"      ⚠️ equity per fill non leggibile ({type(ex).__name__}): "
              f"ricado sull'equity unica")
    return _pd.Series([float("nan")] * len(ts), index=ts)


def _equity_osservata(default: float = 636.0) -> float:
    """L'equity reale dal watermark del libro live — DERIVATA, non ridichiarata (P1).

    Se il file non c'e' o non e' leggibile si ricade sul valore storico e **lo si dice**: un
    default silenzioso qui rimetterebbe esattamente il difetto che questa funzione ripara.
    """
    import json
    f = ROOT / "data" / "live" / "equity_seen.json"
    try:
        v = float(json.loads(f.read_text())["real_equity"])
        if v > 0:
            return v
    except Exception as e:  # noqa: BLE001
        print(f"      ⚠️ watermark non leggibile ({type(e).__name__}): uso il default ${default:,.0f}")
    return default


def ci_mean(x, boot=True):
    x = np.asarray(x, float)
    x = x[np.isfinite(x)]
    n = len(x)
    if n < 2:
        return dict(n=n, mean=np.nan, lo=np.nan, hi=np.nan, sd=np.nan, med=np.nan)
    m, sd = x.mean(), x.std(ddof=1)
    if boot:
        bs = RNG.choice(x, size=(N_BOOT, n), replace=True).mean(axis=1)
        lo, hi = np.percentile(bs, [2.5, 97.5])
    else:
        from scipy import stats
        h = stats.t.ppf(0.975, n - 1) * sd / np.sqrt(n)
        lo, hi = m - h, m + h
    return dict(n=n, mean=m, lo=lo, hi=hi, sd=sd, med=float(np.median(x)))


def fmt(d, u="bps"):
    if not np.isfinite(d["mean"]):
        return "n/d"
    z = "" if (d["lo"] <= 0 <= d["hi"]) else "  <- ZERO ESCLUSO"
    return f"{d['mean']:+7.2f} {u}  IC95 [{d['lo']:+7.2f}, {d['hi']:+7.2f}]  (med {d['med']:+7.2f}, n={d['n']}){z}"


def main():
    print("=" * 96)
    print("  r0822 SLIP-AUDIT — l'attrito VERO del libro live (18 esecuzioni, dal 2026-07-14)")
    print("=" * 96)

    fills = load_fills()
    t0 = fills["fill_ts"].min().floor("D") - pd.Timedelta(days=3)
    t1 = fills["fill_ts"].max().ceil("h") + pd.Timedelta(hours=2)

    # ---------------------------------------------------------------- 0. difetti del dato
    print("\n[0] DIFETTI DEL DATO — trovati PRIMA di misurare\n" + "-" * 96)
    lag = (fills["fill_ts"] - fills["model_bar"]).dt.total_seconds() / 3600.0
    print(f"  (1) `ts_utc` nel jsonl = etichetta dell'ultima barra del feed, NON l'ora del fill.")
    print(f"      18/18 record dicono 00:00:00. Ora vera recuperata da logs/cron_book.log.")
    print(f"      Scarto fill - etichetta: mediana {lag.median():.1f}h, max {lag.max():.1f}h "
          f"({(lag > 26).sum()} fill oltre le 26h).")
    if (lag > 26).any():
        w = fills.loc[lag.idxmax()]
        print(f"      Il caso estremo e' il fill #{w['i']}: etichetta {w['model_bar'].date()}, "
              f"eseguito {w['fill_ts']} = feed-freeze del 14/07 (gia' noto al progetto).")
    print(f"  (2) Feed certificato = INVERSE ({CCXT_INVERSE['BTC']}), libro tradato = LINEARE "
          f"({CCXT_LINEAR['BTC']}).")
    print(f"      La base fra i due e' misurata sotto, non assunta.")

    # ---------------------------------------------------------------- 1. convenzione fee
    print("\n[1] CONTROLLO DI CONVENZIONE (lineare USDC) — la trappola da ~8 ordini di grandezza\n"
          + "-" * 96)
    fills["fee_bps"] = fills["fee"] / fills["notional"] * 1e4
    fb = ci_mean(fills["fee_bps"])
    print(f"  fee_bps = fee_usdc / (filled * price) * 1e4   [LINEARE: amount=base, fee gia' in USDC]")
    print(f"    media {fb['mean']:.3f} bps/lato · mediana {fb['med']:.3f} · "
          f"min {fills['fee_bps'].min():.3f} · max {fills['fee_bps'].max():.3f}")
    print(f"    tier base dichiarato da fee_watch: {FEE_TAKER_BPS_SIDE:.2f} bps/lato -> "
          f"scarto {fb['med'] - FEE_TAKER_BPS_SIDE:+.3f} bps")
    bad = (fills["fee"] / fills["filled"] * 1e4).median()
    print(f"    [se avessi usato la convenzione INVERSE: {bad:,.0f} bps -> "
          f"{bad / FEE_TAKER_BPS_SIDE:,.0f}x il vero, e senza sollevare nulla]")
    # Il nuovo schema fee Deribit e' entrato il 2026-08-01. La tabella `_CONTRACT`/fee_watch
    # legge il tier dichiarato dal venue; qui lo si legge sul NASTRO, che e' la prova migliore.
    CUT = pd.Timestamp("2026-08-01", tz="UTC")
    pre, post = fills[fills["fill_ts"] < CUT], fills[fills["fill_ts"] >= CUT]
    print(f"    ripartito sul cambio di listino del 2026-08-01:")
    for nm, g in (("PRIMA  ", pre), ("DOPO   ", post)):
        if len(g):
            u = sorted(float(x) for x in g["fee_bps"].round(2).unique())
            print(f"      {nm} n={len(g):2d}  fee/lato {u}  "
                  f"(nozionale ${g['notional'].min():.0f}-${g['notional'].max():.0f})")
    if len(pre) and len(post):
        print(f"    => il taglio {pre['fee_bps'].median():.2f} -> {post['fee_bps'].median():.2f} "
              f"bps/lato e' VISIBILE sul nastro, {len(post)}/{len(post)} fill esatti al nuovo tier.")
        print(f"       Non e' un effetto taglia (i due fill a {pre['fee_bps'].median():.2f} bps sono da "
              f"${pre['notional'].median():.0f}, sopra la mediana del campione).")
        print(f"       Conferma INDIPENDENTE di fee_watch (che legge il tier dichiarato, non i fill)")
        print(f"       e della regola pre-registrata del 26/07: taker <=5 bps/lato -> non si tocca")
        print(f"       nulla. Il listino si e' mosso NEL VERSO BUONO.")

    # ---------------------------------------------------------------- 2. potenza, PRIMA
    print("\n[2] POTENZA DICHIARATA — calcolata PRIMA di guardare lo slippage\n" + "-" * 96)
    sig5 = {}
    for a in sorted(fills["asset"].unique()):
        d = _certified(a, "5m", t0, t1)
        r = np.log(d["close"]).diff().dropna() * 1e4
        sig5[a] = float(r.std())
    sig = float(np.mean(list(sig5.values())))
    n = len(fills)
    mde_bps = 2.80 * sig / np.sqrt(n)          # 80% potenza, bilaterale alpha=0.05
    print(f"  Rumore strutturale: sigma dei ritorni 5m nella finestra = "
          + " · ".join(f"{a} {v:.1f} bps" for a, v in sig5.items()) + f" (media {sig:.1f})")
    print(f"  Il riferimento 5m dista dal fill 0-5 min -> il rumore e' di QUELL'ordine.")
    print(f"  MDE in bps a n={n}, potenza 80%, alpha 5%:  2.80 * {sig:.1f} / sqrt({n}) = "
          f"**{mde_bps:.1f} bps/lato**")
    print(f"  Il margine da difendere e' {MARGIN_BPS_SIDE:.1f} bps/lato (3 bps RT).")
    print(f"  => VERDETTO DI POTENZA, dichiarato prima del risultato: la via in bps NON puo'")
    print(f"     risolvere {MARGIN_BPS_SIDE:.1f} bps ({mde_bps / MARGIN_BPS_SIDE:.0f}x troppo grossolana). "
          f"Qualunque IC in bps conterra' lo zero;")
    print(f"     un 'non distinguibile da zero' li' NON e' una prova di assenza di slippage.")
    mde_q = 2.80 * (1 / np.sqrt(12)) / np.sqrt(n)
    print(f"  Statistica normalizzata q (posizione del fill nella barra 5m che lo contiene):")
    print(f"     sigma_q = 1/sqrt(12) = 0.289 -> MDE_q = {mde_q:.3f} di range di barra.")
    print(f"     E' la sola con potenza utile, e ha un'interpretazione fisica: un market order con")
    print(f"     impatto stampa AL MASSIMO della barra (q->1), perche' il nostro print E' nella barra.")

    # ---------------------------------------------------------------- 3. base inverse/lineare
    print("\n[3] BASE inverse-vs-lineare — MISURATA, non assunta (difetto 2)\n" + "-" * 96)
    lin, inv = {}, {}
    use_net = "--no-net" not in sys.argv
    for a in sorted(fills["asset"].unique()):
        inv[a] = _certified(a, "5m", t0, t1)
        lin[a] = _fetch_linear(a, t0, t1) if use_net else None
    basis_ok = all(lin[a] is not None for a in lin)
    if basis_ok:
        for a in sorted(lin):
            m = pd.merge(inv[a][["ts", "close"]], lin[a][["ts", "close"]], on="ts",
                         suffixes=("_inv", "_lin"))
            b = (m["close_lin"] - m["close_inv"]) / m["close_inv"] * 1e4
            print(f"  {a}: n={len(m):,} barre 5m · base mediana {b.median():+.2f} bps · "
                  f"media {b.mean():+.2f} · IQR [{b.quantile(.25):+.2f}, {b.quantile(.75):+.2f}] · "
                  f"|base| p95 {b.abs().quantile(.95):.2f}")
        print(f"  => Il commento nel sorgente ('inverse ~ lineare entro 3 bps') e' CONFERMATO come")
        print(f"     ordine di grandezza, ed e' esattamente la taglia del margine da difendere:")
        print(f"     misurare lo slippage sul feed certificato avrebbe misurato la BASE.")
    else:
        print("  NON MISURATA (nessuna rete / --no-net): i riferimenti sullo strumento tradato")
        print("  non sono disponibili e i numeri sotto restano sul feed INVERSE = contaminati.")

    # ---------------------------------------------------------------- 4. i tre riferimenti
    print("\n[4] I TRE RIFERIMENTI — segno orientato al lato (positivo = COSTO)\n" + "-" * 96)
    inv1h = {a: _certified(a, "1h", t0 - pd.Timedelta(days=10), t1)
             for a in sorted(fills["asset"].unique())}
    daily = {}
    for a, d in inv1h.items():
        g = d.set_index("ts").resample("1D")
        daily[a] = pd.DataFrame(dict(close=g["close"].last())).dropna()

    rows = []
    for _, f in fills.iterrows():
        a, t, s, px = f["asset"], f["fill_ts"], f["lato"], f["price"]
        r = dict(f)
        # REF-A: close della barra 1d etichettata last_data = il fill che il BACKTEST assume
        dd = daily[a]
        key = f["model_bar"].tz_convert("UTC").normalize()
        r["ref_A"] = float(dd.loc[key, "close"]) if key in dd.index else np.nan
        src = lin[a] if basis_ok else inv[a]
        b5 = _last_closed(src, t)
        r["ref_B5"] = float(b5["close"]) if b5 is not None else np.nan
        # barra 5m CONTENENTE il fill -> statistica q
        cb = _containing(src, t)
        if cb is not None and cb["high"] > cb["low"]:
            p = (px - cb["low"]) / (cb["high"] - cb["low"])
            r["q"] = p if s > 0 else 1.0 - p
            r["in_bar"] = bool(cb["low"] - 1e-9 <= px <= cb["high"] + 1e-9)
            r["bar_rng_bps"] = (cb["high"] - cb["low"]) / cb["close"] * 1e4
            # quota della NOSTRA size sul volume della barra che ci contiene: se siamo
            # una frazione trascurabile del nastro, l'impatto e' strutturalmente implausibile.
            r["partecip"] = f["filled"] / cb["volume"] if cb["volume"] > 0 else np.nan
        else:
            r["q"], r["in_bar"], r["bar_rng_bps"], r["partecip"] = np.nan, None, np.nan, np.nan
        # REF-B1h e REF-C VWAP dell'ora contenente
        h0 = t.floor("h")
        hb = src[(src["ts"] >= h0) & (src["ts"] < h0 + pd.Timedelta(hours=1))]
        r["ref_C"] = float((hb["close"] * hb["volume"]).sum() / hb["volume"].sum()) \
            if len(hb) and hb["volume"].sum() > 0 else np.nan
        prev = src[src["ts"] + pd.Timedelta(minutes=5) <= h0]
        r["ref_B1"] = float(prev.iloc[-1]["close"]) if len(prev) else np.nan
        for k in ("A", "B5", "B1", "C"):
            v = r.get(f"ref_{k}")
            r[f"slip_{k}"] = s * (px - v) / v * 1e4 if v and np.isfinite(v) else np.nan
        rows.append(r)
    R = pd.DataFrame(rows)

    lab = {"A": "REF-A  barra 1d del modello (= fill assunto dal BACKTEST)",
           "B5": "REF-B5 close ultima barra 5m CHIUSA  (= SLIPPAGE d'esecuzione)",
           "B1": "REF-B1 close ultima barra 1h CHIUSA",
           "C": "REF-C  VWAP dell'ora contenente (non causale)"}
    strumento = "LINEARE (tradato)" if basis_ok else "INVERSE (certificato) — CONTAMINATO da base"
    print(f"  riferimenti B5/B1/C calcolati sullo strumento {strumento}\n")
    for k in ("A", "B5", "B1", "C"):
        print(f"  {lab[k]}\n      {fmt(ci_mean(R[f'slip_{k}']))}")

    print(f"\n  ⚠️  REF-A NON e' slippage e il suo SEGNO non e' un guadagno. La barra 1d etichettata")
    print(f"      `last_data` chiude fino a ~23h DOPO il fill: su questa finestra (BTC 66k->78k in")
    print(f"      4 giorni) un trend-follower che compra alle 01:07 batte per costruzione la")
    print(f"      chiusura serale, ma il backtest tiene la posizione DA quella chiusura — e' una")
    print(f"      differenza di FASE, non un attrito. La domanda 'quanto costa la barra parziale'")
    print(f"      e' gia' stata misurata bene il 26/07 (r0726_tp01_partial_day, 24 ancore appaiate:")
    print(f"      ±0.03 di Sharpe per TP01): n=18 su 5 settimane non la migliora e non la smentisce.")
    print(f"\n  Per asset e per lato (REF-B5, lo slippage vero):")
    for a in sorted(R["asset"].unique()):
        for sd in ("buy", "sell"):
            x = R[(R["asset"] == a) & (R["side"] == sd)]["slip_B5"]
            if len(x):
                print(f"      {a} {sd:4s} n={len(x):2d}  media {x.mean():+7.2f}  med {x.median():+7.2f}")

    # La media di REF-B5 e' dominata da UN giorno: la si mostra con e senza, e si dice quale
    # si cita. Non e' cherry-picking se si pubblicano entrambe e la regola e' dichiarata.
    wild = R.nlargest(3, "bar_rng_bps")
    day = wild["fill_ts"].dt.date.mode()[0]
    keep = R[R["fill_ts"].dt.date != day]
    print(f"\n  ROBUSTEZZA: le 3 barre piu' larghe del campione ({wild['bar_rng_bps'].min():.0f}-"
          f"{wild['bar_rng_bps'].max():.0f} bps di range) cadono tutte il {day}")
    print(f"      (ETH +14% in 6 ore). Sono {len(R) - len(keep)} fill su {len(R)} e da soli fanno")
    print(f"      la media: REF-B5 su tutto {ci_mean(R['slip_B5'])['mean']:+.2f} bps -> "
          f"{ci_mean(keep['slip_B5'])['mean']:+.2f} bps senza quel giorno")
    print(f"      (IC95 [{ci_mean(keep['slip_B5'])['lo']:+.2f}, {ci_mean(keep['slip_B5'])['hi']:+.2f}], "
          f"n={len(keep)}). La MEDIANA invece non si muove: "
          f"{R['slip_B5'].median():+.2f} -> {keep['slip_B5'].median():+.2f}.")
    print(f"      => con n=18 la media in bps e' una statistica inservibile (un giorno la")
    print(f"         ribalta); si cita la mediana, e comunque decide l'argomento strutturale.")

    # ---------------------------------------------------------------- 5. la statistica con potenza
    print("\n[5] LA STATISTICA CON POTENZA — q, posizione del fill nella barra 5m\n" + "-" * 96)
    q = R["q"].dropna()
    dq = ci_mean(q)
    print(f"  q = (fill-low)/(high-low) per i BUY, 1-quello per i SELL. E[q]=0.5 senza impatto.")
    print(f"      media q {dq['mean']:.3f}  IC95 [{dq['lo']:.3f}, {dq['hi']:.3f}]  "
          f"(med {dq['med']:.3f}, n={dq['n']})")
    print(f"      scostamento da 0.5: {dq['mean'] - 0.5:+.3f}  (MDE dichiarata {mde_q:.3f})")
    outside = (R["in_bar"] == False).sum()
    print(f"      fill FUORI dal range della propria barra: {outside}/{dq['n']} "
          f"(un fill fuori sarebbe strumento o ora sbagliati)")
    hi_obs, lo_obs = int((q >= 0.99).sum()), int((q <= 0.01).sum())
    print(f"      fill sull'ESTREMO avverso (q>=0.99 = firma di impatto): {hi_obs}/{dq['n']}"
          f" · sull'estremo favorevole (q<=0.01): {lo_obs}/{dq['n']}")
    rng_med = float(R["bar_rng_bps"].median())
    print(f"      range mediano di barra 5m: {rng_med:.1f} bps -> lo scostamento misurato vale")
    print(f"      ~{(dq['mean'] - 0.5) * rng_med:+.2f} bps/lato "
          f"[IC95 {(dq['lo'] - 0.5) * rng_med:+.2f}, {(dq['hi'] - 0.5) * rng_med:+.2f}]")

    # -- null EMPIRICO: "0.5" e' un'assunzione (uniforme). Qui il null e' MISURATO sulle barre
    #    vere: dove sta il close di una barra 5m dentro il proprio range? Segno randomizzato
    #    per imitare l'orientamento al lato.
    qc = []
    for a in sorted(fills["asset"].unique()):
        src = lin[a] if basis_ok else inv[a]
        m = src[src["high"] > src["low"]]
        qc.append(((m["close"] - m["low"]) / (m["high"] - m["low"])).to_numpy())
    qc = np.concatenate(qc)
    flip = RNG.random((N_BOOT, dq["n"])) < 0.5
    draw = RNG.choice(qc, size=(N_BOOT, dq["n"]), replace=True)
    null = np.where(flip, draw, 1.0 - draw).mean(axis=1)
    pval = float((null >= dq["mean"]).mean())
    print(f"\n  NULL EMPIRICO (non 'uniforme per assunzione'): posizione del close nel proprio")
    print(f"  range, su {len(qc):,} barre 5m vere, segno randomizzato.")
    sd_unif = (1 / np.sqrt(12)) / np.sqrt(dq["n"])
    print(f"      null: media {null.mean():.3f} · sd {null.std():.3f} "
          f"(l'uniforme darebbe sd {sd_unif:.3f})")
    print(f"      p(one-sided, impatto = q piu' alto del null) = {pval:.3f}"
          f"  -> {'IMPATTO RILEVATO' if pval < 0.05 else 'nessun impatto rilevabile'}")
    # ...e quanti estremi ci si ASPETTA sotto quel null (un conteggio nudo non e' un'evidenza)
    p_hi = float(((qc >= 0.99).mean() + (qc <= 0.01).mean()) / 2)
    print(f"      attesi sotto il null: {p_hi * dq['n']:.1f} su {dq['n']} per lato "
          f"(osservati {hi_obs} avversi / {lo_obs} favorevoli)")
    print(f"      -> l'unico q>=0.99 e' il fill #3 (ETH sell, q=0.996) su una barra da 7.5 bps di")
    print(f"         range: e' 0.03 bps di prezzo. Un estremo su una barra piatta non e' impatto.")
    mde_q_true = 2.80 * float(null.std())
    print(f"\n  ⚠️  CORREZIONE ALLA MIA STESSA POTENZA DICHIARATA. Il null vero e' "
          f"{null.std() / sd_unif:.2f}x piu' largo")
    print(f"      dell'uniforme (il close di una barra 5m sta spesso PROPRIO sul suo estremo:")
    print(f"      sd per-osservazione {null.std() * np.sqrt(dq['n']):.3f} contro 0.289 dell'uniforme).")
    print(f"      MDE_q vera = {mde_q_true:.3f} di range = {mde_q_true * rng_med:.1f} bps, non "
          f"{mde_q * rng_med:.1f}: la mia stima")
    print(f"      a priori era ottimista del {(mde_q_true / mde_q - 1) * 100:.0f}%. => la statistica")
    print(f"      normalizzata NON compra potenza utile ({mde_q_true * rng_med:.1f} bps contro "
          f"{mde_bps:.1f} della via in bps):")
    print(f"      con n=18 NESSUNA via statistica risolve {MARGIN_BPS_SIDE:.1f} bps. Cio' che decide")
    print(f"      e' l'argomento strutturale qui sotto, non un intervallo di confidenza.")

    # -- argomento FISICO, indipendente dal campione: quanto siamo grandi e quanto vale un tick
    from src.live.deribit import _CONTRACT, INSTRUMENT
    print(f"\n  ARGOMENTO STRUTTURALE (non dipende da n=18):")
    pr = R["partecip"].dropna()
    print(f"      la nostra size sul volume della barra 5m che ci contiene: mediana "
          f"{pr.median() * 100:.2f}% · p90 {pr.quantile(.9) * 100:.2f}% · max {pr.max() * 100:.2f}%")
    top = R.dropna(subset=["partecip"]).nlargest(1, "partecip").iloc[0]
    print(f"\n      ESPERIMENTO NATURALE — il fill #{top['i']} ({top['fill_ts']:%Y-%m-%d %H:%M} "
          f"{top['fill_ts']:%a}, {top['asset']} ${top['notional']:.0f}):")
    print(f"        e' stato il {top['partecip'] * 100:.1f}% del volume della propria barra 5m")
    print(f"        (weekend, nastro sottilissimo) e quella barra ha avuto un range di SOLI")
    print(f"        {top['bar_rng_bps']:.2f} bps, con il nostro print a meta' (q={top['q']:.2f}).")
    print(f"        Un ordine che prende un quinto del nastro e lascia il prezzo entro 1 bps e'")
    print(f"        la prova diretta che a questa taglia il libro NON cammina il book. Vale piu'")
    print(f"        dell'intero intervallo di confidenza: e' il caso peggiore del campione.")
    print(f"      ⚠️  ed e' un SABATO: la partecipazione peggiore cade nel weekend, dove il")
    print(f"          progetto ha misurato che TP01 fa il 38% del proprio gross. Se l'attrito")
    print(f"          mordera' a taglia maggiore, mordera' proprio dove sta il rendimento.")
    for a in sorted(R["asset"].unique()):
        tick = _CONTRACT[INSTRUMENT[a]]["tick"]
        px = float(R[R["asset"] == a]["price"].median())
        print(f"      {a}: tick {tick} su ~${px:,.0f} = {tick / px * 1e4:.4f} bps -> anche uno")
        print(f"          spread di 5 tick costerebbe {2.5 * tick / px * 1e4:.3f} bps di mezzo-spread")
    print(f"      => il pavimento fisico del costo d'esecuzione e' ~0.1 bps, DUE ordini di")
    print(f"         grandezza sotto il margine di {MARGIN_BPS_SIDE:.1f} bps. Perche' lo slippage")
    print(f"         sfondi il margine il libro dovrebbe CAMMINARE il book — e a <1% del nastro,")
    print(f"         con MENO estremi avversi di quanti il null ne preveda ({hi_obs} contro "
          f"{p_hi * dq['n']:.1f}), non c'e'")
    print(f"         traccia che lo faccia.")

    # -- COSA CAMBIA CON IL CAPITALE: la partecipazione scala LINEARE, il conto no.
    print(f"\n  SCALA — l'unica parte di questa misura che ha una data di scadenza:")
    # ⚠️ 2026-08-25: era CABLATA a 636.0 con un commento che diceva "equity reale osservata
    # (data/live/equity_seen.json)" — cioe' DICHIARAVA la fonte senza leggerla. E' P1: un
    # misuratore deve DERIVARE il proprio bersaglio, mai ridichiararlo. Il giorno del versamento
    # (equity $667 -> $2.067) la riga avrebbe continuato a stampare $636 e l'intera colonna
    # "scala" sarebbe stata sbagliata di 3,3x senza che nulla lo segnalasse — proprio nella
    # sezione che esiste per dire quando la misura scade.
    eq = _equity_osservata()
    eq_fill = _equity_per_fill(R.loc[pr.index, "fill_ts"])
    if eq_fill.notna().all():
        # partecipazione PER DOLLARO di equity: invariante alla taglia del conto al momento
        ppd = pr.values / eq_fill.values
        base = "per-fill (ogni fill sulla SUA equity)"
    else:
        ppd = pr.values / eq
        base = f"equity unica ${eq:,.0f} — RIPIEGO, i fill hanno taglie di conto diverse"
    print(f"      la partecipazione scala lineare col capitale. Normalizzazione: {base}.")
    print(f"      equity al momento dei fill: ${np.nanmin(eq_fill):,.0f} - ${np.nanmax(eq_fill):,.0f}"
          if eq_fill.notna().any() else "")
    print(f"      {'capitale':>10} {'partecip. mediana':>19} {'p90':>9} {'max osservato':>15}")
    for cap in (600, 5_000, 20_000, 100_000, 272_000):
        mx = float(np.nanmax(ppd)) * cap * 100
        print(f"      {'$' + format(cap, ','):>10} {float(np.nanmedian(ppd)) * cap * 100:>18.2f}% "
              f"{float(np.nanpercentile(ppd, 90)) * cap * 100:>8.2f}% "
              + (f"{mx:>14.1f}%" if mx <= 100 else f"{'>tutta la barra':>15}"))
    print(f"      => a $600 l'impatto e' strutturalmente impossibile e questa misura lo conferma.")
    _mx = float(np.nanmax(ppd))
    print(f"         Ma il caso peggiore osservato e' gia' {pr.max() * 100:.0f}% di una barra 5m: a $5k")
    print(f"         diventa {_mx * 5000 * 100:.0f}%, a $20k {_mx * 20000 * 100:.0f}%. "
          f"**La misura di oggi NON si estrapola**")
    print(f"         al capitale del piano: va RIFATTA a ogni salto di taglia. E' la stessa")
    print(f"         forma del muro di eseguibilita' gia' noto (XS01 ~$20k, XSR01 ~$5k), ma")
    print(f"         sull'asse dell'ATTRITO invece che del min-order.")

    # ---------------------------------------------------------------- 6. conversione
    print("\n[6] CONVERSIONE IN SHARPE/CAGR — curva PUBBLICATA, non ricalcolata\n" + "-" * 96)
    print(f"  r0726_fee_sensitivity: {DSHARPE_PER_BPS_SIDE:+.3f} Sharpe/bps-lato · "
          f"{DCAGR_PER_BPS_SIDE:+.2f} pp CAGR/bps-lato (SKH01 ~4x piu' sensibile di TP01)")
    b5 = ci_mean(R["slip_B5"])
    qs_mean = (dq["mean"] - 0.5) * rng_med
    qs_lo, qs_hi = (dq["lo"] - 0.5) * rng_med, (dq["hi"] - 0.5) * rng_med
    for nm, m, lo, hi in (("REF-B5 (bps, senza potenza)", b5["mean"], b5["lo"], b5["hi"]),
                          ("q normalizzata (con potenza)", qs_mean, qs_lo, qs_hi)):
        print(f"    {nm:<30} slip {m:+6.2f} bps -> dSharpe {DSHARPE_PER_BPS_SIDE * m:+.4f} "
              f"· dCAGR {DCAGR_PER_BPS_SIDE * m:+.3f} pp   "
              f"[IC95 dSharpe {DSHARPE_PER_BPS_SIDE * hi:+.4f}..{DSHARPE_PER_BPS_SIDE * lo:+.4f}]")
    print(f"\n  Margine gia' pagato dal modello: {MARGIN_BPS_SIDE:.1f} bps/lato "
          f"(backtest 10 bps RT - fee vera 7 bps RT).")
    print(f"  Slippage con potenza (q) dentro il margine?  "
          f"{'SI' if qs_hi <= MARGIN_BPS_SIDE else 'NON DIMOSTRATO'} "
          f"(estremo alto IC95 {qs_hi:+.2f} vs {MARGIN_BPS_SIDE:.1f})")

    # ---------------------------------------------------------------- 7. taglia e ora
    print("\n[7] DIPENDENZA DA TAGLIA E ORA — il campione lo permette?\n" + "-" * 96)
    from scipy import stats
    sub = R.dropna(subset=["q"])
    rho, p = stats.spearmanr(sub["notional"], sub["q"])
    print(f"  taglia ESEGUITA ${R['notional'].min():.0f}-${R['notional'].max():.0f} "
          f"(mediana ${R['notional'].median():.0f}). NB: e' l'INCREMENTO mandato al venue, non il")
    print(f"  target di posizione (che arriva a $189): il libro riconcilia, non ri-apre.")
    print(f"  Spearman(notional, q) = {rho:+.3f} (p={p:.2f})")
    big = sub[sub["notional"] >= sub["notional"].median()]["q"]
    small = sub[sub["notional"] < sub["notional"].median()]["q"]
    print(f"    meta' grande n={len(big)} q={big.mean():.3f} · meta' piccola n={len(small)} "
          f"q={small.mean():.3f} · differenza {big.mean() - small.mean():+.3f} "
          f"(MDE su un confronto a meta' campione ~{2.80 * (1 / np.sqrt(12)) * np.sqrt(2 / (n / 2)):.3f})")
    hrs = sorted(int(h) for h in R["fill_ts"].dt.hour.unique())
    print(f"  ore del giorno coperte: {hrs} ({len(hrs)} ore distinte su 24, "
          f"{R['fill_ts'].dt.hour.value_counts().max()} fill nell'ora piu' popolata)")
    print(f"  => con n={n} e {len(hrs)} ore distinte, l'effetto ora NON e' stimabile: "
          f"lo dichiaro non misurato, non 'assente'.")

    # ---------------------------------------------------------------- 8. netting
    print("\n[8] NETTING TP01/SKH01 — cosa NON e' misurabile da questi dati\n" + "-" * 96)
    print(f"  Il libro netta TP01 e SKH01 su UNO strumento: ogni fill e' la RISULTANTE, mai")
    print(f"  il trade di uno sleeve. Quindi lo slippage misurato e' quello del LIBRO, e non si")
    print(f"  puo' attribuire a TP01 o a SKH01 — proprio mentre SKH01 e' ~4x piu' sensibile.")
    print(f"  Il netting RIDUCE il turnover eseguito: {n} ordini in "
          f"{(fills['fill_ts'].max() - fills['fill_ts'].min()).days} giorni, "
          f"nozionale totale ${R['notional'].sum():,.0f}.")

    print("\n" + "=" * 96)
    out = SCRATCH / "r0822_slip_fills.csv"
    SCRATCH.mkdir(parents=True, exist_ok=True)
    R.to_csv(out, index=False)
    print(f"  dettaglio per fill: {out}")
    print("=" * 96)


if __name__ == "__main__":
    main()