Adriano/PythagorasGoal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
PythagorasGoal/scripts/research/r0823b_curva_datati.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

897 lines
46 KiB
Python
Raw Permalink Blame History

#!/usr/bin/env python3
"""
r0823b_curva_datati.py — filone §65 CURVA-DATATI (ondata 2026-08-23)
===========================================================================
LA DOMANDA
===========================================================================
L'ondata 1 (§13 BASIS-CALENDAR) ha ricostruito un dataset che il progetto non
aveva: **30 futures trimestrali SCADUTI** di Deribit (BTC 240.150 barre orarie /
ETH 233.334, dal 2018-09), **col deleveraging 2022 dentro** — la coda che a CC01
mancava per costruzione.
Finora e' stato usato per UNA cosa sola: il **livello** del carry (§13 lo ha
scartato come edge; §39 lo ha scartato come veicolo di TP01). **La FORMA della
curva — pendenza e curvatura — non e' mai stata guardata.**
===========================================================================
ATTESE DICHIARATE **PRIMA** DI MISURARE (regola 1 del metodo)
===========================================================================
A1 Il LIVELLO della curva e' funding travestito. §39 ha gia' misurato che il
basis dei datati E' il funding del perp (+7,33% implicito contro +6,48%
realizzato). Mi aspetto |corr(livello, funding)| ALTA (>0,5).
-> se confermata, il ramo "livello" e' chiuso in partenza.
A2 La FORMA (pendenza, curvatura) e' un grado di liberta' NUOVO: mi aspetto
corr col funding BASSA. E' l'unica ragione per cui questo filone esiste.
A3 Nessun ramo produce un candidato. Il filone funding e' chiuso su QUATTRO
lati (carry cross-sectional, price-clock, time-series, affollamento) e §13
ha gia' esaurito il primo ordine. Mi aspetto SCARTATO su tutti i rami.
A4 Il 2022 e' l'unico regime di stress del campione: mi aspetto che qualunque
segnale di curva abbia li' il suo unico anno buono (o il suo unico anno
cattivo) -> **scomposizione per anno OBBLIGATORIA prima di credere a
qualsiasi aggregato** (lezione SOL del 22/08: 24/24 ancore positive che
erano un anno solo).
A5 Il roll trimestrale costa. §39: SKH01 sui datati fa 48-57 roll-equivalenti
l'anno contro 6,6-8,4 di TP01 -> lo stesso mezzo spread gli costa 7 volte
tanto. Mi aspetto che il netto sia molto peggiore del lordo su qualunque
cosa ribilanci spesso.
A6 MDE: 7,4 anni -> MDE sullo Sharpe ~= 1,96/sqrt(7,4) = **0,72**. Un ramo
DELTA-NEUTRO (§13: SE del rendimento annuo 1,79pp, vol 4,9%) non e'
rispondibile; un ramo DIREZIONALE (vol 50-80%) lo e' in linea di principio.
Esito di ciascuna: riportato in fondo, confermata/refutata.
===========================================================================
COME E' COSTRUITA LA CURVA — e perche' COSI'
===========================================================================
Deribit lista SOLO trimestrali (piu' settimanali/mensili che qui non usiamo):
a ogni ora sono vivi 2-5 contratti (**4 nel 73% delle ore**). Il problema e' che
la scadenza del PRIMO oscilla fra 0 e 91 giorni (dente di sega deterministico).
Interpolare a maturita' costante richiede estrapolazione nel 41-70% delle ore
(misurato). Quindi la curva e' costruita in **spazio FORWARD**, che non ha il
problema:
b1 = ln(F1/S) * 365,25/dte1 carry spot->primo (denominatore
che si annulla a scadenza: rumoroso)
fwd12 = ln(F2/F1) * 365,25/(dte2-dte1) forward carry, finestra ~91g FISSA
fwd23, fwd34 idem
slope = fwd23 - fwd12 curvatura = fwd34 - 2*fwd23 + fwd12
Ogni forward carry e' misurato su una finestra di ~91 giorni SEMPRE, quindi non
c'e' denominatore che si annulla e non c'e' dente di sega nell'annualizzazione.
Cio' che resta di sega (la finestra del fwd12 PARTE a dte1, che oscilla) e' una
proprieta' vera della curva, e viene comunque controllata con un PLACEBO
esplicito (un segnale che e' funzione SOLA di dte1).
📌 Tutto in LOG del rapporto dei prezzi, mai medie aritmetiche di differenze:
la media aritmetica contiene (varA - varB)/2, che su questi dati vale
1,48-2,60%/anno = 2-4x l'effetto cercato (regola gia' pagata, §39).
===========================================================================
IL LAG — verificato, non assunto (§39: corr 0,996 al lag +1)
===========================================================================
`get_tradingview_chart_data` etichetta la barra all'INIZIO: il close della barra
etichettata T e' il prezzo a T+1h. `get_funding_rate_history` etichetta
l'indice all'ISTANTE T. Appaiarli senza lag sfasa di un'ora.
Il lag NON e' assunto: e' scelto misurando la sd del basis perp-vs-indice a
k = -2..+2 e la convergenza a scadenza. Vedi §0.
===========================================================================
DISCIPLINA
===========================================================================
* SOLA LETTURA. Nessuna rete: tutto da `data/raw/fut_deribit/` (persistito
dall'ondata 1). Nessun ordine, nessuna chiave.
* Nessun file di produzione toccato.
* Famiglia DICHIARATA PRIMA e contata AL RIALZO: 86 backtest + 5 regressioni.
Il deflated-Sharpe e' riportato sia per RAMO sia sull'UNIONE, perche' la
stessa griglia da' verdetti diversi a seconda di come la si partiziona (§10,
§24): la partizione e' un secondo posto dove barare e' indolore e invisibile.
USO
nice -n 19 timeout 900 uv run python scripts/research/r0823b_curva_datati.py
"""
from __future__ import annotations
import sys
import warnings
from pathlib import Path
import numpy as np
import pandas as pd
from scipy import stats
PROJECT_ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(PROJECT_ROOT))
sys.path.insert(0, str(PROJECT_ROOT / "scripts" / "research" / "alt"))
import altlib as al # noqa: E402
from src.strategies.trend_portfolio import CANONICAL, TrendPortfolio # noqa: E402
warnings.filterwarnings("ignore", category=RuntimeWarning)
DATA = PROJECT_ROOT / "data" / "raw" / "fut_deribit"
MONTHS = ["JAN", "FEB", "MAR", "APR", "MAY", "JUN",
"JUL", "AUG", "SEP", "OCT", "NOV", "DEC"]
YR = 365.25
ASSETS = ("BTC", "ETH")
HOLDOUT = al.HOLDOUT
SIG_HOUR = 22 # ora di lettura della curva: nota a 23:00 UTC, un'ora
# PRIMA della chiusura della barra giornaliera -> causale
# con margine. La sensibilita' all'ora e' misurata in §6.
# ---- famiglia DICHIARATA PRIMA (contata al rialzo) ------------------------
VARS_A = ("b1", "fwd12", "slope", "slope_o", "curv") # 5
ZWINS = (90, 180, 365) # 3
SIGNS = (+1, -1) # 2
MAPS = ("sgn", "lin") # 2 -> 60 celle
VARS_D = ("fwd12", "slope_o", "curv") # 3
PCTLS = (0.30, 0.50, 0.70) # 3
DIRS = (+1, -1) # 2 -> 18 celle
VARS_B = ("curv", "slope_o") # 2
ZWINS_B = (90, 365) # 2
SIGNS_B = (+1, -1) # 2 -> 8 celle
N_DECLARED_BT = 60 + 18 + 8 # 86 backtest
N_DECLARED_REG = 5 # 5 regressioni ramo C
FEE_SIDE = al.FEE_SIDE # 5 bps/lato: la convenzione dei backtest del progetto
def hr(t: str = "") -> None:
print("\n" + "=" * 92)
if t:
print(t)
print("=" * 92)
# ===========================================================================
# §0 DATO — caricamento, LAG verificato, certificazione
# ===========================================================================
def expiry_of(name: str) -> pd.Timestamp:
_, tag = name.split("-")
return pd.Timestamp(2000 + int(tag[5:]), MONTHS.index(tag[2:5]) + 1,
int(tag[:2]), 8, tz="UTC")
def load_contracts(asset: str):
closes, exps, vols = {}, {}, {}
for f in sorted(DATA.glob(f"{asset}-*.parquet")):
nm = f.stem
df = pd.read_parquet(f)
if len(df) < 200:
continue
s = df.set_index("ts")["close"].astype(float)
s = s[s > 0]
s.index = s.index.tz_convert("UTC")
v = df.set_index("ts")["volume"].astype(float)
v.index = v.index.tz_convert("UTC")
e = expiry_of(nm)
s = s[s.index < e - pd.Timedelta(hours=2)] # esclude l'ora di settlement
closes[nm], exps[nm], vols[nm] = s, e, v
return closes, exps, vols
def load_fund(asset: str) -> pd.DataFrame:
f = pd.read_parquet(DATA / f"fund_{asset}.parquet").set_index("ts").sort_index()
f.index = f.index.tz_convert("UTC")
return f
def lag_check(asset: str) -> int:
"""Sceglie il lag MISURANDOLO (§39: appaiare due serie dello stesso progetto con
convenzioni di etichettatura diverse e' un bug gia' pagato). Ritorna il lag in ore."""
fund = load_fund(asset)
idx = fund["index"].astype(float)
p = al.get(asset, "1h")
perp = pd.Series(p["close"].values.astype(float),
index=pd.DatetimeIndex(pd.to_datetime(p["datetime"], utc=True)))
out = {}
for k in (-2, -1, 0, 1, 2):
ii = idx.copy()
ii.index = ii.index - pd.Timedelta(hours=k)
j = pd.concat({"p": perp, "i": ii}, axis=1, join="inner").dropna()
b = np.log(j["p"] / j["i"]) * 1e4
out[k] = (len(j), float(b.median()), float(b.std()))
best = min(out, key=lambda k: out[k][2])
print(f" {asset}: basis perp-vs-indice per lag (sd = il criterio)")
for k, (n, m, s) in out.items():
mark = " <- SCELTO (sd minima)" if k == best else ""
print(f" lag {k:+d}h n={n:,} mediana {m:+7.2f} bps sd {s:8.2f} bps{mark}")
return best
def build_curve(asset: str, lag_h: int) -> pd.DataFrame:
closes, exps, vols = load_contracts(asset)
fund = load_fund(asset)
idx = fund["index"].astype(float)
f1h = fund["f1h"].astype(float)
wide = pd.DataFrame(closes).sort_index()
S = idx.reindex(wide.index + pd.Timedelta(hours=lag_h))
S.index = wide.index
keep = S.notna()
wide, S = wide[keep], S[keep]
DTE = pd.DataFrame({nm: (exps[nm] - wide.index).total_seconds() / 86400.0
for nm in wide.columns}, index=wide.index).where(wide.notna())
dt_v, px_v, sv = DTE.values, wide.values, S.values
n = len(wide)
out = np.full((n, 6), np.nan)
for i in range(n):
m = np.isfinite(dt_v[i]) & np.isfinite(px_v[i]) & (dt_v[i] > 0.5)
d, p = dt_v[i][m], px_v[i][m]
o = np.argsort(d)
d, p = d[o], p[o]
k = len(d)
out[i, 5] = k
if k >= 1:
out[i, 0] = d[0]
out[i, 1] = np.log(p[0] / sv[i]) * YR / d[0]
if k >= 2:
out[i, 2] = np.log(p[1] / p[0]) * YR / (d[1] - d[0])
if k >= 3:
out[i, 3] = np.log(p[2] / p[1]) * YR / (d[2] - d[1])
if k >= 4:
out[i, 4] = np.log(p[3] / p[2]) * YR / (d[3] - d[2])
C = pd.DataFrame(out, index=wide.index,
columns=["dte1", "b1", "fwd12", "fwd23", "fwd34", "nlive"])
C["slope"] = C["fwd23"] - C["fwd12"]
C["curv"] = C["fwd34"] - 2 * C["fwd23"] + C["fwd12"]
C["fund24"] = f1h.reindex(C.index).rolling(24).mean() * 24 * YR
C["S"] = S
return C
def causal_orth(y: pd.Series, x: pd.Series, min_obs: int = 180) -> pd.Series:
"""Residuo di y su x con beta stimato SOLO sul passato (media espandente, shift 1).
Se il beta usasse tutto il campione sarebbe look-ahead che `causality_ok` non vede."""
j = pd.concat({"y": y, "x": x}, axis=1)
v = j.notna().all(axis=1)
yy = j["y"].where(v)
xx = j["x"].where(v)
n = v.astype(float).cumsum().shift(1)
sx = xx.fillna(0).cumsum().shift(1)
sy = yy.fillna(0).cumsum().shift(1)
sxx = (xx.fillna(0) ** 2).cumsum().shift(1)
sxy = (xx.fillna(0) * yy.fillna(0)).cumsum().shift(1)
cov = sxy / n - (sx / n) * (sy / n)
var = sxx / n - (sx / n) ** 2
beta = (cov / var).where(n >= min_obs)
alpha = (sy / n) - beta * (sx / n)
return (yy - (alpha + beta * xx)).where(n >= min_obs)
def daily_curve(asset: str, C: pd.DataFrame, hour: int = SIG_HOUR) -> pd.DataFrame:
d = C[C.index.hour == hour].copy()
d.index = d.index.normalize()
d = d[~d.index.duplicated(keep="last")]
d["slope_o"] = causal_orth(d["slope"], d["fwd12"])
return d
# ===========================================================================
# SEGNALI E BACKTEST
# ===========================================================================
def zscore(s: pd.Series, w: int) -> pd.Series:
m = s.rolling(w, min_periods=max(30, w // 2)).mean()
sd = s.rolling(w, min_periods=max(30, w // 2)).std()
return ((s - m) / sd).replace([np.inf, -np.inf], np.nan)
def direction_from(z: pd.Series, sign: int, mp: str) -> pd.Series:
if mp == "sgn":
d = np.sign(z)
else:
d = np.clip(z / 2.0, -1.0, 1.0)
return (sign * d).fillna(0.0)
def _align_to_bars(df: pd.DataFrame, s: pd.Series, fill: float = 0.0) -> np.ndarray:
idx = pd.DatetimeIndex(pd.to_datetime(df["datetime"], utc=True))
return s.reindex(idx).fillna(fill).values.astype(float)
def _ms(idx: pd.DatetimeIndex) -> np.ndarray:
"""Epoca in MILLISECONDI, esplicita.
⚠️ `idx.view("int64")` NON e' sicuro: i parquet dei datati hanno risoluzione
`datetime64[ms]` e `al.get` `[ns]` -> la stessa riga di codice ritorna ms in un
posto e ns nell'altro, senza eccezione e senza NaN. E' la famiglia di bug
codificata il 01/07 (`.view("int64")` su indici tz-aware non-ns) e ripagata in
§35. Qui e' stata ripagata una TERZA volta, da me, e l'ha resa visibile un
`n=0` — non un numero sbagliato."""
return (pd.DatetimeIndex(idx).tz_convert("UTC").as_unit("ns")
.astype("int64") // 10 ** 6)
def expanding_pctl(s: pd.Series, min_obs: int = 180) -> pd.Series:
"""Percentile ESPANDENTE causale: quota delle osservazioni STRETTAMENTE
precedenti che stanno sotto quella corrente. O(n log n) con bisect."""
import bisect
vals, seen, out = s.values.astype(float), [], np.full(len(s), np.nan)
for i, v in enumerate(vals):
if np.isfinite(v):
if len(seen) >= min_obs:
out[i] = bisect.bisect_right(seen, v) / len(seen)
bisect.insort(seen, v)
return pd.Series(out, index=s.index)
def cell_daily_A(dcur: dict, var: str, w: int, sign: int, mp: str,
fee_side: float = FEE_SIDE) -> pd.Series:
"""Serie giornaliera netta 50/50 BTC+ETH del ramo A (stessa convenzione di
`altlib.candidate_daily`: eval_weights fa lo shift, non si puo' fare leak)."""
ser = {}
for a in ASSETS:
df = al.get(a, "1d")
d = dcur[a]
dirn = direction_from(zscore(d[var], w), sign, mp)
tgt = al.vol_target(_align_to_bars(df, dirn), df, target_vol=0.20,
vol_win_days=30, leverage_cap=2.0)
ev = al.eval_weights(df, tgt, fee_side=fee_side)
ser[a] = pd.Series(ev["net"], index=ev["idx"])
J = pd.concat(ser, axis=1, join="inner").fillna(0.0)
return al._to_daily(0.5 * J[ASSETS[0]] + 0.5 * J[ASSETS[1]])
def tp01_targets() -> dict:
tp = TrendPortfolio(**CANONICAL)
return {a: tp.target_series(al.get(a, "1d")) for a in ASSETS}
def cell_daily_D(dcur: dict, tgts: dict, var: str, pctl: float, dirn: int,
g_off: float = 0.0, fee_side: float = FEE_SIDE,
t0: pd.Timestamp | None = None) -> pd.Series:
"""TP01 moltiplicato per un gate di curva. Il percentile e' ESPANDENTE e causale."""
ser = {}
for a in ASSETS:
df = al.get(a, "1d")
d = dcur[a]
rk = expanding_pctl(d[var])
on = (rk >= pctl) if dirn > 0 else (rk <= pctl)
g = pd.Series(np.where(on.fillna(False), 1.0, g_off), index=d.index)
g[rk.isna()] = 1.0 # prima del warm-up: gate INATTIVO
# ⚠️ fill=1.0, NON 0.0: fuori dalla finestra della curva il gate non esiste,
# e riempire a zero spegnerebbe TP01 su meta' della sua storia. E' il difetto
# che il controllo d'identita' ha catturato (max|diff| 2,4e-02, non 0).
tgt = tgts[a] * _align_to_bars(df, g, fill=1.0)
ev = al.eval_weights(df, tgt, fee_side=fee_side)
ser[a] = pd.Series(ev["net"], index=ev["idx"])
J = pd.concat(ser, axis=1, join="inner").fillna(0.0)
out = al._to_daily(0.5 * J[ASSETS[0]] + 0.5 * J[ASSETS[1]])
return out[out.index >= t0] if t0 is not None else out
def beta_baseline(fee_side: float = FEE_SIDE) -> pd.Series:
"""LONG SEMPRE, vol-targeted 20% con lo stesso motore: e' il benchmark che il
ramo A deve battere. Senza di lui una cella 'quasi sempre lunga' su un campione
2019-2026 con due tori sembra un segnale e non lo e'."""
ser = {}
for a in ASSETS:
df = al.get(a, "1d")
tgt = al.vol_target(np.ones(len(df)), df, target_vol=0.20,
vol_win_days=30, leverage_cap=2.0)
ev = al.eval_weights(df, tgt, fee_side=fee_side)
ser[a] = pd.Series(ev["net"], index=ev["idx"])
J = pd.concat(ser, axis=1, join="inner").fillna(0.0)
return al._to_daily(0.5 * J[ASSETS[0]] + 0.5 * J[ASSETS[1]])
def mean_pos(dcur: dict, var: str, w: int, sign: int, mp: str) -> float:
"""Esposizione media della cella: se e' ~1 la cella e' un LONG travestito."""
v = []
for a in ASSETS:
df = al.get(a, "1d")
dirn = direction_from(zscore(dcur[a][var], w), sign, mp)
v.append(float(np.mean(_align_to_bars(df, dirn))))
return float(np.mean(v))
def sh(s) -> float:
return al._sh(s)
def dd(s) -> float:
return al._dd_ret(s)
def split(s: pd.Series):
return s[s.index < HOLDOUT], s[s.index >= HOLDOUT]
def per_year(s: pd.Series) -> dict:
return {int(y): round(sh(g), 2) for y, g in s.groupby(s.index.year) if len(g) > 60}
# ===========================================================================
def main() -> None:
print(__doc__.split("USO")[0])
# ---------------------------------------------------------------- §0
hr("§0 IL DATO — lag MISURATO (non assunto), curva, certificazione")
lags = {}
for a in ASSETS:
lags[a] = lag_check(a)
lag = lags["BTC"]
print(f"\n -> lag adottato: +{lag}h su ENTRAMBI gli asset "
f"(BTC {lags['BTC']:+d}, ETH {lags['ETH']:+d})")
curves, dcur = {}, {}
for a in ASSETS:
C = build_curve(a, lag)
curves[a] = C
dcur[a] = daily_curve(a, C)
nl = C["nlive"].value_counts().sort_index()
print(f"\n {a}: {len(C):,} ore {C.index.min().date()} .. {C.index.max().date()}"
f" ({(C.index.max()-C.index.min()).days/365.25:.1f} anni)")
print(f" contratti vivi/ora: " +
" ".join(f"{int(k)}:{v/len(C)*100:.0f}%" for k, v in nl.items()))
print(f" giorni con curva (ora {SIG_HOUR:02d}): {len(dcur[a]):,}")
print("\n CONVERGENZA A SCADENZA |ln(F/indice)| ultime 4h — col lag e senza")
for a in ASSETS:
closes, exps, _ = load_contracts(a)
idx = load_fund(a)["index"].astype(float)
for k in (0, lag):
v = []
for nm, s in closes.items():
tail = s[s.index >= exps[nm] - pd.Timedelta(hours=6)]
if not len(tail):
continue
ii = idx.reindex(tail.index + pd.Timedelta(hours=k))
x = np.abs(np.log(tail.values / ii.values)) * 1e4
x = x[np.isfinite(x)]
if len(x):
v.append(float(np.median(x)))
print(f" {a} lag +{k}h: mediana {np.median(v):5.1f} bps (n={len(v)})")
print(" ⚠️ §13 pubblico' 33-47 bps: era calcolato SENZA il lag (merge_asof 'nearest').")
print(" Correzione a un numero pubblicato: il dato e' migliore di come e' stato descritto.")
# ---------------------------------------------------------------- §1
hr("§1 LA FORMA DELLA CURVA — livello, pendenza, curvatura (mai guardate prima)")
print(" Tutte le grandezze sono carry ANNUALIZZATI in log; fwdXY = forward carry")
print(" su finestra ~91g FISSA (niente denominatore che si annulla).\n")
for a in ASSETS:
C = curves[a]
print(f" --- {a} ---")
print(" var n mediana p05 p95 sd")
for c in ("b1", "fwd12", "fwd23", "fwd34", "slope", "curv", "fund24"):
s = C[c].dropna()
print(f" {c:9s} {len(s):7,} {s.median()*100:8.2f}% {s.quantile(.05)*100:9.2f}% "
f"{s.quantile(.95)*100:9.2f}% {s.std()*100:8.2f}%")
# ---------------------------------------------------------------- §2
hr("§2 PRIOR A1/A2 — la curva e' funding travestito? (si misura PRIMA di backtestare)")
print(" Attesa dichiarata: LIVELLO si (|corr|>0,5), FORMA no.\n")
corr_tab = {}
for a in ASSETS:
d = dcur[a]
print(f" --- {a} (giornaliero, n={len(d):,}) ---")
for c in ("b1", "fwd12", "fwd23", "fwd34", "slope", "slope_o", "curv"):
j = d[[c, "fund24"]].dropna()
r = float(j[c].corr(j["fund24"]))
corr_tab[(a, c)] = r
tag = " <- LIVELLO: funding travestito" if abs(r) > 0.5 else ""
print(f" corr({c:8s}, funding24h) = {r:+.3f} n={len(j):,}{tag}")
j = d[["slope", "fwd12"]].dropna()
print(f" [contorno] corr(slope, livello) = {j['slope'].corr(j['fwd12']):+.3f}"
f" -> la pendenza GREZZA e' in gran parte il livello col segno rovesciato")
print("\n CONTROLLO POSITIVO del rilevatore di correlazione (un rilevatore che non trova")
print(" nulla e' indistinguibile da uno rotto): rumore bianco vs funding, e funding vs se stesso")
rng = np.random.default_rng(20260823)
d = dcur["BTC"]
noise = pd.Series(rng.normal(size=len(d)), index=d.index)
j = pd.concat({"n": noise, "f": d["fund24"]}, axis=1).dropna()
print(f" corr(rumore, funding) = {j['n'].corr(j['f']):+.3f} (atteso ~0)")
print(f" corr(funding, funding) = {1.0:+.3f} (atteso +1)")
print("\n 📌 ESITO: A1 CONFERMATA (b1 +0,6 su entrambi), A2 CONFERMATA — e in modo piu'")
print(" forte dell'atteso: la pendenza ORTOGONALIZZATA al livello e la curvatura")
print(" sono praticamente scorrelate dal funding. **La forma e' un grado di")
print(" liberta' nuovo: il ramo non e' chiuso in partenza.**")
# ---------------------------------------------------------------- §3
hr("§3 PRE-TEST DI REGRESSIONE — e il suo CONTROLLO POSITIVO, che FALLISCE")
print(" Regressione dei rendimenti spot futuri sulla variabile di curva (z-unit).")
print(" t corretto grossolanamente per sovrapposizione: t_naive/sqrt(h).\n")
reg_rows = []
for a in ASSETS:
d = dcur[a]
lp = np.log(d["S"])
print(f" --- {a} ---")
print(" var h=5g h=21g h=63g")
for v in list(VARS_A) + ["mom90(CTRL+)"]:
if v.startswith("mom90"):
x0 = lp - lp.shift(90)
else:
x0 = d[v]
row = f" {v:11s}"
for h in (5, 21, 63):
fw = lp.shift(-h) - lp
jj = pd.concat([x0, fw], axis=1).dropna()
jj.columns = ["x", "y"]
if len(jj) < 100 or jj["x"].std() == 0:
row += " n/a "
continue
xz = (jj["x"] - jj["x"].mean()) / jj["x"].std()
sl, ic, rr, pv, se = stats.linregress(xz, jj["y"])
t = sl / se / np.sqrt(h)
row += f" {sl*100:+6.2f}%(t{t:+5.2f})"
if not v.startswith("mom90"):
reg_rows.append(dict(asset=a, var=v, h=h, beta=sl, t=t))
print(row)
print("\n 🚨 IL CONTROLLO POSITIVO NON PASSA: il momentum a 90 giorni — il segnale che")
print(" REGGE il 75% del libro live (TP01, Sharpe FULL 1,31 = t~3,6 su 7,4 anni) —")
print(" in questa lente da |t| 0,4-1,4. **La regressione predittiva univariata su")
print(" rendimenti crypto NON HA POTENZA a questo campione**: un t basso qui non e'")
print(" evidenza di assenza. -> il ramo si giudica sui BACKTEST (§4-§5), non su questi t.")
print(" (Le regressioni restano a registro: 5 variabili x 3 orizzonti x 2 asset.)")
# ---------------------------------------------------------------- §4
hr("§4 RAMO A — la curva come segnale DIREZIONALE (60 celle dichiarate)")
B = al.tp01_baseline_daily()
print(f" ✅ REPLICA prima di produrre numeri nuovi: TP01 baseline 50/50")
print(f" Sharpe FULL {sh(B):.3f} (pubblicato ~1,30) HOLD {sh(B[B.index>=HOLDOUT]):.3f}")
print(f" ⚠️ l'hold-out pubblicato e' ~0,31-0,44 a seconda del VINTAGE di data/raw/")
print(f" (gitignored, riscritto ogni notte): la terza cifra non e' confrontabile.\n")
BB = beta_baseline()
bi, bh = split(BB)
print(f" BENCHMARK BETA (long sempre, vol-target 20%, stesso motore):")
print(f" Sharpe IS {sh(bi):+.3f} FULL {sh(BB):+.3f} HOLD {sh(bh):+.3f}"
f" maxDD {dd(BB)*100:.1f}%")
print(" ⚠️ e' IL benchmark del ramo A: su 2019-2026 (due tori) una cella che sta")
print(" long quasi sempre eredita questo Sharpe senza contenere informazione.\n")
rowsA = []
for var in VARS_A:
for w in ZWINS:
for sg in SIGNS:
for mp in MAPS:
s = cell_daily_A(dcur, var, w, sg, mp)
ins, hol = split(s)
rowsA.append(dict(var=var, w=w, sign=sg, map=mp, daily=s,
is_sh=sh(ins) if len(ins) > 60 else np.nan,
full=sh(s), hold=sh(hol) if len(hol) > 60 else np.nan,
dd=dd(s)))
assert len(rowsA) == 60, len(rowsA)
print(f" celle valutate: {len(rowsA)} (dichiarate 60)")
validA = [r for r in rowsA if np.isfinite(r["is_sh"])]
chosenA = max(validA, key=lambda r: r["is_sh"])
print(f" distribuzione dei 60 Sharpe FULL: mediana {np.median([r['full'] for r in rowsA]):+.3f}"
f" min {min(r['full'] for r in rowsA):+.3f} max {max(r['full'] for r in rowsA):+.3f}")
print(f" frazione di celle con FULL>0: {np.mean([r['full']>0 for r in rowsA]):.0%}"
f" con HOLD>0: {np.mean([r['hold']>0 for r in rowsA]):.0%}")
print("\n CELLA SCELTA IN-SAMPLE-ONLY (nessuna occhiata all'hold-out):")
print(f" {chosenA['var']} w={chosenA['w']} sign={chosenA['sign']:+d} map={chosenA['map']}"
f" Sh IS {chosenA['is_sh']:+.3f} FULL {chosenA['full']:+.3f}"
f" HOLD {chosenA['hold']:+.3f} maxDD {chosenA['dd']*100:.1f}%")
best_hold = max(validA, key=lambda r: r["hold"])
print(f" (per contrasto, la cella a MASSIMO hold-out — che NON e' selezionabile —:")
print(f" {best_hold['var']} w={best_hold['w']} sign={best_hold['sign']:+d} "
f"map={best_hold['map']} IS {best_hold['is_sh']:+.3f} HOLD {best_hold['hold']:+.3f})")
print("\n top-8 per Sharpe IN-SAMPLE (pos = esposizione media: ~1 => LONG travestito)")
print(" var w sgn map ShIS ShFULL ShHOLD maxDD pos")
for r in sorted(validA, key=lambda r: r["is_sh"], reverse=True)[:8]:
mp_ = mean_pos(dcur, r["var"], r["w"], r["sign"], r["map"])
print(f" {r['var']:9s} {r['w']:4d} {r['sign']:+2d} {r['map']:4s}"
f" {r['is_sh']:+7.3f} {r['full']:+8.3f} {r['hold']:+7.3f} {r['dd']*100:7.1f}%"
f" {mp_:+6.2f}")
print(f" 🚨 NESSUNA delle 60 celle batte il benchmark beta ({sh(BB):+.3f} FULL): la")
print(f" migliore in-sample fa {chosenA['full']:+.3f}, la migliore in assoluto "
f"{max(r['full'] for r in rowsA):+.3f}.")
allA = [r["full"] for r in rowsA]
dsrA, sr0A = al.deflated_sharpe(chosenA["full"], allA, chosenA["daily"])
print(f"\n deflated-Sharpe (ramo A, N=60): {dsrA:.3f} "
f"(massimo atteso dal PURO RUMORE {sr0A:+.3f}) "
f"{'PASS' if dsrA >= 0.95 else 'FAIL'}")
mgA = al.marginal_vs_tp01(chosenA["daily"])
print(f" marginal_vs_tp01: {mgA['marginal_verdict']} corr {mgA['corr_full']:+.3f} "
f"beta {mgA['beta_to_tp01']:+.3f} alpha {mgA['alpha_ann']*100:+.2f}%/anno")
print(f" uplift w25 FULL {mgA['blends']['w25']['uplift_full']:+.3f} "
f"HOLD {mgA['blends']['w25']['uplift_hold']:+.3f} "
f"robust_oos {mgA['robust_oos']} has_insample_edge {mgA['has_insample_edge']} "
f"is_hedge {mgA['is_hedge']}")
earnsA = bool(mgA["marginal_verdict"] == "ADDS" and dsrA >= 0.95)
print(f" -> earns_slot_honest (ramo A) = {earnsA}")
print("\n PER ANNO della cella scelta (A4: un aggregato si scompone prima di crederci):")
py = per_year(chosenA["daily"])
print(" " + " ".join(f"{y}:{v:+.2f}" for y, v in py.items()))
pos = sum(1 for v in py.values() if v > 0)
print(f" anni positivi {pos}/{len(py)} 2022 (l'unico stress del campione): "
f"{py.get(2022, float('nan')):+.2f}")
print("\n PLACEBO DEL DENTE DI SEGA (il rischio strutturale di questo dataset):")
print(" un segnale che e' funzione SOLA di dte1 (giorni alla prima scadenza) non")
print(" contiene informazione di curva. Se battesse la cella scelta, il ramo A")
print(" starebbe misurando il calendario dei roll, non la forma della curva.")
dsaw = {a: dcur[a].assign(**{"__saw": dcur[a]["dte1"]}) for a in ASSETS}
for w in ZWINS:
s = cell_daily_A(dsaw, "__saw", w, +1, "lin")
s2 = cell_daily_A(dsaw, "__saw", w, -1, "lin")
print(f" dte1 w={w:3d}: Sh FULL {sh(s):+.3f} / {sh(s2):+.3f} (segno +/-)"
f" pos {mean_pos(dsaw, '__saw', w, +1, 'lin'):+.2f}")
print(" 🚨 IL PLACEBO BATTE OGNI CELLA DI CURVA — e la diagnosi e' il contrario di")
print(" quella che sembra. I due segni sono SPECCHI quasi esatti e l'esposizione")
print(" media e' ~+1: il placebo non sta misurando il dente di sega, sta stando")
print(f" LONG, e riproduce il benchmark beta ({sh(BB):+.3f}). Cio' che dimostra e'")
print(" piu' severo: **nessun segnale di curva batte 'stai lungo e basta'** su")
print(" questo campione. Un placebo va letto sulla sua ESPOSIZIONE prima che sul")
print(" suo Sharpe, o si conclude l'opposto.")
# ---------------------------------------------------------------- §5
hr("§5 RAMO D — la curva come GATE DI REGIME su TP01 (18 celle dichiarate)")
tgts = tp01_targets()
T0 = max(dcur[a].index.min() for a in ASSETS)
base_gate = cell_daily_D(dcur, tgts, "fwd12", 0.0, +1) # gate sempre acceso
print(f" ✅ CONTROLLO DI IDENTITA': gate sempre acceso vs TP01 baseline sulla finestra comune")
j = pd.concat({"g": base_gate, "b": B}, axis=1, join="inner").dropna()
print(f" max|diff| = {float((j['g']-j['b']).abs().max()):.2e} "
f"Sh {sh(j['g']):.4f} vs {sh(j['b']):.4f}")
ref = base_gate[base_gate.index >= T0]
rowsD = []
for var in VARS_D:
for p in PCTLS:
for dr in DIRS:
s = cell_daily_D(dcur, tgts, var, p, dr, t0=T0)
ins, hol = split(s)
rowsD.append(dict(var=var, p=p, dir=dr, daily=s,
is_sh=sh(ins), full=sh(s), hold=sh(hol), dd=dd(s)))
assert len(rowsD) == 18, len(rowsD)
chosenD = max(rowsD, key=lambda r: r["is_sh"])
ri, rh = split(ref)
print(f"\n riferimento (TP01 sulla finestra della curva): IS {sh(ri):+.3f} "
f"FULL {sh(ref):+.3f} HOLD {sh(rh):+.3f} maxDD {dd(ref)*100:.1f}%")
print(" celle (18): var pctl dir ShIS ShFULL ShHOLD maxDD dShFULL")
for r in sorted(rowsD, key=lambda r: r["is_sh"], reverse=True):
mark = " <- scelta IS" if r is chosenD else ""
print(f" {r['var']:9s} {r['p']:.2f} {r['dir']:+d} {r['is_sh']:+7.3f}"
f" {r['full']:+8.3f} {r['hold']:+7.3f} {r['dd']*100:7.1f}% "
f"{r['full']-sh(ref):+8.3f}{mark}")
allD = [r["full"] for r in rowsD]
dsrD, sr0D = al.deflated_sharpe(chosenD["full"], allD, chosenD["daily"])
print(f"\n deflated-Sharpe (ramo D, N=18): {dsrD:.3f} (rumore max atteso {sr0D:+.3f}) "
f"{'PASS' if dsrD >= 0.95 else 'FAIL'}")
print("\n 🚨 NULL DEL DE-LEVERING (7 occorrenze pagate dal progetto): un gate SPEGNE")
print(" esposizione, quindi va confrontato con un semplice k*TP01 a PARI maxDD.")
tgt_dd = chosenD["dd"]
lo, hi = 0.05, 1.0
for _ in range(40):
k = (lo + hi) / 2
if dd(k * ref) > tgt_dd:
hi = k
else:
lo = k
k = (lo + hi) / 2
print(f" gate scelto: maxDD {tgt_dd*100:.2f}% ShFULL {chosenD['full']:+.3f} "
f"ShHOLD {chosenD['hold']:+.3f}")
print(f" k*TP01 a pari maxDD (k={k:.3f}): maxDD {dd(k*ref)*100:.2f}% "
f"ShFULL {sh(k*ref):+.3f} ShHOLD {sh((k*ref)[(k*ref).index>=HOLDOUT]):+.3f}")
beats = chosenD["full"] > sh(k * ref) + 1e-9
print(f" -> il gate batte il de-levering puro? {beats}")
print(" (nota: lo Sharpe e' invariante alla scala, quindi il de-levering NON puo'")
print(" cambiare lo Sharpe: il confronto giusto e' 'a pari DD lo Sharpe migliora?'")
print(" -> qui la domanda si riduce a dShFULL > 0, e il null e' DEGENERE (§24 (d)).")
print(" Il fatto che conta e' che il gate cambia lo Sharpe, non che riduca il DD.")
print("\n PER ANNO della cella D scelta, contro TP01:")
pyD, pyR = per_year(chosenD["daily"]), per_year(ref)
for y in sorted(set(pyD) | set(pyR)):
print(f" {y}: gate {pyD.get(y, float('nan')):+6.2f} TP01 {pyR.get(y, float('nan')):+6.2f}"
f" d {pyD.get(y,np.nan)-pyR.get(y,np.nan):+6.2f}")
# ---------------------------------------------------------------- §6
hr("§6 SENSIBILITA' ALL'ORA DI LETTURA (robustezza alla SCELTA, non un IC — §57)")
print(" 24 ore di lettura della curva, stessa griglia giornaliera 00:00. Tutte causali.")
bandA, bandD = [], []
for h in range(24):
dc = {a: daily_curve(a, curves[a], hour=h) for a in ASSETS}
bandA.append(sh(cell_daily_A(dc, chosenA["var"], chosenA["w"],
chosenA["sign"], chosenA["map"])))
bandD.append(sh(cell_daily_D(dc, tgts, chosenD["var"], chosenD["p"],
chosenD["dir"], t0=T0)))
for nm, band, canon in (("A", bandA, chosenA["full"]), ("D", bandD, chosenD["full"])):
b = np.array(band)
print(f" ramo {nm}: canonica(h={SIG_HOUR}) {canon:+.3f} mediana {np.median(b):+.3f}"
f" p10 {np.percentile(b,10):+.3f} p90 {np.percentile(b,90):+.3f}"
f" pctl della canonica {float((b<=canon).mean())*100:.0f}°")
bD = np.array(bandD) - sh(ref)
print(f" ramo D, dShFULL vs TP01 su 24 ore: mediana {np.median(bD):+.3f}, "
f"positivo in {int((bD>0).sum())}/24 (⚠️ ~2 osservazioni indipendenti, §57)")
# ---------------------------------------------------------------- §7
hr("§7 RAMO B — la BUTTERFLY delta-neutra (8 celle) e il muro dell'MDE")
print(" F1 - 2*F2 + F3 a nozionale netto ZERO: e' esattamente la CURVATURA, quotata.")
print(" §13 aveva testato il calendar spread (F1 vs F2, 108 celle); la butterfly e'")
print(" il secondo ordine e non era mai stata guardata.\n")
fly = {}
for a in ASSETS:
C = curves[a]
closes, exps, _ = load_contracts(a)
wide = pd.DataFrame(closes).sort_index().reindex(C.index)
DTEa = pd.DataFrame({nm: (exps[nm] - wide.index).total_seconds() / 86400.0
for nm in wide.columns}, index=wide.index).where(wide.notna())
lr = np.log(wide).diff()
dv, lv = DTEa.values, lr.values
pv = wide.values
n = len(wide)
out = np.full(n, np.nan)
turn = np.zeros(n)
prev = None
for i in range(n):
m = np.isfinite(dv[i]) & np.isfinite(pv[i]) & (dv[i] > 0.5)
ordr = np.argsort(np.where(m, dv[i], np.inf))
sel = ordr[:3]
if not np.all(m[sel]) or len(sel) < 3:
prev = None
continue
wts = np.array([1.0, -2.0, 1.0])
if prev is not None and np.array_equal(prev, sel):
r = np.nansum(wts * lv[i][sel])
out[i] = r
else:
turn[i] = 4.0 # 4 gambe di nozionale unitario cambiate (1+2+1)
prev = sel
fly[a] = pd.Series(out, index=wide.index), pd.Series(turn, index=wide.index)
print(" rendimento LORDO della butterfly (nozionale lordo 4x, hold continuo):")
for a in ASSETS:
r, t = fly[a]
rd = r.resample("1D").sum().dropna()
yrs = (rd.index[-1] - rd.index[0]).days / YR
mu = float(rd.mean() * YR)
sd = float(rd.std() * np.sqrt(YR))
se = sd / np.sqrt(yrs)
n_roll = int((t > 0).sum())
print(f" {a}: drift {mu*100:+6.2f}%/anno vol {sd*100:5.2f}% Sharpe {mu/sd:+.2f}"
f" SE(drift) {se*100:.2f}pp roll {n_roll} in {yrs:.1f} anni")
print(f" 🚨 MDE: serve |drift| > {1.96*se*100:.2f}%/anno per distinguerlo da zero.")
rowsB = []
for var in VARS_B:
for w in ZWINS_B:
for sg in SIGNS_B:
ser, ser0 = {}, {}
for a in ASSETS:
r, t = fly[a]
d = dcur[a]
z = zscore(d[var], w)
pos = (sg * np.clip(z / 2, -1, 1)).fillna(0.0)
rd = r.resample("1D").sum()
td = t.resample("1D").sum()
idx = rd.index.intersection(pos.index)
p = pos.reindex(idx).shift(1).fillna(0.0)
gross = p * rd.reindex(idx)
cost = (p.abs() * td.reindex(idx).fillna(0.0) +
4.0 * p.diff().abs().fillna(0.0))
ser[a] = gross - FEE_SIDE * cost
ser0[a] = gross
J = pd.concat(ser, axis=1, join="inner").fillna(0.0)
J0 = pd.concat(ser0, axis=1, join="inner").fillna(0.0)
s = 0.5 * J[ASSETS[0]] + 0.5 * J[ASSETS[1]]
s0 = 0.5 * J0[ASSETS[0]] + 0.5 * J0[ASSETS[1]]
ins, hol = split(s)
rowsB.append(dict(var=var, w=w, sign=sg, daily=s, is_sh=sh(ins),
full=sh(s), hold=sh(hol), gross=sh(s0)))
assert len(rowsB) == 8, len(rowsB)
chosenB = max(rowsB, key=lambda r: r["is_sh"])
print("\n celle (8): var w sgn Sh LORDO ShIS ShFULL ShHOLD")
for r in sorted(rowsB, key=lambda r: r["is_sh"], reverse=True):
mark = " <- scelta IS" if r is chosenB else ""
print(f" {r['var']:9s} {r['w']:4d} {r['sign']:+d} {r['gross']:+9.3f}"
f" {r['is_sh']:+7.3f} {r['full']:+8.3f} {r['hold']:+7.3f}{mark}")
gmax = max(r["gross"] for r in rowsB)
print(f"\n 📌 il MIGLIOR lordo della famiglia e' {gmax:+.3f}: sotto l'MDE (Sharpe 0,72)")
print(" ANCHE a fee zero. Quindi NON e' morte-per-fee — la fee la seppellisce, ma")
print(" sotto non c'e' un edge lordo da salvare. (Distinzione che il progetto fa")
print(" sistematicamente: §42 TP01-LS, §7 CRT.)")
# ---------------------------------------------------------------- §8
hr("§8 RAMO C — la curva predice la VOLATILITA' futura? (5 regressioni)")
print(" RV30 futura su curva, CONTROLLANDO RV30 corrente e DVOL (che e' la previsione")
print(" di mercato). Senza i controlli si misura solo 'la vol e' persistente'.\n")
for a in ASSETS:
d = dcur[a]
lp = np.log(d["S"])
r = lp.diff()
rv = r.rolling(30).std() * np.sqrt(YR)
fut = rv.shift(-30)
ts = pd.DataFrame({"timestamp": _ms(d.index)})
dv = pd.Series(al.dvol(ts, a), index=d.index) / 100.0
assert dv.notna().sum() > 500 and 0.2 < float(dv.median()) < 1.5, (
"controllo positivo DVOL fallito: merge vuoto o fuori scala "
f"(n={int(dv.notna().sum())}, mediana={float(dv.median()):.3f})")
base = pd.concat({"y": fut, "rv": rv, "dv": dv}, axis=1).dropna()
print(f" --- {a} (n={len(base):,}, dal {base.index.min().date()}: DVOL parte 2021-03) ---")
for v in VARS_A:
jj = pd.concat([base, d[v].rename("x")], axis=1).dropna()
if len(jj) < 200:
print(f" {v:9s} n troppo piccolo")
continue
X = np.column_stack([np.ones(len(jj)), jj["rv"], jj["dv"],
(jj["x"] - jj["x"].mean()) / jj["x"].std()])
bb, *_ = np.linalg.lstsq(X, jj["y"].values, rcond=None)
res = jj["y"].values - X @ bb
s2 = res @ res / (len(jj) - 4)
se = np.sqrt(np.diag(s2 * np.linalg.pinv(X.T @ X)))[3]
print(f" {v:9s} beta {bb[3]*100:+6.2f}pp di vol per 1sd "
f"t {bb[3]/se:+6.2f} (naive) t/sqrt(30) {bb[3]/se/np.sqrt(30):+5.2f}")
print("\n ⚠️ stesso limite di §3: nessun controllo positivo passa a questa risoluzione,")
print(" quindi queste righe sono descrittive, non decisive.")
# ---------------------------------------------------------------- §9
hr("§9 DEFLATED-SHARPE SULL'UNIONE, e la sua dipendenza dalla PARTIZIONE (§10/§24)")
allBT = allA + allD + [r["full"] for r in rowsB]
print(f" backtest dichiarati: {N_DECLARED_BT}, valutati: {len(allBT)} "
f"(+{N_DECLARED_REG} regressioni ramo C)")
for nm, pool, ch in (("ramo A (60)", allA, chosenA), ("ramo D (18)", allD, chosenD),
("ramo B (8)", [r["full"] for r in rowsB], chosenB),
("UNIONE (86)", allBT, chosenA)):
dsr, sr0 = al.deflated_sharpe(ch["full"], pool, ch["daily"])
print(f" {nm:12s} N={len(pool):3d} sd(Sharpe di griglia) {np.std(pool, ddof=1):.3f}"
f" max atteso dal rumore {sr0:+.3f} DSR {dsr:.3f} "
f"{'PASS' if dsr >= 0.95 else 'FAIL'}")
print(" 📌 La stessa griglia da' verdetti diversi a seconda di come la si partiziona:")
print(" e' il secondo posto dove barare e' indolore e invisibile (dopo il conteggio).")
# ---------------------------------------------------------------- §10
hr("§10 ESEGUIBILITA' — il roll trimestrale, e il muro che nessun backtest supera")
lb = Path("/tmp/claude-1001/-opt-docker-PythagorasGoal/"
"b6cc75e7-14f8-4c32-bd07-ab8a0d2aaee6/scratchpad/basis_cache/live_books.parquet")
if lb.exists():
d = pd.read_parquet(lb)
print(" spread bid-ask VIVI (fotografia del 22/08, non storia — va detto):")
for _, r in d.iterrows():
print(f" {r['instrument']:16s} spread {r['spread_bps']:6.2f} bps"
f" mezzo-spread {r['spread_bps']/2:5.2f} bps")
q = d[d.instrument.str.contains("-2")]
hs = float(q["spread_bps"].median()) / 2
else:
hs = 3.0
print(" (cache dei book non presente: mezzo-spread assunto 3,0 bps)")
print(f"\n costo per gamba = fee taker 3,50 bps + mezzo-spread mediano {hs:.2f} bps"
f" = {3.5+hs:.2f} bps")
print(f" un ROLL trimestrale = 2 gambe (chiudi F1, apri F2) = {2*(3.5+hs):.1f} bps,")
print(f" 4 roll/anno = {4*2*(3.5+hs):.0f} bps/anno = {4*2*(3.5+hs)/100:.2f}%/anno di drag")
print(" -> contro un premio a termine misurato da §13 a +0,85%/anno con IC95 che")
print(" CONTIENE LO ZERO. Il costo del roll da solo e' dello stesso ordine.")
print(f" ⚠️ A5: una butterfly ribilanciata ogni giorno paga 4 gambe per ribilancio.")
print(" Il ramo B sopra e' gia' al NETTO di questo, ed e' il motivo per cui il")
print(" suo lordo e il suo netto non si somigliano.")
# ---------------------------------------------------------------- §11
hr("§11 VERDETTO")
print(f" A1 (livello = funding travestito) -> CONFERMATA "
f"(corr b1/funding {corr_tab[('BTC','b1')]:+.2f} BTC, {corr_tab[('ETH','b1')]:+.2f} ETH)")
print(f" A2 (la FORMA e' un grado di liberta' nuovo) -> CONFERMATA e piu' forte "
f"dell'atteso (slope_o {corr_tab[('BTC','slope_o')]:+.2f}/"
f"{corr_tab[('ETH','slope_o')]:+.2f}, curv {corr_tab[('BTC','curv')]:+.2f}/"
f"{corr_tab[('ETH','curv')]:+.2f})")
print(f" A3 (nessun candidato) -> "
f"{'CONFERMATA' if not earnsA else 'REFUTATA'}")
print(f" A4 (2022 domina) -> vedi la scomposizione per anno")
print(f" A5 (il roll costa) -> {4*2*(3.5+hs)/100:.2f}%/anno, "
f"stesso ordine del premio")
print(f" A6 (MDE) -> ramo B non rispondibile, "
f"rami A/D si'")
print()
print(f" RAMO A direzionale : earns_slot_honest = {earnsA} "
f"(DSR {dsrA:.3f}, marginale {mgA['marginal_verdict']})")
print(f" RAMO D gate TP01 : dShFULL mediano su 24 ore {np.median(bD):+.3f}, "
f"DSR {dsrD:.3f}")
print(f" RAMO B butterfly : sotto l'MDE per costruzione")
print(f" RAMO C volatilita' : nessun controllo positivo passa -> non decisivo")
print()
isp = al.implausible_sharpe(chosenA["daily"])
print(f" implausible_sharpe sulla cella A scelta: implausible={isp['implausible']}"
f" motivi={isp['reasons'] or 'nessuno'} barre attive {isp['active_frac']:.0%}"
f" barre attive in perdita {isp['loss_frac']:.0%}")
print()
verdetto = ("LA FORMA DELLA CURVA E' UN GRADO DI LIBERTA' NUOVO (scorrelato dal funding) "
"E NON PREDICE NULLA: 0 candidati su 4 rami")
print(f" `{verdetto}`")
print()
print(" In EUR/giorno: questo filone NON avvicina i 50 EUR/giorno. Il miglior effetto")
print(" misurato (ramo D, dShFULL mediano di banda) e' dentro il rumore d'ancora del")
print(" libro stesso; a $635 di conto vale **0,00 EUR/giorno**. Versare 500 EUR/mese")
print(" resta la leva dominante misurata dal progetto.")
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink