Adriano Dal Pastro
510 lines
28 KiB
Python
510 lines
28 KiB
Python
"""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 q e' un'estrazione ~uniforme -> E[q] = 0.5. sigma_q = 1/sqrt(12) = 0.289 e' ~20x
|
|
piu' piccolo del rumore in bps.
|
|
|
|
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 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]")
|
|
odd = fills[(fills["fee_bps"] - FEE_TAKER_BPS_SIDE).abs() > 0.05]
|
|
if len(odd):
|
|
print(f" ⚠️ {len(odd)}/{len(fills)} fill NON al tier {FEE_TAKER_BPS_SIDE:.2f}:")
|
|
for _, o in odd.iterrows():
|
|
print(f" #{o['i']} {o['fill_ts']} {o['asset']} {o['side']} "
|
|
f"{o['filled']} @ {o['price']:.2f} = ${o['notional']:.2f} · "
|
|
f"fee {o['fee']:.8f} = {o['fee_bps']:.2f} bps")
|
|
print(f" -> nozionale ${odd['notional'].min():.2f}: sotto i ~$7 la fee sembra")
|
|
print(f" arrotondata AL RIALZO. E' un costo REALE che il modello non ha, e")
|
|
print(f" morde solo gli ordini minuscoli (dove il libro vive a $600).")
|
|
|
|
# ---------------------------------------------------------------- 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}")
|
|
|
|
# ---------------------------------------------------------------- 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)")
|
|
print(f" q=1.00 esatti (fill AL massimo = firma di impatto): {(q >= 0.999).sum()}/{dq['n']}"
|
|
f" · q=0.00 esatti: {(q <= 0.001).sum()}/{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.")
|
|
print(f" null: media {null.mean():.3f} · sd {null.std():.3f} "
|
|
f"(l'uniforme darebbe sd {(1 / np.sqrt(12)) / np.sqrt(dq['n']):.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'}")
|
|
|
|
# -- 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}% · max {pr.max() * 100:.2f}%")
|
|
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 0/18 fill al massimo di barra non c'e' traccia che lo faccia.")
|
|
|
|
# ---------------------------------------------------------------- 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()
|