Adriano/PythagorasGoal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

skeptic(spot): il lead REGGE a $600 e si ANNULLA alla taglia del muro che sposta

Browse Source 
§52 SKEPTIC-SPOT — attacco deliberato a §39/§45 (spot al posto del perpetual per
la gamba TP01). Sei attacchi, replica bit-exact prima di ognuno (max|dif| = 0.0 su
tp01_realistic; +2,106% di sleeve e +1,580% di libro protetti da assert).

Esiti: 1 allineamento REGGE · 2 controfattuale INCRINA · 3 tetto 1,0x REGGE ·
4 spread/profondita' INCRINA · 5 haircut REGGE · 6 disaster-SL REGGE.

Il risultato che cambia una conclusione e non una presentazione: il differenziale
di cammino spot-vs-perp e' +0,72 bps a $600 (-0,04%/anno) e +25,66 bps a $272k
(-1,54%/anno = il 97% del lead lordo). Il +1,55% e' misurato a una taglia a cui
l'esecuzione e' gratis e speso dentro un muro da $290k, dove non lo e': il muro
NON si sposta del 10,9% pubblicato.

Trovato invece che la storia dello spot ESISTE (get_tradingview_chart_data serve
BTC_USDC/ETH_USDC dal 2023-04-24, 29.196 barre orarie, 0 gap): §39 la dichiarava
assente. TP01 girato sul prezzo spot VERO contro il perp, 24 ancore appaiate, da'
+0,166%/anno di sleeve con la banda che contiene lo zero (16/24) -> il
controfattuale sul PREZZO e' innocuo. Ma lo strumento non esisteva per il 55,3%
del campione e nel 2023 aveva ~9% di ore senza scambi: il lordo passa da +1,580%
(7,4 anni) a +1,450% (era spot) a +1,180% (era liquida, 2024+).

Attacchi falliti, e vanno detti: il tetto 1,0x non morde (3 giorni su BTC, tutti
nel 2018-11 e in PERDITA: troncare avrebbe fatto guadagnare); nessun haircut <=100%
rende binding il margine (copertura 50,0x replicata al decimo); il disaster-SL tolto
a TP01 vale 0,15%/anno ammortizzato allo scenario operativo.

Tre errori miei, catturati e dichiarati nello script: (a) confrontavo il costo di
SPREAD dello spot con la banda NETTA di §45 — mele contro pere, nel verso che mi
conveniva: like-with-like le due bande si sovrappongono; (b) l'aritmetica del
margine metteva SKH01 a 1,0x su ciascun asset invece che sul proprio sleeve
(25,0x invece di 50,0x: §45 aveva ragione); (c) ammortizzavo su 7,4 anni la coda
di un blackout da 30 giorni preso come argmax, ottenendo 1,97%/anno = il 127% del
lead da un evento mai accaduto.

Squalificato dal proprio controllo positivo: gli stimatori di spread da OHLC
(Roll, Corwin-Schultz) attribuiscono 4-16 bps al perpetual, che e' a un tick ->
numeri cancellati, non riportati.

Correzione a §45: «lo spot non si liquida» e' falso — BTC/ETH hanno
in_cross_collateral_pool: true, quindi sono collaterale liquidabile (non binding).

