#!/usr/bin/env python """r0822b_maker.py — l'ESECUZIONE PASSIVA (maker) vale qualcosa per il libro live? CONTESTO (fatti gia' nel progetto, non ri-misurati qui) ------------------------------------------------------- * Il libro Deribit (TP01 75% + SKH01 25%, nettati su UNA sola posizione per strumento) manda **solo ordini market** -> paga sempre il **taker**. `src/live/execution.py::_submit` ha `order_type="market"` come default e `rebalance_signed` non passa mai altro. * Fee misurate: **taker 3,50 bps/lato** (fee_watch + 16/16 fill sul nastro, 22/08), **maker 1,50 bps/lato** (listino base dal 2026-08-01; fino al 31/07 era 0,00). I backtest del progetto modellano **5,00 bps/lato** (0,10% RT) -> gia' conservativi. * Il progetto ha guardato il lato passivo UNA volta (raccomandazione "T1" del 26/07: il TP di SKH01 come limit resting) e l'ha chiusa **non per la fee** ma perche' un ordine al livello di SKH chiuderebbe anche la quota TP01 (**netting su strumento unico**, `src/live/book.py`). ⚠️ QUEL VINCOLO NON SI APPLICA QUI: rendere passivo **l'ordine netto di ribilanciamento** e' UN ordine sulla posizione netta -> niente conflitto fra sleeve. Il netting uccide i bracket per-sleeve, non l'ordine netto. E' la ragione per cui questa domanda e' nuova. * Slippage misurato il 22/08: **~0 a questa taglia** (un fill ha preso il 22% del volume della sua barra 5m, range 1,08 bps) -> il differenziale maker/taker e' fee PURA, e il mezzo spread catturato da un passivo e' <0,005 bps (tick BTC 0,1 su ~$110k; ETH 0,01 su ~$2.500) = ZERO. Percio' qui il passivo NON guadagna spread: guadagna 2,00 bps di fee e rischia il non-fill. * Il disaster-SL −30% e' uno `stop_market` on-book: NON e' toccato da nulla di quanto segue. LA DOMANDA CHE NESSUNO HA FATTO ------------------------------- Un limit resting non e' un market order piu' economico: **puo' non riempirsi**. E non riempirsi su un segnale di trend significa **perdere il movimento**, che e' esattamente il caso in cui non si e' riempito (il prezzo e' scappato dalla parte giusta). Quindi: costo atteso del passivo = P(fill)*(−2,00 bps) + P(no fill)*E[mossa avversa | no fill] e la seconda parte e' **avversa per costruzione**: un buy limit al riferimento non si riempie se e solo se il prezzo non e' MAI sceso al livello, cioe' e' salito. COSA FA QUESTO SCRIPT --------------------- 1. il differenziale in Sharpe/CAGR di libro, riusando la curva fee gia' pubblicata (`r0726_fee_sensitivity.py`: −0,017 Sharpe/bps, −0,23% CAGR/bps) E ri-misurandola a 4 punti di fee sul path del libro (verifica indipendente, non ri-derivazione della curva); 2. ricostruisce gli **ORDINI** del libro live (target netto orario, banda min-order $5) e li confronta con i 18 ordini veri di `data/live/book_executions.jsonl` (controllo di realta'); 3. per OGNI ordine misura su 5m certificati: P(fill) di un limit al riferimento entro N minuti, e il ritorno condizionato **ai casi in cui NON si riempie**. Confronto **APPAIATO PER ORDINE** (mai per esito: allineare sui riempiti tiene solo i casi in cui non e' successo niente — trappola gia' presa 3 volte in questo progetto); 4. trasforma il costo per-ordine in una serie di P&L additiva ai rendimenti giornalieri del libro -> Sharpe/CAGR/maxDD della politica passiva, e il numero in **dollari all'anno**; 5. mette il risultato accanto alle altre leve gia' misurate. LENTI E SEMPLIFICAZIONI — DICHIARATE (tutte GENEROSE verso il maker, di proposito) --------------------------------------------------------------------------------- (a) griglia oraria: il cron gira a :07. Decisione con le barre 5m chiuse fino a h:05, prezzo di riferimento = close della barra 5m etichettata h:00. Verificato sul log vero: gli ordini cadono a :07 di un'ora qualsiasi, e TP01 si aggiorna alle 01:07 UTC. (b) TP01 cambia target una volta al giorno, all'ora 01:00 UTC (lente backtest; l'effetto della barra parziale e' misurato +-0,03 Sharpe, 26/07 -> irrilevante per la TAGLIA degli ordini). (c) SKH01: ingressi alla CHIUSURA del bin 230m (lente backtest), uscite al primo confine orario dopo il tocco 5m (lente live, `_skyhook_positions` itera nel bin parziale). L'ingresso vero del live e' INTRA-BIN, cioe' piu' vicino al breakout -> peggiore per un limit passivo: questa lente e' **generosa** verso il maker. (d) fill: un limit BUY al prezzo P si considera riempito se `low <= P` nella finestra. Ignora la coda del book (al livello si sta DIETRO agli ordini gia' presenti: un semplice tocco non riempie) -> **generosa** verso il maker. Finestra di scansione da h:05 (l'ordine e' a h:07: regaliamo 2 minuti di opportunita' di fill). (e) ordini tutto-o-niente, nessun fill parziale. (f) niente cap per-asset nella ricostruzione (a $635 il cap $318 morde di rado e riduce PROPORZIONALMENTE guadagno e danno: non cambia il segno). nice -n 19 timeout 900 uv run python scripts/research/r0822b_maker.py """ from __future__ import annotations import json 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)) from src.backtest.harness import backtest_signals # noqa: E402 from src.data.downloader import load_data # noqa: E402 from src.portfolio.portfolio import combine_outer, metrics, to_daily # noqa: E402 from src.strategies.skyhook import (LTF_MIN, SKH01_V2_DD, build_frames, # noqa: E402 skyhook_entries) from src.strategies.trend_portfolio import (CANONICAL, TrendPortfolio, # noqa: E402 resample_1d, simple_returns) ASSETS = ("BTC", "ETH") W_TP01, W_SKH = 0.75, 0.25 # pesi del book Deribit (src/live/book.py) WEIGHT = 0.5 # book 50/50 BTC+ETH MIN_ORDER_USD = 5.0 # config/live.json EQUITY_LIVE = 635.0 # conto vero (22/08) FEE_TAKER_BPS = 3.50 # misurata sul nastro + fee_watch FEE_MAKER_BPS = 1.50 # listino base dal 2026-08-01 FEE_MODEL_BPS = 5.00 # cio' che i backtest modellano FEE_GRID_BPS = (0.0, FEE_MAKER_BPS, FEE_TAKER_BPS, FEE_MODEL_BPS) # curva PUBBLICATA (r0726_fee_sensitivity.py) — riusata, non ri-derivata PUB_SH_PER_BPS = -0.017 PUB_CAGR_PER_BPS = -0.23 # punti percentuali di CAGR per bps/lato HORIZONS_MIN = (5, 15, 30, 60, 120) # attesa massima del limit prima di inseguire # CODA DEL BOOK — il parametro che DECIDE, e che l'OHLC non contiene. # Un limit al livello sta DIETRO agli ordini gia' presenti: "il prezzo ha toccato il mio livello" # NON e' "sono stato riempito". `q` = quanti bps il prezzo deve attraversare OLTRE il livello # perche' la coda davanti sia consumata. q=0 e' il LIMITE SUPERIORE ottimistico (e a q=0 il fill # e' quasi automatico: l'apertura della prima barra di scansione COINCIDE col livello, quindi # `low <= P` chiede solo che il prezzo non scappi istantaneamente). QUEUE_BPS = (0.0, 1.0, 2.0, 5.0, 10.0) P2_MAX_HOURS = 24 # politica "mai inseguire": ri-posta ogni ora, tetto 24h CAPITALS = (635.0, 2000.0, 5000.0) HOLDOUT = pd.Timestamp("2025-01-01", tz="UTC") SEED = 20260822 # =========================================================================================== # 1. SEGNALI — calcolati UNA volta per asset (non dipendono dalla fee) # =========================================================================================== def build_asset(asset: str) -> dict: """Tutto cio' che serve per l'asset: TP01 daily, path SKH01, griglia 5m e oraria.""" df5 = load_data(asset, "5m") ts5 = pd.to_datetime(df5["datetime"], utc=True) A = dict( ts5=ts5.values.astype("datetime64[ns]"), o5=df5["open"].values.astype(float), h5=df5["high"].values.astype(float), l5=df5["low"].values.astype(float), c5=df5["close"].values.astype(float), ) # --- SKH01: ingressi/uscite sulla griglia 230m, poi tempo di RILEVAZIONE oraria ltf, htf = build_frames(df5) ent = skyhook_entries(ltf, htf, SKH01_V2_DD) H, L = ltf["high"].values, ltf["low"].values ltf_ts = pd.to_datetime(ltf["datetime"], utc=True) n = len(ltf) trades = [] i = 0 while i < n: e = ent[i] if e is None: i += 1 continue d, sl, tp, mb = e["dir"], e["sl"], e["tp"], e["max_bars"] exit_idx = None for s in range(1, mb + 1): j = i + s if j >= n: break hit = (L[j] <= sl or H[j] >= tp) if d == 1 else (H[j] >= sl or L[j] <= tp) if hit or s == mb: exit_idx = j break if exit_idx is None: # trade ancora aperto a fine storia trades.append((i, None, d, sl, tp)) break trades.append((i, exit_idx, d, sl, tp)) i = exit_idx + 1 # tempi di rilevazione: ingresso a chiusura bin (lente backtest); # uscita al primo confine orario dopo il TOCCO 5m (lente live) ltf_close = ltf_ts + pd.Timedelta(minutes=LTF_MIN) skh_events = [] # (ts_rilevazione, segno) for (i0, j0, d, sl, tp) in trades: skh_events.append((_ceil_hour(ltf_close.iloc[i0]), int(d))) if j0 is None: continue t_touch = _first_touch_5m(A, ltf_ts.iloc[j0], ltf_close.iloc[j0], d, sl, tp) skh_events.append((_ceil_hour(t_touch), 0)) # --- TP01: target giornaliero, visibile dall'ora 01:00 del giorno DOPO la chiusura d1 = resample_1d(load_data(asset, "1h")) tp_cfg = TrendPortfolio(**CANONICAL) tgt = tp_cfg.target_series(d1) d1_ts = pd.to_datetime(d1["datetime"], utc=True) tp_events = [(pd.Timestamp(t).normalize() + pd.Timedelta(hours=25), float(v)) for t, v in zip(d1_ts, tgt)] # barra del giorno D -> vista alle 01:00 di D+1 # --- griglia oraria dei momenti di decisione (barre 5m etichettate :00) m = pd.DatetimeIndex(A["ts5"]) hmask = (m.minute == 0) A["hours"] = m[hmask].tz_localize("UTC") A["hidx"] = np.flatnonzero(hmask) # indice 5m della barra di riferimento A["ref"] = A["c5"][A["hidx"]] # prezzo di riferimento (close 5m di h:00) A["skh_events"] = skh_events A["tp_events"] = tp_events A["daily_ret"] = simple_returns(d1["close"].values.astype(float)) A["daily_tgt"] = tgt A["d1_ts"] = d1_ts A["ltf"] = ltf A["ent"] = ent return A def _ceil_hour(t: pd.Timestamp) -> pd.Timestamp: return t.ceil("h") def _first_touch_5m(A: dict, t_open, t_close, d: int, sl: float, tp: float) -> pd.Timestamp: """Primo istante 5m, dentro il bin 230m d'uscita, in cui SL o TP e' toccato. Se nessuno e' toccato (uscita per max_bars) ritorna la chiusura del bin.""" lo = np.searchsorted(A["ts5"], np.datetime64(t_open.tz_localize(None))) hi = np.searchsorted(A["ts5"], np.datetime64(t_close.tz_localize(None))) for k in range(lo, min(hi, len(A["ts5"]))): hit = (A["l5"][k] <= sl or A["h5"][k] >= tp) if d == 1 else \ (A["h5"][k] >= sl or A["l5"][k] <= tp) if hit: return pd.Timestamp(A["ts5"][k], tz="UTC") + pd.Timedelta(minutes=5) return t_close def exposure_path(A: dict) -> pd.DataFrame: """Esposizione-bersaglio del libro su questo asset, ORA per ORA, in frazione di equity. expo = WEIGHT * (W_TP01*tp_frac + W_SKH*skh_sign) — la formula di src/live/book.py.""" hours = A["hours"] tp = pd.Series(0.0, index=hours) for t, v in A["tp_events"]: tp.loc[tp.index >= t] = v sk = pd.Series(0.0, index=hours) for t, v in sorted(A["skh_events"]): sk.loc[sk.index >= t] = v expo = WEIGHT * (W_TP01 * tp.values + W_SKH * sk.values) src = np.where(np.abs(np.diff(sk.values, prepend=0.0)) > 0, "SKH", np.where(np.abs(np.diff(tp.values, prepend=0.0)) > 0, "TP01", "")) return pd.DataFrame(dict(ts=hours, expo=expo, ref=A["ref"], hidx=A["hidx"], src=src)) # =========================================================================================== # 2. ORDINI — banda min-order, esattamente come rebalance_signed # =========================================================================================== def orders_from_path(P: pd.DataFrame, equity: float, min_usd: float = MIN_ORDER_USD) -> pd.DataFrame: """Ordini che il libro manderebbe: si trada solo se |target-posizione| >= min_usd.""" band = min_usd / equity expo = P["expo"].values held = 0.0 rows = [] for k in range(len(expo)): delta = expo[k] - held if abs(delta) < band: continue rows.append((P["ts"].iloc[k], delta, P["ref"].iloc[k], P["hidx"].iloc[k], P["src"].iloc[k])) held = expo[k] return pd.DataFrame(rows, columns=["ts", "dfrac", "ref", "hidx", "src"]) # =========================================================================================== # 3. FILL E SELEZIONE AVVERSA # =========================================================================================== def measure_orders(A: dict, O: pd.DataFrame, horizons=HORIZONS_MIN, q_bps: float = 0.0, strict: bool = True) -> dict: """Per ogni ordine e ogni orizzonte N: il limit e' stato riempito? e se no, quanto e' scappato il prezzo? `q_bps` = attraversamento richiesto oltre il livello (coda del book). `strict` E' IL PARAMETRO CHE DECIDE IL SEGNO DI TUTTO IL RISULTATO. strict=False -> `low <= limite`: il prezzo ha TOCCATO il livello. Include il caso in cui la barra APRE esattamente al livello e non ci torna mai: li' non e' stato scambiato NIENTE al nostro prezzo dopo che l'ordine e' arrivato, quindi un passivo in fondo alla coda NON e' riempito. E' un limite superiore FISICAMENTE NON DIFENDIBILE. strict=True -> `low < limite`: il prezzo ha ATTRAVERSATO il livello. E' la condizione minima perche' esista uno scambio al nostro prezzo. Anche cosi' si assume di essere DAVANTI a tutta la coda gia' presente -> resta generosa, ma difendibile.""" l5, h5, c5 = A["l5"], A["h5"], A["c5"] nb = len(c5) idx = O["hidx"].values.astype(int) ref = O["ref"].values.astype(float) buy = (O["dfrac"].values > 0) q = q_bps / 1e4 out = {} for N in horizons: steps = N // 5 filled = np.zeros(len(O), bool) adverse = np.zeros(len(O)) for k in range(len(O)): i0 = idx[k] + 1 # barra 5m etichettata h:05 (ordine a h:07: generoso) i1 = min(i0 + steps, nb) P = ref[k] if i0 >= i1: filled[k] = True continue lim = P * (1.0 - q) if buy[k] else P * (1.0 + q) if buy[k]: hit = (l5[i0:i1].min() < lim) if strict else (l5[i0:i1].min() <= lim) else: hit = (h5[i0:i1].max() > lim) if strict else (h5[i0:i1].max() >= lim) filled[k] = hit if not hit: end = c5[i1 - 1] adverse[k] = (end - P) / P if buy[k] else (P - end) / P out[N] = pd.DataFrame(dict(filled=filled, adverse=adverse), index=O.index) return out def measure_p2(A: dict, O: pd.DataFrame, P: pd.DataFrame, max_hours=P2_MAX_HOURS, q_bps: float = 0.0, strict: bool = True) -> pd.DataFrame: """Politica 'MAI INSEGUIRE': il limit resta e viene RI-POSTATO al nuovo riferimento a ogni cron orario, fino a `max_hours`; solo allora si insegue a market. Il costo e' cumulativo.""" l5, h5, c5 = A["l5"], A["h5"], A["c5"] hidx = P["hidx"].values.astype(int) ref = P["ref"].values.astype(float) nb, nh = len(c5), len(hidx) pos_of = {int(v): k for k, v in enumerate(hidx)} q = q_bps / 1e4 cycles, adverse, chased = [], [], [] for _, r in O.iterrows(): k0 = pos_of[int(r["hidx"])] P0 = float(r["ref"]); buy = r["dfrac"] > 0 done, adv, cyc = False, 0.0, 0 for k in range(k0, min(k0 + max_hours, nh)): Pk = ref[k] lim = Pk * (1.0 - q) if buy else Pk * (1.0 + q) i0 = hidx[k] + 1 i1 = min(i0 + 12, nb) if i0 >= i1: break if buy: hit = (l5[i0:i1].min() < lim) if strict else (l5[i0:i1].min() <= lim) else: hit = (h5[i0:i1].max() > lim) if strict else (h5[i0:i1].max() >= lim) cyc = k - k0 + 1 if hit: adv = (lim - P0) / P0 if buy else (P0 - lim) / P0 done = True break if not done: k = min(k0 + max_hours, nh - 1) adv = (ref[k] - P0) / P0 if buy else (P0 - ref[k]) / P0 cycles.append(cyc); adverse.append(adv); chased.append(not done) return pd.DataFrame(dict(cycles=cycles, adverse=adverse, chased=chased), index=O.index) def per_order_delta(O: pd.DataFrame, M: pd.DataFrame, *, miss_fee_bps: float) -> np.ndarray: """Delta di P&L in FRAZIONE DI EQUITY DI LIBRO, per ORDINE, del passivo contro il taker. >0 = il passivo guadagna. Confronto APPAIATO PER ORDINE: ogni ordine compare una volta in entrambe le politiche, riempito o no (allineare sui soli riempiti terrebbe i casi in cui non e' successo niente — trappola gia' presa 3 volte in questo progetto).""" absd = O["dfrac"].abs().values filled = M["filled"].values if "filled" in M.columns else np.ones(len(O), bool) fee_bps = np.where(filled, FEE_MAKER_BPS, miss_fee_bps) return (FEE_TAKER_BPS - fee_bps) / 1e4 * absd - M["adverse"].values * absd def weighted_stats(F: np.ndarray, Ad: np.ndarray, W: np.ndarray) -> dict: """Tutto PONDERATO PER NOZIONALE (e' il nozionale che paga la fee, non il conteggio). `p_be` = la frazione di fill che il passivo deve raggiungere per PAREGGIARE il taker: p*(-gap) + (1-p)*E[avversa|no fill] = 0 -> p_be = E/(E+gap) E' la forma model-free del risultato: non dipende dall'ipotesi di coda, solo da quanto scappa il prezzo quando il limit resta a terra.""" gap = (FEE_TAKER_BPS - FEE_MAKER_BPS) / 1e4 wf = W[F].sum() / W.sum() if W.sum() > 0 else 1.0 miss_w = W[~F] e_adv = float(np.sum(miss_w * Ad[~F]) / miss_w.sum()) if miss_w.sum() > 0 else 0.0 cost = float(np.sum(W * (np.where(F, FEE_MAKER_BPS, FEE_TAKER_BPS) / 1e4 + Ad)) / W.sum()) return dict(p_fill=wf, e_adv=e_adv, cost_bps=cost * 1e4, p_be=e_adv / (e_adv + gap) if (e_adv + gap) > 0 else 1.0) def boot_ci(x: np.ndarray, w: np.ndarray, n: int = 2000, seed: int = SEED) -> tuple: """IC95% bootstrap della media PONDERATA PER NOZIONALE (in bps del nozionale tradato).""" rng = np.random.default_rng(seed) m = len(x) idx = rng.integers(0, m, size=(n, m)) num = x[idx].sum(axis=1) den = w[idx].sum(axis=1) b = num / np.where(den > 0, den, np.nan) * 1e4 return float(np.nanpercentile(b, 2.5)), float(np.nanpercentile(b, 97.5)) # =========================================================================================== # 4. LIBRO — rendimenti giornalieri modellati, per innestarci il costo d'esecuzione # =========================================================================================== def book_daily(cache: dict, fee_bps_side: float) -> pd.Series: fs = fee_bps_side / 1e4 tp_s, sk_s = {}, {} for a in ASSETS: A = cache[a] r, tgt = A["daily_ret"], A["daily_tgt"] held = np.zeros(len(tgt)); held[1:] = tgt[:-1] net = held * r - fs * np.abs(np.diff(held, prepend=0.0)); net[0] = 0.0 tp_s[a] = pd.Series(np.clip(net, -0.99, None), index=A["d1_ts"]) m = backtest_signals(A["ltf"], A["ent"], fee_rt=2 * fs, leverage=1.0, asset=a, tf="230m") s = pd.Series(m.equity, index=pd.DatetimeIndex(pd.to_datetime(m.eq_index, utc=True))) sk_s[a] = s.resample("1D").last().ffill().pct_change().dropna() Jt = pd.concat(tp_s, axis=1, join="inner").fillna(0.0) Js = pd.concat(sk_s, axis=1, join="inner").fillna(0.0) tp = to_daily(pd.Series(0.5 * Jt["BTC"].values + 0.5 * Jt["ETH"].values, index=Jt.index)) sk = to_daily(pd.Series(0.5 * Js["BTC"].values + 0.5 * Js["ETH"].values, index=Js.index)) return combine_outer({"TP01": tp, "SKH01": sk}, {"TP01": W_TP01, "SKH01": W_SKH}) def graft(base: pd.Series, deltas: dict, O_live: dict) -> pd.Series: """Innesta il delta per-ordine (frazione di equity) nei rendimenti giornalieri del libro.""" ser = [] for a in ASSETS: ts = pd.DatetimeIndex(O_live[a]["ts"]).tz_convert("UTC").normalize() ser.append(pd.Series(deltas[a], index=ts)) add = pd.concat(ser).groupby(level=0).sum() return base.add(add.reindex(base.index).fillna(0.0), fill_value=0.0) def show(tag: str, d: pd.Series) -> dict: m = metrics(d) h = metrics(d[d.index >= HOLDOUT]) print(f" {tag:<36} Sh {m['sharpe']:>6.3f} | HOLD {h['sharpe']:>6.3f} | " f"maxDD {m['maxdd']*100:>5.2f}% | CAGR {m['cagr']*100:>6.2f}%") return dict(sharpe=m["sharpe"], hold=h["sharpe"], maxdd=m["maxdd"], cagr=m["cagr"]) # =========================================================================================== def main() -> None: rng = np.random.default_rng(SEED) print("=" * 104) print(" r0822b — ESECUZIONE PASSIVA (MAKER) SUL LIBRO LIVE:") print(" quanto vale il differenziale, quanto costa non riempirsi") print("=" * 104) print(f" taker {FEE_TAKER_BPS:.2f} bps/lato · maker {FEE_MAKER_BPS:.2f} bps/lato · " f"differenziale {FEE_TAKER_BPS-FEE_MAKER_BPS:.2f} bps/lato · modello backtest {FEE_MODEL_BPS:.2f}") print(" NETTING: il libro ha UNA posizione netta per strumento -> l'ordine di ribilanciamento") print(" e' UNO solo. Renderlo passivo NON e' bloccato dal motivo che uccise T1 il 26/07") print(" (quello uccide i BRACKET per-sleeve, non l'ordine netto). Il disaster-SL -30% resta") print(" uno stop_market on-book e non e' toccato da niente di quanto segue.") cache = {a: build_asset(a) for a in ASSETS} # ---------------------------------------------------------------- 1. il differenziale print("\n" + "-" * 104) print(" 1. QUANTO VALE IL DIFFERENZIALE (limite superiore: TUTTO il flusso diventa maker,") print(" zero mancati riempimenti — cioe' il numero che NON si puo' superare)") print("-" * 104) curve = {bps: show(f"libro @ {bps:.2f} bps/lato", book_daily(cache, bps)) for bps in FEE_GRID_BPS} gap = FEE_TAKER_BPS - FEE_MAKER_BPS dsh = curve[FEE_MAKER_BPS]["sharpe"] - curve[FEE_TAKER_BPS]["sharpe"] dcg = (curve[FEE_MAKER_BPS]["cagr"] - curve[FEE_TAKER_BPS]["cagr"]) * 100 print(f"\n misurato qui : dSharpe {dsh:+.4f} dCAGR {dcg:+.3f} pp " f"({dsh/gap:+.4f} Sh/bps, {dcg/gap:+.3f} pp/bps)") print(f" curva 26/07 : dSharpe {-PUB_SH_PER_BPS*gap:+.4f} dCAGR {-PUB_CAGR_PER_BPS*gap:+.3f} pp " f"({-PUB_SH_PER_BPS:+.4f} Sh/bps, {-PUB_CAGR_PER_BPS:+.3f} pp/bps) <- replica indipendente OK") print(" in DOLLARI all'anno, se tutto il flusso fosse maker:") for cap, note in ((EQUITY_LIVE, "conto vero oggi"), (5000.0, "dopo i versamenti"), (258338.0, "capitale-rendita, muro 07/08")): print(f" ${cap:>9,.0f} -> ${cap*dcg/100:>9,.2f}/anno ({note})") # ---------------------------------------------------------------- 2. gli ordini print("\n" + "-" * 104) print(" 2. GLI ORDINI DEL LIBRO (target netto orario + banda min-order $5, formula di book.py)") print("-" * 104) paths = {a: exposure_path(cache[a]) for a in ASSETS} yrs = (paths["BTC"]["ts"].iloc[-1] - paths["BTC"]["ts"].iloc[0]).days / 365.25 print(f" {'capitale':>9} {'banda':>7} |" + "".join(f"{a+' ord/anno':>15}{'turnover/anno':>16}" for a in ASSETS)) ORD = {} for cap in CAPITALS: line, tot = "", {} for a in ASSETS: O = orders_from_path(paths[a], cap) tot[a] = O line += f"{len(O)/yrs:>15.1f}{O['dfrac'].abs().sum()/yrs*cap:>15,.0f}$" ORD[cap] = tot print(f" ${cap:>8,.0f} {MIN_ORDER_USD/cap*100:>6.2f}% |" + line) O_live = ORD[EQUITY_LIVE] n_live = sum(len(O_live[a]) for a in ASSETS) turn = sum(O_live[a]["dfrac"].abs().sum() for a in ASSETS) / yrs print(f"\n ricostruzione a ${EQUITY_LIVE:,.0f}: {n_live} ordini su {yrs:.1f} anni = " f"{n_live/yrs:.1f}/anno di libro ({n_live/yrs/2:.1f} per asset), " f"turnover {turn:.1f}x equity/anno") print(f" coerenza col punto 1: {turn:.1f}x * {gap:.2f} bps = {turn*gap/1e4*100:.3f} pp di CAGR " f"contro i {dcg:.3f} pp misurati sul libro modellato " f"({turn*gap/1e4*100/dcg*100:.0f}% — la ricostruzione tiene meno turnover del modello)") exe = ROOT / "data" / "live" / "book_executions.jsonl" if exe.exists(): rows = [json.loads(l) for l in exe.read_text().splitlines() if l.strip()] gg = max((pd.Timestamp(rows[-1]["ts_utc"]) - pd.Timestamp(rows[0]["ts_utc"])).days, 1) print(f" controllo di realta': {len(rows)} ordini VERI in {gg} giorni = " f"{len(rows)/gg*365.25:.0f}/anno di libro. Stesso ordine di grandezza; la finestra e'") print(" troppo corta per una stima (18 ordini), serve solo a escludere un errore di fattore 10.") for a in ASSETS: v = O_live[a]["src"].value_counts() print(f" {a}: origine " + " ".join(f"{(k or 'drift')} {n}" for k, n in v.items())) # ---------------------------------------------------------------- 3. il non-fill print("\n" + "-" * 104) print(" 3. IL COSTO DEL MANCATO RIEMPIMENTO (limit al riferimento, APPAIATO PER ORDINE)") print("-" * 104) print(" Il parametro che decide NON e' l'orizzonte d'attesa: e' cosa si chiama 'riempito'.") print(" Un limit al livello sta DIETRO a tutta la coda gia' presente, quindi:") print(" TOCCO (low <= limite) -> include la barra che APRE al livello e non ci") print(" torna mai: al nostro prezzo non e' stato scambiato") print(" NIENTE. Fisicamente non difendibile: LIMITE SUPERIORE.") print(" ATTRAVERSAMENTO (low < limite) -> condizione MINIMA perche' uno scambio al nostro") print(" prezzo esista. Assume ancora di essere davanti a") print(" tutta la coda -> generosa, ma difendibile.") print(" q bps in piu' -> quanto il prezzo deve andare OLTRE perche' la coda") print(" davanti a noi sia consumata. L'OHLC non contiene la") print(" profondita' del book: `q` si dichiara, non si stima.") print(" E la colonna che decide e' `serve`: la frazione di fill necessaria per PAREGGIARE il") print(" taker = E[avversa|no fill]/(E[avversa|no fill] + 2,00 bps). Model-free.") print(f"\n {'ipotesi':>10} {'N(min)':>7} | {'P(fill)':>8} {'serve':>8} {'esito':>6} | {'E[avv|no]':>10}" f" {'peggio':>8} | {'costo maker':>12} | {'vs taker':>9} | {'IC95 del delta':>22}" f" | {'ordini vinti':>12}") grid_cells = 6 # le 6 celle della griglia fine calcolate piu' sotto tables = {} variants = [("TOCCO <=", 0.0, False)] + [(f"attrav. q={q:.0f}", q, True) for q in QUEUE_BPS] for lab, q, strict in variants: MEAS = {a: measure_orders(cache[a], O_live[a], q_bps=q, strict=strict) for a in ASSETS} tables[(q, strict)] = MEAS for N in HORIZONS_MIN: grid_cells += 1 F = np.concatenate([MEAS[a][N]["filled"].values for a in ASSETS]) Ad = np.concatenate([MEAS[a][N]["adverse"].values for a in ASSETS]) W = np.concatenate([O_live[a]["dfrac"].abs().values for a in ASSETS]) D = np.concatenate([per_order_delta(O_live[a], MEAS[a][N], miss_fee_bps=FEE_TAKER_BPS) for a in ASSETS]) S = weighted_stats(F, Ad, W) lo, hi = boot_ci(D, W) ok = "SI" if S["p_fill"] >= S["p_be"] else "no" print(f" {lab:>10} {N:>7} | {S['p_fill']*100:>7.1f}% {S['p_be']*100:>7.1f}% {ok:>6} | " f"{S['e_adv']*1e4:>8.0f}b {(Ad[~F].max()*1e4 if (~F).any() else 0):>7.0f}" f" | {S['cost_bps']:>8.2f} bps | {S['cost_bps']-FEE_TAKER_BPS:>+8.2f} | " f"[{lo:>+7.2f},{hi:>+7.2f}] bps | {(D>0).mean()*100:>11.1f}%") print() # -- la sintesi che non dipende dalle ipotesi di coda print(" SINTESI — IL SEGNO DEL RISULTATO STA TUTTO IN UNA DISUGUAGLIANZA STRETTA:") print(f" {'N(min)':>7} | {'TOCCO <= (limite sup.)':>28} | {'ATTRAVERSAMENTO < (minimo difendibile)':>42}") for N in HORIZONS_MIN: row = [] for key in ((0.0, False), (0.0, True)): F = np.concatenate([tables[key][a][N]["filled"].values for a in ASSETS]) Ad = np.concatenate([tables[key][a][N]["adverse"].values for a in ASSETS]) W = np.concatenate([O_live[a]["dfrac"].abs().values for a in ASSETS]) S = weighted_stats(F, Ad, W) keep = (FEE_TAKER_BPS - S["cost_bps"]) / (FEE_TAKER_BPS - FEE_MAKER_BPS) row.append(f"P(fill) {S['p_fill']*100:>5.1f}% resta {FEE_TAKER_BPS-S['cost_bps']:>+5.2f} bps" f" = {keep*100:>6.1f}%") print(f" {N:>7} | {row[0]:>28} | {row[1]:>42}") print(" Fra le due colonne cambia UNA cosa: se la barra che apre esattamente al nostro") print(" prezzo, e da li' se ne va, conta come riempimento. Non conta: al nostro prezzo non") print(" e' stato scambiato niente dopo che l'ordine e' arrivato. Con la definizione fisica") print(" il passivo PERDE gia' prima di aggiungere un solo bps di coda.") # griglia FINE vicino a zero: dove sta il salto? (calcolata qui, non citata a memoria) print("\n griglia FINE a N=60 (attraversamento), per vedere dove sta il salto:") fine = [] for q in (0.0, 0.005, 0.05, 0.25, 0.5, 1.0): num = den = wf = 0.0 for a in ASSETS: M = measure_orders(cache[a], O_live[a], horizons=(60,), q_bps=q, strict=True)[60] w = O_live[a]["dfrac"].abs().values F = M["filled"].values num += np.sum(w * (np.where(F, FEE_MAKER_BPS, FEE_TAKER_BPS) / 1e4 + M["adverse"].values)) den += w.sum(); wf += w[F].sum() fine.append((q, wf / den, num / den * 1e4)) print(" " + " ".join(f"q={q:<5.3f} P(fill) {f*100:>4.1f}% costo {c:>5.2f}" for q, f, c in fine)) Ft = np.concatenate([tables[(0.0, False)][a][60]["filled"].values for a in ASSETS]) At = np.concatenate([tables[(0.0, False)][a][60]["adverse"].values for a in ASSETS]) Wt = np.concatenate([O_live[a]["dfrac"].abs().values for a in ASSETS]) touch60 = weighted_stats(Ft, At, Wt)["cost_bps"] print(f" Il salto e' TUTTO al primo passo (tocco {touch60:.2f} -> attraversamento sub-tick " f"{fine[1][2]:.2f} bps): non e' la CODA a decidere, e' la DISUGUAGLIANZA. Poi il costo") print(f" sale piano ({fine[-1][2]:.2f} a q=1 bps) e non torna mai sotto il taker " f"({FEE_TAKER_BPS:.2f}): non esiste un q positivo di pareggio.") # -- decomposizione: chi paga il non-fill? q0, N0 = 2.0, 60 MEAS = tables[(q0, True)] print(f" DECOMPOSIZIONE ad attraversamento q={q0:.0f} bps, N={N0} min — non tutti gli ordini sono uguali:") print(f" {'gruppo':<22}{'n':>6}{'P(fill)':>10}{'E[avversa] su TUTTI':>22}{'E[avv|no fill]':>17}") groups = {} for a in ASSETS: O = O_live[a]; M = MEAS[a][N0] groups.setdefault("SKH (breakout)", []).append((M, (O["src"] == "SKH").values)) groups.setdefault("TP01 (drift 01:00)", []).append((M, (O["src"] == "TP01").values)) groups.setdefault("aumenta esposizione", []).append((M, _increases(O))) groups.setdefault("riduce esposizione", []).append((M, ~_increases(O))) for lab, parts in groups.items(): F = np.concatenate([m["filled"].values[k] for m, k in parts]) Ad = np.concatenate([m["adverse"].values[k] for m, k in parts]) if not len(F): continue print(f" {lab:<22}{len(F):>6}{F.mean()*100:>9.1f}%{Ad.mean()*1e4:>19.1f} bps" f"{(Ad[~F].mean()*1e4 if (~F).any() else 0):>13.0f} bps") print(" NB: un non-fill che RIDUCE l'esposizione lascia la posizione aperta piu' a lungo —") print(" e' rischio, non solo prezzo, e la media in bps non lo mostra.") # -- placebo print(f"\n PLACEBO a q={q0:.0f}, N={N0} — stesse taglie a ORE CASUALI (quanto pesa il") print(" condizionamento sul segnale? se il placebo e' uguale, il segnale non seleziona nulla):") grid_cells += 2 for a in ASSETS: P = paths[a] k = rng.integers(1, len(P) - 30, size=len(O_live[a])) Of = pd.DataFrame(dict(ts=P["ts"].values[k], dfrac=O_live[a]["dfrac"].values, ref=P["ref"].values[k], hidx=P["hidx"].values[k], src="")) Mf = measure_orders(cache[a], Of, horizons=(N0,), q_bps=q0, strict=True)[N0] Mr = MEAS[a][N0] print(f" {a}: reale P(fill) {Mr['filled'].mean()*100:>5.1f}% " f"E[avversa su tutti] {Mr['adverse'].mean()*1e4:>6.1f} bps" f" | placebo P(fill) {Mf['filled'].mean()*100:>5.1f}% " f"E[avversa] {Mf['adverse'].mean()*1e4:>6.1f} bps") # -- politica "mai inseguire" print(f"\n POLITICA 'MAI INSEGUIRE' a q={q0:.0f} bps (ri-posta ogni ora, tetto {P2_MAX_HOURS}h):") grid_cells += 1 P2 = {a: measure_p2(cache[a], O_live[a], paths[a], q_bps=q0, strict=True) for a in ASSETS} c2 = np.concatenate([P2[a]["cycles"].values for a in ASSETS]) a2 = np.concatenate([P2[a]["adverse"].values for a in ASSETS]) w2 = np.concatenate([O_live[a]["dfrac"].abs().values for a in ASSETS]) ch2 = np.concatenate([P2[a]["chased"].values for a in ASSETS]) cost2 = (np.sum(w2 * (FEE_MAKER_BPS / 1e4 + a2)) / w2.sum()) * 1e4 print(f" cicli orari mediana {np.median(c2):.0f} p90 {np.percentile(c2,90):.0f} " f"max {c2.max():.0f} | mai riempito entro {P2_MAX_HOURS}h: {ch2.mean()*100:.1f}%") contrib = w2 * a2 top5 = np.sort(contrib)[-5:].sum() / contrib.sum() if contrib.sum() > 0 else 0.0 print(f" avversa: mediana {np.median(a2)*1e4:+.1f} bps p90 {np.percentile(a2,90)*1e4:+.1f} " f"MEDIA {a2.mean()*1e4:+.1f} peggiore {a2.max()*1e4:+.0f}") print(f" -> costo {cost2:.2f} bps/lato ({cost2-FEE_TAKER_BPS:+.2f} contro il taker), e " f"i 5 ordini peggiori su {len(a2)} fanno il {top5*100:.0f}% del danno totale.") print(" E' la FIRMA DELLA TRONCATURA, vista dall'altra parte: il passivo vince PICCOLO e") print(" SPESSO (mediana favorevole) e perde GROSSO e di RADO. Stessa forma del profit-take") print(" di VRP01 (30/07), dove una regola che scattava sull'86-91% dei vincenti non era") print(" protezione ma troncatura. Qui la media e' il numero che decide, non la mediana.") # ---------------------------------------------------------------- 4. impatto sul libro print("\n" + "-" * 104) print(" 4. IMPATTO SUL LIBRO (delta per-ordine innestato nei rendimenti giornalieri)") print("-" * 104) base = book_daily(cache, FEE_TAKER_BPS) b = show("TAKER 3,50 bps = il libro di oggi", base) res = {} for q, strict in [(0.0, False), (0.0, True), (1.0, True), (2.0, True), (5.0, True)]: for N in (5, 60, 120): alt = graft(base, {a: per_order_delta(O_live[a], tables[(q, strict)][a][N], miss_fee_bps=FEE_TAKER_BPS) for a in ASSETS}, O_live) tag = "TOCCO<=" if not strict else f"attrav q={q:.0f}" m = show(f"MAKER {tag:>10} N={N:>3} min, poi market", alt) res[(tag, N)] = m alt2 = graft(base, {a: per_order_delta(O_live[a], P2[a], miss_fee_bps=FEE_MAKER_BPS) for a in ASSETS}, O_live) m2 = show(f"MAKER 'mai inseguire' q={q0:.0f} bps", alt2) print("\n delta contro il baseline TAKER (a $635 di conto vero):") for (q, N), m in res.items(): print(f" {q:>10} N={N:>3} dSharpe {m['sharpe']-b['sharpe']:+.4f} " f"dHOLD {m['hold']-b['hold']:+.4f} dCAGR {(m['cagr']-b['cagr'])*100:+.3f} pp " f"dmaxDD {(m['maxdd']-b['maxdd'])*100:+.2f} pp -> ${EQUITY_LIVE*(m['cagr']-b['cagr']):+.2f}/anno") print(f" {'mai-inseg.':>10} dSharpe {m2['sharpe']-b['sharpe']:+.4f} " f"dHOLD {m2['hold']-b['hold']:+.4f} dCAGR {(m2['cagr']-b['cagr'])*100:+.3f} pp " f"dmaxDD {(m2['maxdd']-b['maxdd'])*100:+.2f} pp -> ${EQUITY_LIVE*(m2['cagr']-b['cagr']):+.2f}/anno") # ---------------------------------------------------------------- 5. robustezza print("\n" + "-" * 104) print(" 5. ROBUSTEZZA (banda RIDOTTA: gli ordini spostati di -1h/0/+1h. NON e' la banda a 24") print(" ancore: qui si sposta il momento dell'ORDINE, non l'ancora della strategia)") print("-" * 104) grid_cells += 3 for sh in (-1, 0, 1): num, den = 0.0, 0.0 for a in ASSETS: P = paths[a] pos = {int(v): k for k, v in enumerate(P["hidx"].values)} kk = np.array([min(max(pos[int(h)] + sh, 0), len(P) - 1) for h in O_live[a]["hidx"]]) Osh = pd.DataFrame(dict(ts=P["ts"].values[kk], dfrac=O_live[a]["dfrac"].values, ref=P["ref"].values[kk], hidx=P["hidx"].values[kk], src="")) M = measure_orders(cache[a], Osh, horizons=(N0,), q_bps=q0, strict=True)[N0] w = Osh["dfrac"].abs().values num += np.sum(w * (np.where(M["filled"].values, FEE_MAKER_BPS, FEE_TAKER_BPS) / 1e4 + M["adverse"].values)) den += w.sum() print(f" ordini spostati di {sh:+d}h costo maker {num/den*1e4:>6.2f} bps/lato " f"({num/den*1e4-FEE_TAKER_BPS:+.2f} contro il taker)") print(" -1h e' informativo: sposta l'ordine PRIMA che il segnale sia visibile, cioe' verso") print(" dove sta l'ingresso VERO di SKH01 (intra-bin, piu' vicino al breakout della lente") print(" usata qui). La lente di questo script e' GENEROSA verso il maker anche per questo.") # ---------------------------------------------------------------- 6. ordini di grandezza print("\n" + "-" * 104) print(" 6. LA DOMANDA STRUTTURALE — la leva accanto alle altre gia' misurate nel progetto") print("-" * 104) bq, bN = max(res, key=lambda k: res[k]["sharpe"]) bm = res[(bq, bN)] print(f" maker, LIMITE SUPERIORE (0 non-fill) dSharpe {dsh:+.4f} " f"${EQUITY_LIVE*dcg/100:>+7.2f}/anno a ${EQUITY_LIVE:,.0f} ${5000*dcg/100:>+7.2f} a $5.000") print(f" maker, migliore cella misurata ({bq},N={bN}) dSharpe {bm['sharpe']-b['sharpe']:+.4f} " f"${EQUITY_LIVE*(bm['cagr']-b['cagr']):>+7.2f}/anno a ${EQUITY_LIVE:,.0f}") worst = min(res, key=lambda k: res[k]["sharpe"]) print(f" maker, peggiore cella misurata ({worst[0]},N={worst[1]}) " f"dSharpe {res[worst]['sharpe']-b['sharpe']:+.4f} " f"${EQUITY_LIVE*(res[worst]['cagr']-b['cagr']):>+7.2f}/anno") print(f" peso di SKH01 ottimizzato dSharpe +0.0300 GATE FALLITO (26/07)") print(f" banda d'ancora del libro (FULL) dSharpe +-0.20 (26/07, 2000 estrazioni)") print(f" banda d'ancora del libro (HOLD-OUT) 2.36 canonico -> 1.54 onesto (26/07)") print(f" raddoppio del taker (regola 26/07) dSharpe -0.08 'non si tocca nulla'") print(f" versare 250 EUR/mese invece di 0 da MAI a 16-20 anni (26-27/07)") # -- rischio operativo: quanti ordini resterebbero APERTI al giro successivo del cron print("\n RISCHIO OPERATIVO (non e' un numero di rendimento, e' un modo nuovo di rompersi):") for key, lab in (((0.0, True), "attraversamento q=0"), ((2.0, True), "attraversamento q=2")): F = np.concatenate([tables[key][a][60]["filled"].values for a in ASSETS]) print(f" {lab:<22} N=60 min: {(1-F.mean())*100:>4.1f}% degli ordini ancora APERTI al giro") print(f" {'':<22} successivo = {(1-F.mean())*n_live/yrs:>4.1f} volte l'anno in cui l'esecutore") print(f" {'':<22} deve cancellare-e-rimpiazzare correttamente, su un conto vero.") print(" Oggi `rebalance_signed` manda market e finisce: nessun ordine sopravvive al giro.") print(" Un limit resting introduce ordini orfani e doppi fill, e il libro ha gia' pagato") print(" il prezzo di un percorso che si rompe in silenzio (`fresh_5m`, 26-29/07).") print("\n " + "=" * 100) print(" VERDETTO: SCARTATO. Il differenziale c'e' (2,00 bps/lato = +0,033 Sharpe = $2,96/anno") print(" a $635) ma e' il TETTO, e la selezione avversa se lo mangia tutto gia' alla condizione") print(" di riempimento piu' generosa che sia fisicamente difendibile. Il segno del risultato") print(" dipende da una disuguaglianza stretta (`<=` contro `<`), il che e' un altro modo di") print(" dire che l'effetto e' dentro il rumore: l'IC95 del delta contiene lo zero in 10 celle") print(" su 10 fra tocco e attraversamento-q=0. E il premio massimo immaginabile ($2,96/anno)") print(" e' un ordine di grandezza sotto la leva piu' piccola gia' respinta dal progetto.") print(" " + "=" * 100) print(f"\n GRIGLIA DICHIARATA: {grid_cells} celle valutate " f"(({len(QUEUE_BPS)}q attrav. + 1 tocco) x {len(HORIZONS_MIN)}N + placebo x2 + " f"mai-inseguire + 3 shift + 6 celle di griglia fine) " f"+ {len(FEE_GRID_BPS)} punti di fee.") print("=" * 104) def _increases(O: pd.DataFrame) -> np.ndarray: """True se l'ordine AUMENTA il valore assoluto dell'esposizione (entrata/incremento).""" d = O["dfrac"].values held = np.zeros(len(d)) acc = 0.0 for i, x in enumerate(d): held[i] = acc acc += x return np.abs(held + d) > np.abs(held) if __name__ == "__main__": main()