Adriano Dal Pastro
710 lines
39 KiB
Python
710 lines
39 KiB
Python
#!/usr/bin/env python
|
|
"""r0822_oi_pin.py — OI-PIN: il prezzo e' attratto verso gli strike a massimo open interest
|
|
nelle ore che precedono la scadenza Deribit? (ondata 2026-08-22, filone OI-PIN)
|
|
|
|
CONTESTO / PERCHE' QUESTO FILONE NON E' "CALENDARIO GIA' MORTO"
|
|
--------------------------------------------------------------
|
|
Il 02/07 il progetto ha bocciato il **calendario** delle scadenze Deribit (drift post-expiry
|
|
weekly/monthly/quarterly): 0/24 celle a Bonferroni, e il pattern si INVERTE proprio sul
|
|
quarterly, dove l'open interest massimo dovrebbe amplificarlo. Quel test guardava le DATE.
|
|
Qui si guarda l'OPEN INTEREST VERO, per strike, ora per ora (`data/raw/cb_chain/`):
|
|
il meccanismo proposto non e' "il venerdi' e' speciale" ma "il market maker corto gamma
|
|
ricompra/rivende verso lo strike dove sta la sua esposizione". E' un meccanismo NON di
|
|
calendario, quindi ammissibile secondo la sezione 5 del brief.
|
|
|
|
IPOTESI (dichiarata PRIMA di misurare)
|
|
--------------------------------------
|
|
H1 Il prezzo deriva VERSO il livello a massima concentrazione di OI / gamma / max-pain nelle
|
|
ultime ore prima della scadenza (statistica `toward = sign(K - S_t) * r_{t->T}` > 0).
|
|
H2 L'effetto e' piu' forte a poche ore dalla scadenza e sulle scadenze grandi (venerdi').
|
|
H0-ATTESA DELL'AUTORE: **negativa**. Motivi dichiarati in anticipo:
|
|
(a) il crypto e' 24/7, senza il "chiusura di borsa" che genera il pinning azionario classico;
|
|
(b) l'OI Deribit e' grande ma il perp/spot che lo dovrebbe muovere e' molto piu' grande;
|
|
(c) il progetto ha gia' misurato che ogni effetto di finestra su BTC/ETH e' rumore;
|
|
(d) e soprattutto: `K_oi` sta quasi sempre VICINO allo spot, quindi qualunque statistica
|
|
"il prezzo va verso K" e' contaminata dalla POSIZIONE del livello, non dal suo OI.
|
|
Per questo il null location-matched (regola codificata il 02/07 sui livelli di Fibonacci)
|
|
non e' un contorno: e' IL test.
|
|
|
|
CONVENZIONE CAUSALE (verificata da un assert nel codice)
|
|
--------------------------------------------------------
|
|
Barre 1h open-labeled: la barra etichettata `tau` copre [tau, tau+1h) e il suo close e' il
|
|
prezzo a `tau+1h`.
|
|
* snapshot catena all'ora `hh` -> righe con ts in [hh, hh+1h)
|
|
* decisione a fine barra `hh` -> prezzo eseguibile = close(hh) = prezzo a hh+1h
|
|
* quindi entrata a `hte` ore dalla scadenza <=> snapshot a dte = hte+1 ore
|
|
* uscita al prezzo di regolamento = close della barra (E-1h) = prezzo a E (08:00 UTC)
|
|
`A.causality_ok` NON e' applicabile (il segnale non e' funzione del solo feed OHLC): la
|
|
causalita' e' garantita per costruzione e verificata da `_assert_causalita()`.
|
|
|
|
Uso: nice -n 19 timeout 900 uv run python scripts/research/r0822_oi_pin.py
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import glob
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
sys.path.insert(0, str(ROOT))
|
|
sys.path.insert(0, str(ROOT / "scripts" / "research" / "alt"))
|
|
import altlib as A # noqa: E402
|
|
|
|
STORE = ROOT / "data" / "raw" / "cb_chain"
|
|
ASSETS = ("BTC", "ETH")
|
|
FEE_SIDE_REAL = 0.00035 # 3,5 bps/lato = fee Deribit REALE del libro (7 bps RT)
|
|
FEE_SIDE_CONS = A.FEE_SIDE # 5 bps/lato = convenzione conservativa dei backtest del progetto
|
|
MAX_H = 50 # ore prima della scadenza da tenere in memoria
|
|
HTE_GRID = (1, 2, 3, 6, 12, 24, 36, 48) # ore-alla-scadenza all'INGRESSO
|
|
RULES = ("oi", "gex", "mp") # max OI · max gamma*OI · max pain
|
|
PLACEBOS_GLOB = ("rand", "mid", "mr7") # non dipendono dalla regola
|
|
PLACEBOS_RULE = ("adj", "jit") # costruiti sulla distanza DELLA regola
|
|
SEED = 20260822
|
|
RNG = np.random.default_rng(SEED)
|
|
|
|
|
|
# =========================================================================== dati
|
|
def _files() -> list[str]:
|
|
fs = sorted(glob.glob(str(STORE / "*.parquet")))
|
|
if not fs:
|
|
raise FileNotFoundError(f"{STORE} vuoto o assente")
|
|
return fs
|
|
|
|
|
|
def load_near(max_h: int = MAX_H) -> pd.DataFrame:
|
|
"""Catena filtrata alle sole ore vicine a scadenza, letta file per file (RAM: 7 GB in due)."""
|
|
cols = ["asset", "strike", "option_type", "ts", "exp",
|
|
"open_interest", "gamma", "quote_status"]
|
|
out = []
|
|
for p in _files():
|
|
d = pd.read_parquet(p, columns=cols)
|
|
d["ts"] = pd.to_datetime(d["ts"], utc=True)
|
|
d["exp"] = pd.to_datetime(d["exp"], utc=True)
|
|
d["hh"] = d["ts"].dt.floor("h")
|
|
d["dte_h"] = (d["exp"] - d["hh"]).dt.total_seconds() / 3600.0
|
|
d = d[(d["dte_h"] > 0) & (d["dte_h"] <= max_h)]
|
|
if len(d):
|
|
out.append(d)
|
|
del d
|
|
d = pd.concat(out, ignore_index=True)
|
|
# una riga presente non e' un dato presente: OI mancante o quota rotta non conta
|
|
d = d[d["open_interest"].notna() & (d["open_interest"] > 0)]
|
|
d = (d.sort_values("ts")
|
|
.drop_duplicates(subset=["asset", "exp", "hh", "strike", "option_type"], keep="last")
|
|
.reset_index(drop=True))
|
|
return d
|
|
|
|
|
|
def spot_1h() -> dict[str, pd.Series]:
|
|
s = {}
|
|
for a in ASSETS:
|
|
df = A.get(a, "1h")
|
|
s[a] = pd.Series(df["close"].to_numpy(float),
|
|
index=pd.DatetimeIndex(pd.to_datetime(df["datetime"], utc=True)))
|
|
return s
|
|
|
|
|
|
# =========================================================================== livelli
|
|
def _max_pain(K: np.ndarray, oi_c: np.ndarray, oi_p: np.ndarray) -> float:
|
|
"""Strike che minimizza il payout complessivo verso i detentori (max pain classico)."""
|
|
d = K[None, :] - K[:, None] # d[i,j] = K_j - K_i (K_i = candidato)
|
|
pay = (np.maximum(-d, 0.0) * oi_c[None, :]).sum(1) + (np.maximum(d, 0.0) * oi_p[None, :]).sum(1)
|
|
return float(K[int(np.argmin(pay))])
|
|
|
|
|
|
def levels_table(chain: pd.DataFrame, spot: dict[str, pd.Series]) -> pd.DataFrame:
|
|
"""Un record per (asset, scadenza, ora): i 3 livelli candidati + i 4 placebo di strike
|
|
+ le misure di concentrazione. I placebo sono LOCATION-MATCHED (regola 02/07)."""
|
|
rows = []
|
|
piv = (chain.groupby(["asset", "exp", "hh", "strike", "option_type"], sort=False)
|
|
.agg(oi=("open_interest", "sum"), gm=("gamma", "mean")).reset_index())
|
|
for (asset, exp, hh), g in piv.groupby(["asset", "exp", "hh"], sort=False):
|
|
S = spot[asset].asof(hh) # prezzo a hh+1h = close(barra hh)
|
|
if not np.isfinite(S) or S <= 0:
|
|
continue
|
|
K = np.sort(g["strike"].unique().astype(float))
|
|
if len(K) < 3:
|
|
continue
|
|
oi_c = np.zeros(len(K)); oi_p = np.zeros(len(K)); gx = np.zeros(len(K))
|
|
idx = {k: i for i, k in enumerate(K)}
|
|
for k, ot, o, gm in zip(g["strike"].to_numpy(float), g["option_type"].to_numpy(),
|
|
g["oi"].to_numpy(float), g["gm"].to_numpy(float)):
|
|
i = idx[float(k)]
|
|
(oi_c if ot == "C" else oi_p)[i] += o
|
|
gx[i] += o * (gm if np.isfinite(gm) else 0.0)
|
|
oi = oi_c + oi_p
|
|
tot = oi.sum()
|
|
if tot <= 0:
|
|
continue
|
|
i_oi = int(np.argmax(oi))
|
|
lv = {
|
|
"oi": float(K[i_oi]),
|
|
"gex": float(K[int(np.argmax(gx))]) if gx.max() > 0 else np.nan,
|
|
"mp": _max_pain(K, oi_c, oi_p),
|
|
}
|
|
# ---- placebo LOCATION-MATCHED, uno per OGNI regola --------------------------------
|
|
# ⚠ difetto trovato e corretto in sessione: la prima stesura costruiva `adj`/`jit`
|
|
# attorno al SOLO max-OI e poi li usava come null anche per max-pain e gamma. Ma
|
|
# |K_oi - S| mediano e' ~2,4% e |K_mp - S| ~1,1%: confrontare mp con un placebo
|
|
# tarato sulla distanza di oi NON e' location-matched, e' proprio l'errore che il
|
|
# null doveva prevenire. Ogni regola ha ora i placebo alla PROPRIA distanza.
|
|
dist = np.abs(K - S)
|
|
for nm, kv in list(lv.items()):
|
|
if not np.isfinite(kv):
|
|
out_adj = out_jit = np.nan
|
|
else:
|
|
i_r = int(np.argmin(np.abs(K - kv)))
|
|
j = i_r + (1 if (i_r == 0 or (i_r < len(K) - 1 and RNG.random() < 0.5)) else -1)
|
|
out_adj = float(K[int(np.clip(j, 0, len(K) - 1))])
|
|
d0 = abs(kv - S)
|
|
cand = np.where((dist >= 0.5 * d0) & (dist <= 1.5 * d0) & (K != kv))[0]
|
|
out_jit = float(K[cand[RNG.integers(len(cand))]]) if len(cand) else np.nan
|
|
lv[f"adj_{nm}"] = out_adj
|
|
lv[f"jit_{nm}"] = out_jit
|
|
# placebo GLOBALI (non dipendono dalla regola)
|
|
# rand: permutare l'OI fra gli strike e riprenderne l'argmax equivale ESATTAMENTE a
|
|
# estrarre uno strike uniforme dalla griglia (l'argmax di un vettore permutato e'
|
|
# uniforme sulle posizioni) -> lo si implementa cosi', ed e' esatto, non simulato.
|
|
lv["rand"] = float(K[RNG.integers(len(K))])
|
|
# mid: centro della griglia quotata = puro artefatto di copertura, zero informazione OI
|
|
lv["mid"] = float(0.5 * (K[0] + K[-1]))
|
|
rows.append(dict(
|
|
asset=asset, exp=exp, hh=hh, dte_h=float((exp - hh).total_seconds() / 3600.0),
|
|
S=float(S), n_k=len(K), tot_oi=float(tot),
|
|
top_share=float(oi[i_oi] / tot),
|
|
hhi=float(((oi / tot) ** 2).sum()),
|
|
k_step=float(np.median(np.diff(K))) if len(K) > 1 else np.nan,
|
|
**{f"L_{k}": v for k, v in lv.items()},
|
|
))
|
|
return pd.DataFrame(rows)
|
|
|
|
|
|
# =========================================================================== il FATTO
|
|
def build_events(lev: pd.DataFrame, spot: dict[str, pd.Series]) -> pd.DataFrame:
|
|
"""Un evento per (asset, scadenza, hte): livelli, prezzo d'ingresso, prezzo di regolamento."""
|
|
ev = []
|
|
mr7 = {a: spot[a].rolling(24 * 7, min_periods=24).mean() for a in ASSETS}
|
|
for a in ASSETS:
|
|
s = spot[a]
|
|
sub = lev[lev["asset"] == a]
|
|
for exp, g in sub.groupby("exp"):
|
|
t_set = pd.Timestamp(exp) - pd.Timedelta(hours=1) # barra il cui close e' a E
|
|
if t_set not in s.index:
|
|
continue
|
|
S_T = float(s.loc[t_set])
|
|
gi = g.set_index("hh")
|
|
for hte in HTE_GRID:
|
|
hh = pd.Timestamp(exp) - pd.Timedelta(hours=hte + 1)
|
|
if hh not in gi.index or hh not in s.index:
|
|
continue
|
|
r = gi.loc[hh]
|
|
S = float(s.loc[hh]) # close(hh) = prezzo a E-hte
|
|
if not np.isfinite(S) or S <= 0:
|
|
continue
|
|
rec = dict(asset=a, exp=exp, hte=hte, hh=hh, S=S, S_T=S_T,
|
|
ret=S_T / S - 1.0, n_k=int(r["n_k"]), tot_oi=float(r["tot_oi"]),
|
|
top_share=float(r["top_share"]), hhi=float(r["hhi"]),
|
|
k_step=float(r["k_step"]),
|
|
dow=int(pd.Timestamp(exp).dayofweek))
|
|
lvls = {c[2:]: r[c] for c in r.index if str(c).startswith("L_")}
|
|
lvls["mr7"] = float(mr7[a].asof(hh))
|
|
for nm, K in lvls.items():
|
|
K = float(K) if K is not None else np.nan
|
|
if not np.isfinite(K) or K <= 0:
|
|
rec[f"d_{nm}"] = np.nan; rec[f"tw_{nm}"] = np.nan
|
|
rec[f"sh_{nm}"] = np.nan
|
|
continue
|
|
d = (K - S) / S
|
|
rec[f"K_{nm}"] = K
|
|
rec[f"d_{nm}"] = d
|
|
rec[f"tw_{nm}"] = np.sign(d) * rec["ret"] # verso il livello
|
|
rec[f"sh_{nm}"] = (abs(S - K) - abs(S_T - K)) / S # avvicinamento
|
|
ev.append(rec)
|
|
E = pd.DataFrame(ev)
|
|
# NULL STATICO CAUSALE (lezione 25/07: il primo null statico di STATARB usava
|
|
# sign(mean(segnale)) su tutto il campione = look-ahead; qui la maggioranza e' ESPANDENTE
|
|
# e shiftata). Risponde a: "il candidato e' solo 'sempre nello stesso verso' prima della
|
|
# scadenza?" — che su 3,7 mesi di mercato direzionale e' l'ipotesi piu' probabile.
|
|
E = E.sort_values(["asset", "hte", "hh"]).reset_index(drop=True)
|
|
sgn = np.sign(E["d_mp"].to_numpy(float))
|
|
maj = (pd.Series(sgn).groupby([E["asset"], E["hte"]])
|
|
.transform(lambda x: x.expanding().mean().shift(1)))
|
|
E["d_maj"] = np.where(np.isfinite(maj), np.sign(maj) * 0.01, np.nan)
|
|
E["tw_maj"] = np.sign(E["d_maj"]) * E["ret"]
|
|
E["sh_maj"] = np.nan
|
|
E["d_long"] = 0.01; E["tw_long"] = E["ret"]; E["sh_long"] = np.nan
|
|
E["d_short"] = -0.01; E["tw_short"] = -E["ret"]; E["sh_short"] = np.nan
|
|
return E
|
|
|
|
|
|
def _assert_causalita(chain: pd.DataFrame, ev: pd.DataFrame) -> str:
|
|
"""Il livello usato a `hh` deve venire da quote osservate PRIMA del prezzo d'ingresso."""
|
|
bad = int((chain["ts"] < chain["hh"]).sum()
|
|
+ (chain["ts"] >= chain["hh"] + pd.Timedelta("1h")).sum())
|
|
assert bad == 0, "snapshot fuori dalla propria ora"
|
|
dte = (pd.to_datetime(ev["exp"]) - pd.to_datetime(ev["hh"])).dt.total_seconds() / 3600.0
|
|
assert bool((dte == ev["hte"] + 1).all()), "hte non coerente con lo snapshot"
|
|
assert bool(((pd.to_datetime(ev["hh"]) + pd.Timedelta(hours=1))
|
|
<= pd.to_datetime(ev["exp"])).all()), "ingresso oltre la scadenza"
|
|
return (f"OK — 0/{len(chain)} righe di catena fuori dalla propria ora; "
|
|
f"{len(ev)} eventi con dte == hte+1 (ingresso = close(hh), quota vista in [hh,hh+1h))")
|
|
|
|
|
|
# =========================================================================== inferenza
|
|
def cluster_boot(x: np.ndarray, clusters: np.ndarray, n: int = 2000, seed: int = SEED):
|
|
"""IC95 della media con ricampionamento dei CLUSTER (una scadenza = 1 cluster: BTC ed ETH
|
|
della stessa scadenza non sono osservazioni indipendenti)."""
|
|
x = np.asarray(x, float)
|
|
m = np.isfinite(x)
|
|
x, clusters = x[m], np.asarray(clusters)[m]
|
|
if len(x) < 5:
|
|
return np.nan, np.nan, np.nan, 0
|
|
uc = np.unique(clusters)
|
|
groups = [x[clusters == c] for c in uc]
|
|
rng = np.random.default_rng(seed)
|
|
draws = np.empty(n)
|
|
for i in range(n):
|
|
pick = rng.integers(0, len(groups), len(groups))
|
|
draws[i] = np.mean(np.concatenate([groups[j] for j in pick]))
|
|
return float(np.mean(x)), float(np.percentile(draws, 2.5)), float(np.percentile(draws, 97.5)), len(uc)
|
|
|
|
|
|
def mde(x: np.ndarray, clusters: np.ndarray) -> float:
|
|
"""Effetto minimo rilevabile (alpha .05 bilaterale, potenza 80%) con questo campione."""
|
|
x = np.asarray(x, float)
|
|
m = np.isfinite(x)
|
|
x, cl = x[m], np.asarray(clusters)[m]
|
|
if len(x) < 3:
|
|
return np.nan
|
|
nc = max(len(np.unique(cl)), 1)
|
|
return float(2.802 * np.std(x) / np.sqrt(nc))
|
|
|
|
|
|
# =========================================================================== strategia
|
|
_W: dict = {}
|
|
|
|
|
|
def win_1h(asset: str, lo: pd.Timestamp, hi: pd.Timestamp) -> pd.DataFrame:
|
|
"""Feed 1h tagliato alla finestra della catena: fuori da li' la posizione e' zero per
|
|
costruzione, e valutare 70k barre 288 volte costerebbe minuti per nulla."""
|
|
key = (asset, lo, hi)
|
|
if key not in _W:
|
|
df = A.get(asset, "1h")
|
|
dt = pd.DatetimeIndex(pd.to_datetime(df["datetime"], utc=True))
|
|
m = (dt >= lo - pd.Timedelta("2D")) & (dt <= hi + pd.Timedelta("2D"))
|
|
_W[key] = df.loc[m].reset_index(drop=True)
|
|
return _W[key]
|
|
|
|
|
|
def hourly_target(ev: pd.DataFrame, asset: str, rule: str, hte: int,
|
|
band: tuple[float, float] | None, sizing: str,
|
|
subsample: str, df1h: pd.DataFrame) -> np.ndarray:
|
|
"""Posizione oraria: entra `hte` ore prima della scadenza nel verso del livello, esce a
|
|
scadenza. Se piu' finestre si sovrappongono vince la scadenza PIU' VICINA (una posizione
|
|
alla volta = cio' che il conto Deribit puo' davvero fare: strumento unico, netting)."""
|
|
idx = pd.DatetimeIndex(pd.to_datetime(df1h["datetime"], utc=True))
|
|
pos = np.zeros(len(idx))
|
|
loc = pd.Series(np.arange(len(idx)), index=idx)
|
|
e = ev[(ev["asset"] == asset) & (ev["hte"] == hte)].copy()
|
|
if subsample == "fri":
|
|
e = e[e["dow"] == 4]
|
|
elif subsample == "big":
|
|
e = e[e["tot_oi"] >= e["tot_oi"].median()]
|
|
d = e[f"d_{rule}"].to_numpy(float)
|
|
keep = np.isfinite(d)
|
|
if band is not None:
|
|
keep &= (np.abs(d) >= band[0]) & (np.abs(d) <= band[1])
|
|
e = e[keep]
|
|
d = d[keep]
|
|
if len(e) == 0:
|
|
return pos
|
|
w = np.sign(d) if sizing == "sign" else np.clip(d / 0.01, -1.0, 1.0)
|
|
order = np.argsort(e["exp"].to_numpy())[::-1] # lontane prima: la scadenza VICINA sovrascrive
|
|
for j in order:
|
|
hh = pd.Timestamp(e["hh"].iloc[j]); exp = pd.Timestamp(e["exp"].iloc[j])
|
|
last = exp - pd.Timedelta(hours=2)
|
|
if hh not in loc.index or last not in loc.index:
|
|
continue
|
|
i0, i1 = int(loc[hh]), int(loc[last])
|
|
if i1 >= i0:
|
|
pos[i0:i1 + 1] = w[j]
|
|
return pos
|
|
|
|
|
|
def strat_daily(ev: pd.DataFrame, rule: str, hte: int, band, sizing: str, subsample: str,
|
|
fee_side: float) -> tuple[pd.Series, dict]:
|
|
"""Serie giornaliera netta 50/50 BTC+ETH + metriche, sulla sola finestra della catena."""
|
|
nets, info = {}, {}
|
|
lo_w = pd.Timestamp(ev["hh"].min()); hi_w = pd.Timestamp(ev["exp"].max())
|
|
for a in ASSETS:
|
|
df = win_1h(a, lo_w, hi_w)
|
|
tgt = hourly_target(ev, a, rule, hte, band, sizing, subsample, df)
|
|
r = A.eval_weights(df, tgt, fee_side=fee_side)
|
|
nets[a] = pd.Series(r["net"], index=r["idx"])
|
|
info[a] = int(np.sum(np.abs(np.diff(np.r_[0.0, tgt])) > 1e-12))
|
|
J = pd.concat(nets, axis=1, join="inner").fillna(0.0)
|
|
daily = A._to_daily(0.5 * J[ASSETS[0]] + 0.5 * J[ASSETS[1]])
|
|
lo = pd.Timestamp(ev["hh"].min()).normalize()
|
|
daily = daily[daily.index >= lo]
|
|
return daily, info
|
|
|
|
|
|
def metr(s: pd.Series) -> dict:
|
|
r = s.dropna()
|
|
if len(r) < 5:
|
|
return dict(sharpe=np.nan, dd=np.nan, cagr=np.nan, n=len(r))
|
|
eq = np.cumprod(1 + r.to_numpy(float)); pk = np.maximum.accumulate(eq)
|
|
yrs = max((r.index[-1] - r.index[0]).total_seconds() / 86400 / 365.25, 1e-6)
|
|
return dict(sharpe=A._sh(r), dd=float(np.max((pk - eq) / pk)),
|
|
cagr=float(eq[-1] ** (1 / yrs) - 1), n=int(len(r)))
|
|
|
|
|
|
# =========================================================================== main
|
|
def main() -> None:
|
|
P = print
|
|
P("=" * 100)
|
|
P("OI-PIN — pinning verso gli strike a massimo open interest (catena Deribit oraria)")
|
|
P("=" * 100)
|
|
|
|
# ---------------------------------------------------------------- 1. DATO
|
|
chain = load_near()
|
|
spot = spot_1h()
|
|
lev = levels_table(chain, spot)
|
|
ev = build_events(lev, spot)
|
|
P("\n[1] DATO")
|
|
P(f" righe catena a <= {MAX_H}h dalla scadenza : {len(chain):,}")
|
|
P(f" scadenze distinte : {ev.groupby('asset').exp.nunique().to_dict()}")
|
|
P(f" finestra : {ev.hh.min()} -> {ev.exp.max()}")
|
|
P(f" eventi (asset x scadenza x hte) : {len(ev)}")
|
|
P(f" strike per snapshot mediana/min : {lev.n_k.median():.0f} / {lev.n_k.min():.0f}")
|
|
P(f" causalita' : {_assert_causalita(chain, ev)}")
|
|
P(" ⚠ il collettore scarta gli strumenti con OI<100 -> la griglia e' TRONCATA sulle code:")
|
|
P(" il max-OI ne soffre poco (e' un argmax), il MAX PAIN si', perche' e' una somma su tutta")
|
|
P(" la catena. Il verdetto su 'mp' va letto come 'max pain sulla catena OI>=100'.")
|
|
|
|
# ---------------------------------------------------------------- 2. POTENZA (PRIMA del risultato)
|
|
P("\n[2] POTENZA DICHIARATA — calcolata PRIMA di leggere qualunque effetto")
|
|
P(" Unita' statistica = una SCADENZA (BTC ed ETH della stessa data non sono indipendenti).")
|
|
P(f" {'hte':>4} {'n oss':>6} {'n clus':>7} {'sd(ret)':>9} {'MDE medio/trade':>16} {'MDE ann.':>10}")
|
|
for hte in HTE_GRID:
|
|
e = ev[ev["hte"] == hte]
|
|
m = mde(e["ret"].to_numpy(float), e["exp"].to_numpy())
|
|
ann = m * (365.25 * 24 / hte) if np.isfinite(m) else np.nan
|
|
P(f" {hte:>4} {len(e):>6} {e.exp.nunique():>7} {np.nanstd(e['ret']):>9.4f}"
|
|
f" {m:>16.4f} {ann:>9.1%}")
|
|
P(" LETTURA OBBLIGATORIA: il campione rileva solo effetti ENORMI. A 24h l'MDE e' ~0,5-0,7%")
|
|
P(" per trade contro 0,07% di fee: un pinning realistico (pochi bps) e' INVISIBILE qui.")
|
|
P(" Quindi un 'non significativo' NON prova assenza; un 'significativo' sarebbe cosi' grande")
|
|
P(" da essere sospetto. Il verdetto massimo raggiungibile e' LEAD.")
|
|
|
|
# ---------------------------------------------------------------- 3. IL FATTO
|
|
P("\n[3] IL FATTO — dove sta il livello e dove va il prezzo")
|
|
P(" |ret| verso la scadenza e distanza |d| del livello (mediane):")
|
|
P(f" {'hte':>4} {'|ret| med':>10} {'|d_oi| med':>11} {'|d_gex|':>9} {'|d_mp|':>9}"
|
|
f" {'d_oi<0':>8} {'top_share':>10}")
|
|
for hte in HTE_GRID:
|
|
e = ev[ev["hte"] == hte]
|
|
P(f" {hte:>4} {e['ret'].abs().median():>10.4f} {e['d_oi'].abs().median():>11.4f}"
|
|
f" {e['d_gex'].abs().median():>9.4f} {e['d_mp'].abs().median():>9.4f}"
|
|
f" {(e['d_oi'] < 0).mean():>8.2f} {e['top_share'].median():>10.3f}")
|
|
P(" ⚠ se 'd_oi<0' e' lontano da 0.50 il livello ha un BIAS DI SEGNO (griglia troncata o")
|
|
P(" mercato in trend): una statistica 'il prezzo va verso K' lo erediterebbe.")
|
|
|
|
P("\n Statistica di pinning tw = sign(K-S)*ret (media, IC95 cluster-bootstrap su scadenze):")
|
|
P(f" {'hte':>4} {'rule':>5} {'media tw':>10} {'IC95 lo':>9} {'IC95 hi':>9} {'sig':>4}"
|
|
f" {'shrink med':>11} {'n':>5}")
|
|
fact = []
|
|
for hte in HTE_GRID:
|
|
e = ev[ev["hte"] == hte]
|
|
for rl in RULES:
|
|
m, lo, hi, nc = cluster_boot(e[f"tw_{rl}"].to_numpy(float), e["exp"].to_numpy())
|
|
sg = "***" if (np.isfinite(lo) and lo > 0) else ("neg" if (np.isfinite(hi) and hi < 0) else "-")
|
|
P(f" {hte:>4} {rl:>5} {m:>10.5f} {lo:>9.5f} {hi:>9.5f} {sg:>4}"
|
|
f" {np.nanmedian(e[f'sh_{rl}']):>11.5f} {int(np.isfinite(e[f'tw_{rl}']).sum()):>5}")
|
|
fact.append(dict(hte=hte, rule=rl, mean=m, lo=lo, hi=hi))
|
|
F = pd.DataFrame(fact)
|
|
P(f" celle con IC95 tutto sopra zero: {int((F.lo > 0).sum())}/{len(F)}"
|
|
f" | tutto sotto zero: {int((F.hi < 0).sum())}/{len(F)}")
|
|
|
|
P("\n LA STESSA STATISTICA SUI LIVELLI PLACEBO — se sale anche li', NON e' pinning:")
|
|
P(f" {'hte':>4} " + " ".join(f"{('tw_' + n):>10}" for n in
|
|
("oi", "mp", "rand", "jit_mp", "mid", "mr7", "maj", "short")))
|
|
for hte in HTE_GRID:
|
|
e = ev[ev["hte"] == hte]
|
|
P(f" {hte:>4} " + " ".join(
|
|
f"{np.nanmean(e[f'tw_{n}']):>10.5f}" for n in
|
|
("oi", "mp", "rand", "jit_mp", "mid", "mr7", "maj", "short")))
|
|
P("\n QUANTO SONO LO STESSO SEGNALE? accordo del SEGNO di (K-S) fra max-pain e i placebo:")
|
|
for n in ("mid", "mr7", "oi", "maj"):
|
|
agr = [float(np.nanmean(np.sign(ev.loc[ev.hte == h, "d_mp"])
|
|
== np.sign(ev.loc[ev.hte == h, f"d_{n}"]))) for h in HTE_GRID]
|
|
P(f" mp vs {n:<5}: " + " ".join(f"h{h}={a:.2f}" for h, a in zip(HTE_GRID, agr)))
|
|
P(f" base rate sign<0 : " + " ".join(
|
|
f"{n}={np.nanmean(np.sign(ev[f'd_{n}']) < 0):.2f}" for n in ("oi", "gex", "mp", "mid", "mr7")))
|
|
for a in ASSETS:
|
|
sp_ = spot[a]
|
|
w = sp_[(sp_.index >= ev.hh.min()) & (sp_.index <= ev.exp.max())]
|
|
P(f" deriva {a} nella finestra: {w.iloc[-1] / w.iloc[0] - 1:+.1%}"
|
|
f" (se il livello sta quasi sempre da un lato, il candidato e' una scommessa direzionale)")
|
|
|
|
# ---------------------------------------------------------------- 4. NULL LOCATION-MATCHED
|
|
P("\n[4] NULL LOCATION-MATCHED (regola 02/07: e' la POSIZIONE o e' l'OI?)")
|
|
P(" Confronto APPAIATO per evento: tw(livello speciale) - tw(placebo alla stessa distanza).")
|
|
P(" placebo PER REGOLA: adj_r = strike adiacente a K_r · jit_r = strike arbitrario a")
|
|
P(" distanza 0.5-1.5x |K_r - S| (analogo del Fib±jitter del 02/07).")
|
|
P(" placebo GLOBALI: rand = strike uniforme (== permutazione dell'OI, equivalenza esatta) ·")
|
|
P(" mid = centro della griglia quotata (zero info OI) · mr7 = media 7g dello spot.")
|
|
P(f" {'hte':>4} {'rule':>5} {'Δ vs adj_r':>12} {'Δ vs jit_r':>12}"
|
|
f" {'Δ vs rand':>12} {'Δ vs mid':>12} {'Δ vs mr7':>12}")
|
|
npos = ntot = 0
|
|
npos_c = ntot_c = 0 # solo i null PULITI (bilanciati nel segno): rand/mid/mr7
|
|
for hte in HTE_GRID:
|
|
e = ev[ev["hte"] == hte]
|
|
for rl in RULES:
|
|
cells = []
|
|
for p in (f"adj_{rl}", f"jit_{rl}") + PLACEBOS_GLOB:
|
|
dd = e[f"tw_{rl}"].to_numpy(float) - e[f"tw_{p}"].to_numpy(float)
|
|
m, lo, hi, _ = cluster_boot(dd, e["exp"].to_numpy())
|
|
sig = bool(np.isfinite(lo) and lo > 0)
|
|
cells.append(f"{m:>11.5f}{'*' if sig else ' '}")
|
|
ntot += 1; npos += int(sig)
|
|
if p in PLACEBOS_GLOB:
|
|
ntot_c += 1; npos_c += int(sig)
|
|
P(f" {hte:>4} {rl:>5} " + " ".join(cells))
|
|
P(f" celle (rule x hte x placebo) in cui lo SPECIALE batte il placebo con IC95>0:"
|
|
f" {npos}/{ntot} (atteso per caso ~{0.025 * ntot:.0f})")
|
|
P(f" ... contando SOLO i null bilanciati nel segno (rand/mid/mr7): {npos_c}/{ntot_c}"
|
|
f" (atteso ~{0.025 * ntot_c:.0f}) <-- e' QUESTO il numero da citare")
|
|
P("\n ⚠ UN PLACEBO SI CONTROLLA PRIMA DI USARLO: se il suo segno non e' bilanciato rispetto")
|
|
P(" alla regola, il Δ non misura l'informazione, misura il ROVESCIAMENTO del segnale.")
|
|
P(f" quota di eventi con sign(K_placebo-S) == sign(K_regola-S), su tutti gli hte:")
|
|
for rl in RULES:
|
|
agr = {p: float(np.nanmean(np.sign(ev[f"d_{rl}"]) == np.sign(ev[f"d_{p}"])))
|
|
for p in (f"adj_{rl}", f"jit_{rl}") + PLACEBOS_GLOB}
|
|
P(" " + rl + ": " + " ".join(f"{k}={v:.2f}" for k, v in agr.items()))
|
|
P(" un null PULITO sta vicino a 0.50 (rand/mid/mr7). Sotto ~0.35 il placebo e' la")
|
|
P(" strategia INVERTITA e il suo Δ va letto come ~2x il segnale, non come un test.")
|
|
P(" LETTURA: `adj_r`/`jit_r` chiedono 'e' proprio QUELLO strike?'; `mid`/`mr7` chiedono")
|
|
P(" 'serve l'open interest, o basta il centro recente del prezzo?'. La seconda e' la")
|
|
P(" domanda che uccide o salva il filone.")
|
|
|
|
# ---------------------------------------------------------------- 5. CONDIZIONAMENTO
|
|
P("\n[5] CONDIZIONAMENTO — la concentrazione di OI e le scadenze grandi cambiano qualcosa?")
|
|
P(f" {'hte':>4} {'gruppo':>16} {'media tw_oi':>12} {'IC95 lo':>9} {'IC95 hi':>9} {'n':>4}")
|
|
for hte in (3, 12, 24):
|
|
e = ev[ev["hte"] == hte]
|
|
hi_c = e["top_share"] >= e["top_share"].median()
|
|
big = e["tot_oi"] >= e["tot_oi"].quantile(0.75)
|
|
for nm, sel in (("conc ALTA", hi_c), ("conc BASSA", ~hi_c),
|
|
("OI top-25%", big), ("venerdi'", e["dow"] == 4)):
|
|
x = e.loc[sel, "tw_oi"].to_numpy(float)
|
|
m, lo, hh_, _ = cluster_boot(x, e.loc[sel, "exp"].to_numpy())
|
|
P(f" {hte:>4} {nm:>16} {m:>12.5f} {lo:>9.5f} {hh_:>9.5f} {int(np.isfinite(x).sum()):>4}")
|
|
|
|
# ---------------------------------------------------------------- 6. STRATEGIA
|
|
P("\n[6] STRATEGIA — expectancy netta fee (3,5 bps/lato reali) e serie giornaliera")
|
|
bands = {"nessuna": None, "0.2-3%": (0.002, 0.03)}
|
|
subs = ("all", "big", "fri")
|
|
sizings = ("sign", "prop")
|
|
cells, n_trials = [], 0
|
|
for rl in RULES:
|
|
for hte in HTE_GRID:
|
|
for bn, bd in bands.items():
|
|
for sz in sizings:
|
|
for sb in subs:
|
|
n_trials += 1
|
|
d, info = strat_daily(ev, rl, hte, bd, sz, sb, FEE_SIDE_REAL)
|
|
mm = metr(d)
|
|
cells.append(dict(rule=rl, hte=hte, band=bn, sizing=sz, sub=sb,
|
|
**mm, ntrade=sum(info.values())))
|
|
C = pd.DataFrame(cells)
|
|
P(f" griglia strategia valutata: {n_trials} celle "
|
|
f"({len(RULES)} rule x {len(HTE_GRID)} hte x {len(bands)} bande x {len(sizings)} sizing x {len(subs)} sottocampioni)")
|
|
P(f" celle con Sharpe > 0: {(C.sharpe > 0).sum()}/{len(C)} (una famiglia di RUMORE sta a ~50%)")
|
|
P(" migliori 8 celle per Sharpe (NON e' una selezione: e' il MASSIMO della griglia, che serve al DSR):")
|
|
P(C.sort_values("sharpe", ascending=False).head(8).to_string(index=False,
|
|
float_format=lambda v: f"{v:.3f}"))
|
|
P(" peggiori 3:")
|
|
P(C.sort_values("sharpe").head(3).to_string(index=False, float_format=lambda v: f"{v:.3f}"))
|
|
|
|
P("\n [6-bis] LA STESSA MACCHINA SUI LIVELLI PLACEBO (stesse finestre, stessa fee, stesso")
|
|
P(" sizing): se un placebo regge quanto il livello 'speciale', l'OI non aggiunge nulla.")
|
|
nulls = []
|
|
for rl in ("mp", "oi", "mid", "mr7", "rand", "jit_mp", "maj", "short", "long"):
|
|
for hte in (3, 6, 12, 24):
|
|
for bn, bd in bands.items():
|
|
d, _i = strat_daily(ev, rl, hte, bd, "sign", "all", FEE_SIDE_REAL)
|
|
mm = metr(d)
|
|
nulls.append(dict(rule=rl, hte=hte, band=bn, sharpe=mm["sharpe"],
|
|
cagr=mm["cagr"], dd=mm["dd"]))
|
|
N = pd.DataFrame(nulls)
|
|
piv = N.pivot_table(index=["hte", "band"], columns="rule", values="sharpe")
|
|
P(piv.reindex(columns=["mp", "oi", "mid", "mr7", "rand", "jit_mp", "maj", "short", "long"])
|
|
.to_string(float_format=lambda v: f"{v:6.2f}"))
|
|
P(f" celle placebo valutate: {len(N)} (contano come trial: la griglia dichiarata sale)")
|
|
|
|
best = C.sort_values("sharpe", ascending=False).iloc[0]
|
|
P(f"\n CELLA MIGLIORE = {best['rule']}/hte{int(best['hte'])}/{best['band']}/{best['sizing']}/{best['sub']}")
|
|
|
|
# expectancy per trade, in % e in R, sulla cella migliore
|
|
e = ev[ev["hte"] == int(best["hte"])].copy()
|
|
if best["sub"] == "big":
|
|
e = e[e["tot_oi"] >= e["tot_oi"].median()]
|
|
elif best["sub"] == "fri":
|
|
e = e[e["dow"] == 4]
|
|
dd = e[f"d_{best['rule']}"].to_numpy(float)
|
|
keep = np.isfinite(dd)
|
|
if best["band"] != "nessuna":
|
|
keep &= (np.abs(dd) >= 0.002) & (np.abs(dd) <= 0.03)
|
|
tw = e.loc[keep, f"tw_{best['rule']}"].to_numpy(float)
|
|
dk = np.abs(dd[keep])
|
|
net = tw - 2 * FEE_SIDE_REAL
|
|
wins, losses = net[net > 0], net[net < 0]
|
|
rr = (wins.mean() / abs(losses.mean())) if len(wins) and len(losses) else np.nan
|
|
P(f" trade: {len(net)} · WR {np.mean(net > 0):.1%} · RR medio {rr:.2f}"
|
|
f" · WR-knob atteso 1/(1+RR) = {1 / (1 + rr):.1%}" if np.isfinite(rr) else " RR n.d.")
|
|
P(f" expectancy NETTA per trade: {np.mean(net):+.5f} ({np.mean(net) * 100:+.3f}%)"
|
|
f" | in R (R = distanza dal livello): {np.mean(net / np.maximum(dk, 1e-6)):+.3f} R")
|
|
P(f" a fee ZERO: {np.mean(tw):+.5f} · la fee costa {2 * FEE_SIDE_REAL:.5f}/trade ="
|
|
f" {200 * FEE_SIDE_REAL / max(np.mean(tw), 1e-9):.0f}% del lordo"
|
|
f" -> {'MORTE PER FEE' if np.mean(tw) <= 2 * FEE_SIDE_REAL else 'la fee NON e il vincolo'}")
|
|
|
|
dbest, info = strat_daily(ev, best["rule"], int(best["hte"]),
|
|
None if best["band"] == "nessuna" else (0.002, 0.03),
|
|
best["sizing"], best["sub"], FEE_SIDE_REAL)
|
|
dcons, _ = strat_daily(ev, best["rule"], int(best["hte"]),
|
|
None if best["band"] == "nessuna" else (0.002, 0.03),
|
|
best["sizing"], best["sub"], FEE_SIDE_CONS)
|
|
mb, mc = metr(dbest), metr(dcons)
|
|
P(f" serie giornaliera cella migliore fee 3,5bps: Sharpe {mb['sharpe']:+.2f}"
|
|
f" · maxDD {mb['dd']:.2%} · CAGR {mb['cagr']:+.2%} · n={mb['n']}g")
|
|
P(f" fee 5,0bps: Sharpe {mc['sharpe']:+.2f}"
|
|
f" · maxDD {mc['dd']:.2%} · CAGR {mc['cagr']:+.2%}")
|
|
# scomposizione: per mese e per ERA DI COLLETTORE (bite:research fino al 30/07, poi pyg)
|
|
P("\n SCOMPOSIZIONE (regola 22/08: un contributo positivo si scompone prima di crederci)")
|
|
mo = dbest.groupby([dbest.index.year, dbest.index.month])
|
|
P(" " + " · ".join(f"{y}-{m:02d}: Sh {A._sh(g):+.2f} ret {float(np.prod(1 + g) - 1):+.2%} ({len(g)}g)"
|
|
for (y, m), g in mo))
|
|
cut = pd.Timestamp("2026-07-30", tz="UTC")
|
|
e1, e2 = dbest[dbest.index < cut], dbest[dbest.index >= cut]
|
|
P(f" era collettore bite:research (<30/07): Sh {A._sh(e1):+.2f} ret {float(np.prod(1 + e1) - 1):+.2%} ({len(e1)}g)"
|
|
f" | pyg (>=30/07): Sh {A._sh(e2):+.2f} ret {float(np.prod(1 + e2) - 1):+.2%} ({len(e2)}g)")
|
|
top5 = dbest.sort_values(ascending=False).head(5)
|
|
P(f" concentrazione: i 5 giorni migliori valgono {float(np.prod(1 + top5) - 1):+.2%}"
|
|
f" su {float(np.prod(1 + dbest) - 1):+.2%} totali"
|
|
f" = {100 * np.log1p(top5).sum() / max(np.log1p(dbest).sum(), 1e-9):.0f}% del log-equity")
|
|
P(" ⚠ HOLD-OUT: NON ESISTE. La catena parte dal 2026-05-01 e l'hold-out del progetto e'")
|
|
P(" 2025-01-01: TUTTA la serie sta dentro l'hold-out, quindi non c'e' nessuna finestra")
|
|
P(" fuori campione da mostrare. Questo NON e' un dettaglio: e' il motivo per cui")
|
|
P(" 3,7 mesi di catena non possono produrre uno sleeve.")
|
|
|
|
# ---------------------------------------------------------------- 7. GATE
|
|
P("\n[7] GATE")
|
|
P(f" causalita' : {_assert_causalita(chain, ev)}")
|
|
P(" A.causality_ok : NON GIRATO — non applicabile (il segnale non e' funzione del solo")
|
|
P(" feed OHLC; la causalita' e' garantita dalla convenzione hte=dte-1")
|
|
P(" e verificata dall'assert qui sopra).")
|
|
|
|
# DSR: conta TUTTI i trial al rialzo
|
|
all_sr = [float(v) for v in C["sharpe"].to_numpy(float) if np.isfinite(v)]
|
|
all_sr_wide = all_sr + [float(v) for v in N["sharpe"].to_numpy(float) if np.isfinite(v)]
|
|
n_fact = len(HTE_GRID) * len(RULES) * (1 + len(PLACEBOS_GLOB) + len(PLACEBOS_RULE)) + 3 * 4 # celle del fatto + condiz.
|
|
trials_tot = len(all_sr_wide) + n_fact
|
|
dsr, sr0 = A.deflated_sharpe(mb["sharpe"], all_sr, dbest, dpy=365.25)
|
|
dsr_w, _ = A.deflated_sharpe(mb["sharpe"], all_sr_wide, dbest, dpy=365.25)
|
|
P(f" deflated Sharpe : DSR {dsr:.3f} (soglia 0.95) · Sharpe-null atteso {sr0:.2f}"
|
|
f" · trial contati {len(all_sr)} (griglia strategia)")
|
|
P(f" DSR contando ANCHE le celle placebo ({len(all_sr_wide)} trial): {dsr_w:.3f}")
|
|
P(f" trial TOTALI dichiarati al rialzo (fatto+null+condiz.+strategia): {trials_tot}")
|
|
|
|
# marginal vs TP01
|
|
mv = A.marginal_vs_tp01(dbest)
|
|
P(f" marginal vs TP01 : {mv.get('marginal_verdict')} · corr {mv.get('corr_full')}"
|
|
f" · uplift w25 full {mv.get('blends', {}).get('w25', {}).get('uplift_full')}"
|
|
f" · giorni in comune {mv.get('n_days')}")
|
|
P(" ⚠ il gate NON PUO' dare ADDS con questo campione: `multicut_persistent`")
|
|
P(" richiede >=2 tagli annuali da >=120 giorni e la serie ne ha ~113 in UN")
|
|
P(" anno solo; e `has_insample_edge` e' True per DEFAULT perche' non c'e'")
|
|
P(" alcun pre-2025. Il verdetto marginale qui e' strutturale, non informativo.")
|
|
|
|
# null del de-levering (obbligatorio su ogni claim di DD)
|
|
B = A.tp01_baseline_daily()
|
|
J = pd.concat({"B": B, "C": dbest}, axis=1, join="inner").dropna()
|
|
if len(J) > 20:
|
|
bl = 0.75 * J["B"] + 0.25 * J["C"]
|
|
dd_bl = A._dd_ret(bl)
|
|
ks = np.linspace(0.05, 1.0, 96)
|
|
ok = [(k, A._sh(k * J["B"]), A._dd_ret(k * J["B"])) for k in ks]
|
|
cand = [(k, s) for k, s, d_ in ok if d_ <= dd_bl]
|
|
kbest = max(cand, key=lambda t: t[1]) if cand else None
|
|
P(f" null de-levering : blend 0.75TP01+0.25cand -> Sharpe {A._sh(bl):+.2f} / DD {dd_bl:.2%}"
|
|
f" | k*TP01 a pari DD -> k={kbest[0]:.2f} Sharpe {kbest[1]:+.2f}"
|
|
if kbest else " null de-levering : non calcolabile")
|
|
if kbest:
|
|
P(f" esito: {'REFUTED (il de-levering fa meglio)' if kbest[1] >= A._sh(bl) else 'superato'}")
|
|
else:
|
|
P(" null de-levering : NON GIRATO (sovrapposizione con TP01 troppo corta)")
|
|
|
|
# implausible
|
|
imp = A.implausible_sharpe(dbest, n_trades=int(best["ntrade"]))
|
|
P(f" implausible_sharpe : implausible={imp['implausible']} · {imp.get('reasons')}")
|
|
|
|
# anchor: l'ANCORA di questa strategia e' hte (quale ora prima della scadenza si entra)
|
|
ab = A.anchor_luck_band(
|
|
lambda h: strat_daily(ev, best["rule"], int(h),
|
|
None if best["band"] == "nessuna" else (0.002, 0.03),
|
|
best["sizing"], best["sub"], FEE_SIDE_REAL)[0],
|
|
offsets=list(HTE_GRID), canonical=int(best["hte"]))
|
|
P(f" anchor (hte) : canonico {ab.get('canonical'):.2f} al {100 * ab.get('canonical_pctl', np.nan):.0f}° pctl"
|
|
f" · MEDIANA ONESTA {ab.get('median'):.2f} · banda [{ab.get('lo'):.2f},{ab.get('hi'):.2f}]"
|
|
f" · positive {ab.get('frac_positive'):.0%} · gate_pass={ab.get('gate_pass')}")
|
|
|
|
# LA DOMANDA DECISIVA, sulla banda d'ancora: max-pain aggiunge qualcosa a un livello
|
|
# che NON usa opzioni? Mediana delle DIFFERENZE APPAIATE sui 8 hte (mai differenza
|
|
# delle mediane: lezione 26/07).
|
|
def _fn(rule):
|
|
return lambda h: strat_daily(ev, rule, int(h), (0.002, 0.03), "sign", "all",
|
|
FEE_SIDE_REAL)[0]
|
|
P(" ⚠ le 8 ancore hte NON sono 8 osservazioni indipendenti: guardano le STESSE 73 scadenze")
|
|
P(" e lo STESSO percorso di prezzo. '8/8 positive' dice che il segno non dipende")
|
|
P(" dall'ancora, NON che l'effetto sia significativo — quello lo dice il cluster-bootstrap")
|
|
P(" della sezione [4], che sui placebo senza opzioni (mid/mr7) NON e' significativo.")
|
|
for other in ("mr7", "mid", "oi", "rand"):
|
|
dl = A.anchor_luck_delta(_fn("mp"), _fn(other), offsets=list(HTE_GRID))
|
|
P(f" Δ appaiato mp-{other:<4}: mediana {dl.get('median_paired'):+.2f}"
|
|
f" · positiva in {dl.get('n_positive')}/{dl.get('n_anchors')} hte"
|
|
f" · banda [{dl.get('lo'):+.2f},{dl.get('hi'):+.2f}] · gate_pass={dl.get('gate_pass')}")
|
|
|
|
# eseguibilita' a $600
|
|
P(" eseguibilita' $600 :", end=" ")
|
|
hc = []
|
|
for a in ASSETS:
|
|
df = win_1h(a, pd.Timestamp(ev["hh"].min()), pd.Timestamp(ev["exp"].max()))
|
|
tgt = hourly_target(ev, a, best["rule"], int(best["hte"]),
|
|
None if best["band"] == "nessuna" else (0.002, 0.03),
|
|
best["sizing"], best["sub"], df)
|
|
sc = A.eval_weights_smallcap(df, tgt, capital=300.0, fee_side=FEE_SIDE_REAL)
|
|
hc.append(f"{a} haircut {sc['sharpe_haircut']:+.3f} ({sc['n_executed_trades']} trade eseguiti)")
|
|
P(" · ".join(hc))
|
|
P(" (300$/asset = il cap per-asset del libro live a questo capitale;")
|
|
P(" una gamba ±1 muove 300$ >> min-order 5$ -> nessun ordine saltato)")
|
|
|
|
# ---------------------------------------------------------------- 8. SINTESI
|
|
P("\n[8] SINTESI")
|
|
n_fact_pos = int((F.lo > 0).sum())
|
|
P(f" fatto : {n_fact_pos}/{len(F)} celle (rule x hte) con pinning significativo a IC95")
|
|
P(f" null loc. : {npos}/{ntot} confronti appaiati vinti dal livello 'speciale';"
|
|
f" {npos_c}/{ntot_c} contando solo i null PULITI")
|
|
P(f" di questi, contro il placebo SENZA OPZIONI (media 7g dello spot): "
|
|
f"{sum(1 for h in HTE_GRID for rl in RULES if (lambda t: np.isfinite(t[1]) and t[1] > 0)(cluster_boot((ev[ev.hte == h][f'tw_{rl}'] - ev[ev.hte == h]['tw_mr7']).to_numpy(float), ev[ev.hte == h]['exp'].to_numpy())))}/24")
|
|
P(f" strategia : miglior Sharpe della griglia {mb['sharpe']:+.2f} su {mb['n']} giorni, DSR {dsr:.3f}")
|
|
P(f" marginale : {mv.get('marginal_verdict')} (strutturalmente non ADDS-abile a 3,7 mesi)")
|
|
P("=" * 100)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|