Numero onesto rivisto: [+1,38%, +1,58%]/anno a $600, -0,11%/anno a $272k.
Sola lettura provata (git status -- src/ config/ scripts/live/ tests/ data/ vuoto),
0 ordini, solo GET pubbliche con pacing e astensione nei minuti :05-:10 e :24-:30.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_018M8Ncho6QV9FWLdyy4VyQf
This commit is contained in:
Adriano Dal Pastro
2026-08-23 02:19:33 +00:00
parent 892ca6e214
commit f5ebe689da
1 changed files with 943 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/research/r0823_skeptic_spot.py
+943
View File
@@ -0,0 +1,943 @@
#!/usr/bin/env python3
"""r0823_skeptic_spot.py — §52 SKEPTIC-SPOT: attacco deliberato al lead dello SPOT.
BERSAGLIO
---------
§39 (`r0822e_funding_avoid`) e §45 (`r0823_spot_netting`) concludono che esprimere la gamba
TP01 sullo SPOT Deribit (`BTC_USDC`/`ETH_USDC`) invece che sul perpetual vale
**+1,55%/anno di drift di libro**, muro $325,0k -> $289,6k (-10,9%), netto [+1,55%, +1,66%].
Questo script parte dall'ipotesi che il numero sia SBAGLIATO e cerca dove. Sei attacchi,
in ordine di priorita' dichiarato. Ogni attacco chiude con REGGE / INCRINA / FALSIFICA.
REGOLE DEL PROGETTO APPLICATE A ME STESSO
-----------------------------------------
* **Riprodurre prima di attaccare.** Le sezioni 1 e 2 riproducono -2,16%/anno di funding,
+2,106% di sleeve e +1,580% di libro PRIMA di qualunque delta, con `assert`.
* **Il carry si misura sul LOG del rapporto dei prezzi**, mai sulla media aritmetica delle
differenze (contiene `(varA-varB)/2`).
* **Ogni delta su griglia ancorata eredita la fortuna d'ancora** -> mediana delle differenze
APPAIATE sulle 24 ancore giornaliere, mai la differenza delle mediane.
* **Epoca in ms ESPLICITA** (mai `.view`/`.astype("int64")` su DatetimeIndex tz-aware).
* Etichette di provenienza: [VENUE] misurato dall'API / [LOG] dai log / [CALC] aritmetica /
[DEDOTTO] / [ASSUNTO].
* **Un controllo positivo che fallisce squalifica lo STRUMENTO, non il bersaglio**: la sez. 5
butta via i propri stimatori di spread perche' non superano il controllo positivo sul perp.
SOLA LETTURA / RETE
-------------------
Nessuna scrittura nel repo. Nessun ordine, mai (solo GET **pubbliche**, nessuna chiave).
Cache dello storico spot FUORI dal repo. Pacing <=2 req/s e guardia che si astiene nei
minuti :05-:10 e :24-:30 (finestre del cron di produzione).
uv run python scripts/research/r0823_skeptic_spot.py # con rete
uv run python scripts/research/r0823_skeptic_spot.py --no-net # solo cache
"""
from __future__ import annotations
import contextlib
import datetime as dt
import io
import json
import sys
import time
import urllib.request
from pathlib import Path
import numpy as np
import pandas as pd
ROOT = Path("/opt/docker/PythagorasGoal")
for _p in (ROOT / "scripts" / "research", ROOT / "scripts" / "research" / "alt", ROOT):
sys.path.insert(0, str(_p))
# r0822d_funding legge `--no-net` da sys.argv all'import: qui il funding viene dalla sua
# cache. Le uniche chiamate di rete di QUESTO script sono le mie, sotto, col mio flag.
_NET = "--no-net" not in sys.argv
if "--no-net" not in sys.argv:
sys.argv.append("--no-net")
import altlib as al # noqa: E402
import r0702_tp01_offset as r02t # noqa: E402 (resample_offset: le 24 ancore)
import r0725_capcurve as CC # noqa: E402 (DERIBIT_FEE_SIDE/MIN_ORDER)
import r0822d_funding as FD # noqa: E402 (build_rate/daily_funding/stat)
from src.data.downloader import load_data # noqa: E402
from src.strategies.trend_portfolio import ( # noqa: E402
CANONICAL, TrendPortfolio, resample_1d, simple_returns)
# ---------------------------------------------------------------------------
# Costanti dichiarate
# ---------------------------------------------------------------------------
ASSETS = ("BTC", "ETH")
ALLOC = 600.0 # il capitale su cui girano §39/§45
W_TP, W_SKH = 0.75, 0.25
EPOCH = pd.Timestamp("1970-01-01", tz="UTC")
MSH = 3_600_000
SPOT = {"BTC": "BTC_USDC", "ETH": "ETH_USDC"}
PERP_LIN = {"BTC": "BTC_USDC-PERPETUAL", "ETH": "ETH_USDC-PERPETUAL"}
# [VENUE] creation_timestamp di get_instruments?currency=USDC&kind=spot (letto in sessione)
SPOT_BIRTH = pd.Timestamp("2023-04-24 13:00", tz="UTC")
API = "https://www.deribit.com/api/v2/public/"
_SP = Path("/tmp/claude-1001/-opt-docker-PythagorasGoal/"
"b6cc75e7-14f8-4c32-bd07-ab8a0d2aaee6/scratchpad")
CACHE = _SP if _SP.parent.exists() else Path("/tmp/pythagoras_skeptic_spot")
# §45, da citare e attaccare
S45_HALF_SPREAD_BPS = (1.5, 3.1) # mezzo spread misurato su DUE istantanee a 4 minuti
S45_ESEC_BAND = (-0.034, +0.077) # contributo netto d'esecuzione, %/anno
S45_NETTO = (1.55, 1.66) # il numero da citare, %/anno di libro
S45_MARGIN_COVER = 50.0 # copertura del margine di SKH01
CAP_TETTO_PUBBL = 0.102 # % di giorni in cui TP01 supera 1,0x (§39)
_last = [0.0]
def sez(t: str) -> None:
print("\n" + "=" * 94 + f"\n{t}\n" + "=" * 94)
def _guard_window() -> None:
"""Astensione nei minuti del cron di produzione (:05-:10 e :24-:30)."""
while True:
m = dt.datetime.now(dt.timezone.utc).minute
if not (5 <= m <= 10 or 24 <= m <= 30):
return
tgt = 11 if m <= 10 else 31
w = (tgt - m) * 60 - dt.datetime.now(dt.timezone.utc).second
print(f" [rete] minuto :{m:02d} e' finestra del cron -> attendo {max(w,1)}s")
time.sleep(max(w, 1))
def _get(path: str, tries: int = 4):
"""GET pubblica, pacing <=2 req/s, guardia sulle finestre del cron."""
if not _NET:
raise RuntimeError("--no-net: rete disabilitata")
_guard_window()
for k in range(tries):
d = time.time() - _last[0]
if d < 0.55:
time.sleep(0.55 - d)
_last[0] = time.time()
try:
with urllib.request.urlopen(API + path, timeout=30) as r:
return json.load(r)
except Exception as ex: # noqa: BLE001
if k == tries - 1:
print(f" [rete] KO {path[:60]}: {ex}")
return None
time.sleep(1.5 * (k + 1))
return None
def to_ms(idx: pd.DatetimeIndex) -> np.ndarray:
"""Epoca in ms ESPLICITA. Mai `.view`/`.astype('int64')` su tz-aware."""
return ((pd.DatetimeIndex(idx) - EPOCH) // pd.Timedelta(milliseconds=1)).values.astype(np.int64)
# ===========================================================================
# 0) LO SPOT: esistenza, nascita, storia
# ===========================================================================
def spot_1h(a: str) -> pd.DataFrame:
"""Storia oraria dello SPOT Deribit, cache fuori dal repo. Schema `al.get`-compatibile."""
f = CACHE / f"spot_{a.lower()}_1h.parquet"
have = pd.read_parquet(f) if f.exists() else pd.DataFrame()
if len(have) and (not _NET or
pd.Timestamp(have["ts"].max()) > pd.Timestamp.now(tz="UTC") - pd.Timedelta("6h")):
d = have.copy()
else:
out = ([have] if len(have) else [])
t = int(SPOT_BIRTH.timestamp() * 1000) if not len(have) else int(
pd.Timestamp(have["ts"].max()).timestamp() * 1000) + MSH
end = int(pd.Timestamp.now(tz="UTC").floor("h").timestamp() * 1000)
while t < end:
e = min(t + 900 * MSH, end)
r = _get(f"get_tradingview_chart_data?instrument_name={SPOT[a]}"
f"&start_timestamp={t}&end_timestamp={e}&resolution=60")
r = (r or {}).get("result", {})
if r.get("status") != "ok" or not r.get("ticks"):
break
out.append(pd.DataFrame(dict(
ts=pd.to_datetime(r["ticks"], unit="ms", utc=True), open=r["open"],
high=r["high"], low=r["low"], close=r["close"], volume=r["volume"])))
t = int(r["ticks"][-1]) + MSH
d = pd.concat(out).drop_duplicates("ts").sort_values("ts").reset_index(drop=True)
CACHE.mkdir(parents=True, exist_ok=True)
d.to_parquet(f)
d = d.rename(columns={"ts": "datetime"})
d["datetime"] = pd.DatetimeIndex(pd.to_datetime(d["datetime"], utc=True))
d["timestamp"] = to_ms(d["datetime"])
return d[["timestamp", "open", "high", "low", "close", "volume", "datetime"]]
def perp_1h(a: str) -> pd.DataFrame:
d = al.get(a, "1h").copy()
d["datetime"] = pd.DatetimeIndex(pd.to_datetime(d["datetime"], utc=True))
return d
# ===========================================================================
# 1) REPLICA — riprodurre i numeri PRIMA di attaccarli
# ===========================================================================
def tp01_leg(df1h: pd.DataFrame, cap_asset: float, rate: dict | None, a: str) -> tuple:
"""Replica per-asset di `FD.tp01_realistic`, con il prezzo come grado di liberta'.
Con (feed perp, rate) deve dare bit-exact `FD.tp01_realistic`: e' la replica di controllo."""
tp = TrendPortfolio(**CANONICAL)
df = resample_1d(df1h)
idx = pd.DatetimeIndex(pd.to_datetime(df["datetime"], utc=True))
r = simple_returns(df["close"].values.astype(float))
tgt = np.nan_to_num(np.asarray(tp.target_series(df), float))
held = np.empty(len(tgt))
cur = 0.0
for i in range(len(tgt)):
if abs(tgt[i] - cur) * cap_asset >= CC.DERIBIT_MIN_ORDER:
cur = tgt[i]
held[i] = cur
pos = np.zeros(len(held))
pos[1:] = held[:-1]
net = pos * r - CC.DERIBIT_FEE_SIDE * np.abs(np.diff(pos, prepend=0.0))
if rate is not None:
net = net - pos * FD.daily_funding(rate[a], idx)
net[0] = 0.0
return pd.Series(np.clip(net, -0.99, None), index=idx), tgt, pos, held
def tp01_book(dfs: dict, alloc: float, rate: dict | None) -> pd.Series:
cap = alloc / len(ASSETS)
s = {a: tp01_leg(dfs[a], cap, rate, a)[0] for a in ASSETS}
J = pd.concat(s, axis=1, join="inner").fillna(0.0)
out = pd.Series(0.5 * J["BTC"].values + 0.5 * J["ETH"].values, index=J.index)
if out.index.tz is None:
out.index = out.index.tz_localize("UTC")
return out
def sezione1() -> dict:
sez("1) REPLICA DI CONTROLLO — i numeri attaccati, riprodotti prima di toccarli")
with contextlib.redirect_stdout(io.StringIO()):
F = FD.sezione0()["F"]
rate = FD.build_rate(F)
ref, dg = FD.tp01_realistic(ALLOC * W_TP, rate)
perp = {a: perp_1h(a) for a in ASSETS}
mine = tp01_book(perp, ALLOC * W_TP, rate)
c = pd.concat([ref, mine], axis=1, join="inner")
dmax = float(np.max(np.abs(c.iloc[:, 0] - c.iloc[:, 1])))
print(f" [CALC] la mia replica di TP01 contro `FD.tp01_realistic`: max|dif| = {dmax:.1e}")
assert dmax == 0.0, "replica non bit-exact: nessun attacco oltre questo punto"
nc = FD.tp01_realistic(ALLOC * W_TP, None)[0]
a0, a1 = FD.stat(nc), FD.stat(ref)
risp = a0["drift"] - a1["drift"]
print(f" TP01 sleeve: senza carry drift {a0['drift']:+.3%} / con perp {a1['drift']:+.3%} "
f"-> risparmio {risp:+.3%} (§45 pubblica +2,11%)")
print(f" a peso di libro x{W_TP}: {risp*W_TP:+.3%} (§39/§45 pubblicano +1,55/+1,58%)")
assert abs(risp - 0.02106) < 3e-4, "il +2,11% di sleeve non replica"
assert abs(risp * W_TP - 0.0158) < 3e-4, "il +1,58% di libro non replica"
print(f" vol dello sleeve: {a0['vol']:.2%} senza carry contro {a1['vol']:.2%} col perp "
f"-> il carry e' un DRIFT, non rischio (gate iso-vol di §39)")
print(" ✅ REPLICATO. Da qui in poi ogni numero e' un delta su questa base.")
return dict(rate=rate, perp=perp, nc=nc, perp_s=ref, risp=risp, dg=dg)
# ===========================================================================
# 2) ATTACCO 1 — il funding e lo spot sono la stessa quantita'? il LAG
# ===========================================================================
def attacco1(rep: dict) -> str:
sez("ATTACCO 1 — «funding misurato su una serie, spot su un'altra»: il LAG e l'oggetto")
print(" §39 ha catturato in sessione uno sfasamento di UN'ORA fra `fund_*.parquet`")
print(" (indice etichettato all'ISTANTE T) e `al.get(a,'1h')` (barra etichettata")
print(" all'INIZIO). Riparato con `aligned_index(...).shift(-1)`. Domanda: la stessa")
print(" famiglia e' altrove? E le due serie parlano davvero dello stesso oggetto?\n")
FUT = ROOT / "data" / "raw" / "fut_deribit"
ok_lag = True
print(" (a) TRE serie indipendenti, correlazione dei log-rendimenti orari per LAG")
print(f" {'asset':>5} {'coppia':<22} {'lag -1':>9} {'lag 0':>9} {'lag +1':>9} verdetto")
for a in ASSETS:
fu = pd.read_parquet(FUT / f"fund_{a}.parquet").set_index("ts").sort_index()
pe = perp_1h(a).set_index("datetime")["close"].astype(float)
sp = spot_1h(a).set_index("datetime")["close"].astype(float) if True else None
g = pe.index.intersection(fu.index).intersection(sp.index)
ix = fu["index"].reindex(g).values.astype(float)
pv = pe.reindex(g).values.astype(float)
sv = sp.reindex(g).values.astype(float)
def corr_lag(x, y, k):
rx, ry = np.diff(np.log(x)), np.diff(np.log(y))
if k > 0:
rx, ry = rx[:-k], ry[k:]
elif k < 0:
rx, ry = rx[-k:], ry[:k]
m = np.isfinite(rx) & np.isfinite(ry)
return float(np.corrcoef(rx[m], ry[m])[0, 1])
for lbl, x, y in (("perp vs indice", pv, ix), ("SPOT vs indice", sv, ix),
("SPOT vs perp", sv, pv)):
cs = [corr_lag(x, y, k) for k in (-1, 0, 1)]
best = int(np.argmax(cs)) - 1
good = (best == 1) if "indice" in lbl else (best == 0)
ok_lag &= good
print(f" {a:>5} {lbl:<22} {cs[0]:>+9.4f} {cs[1]:>+9.4f} {cs[2]:>+9.4f} "
f"max a lag {best:+d} {'(atteso)' if good else '⚠️ INATTESO'}")
print(" -> l'indice va confrontato a lag +1 (shift(-1) = cio' che `aligned_index` fa);")
print(" lo SPOT, scaricato da un ENDPOINT DIVERSO, ha la stessa etichettatura del")
print(" perp (max a lag 0) = TERZA conferma indipendente della convenzione.")
print("\n (b) LA DOMANDA VERA: da dove viene il +1,55%? Non dall'indice.")
grid = {}
for a in ASSETS:
fu = pd.read_parquet(FUT / f"fund_{a}.parquet").set_index("ts").sort_index()
pe = perp_1h(a).set_index("datetime")["close"].astype(float)
g = pe.index.intersection(fu.index)
ix = fu["index"].reindex(g).shift(-1).values.astype(float)
pv = pe.reindex(g).values.astype(float)
m = np.isfinite(ix) & np.isfinite(pv)
rlog = np.diff(np.log(pv[m])) - np.diff(np.log(ix[m]))
rari = np.diff(pv[m]) / pv[m][:-1] - np.diff(ix[m]) / ix[m][:-1]
fund = float(np.mean(rep["rate"][a]["i1h"])) * 24 * 365.25
grid[a] = (float(np.mean(rlog)) * 24 * 365.25, float(np.mean(rari)) * 24 * 365.25, fund)
print(f" {a}: `spot_rate` (perp-indice) LOG {grid[a][0]:+.3%}/anno "
f"ARITMETICO {grid[a][1]:+.3%}/anno (artefatto {grid[a][1]-grid[a][0]:+.3%})")
print(f" funding orario medio, annualizzato: {fund:+.3%}/anno")
print(" -> il PREZZO del perpetual non si stacca dall'indice (log ~0): il funding e'")
print(" un FLUSSO DI CASSA, non un drift di prezzo. Quindi `spot_rate` vale ~0 e")
print(" la variante E) di §39 e' quasi identica alla F) «nessun carry».")
e_min_f = float(np.mean([abs(grid[a][0]) for a in ASSETS])) * W_TP
print(f" -> **il +1,55% NON viene dall'indice: viene dalla serie `rate` del funding**")
print(f" (contributo dell'indice al numero di libro: |{e_min_f:.3%}|/anno).")
print(" E quella serie e' gia' validata contro il conto vero in §35 (ETH -0,2% /")
print(" BTC -7,7% contro `realized_pnl` di sessione). Il vettore d'attacco 1")
print(" punta a un pezzo che pesa quasi nulla.")
v = "REGGE" if ok_lag else "INCRINA"
print(f"\n ESITO ATTACCO 1: **{v}** — nessuno sfasamento residuo su 6 coppie/2 asset, e la")
print(" quantita' attaccata (l'indice) contribuisce <0,01%/anno al numero pubblicato.")
return v
# ===========================================================================
# 3) ATTACCO 2 — controfattuale o misura? lo SPOT VERO
# ===========================================================================
def attacco2(rep: dict) -> tuple[str, dict]:
sez("ATTACCO 2 — «lo spot e' un mercato a se'»: quanto del lead e' MISURATO e quanto MODELLATO")
print(" §39 dichiara in un commento: «qui lo spot e' modellato con l'INDICE Deribit, non")
print(" col libro spot vero (che non ha storia in questo progetto)». Verifico se e' vero")
print(" che la storia non c'e'.\n")
print(" (a) [VENUE] `get_instruments?currency=USDC&kind=spot` -> BTC_USDC/ETH_USDC hanno")
print(f" `creation_timestamp` = {SPOT_BIRTH.date()}. E `get_tradingview_chart_data`")
print(" SERVE la storia oraria di uno strumento spot: **la storia c'era.**")
sp = {a: spot_1h(a) for a in ASSETS}
for a in ASSETS:
d = sp[a]
gridfull = pd.date_range(d["datetime"].min(), d["datetime"].max(), freq="1h")
print(f" {a}: {len(d):,} barre orarie {d['datetime'].min().date()} .. "
f"{d['datetime'].max().date()}, ore mancanti nella griglia "
f"{len(gridfull) - len(d)}")
tp_full = rep["nc"]
quota = float((tp_full.index >= SPOT_BIRTH).mean())
print(f"\n (b) 🚨 LA PARTE CHE §39/§45 NON DICONO: lo strumento **non esisteva** per il")
print(f" {1-quota:.1%} del campione di TP01 (il libro parte dal 2019-03, lo spot dal")
print(f" {SPOT_BIRTH.date()}). Su quella parte il lead non e' misurabile NEMMENO IN")
print(" LINEA DI PRINCIPIO: non c'e' un mercato su cui l'ordine sarebbe andato.")
a0 = FD.stat(rep["nc"])
a1 = FD.stat(rep["perp_s"])
e0 = FD.stat(rep["nc"][rep["nc"].index >= SPOT_BIRTH])
e1 = FD.stat(rep["perp_s"][rep["perp_s"].index >= SPOT_BIRTH])
r_full = (a0["drift"] - a1["drift"]) * W_TP
r_era = (e0["drift"] - e1["drift"]) * W_TP
print(f" risparmio di libro sul campione PIENO ({a0['n']:,} g): {r_full:+.3%}/anno")
print(f" risparmio di libro sull'ERA SPOT ({e0['n']:,} g): {r_era:+.3%}/anno "
f"({r_era/r_full-1:+.1%} contro il pubblicato)")
LIQ = pd.Timestamp("2024-01-01", tz="UTC")
l0 = FD.stat(rep["nc"][rep["nc"].index >= LIQ])
l1 = FD.stat(rep["perp_s"][rep["perp_s"].index >= LIQ])
r_liq = (l0["drift"] - l1["drift"]) * W_TP
print(f" risparmio di libro sull'era LIQUIDA (2024+, {l0['n']:,} g): {r_liq:+.3%}/anno "
f"({r_liq/r_full-1:+.1%})")
print(" -> il lead NON e' concentrato nel periodo non verificabile: sopravvive quasi")
print(" intero anche guardando solo dove lo strumento esiste. **Questo assolve**")
print(" §39 sull'accusa piu' grave che gli si poteva muovere.")
print("\n (c) IL TEST CHE §39 NON HA POTUTO FARE: TP01 girato sul prezzo SPOT VERO")
print(" (segnale E rendimenti dallo spot) contro TP01 sul perp, stessa finestra,")
print(" **24 ancore giornaliere, mediana delle differenze APPAIATE**.")
per = {a: perp_1h(a) for a in ASSETS}
lo = max(max(sp[a]["datetime"].min() for a in ASSETS),
max(per[a]["datetime"].min() for a in ASSETS))
hi = min(min(sp[a]["datetime"].max() for a in ASSETS),
min(per[a]["datetime"].max() for a in ASSETS))
cut = {}
for a in ASSETS:
for nm, d in (("spot", sp[a]), ("perp", per[a])):
x = d[(d["datetime"] >= lo) & (d["datetime"] <= hi)].reset_index(drop=True)
cut[(nm, a)] = x
cap = ALLOC * W_TP / 2
rows = []
for h in range(24):
st = {}
for nm in ("spot", "perp"):
legs = {}
for a in ASSETS:
dd = r02t.resample_offset(cut[(nm, a)], h)
tp = TrendPortfolio(**CANONICAL)
idx = pd.DatetimeIndex(pd.to_datetime(dd["datetime"], utc=True))
r = simple_returns(dd["close"].values.astype(float))
tgt = np.nan_to_num(np.asarray(tp.target_series(dd), float))
held = np.empty(len(tgt))
cur = 0.0
for i in range(len(tgt)):
if abs(tgt[i] - cur) * cap >= CC.DERIBIT_MIN_ORDER:
cur = tgt[i]
held[i] = cur
pos = np.zeros(len(held))
pos[1:] = held[:-1]
net = pos * r - CC.DERIBIT_FEE_SIDE * np.abs(np.diff(pos, prepend=0.0))
net[0] = 0.0
legs[a] = pd.Series(np.clip(net, -0.99, None), index=idx)
J = pd.concat(legs, axis=1, join="inner").fillna(0.0)
s = pd.Series(0.5 * J["BTC"].values + 0.5 * J["ETH"].values, index=J.index)
st[nm] = FD.stat(s)
rows.append((h, st["spot"]["drift"] - st["perp"]["drift"],
st["spot"]["full"] - st["perp"]["full"]))
dd_ = np.array([r[1] for r in rows])
ds_ = np.array([r[2] for r in rows])
med = float(np.median(dd_))
print(f" finestra comune {lo.date()} .. {hi.date()} ({(hi-lo).days} giorni)")
print(f" Δdrift (spot vero - perp), 24 ancore: mediana APPAIATA {med:+.3%}/anno di")
print(f" sleeve, p10 {np.percentile(dd_,10):+.3%} p90 {np.percentile(dd_,90):+.3%}, "
f"positivo in {int((dd_>0).sum())}/24")
print(f" ΔSharpe: mediana {float(np.median(ds_)):+.3f}, positivo in "
f"{int((ds_>0).sum())}/24")
err = abs(med) * W_TP
print(f" -> la banda CONTIENE LO ZERO (16/24 positive = monetina) e il punto stimato e'")
print(f" a FAVORE dello spot. **L'errore di modellazione del prezzo vale al piu'")
print(f" |{err:.3%}|/anno di libro** ({err/abs(r_full)*100:.0f}% del +{r_full*100:.2f}% pubblicato),")
print(" e non ha nemmeno segno stabilito: il controfattuale sul PREZZO e' innocuo")
print(" dove e' verificabile. ⚠️ Cio' non lo rende innocuo dove non lo e'.")
print("\n (d) MA la base spot/perp non e' zero, e la sua storia dice un'altra cosa:")
print(f" {'anno':>6} {'vol=0':>8} {'flat':>8} {'base med':>10} {'sd base':>9} "
f"{'vol mediano/ora':>18}")
liq = {}
for a in ASSETS:
S = sp[a].set_index("datetime")
P = per[a].set_index("datetime")
g = S.index.intersection(P.index)
S, P = S.loc[g], P.loc[g]
print(f" {a}:")
for y, gr in S.groupby(S.index.year):
pr = P.loc[gr.index]
b = (gr["close"] / pr["close"] - 1.0).values
print(f" {y:>6} {float((gr['volume']<=0).mean()):>7.2%} "
f"{float(((gr['high']-gr['low'])<=0).mean()):>7.2%} "
f"{float(np.nanmedian(b))*1e4:>+9.2f}b {float(np.nanstd(b))*1e4:>8.1f}b "
f"{float(gr['volume'].median()):>16.3f} {a}")
liq[a] = (float((S["volume"] <= 0).mean()), float(((S["high"] - S["low"]) <= 0).mean()))
print(" -> il libro spot **e' maturato**: nel 2023 aveva ~9% di ore senza scambi e")
print(" ~15% di barre piatte (la soglia con cui questo progetto ESCLUSE gli alt");
print(" Deribit nel giugno), nel 2025-26 e' continuo. **Il lead e' verificabile")
print(" solo dove lo strumento e' anche liquido: dal 2024, cioe' 2,6 anni.**")
v = "INCRINA"
print(f"\n ESITO ATTACCO 2: **{v}** — non falsifica (dove misurabile l'errore vale")
print(f" {err:.3%}/anno, il {err/abs(r_full)*100:.0f}% del lead), ma **la dichiarazione di")
print(" §45 «il numero da citare e' [+1,55%, +1,66%]» nasconde che il 55,3% del campione e'")
print(" su uno strumento che NON ESISTEVA, e un altro 10% su un libro che sarebbe stato")
print(" scartato dalla certificazione di questo progetto per barre piatte.**")
return v, dict(r_full=r_full, r_era=r_era, r_liq=r_liq, err=err, quota=quota, dd=dd_)
# ===========================================================================
# 4) ATTACCO 3 — il tetto 1,0x
# ===========================================================================
def attacco3() -> str:
sez("ATTACCO 3 — il tetto 1,0x: e se quello 0,102% di giorni fossero i giorni BUONI?")
print(" Sullo spot non si e' long piu' del cash. §39 dice che TP01 supera 1,0x nello")
print(" 0,102% dei giorni e conclude «il tetto non morde». Un troncamento nello 0,1% dei")
print(" giorni puo' costare molto se quei giorni sono i migliori (precedente del progetto:")
print(" il backtest di SKH01 concentra ~39% del log-equity in 5 giorni).\n")
tp = TrendPortfolio(**CANONICAL)
tot_n = tot_m = 0
persa = 0.0
for a in ASSETS:
df = resample_1d(load_data(a, "1h"))
idx = pd.DatetimeIndex(pd.to_datetime(df["datetime"], utc=True))
tgt = np.nan_to_num(np.asarray(tp.target_series(df), float))
r = simple_returns(df["close"].values.astype(float))
pos = np.zeros(len(tgt))
pos[1:] = tgt[:-1]
m = tgt > 1.0
mm = np.zeros(len(tgt), bool)
mm[1:] = m[:-1]
tot_n += len(tgt)
tot_m += int(m.sum())
gross = pos * r
print(f" {a}: {len(tgt):,} giorni, sopra 1,0x {int(m.sum())} ({m.mean():.4%}), "
f"target massimo {tgt.max():.3f}")
if m.sum():
print(f" date: {', '.join(str(d.date()) for d in idx[m])}")
print(f" rendimento LORDO di quei giorni {gross[mm].sum():+.4%} contro "
f"{gross.sum():+.3%} di tutto il campione = {gross[mm].sum()/gross.sum():+.3%}")
exc = np.clip(pos - 1.0, 0.0, None)
d = float((exc * r)[mm].sum())
persa += d
print(f" effetto del troncamento a 1,0x: {d:+.5%} di lordo TOTALE "
f"({'un GUADAGNO' if d < 0 else 'una perdita'} per il tetto)")
print(f"\n totale: {tot_m} giorni su {tot_n:,} asset-giorni; **su BTC da solo e' lo 0,1023%,")
print(f" cioe' esattamente lo {CAP_TETTO_PUBBL:.3f}% pubblicato da §39** (il numero era")
print(f" per-asset, non di portafoglio) — effetto netto del tetto {-persa:+.5%} di lordo.")
print(" 📌 E i giorni sono 2018-11-10/11/12: **PRIMA dell'era del libro** (2019-03) e in")
print(" PERDITA. Il tetto non solo non morde: troncare avrebbe fatto GUADAGNARE.")
print("\n ESITO ATTACCO 3: **REGGE** (l'attacco fallisce, e il segno e' l'opposto di quello")
print(" che cercavo: e' l'unico dei sei in cui il bersaglio esce piu' forte).")
return "REGGE"
# ===========================================================================
# 5) ATTACCO 4 — lo spread: due istantanee non sono una distribuzione
# ===========================================================================
def walk_cost_bps(book: dict, usd: float, side: str) -> float | None:
"""Costo effettivo vs mid di un ordine da `usd`, camminando il libro (stessa convenzione
di `r0822e_funding_avoid.walk_cost_bps`). None se il libro visibile NON basta — e allora
si dichiara, non si estrapola."""
if not book or not book.get("bid") or not book.get("ask"):
return None
mid = (book["bid"] + book["ask"]) / 2.0
lv = book.get("asks" if side == "buy" else "bids") or []
need, got, cost = usd / mid, 0.0, 0.0
for p, q in lv:
take = min(q, need - got)
got += take
cost += take * p
if got >= need - 1e-12:
break
if got < need - 1e-12:
return None
return (cost / got / mid - 1.0) * 1e4 * (1.0 if side == "buy" else -1.0)
def roll_bps(c: np.ndarray) -> float:
r = np.diff(np.log(c))
r = r[np.isfinite(r)]
cv = float(np.cov(r[:-1], r[1:])[0, 1])
return 2 * np.sqrt(-cv) * 1e4 if cv < 0 else float("nan")
def cs_bps(h: np.ndarray, l: np.ndarray) -> float:
"""Corwin-Schultz (2012), mediana dello spread relativo in bps."""
b = (np.log(h[1:] / l[1:])) ** 2 + (np.log(h[:-1] / l[:-1])) ** 2
H = np.maximum(h[1:], h[:-1])
L = np.minimum(l[1:], l[:-1])
g = (np.log(H / L)) ** 2
k = 3 - 2 * np.sqrt(2.0)
a = (np.sqrt(2 * b) - np.sqrt(b)) / k - np.sqrt(g / k)
s = 2 * (np.exp(a) - 1) / (1 + np.exp(a))
return float(np.nanmedian(np.clip(np.where(np.isfinite(s), s, np.nan), 0, None))) * 1e4
def attacco4(rep: dict, ROUNDS: int = 10) -> tuple[str, dict]:
sez("ATTACCO 4 — lo SPREAD: §45 ha DUE istantanee a 4 minuti che differiscono del 79%")
print(" (a) PRIMA PROVA, e va buttata via. Stimatori di spread dalle sole OHLC (Roll,")
print(" Corwin-Schultz) su spot e perp. **Controllo positivo obbligatorio:** il perp")
print(" e' un mercato a UN TICK (mezzo spread noto ~0,01-0,04 bps); se lo stimatore")
print(" non lo riconosce, non ha risoluzione per questa domanda.")
sp = {a: spot_1h(a) for a in ASSETS}
per = {a: perp_1h(a) for a in ASSETS}
bad = 0
for a in ASSETS:
S = sp[a].set_index("datetime")
P = per[a].set_index("datetime")
g = S.index.intersection(P.index)
S, P = S.loc[g], P.loc[g]
rs, rp = roll_bps(S["close"].values), roll_bps(P["close"].values)
cssp, cspp = cs_bps(S["high"].values, S["low"].values), cs_bps(P["high"].values, P["low"].values)
print(f" {a}: Roll spot {rs:6.2f} / perp {rp:6.2f} bps | "
f"Corwin-Schultz spot {cssp:5.2f} / perp {cspp:5.2f} bps")
bad += int(rp > 1.0) + int(cspp > 1.0)
print(f" -> ⚠️ CONTROLLO POSITIVO FALLITO {bad}/4: attribuiscono al perpetual 4-16 bps")
print(" di spread dove ce ne sono ~0,02. **Questi stimatori NON hanno potenza qui**")
print(" (misurano la volatilita' per barra, non lo spread) -> i loro numeri sono")
print(" cancellati, non riportati. *Uno strumento va squalificato dal suo controllo")
print(" positivo, non dal fatto che dia la risposta che mi serviva.*")
print(f"\n (b) SECONDA PROVA: campionamento del BOOK VERO, {ROUNDS} giri (non 2).")
obs = {k: [] for k in ("BTC_spot", "ETH_spot", "BTC_perp", "ETH_perp")}
depth = {}
for i in range(ROUNDS if _NET else 0):
for a in ASSETS:
for nm, inst in (("spot", SPOT[a]), ("perp", PERP_LIN[a])):
r = _get(f"get_order_book?instrument_name={inst}&depth=20")
r = (r or {}).get("result") or {}
b, k = r.get("bids") or [], r.get("asks") or []
if not b or not k:
continue
mid = (b[0][0] + k[0][0]) / 2
obs[f"{a}_{nm}"].append((k[0][0] - b[0][0]) / 2 / mid * 1e4)
depth.setdefault(f"{a}_{nm}", []).append(dict(
bid=b[0][0], ask=k[0][0], bids=b, asks=k,
bid_top=b[0][0] * b[0][1], ask_top=k[0][0] * k[0][1],
bid20=sum(p * q for p, q in b), ask20=sum(p * q for p, q in k),
span=abs(k[-1][0] - k[0][0]) / k[0][0] * 1e4))
if i < ROUNDS - 1:
time.sleep(12)
ms = {}
print(f" {'strumento':<12} {'n':>3} {'mediana':>9} {'min':>8} {'max':>8} [VENUE]")
for k, v in obs.items():
if not v:
continue
ms[k] = float(np.median(v))
print(f" {k:<12} {len(v):>3} {np.median(v):>8.2f}b {min(v):>7.2f}b {max(v):>7.2f}b")
if not ms:
print(" (nessun campione: --no-net)")
return "NON PORTATO", {}
hs = float(np.mean([ms.get("BTC_spot", np.nan), ms.get("ETH_spot", np.nan)]))
hp = float(np.mean([ms.get("BTC_perp", np.nan), ms.get("ETH_perp", np.nan)]))
print(f" -> mezzo spread SPOT mediano {hs:.2f} bps contro PERP {hp:.3f} bps.")
print(f" §45 misuro' {S45_HALF_SPREAD_BPS[0]}-{S45_HALF_SPREAD_BPS[1]} bps: il mio")
print(f" campione sta **{hs/np.mean(S45_HALF_SPREAD_BPS):.1f}x** sopra il centro di")
print(" quella banda. ⚠️ Anche il mio e' UNA finestra oraria: dice che 2 istantanee")
print(" sottostimavano, non qual e' la distribuzione giornaliera.")
print("\n (c) IL COSTO — e qui ho commesso IO l'errore che stavo cercando negli altri.")
print(" La prima stesura confrontava il costo di SPREAD dello spot con la banda")
print(f" [{S45_ESEC_BAND[0]:+.3f}%, {S45_ESEC_BAND[1]:+.3f}%] di §45, che e' il contributo NETTO")
print(" dell'esecuzione: **spread pagato MENO commissione risparmiata** (lo spot ha")
print(" taker 0,00). Mele contro pere, e nel verso che mi conveniva. Rifatto")
print(" like-with-like, sul turnover di TP01 MISURATO qui (non ereditato):")
tover = {}
for a in ASSETS:
df = resample_1d(load_data(a, "1h"))
tp = TrendPortfolio(**CANONICAL)
tgt = np.nan_to_num(np.asarray(tp.target_series(df), float))
yrs = len(tgt) / 365.25
for capbook in (ALLOC, 5_000.0, 50_000.0, 272_000.0):
cap = capbook * W_TP / 2
held = np.empty(len(tgt))
cur = 0.0
for i in range(len(tgt)):
if abs(tgt[i] - cur) * cap >= CC.DERIBIT_MIN_ORDER:
cur = tgt[i]
held[i] = cur
d = np.abs(np.diff(held, prepend=0.0)) * cap
nz = d[d > 0]
tover.setdefault(capbook, {})[a] = dict(
turn=float(d.sum() / yrs), n=len(nz) / yrs, sizes=nz,
med=float(np.median(nz)), p95=float(np.percentile(nz, 95)), mx=float(nz.max()))
T600 = sum(tover[ALLOC][a]["turn"] for a in ASSETS)
print(f" turnover spot della gamba TP01: ${T600:,.0f}/anno su ${ALLOC:.0f} di equity")
print(f" {'mezzo spread':<26} {'spread pagato':>14} {'commiss. risp.':>15} {'NETTO':>10}")
esec_tab = {}
for lbl, h in (("§45 (2 istantanee)", float(np.mean(S45_HALF_SPREAD_BPS))),
("mio campione (10 giri)", hs)):
for flbl, f in (("fee modellata 5,0b", CC.DERIBIT_FEE_SIDE * 1e4),
("fee reale 3,5b", 3.50)):
pag = T600 * h / 1e4 / ALLOC
ris = T600 * f / 1e4 / ALLOC
esec_tab[(lbl, flbl)] = ris - pag
print(f" {lbl:<26} {-pag:>13.3%} {ris:>14.3%} {ris-pag:>+9.3%} [{flbl}]")
esec = float(np.median(list(esec_tab.values())))
lo_e, hi_e = min(esec_tab.values()), max(esec_tab.values())
print(f" -> contributo NETTO dell'esecuzione: **[{lo_e:+.3%}, {hi_e:+.3%}]**, contro la")
print(f" banda pubblicata da §45 [{S45_ESEC_BAND[0]:+.3f}%, {S45_ESEC_BAND[1]:+.3f}%].")
print(" 📌 **Le due bande si SOVRAPPONGONO.** Il mezzo spread e' 1,8x il pubblicato, ma")
print(" il numero che entra nel lead non cambia, perche' la commissione risparmiata")
print(" (lo spot e' a fee ZERO) e' dello stesso ordine di grandezza e va nell'altro")
print(" verso. **Attaccare lo spread da solo non poteva funzionare, e il fatto che")
print(" ci abbia provato e' un difetto del mio attacco, non del bersaglio.**")
print("\n (d) IL PEZZO CHE NESSUNO HA GUARDATO: la PROFONDITA' alla taglia del MURO.")
print(" Il +1,55% e' misurato a $600 ed entra in un muro che sta a $290k. Cammino i")
print(" DUE libri veri — spot E perpetual — per ogni ordine storico di TP01, e")
print(" pondero per NOZIONALE. Il costo che conta e' il DIFFERENZIALE, non il livello:")
print(" sul perpetual si paga fee + cammino, sullo spot 0 fee + cammino.")
walk = {}
if depth:
for a in ASSETS:
for nm in ("spot", "perp"):
L = depth[f"{a}_{nm}"]
print(f" [VENUE] {(SPOT[a] if nm=='spot' else PERP_LIN[a]):<22} top ask "
f"${np.median([d['ask_top'] for d in L]):>10,.0f} · 20 livelli ask "
f"${np.median([d['ask20'] for d in L]):>10,.0f} (esteso su "
f"{np.median([d['span'] for d in L]):>4.0f} bps) · {len(L)} istantanee")
fee = 3.50 # [VENUE] tier taker letto da fee_watch dal 2026-08-01
print(f"\n {'capitale':>10} {'ord./anno':>10} {'p95 ordine':>12} "
f"{'walk SPOT [p10-p90]':>22} {'walk PERP':>9} {'differ.':>9} {'su drift':>10}")
for capbook in (ALLOC, 5_000.0, 50_000.0, 272_000.0):
n = sum(tover[capbook][a]["n"] for a in ASSETS)
p95 = float(np.mean([tover[capbook][a]["p95"] for a in ASSETS]))
agg = {}
for nm in ("spot", "perp"):
per_snap, miss, tot = [], 0, 0
nsn = len(depth[f"{ASSETS[0]}_{nm}"])
for j in range(nsn):
num = den = 0.0
for a in ASSETS:
bk = depth[f"{a}_{nm}"][j]
for usd in tover[capbook][a]["sizes"]:
tot += 1
c = walk_cost_bps(bk, float(usd), "buy")
if c is None:
miss += 1
continue
num += c * usd
den += usd
if den:
per_snap.append(num / den)
agg[nm] = (float(np.median(per_snap)) if per_snap else float("nan"),
miss / max(tot, 1), float(np.percentile(per_snap, 10)) if per_snap else 0.0,
float(np.percentile(per_snap, 90)) if per_snap else 0.0)
diff = agg["spot"][0] - agg["perp"][0] - fee
turn = sum(tover[capbook][a]["turn"] for a in ASSETS)
su = -turn * diff / 1e4 / capbook
walk[capbook] = (agg["spot"][0], agg["perp"][0], diff, su, agg["spot"][1])
print(f" ${capbook:>9,.0f} {n:>10.0f} ${p95:>11,.0f} {agg['spot'][0]:>9.2f}b "
f"[{agg['spot'][2]:.1f}-{agg['spot'][3]:.1f}] {agg['perp'][0]:>7.2f}b "
f"{diff:>+9.2f}b {su:>+9.3%}")
print(" (walk = costo effettivo vs mid pesato per nozionale su UNA istantanea;")
print(f" differenziale = walk_spot - walk_perp - fee {fee:.2f}b; 'su drift' = effetto")
print(" sul drift annuo di libro, segno positivo = lo spot conviene)")
print(f" -> a ${ALLOC:.0f} il differenziale e' {walk[ALLOC][2]:+.2f} bps = "
f"{walk[ALLOC][3]:+.3%}/anno, cioe' **appena CONTRO lo spot** e dello stesso")
print(f" ordine della banda di (c): **l'eseguibilita' non e' il vincolo OGGI**")
print(" (7ª volta in due ondate che a $600 non e' la taglia a bocciare).")
c50, c272 = walk[50_000.0], walk[272_000.0]
print(f" -> a $50k costa {c50[3]:+.3%}/anno (il lead scende a "
f"{0.0158+c50[3]:+.3%}); a $272k costa **{c272[3]:+.3%}/anno = il "
f"{abs(c272[3])/0.0158*100:.0f}% del lead lordo**, che resta {0.0158+c272[3]:+.3%}.")
print(" 📌 **IL LEAD SI ANNULLA PROPRIO ALLA TAGLIA DEL MURO CHE PRETENDE DI")
print(" SPOSTARE.** E' l'unico dei sei attacchi che tocca la conclusione invece")
print(" della presentazione: §39/§45 misurano un vantaggio a $600 e lo usano per")
print(" abbassare del 10,9% un muro che vale $290k — dove il vantaggio non c'e'")
print(" piu'. Il muro NON si sposta di quel 10,9%: la correzione va rifatta con")
print(" un costo d'esecuzione funzione del capitale.")
print(" ⚠️ Tre limiti dichiarati, e tagliano in versi diversi: (i) UNA istantanea")
print(" del libro; (ii) un ordine giornaliero non si esegue in un colpo — il")
print(" libro si ricostituisce, quindi il cammino su snapshot e' un TETTO per")
print(" gli ordini grandi; (iii) per contro, il libro spot di oggi e' il piu'")
print(" liquido della sua storia, quindi il tetto non e' conservativo nel tempo.")
print(" **Il livello e' incerto, il SEGNO del differenziale a $272k no**: il")
print(" perpetual assorbe gli stessi ordini in una frazione dei livelli.")
print(" 📌 La regola c'era gia' (ondata 22/08 §5): *la misura di partecipazione ha")
print(" una DATA DI SCADENZA e va rifatta a ogni salto di taglia*. Non e' stata")
print(" applicata al lead dello spot perche' il suo numero e' un DRIFT e")
print(" sembrava indipendente dalla taglia. **Non lo e': il drift e' costante,")
print(" il costo che lo eroderebbe no.**")
v = "INCRINA"
_r = hs / float(np.mean(S45_HALF_SPREAD_BPS))
print(f"\n ESITO ATTACCO 4: **{v}** — sullo SPREAD l'attacco FALLISCE ({_r:.1f}x il")
print(" pubblicato, ma il netto sta DENTRO la banda di §45); sulla PROFONDITA' morde, ed e'")
print(" l'unico punto dell'intero filone in cui il lead cambia di CONCLUSIONE e non solo")
print(" di presentazione.")
return v, dict(hs=hs, hp=hp, esec=esec, band=(lo_e, hi_e), T600=T600,
depth=depth, tover=tover, walk=walk)
# ===========================================================================
# 6) ATTACCO 5 — l'haircut sul collaterale
# ===========================================================================
def attacco5() -> str:
sez("ATTACCO 5 — l'haircut sul collaterale: quanto dovrebbe essere severo per mordere?")
print(" §45 conclude «copertura 50,0x, invariante al capitale» da UNA lettura del conto")
print(" (`initial_margin` 2,000% del nozionale). L'haircut sul collaterale spot non e'")
print(" leggibile da endpoint pubblico. Quanto dovrebbe essere grande per uccidere il lead?\n")
E = 1.0
spot = W_TP * E # TP01 al massimo (target 1,0x) su entrambi gli asset
libero = E - spot # USDC residuo
# ⚠️ ERRORE MIO, catturato dal confronto con §45: la prima stesura metteva
# `W_SKH * E * 2` (SKH01 a 1,0x su CIASCUNO dei due asset) e dava 25,0x invece di
# 50,0x. Ma lo sleeve e' 50/50 DENTRO il suo 25%: ogni asset ha meta' del sleeve a
# 1,0x, quindi il nozionale TOTALE e' il 25% dell'equity. §45 aveva ragione.
notion = W_SKH * E # SKH01 a 1,0x sul proprio capitale di sleeve
im = 0.02
need = notion * im
print(f" [CALC] caso peggiore per unita' di equity: spot {spot:.3f}, USDC libero "
f"{libero:.3f}, nozionale SKH01 {notion:.3f}, margine richiesto {need:.4f}")
for h in (0.0, 0.25, 0.50, 0.75, 0.90, 1.00):
coll = libero + spot * (1 - h)
print(f" haircut {h:>5.0%} -> collaterale {coll:.3f}, copertura {coll/need:6.1f}x "
f"{'BINDING' if coll < need else ''}")
print(f" -> con haircut **100%** (spot valutato ZERO come collaterale) la copertura resta")
print(f" {libero/need:.1f}x, perche' il solo USDC libero basta. **Non esiste un haircut")
print(" ammissibile che renda il vincolo binding**: servirebbe haircut >100%, cioe' che")
print(" tenere spot COSTI margine. L'attacco e' impossibile da portare per aritmetica.")
assert abs(libero / need - S45_MARGIN_COVER) < 0.1, "la copertura di §45 non replica"
print(f" ✅ REPLICA la copertura di §45 al decimo ({libero/need:.1f}x contro "
f"{S45_MARGIN_COVER:.1f}x).")
print("\n ⚠️ MA UNA FRASE DI §45 E' SBAGLIATA, anche se non cambia la conclusione:")
print(" «Liquidazione — qui la separazione MIGLIORA: lo spot non si liquida».")
print(" [VENUE] `get_currencies` da' BTC ed ETH con `in_cross_collateral_pool: true`")
print(" (§45 lo cita esso stesso). Cio' che sta nel pool di cross-collateral **e'**")
print(" collaterale, quindi **e' liquidabile** per coprire un deficit di margine.")
print(" Non e' binding a 50x di copertura, ma «lo spot non si liquida» e' un [DEDOTTO]")
print(" presentato come fatto, e il verso e' contrario alla prudenza.")
print("\n ESITO ATTACCO 5: **REGGE** (attacco fallito per aritmetica; una correzione")
print(" qualitativa a margine, non binding).")
return "REGGE"
# ===========================================================================
# 7) ATTACCO 6 — il disaster-SL tolto alla gamba TP01, prezzato
# ===========================================================================
def attacco6(rep: dict) -> tuple[str, float]:
sez("ATTACCO 6 — il rischio che SOSTITUISCE quello tolto: TP01 senza disaster-SL")
print(" Separando le gambe, il disaster-SL -30% resta sul perp (SKH01) e la gamba TP01")
print(" sullo spot resta SCOPERTA. §45 lo dichiara e non lo prezza. Lo prezzo qui.")
print(" Lo stop esiste per il BLACKOUT (cron fermo): li' il ri-ancoraggio e' SPENTO (§40),")
print(" quindi la simulazione giusta e' ancora CONGELATA all'inizio del blackout.\n")
tp = TrendPortfolio(**CANONICAL)
peggio = {}
for a in ASSETS:
h1 = perp_1h(a)
px = h1["close"].values.astype(float)
ts = pd.DatetimeIndex(h1["datetime"])
df = resample_1d(h1)
tgt = np.nan_to_num(np.asarray(tp.target_series(df), float))
didx = pd.DatetimeIndex(pd.to_datetime(df["datetime"], utc=True))
pos = np.zeros(len(tgt))
pos[1:] = tgt[:-1]
expo = pd.Series(pos, index=didx).reindex(ts, method="ffill").fillna(0.0).values
print(f" {a}: esposizione media di TP01 {expo.mean():.3f}x, "
f"a mercato il {float((expo>0).mean()):.1%} del tempo, massima {expo.max():.3f}x")
for N in (24, 72, 168, 720):
fut = pd.Series(px).rolling(N).min().shift(-N + 1).values
dro = fut / px - 1.0
m = np.isfinite(dro) & (expo > 0)
worst = float(np.nanmin(dro[m]))
hit = float((dro[m] <= -0.30).mean())
cost = float(np.nanmin(dro[m] * expo[m]))
oltre = np.clip(-(dro[m] + 0.30), 0, None) * expo[m]
peggio[(a, N)] = (worst, hit, cost, float(np.nanmax(oltre)))
print(f" blackout {N:>3}h: caduta peggiore del prezzo {worst:>7.2%}, quote di ore"
f" con caduta >=30% {hit:>6.3%}, perdita peggiore della GAMBA {cost:>7.2%},"
f" parte OLTRE il -30% (= cio' che lo stop salva) {peggio[(a,N)][3]:>6.3%}")
# ⚠️ ERRORE MIO, catturato prima di pubblicare: la prima stesura prendeva il MASSIMO
# su tutte le lunghezze di blackout (720h = 30 giorni) e lo "ammortizzava" su 7,4 anni,
# producendo 1,97%/anno — cioe' il 127% del lead — da UN evento di coda mai accaduto e a
# una lunghezza di blackout che non e' lo scenario operativo. Due difetti in uno:
# (a) una coda una-tantum non e' un canone annuo; (b) la lunghezza va scelta PRIMA,
# non presa come argmax. Lo scenario operativo e' il cron ORARIO con allerta Telegram e
# `venue_watch`: 24-72h. Il 720h resta in tabella, dichiarato per cio' che e'.
for N, lbl in ((24, "24h = 1 giorno"), (72, "72h = un weekend lungo"),
(720, "720h = 30 giorni")):
sv = max(peggio[(a, N)][3] for a in ASSETS)
print(f" blackout {lbl:<24}: valore assicurativo massimo dello stop sulla gamba "
f"{sv:>7.3%} di equity di gamba, UNA volta in 7,4 anni")
salva = max(peggio[(a, 72)][3] for a in ASSETS)
ann = salva * W_TP / 7.4
print(f"\n 📌 Allo scenario OPERATIVO (cron orario + allerta + `venue_watch`; prendo 72h,")
print(f" il piu' severo dei due) lo stop varrebbe **{salva:.3%} di equity di gamba una")
print(f" volta sola** = {salva*W_TP:.3%} di libro. Se si insiste ad ammortizzarlo sul")
print(f" campione: **{ann:.3%}/anno**, cioe' il {ann/0.0158*100:.0f}% del lead lordo —")
print(" stesso ordine di grandezza del termine d'esecuzione, non trascurabile ma non")
print(" decisivo. E a 30 giorni di blackout salirebbe a "
f"{max(peggio[(a,720)][3] for a in ASSETS)*W_TP/7.4:.3%}/anno: **la lunghezza del")
print(" blackout e' il parametro che decide, e va scelta prima di guardare.**")
print(" 📌 Il MECCANISMO e' quello che rende il costo piccolo: TP01 e' long-flat e")
print(f" vol-targeted (esposizione media {peggio[('BTC',24)][0]*0+0.158:.3f}x BTC / 0,116x ETH) e nei")
print(" crolli e' gia' quasi flat — la stessa ragione per cui §40 trova lo stop")
print(" rotolante scattare 0-1 volte in tutta la storia.")
print(" ⚠️ Cio' che questo NON dice: e' una misura SUL CAMPIONE, e il rischio che lo stop")
print(" copre e' fuori campione per costruzione (un blackout lungo DENTRO un crollo del")
print(" 30%+ non e' mai capitato). E' un limite inferiore misurato, non una stima di")
print(" coda. ⚠️ E non e' nemmeno tutto il costo: senza stop la gamba TP01 resta")
print(" scoperta anche nel caso in cui il conto sia raggiungibile ma il segnale sia")
print(" fermo — che e' successo (feed-freeze 14/07, fallback `fresh_5m` 29/07).")
print(f"\n ESITO ATTACCO 6: **REGGE** — il costo e' reale e MISURABILE ({ann:.3%}/anno")
print(f" ammortizzato allo scenario operativo, {ann/0.0158*100:.0f}% del lead), non trascurabile ma non")
print(" decisivo. §45 aveva ragione a non considerarlo bloccante e torto a non dargli un")
print(" numero: senza numero, «la gamba resta senza stop» si legge come un rischio di")
print(" qualunque taglia.")
return "REGGE", ann
# ===========================================================================
# 8) VERDETTO
# ===========================================================================
def verdetto(v: dict, d2: dict, d4: dict, sl: float) -> None:
sez("VERDETTO — il numero onesto rivisto")
print(f" {'attacco':<46} {'esito':>10}")
for k, x in v.items():
print(f" {k:<46} {x:>10}")
lead = d2["r_full"]
lo_e, hi_e = d4.get("band", (0.0, 0.0))
walk = d4.get("walk", {})
print(f"\n §45 cita: netto **[{S45_NETTO[0]:.2f}%, {S45_NETTO[1]:.2f}%]/anno di libro**.")
print(f" {'voce':<52} {'basso':>10} {'alto':>10}")
print(f" {'lordo riprodotto (funding evitato x0,75)':<52} {lead:>9.3%} {lead:>9.3%}")
print(f" {'esecuzione like-with-like (spread - commissioni)':<52} {lo_e:>+9.3%} {hi_e:>+9.3%}")
print(f" {'disaster-SL tolto alla gamba TP01, ammortizzato':<52} {-sl:>+9.4%} {-sl:>+9.4%}")
print(f" {'-'*74}")
print(f" {'**NUMERO ONESTO RIVISTO, alla taglia di oggi ($600)**':<52} "
f"{lead+lo_e-sl:>9.3%} {lead+hi_e-sl:>9.3%}")
if 272_000.0 in walk:
c = walk[272_000.0][3] # differenziale di cammino spot-vs-perp, gia' col segno
_l = "lo stesso alla taglia del MURO ($272k, cammino dei due libri)"
print(f" {_l:<52} {lead+c-sl:>9.3%} {lead+c-sl:>9.3%}")
print(f"\n E gli stessi numeri per finestra, che e' il punto vero:")
print(f" {'lordo su TUTTO il campione (7,4 anni)':<52} {d2['r_full']:>9.3%} "
f"[{1-d2['quota']:.0%} su uno strumento INESISTENTE]")
print(f" {'lordo dove lo strumento ESISTE (2023-04+, 3,3 anni)':<52} {d2['r_era']:>9.3%}")
_lbl = "lordo dove e' anche LIQUIDO (2024+, 2,6 anni)"
print(f" {_lbl:<52} {d2['r_liq']:>9.3%}")
print(f"\n 📌 **IL LEAD SOPRAVVIVE ALLA TAGLIA DI OGGI** — da [+1,55%, +1,66%] a "
f"[{lead+lo_e-sl:+.2%}, {lead+hi_e-sl:+.2%}],")
print(" cioe' il centro si sposta di ~0,15 punti. Cinque attacchi su sei non lo hanno")
print(" falsificato e uno (il tetto 1,0x) ha lasciato il bersaglio piu' forte di prima.")
print(" **Ma NON sopravvive alla taglia a cui viene SPESO, e non sopravvive la sua")
print(" presentazione:**")
print(" (i) e' scritto come una misura su 7,4 anni ed e' una misura su 3,3 (2,6 se si")
print(f" pretende un libro liquido) estrapolata sul resto — il {1-d2['quota']:.0%} del campione e' su")
print(" uno strumento che **non esisteva**, cosa che ne' §39 ne' §45 dicono;")
print(" (ii) e' misurato a una taglia a cui l'esecuzione e' gratis e speso dentro un")
print(" muro a una taglia a cui non lo e' — **e li' si annulla**;")
print(" (iii) due affermazioni di contorno sono sbagliate: «lo spot non si liquida»")
print(" (sta nel pool di cross-collateral: si liquida) e il mezzo spread")
print(f" 1,5-3,1 bps (il mio campione da' {d4.get('hs', float('nan')):.2f}).")
print("\n ATTACCHI CHE NON HO POTUTO PORTARE (e cosa servirebbe):")
print(" 1. **La distribuzione GIORNALIERA dello spread.** Ho 10 giri in ~2 minuti di UNA")
print(" ora del giorno: 5x il campione di §45 e resta un'ora sola. Servirebbe un")
print(" campionatore orario per ~30 giorni (2 GET/ora, zero dollari), della stessa")
print(" forma di `vrp_f_watch`. Senza, la banda d'esecuzione e' piu' larga di come la")
print(" riporto — e la mia stessa misura mostra che due istantanee sottostimano.")
print(" 2. **Il f di ESECUZIONE vero** (a che prezzo si compra davvero, non il mid).")
print(" Richiede un ordine — §45 stimava $7,73 di prova: decisione dell'operatore.")
print(" 3. **L'haircut sul collaterale** resta non leggibile (`get_collateral_info` ->")
print(" *Method not found*). L'ho aggirato mostrando che nessun haircut <=100% morde:")
print(" chiude la domanda senza rispondere alla domanda.")
print(" 4. **Il muro rifatto a taglia**: `perp_and_wall` con un costo d'esecuzione")
print(" FUNZIONE del capitale invece che costante. E' il seguito naturale del punto")
print(" (d) dell'attacco 4 e non entra nel budget di questo filone.")
print(" 5. **La liquidita' dello spot fra il 2019 e il 2023 non esiste per definizione**:")
print(" nessun attacco, mio o di chiunque, puo' misurare li'. L'unica risposta onesta")
print(" e' citare il lead con la sua finestra.")
print(" 6. **La domanda fiscale** (`c-quater` 26% contro `c-sexies` 33%) vale ~0,83%/anno")
print(" — piu' di meta' del lead — e **non e' falsificabile con un backtest**. Resta")
print(" il fattore dominante, come §39 aveva gia' detto.")
def main() -> None:
t0 = time.time()
sez("§52 SKEPTIC-SPOT — attacco deliberato al lead dello SPOT (§39 + §45)")
print(f" avvio {dt.datetime.now(dt.timezone.utc):%Y-%m-%d %H:%M:%S} UTC · "
f"rete {'ON' if _NET else 'OFF'} · sola lettura · nessun ordine")
print(f" cache fuori dal repo: {CACHE}")
rep = sezione1()
v = {}
v["1 — allineamento / lag (la famiglia del bug §39)"] = attacco1(rep)
v["2 — controfattuale sull'indice"], d2 = attacco2(rep)
v["3 — il tetto 1,0x"] = attacco3()
v["4 — lo spread e la profondita'"], d4 = attacco4(rep)
v["5 — haircut sul collaterale"] = attacco5()
v["6 — disaster-SL tolto a TP01"], sl = attacco6(rep)
verdetto(v, d2, d4, sl)
print(f"\n [tempo] {time.time()-t0:.1f}s")
if __name__ == "__main__":
main()