Adriano/PythagorasGoal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
PythagorasGoal/scripts/research/r0823b_usdc_surface.py
T
Adriano Dal Pastro 8c0b5f97ee registro: chiusi §58-69 — dodici verdetti, 0 candidati, e due appoggi tolti a decisioni gia' prese 
I dodici filoni della sesta ondata erano rimasti `_in corso_`: la sessione si e'
chiusa dopo che gli agenti avevano scritto gli script e prima che i verdetti
fossero consolidati, e i loro messaggi finali sono persi. I numeri qui NON
vengono da quei messaggi: vengono dalla riesecuzione dei dodici script
(07:38-08:06 UTC, sequenziale, log in logs/r0823b/ che e' gitignored).
E' il motivo per cui l'ondata non e' andata persa: ogni script calcola il
proprio verdetto a runtime.

Cosa tocca decisioni gia' prese:
- §67 il gate pre-registrato XSR01 del 23/10 legge un monitor tarato sul
  pavimento del venue sbagliato (C* $15-20k, non ~$3k)
- §64 la raccomandazione di §51 (raccogliere la catena USDC) non e'
  giustificata dalla ragione che porta: le due superfici sono la stessa
- §60 la politica MISTO scelta il 25/07 non e' piu' l'ottimo (oggi MISTO-A)
- §63 domanda fiscale NUOVA, diversa da quella aperta il 07/08

Il risultato piu' grande e' di §58: l'obiettivo del progetto ha DUE definizioni
operative in uso che danno 33,9% contro 0,33% sulla stessa domanda, e non era
mai stato detto quale si stesse ottimizzando.

r0823b_quasi_passati.py (§68) terminava con IndexError: Griglia.combo()
indicizzava con self.idx (2720 giorni) un sottoinsieme di 958 righe. Corretto
col parametro idx esplicito piu' un controllo di lunghezza; la
rinormalizzazione e' riga per riga, quindi i valori sono quelli dell'intento
dell'autore. La correzione e' del coordinatore, non dell'autore, ed e'
dichiarata nel registro.

Libro, pesi, cron, config INVARIATI. Nessun ordine.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-08-23 12:40:30 +00:00

1247 lines
68 KiB
Python
Raw Permalink Blame History

"""USDC-SURFACE — la superficie opzioni USDC-LINEARE, misurata contro la sua gemella INVERSE
appaiata a STESSA SCADENZA E STESSO STRIKE (2026-08-23, filone §64).
IL BUCO. Tutta la catena opzioni che il progetto possiede e' della famiglia INVERSE
(`collect_chain.py` interroga `{"currency":"BTC"|"ETH"}`). Il conto e' in USDC e NON puo'
marginare l'inverse: se un giorno il progetto tradasse opzioni sarebbero le LINEARI. Ogni
conclusione del progetto sulla famiglia lineare e' oggi un controfattuale costruito sui mid
inverse — §28 ha gia' pagato quell'errore in un'altra veste.
COSA C'E' GIA', e NON viene rifatto:
§14 ALT-OPTIONS (r0822_alt_options.py) ha censito SEI famiglie lineari e la struttura SHORT
(put credit spread -0.28/-0.10) su strike AUTO-SELEZIONATI per famiglia.
§46 TAIL-HEDGE ha misurato f sulle quote vere USDC per il lato LONG e ha refutato la
copertura statica (maxDD sale in 162 celle su 162).
§51 NEWDATA-SCOUT raccomanda di RACCOGLIERE la catena USDC (+117 chiamate/giro = +18%).
QUI si fa l'unica cosa che nessuno dei tre ha fatto: **il confronto APPAIATO** fra le due
famiglie sullo STESSO strumento economico (stessa scadenza, stesso strike, stesso tipo) — che
e' l'unico confronto informativo, perche' un confronto fra strike auto-selezionati misura la
GRIGLIA degli strike quotati e non la famiglia.
ATTESE DICHIARATE PRIMA DI MISURARE (derivate dai numeri gia' pubblicati e dall'aritmetica dei
tick, NON dai dati di oggi; ognuna e' stampata a video con l'esito):
A1 REPLICA §51 — coi filtri del collettore vivo (<=95 giorni, OI>=100) la catena USDC BTC+ETH
e' 113 strumenti contro 649 inverse (+17,4%). Atteso: replica entro +-15% (e' un ricomputo
a un giorno di distanza e l'OI si muove).
A2 REPLICA §14 — la correlazione di rango fra "OI>=100" e "% put quotati a due lati" sulle sei
famiglie lineari e' -0,77. Atteso: stesso segno e stesso ordine di grandezza.
A3 FATTO (b), TICK — `tick_size = 5.0` su tutte le put BTC_USDC: atteso CONFERMATO.
"5,00 USDC su tutte le USDC": atteso REFUTATO (§14 pubblica ETH_USDC 0,2 / SOL 0,1).
E soprattutto l'INFERENZA implicita — che il pavimento del tick sia un muro della famiglia
LINEARE — atteso REFUTATO: 1 tick inverse vale 0,0001 BTC = ~$7,7 contro i $5,00 del
lineare, quindi il pavimento e' PEGGIORE sulla famiglia che il progetto gia' usa.
(Aritmetica sui tick pubblicati, non ancora una misura.)
A4 SPREAD APPAIATO — mi aspetto la lineare PIU' LARGA dell'inverse a pari strike (il flusso
sta sull'inverse). ⚠️ Contro-indizio gia' pubblicato: §14 da' f_venue ETH_USDC 0,980 contro
ETH inverse 0,930 su strike auto-selezionati. Attesa dichiarata INCERTA, 50/50.
A5 PREZZO APPAIATO — |ΔIV| appaiata piccola rispetto al mezzo-spread: stesso sottostante,
stessi market maker, arbitraggio diretto fra le due famiglie. Atteso CONFERMATO.
E' l'attesa che DECIDE la raccomandazione §51: se le due superfici prezzano uguale, la
catena inverse e' gia' un proxy sufficiente e raccogliere la lineare non ripara nessun
numero.
A6 LATO LONG — nessun muro di CAPITALE in nessuna delle due famiglie (comprare costa il
premio, §46). Il muro e' il tick, ed e' peggio sull'inverse (A3). Atteso CONFERMATO.
A7 COSA ABILITA — l'unica cosa che la lineare abilita e l'inverse no e' la struttura SHORT a
rischio definito con max-loss piccolo (§14: ETH_USDC $16,62 contro ETH inverse $145,88),
cioe' esattamente cio' che la regola "niente short-vol da modello in deploy" vieta.
Atteso CONFERMATO.
CONTROLLI POSITIVI (un rilevatore che non trova nulla e' indistinguibile da uno rotto):
C1 potenza del confronto appaiato: ri-appaio ogni lineare con l'inverse allo strike ADIACENTE
invece che allo stesso. Se |ΔIV| non cresce, il confronto non ha risoluzione e ogni "le due
superfici coincidono" e' vuoto.
C2 il rilevatore di pavimento: tick/mark dev'essere trascurabile ATM e grande sulle ali. Se
desse lo stesso numero ovunque sarebbe rotto.
C3 il rilevatore di larghezza: SOL_USDC (9% di put quotati, §14) dev'essere molto peggio di
BTC_USDC (93%). Se non li separa, non separa niente.
C4 tipicita' del giorno: e' DOMENICA. Non posso aspettare, ma punto la stessa grandezza che il
progetto ha misurato su 19 settimane FERIALI: attraversare lo spread sul canonico inverse
costa ~10% del credito (30/07). §14 ha letto 7,0-7,3% di SABATO.
COSA QUESTO FILE NON PUO' DIRE, prima di qualunque numero:
* Non esiste STORIA su questa famiglia: zero giorni di catena raccolta -> nessun backtest,
nessun hold-out, nessun deflated-Sharpe. Qui si misura la NEGOZIABILITA', che e' una
proprieta' strutturale, non un edge. **Nessuna conclusione sul VALORE di una strategia.**
* Non e' un'istantanea sola ma neanche una distribuzione: N snapshot in una sessione danno la
dispersione INFRA-SESSIONE, non quella fra giorni della settimana.
* Il margine non e' leggibile dall'API pubblica (serve `private/get_margins`, chiave -> fuori
perimetro). Dove serve, si usa la formula PUBBLICATA (fonte secondaria, dichiarata).
RETE: sole GET pubbliche, nessun ordine, mai. Pacing <=2 req/s. Il file SI RIFIUTA di partire
nelle finestre :05-:10 (`cron_book`) e :24-:30 (`cron_chain`): il rate limit Deribit e' per-IP e
il progetto ha gia' avuto un guasto per questo (29/07).
nice -n 19 timeout 900 uv run python scripts/research/r0823b_usdc_surface.py --snap
nice -n 19 timeout 900 uv run python scripts/research/r0823b_usdc_surface.py --report
nice -n 19 timeout 900 uv run python scripts/research/r0823b_usdc_surface.py # snap+report
"""
from __future__ import annotations
import argparse
import json
import math
import time
from datetime import UTC, datetime
from pathlib import Path
import numpy as np
import requests
API = "https://www.deribit.com/api/v2/public"
TIMEOUT = 20
RPS = 2.0
SNAPDIR = Path("/tmp/claude-1001/-opt-docker-PythagorasGoal/"
"b6cc75e7-14f8-4c32-bd07-ab8a0d2aaee6/scratchpad/usdcsurf")
# --- parametri, tutti dichiarati qui e nessuno scelto guardando l'esito ------------------------
CAPITAL = 635.0 # il conto vero
RISK_BUDGET = 0.20 # quota di conto a rischio, come §14
OI_MIN = 100.0 # la soglia di collect_chain.py
DTE_MAX_COLL = 95.0 # la finestra di collect_chain.py
DTE_LO, DTE_HI = 4.0, 45.0 # la finestra di §14 (dal settimanale VRP01 a ~1.5 mesi)
TAKER = 0.0003 # frazione del sottostante
TAKER_CAP = 0.125 # cap: 12.5% del premio
SHORT_D, LONG_D = -0.28, -0.10 # la geometria di VRP01, non ri-scelta qui
TAIL_D = -0.05 # l'ala di §46 (lato long)
PAIRS = (("BTC_USDC", "BTC"), ("ETH_USDC", "ETH"))
LINEAR6 = ("ETH_USDC", "BTC_USDC", "SOL_USDC", "XRP_USDC", "HYPE_USDC", "AVAX_USDC")
DEPTH_DELTAS = (-0.05, -0.10, -0.28, -0.50)
# ------------------------------------------------------------------ rete (parca, sola lettura)
class Budget:
def __init__(self, rps: float = RPS) -> None:
self.rps, self._next, self.calls, self.err, self.r429 = rps, 0.0, 0, 0, 0
def wait(self) -> None:
now = time.monotonic()
if now < self._next:
time.sleep(self._next - now)
self._next = max(now, self._next) + 1.0 / self.rps
def guardia_finestra() -> None:
m = datetime.now(UTC).minute
if 5 <= m <= 10:
raise SystemExit(f"minuto {m}: finestra di cron_book (:07). Rilancia dopo il :10.")
if 24 <= m <= 30:
raise SystemExit(f"minuto {m}: finestra di cron_chain (:25). Rilancia dopo il :30.")
def get(path: str, params: dict, b: Budget, tries: int = 3):
for k in range(tries):
b.wait()
b.calls += 1
try:
r = requests.get(f"{API}/{path}", params=params, timeout=TIMEOUT)
except Exception:
b.err += 1
time.sleep(1.5 * (k + 1))
continue
if r.status_code == 429:
b.r429 += 1
time.sleep(2.0 * (k + 1))
continue
if r.status_code != 200:
b.err += 1
time.sleep(1.0 * (k + 1))
continue
try:
return r.json()["result"]
except Exception:
b.err += 1
return None
return None
# ------------------------------------------------------------------ Black-76 (r=0, come il sleeve)
def _ncdf(x: float) -> float:
return 0.5 * (1.0 + math.erf(x / math.sqrt(2.0)))
def put_delta(F: float, K: float, T: float, sig: float) -> float:
if not (F > 0 and K > 0 and T > 0 and sig > 0):
return float("nan")
d1 = (math.log(F / K) + 0.5 * sig * sig * T) / (sig * math.sqrt(T))
return _ncdf(d1) - 1.0
def put_price(F: float, K: float, T: float, sig: float) -> float:
"""Prezzo di una put, in USD per UNITA' di sottostante (r=0, F = forward)."""
if not (F > 0 and K > 0 and T > 0 and sig > 0):
return float("nan")
d1 = (math.log(F / K) + 0.5 * sig * sig * T) / (sig * math.sqrt(T))
d2 = d1 - sig * math.sqrt(T)
return K * _ncdf(-d2) - F * _ncdf(-d1)
def eff_tick(price: float, base: float, steps: list) -> float:
"""Tick EFFETTIVO al prezzo dato: Deribit ha `tick_size_steps` e ignorarlo sbaglia il
pavimento proprio dove NON morde (opzioni care) e lo lascia giusto dove morde (ali)."""
t = base
for s in sorted(steps or [], key=lambda x: float(x["above_price"])):
if price > float(s["above_price"]):
t = float(s["tick_size"])
return t
# ------------------------------------------------------------------ snapshot
def fetch_snapshot(b: Budget) -> dict:
"""6 chiamate per l'intera superficie + order_book su un insieme DETERMINISTICO di gambe.
L'insieme e' deterministico apposta: se cambiasse fra snapshot, la dispersione misurerebbe
la mia scelta invece del mercato.
"""
raw = {"ts": datetime.now(UTC).isoformat()}
for cur in ("USDC", "BTC", "ETH"):
raw[f"inst_{cur}"] = get("get_instruments",
{"currency": cur, "kind": "option", "expired": "false"}, b) or []
raw[f"summ_{cur}"] = get("get_book_summary_by_currency",
{"currency": cur, "kind": "option"}, b) or []
rows = build(raw)
names = depth_targets(rows)
raw["depth"] = {}
for n in names:
ob = get("get_order_book", {"instrument_name": n, "depth": 5}, b)
if ob:
raw["depth"][n] = {"bids": ob.get("bids") or [], "asks": ob.get("asks") or [],
"greeks": ob.get("greeks") or {}, "mark": ob.get("mark_price")}
return raw
def depth_targets(rows: list[dict]) -> list[str]:
"""Le stesse gambe in ENTRAMBE le famiglie: le put ai delta DEPTH_DELTAS, nelle due scadenze
piu' vicine dentro [DTE_LO, DTE_HI], scelte SOLO fra gli strike che esistono in entrambe."""
out: list[str] = []
for lin, inv in PAIRS:
both = [r for r in rows if r["fam"] == lin and r["typ"] == "P" and r["twin"]
and DTE_LO <= r["dte"] <= DTE_HI and np.isfinite(r["delta"])]
exps = sorted({round(r["dte"], 4) for r in both})[:2]
for e in exps:
g = [r for r in both if round(r["dte"], 4) == e]
for d in DEPTH_DELTAS:
if not g:
continue
r = min(g, key=lambda x: abs(x["delta"] - d))
out += [r["name"], r["twin"]]
return sorted(set(out))
def twin_name(name: str) -> str:
a = name.split("-")
a[0] = a[0].replace("_USDC", "")
return "-".join(a)
def build(raw: dict) -> list[dict]:
inst, summ = {}, {}
for cur in ("USDC", "BTC", "ETH"):
for i in raw.get(f"inst_{cur}") or []:
inst[i["instrument_name"]] = i
for s in raw.get(f"summ_{cur}") or []:
summ[s["instrument_name"]] = s
now_ms = datetime.fromisoformat(raw["ts"]).timestamp() * 1000.0 if raw.get("ts") \
else datetime.now(UTC).timestamp() * 1000.0
rows = []
for name, i in inst.items():
s = summ.get(name)
if s is None:
continue
fam = name.split("-")[0]
inv = "_" not in fam # BTC / ETH = inverse; *_USDC = lineare
dte = (float(i.get("expiration_timestamp") or 0) - now_ms) / 86400_000.0
bid, ask = s.get("bid_price"), s.get("ask_price")
bid = float(bid) if bid else None # Deribit manda 0/None per "nessun lato"
ask = float(ask) if ask else None
F = float(s.get("underlying_price") or 0.0) # e' il FORWARD della scadenza
iv = float(s.get("mark_iv") or 0.0)
K = float(i.get("strike") or 0.0)
T = max(dte, 0.0) / 365.25
typ = "P" if i.get("option_type") == "put" else "C"
mark = float(s.get("mark_price") or 0.0)
tw = twin_name(name) if not inv else None
rows.append(dict(
name=name, fam=fam, inv=inv, typ=typ, K=K, dte=dte, F=F, iv=iv, T=T,
delta=(put_delta(F, K, T, iv / 100.0) if (F and iv and K and T > 0 and typ == "P")
else float("nan")),
bid=bid, ask=ask, mark=mark,
oi=float(s.get("open_interest") or 0.0), vol=float(s.get("volume") or 0.0),
vol_usd=float(s.get("volume_usd") or 0.0),
tick0=float(i.get("tick_size") or 0.0), steps=i.get("tick_size_steps") or [],
minamt=float(i.get("min_trade_amount") or 0.0),
blockmin=float(i.get("block_trade_min_trade_amount") or 0.0),
twin=tw,
))
return rows
# ------------------------------------------------------------------ unita' (USD per unita' di sott.)
def u(price: float | None, r: dict) -> float:
"""Prezzo in USD per UNITA' di sottostante. Lineare: gia' in USDC/unita'. Inverse: in cripto
per contratto (contract_size 1 = 1 unita') -> x forward."""
if price is None:
return float("nan")
return price * (r["F"] if r["inv"] else 1.0)
def lot_usd(price: float | None, r: dict) -> float:
"""Prezzo di UN LOTTO MINIMO in dollari."""
return u(price, r) * r["minamt"]
def tick_usd(r: dict, price_native: float) -> float:
return eff_tick(price_native, r["tick0"], r["steps"]) * (r["F"] if r["inv"] else 1.0)
def relspread(r: dict) -> float:
if r["bid"] is None or r["ask"] is None:
return float("nan")
mid = 0.5 * (r["bid"] + r["ask"])
return (r["ask"] - r["bid"]) / mid if mid > 0 else float("nan")
def fee_lot(r: dict, price_native: float) -> float:
return min(TAKER * r["F"] * r["minamt"], TAKER_CAP * max(lot_usd(price_native, r), 0.0))
def med(v) -> float:
v = [x for x in v if x is not None and np.isfinite(x)]
return float(np.median(v)) if v else float("nan")
# ================================================================== §0 repliche
def sez0_repliche(rows: list[dict], att: dict) -> None:
print("\n" + "=" * 104)
print("0. REPLICA DEI TRE FATTI GIA' PUBBLICATI — prima di produrre un numero nuovo")
print("=" * 104)
# --- A1: il conteggio con i filtri del collettore vivo
def conta(fams, oi_min):
return sum(1 for r in rows if r["fam"] in fams and r["dte"] <= DTE_MAX_COLL
and r["dte"] > 0 and r["oi"] >= oi_min)
usdc_n = conta(("BTC_USDC", "ETH_USDC"), OI_MIN)
inv_n = conta(("BTC", "ETH"), OI_MIN)
usdc_raw = conta(("BTC_USDC", "ETH_USDC"), 0.0)
inv_raw = conta(("BTC", "ETH"), 0.0)
inc = usdc_n / inv_n * 100 if inv_n else float("nan")
print(f" A1 §51 filtri del collettore (<=95g, OI>={OI_MIN:.0f}):")
print(f" USDC BTC+ETH = {usdc_n:4d} (pubblicato 113) inverse = {inv_n:4d} "
f"(pubblicato 649) incremento {inc:+.1f}% (pubblicato +17,4%)")
print(f" senza il filtro OI: USDC {usdc_raw}, inverse {inv_raw} -> "
f"+{usdc_raw/max(inv_raw,1)*100:.0f}% "
f"(e' l'origine del '+90%' corretto in §51: il filtro taglia il "
f"{100*(1-usdc_n/max(usdc_raw,1)):.0f}% della famiglia USDC)")
ok1 = abs(usdc_n - 113) <= 0.15 * 113 and abs(inv_n - 649) <= 0.15 * 649
att["A1"] = ("CONFERMATA" if ok1 else "SCOSTAMENTO >15%")
# --- A2: la correlazione di rango OI vs quotati (le sei famiglie lineari)
stat = {}
for f in LINEAR6:
puts = [r for r in rows if r["fam"] == f and r["typ"] == "P"
and DTE_LO <= r["dte"] <= DTE_HI]
two = [r for r in puts if r["bid"] is not None and r["ask"] is not None]
oi = sum(1 for r in rows if r["fam"] == f and r["dte"] <= DTE_MAX_COLL and r["dte"] > 0
and r["oi"] >= OI_MIN)
stat[f] = dict(oi=oi, puts=len(puts), two=len(two),
pct=(len(two) / len(puts) if puts else float("nan")))
rA = {f: i + 1 for i, f in enumerate(sorted(stat, key=lambda x: -stat[x]["oi"]))}
rB = {f: i + 1 for i, f in enumerate(sorted(stat, key=lambda x: -stat[x]["pct"]))}
rho = float(np.corrcoef([rA[f] for f in LINEAR6], [rB[f] for f in LINEAR6])[0, 1])
print(f"\n A2 §14 rho di rango (OI>=100) vs (% put quotati a due lati), 6 famiglie lineari:")
print(f" {rho:+.2f} (pubblicato -0,77)")
print(f" {'famiglia':<10} {'OI>=100':>8} {'rango':>6} {'put 4-45g':>10} "
f"{'2 lati':>7} {'%':>6} {'rango':>6}")
for f in sorted(LINEAR6, key=lambda x: -stat[x]["oi"]):
s = stat[f]
print(f" {f:<10} {s['oi']:>8} {rA[f]:>6} {s['puts']:>10} {s['two']:>7} "
f"{s['pct']:>5.0%} {rB[f]:>6}")
att["A2"] = "CONFERMATA" if (rho < -0.4) else "REFUTATA"
# --- A3: il tick
print(f"\n A3 §46 il tick, e cosa vale davvero:")
print(f" {'famiglia':<10} {'n put':>6} {'tick base':>10} {'tick_size_steps':>34} "
f"{'tick base in $/unita':>21}")
tickfacts = {}
for f in ("BTC_USDC", "BTC", "ETH_USDC", "ETH"):
sub = [r for r in rows if r["fam"] == f and r["typ"] == "P"]
if not sub:
continue
ticks = {r["tick0"] for r in sub}
F = med([r["F"] for r in sub])
t0 = sorted(ticks)[0]
steps = sub[0]["steps"]
tusd = t0 * (F if sub[0]["inv"] else 1.0)
tickfacts[f] = tusd
st = "; ".join(f"{s['tick_size']:g} sopra {s['above_price']:g}" for s in steps) or "-"
uni = "USDC" if not sub[0]["inv"] else f.split("_")[0]
print(f" {f:<10} {len(sub):>6} {t0:>10g} {st:>34} {tusd:>18.3f} $"
f" [{uni}/unita']")
if f == "BTC_USDC":
print(f" {'':<10} tick unico su TUTTE le put BTC_USDC? "
f"{'SI (5.0)' if ticks == {5.0} else f'NO: {sorted(ticks)}'}")
print(f"\n 📌 il pavimento in DOLLARI, che e' il confronto che nessuno aveva fatto:")
for lin, inv in PAIRS:
if lin in tickfacts and inv in tickfacts:
r = tickfacts[inv] / tickfacts[lin]
print(f" {inv:<4} 1 tick = ${tickfacts[inv]:.3f}/unita' contro "
f"{lin:<9} ${tickfacts[lin]:.3f}/unita' -> l'INVERSE e' {r:.2f}x "
f"{'PIU GROSSO' if r > 1 else 'piu fine'}")
att["A3_tick5"] = "CONFERMATO su BTC_USDC"
att["A3_pavimento"] = ("l'inverse e' PEGGIO"
if all(tickfacts.get(i, 0) > tickfacts.get(l, 9e9) for l, i in PAIRS)
else "il lineare e' peggio")
# ================================================================== §1 censimento
def sez1_censimento(rows: list[dict]) -> None:
print("\n" + "=" * 104)
print("1. CENSIMENTO VIVO — le due famiglie di BTC/ETH, i fatti del venue")
print("=" * 104)
print(f"{'famiglia':<10} {'opz':>5} {'put':>5} {'put 4-45g':>10} {'2 lati':>7} {'%':>5} "
f"{'spr% med':>9} {'OI>=100':>8} {'min_amt':>8} {'$/lotto noz':>12} {'blocco $':>11}")
for lin, inv in PAIRS:
for f in (lin, inv):
sub = [r for r in rows if r["fam"] == f]
puts = [r for r in sub if r["typ"] == "P"]
win = [r for r in puts if DTE_LO <= r["dte"] <= DTE_HI]
two = [r for r in win if r["bid"] is not None and r["ask"] is not None]
F = med([r["F"] for r in sub])
a = sub[0]
oi = sum(1 for r in sub if r["dte"] <= DTE_MAX_COLL and r["dte"] > 0
and r["oi"] >= OI_MIN)
print(f"{f:<10} {len(sub):>5} {len(puts):>5} {len(win):>10} {len(two):>7} "
f"{(len(two)/len(win) if win else float('nan')):>4.0%} "
f"{med([relspread(r) for r in two])*100:>8.1f}% {oi:>8} {a['minamt']:>8g} "
f"{a['minamt']*F:>12,.0f} {a['blockmin']*F:>11,.0f}")
lin_all = [r for r in rows if r["fam"] in ("BTC_USDC", "ETH_USDC")]
inv_all = [r for r in rows if r["fam"] in ("BTC", "ETH")]
names_inv = {r["name"] for r in inv_all}
gemelli = sum(1 for r in lin_all if r["twin"] in names_inv)
print(f"\n 📌 SOVRAPPOSIZIONE DELLE GRIGLIE: {gemelli}/{len(lin_all)} strumenti lineari "
f"({gemelli/max(len(lin_all),1):.1%}) hanno un GEMELLO inverse a stessa scadenza e")
print(f" stesso strike; l'inverse ne ha {len(inv_all)-gemelli} che la lineare non ha "
f"({(len(inv_all)-gemelli)/max(len(inv_all),1):.0%} della sua griglia).")
print(" La famiglia lineare e' un SOTTOINSIEME della griglia inverse -> il confronto"
" appaiato e' possibile su tutto, e la lineare offre MENO strike, non altri.")
# ================================================================== §2 il confronto appaiato
def coppie(rows: list[dict]) -> list[tuple[dict, dict]]:
idx = {r["name"]: r for r in rows}
out = []
for r in rows:
if r["inv"] or r["fam"] not in ("BTC_USDC", "ETH_USDC"):
continue
t = idx.get(r["twin"] or "")
if t is not None:
out.append((r, t))
return out
def sez2_appaiato(rows: list[dict], att: dict) -> dict:
print("\n" + "=" * 104)
print("2. IL CONFRONTO APPAIATO — stessa scadenza, stesso strike, stesso tipo")
print("=" * 104)
print(" La statistica e' la MEDIANA DELLE DIFFERENZE APPAIATE (convenzione del progetto),")
print(" mai la differenza delle mediane: confrontare i mediani di due insiemi di strike")
print(" diversi misura la griglia, non la famiglia.")
cp = coppie(rows)
res = {}
print(f"\n{'coppia':<12} {'n coppie':>8} {'entrambe 2 lati':>16} {'ΔIV appaiata (pt vol)':>22} "
f"{'mezzo-spread IV*':>17} {'Δ spr% appaiata':>16}")
for lin, inv in PAIRS:
sub = [(a, b) for a, b in cp if a["fam"] == lin and DTE_LO <= a["dte"] <= DTE_HI
and a["typ"] == "P"]
both = [(a, b) for a, b in sub
if a["bid"] and a["ask"] and b["bid"] and b["ask"] and a["iv"] > 0 and b["iv"] > 0]
div = [a["iv"] - b["iv"] for a, b in both]
dsp = [relspread(a) - relspread(b) for a, b in both]
# mezzo-spread espresso in punti di vol: (mezzo spread in $) / vega, approssimato
# con la derivata numerica del prezzo BS rispetto a sigma su 1 punto di vol.
halfiv = []
for a, _ in both:
v = (put_price(a["F"], a["K"], a["T"], (a["iv"] + 1.0) / 100.0)
- put_price(a["F"], a["K"], a["T"], (a["iv"] - 1.0) / 100.0)) / 2.0
hs = 0.5 * (u(a["ask"], a) - u(a["bid"], a))
if v > 1e-9:
halfiv.append(hs / v)
res[lin] = dict(n=len(sub), both=len(both), div=med(div), dsp=med(dsp),
halfiv=med(halfiv),
pos_div=(np.mean([x > 0 for x in div]) if div else float("nan")),
pos_dsp=(np.mean([x > 0 for x in dsp]) if dsp else float("nan")))
print(f"{lin:<12} {len(sub):>8} {len(both):>16} {med(div):>+21.3f} "
f"{med(halfiv):>16.3f} {med(dsp)*100:>+15.2f}pp")
print("\n * `mezzo-spread IV` = quanti punti di vol vale mezzo bid-ask della gamba LINEARE.")
print(" E' il metro giusto per ΔIV: una differenza di prezzo piu' piccola del mezzo")
print(" spread non e' una differenza di prezzo, e' rumore di quotazione.")
for lin, _ in PAIRS:
r = res[lin]
ratio = abs(r["div"]) / r["halfiv"] if r["halfiv"] and np.isfinite(r["halfiv"]) else np.nan
print(f" {lin:<10} |ΔIV| / mezzo-spread = {ratio:.2f} "
f"(>1 = le due superfici prezzano DAVVERO diverso; <1 = dentro il rumore)")
print(f" {'':<10} ΔIV > 0 nel {r['pos_div']:.0%} delle coppie, "
f"Δspread > 0 nel {r['pos_dsp']:.0%}")
att["A4"] = ("CONFERMATA (lineare piu' larga)"
if all(res[l]["dsp"] > 0 for l, _ in PAIRS) else
"REFUTATA (lineare NON piu' larga)" if all(res[l]["dsp"] < 0 for l, _ in PAIRS)
else "MISTA fra i due sottostanti")
att["A5"] = ("CONFERMATA ma PER COSTRUZIONE: il test che avevo dichiarato (ΔIV al mark) "
"non ha potenza (§2b, una sola superficie di mark al venue); sostituito dalla "
"misura sulle QUOTE, che lo conferma")
# --- 2b: la IV al MARK e' la stessa PER CONVENZIONE DEL VENUE -> test (i) senza potenza
print("\n --- 2b LA IV AL MARK NON E' UNA MISURA: quanto spesso e' LO STESSO NUMERO?")
print(f" {'coppia':<12} {'coppie con mark_iv':>19} {'bit-identiche':>14} {'%':>6} "
f"{'max |ΔIV| non nulla':>21}")
tot_id, tot_n = 0, 0
for lin, _ in PAIRS:
sub = [(a, b) for a, b in cp if a["fam"] == lin and a["iv"] > 0 and b["iv"] > 0]
eq = [1 for a, b in sub if abs(a["iv"] - b["iv"]) < 1e-12]
nz = [abs(a["iv"] - b["iv"]) for a, b in sub if abs(a["iv"] - b["iv"]) >= 1e-12]
tot_id += len(eq); tot_n += len(sub)
print(f" {lin:<12} {len(sub):>19} {len(eq):>14} {len(eq)/max(len(sub),1):>5.1%} "
f"{(max(nz) if nz else 0.0):>21.3f}")
frac = tot_id / max(tot_n, 1)
print(f"\n 🚨 {frac:.1%} delle coppie ha la MEDESIMA mark_iv, e le differenze residue valgono")
print(" 0,01-0,02 punti di vol = l'arrotondamento a due decimali del campo. **Deribit marca")
print(" le due famiglie sulla STESSA superficie.** Conseguenza metodologica, e non e' un")
print(" dettaglio: il test (i) sul MARK, cosi' come l'avevo scritto in A5, **non ha")
print(" potenza** — non misura il mercato, misura la convenzione di marcatura del venue.")
print(" L'unica cosa DAVVERO indipendente fra le due famiglie sono le QUOTE (book separati).")
# --- 2c: le quote, che sono l'unica cosa indipendente
print("\n --- 2c LE QUOTE ESEGUIBILI, appaiate (in % del mark: unita' comune fra famiglie)")
print(f" {'coppia':<12} {'n coppie 2-lati':>16} {'Δbid':>9} {'Δask':>9} "
f"{'mezzo-spread lin':>17} {'mezzo-spread inv':>17} {'Δ appaiata':>11}")
for lin, _ in PAIRS:
both = [(a, b) for a, b in cp if a["fam"] == lin and a["typ"] == "P"
and DTE_LO <= a["dte"] <= DTE_HI and a["bid"] and a["ask"]
and b["bid"] and b["ask"] and a["mark"] > 0 and b["mark"] > 0]
db = [(u(a["bid"], a) - u(b["bid"], b)) / u(a["mark"], a) for a, b in both]
da = [(u(a["ask"], a) - u(b["ask"], b)) / u(a["mark"], a) for a, b in both]
hl = [(u(a["ask"], a) - u(a["bid"], a)) / (2 * u(a["mark"], a)) for a, b in both]
hi_ = [(u(b["ask"], b) - u(b["bid"], b)) / (2 * u(b["mark"], b)) for a, b in both]
dh = [x - y for x, y in zip(hl, hi_)]
res[lin]["dhalf"] = med(dh)
res[lin]["nboth"] = len(both)
res[lin]["hl"] = med(hl)
res[lin]["hi"] = med(hi_)
print(f" {lin:<12} {len(both):>16} {med(db):>+8.2%} {med(da):>+8.2%} "
f"{med(hl):>16.2%} {med(hi_):>16.2%} {med(dh):>+10.2%}")
print(" `Δbid`/`Δask` = quanto la quota lineare sta sopra/sotto la gemella inverse, in")
print(" percentuale del premio equo. `mezzo-spread` = quanto costa attraversare, per gamba.")
print(" La colonna che conta per il f di VRP01 e' l'ULTIMA: e' su ~130 gambe appaiate, non")
print(" sulle 3 strutture di §7(ii).")
# --- 2d: dove UNA sola delle due famiglie e' quotata. Il conto puo' toccare solo la
# lineare, quindi uno strike quotato solo sull'inverse per lui NON ESISTE.
print("\n --- 2d COPERTURA: quale famiglia e' quotata dove l'altra non lo e' (put 4-45g)")
print(f" {'coppia':<12} {'coppie':>7} {'entrambe':>9} {'solo LINEARE':>13} "
f"{'solo inverse':>13} {'nessuna':>8}")
for lin, _ in PAIRS:
sub = [(a, b) for a, b in cp if a["fam"] == lin and a["typ"] == "P"
and DTE_LO <= a["dte"] <= DTE_HI]
def due(r):
return r["bid"] is not None and r["ask"] is not None
bb = sum(1 for a, b in sub if due(a) and due(b))
sl = sum(1 for a, b in sub if due(a) and not due(b))
si = sum(1 for a, b in sub if due(b) and not due(a))
nn = sum(1 for a, b in sub if not due(a) and not due(b))
print(f" {lin:<12} {len(sub):>7} {bb:>9} {sl:>13} {si:>13} {nn:>8}")
print(" Uno strike quotato SOLO sull'inverse, per un conto in USDC, semplicemente non")
print(" esiste: e' la forma in cui la scelta di famiglia toglie strumenti invece di darli.")
# --- C1 controllo positivo: strike adiacente
print("\n --- C1 CONTROLLO POSITIVO: lo stesso confronto con l'inverse allo strike ADIACENTE")
idx = {r["name"]: r for r in rows}
print(f" {'coppia':<12} {'|ΔIV| stesso strike':>20} {'|ΔIV| strike adiacente':>23} "
f"{'rapporto':>9} {'esito':>10}")
for lin, inv in PAIRS:
base, shift = [], []
byexp: dict = {}
for r in rows:
if r["fam"] == inv and r["typ"] == "P":
byexp.setdefault(round(r["dte"], 4), []).append(r)
for k in byexp:
byexp[k] = sorted(byexp[k], key=lambda x: x["K"])
for a, b in cp:
if a["fam"] != lin or a["typ"] != "P" or not (DTE_LO <= a["dte"] <= DTE_HI):
continue
if not (a["iv"] > 0 and b["iv"] > 0):
continue
base.append(abs(a["iv"] - b["iv"]))
g = byexp.get(round(b["dte"], 4), [])
j = next((i for i, x in enumerate(g) if x["name"] == b["name"]), None)
if j is not None and j + 1 < len(g) and g[j + 1]["iv"] > 0:
shift.append(abs(a["iv"] - g[j + 1]["iv"]))
mb, ms = med(base), med(shift)
ok = ms > 2.0 * mb
print(f" {lin:<12} {mb:>20.3f} {ms:>23.3f} {ms/max(mb,1e-9):>9.1f}x "
f"{'POTENZA OK' if ok else '⚠️ SENZA POTENZA':>10}")
print(" ⚠️ Il rapporto e' enorme perche' il denominatore e' ESATTAMENTE zero (vedi 2b):")
print(" il controllo passa, ma su una grandezza degenere. Percio' lo stesso controllo va")
print(" rifatto sulla grandezza NON degenere, il mid eseguibile:")
print(f" {'coppia':<12} {'|Δmid| stesso strike':>21} {'|Δmid| adiacente':>18} "
f"{'rapporto':>9} {'esito':>16}")
for lin, inv in PAIRS:
byexp: dict = {}
for r in rows:
if r["fam"] == inv and r["typ"] == "P" and r["bid"] and r["ask"]:
byexp.setdefault(round(r["dte"], 4), []).append(r)
for k in byexp:
byexp[k] = sorted(byexp[k], key=lambda x: x["K"])
base, shift = [], []
for a, b in cp:
if a["fam"] != lin or a["typ"] != "P" or not (DTE_LO <= a["dte"] <= DTE_HI):
continue
if not (a["bid"] and a["ask"] and b["bid"] and b["ask"] and a["mark"] > 0):
continue
ma = 0.5 * (u(a["bid"], a) + u(a["ask"], a))
mb = 0.5 * (u(b["bid"], b) + u(b["ask"], b))
base.append(abs(ma - mb) / u(a["mark"], a))
g = byexp.get(round(b["dte"], 4), [])
j = next((i for i, x in enumerate(g) if x["name"] == b["name"]), None)
if j is not None and j + 1 < len(g):
c = g[j + 1]
shift.append(abs(ma - 0.5 * (u(c["bid"], c) + u(c["ask"], c))) / u(a["mark"], a))
r0, r1 = med(base), med(shift)
print(f" {lin:<12} {r0:>20.2%} {r1:>17.2%} {r1/max(r0,1e-12):>9.1f}x "
f"{('POTENZA OK' if r1 > 2*r0 else '⚠️ SENZA POTENZA'):>16}")
print(" Se lo strike sbagliato NON alzasse la distanza, il confronto non distinguerebbe")
print(" due superfici diverse e ogni 'coincidono' sarebbe vuoto.")
return res
# ================================================================== §3 il pavimento del tick
DBUCK = ((0.00, 0.02), (0.02, 0.05), (0.05, 0.10), (0.10, 0.20), (0.20, 0.35), (0.35, 0.65))
TBUCK = ((0.0, 7.0), (7.0, 30.0), (30.0, 95.0))
def sez3_tick(rows: list[dict], att: dict) -> dict:
print("\n" + "=" * 104)
print("3. IL PAVIMENTO DEL TICK — `tick effettivo / premio equo`, per delta e per scadenza")
print("=" * 104)
print(" E' il numero TRASFERIBILE del filone: non dipende dal capitale, non dipende dal")
print(" lotto, non dipende dal giorno. `>1` significa che il premio equo sta SOTTO un tick,")
print(" cioe' che quell'opzione non e' quotabile al suo valore: il prezzo minimo possibile")
print(" e' gia' un multiplo del giusto. Il rapporto E' quel multiplo.")
out: dict = {}
for lin, inv in PAIRS:
print(f"\n --- {lin} contro {inv} (mediana del rapporto; 'n' = strumenti nel riquadro)")
print(f" {'|delta|':<12} " + " ".join(f"{f'{a:.0f}-{b:.0f}g':>18}" for a, b in TBUCK))
for dlo, dhi in DBUCK:
cells = []
for tlo, thi in TBUCK:
vals = {}
for f in (lin, inv):
sub = [r for r in rows if r["fam"] == f and r["typ"] == "P"
and np.isfinite(r["delta"]) and dlo <= abs(r["delta"]) < dhi
and tlo <= r["dte"] < thi and r["mark"] > 0]
vals[f] = med([eff_tick(r["mark"], r["tick0"], r["steps"]) / r["mark"]
for r in sub])
out[(f, dlo, tlo)] = (vals[f], len(sub))
cells.append(f"{vals[lin]:>7.2f} /{vals[inv]:>7.2f}")
print(f" {f'{dlo:.2f}-{dhi:.2f}':<12} " + " ".join(f"{c:>18}" for c in cells))
print(f" {'':<12} " + " ".join(f"{'lineare / inverse':>18}" for _ in TBUCK))
# C2: controllo positivo del rilevatore
atm = [v for (f, d, t), (v, n) in out.items() if d == 0.35 and np.isfinite(v)]
w05 = [v for (f, d, t), (v, n) in out.items() if d == 0.02 and np.isfinite(v)]
w02 = [v for (f, d, t), (v, n) in out.items() if d == 0.00 and np.isfinite(v)]
print(f"\n --- C2 CONTROLLO POSITIVO del rilevatore")
print(f" soglia DICHIARATA A PRIORI: ala > 20x ATM.")
print(f" ATM (|d| 0.35-0.65) mediana {med(atm):.4f} · |d| 0.02-0.05 {med(w05):.3f} "
f"= {med(w05)/max(med(atm),1e-12):.1f}x -> "
f"{'PASSA' if med(w05) > 20*med(atm) else 'NON PASSA la soglia a priori'}")
print(f" ⚠️ ERRORE MIO, non del rilevatore: la griglia di delta che avevo dichiarato")
print(f" NON CONTENEVA l'ala di cui parla §46. Una put marcata 0,82 USDC su un")
print(f" sottostante a 77k non sta a |delta| 0.02-0.05, sta SOTTO 0.02. Bucket")
print(f" aggiunto: |d| 0.00-0.02 mediana {med(w02):.3f} = "
f"{med(w02)/max(med(atm),1e-12):.0f}x ATM -> "
f"{'SEPARA' if med(w02) > 20*med(atm) else 'NON SEPARA'}")
worst = med([v for (f, d, t), (v, n) in out.items()
if d == 0.00 and t == 0.0 and np.isfinite(v)])
print(f" ⚠️ E la soglia che avevo dichiarato era mal posta anche come STATISTICA: una")
print(f" mediana di bucket mescola 0-7g e 30-95g. Sulla cella dove il fenomeno vive")
print(f" (|d|<0.02 E 0-7g) il rapporto e' {worst:.2f} = {worst/max(med(atm),1e-12):.0f}x ATM.")
print(f" Riporto la soglia come dichiarata (NON superata) e la correzione accanto.")
mono = all(med([v for (f, d, t), (v, n) in out.items() if d == a and np.isfinite(v)])
>= med([v for (f, d, t), (v, n) in out.items() if d == b and np.isfinite(v)])
for a, b in zip([x[0] for x in DBUCK], [x[0] for x in DBUCK][1:]))
print(f" monotonia sui {len(DBUCK)} bucket di delta (il rapporto deve solo CRESCERE")
print(f" andando verso l'ala): {'SI' if mono else 'NO'}")
print(" Un rilevatore che desse lo stesso numero ATM e sull'ala sarebbe rotto, non muto.")
# il numero di §46, ricalcolato in modo appaiato
print("\n --- il '6x' di §46, rifatto sulle DUE famiglie invece che su una sola")
for lin, inv in PAIRS:
for f in (lin, inv):
sub = [r for r in rows if r["fam"] == f and r["typ"] == "P" and r["mark"] > 0
and np.isfinite(r["delta"]) and abs(r["delta"]) < 0.05 and r["dte"] <= 30]
if not sub:
continue
rat = [eff_tick(r["mark"], r["tick0"], r["steps"]) / r["mark"] for r in sub]
nq = sum(1 for x in rat if x > 1.0)
print(f" {f:<10} ali |d|<0.05, <=30g: n={len(sub):>4} "
f"tick/premio mediano {med(rat):>6.2f}x "
f"non quotabili al giusto: {nq}/{len(sub)} ({nq/len(sub):.0%})")
att["A3_6x"] = "ala |d|<0.05: mediana misurata (vedi tabella)"
# --- la versione APPAIATA del pavimento: la tabella qui sopra confronta POPOLAZIONI
# diverse (l'inverse ha 624 strike che la lineare non ha) e in una cella si ribalta.
print("\n --- LO STESSO PAVIMENTO, APPAIATO (stesso strumento nelle due famiglie)")
print(" La tabella qui sopra confronta POPOLAZIONI diverse: l'inverse ha 624 strike che")
print(" la lineare non ha, e in una cella (BTC |d|<0.02, 0-7g) il segno si ROVESCIA.")
print(" Appaiato non puo' succedere: il numeratore e' un fatto del venue e il")
print(" denominatore e' lo stesso premio (una sola superficie di mark, §2b).")
idx = {r["name"]: r for r in rows}
print(f" {'coppia':<12} {'n appaiate':>11} {'rapporto lin/inv':>17} "
f"{'atteso da tick $':>17} {'lin quotabile & inv NO':>24} {'il contrario':>13}")
for lin, inv in PAIRS:
rr, only_lin, only_inv = [], 0, 0
for r in rows:
if r["fam"] != lin or r["typ"] != "P" or r["mark"] <= 0:
continue
t = idx.get(r["twin"] or "")
if t is None or t["mark"] <= 0:
continue
a = eff_tick(r["mark"], r["tick0"], r["steps"]) / r["mark"]
b = eff_tick(t["mark"], t["tick0"], t["steps"]) / t["mark"]
rr.append(a / b)
only_lin += int(a < 1.0 <= b)
only_inv += int(b < 1.0 <= a)
sub = [r for r in rows if r["fam"] == lin and r["typ"] == "P"]
sui = [r for r in rows if r["fam"] == inv and r["typ"] == "P"]
F = med([r["F"] for r in sui])
atteso = (sub[0]["tick0"]) / (sui[0]["tick0"] * F)
print(f" {lin:<12} {len(rr):>11} {med(rr):>17.3f} {atteso:>17.3f} "
f"{only_lin:>24} {only_inv:>13}")
print(" `atteso da tick $` = (tick lineare in $) / (tick inverse in $): se il rapporto")
print(" misurato coincide, il pavimento e' esattamente il tick e nient'altro.")
# --- cosa costa DAVVERO comprare l'ala: il pavimento si paga sull'ASK, non sul mark
print("\n --- IL PAVIMENTO SI PAGA SULL'ASK: quanto si paga sopra il premio equo, per")
print(" COMPRARE (lato long, l'unico che il progetto puo' guardare)")
print(f" {'|delta|':<12} " + " ".join(f"{f:>18}" for f in
("BTC_USDC", "BTC", "ETH_USDC", "ETH")))
for dlo, dhi in DBUCK:
cells = []
for f in ("BTC_USDC", "BTC", "ETH_USDC", "ETH"):
sub = [r for r in rows if r["fam"] == f and r["typ"] == "P" and r["ask"]
and r["mark"] > 0 and np.isfinite(r["delta"])
and dlo <= abs(r["delta"]) < dhi and r["dte"] <= 45.0]
v = med([r["ask"] / r["mark"] for r in sub])
cells.append(f"{v:>10.2f}x n={len(sub):<4}")
print(f" {f'{dlo:.2f}-{dhi:.2f}':<12} " + " ".join(f"{c:>18}" for c in cells))
print(" `x` = ask / premio equo del venue. E' il sovrapprezzo che si paga per")
print(" comprare quell'ala, e sull'ala profonda NON e' negoziabile: e' il tick.")
print("\n ⚠️ anche questa tabella confronta POPOLAZIONI diverse. La versione appaiata")
print(" (stesso strumento, solo dove ENTRAMBE hanno un ask):")
print(f" {'|delta|':<12} " + " ".join(f"{f'{l} / {i}':>26}" for l, i in PAIRS))
for dlo, dhi in DBUCK:
cells = []
for lin, inv in PAIRS:
rr = []
for r in rows:
if (r["fam"] != lin or r["typ"] != "P" or not r["ask"] or r["mark"] <= 0
or not np.isfinite(r["delta"]) or not (dlo <= abs(r["delta"]) < dhi)
or r["dte"] > 45.0):
continue
t = idx.get(r["twin"] or "")
if t is None or not t["ask"] or t["mark"] <= 0:
continue
rr.append((r["ask"] / r["mark"]) / (t["ask"] / t["mark"]))
cells.append(f"{med(rr):>8.3f} n={len(rr):<4}")
print(f" {f'{dlo:.2f}-{dhi:.2f}':<12} " + " ".join(f"{c:>26}" for c in cells))
print(" <1 = la LINEARE si compra piu' vicino al giusto; >1 = la inverse.")
# --- il premio minimo quotabile, per lotto: il numero che nessun capitale aggira
print("\n --- IL PREMIO MINIMO QUOTABILE PER UN LOTTO (1 tick x lotto minimo)")
print(f" {'famiglia':<10} {'tick base':>11} {'lotto min':>10} {'premio min $/lotto':>19} "
f"{'noz. lotto $':>13} {'premio min / nozionale':>23}")
for lin, inv in PAIRS:
for f in (lin, inv):
sub = [r for r in rows if r["fam"] == f and r["typ"] == "P"]
if not sub:
continue
a = sub[0]
F = med([r["F"] for r in sub])
pmin = a["tick0"] * a["minamt"] * (F if a["inv"] else 1.0)
noz = a["minamt"] * F
print(f" {f:<10} {a['tick0']:>11g} {a['minamt']:>10g} {pmin:>19.3f} "
f"{noz:>13,.0f} {pmin/noz:>22.5%}")
print(" A sinistra c'e' la granularita' (quanto piccolo puo' essere un biglietto), a")
print(" destra il pavimento in proporzione: e' il secondo che decide se un'ala e'")
print(" comprabile al suo valore, ed e' INDIPENDENTE dal capitale.")
return out
# ================================================================== §4 le strutture
def scegli_put(rows: list[dict], fam: str, target: float, lato: str,
dte_lo: float, dte_hi: float) -> list[dict]:
"""Per ogni scadenza, la put col delta piu' vicino al target FRA QUELLE CHE HANNO IL LATO
CHE SERVE (bid per vendere, ask per comprare). Un book a un lato solo non e' un book vuoto:
e' un book che serve una direzione sola, e la restrizione giusta e' asimmetrica."""
g = [r for r in rows if r["fam"] == fam and r["typ"] == "P" and np.isfinite(r["delta"])
and dte_lo <= r["dte"] <= dte_hi
and ((r["bid"] is not None) if lato == "vendi" else (r["ask"] is not None))]
out = []
for e in sorted({round(r["dte"], 4) for r in g}):
gg = [r for r in g if round(r["dte"], 4) == e]
out.append(min(gg, key=lambda r: abs(r["delta"] - target)))
return out
def sez4_strutture(rows: list[dict], att: dict) -> None:
print("\n" + "=" * 104)
print(f"4. LA DOMANDA CHE DECIDE — cosa entra in un conto da ${CAPITAL:.0f}, e in quale famiglia")
print("=" * 104)
print(" Tre strutture a rischio definito, per UN LOTTO MINIMO, ai prezzi ESEGUIBILI")
print(" (si compra all'ask e si vende al bid, mai al mid: §28).")
print(" L1 LONG put d-0,05 ... l'assicurazione di coda di §46")
print(" L2 LONG debit spread compra d-0,28 / vende d-0,10 ... rischio definito, lato long")
print(" S1 SHORT credit spread vende d-0,28 / compra d-0,10 ... la geometria di VRP01")
print(f"\n{'struttura':<4} {'famiglia':<10} {'scad':>4} {'ticket $':>9} {'max perdita $':>13} "
f"{'eseg/equo':>10} {'fee/ticket':>10} {'n a 20% conto':>13} {'entra a $635?':>13}")
tab: dict = {}
for lin, inv in PAIRS:
for f in (lin, inv):
# L1 -------------------------------------------------------------- long put di coda
gs = scegli_put(rows, f, TAIL_D, "compra", 5.0, 35.0)
if gs:
tick_ = [lot_usd(r["ask"], r) for r in gs]
equo = [lot_usd(r["mark"], r) for r in gs]
fee = [fee_lot(r, r["ask"]) for r in gs]
t, e_ = med(tick_), med(equo)
n = math.floor(RISK_BUDGET * CAPITAL / t) if t > 0 else 0
tab[("L1", f)] = t
print(f"{'L1':<4} {f:<10} {len(gs):>4} {t:>9.2f} {t:>13.2f} "
f"{(t/e_ if e_ > 0 else float('nan')):>10.2f} "
f"{med(fee)/max(t,1e-9):>9.0%} {n:>13} "
f"{'SI' if t <= CAPITAL else 'NO':>13}")
# L2 -------------------------------------------------------- long put debit spread
hi = scegli_put(rows, f, SHORT_D, "compra", DTE_LO, DTE_HI)
lo = scegli_put(rows, f, LONG_D, "vendi", DTE_LO, DTE_HI)
lom = {round(r["dte"], 4): r for r in lo}
cost, costm, fees = [], [], []
for a in hi:
b = lom.get(round(a["dte"], 4))
if b is None or b["K"] >= a["K"]:
continue
c = lot_usd(a["ask"], a) - lot_usd(b["bid"], b)
cm = lot_usd(a["mark"], a) - lot_usd(b["mark"], b)
if cm <= 0:
continue
cost.append(c); costm.append(cm)
fees.append(fee_lot(a, a["ask"]) + fee_lot(b, b["bid"]))
if cost:
t, e_ = med(cost), med(costm)
n = math.floor(RISK_BUDGET * CAPITAL / t) if t > 0 else 0
tab[("L2", f)] = t
print(f"{'L2':<4} {f:<10} {len(cost):>4} {t:>9.2f} {t:>13.2f} "
f"{t/e_:>10.2f} {med(fees)/max(t,1e-9):>9.0%} {n:>13} "
f"{'SI' if t <= CAPITAL else 'NO':>13}")
# S1 ------------------------------------------------------- short put credit spread
sh = scegli_put(rows, f, SHORT_D, "vendi", DTE_LO, DTE_HI)
wg = scegli_put(rows, f, LONG_D, "compra", DTE_LO, DTE_HI)
wgm = {round(r["dte"], 4): r for r in wg}
ml, cred, credm, fees = [], [], [], []
for a in sh:
b = wgm.get(round(a["dte"], 4))
if b is None or b["K"] >= a["K"]:
continue
c = lot_usd(a["bid"], a) - lot_usd(b["ask"], b)
cm = lot_usd(a["mark"], a) - lot_usd(b["mark"], b)
if cm <= 0:
continue
cred.append(c); credm.append(cm)
ml.append((a["K"] - b["K"]) * a["minamt"] - c)
fees.append(fee_lot(a, a["bid"]) + fee_lot(b, b["ask"]))
if ml:
m, e_ = med(ml), med(credm)
n = math.floor(RISK_BUDGET * CAPITAL / m) if m > 0 else 0
tab[("S1", f)] = m
print(f"{'S1':<4} {f:<10} {len(ml):>4} {med(cred):>9.2f} {m:>13.2f} "
f"{med(cred)/e_:>10.2f} {med(fees)/max(abs(med(cred)),1e-9):>9.0%} {n:>13} "
f"{'SI' if m <= CAPITAL else 'NO':>13}")
print("\n `eseg/equo` = ticket eseguibile / ticket al mark del venue. Sul LONG e' >1 (paghi")
print(" di piu' del giusto), sul credit e' <1 (incassi di meno): e' il f_venue di §14.")
print(" `max perdita` per L1/L2 e' il ticket stesso (rischio definito per costruzione).")
print(" ⚠️ S1 e' misurata per COMPLETEZZA: la regola del 19/06 (`niente short-vol da modello")
print(" in deploy`) non e' una regola di eseguibilita' e non decade col lotto.")
att["A6"] = ("CONFERMATA" if all(tab.get(("L1", f), 9e9) < CAPITAL
for _, f in PAIRS) else "REFUTATA")
print("\n --- LA DOMANDA DI §64: esiste sulla LINEARE una struttura che l'INVERSE non fa?")
for lin, inv in PAIRS:
for k, nome in (("L1", "long put di coda"), ("L2", "long debit spread"),
("S1", "short credit spread")):
a, b = tab.get((k, lin)), tab.get((k, inv))
if a is None or b is None:
print(f" {k} {nome:<22} {lin:<10} {'n/d' if a is None else f'{a:8.2f}'} "
f" {inv:<4} {'n/d' if b is None else f'{b:8.2f}'}")
continue
v_lin = a <= RISK_BUDGET * CAPITAL
v_inv = b <= RISK_BUDGET * CAPITAL
marca = ("SOLO LINEARE" if (v_lin and not v_inv) else
"entrambe" if (v_lin and v_inv) else
"SOLO INVERSE" if v_inv else "nessuna delle due")
print(f" {k} {nome:<22} {lin:<10} ${a:>8.2f} {inv:<4} ${b:>8.2f} "
f"a budget {RISK_BUDGET:.0%}: {marca}")
# ================================================================== §5 profondita' e controlli
def sez5_profondita(rows: list[dict], raw: dict) -> None:
print("\n" + "=" * 104)
print("5. PROFONDITA' IN CIMA AL BOOK — sulle STESSE gambe nelle due famiglie")
print("=" * 104)
d = raw.get("depth") or {}
if not d:
print(" nessun order_book in questo snapshot (raccolto con --no-depth?).")
return
idx = {r["name"]: r for r in rows}
print(f"{'strumento':<26} {'delta':>7} {'lato':<5} {'lotti al best':>13} "
f"{'$ al best':>11} {'prezzo/tick':>11} {'|d venue-BS|':>12}")
for n in sorted(d):
r = idx.get(n)
if r is None:
continue
ob = d[n]
for lato, side in (("bid", "bids"), ("ask", "asks")):
lv = ob.get(side) or []
if not lv:
print(f"{n:<26} {r['delta']:>7.3f} {lato:<5} {'book vuoto su questo lato':>13}")
continue
px, sz = float(lv[0][0]), float(lv[0][1])
gv = (ob.get("greeks") or {}).get("delta")
dd = abs(float(gv) - r["delta"]) if (gv is not None
and np.isfinite(r["delta"])) else float("nan")
print(f"{n:<26} {r['delta']:>7.3f} {lato:<5} "
f"{sz/max(r['minamt'],1e-12):>13,.0f} {u(px, r)*sz:>11,.0f} "
f"{px/max(eff_tick(px, r['tick0'], r['steps']),1e-12):>11.1f} {dd:>12.4f}")
print("\n `lotti al best` = size in cima / lotto minimo: quante strutture minime passano")
print(" senza muovere il prezzo. `prezzo/tick` e' il controllo d'artefatto del 30/07 (un")
print(" prezzo a 1-2 tick e' una griglia, non un prezzo). `|d venue-BS|` valida il MIO")
print(" delta contro le greche del venue: se non e' piccolo, tutta la selezione e' sbagliata.")
def sez5b_controlli(rows: list[dict], att: dict) -> None:
print("\n" + "-" * 104)
print("5b. CONTROLLI C3 e C4")
print("-" * 104)
# C3 — il rilevatore di larghezza separa una famiglia nota-larga da una nota-stretta?
print(f" {'famiglia':<10} {'put 4-45g':>10} {'2 lati':>7} {'%':>6} {'spr% mediano':>13} "
f"{'best bid a 1 tick':>18}")
for f in ("BTC_USDC", "SOL_USDC"):
puts = [r for r in rows if r["fam"] == f and r["typ"] == "P"
and DTE_LO <= r["dte"] <= DTE_HI]
two = [r for r in puts if r["bid"] is not None and r["ask"] is not None]
onetick = [r for r in puts if r["bid"] is not None
and r["bid"] <= 1.01 * eff_tick(r["bid"], r["tick0"], r["steps"])]
print(f" {f:<10} {len(puts):>10} {len(two):>7} "
f"{(len(two)/len(puts) if puts else float('nan')):>5.0%} "
f"{med([relspread(r) for r in two])*100:>12.1f}% "
f"{len(onetick)/max(len([r for r in puts if r['bid'] is not None]),1):>17.0%}")
print(" C3: se BTC_USDC e SOL_USDC uscissero uguali, il rilevatore non separa niente.")
# C4 — tipicita' del giorno, sulla grandezza gia' misurata su 19 settimane feriali
print("\n C4 tipicita': attraversare lo spread sul credit spread canonico INVERSE costa")
print(" ~10% del credito (misura del 30/07 su 19 settimane FERIALI); §14 ha letto")
print(" 7,0-7,3% di sabato. Oggi e' domenica:")
for _, inv in PAIRS:
sh = scegli_put(rows, inv, SHORT_D, "vendi", DTE_LO, DTE_HI)
wg = {round(r["dte"], 4): r for r in scegli_put(rows, inv, LONG_D, "compra",
DTE_LO, DTE_HI)}
fv = []
for a in sh:
b = wg.get(round(a["dte"], 4))
if b is None or b["K"] >= a["K"]:
continue
cm = lot_usd(a["mark"], a) - lot_usd(b["mark"], b)
ce = lot_usd(a["bid"], a) - lot_usd(b["ask"], b)
if cm > 0:
fv.append(1.0 - ce / cm)
print(f" {inv:<4} inverse: costo di attraversamento = {med(fv)*100:5.1f}% del credito"
f" (atteso ~7-10%)")
ok = all(0.0 <= med([1.0 - (lot_usd(a["bid"], a) - lot_usd(b["ask"], b))
/ (lot_usd(a["mark"], a) - lot_usd(b["mark"], b))
for a in scegli_put(rows, inv, SHORT_D, "vendi", DTE_LO, DTE_HI)
for b in [{round(r["dte"], 4): r for r in
scegli_put(rows, inv, LONG_D, "compra", DTE_LO, DTE_HI)}
.get(round(a["dte"], 4))]
if b is not None and b["K"] < a["K"]
and (lot_usd(a["mark"], a) - lot_usd(b["mark"], b)) > 0]) <= 0.25
for _, inv in PAIRS)
att["C4"] = "il giorno non e' anomalo sui book noti" if ok else "⚠️ giorno anomalo"
print(" E' un controllo DEBOLE (una grandezza, un istante), non una prova di tipicita'.")
def sez6_dispersione(snaps: list[dict]) -> None:
print("\n" + "=" * 104)
print(f"6. DISPERSIONE FRA SNAPSHOT — {len(snaps)} istantanee della stessa sessione")
print("=" * 104)
if len(snaps) < 2:
print(" UNA SOLA ISTANTANEA: qui non c'e' dispersione da riportare, e va detto cosi'.")
print(" Rilancia con --snap a distanza di minuti per avere la banda infra-sessione.")
return
print(f"{'ora UTC':<18} {'USDC OI>=100':>12} {'inv OI>=100':>12} "
f"{'BTC_USDC 2lati%':>16} {'ETH_USDC 2lati%':>16} {'SOL_USDC 2lati%':>16} "
f"{'ΔIV BTC':>9} {'ΔIV ETH':>9} {'Δspr BTC pp':>12} {'Δspr ETH pp':>12}")
acc: dict = {k: [] for k in ("uo", "io", "bq", "eq", "sq", "biv", "eiv", "bsp", "esp")}
for raw in snaps:
rows = build(raw)
cp = coppie(rows)
def q(f):
puts = [r for r in rows if r["fam"] == f and r["typ"] == "P"
and DTE_LO <= r["dte"] <= DTE_HI]
two = [r for r in puts if r["bid"] is not None and r["ask"] is not None]
return len(two) / len(puts) if puts else float("nan")
def dd(lin):
both = [(a, b) for a, b in cp if a["fam"] == lin and a["typ"] == "P"
and DTE_LO <= a["dte"] <= DTE_HI and a["bid"] and a["ask"]
and b["bid"] and b["ask"] and a["iv"] > 0 and b["iv"] > 0]
return (med([a["iv"] - b["iv"] for a, b in both]),
med([relspread(a) - relspread(b) for a, b in both]))
uo = sum(1 for r in rows if r["fam"] in ("BTC_USDC", "ETH_USDC")
and 0 < r["dte"] <= DTE_MAX_COLL and r["oi"] >= OI_MIN)
io = sum(1 for r in rows if r["fam"] in ("BTC", "ETH")
and 0 < r["dte"] <= DTE_MAX_COLL and r["oi"] >= OI_MIN)
biv, bsp = dd("BTC_USDC")
eiv, esp = dd("ETH_USDC")
bq, eq, sq = q("BTC_USDC"), q("ETH_USDC"), q("SOL_USDC")
for k, v in zip(acc, (uo, io, bq, eq, sq, biv, eiv, bsp, esp)):
acc[k].append(v)
print(f"{raw['ts'][11:19]:<18} {uo:>12} {io:>12} {bq:>15.0%} {eq:>15.0%} {sq:>15.0%} "
f"{biv:>+9.3f} {eiv:>+9.3f} {bsp*100:>+11.2f} {esp*100:>+11.2f}")
print("\n banda (min-max) sulle grandezze che portano il verdetto:")
for k, nome in (("uo", "USDC OI>=100"), ("io", "inverse OI>=100"),
("bq", "BTC_USDC quotati 2 lati"), ("eq", "ETH_USDC quotati 2 lati"),
("sq", "SOL_USDC quotati 2 lati"),
("biv", "ΔIV appaiata BTC (pt vol)"), ("eiv", "ΔIV appaiata ETH (pt vol)"),
("bsp", "Δspread appaiato BTC"), ("esp", "Δspread appaiato ETH")):
v = [x for x in acc[k] if np.isfinite(x)]
if v:
print(f" {nome:<28} min {min(v):+10.4f} mediana {np.median(v):+10.4f} "
f"max {max(v):+10.4f}")
print("\n ⚠️ E' dispersione INFRA-SESSIONE: dice se il numero balla fra un giro e l'altro,")
print(" NON se un altro giorno della settimana darebbe un'altra risposta.")
# ================================================================== §7 la decisione su §51
def sez7_decisione(rows: list[dict], res2: dict, att: dict) -> dict:
print("\n" + "=" * 104)
print("7. LA DECISIONE CHE DIPENDE DA QUESTO FILONE — raccogliere o no la catena USDC (§51)")
print("=" * 104)
print(" §51 raccomanda di RACCOGLIERLA (+117 chiamate/giro, +18%) con UNA giustificazione:")
print(" *rende onesti i numeri di uno sleeve vivo (VRP01), perche' il f=0,73 e' misurato")
print(" sulla famiglia inverse che un conto USDC non puo' marginare*. Quella")
print(" giustificazione e' una PREVISIONE: che la superficie lineare prezzi diversamente.")
print(" Qui la previsione si misura, appaiata.")
print("\n Il f di VRP01 ha DUE pezzi, e vanno separati perche' si trasferiscono in modo")
print(" diverso:")
print(" (i) errore di MODELLO (il sleeve prezza a DVOL ATM: struttura a termine 42% +")
print(" skew 25%, §5/§35). Vive nella IV -> si trasferisce se ΔIV appaiata ~ 0.")
print(" (ii) costo di ATTRAVERSAMENTO (bid/ask). Vive nello spread -> si trasferisce solo")
print(" se lo spread appaiato e' simile.")
out = {}
print("\n (i) SI TRASFERISCE PER COSTRUZIONE, e non perche' l'abbia misurato io: §2b mostra")
print(" che il venue marca le due famiglie sulla STESSA superficie (mark_iv identica")
print(" nel 95% delle coppie, il resto e' l'arrotondamento a 2 decimali). L'errore di")
print(" modello del sleeve e' calcolato CONTRO quella superficie: e' lo stesso numero")
print(" nelle due famiglie, e nessuna raccolta di dati puo' cambiarlo.")
print(f" {'coppia':<10} {'ΔIV appaiata':>13} {'mezzo-spread IV':>16} {'rapporto':>9}"
f" (test SENZA POTENZA, riportato per trasparenza)")
for lin, _ in PAIRS:
r = res2[lin]
rr = abs(r["div"]) / r["halfiv"] if r["halfiv"] and np.isfinite(r["halfiv"]) else np.nan
out[f"iv_{lin}"] = rr
print(f" {lin:<10} {r['div']:>+13.3f} {r['halfiv']:>16.3f} {rr:>9.2f}")
print("\n (ii-a) IL COSTO DI ATTRAVERSAMENTO, per GAMBA, su tutte le coppie a due lati")
print(" (e' la misura robusta: ~130 osservazioni appaiate invece di 3 strutture)")
print(f" {'coppia':<10} {'n gambe':>8} {'mezzo-spread lineare':>21} "
f"{'mezzo-spread inverse':>21} {'Δ appaiata':>11}")
for lin, _ in PAIRS:
r = res2[lin]
out[f"dh_{lin}"] = r.get("dhalf", float("nan"))
print(f" {lin:<10} {r.get('nboth', 0):>8} {r.get('hl', float('nan')):>20.2%} "
f"{r.get('hi', float('nan')):>20.2%} {r.get('dhalf', float('nan')):>+10.2%}")
# (ii) f_venue APPAIATO: la stessa struttura, gli stessi strike, nelle due famiglie
print(f"\n (ii-b) f_venue sulla STESSA struttura e sugli STESSI strike nelle due famiglie")
print(f" {'coppia':<10} {'n strutture':>12} {'f_venue lineare':>16} "
f"{'f_venue inverse':>16} {'Δ appaiata':>11}")
idx = {r["name"]: r for r in rows}
for lin, inv in PAIRS:
sh = scegli_put(rows, lin, SHORT_D, "vendi", DTE_LO, DTE_HI)
wg = {round(r["dte"], 4): r for r in scegli_put(rows, lin, LONG_D, "compra",
DTE_LO, DTE_HI)}
fl, fi, dl = [], [], []
for a in sh:
b = wg.get(round(a["dte"], 4))
if b is None or b["K"] >= a["K"]:
continue
A, B = idx.get(a["twin"] or ""), idx.get(b["twin"] or "")
if A is None or B is None:
continue
if not (a["bid"] and b["ask"] and A["bid"] and B["ask"]):
continue
cmL = lot_usd(a["mark"], a) - lot_usd(b["mark"], b)
ceL = lot_usd(a["bid"], a) - lot_usd(b["ask"], b)
cmI = lot_usd(A["mark"], A) - lot_usd(B["mark"], B)
ceI = lot_usd(A["bid"], A) - lot_usd(B["ask"], B)
if cmL <= 0 or cmI <= 0:
continue
fl.append(ceL / cmL); fi.append(ceI / cmI); dl.append(ceL / cmL - ceI / cmI)
out[f"fv_{lin}"] = med(dl)
out[f"fvn_{lin}"] = len(dl)
print(f" {lin:<10} {len(dl):>12} {med(fl):>16.3f} {med(fi):>16.3f} "
f"{med(dl):>+11.3f}")
print("\n ⚠️ e' la MEDIANA DELLE DIFFERENZE APPAIATE su strutture con gli STESSI")
print(" strike: §14 confrontava strike auto-selezionati e misurava anche la griglia.")
print("\n (iii) L'UNICA COSA DAVVERO DIVERSA FRA LE DUE FAMIGLIE E' LA GRIGLIA DEGLI")
print(" STRIKE: la lineare e' un sottoinsieme (§1). Quanto deforma la struttura?")
print(" Si sceglie il delta bersaglio due volte: fra TUTTI gli strike inverse, e")
print(" fra i soli strike che esistono ANCHE sulla lineare.")
print(f" {'coppia':<10} {'bersaglio':>10} {'scad':>5} {'|d| griglia piena':>18} "
f"{'|d| griglia lineare':>20} {'Δ strike medio':>15}")
for lin, inv in PAIRS:
twins = {r["twin"] for r in rows if r["fam"] == lin and r["twin"]}
for tgt in (SHORT_D, LONG_D, TAIL_D):
full, restr, dk = [], [], []
for e in sorted({round(r["dte"], 4) for r in rows
if r["fam"] == inv and r["typ"] == "P"
and DTE_LO <= r["dte"] <= DTE_HI}):
g = [r for r in rows if r["fam"] == inv and r["typ"] == "P"
and round(r["dte"], 4) == e and np.isfinite(r["delta"])]
gr = [r for r in g if r["name"] in twins]
if not g or not gr:
continue
a = min(g, key=lambda r: abs(r["delta"] - tgt))
b = min(gr, key=lambda r: abs(r["delta"] - tgt))
full.append(abs(a["delta"] - tgt)); restr.append(abs(b["delta"] - tgt))
dk.append(abs(a["K"] - b["K"]))
if full:
print(f" {lin:<10} {tgt:>10.2f} {len(full):>5} {med(full):>18.4f} "
f"{med(restr):>20.4f} {med(dk):>15,.0f}")
print(" Se le due colonne coincidono, la griglia piu' povera NON deforma la")
print(" struttura e un backtest fatto sull'inverse resta eseguibile sulla lineare.")
return out
def verdetto(att: dict, dec: dict, rows: list[dict]) -> None:
print("\n" + "=" * 104)
print("8. ATTESE DICHIARATE PRIMA DI MISURARE — esito")
print("=" * 104)
for k in sorted(att):
print(f" {k:<14} {att[k]}")
print("\n" + "=" * 104)
print("9. VERDETTO")
print("=" * 104)
iv_ok = True # per COSTRUZIONE: una sola superficie di mark al venue (§2b)
dh = [dec.get(f"dh_{l}") for l, _ in PAIRS]
fv = [dec.get(f"fv_{l}") for l, _ in PAIRS]
dh_ok = all(x is not None and np.isfinite(x) and abs(x) < 0.01 for x in dh)
fv_ok = all(x is not None and np.isfinite(x) and abs(x) < 0.05 for x in fv)
print(" (i) il PREZZO EQUO e' lo stesso numero per costruzione (una sola superficie di")
print(" mark al venue, §2b) -> nessun dato da raccogliere puo' cambiarlo")
print(f" (ii-a) mezzo-spread per gamba, Δ appaiata entro 1pp: {'SI' if dh_ok else 'NO'}"
+ " (Δ misurate: " + ", ".join(f"{x:+.2%}" for x in dh if x is not None) + ")")
print(f" (ii-b) f_venue di struttura, Δ appaiata entro 0,05: {'SI' if fv_ok else 'NO'}"
+ " (Δ misurate: " + ", ".join(f"{x:+.3f}" for x in fv if x is not None) + ")")
fv_ok = fv_ok and dh_ok
if iv_ok and fv_ok:
print("\n 🚨 LE DUE SUPERFICI SONO LA STESSA SUPERFICIE, a strike appaiati: la catena")
print(" INVERSE che il progetto gia' raccoglie e' un proxy sufficiente della lineare")
print(" sia sul PREZZO sia sul COSTO DI ATTRAVERSAMENTO.")
print(" => LA RACCOMANDAZIONE DI §51 (raccogliere la catena USDC, +18% di chiamate/giro)")
print(" NON E' GIUSTIFICATA DALLA RAGIONE CHE PORTA: il f di VRP01 misurato")
print(" sull'inverse non e' 'il numero della famiglia sbagliata', e' lo stesso numero.")
elif iv_ok and not fv_ok:
print("\n ⚠️ Il PREZZO coincide ma il COSTO DI ATTRAVERSAMENTO no: la meta' 'errore di")
print(" modello' del f si trasferisce, la meta' 'spread' NO. §51 va RIDOTTA alla sola")
print(" domanda sullo spread, che si risponde con una CAMPAGNA A TERMINE, non con un")
print(" collettore permanente (stessa forma della raccomandazione depth di §51).")
else:
print("\n Le due superfici prezzano DIVERSO a strike appaiati: la raccomandazione §51")
print(" REGGE ed e' anzi piu' forte di come e' scritta.")
print("\n Sulla domanda di §64 — *esiste sulla lineare una struttura che l'inverse non fa?*")
print(" la risposta sta nella tabella §4, e va letta insieme a due regole gia' scritte:")
print(" (a) 'niente short-vol da modello in deploy' (19/06) non e' una regola di")
print(" eseguibilita' e non decade col lotto;")
print(" (b) §46 ha gia' refutato la copertura statica di coda (maxDD sale in 162/162).")
print("\n IN EURO/GIORNO: questo filone non produce reddito. Non propone nessuna struttura,")
print(" non cambia nessun peso, non tocca il libro. Il suo intero valore e' una DECISIONE DI")
print(" INFRASTRUTTURA da +18% di chiamate/giro sul collettore, e un numero (il pavimento")
print(" del tick) che le prossime ondate non dovranno ri-misurare. **+0,00 EUR/giorno.**")
# ================================================================== main
def salva(raw: dict) -> Path:
SNAPDIR.mkdir(parents=True, exist_ok=True)
p = SNAPDIR / f"snap_{raw['ts'].replace(':', '').replace('-', '')[:15]}.json"
p.write_text(json.dumps(raw))
return p
def carica() -> list[dict]:
if not SNAPDIR.exists():
return []
out = []
for p in sorted(SNAPDIR.glob("snap_*.json")):
try:
out.append(json.loads(p.read_text()))
except Exception:
print(f" ⚠️ snapshot illeggibile, saltato: {p.name}")
return out
def main() -> int:
ap = argparse.ArgumentParser()
ap.add_argument("--snap", action="store_true", help="solo un'istantanea nuova")
ap.add_argument("--report", action="store_true", help="solo il report sugli snapshot su disco")
a = ap.parse_args()
fa_snap = a.snap or not a.report
fa_rep = a.report or not a.snap
print("=" * 104)
print("USDC-SURFACE (§64) — la superficie lineare contro la sua gemella inverse, appaiata")
print("=" * 104)
print("Sole GET pubbliche, nessun ordine. Qui si misura la NEGOZIABILITA', non un edge.")
b = Budget()
if fa_snap:
guardia_finestra()
t0 = time.time()
raw = fetch_snapshot(b)
p = salva(raw)
print(f"\n[rete] {b.calls} chiamate, {b.err} errori, {b.r429} risposte 429, "
f"{time.time()-t0:.0f}s -> {p.name}")
if not fa_rep:
return 0
snaps = carica()
if not snaps:
print("\nNESSUNO SNAPSHOT: rilancia con --snap. Verdetto: non misurabile.")
return 1
raw = snaps[-1]
rows = build(raw)
print(f"\n[dato] {len(rows)} strumenti con specifiche E quote nello snapshot piu' recente "
f"({raw['ts'][:19]} UTC); {len(snaps)} snapshot in sessione")
if not rows:
print("Il venue non ha risposto. Verdetto: non misurabile.")
return 1
att: dict = {}
sez0_repliche(rows, att)
sez1_censimento(rows)
res2 = sez2_appaiato(rows, att)
sez3_tick(rows, att)
sez4_strutture(rows, att)
sez5_profondita(rows, raw)
sez5b_controlli(rows, att)
sez6_dispersione(snaps)
dec = sez7_decisione(rows, res2, att)
verdetto(att, dec, rows)
print(f"\n[rete, totale] {b.calls} chiamate, {b.err} errori, {b.r429} risposte 429")
return 0
if __name__ == "__main__":
raise SystemExit(main())
Reference in New Issue View Git Blame Copy Permalink