Adriano Dal Pastro
05816c49f9
Chiesto dall'operatore in quattro battute: hold BTC long o short coperto in
opzioni, scadenza <=15gg, "ridurre la vincita ma bloccare la perdita" (=>
collar, non put protettiva), entrata gated da indicatori ("forte bull"), e
uscita dalle opzioni fra il 50% e il 75% del tempo.
PERCHE' SI POTEVA RIAPRIRE DOPO §46. §46 (tail-hedge) fu refutato sul beta
+0,076 del libro — "non si assicura un libro che nei crash e' gia' quasi
piatto". Qui il sottostante e' un hold di BTC, beta 1,0: quel motivo non si
applica. E §46 dichiarava non provata proprio la copertura gated su regime.
L'entrata non aggiunge un solo parametro: tsmom_blend media tre np.sign() su
(30,90,180) => valori in {-1,-1/3,+1/3,+1}, quindi "forte bull" = |blend|==1
(52,5% dei giorni) contro il confronto dichiarato |blend|>=1/3 (97,1%).
RISULTATI (lente lunga 2021-03 -> 2026-09, 5,44 anni; griglia 48 celle
dichiarata prima + 36 di estensione dichiarata):
- A1 CONFERMATA: il pavimento FUNZIONA, maxDD scende in 36/48 (§46: saliva
in 162/162). La meta' della domanda ha risposta positiva.
- A3 CONFERMATA: il de-levering lo fa meglio in 45/48. Δdrift/ΔmaxDD 7,90
(gate forte) / 1,98 (largo): 2-8 punti di drift per punto di DD.
- C9 in forma pura: il tetto taglia il 46,2% dei cicli VINCENTI, il pavimento
para il 6,7% dei PERDENTI — 7x piu' spesso sui vincenti: troncatura.
- M1: collar Sharpe 0,508 vs TP01 0,852; TP01+10% => +0,000 di Sharpe e
+2,32pp di maxDD.
IL FATTO CHE VALE PIU' DEL VERDETTO. Le 3 celle vincenti stavano tutte sul
BORDO; estesa la famiglia vince 35/36 nell'ANGOLO (dput 0,02 / dcall 0,50,
Sharpe 1,471) — e il limite di quell'angolo e' una COVERED CALL: la pendenza
porta fuori dalla domanda posta e dentro lo short-vol. E quel 1,471 e' il
prezzatore che si paga da solo: DVOL/RV-forward 1,320 a 7g (sopra nel 76,9%
dei giorni) => riprezzato alla vol vera l'angolo cade a 0,511, che e' VRP01
(0,47). Non una scoperta: VRP01 per una strada piu' lunga. §3 lo blocca.
USCITA ANTICIPATA: implementata (exit_frac) e COSTA. Cella onesta gate forte:
drift +9,48% (scadenza) -> +3,84% (50%), esito da VINCE a perde sotto 0,75.
Il meccanismo previsto c'e' (VRP residuo +3,38 -> +1,27pp) ma lo spread lo
travolge. Corregge l'applicazione di §46: "un roll anticipato non paga f"
vale per una copertura solo LONG; in un collar la gamba venduta va
RICOMPRATA, quindi si paga f sulla parte che a scadenza si regolava gratis.
L'asimmetria si INVERTE quando la struttura ha una gamba corta.
CONTROLLI DELL'APPARATO 3/3 (M15): pranzo gratis riconosciuto (maxDD
51,83%->36,76%, drift +10,04%->+35,69%), premio x10 rifiutato, zero-cost
finito. Cinque difetti miei catturati dai controlli, non a occhio: bisezione
zero-cost invertita (dava Sharpe -3,9), dcall=NaN nella cassa, C9 non
consapevole della direzione (S1>S0 non e' "vincente" per uno short), e due di
contabilita' che avrebbero ADULATO il collar (base che rollava lo spot
pagando ~3,6%/a di fee inesistenti; roll che chiudeva lo spot senza motivo).
Corregge anche un muro di §46: il tick da 5 USDC e' della famiglia USDC; la
catena che raccogliamo e' 100% inverse, quindi li' non si applica.
Regole nuove in CLAUDE.md: M29 (un edge da opzioni prezzate a modello si
riprezza alla vol REALIZZATA prima di crederci), M8 esteso (un argmax sul
BORDO e' una pendenza, non una cella), C4 esteso (il segno dell'asimmetria di
f dipende dal verso della gamba).
Libro, pesi, cron, config INVARIATI. Nessun ordine. Suite: 807 passati.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
716 lines
36 KiB
Python
716 lines
36 KiB
Python
"""COLLAR01 — hold BTC (long/short) con COLLAR di opzioni a <=15 giorni.
|
|
|
|
DOMANDA (posta dall'operatore il 2026-09-01):
|
|
"crea una strategia di hold BTC in long o short con copertura con options (max 15gg)"
|
|
+ due precisazioni in corso d'opera:
|
|
"opzione deve essere al max di 15gg" -> vincolo sulla SCADENZA, non sul solo holding
|
|
"voglio ridurre la vincita, ma bloccare la perdita" -> NON e' una put protettiva: e' un COLLAR
|
|
(pavimento comprato, tetto venduto per finanziarlo).
|
|
"l'entrata deve essere gestita da una strategia confortata da indicatori (es. forte bull)"
|
|
"dalle opzioni dobbiamo uscire prima del termine (tra 50% e 75% del tempo)" -> `exit_frac`.
|
|
|
|
PERCHE' QUESTO FILONE NON E' §46 (TAIL-HEDGE, REFUTATO il 2026-08-23).
|
|
§46 compro' put deep-OTM sul LIBRO (TP01+SKH01) e fu refutato *prima del prezzo*: il maxDD
|
|
saliva in 162/162 celle perche' **il beta del libro al sottostante e' +0,076** — "non si
|
|
assicura un libro che nei crash e' gia' quasi piatto".
|
|
Qui il sottostante e' un **hold di BTC nudo, beta 1,0 per costruzione**: il motivo che uccise
|
|
§46 NON si applica, ed e' l'unica ragione per cui questo filone puo' essere riaperto
|
|
(CLAUDE.md: "chi la riapre deve battere il motivo, non ripetere la misura").
|
|
§46 dichiara inoltre di NON aver provato la copertura dinamica e di aver misurato la sola
|
|
statica: anche quella porta e' aperta, ma qui non la uso — il collar e' statico a griglia.
|
|
|
|
COSA E' GIA' NOTO E VIENE RIUSATO, NON RIMISURATO:
|
|
- metriche e i due null di scala (`k_for_same_dd`, `k_for_same_drift`) da r0823_tail_hedge;
|
|
- listino fee opzioni Deribit (taker 0,03% del sottostante, cap 12,5% del premio, consegna
|
|
0,015%) dallo stesso;
|
|
- la catena reale via `cblib.load_chain()` (archivio 2026-05-01 + raccolta propria 2026-07-30).
|
|
|
|
LE DUE LENTI, ed e' una scelta dichiarata:
|
|
(A) LENTE LUNGA (decide): 2021-03-24 -> oggi, la finestra del DVOL — ~5,4 anni. Opzioni
|
|
prezzate a modello (BS, r=0) con DVOL come ancora ATM, CORRETTA da skew e struttura a
|
|
termine MISURATI SULLA CATENA VERA, e spread f misurato per ala. E' l'impianto di §46.
|
|
(B) LENTE REALE (controlla il segno, NON decide): 2026-05-01 -> oggi, solo quote a due lati.
|
|
123 giorni = ~8 cicli a 15 giorni: sotto-potenziata per costruzione, e per giunta su una
|
|
finestra in cui BTC e' salito da ~64,7k a ~77,5k, che e' AVVERSA a un collar (il tetto si
|
|
paga proprio li'). Si legge il segno, mai la taglia.
|
|
|
|
M5 (null del de-levering) E' IL PRIMO TEST, NON L'ULTIMO: ogni claim "meno drawdown" si misura
|
|
contro "tieni semplicemente meno BTC" a iso-maxDD. Se il collar non batte quello, non e'
|
|
protezione: e' una ri-scalatura pagata cara.
|
|
C9 e' il secondo: "una regola d'uscita che scatta piu' spesso sui vincenti che sui perdenti non e'
|
|
protezione, e' troncatura" — qui il tetto e' letteralmente quella regola.
|
|
|
|
NESSUN ORDINE. Libro, pesi, cron, config INVARIATI.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import math
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
from scipy.stats import norm
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
sys.path.insert(0, str(ROOT))
|
|
sys.path.insert(0, str(ROOT / "scripts" / "research"))
|
|
|
|
import cblib # noqa: E402
|
|
from src.data.downloader import load_data # noqa: E402
|
|
|
|
DPY = 365.25
|
|
|
|
# ---- listino Deribit opzioni (identico a r0823_tail_hedge, non ridichiarato a mano) -----------
|
|
OPT_TAKER_UNDERLYING = 0.0003
|
|
OPT_FEE_CAP_FRAC = 0.125
|
|
OPT_DELIVERY_UNDERLYING = 0.00015
|
|
|
|
# ---- GRIGLIA DICHIARATA PRIMA DI GUARDARE I RISULTATI -----------------------------------------
|
|
G_DPUT = (0.10, 0.20, 0.30) # |delta| della put comprata = il PAVIMENTO
|
|
G_DCALL = (0.10, 0.20, 0.30) # delta della call venduta = il TETTO
|
|
G_TENOR = (7, 14) # giorni a scadenza: <=15, vincolo dell'operatore
|
|
N_CELLS = len(G_DPUT) * len(G_DCALL) * len(G_TENOR)
|
|
|
|
# varianti dichiarate a parte (non nella griglia, per non gonfiare i trial)
|
|
V_ZEROCOST = "zero-cost" # tetto scelto per finanziare esattamente il pavimento
|
|
|
|
ATTESE = """
|
|
ATTESE A PRIORI — scritte e stampate PRIMA di misurare (M12: un follow-up contiene una previsione).
|
|
A1 Il maxDD SCENDE davvero, a differenza di §46: il beta e' 1,0 e il pavimento morde.
|
|
(se A1 fallisse, il filone cade come §46 e per lo stesso motivo)
|
|
A2 Il drift scende PIU' del maxDD, perche' il rendimento di BTC vive nella coda destra e il
|
|
tetto la taglia. Previsione: rapporto Δdrift/ΔmaxDD > 1.
|
|
A3 Il collar PERDE contro il null del de-levering a iso-maxDD in piu' della meta' delle celle.
|
|
E' la previsione centrale del filone.
|
|
A4 Lo skew rende il collar delta-simmetrico un DEBITO netto (put piu' cara della call).
|
|
[gia' misurato prima di scrivere: +31% / +19% / +12% a |d| 0,10 / 0,20 / 0,30 -> CONFERMATA]
|
|
A5 Il tick morde MENO che in §46: quella era la famiglia USDC (tick 5 USDC); la catena che
|
|
raccogliamo e' 100% inverse, con tick in BTC.
|
|
A6 Nella lente reale (BTC +20% in 123 giorni) il collar sembrera' pessimo: finestra avversa,
|
|
non evidenza. Va letta come segno, non come taglia.
|
|
A7 Il verso SHORT sara' peggiore del LONG: il drift di BTC e' positivo e lo short lo paga tutto,
|
|
col collar che ne cappa il solo rimbalzo favorevole.
|
|
"""
|
|
|
|
|
|
# ==================================================================== 0. METRICHE E NULL (da §46)
|
|
|
|
def _sh(x) -> float:
|
|
r = np.asarray(pd.Series(x).dropna().values, float)
|
|
return float(np.mean(r) / np.std(r) * math.sqrt(DPY)) if len(r) > 2 and np.std(r) > 0 else 0.0
|
|
|
|
|
|
def _dd(x) -> float:
|
|
r = np.asarray(pd.Series(x).dropna().values, float)
|
|
eq = np.cumprod(1.0 + r)
|
|
pk = np.maximum.accumulate(eq)
|
|
return float(np.max((pk - eq) / pk)) if len(eq) else 0.0
|
|
|
|
|
|
def _drift(x) -> float:
|
|
r = np.asarray(pd.Series(x).dropna().values, float)
|
|
if len(r) < 2:
|
|
return 0.0
|
|
tot = float(np.prod(1.0 + r))
|
|
return -1.0 if tot <= 0 else float(tot ** (DPY / len(r)) - 1.0)
|
|
|
|
|
|
def k_for_same_dd(base: np.ndarray, target_dd: float) -> float:
|
|
"""La scala costante che porta il maxDD del NUDO esattamente a quello del coperto (bisezione)."""
|
|
lo, hi = 1e-4, 1.0
|
|
if _dd(hi * base) <= target_dd:
|
|
return 1.0
|
|
for _ in range(60):
|
|
mid = 0.5 * (lo + hi)
|
|
if _dd(mid * base) > target_dd:
|
|
hi = mid
|
|
else:
|
|
lo = mid
|
|
return 0.5 * (lo + hi)
|
|
|
|
|
|
# ==================================================================== 1. CALIBRAZIONE DALLA CATENA
|
|
|
|
_BANDS = (0.05, 0.10, 0.15, 0.20, 0.25, 0.30, 0.40, 0.50)
|
|
|
|
|
|
def calibra_catena() -> dict:
|
|
"""Skew, struttura a termine e spread f: MISURATI sulla catena vera, mai assunti.
|
|
|
|
Restituisce funzioni di correzione applicate al DVOL (che e' un ATM a 30 giorni) per ottenere
|
|
l'IV di una gamba a delta e scadenza dati. Senza queste il modello prezzerebbe la put e la
|
|
call allo STESSO vol: l'errore andrebbe tutto a favore del collar (A4).
|
|
"""
|
|
df = cblib.load_chain()
|
|
b = df[df["asset"].eq("BTC")].copy()
|
|
b["exp"] = pd.to_datetime(b["exp"], utc=True)
|
|
b["ts"] = pd.to_datetime(b["ts"], utc=True)
|
|
b["dte"] = (b["exp"] - b["ts"]).dt.total_seconds() / 86400.0
|
|
q = b[(b["dte"] > 1) & (b["dte"] <= 15) & b["bid"].gt(0) & b["ask"].gt(0)
|
|
& b["iv"].notna() & b["delta"].notna()].copy()
|
|
q["ad"] = q["delta"].abs()
|
|
|
|
atm = q[q["ad"].between(0.45, 0.55)].groupby(["ts", "exp"])["iv"].mean().rename("iv_atm")
|
|
q = q.join(atm, on=["ts", "exp"])
|
|
q = q[q["iv_atm"].notna() & q["iv_atm"].gt(0)]
|
|
|
|
skew = {}
|
|
for t, nm in (("P", "put"), ("C", "call")):
|
|
xs, ys = [], []
|
|
for d in _BANDS:
|
|
lo, hi = d * 0.85, d * 1.15
|
|
s = q[q["option_type"].eq(t) & q["ad"].between(lo, hi)]
|
|
if len(s) >= 200:
|
|
xs.append(d)
|
|
ys.append(float((s["iv"] / s["iv_atm"]).median()))
|
|
skew[nm] = (np.array(xs), np.array(ys))
|
|
|
|
# struttura a termine: IV ATM a DTE t contro il DVOL 30g dello stesso istante
|
|
dv = cblib.dvol_series("BTC")
|
|
a = q[q["ad"].between(0.45, 0.55)].copy()
|
|
a["giorno"] = a["ts"].dt.floor("D")
|
|
dvd = dv.copy()
|
|
dvd.index = pd.to_datetime(dvd.index, utc=True).floor("D")
|
|
dvd = dvd.groupby(level=0).last()
|
|
a = a.join(dvd.rename("dvol"), on="giorno")
|
|
a = a[a["dvol"].notna() & a["dvol"].gt(0)]
|
|
a["ratio"] = (a["iv"] / 100.0) / (a["dvol"] / 100.0) if a["iv"].median() > 3 else a["iv"] / (a["dvol"] / 100.0)
|
|
ts_x, ts_y = [], []
|
|
for lo, hi in ((1, 4), (4, 8), (8, 12), (12, 15)):
|
|
s = a[a["dte"].between(lo, hi)]
|
|
if len(s) >= 200:
|
|
ts_x.append(0.5 * (lo + hi))
|
|
ts_y.append(float(s["ratio"].median()))
|
|
|
|
# spread relativo per ala e per banda di delta (mezza forchetta su mid) = il costo di f
|
|
spread = {}
|
|
for t, nm in (("P", "put"), ("C", "call")):
|
|
xs, ys = [], []
|
|
for d in _BANDS:
|
|
lo, hi = d * 0.85, d * 1.15
|
|
s = q[q["option_type"].eq(t) & q["ad"].between(lo, hi)]
|
|
s = s[s["mid"].gt(0)]
|
|
if len(s) >= 200:
|
|
xs.append(d)
|
|
ys.append(float((((s["ask"] - s["bid"]) / 2.0) / s["mid"]).median()))
|
|
spread[nm] = (np.array(xs), np.array(ys))
|
|
|
|
return dict(skew=skew, term=(np.array(ts_x), np.array(ts_y)), spread=spread,
|
|
n_quote=len(q), giorni=int(q["ts"].dt.date.nunique()),
|
|
span=(q["ts"].min(), q["ts"].max()))
|
|
|
|
|
|
def _interp(tab, x):
|
|
xs, ys = tab
|
|
if len(xs) == 0:
|
|
return 1.0
|
|
return float(np.interp(x, xs, ys))
|
|
|
|
|
|
# ==================================================================== 2. PREZZATORE
|
|
|
|
def _d1(S, K, T, sig):
|
|
return (math.log(S / K) + 0.5 * sig * sig * T) / (sig * math.sqrt(T))
|
|
|
|
|
|
def bs_call(S, K, T, sig):
|
|
if T <= 0 or sig <= 0:
|
|
return max(S - K, 0.0)
|
|
d1 = _d1(S, K, T, sig)
|
|
return S * norm.cdf(d1) - K * norm.cdf(d1 - sig * math.sqrt(T))
|
|
|
|
|
|
def bs_put(S, K, T, sig):
|
|
if T <= 0 or sig <= 0:
|
|
return max(K - S, 0.0)
|
|
d1 = _d1(S, K, T, sig)
|
|
return K * norm.cdf(-(d1 - sig * math.sqrt(T))) - S * norm.cdf(-d1)
|
|
|
|
|
|
def strike_da_delta(S, T, sig_atm, cal, delta, tipo):
|
|
"""Strike che realizza |delta| dato, risolvendo la circolarita' skew<->delta per punto fisso.
|
|
|
|
L'IV dipende dal delta (skew) e il delta dipende dallo strike che dipende dall'IV: due
|
|
iterazioni bastano (verificato: la terza muove lo strike di <0,1%).
|
|
"""
|
|
sig = sig_atm * _interp(cal["skew"][tipo], delta)
|
|
K = S
|
|
for _ in range(3):
|
|
z = norm.ppf(delta) if tipo == "call" else -norm.ppf(delta)
|
|
K = S * math.exp(0.5 * sig * sig * T - z * sig * math.sqrt(T))
|
|
sig = sig_atm * _interp(cal["skew"][tipo], delta)
|
|
return K, sig
|
|
|
|
|
|
def _fee(prem, S):
|
|
return min(OPT_TAKER_UNDERLYING * S, OPT_FEE_CAP_FRAC * prem) if prem > 0 else 0.0
|
|
|
|
|
|
# ==================================================================== 3. IL MOTORE DEL COLLAR
|
|
|
|
def serie_giornaliera() -> pd.DataFrame:
|
|
"""BTC giornaliero + DVOL, sulla finestra in cui ESISTONO entrambi (2021-03 in poi)."""
|
|
d = load_data("BTC", "1h")
|
|
idx = pd.DatetimeIndex(pd.to_datetime(d["datetime"], utc=True))
|
|
px = pd.Series(d["close"].values.astype(float), index=idx).resample("1D").last()
|
|
dv = pd.read_parquet(ROOT / "data" / "raw" / "dvol_btc.parquet")
|
|
iv = pd.Series(dv["close"].values.astype(float) / 100.0,
|
|
index=pd.to_datetime(dv["timestamp"], unit="ms", utc=True)).resample("1D").last()
|
|
P = pd.concat({"px": px, "iv": iv}, axis=1, join="inner").dropna()
|
|
return P
|
|
|
|
|
|
SPOT_FEE_SIDE = 0.00035 # Deribit taker reale dal 2026-08-01 (3,50 bps/lato), §8
|
|
|
|
|
|
def gate_trend(P: pd.DataFrame, forte: bool) -> np.ndarray:
|
|
"""L'ENTRATA, governata da indicatori — e presa dal progetto, non inventata qui.
|
|
|
|
`trend_portfolio.tsmom_blend` media tre `np.sign()` sugli orizzonti (30, 90, 180) giorni,
|
|
quindi assume SOLO i valori {-1, -1/3, +1/3, +1} (il bucket 2/3 non esiste: e' un errore gia'
|
|
corretto in §57). Questo da' a "forte bull" una definizione che non aggiunge NEMMENO UN
|
|
parametro nuovo:
|
|
forte=True -> |blend| == 1 : tutti e tre gli orizzonti concordi ("forte bull/bear")
|
|
forte=False -> |blend| >= 1/3: almeno la maggioranza (confronto dichiarato)
|
|
Il segno del blend da' la DIREZIONE (long o short), come chiesto.
|
|
"""
|
|
from src.strategies.trend_portfolio import tsmom_blend
|
|
c = P["px"].values.astype(float)
|
|
b = tsmom_blend(c, (30, 90, 180))
|
|
d = np.sign(b)
|
|
d[np.abs(b) < (0.99 if forte else 0.32)] = 0.0
|
|
return np.concatenate([[0.0], d[:-1]]) # ESEGUIBILE: si entra il giorno DOPO il segnale
|
|
|
|
|
|
def collar(P: pd.DataFrame, cal: dict, *, dput: float, dcall: float, tenor: int,
|
|
dirs: np.ndarray, zerocost: bool = False, senza_opzioni: bool = False,
|
|
prem_scale: float = 1.0, solo_pavimento: bool = False,
|
|
exit_frac: float = 1.0) -> dict:
|
|
"""Hold di BTC gated dal trend, coperto da un collar rollato ogni `tenor` giorni (<=15).
|
|
|
|
direzione +1 = LONG BTC -> compra PUT (pavimento), vende CALL (tetto)
|
|
direzione -1 = SHORT BTC -> compra CALL (pavimento), vende PUT (tetto)
|
|
`exit_frac` < 1 esce dalle opzioni PRIMA della scadenza (0,50-0,75 = richiesta dell'operatore
|
|
del 2026-09-01). Non e' un dettaglio di rollo: cambia il segno di due cose opposte.
|
|
- sulla gamba COMPRATA (il pavimento) uscire presto RECUPERA valore temporale: il theta di una
|
|
long accelera verso la scadenza, e §46 aveva gia' misurato che *"f si paga solo sulla parte
|
|
di valore che converge a intrinseco, quindi un roll anticipato non lo paga"*;
|
|
- sulla gamba VENDUTA (il tetto) uscire presto RINUNCIA al theta piu' veloce, cioe' proprio
|
|
alla parte che in questo filone stampava il premio di varianza.
|
|
Il prezzo di entrambe e' che si paga lo SPREAD in uscita su due gambe, invece di regolare a
|
|
intrinseco. Chi lo attiva compra meno short-vol e piu' protezione onesta: e' misurabile, ed e'
|
|
misurato nella sezione dedicata.
|
|
|
|
`prem_scale` e `solo_pavimento` esistono per i CONTROLLI (M15): con premio 0 e solo il
|
|
pavimento la struttura e' un pranzo gratis e DEVE vincere; con premio x10 DEVE perdere ovunque.
|
|
Un apparato che non riconosce il pranzo gratis e' guasto, e §46 lo scopri' a proprie spese.
|
|
|
|
`senza_opzioni=True` produce il MEDESIMO gated hold senza collar: e' la BASE del null M5, e
|
|
dev'essere generata dallo stesso motore o il confronto misurerebbe anche le differenze di
|
|
contabilita' (P1: il null deriva dal codice sorvegliato, non lo ridichiara).
|
|
|
|
Marcatura GIORNALIERA a modello: il pavimento vincola il PAYOFF A SCADENZA, non il mark
|
|
intermedio. Senza marcatura giornaliera il maxDD del collar sarebbe misurato solo ai roll e la
|
|
protezione sembrerebbe migliore di quanto e'.
|
|
"""
|
|
S = P["px"].values.astype(float)
|
|
A = P["iv"].values.astype(float)
|
|
n = len(S)
|
|
if n < tenor + 5:
|
|
return {}
|
|
|
|
tipo_di = {+1: ("put", "call"), -1: ("call", "put")}
|
|
nav = np.full(n, np.nan)
|
|
V = 1.0 # capitale normalizzato
|
|
cash, q, nopt = 1.0, 0.0, 0.0
|
|
Kl = Ks = np.nan
|
|
dc_vivo = 0.20 # delta del tetto VIVO: serve allo spread d'uscita anticipata
|
|
scad = -1
|
|
roll_at = -1 # quando si ESCE (<= scad): con exit_frac<1 e' PRIMA della scadenza
|
|
cur = 0
|
|
cicli = []
|
|
dc_scelti = [] # delta del tetto effettivamente usato a ogni apertura
|
|
apre = 0
|
|
|
|
def _mark(K, T, sig_atm, tipo, S_t):
|
|
if T <= 0 or math.isnan(K):
|
|
return 0.0 if math.isnan(K) else (max(K - S_t, 0.0) if tipo == "put" else max(S_t - K, 0.0))
|
|
sig = sig_atm * _interp(cal["term"], max(T * DPY, 1.0))
|
|
d1 = _d1(S_t, K, T, sig)
|
|
dl = norm.cdf(d1) if tipo == "call" else 1.0 - norm.cdf(d1)
|
|
sig *= _interp(cal["skew"][tipo], min(max(dl, 0.02), 0.60))
|
|
return bs_put(S_t, K, T, sig) if tipo == "put" else bs_call(S_t, K, T, sig)
|
|
|
|
def _chiudi(i, S_t, anche_spot=True):
|
|
"""Regola le opzioni vive; liquida lo spot SOLO se richiesto.
|
|
|
|
⚠️ Il flag esiste per un difetto che falserebbe tutto: al ROLL delle opzioni lo spot non
|
|
ha motivo di muoversi. Chiuderlo e riaprirlo pagherebbe 2x3,5 bps ogni `tenor` giorni
|
|
(a 7 giorni: ~3,6%/anno) di fee che nella realta' non si pagano.
|
|
"""
|
|
nonlocal cash, q, nopt, Kl, Ks, cur
|
|
if nopt > 0 and not senza_opzioni:
|
|
tl, ts = tipo_di[cur]
|
|
T_rem = max(scad - i, 0) / DPY
|
|
vl = _mark(Kl, T_rem, A[i], tl, S_t)
|
|
vs = 0.0 if solo_pavimento else _mark(Ks, T_rem, A[i], ts, S_t)
|
|
if T_rem > 0: # smontaggio anticipato: pago lo spread in USCITA
|
|
vl *= 1.0 - _interp(cal["spread"][tl], dput)
|
|
vs *= 1.0 + _interp(cal["spread"][ts], dc_vivo)
|
|
else:
|
|
for x in (vl, vs):
|
|
if x > 0:
|
|
cash -= nopt * OPT_DELIVERY_UNDERLYING * S_t
|
|
cash += nopt * (vl - vs)
|
|
nopt, Kl, Ks = 0.0, np.nan, np.nan
|
|
if anche_spot and q != 0.0:
|
|
cash += q * S_t - abs(q) * S_t * SPOT_FEE_SIDE
|
|
q = 0.0
|
|
|
|
for i in range(n):
|
|
S_t = S[i]
|
|
d = int(dirs[i])
|
|
fine_ciclo = (i >= roll_at) and cur != 0 and not senza_opzioni
|
|
solo_roll = fine_ciclo and d == cur
|
|
if d != cur or fine_ciclo:
|
|
if cur != 0:
|
|
if nopt > 0 and i >= scad:
|
|
tl, ts = tipo_di[cur]
|
|
il_ = max(Kl - S_t, 0.0) if tl == "put" else max(S_t - Kl, 0.0)
|
|
is_ = 0.0 if solo_pavimento else (
|
|
max(Ks - S_t, 0.0) if ts == "put" else max(S_t - Ks, 0.0))
|
|
cicli.append(dict(i0=apre, i1=i, S0=S[apre], S1=S_t, Kl=Kl, Ks=Ks,
|
|
dirn=cur, cap=is_ > 0, floor=il_ > 0,
|
|
scaduto=bool(fine_ciclo)))
|
|
_chiudi(i, S_t, anche_spot=not solo_roll)
|
|
if not solo_roll:
|
|
V = cash
|
|
if not solo_roll:
|
|
cur = d
|
|
if cur != 0:
|
|
if not solo_roll:
|
|
q = cur * V / S_t
|
|
cash = V - q * S_t - abs(q) * S_t * SPOT_FEE_SIDE
|
|
apre, scad = i, i + tenor
|
|
roll_at = i + max(1, int(round(exit_frac * tenor)))
|
|
if not senza_opzioni:
|
|
tl, ts = tipo_di[cur]
|
|
T = tenor / DPY
|
|
Kl, sl = strike_da_delta(S_t, T, A[i], cal, dput, tl)
|
|
pl = (bs_put if tl == "put" else bs_call)(S_t, Kl, T, sl)
|
|
pl *= 1.0 + _interp(cal["spread"][tl], dput) # compro all'ASK
|
|
pl *= prem_scale
|
|
if zerocost:
|
|
lo, hi = 0.02, 0.60
|
|
for _ in range(40):
|
|
md = 0.5 * (lo + hi)
|
|
Kx, sx = strike_da_delta(S_t, T, A[i], cal, md, ts)
|
|
px_ = (bs_put if ts == "put" else bs_call)(S_t, Kx, T, sx)
|
|
px_ *= 1.0 - _interp(cal["spread"][ts], md) # vendo al BID
|
|
# il premio CRESCE col delta: se incasso meno di quanto pago,
|
|
# il tetto va avvicinato (delta PIU' ALTO) -> lo = md.
|
|
if px_ < pl:
|
|
lo = md
|
|
else:
|
|
hi = md
|
|
dc_use = 0.5 * (lo + hi)
|
|
else:
|
|
dc_use = dcall
|
|
dc_vivo = dc_use
|
|
dc_scelti.append(float(dc_use))
|
|
Ks, ss = strike_da_delta(S_t, T, A[i], cal, dc_use, ts)
|
|
ps = (bs_put if ts == "put" else bs_call)(S_t, Ks, T, ss)
|
|
ps *= (1.0 - _interp(cal["spread"][ts], dc_use)) * prem_scale
|
|
if solo_pavimento:
|
|
ps, Ks = 0.0, np.inf if ts == "call" else 0.0
|
|
nopt = abs(q)
|
|
cash += nopt * (ps - pl) - nopt * (_fee(pl, S_t) + _fee(ps, S_t))
|
|
vl = vs = 0.0
|
|
if nopt > 0 and not senza_opzioni:
|
|
tl, ts = tipo_di[cur]
|
|
T_rem = max(scad - i, 0) / DPY
|
|
vl = _mark(Kl, T_rem, A[i], tl, S_t)
|
|
vs = 0.0 if solo_pavimento else _mark(Ks, T_rem, A[i], ts, S_t)
|
|
nav[i] = cash + q * S_t + nopt * (vl - vs)
|
|
|
|
v = pd.Series(nav, index=P.index).dropna()
|
|
r = (v / v.shift(1) - 1.0).dropna()
|
|
return dict(ret=r, nav=v, cicli=cicli, dc_scelti=dc_scelti)
|
|
|
|
|
|
# ==================================================================== 4. MAIN
|
|
|
|
def _riga(nome, r) -> dict:
|
|
return dict(nome=nome, sh=_sh(r), dd=_dd(r), drift=_drift(r))
|
|
|
|
|
|
def main() -> None:
|
|
print("=" * 96)
|
|
print(" COLLAR01 — hold BTC gated dal trend, coperto da collar di opzioni a <=15 giorni")
|
|
print("=" * 96)
|
|
print(ATTESE)
|
|
|
|
cal = calibra_catena()
|
|
print(f" CALIBRAZIONE dalla catena VERA: {cal['n_quote']:,} quote a due lati su "
|
|
f"{cal['giorni']} giorni, {cal['span'][0].date()} -> {cal['span'][1].date()}")
|
|
for nm in ("put", "call"):
|
|
xs, ys = cal["skew"][nm]
|
|
print(f" skew {nm:5s} IV/ATM : " + " ".join(f"d{x:.2f}={y:.3f}" for x, y in zip(xs, ys)))
|
|
xs, ys = cal["term"]
|
|
print(f" termine IV/DVOL30 : " + " ".join(f"{x:.0f}g={y:.3f}" for x, y in zip(xs, ys)))
|
|
for nm in ("put", "call"):
|
|
xs, ys = cal["spread"][nm]
|
|
print(f" spread {nm:5s} (mezza forchetta/mid): "
|
|
+ " ".join(f"d{x:.2f}={y:.3f}" for x, y in zip(xs, ys)))
|
|
|
|
P = serie_giornaliera()
|
|
print(f"\n LENTE LUNGA: {P.index[0].date()} -> {P.index[-1].date()} "
|
|
f"({len(P)} giorni, {len(P)/DPY:.2f} anni)")
|
|
|
|
bh = (P["px"] / P["px"].shift(1) - 1.0).dropna()
|
|
print(f" BTC buy&hold nudo: Sharpe {_sh(bh):.3f} · maxDD {100*_dd(bh):.2f}% · "
|
|
f"drift {100*_drift(bh):+.2f}%/a")
|
|
|
|
righe = []
|
|
for forte in (True, False):
|
|
dirs = gate_trend(P, forte)
|
|
nome_g = "FORTE (3/3 orizzonti)" if forte else "LARGO (>=2/3)"
|
|
quota = 100.0 * np.mean(dirs != 0)
|
|
qlong = 100.0 * np.mean(dirs > 0)
|
|
base = collar(P, cal, dput=0.2, dcall=0.2, tenor=7, dirs=dirs, senza_opzioni=True)
|
|
rb = base["ret"]
|
|
print(f"\n ── GATE {nome_g}: a mercato {quota:.1f}% dei giorni "
|
|
f"(long {qlong:.1f}%, short {quota-qlong:.1f}%)")
|
|
print(f" BASE gated SENZA opzioni: Sharpe {_sh(rb):.3f} · maxDD {100*_dd(rb):.2f}% · "
|
|
f"drift {100*_drift(rb):+.2f}%/a ← e' il NULL da battere (M5)")
|
|
for tenor in G_TENOR:
|
|
for dp in G_DPUT:
|
|
for dc in G_DCALL:
|
|
out = collar(P, cal, dput=dp, dcall=dc, tenor=tenor, dirs=dirs)
|
|
if not out:
|
|
continue
|
|
r = out["ret"]
|
|
dd_c, dr_c = _dd(r), _drift(r)
|
|
k = k_for_same_dd(rb.values, dd_c)
|
|
dr_null = _drift(k * rb.values)
|
|
righe.append(dict(gate=nome_g, forte=forte, tenor=tenor, dp=dp, dc=dc,
|
|
sh=_sh(r), dd=dd_c, drift=dr_c, k=k, drift_null=dr_null,
|
|
vince=(dr_c > dr_null) and (dd_c < _dd(rb)),
|
|
dd_scende=dd_c < _dd(rb), cicli=out["cicli"], ret=r))
|
|
# varianti zero-cost, dichiarate FUORI griglia
|
|
for tenor in G_TENOR:
|
|
for dp in G_DPUT:
|
|
out = collar(P, cal, dput=dp, dcall=np.nan, tenor=tenor, dirs=dirs, zerocost=True)
|
|
if not out:
|
|
continue
|
|
r = out["ret"]
|
|
dd_c, dr_c = _dd(r), _drift(r)
|
|
k = k_for_same_dd(rb.values, dd_c)
|
|
righe.append(dict(gate=nome_g, forte=forte, tenor=tenor, dp=dp, dc="zc",
|
|
sh=_sh(r), dd=dd_c, drift=dr_c, k=k,
|
|
drift_null=_drift(k * rb.values),
|
|
vince=(dr_c > _drift(k * rb.values)) and (dd_c < _dd(rb)),
|
|
dd_scende=dd_c < _dd(rb), cicli=out["cicli"], ret=r))
|
|
R = pd.DataFrame([x for x in righe if x["gate"] == nome_g])
|
|
print(f"\n {'ten':>4} {'dput':>5} {'dcall':>6} {'Sharpe':>7} {'maxDD':>7} "
|
|
f"{'drift':>8} {'k null':>7} {'drift null':>10} esito")
|
|
for _, x in R.iterrows():
|
|
dcs = x["dc"] if isinstance(x["dc"], str) else f"{x['dc']:.2f}"
|
|
print(f" {x['tenor']:>4} {x['dp']:>5.2f} {dcs:>6} {x['sh']:>7.3f} "
|
|
f"{100*x['dd']:>6.2f}% {100*x['drift']:>7.2f}% {x['k']:>7.3f} "
|
|
f"{100*x['drift_null']:>9.2f}% "
|
|
f"{'VINCE' if x['vince'] else ('dd↓ ma perde' if x['dd_scende'] else 'dd↑')}")
|
|
|
|
R = pd.DataFrame(righe)
|
|
n_dd = int(R["dd_scende"].sum())
|
|
n_win = int(R["vince"].sum())
|
|
print("\n" + "=" * 96)
|
|
print(f" A1 maxDD SCENDE in {n_dd}/{len(R)} celle "
|
|
f"({'CONFERMATA' if n_dd > len(R)//2 else 'REFUTATA'})")
|
|
print(f" A3 il collar BATTE il de-levering in {n_win}/{len(R)} celle "
|
|
f"({'REFUTATA (il collar vince)' if n_win > len(R)//2 else 'CONFERMATA (il collar perde)'})")
|
|
|
|
# ---- A2: rapporto Δdrift/ΔmaxDD ----------------------------------------------------------
|
|
for forte in (True, False):
|
|
sub = R[R["forte"] == forte]
|
|
if not len(sub):
|
|
continue
|
|
dirs = gate_trend(P, forte)
|
|
rb = collar(P, cal, dput=0.2, dcall=0.2, tenor=7, dirs=dirs, senza_opzioni=True)["ret"]
|
|
d0, g0 = _dd(rb), _drift(rb)
|
|
rap = ((g0 - sub["drift"]) / np.maximum(d0 - sub["dd"], 1e-9)).median()
|
|
print(f" A2 gate {'FORTE' if forte else 'LARGO'}: Δdrift/ΔmaxDD mediano {rap:.2f} "
|
|
f"({'CONFERMATA' if rap > 1 else 'REFUTATA'}: >1 = il tetto costa piu' di quanto il "
|
|
f"pavimento renda)")
|
|
|
|
# ---- C9: il tetto scatta piu' sui vincenti o sui perdenti? -------------------------------
|
|
best = R.sort_values("sh", ascending=False).iloc[0]
|
|
cic = pd.DataFrame(best["cicli"])
|
|
if len(cic):
|
|
# ⚠️ "vincente" dipende dalla DIREZIONE: per un ciclo SHORT un prezzo che sale e' una
|
|
# PERDITA. Usare S1>S0 come proxy di vittoria e' il difetto che dava "pavimento 0,0%".
|
|
cic["vinc"] = cic["dirn"] * (cic["S1"] - cic["S0"]) > 0
|
|
cap_w = 100.0 * cic[cic["vinc"]]["cap"].mean()
|
|
flo_l = 100.0 * cic[~cic["vinc"]]["floor"].mean()
|
|
print(f"\n C9 cella migliore (gate {best['gate']}, {best['tenor']}g, dput {best['dp']}, "
|
|
f"dcall {best['dc']}), {len(cic)} cicli:")
|
|
print(f" il TETTO taglia nel {cap_w:.1f}% dei cicli VINCENTI · "
|
|
f"il PAVIMENTO para nel {flo_l:.1f}% dei cicli PERDENTI")
|
|
print(f" -> {'TRONCATURA' if cap_w > flo_l else 'PROTEZIONE'} "
|
|
f"(C9: se scatta piu' sui vincenti che sui perdenti, non e' protezione)")
|
|
|
|
# ---- M1: il confronto che decide se merita uno slot --------------------------------------
|
|
try:
|
|
from src.portfolio.sleeves import _tp01_returns
|
|
tp = _tp01_returns()
|
|
tp.index = pd.to_datetime(tp.index, utc=True)
|
|
tpd = (1.0 + tp).resample("1D").prod() - 1.0
|
|
j = pd.concat({"c": best["ret"], "tp": tpd}, axis=1, join="inner").dropna()
|
|
print(f"\n M1 vs TP01 sulla finestra comune ({len(j)} giorni): "
|
|
f"collar Sharpe {_sh(j['c']):.3f} · TP01 {_sh(j['tp']):.3f} · "
|
|
f"corr {j['c'].corr(j['tp']):+.3f}")
|
|
for w in (0.10, 0.25):
|
|
mix = (1 - w) * j["tp"] + w * j["c"]
|
|
print(f" TP01 + {100*w:.0f}% collar: Sharpe {_sh(mix):.3f} "
|
|
f"({_sh(mix)-_sh(j['tp']):+.3f}) · maxDD {100*_dd(mix):.2f}% "
|
|
f"({100*(_dd(mix)-_dd(j['tp'])):+.2f}pp)")
|
|
except Exception as e: # pragma: no cover
|
|
print(f"\n M1 non calcolato: {type(e).__name__}: {e}")
|
|
|
|
# ---- ESTENSIONE AL BORDO (M4: si riapre la FAMIGLIA, M8: un argmax sul bordo non decide) --
|
|
print("\n" + "=" * 96)
|
|
print(" ESTENSIONE DICHIARATA — le celle vincenti stanno tutte sul BORDO (dput min, dcall max).")
|
|
print(" Sondo dove punta la pendenza. Il limite dput->0 / dcall->0,50 e' la COVERED CALL pura:")
|
|
print(" nessun pavimento, tetto ATM. Trial aggiunti: 2 gate x 2 tenor x 3 x 3 = 36 (dichiarati).")
|
|
ext = []
|
|
for forte in (True, False):
|
|
dirs = gate_trend(P, forte)
|
|
rb = collar(P, cal, dput=0.2, dcall=0.2, tenor=7, dirs=dirs, senza_opzioni=True)["ret"]
|
|
for tenor in (7, 14):
|
|
for dp in (0.02, 0.05, 0.10):
|
|
for dc in (0.30, 0.40, 0.50):
|
|
r = collar(P, cal, dput=dp, dcall=dc, tenor=tenor, dirs=dirs)["ret"]
|
|
k = k_for_same_dd(rb.values, _dd(r))
|
|
ext.append(dict(forte=forte, tenor=tenor, dp=dp, dc=dc, sh=_sh(r),
|
|
dd=_dd(r), drift=_drift(r),
|
|
vince=(_drift(r) > _drift(k * rb.values)) and (_dd(r) < _dd(rb))))
|
|
E = pd.DataFrame(ext)
|
|
top = E.sort_values("sh", ascending=False).iloc[0]
|
|
print(f" -> VINCE in {int(E['vince'].sum())}/{len(E)} celle, e il massimo e' NELL'ANGOLO: "
|
|
f"gate {'FORTE' if top['forte'] else 'LARGO'} {top['tenor']}g dput {top['dp']:.2f} "
|
|
f"dcall {top['dc']:.2f} -> Sharpe {top['sh']:.3f}, maxDD {100*top['dd']:.2f}%, "
|
|
f"drift {100*top['drift']:+.2f}%/a")
|
|
print(" ⚠️ Il pavimento non e' cio' che vince: vince il TETTO. La pendenza porta FUORI da cio'")
|
|
print(" che l'operatore ha chiesto ('bloccare la perdita') e DENTRO lo short-vol.")
|
|
|
|
# ---- LA DIAGNOSTICA CHE DECIDE: quanto dell'edge e' il PREMIO DI VARIANZA? ----------------
|
|
lr = np.log(P["px"] / P["px"].shift(1))
|
|
print("\n RIPREZZATURA ALLA VOLATILITA' REALIZZATA — diagnostica con LOOK-AHEAD DICHIARATO,")
|
|
print(" non e' una strategia: e' il valore equo ex-post. Se l'edge sparisce, era IV-RV.")
|
|
for h in (7, 14):
|
|
fw = (lr[::-1].rolling(h).std()[::-1] * math.sqrt(DPY)).bfill().ffill()
|
|
j = pd.concat({"iv": P["iv"], "fw": fw}, axis=1).dropna()
|
|
print(f" DVOL / RV-forward a {h}g: mediana {(j['iv']/j['fw']).median():.3f} · "
|
|
f"DVOL sopra nel {100*(j['iv']>j['fw']).mean():.1f}% dei giorni")
|
|
for tenor in (7, 14):
|
|
fw = (lr[::-1].rolling(tenor).std()[::-1] * math.sqrt(DPY)).bfill().ffill()
|
|
Pf = P.copy()
|
|
Pf["iv"] = fw.reindex(P.index).values
|
|
for forte in (True, False):
|
|
dirs = gate_trend(P, forte)
|
|
rb = collar(P, cal, dput=0.2, dcall=0.2, tenor=7, dirs=dirs, senza_opzioni=True)["ret"]
|
|
for dp, dc in ((0.02, 0.50), (0.10, 0.30)):
|
|
a = collar(P, cal, dput=dp, dcall=dc, tenor=tenor, dirs=dirs)["ret"]
|
|
b = collar(Pf, cal, dput=dp, dcall=dc, tenor=tenor, dirs=dirs)["ret"]
|
|
ka, kb = k_for_same_dd(rb.values, _dd(a)), k_for_same_dd(rb.values, _dd(b))
|
|
wa = _drift(a) > _drift(ka * rb.values) and _dd(a) < _dd(rb)
|
|
wb = _drift(b) > _drift(kb * rb.values) and _dd(b) < _dd(rb)
|
|
print(f" {'FORTE' if forte else 'LARGO'} {tenor:>2}g dput {dp:.2f} dcall {dc:.2f}: "
|
|
f"a DVOL Sh {_sh(a):6.3f} drift {100*_drift(a):+7.2f}% {'VINCE' if wa else 'perde'}"
|
|
f" | a RV Sh {_sh(b):6.3f} drift {100*_drift(b):+7.2f}% {'VINCE' if wb else 'perde'}"
|
|
f" | VRP {100*(_drift(a)-_drift(b)):+6.2f} pp")
|
|
|
|
# ---- USCITA ANTICIPATA (richiesta dell'operatore: fra il 50% e il 75% del tempo) ---------
|
|
print("\n" + "=" * 96)
|
|
print(" USCITA ANTICIPATA DALLE OPZIONI — exit_frac 0,50 / 0,625 / 0,75 contro la scadenza.")
|
|
print(" §46 misuro' che 'f si paga solo sulla parte di valore che converge a intrinseco,")
|
|
print(" quindi un roll anticipato NON lo paga'. ⚠️ Vale per una copertura SOLO LONG: in un")
|
|
print(" collar c'e' una gamba VENDUTA da RICOMPRARE, e uscire prima paga f esattamente sulla")
|
|
print(" parte che a scadenza si sarebbe regolata gratis. L'asimmetria si INVERTE.")
|
|
lr2 = np.log(P["px"] / P["px"].shift(1))
|
|
for forte in (True, False):
|
|
dirs = gate_trend(P, forte)
|
|
rb = collar(P, cal, dput=0.2, dcall=0.2, tenor=7, dirs=dirs, senza_opzioni=True)["ret"]
|
|
print(f"\n gate {'FORTE' if forte else 'LARGO'} (base drift {100*_drift(rb):+.2f}%/a):")
|
|
for dp, dc in ((0.10, 0.30), (0.20, 0.20), (0.02, 0.50)):
|
|
for ef in (1.00, 0.75, 0.625, 0.50):
|
|
a = collar(P, cal, dput=dp, dcall=dc, tenor=14, dirs=dirs, exit_frac=ef)["ret"]
|
|
hh = max(1, int(round(ef * 14)))
|
|
fw = (lr2[::-1].rolling(hh).std()[::-1] * math.sqrt(DPY)).bfill().ffill()
|
|
Pf = P.copy()
|
|
Pf["iv"] = fw.reindex(P.index).values
|
|
b = collar(Pf, cal, dput=dp, dcall=dc, tenor=14, dirs=dirs, exit_frac=ef)["ret"]
|
|
k = k_for_same_dd(rb.values, _dd(a))
|
|
w = (_drift(a) > _drift(k * rb.values)) and (_dd(a) < _dd(rb))
|
|
print(f" d{dp:.2f}/{dc:.2f} exit {ef:.3f}: Sh {_sh(a):6.3f} DD {100*_dd(a):6.2f}% "
|
|
f"drift {100*_drift(a):+7.2f}% null {100*_drift(k*rb.values):+6.2f}% "
|
|
f"{'VINCE' if w else 'perde'} | a RV drift {100*_drift(b):+7.2f}% "
|
|
f"VRP {100*(_drift(a)-_drift(b)):+6.2f}pp")
|
|
|
|
# ---- CONTROLLI DELL'APPARATO (M15) — un rilevatore che non li passa non misura nulla ----
|
|
print("\n CONTROLLI DELL'APPARATO:")
|
|
dirs = gate_trend(P, True)
|
|
rb = collar(P, cal, dput=0.2, dcall=0.2, tenor=7, dirs=dirs, senza_opzioni=True)["ret"]
|
|
esiti = []
|
|
g = collar(P, cal, dput=0.20, dcall=0.20, tenor=14, dirs=dirs,
|
|
prem_scale=0.0, solo_pavimento=True)["ret"]
|
|
ok_g = (_dd(g) < _dd(rb)) and (_drift(g) > _drift(rb))
|
|
esiti.append(ok_g)
|
|
print(f" pranzo gratis (pavimento a premio ZERO): maxDD {100*_dd(rb):.2f}%->{100*_dd(g):.2f}% "
|
|
f"drift {100*_drift(rb):+.2f}%->{100*_drift(g):+.2f}% {'OK' if ok_g else 'FALLITO'}")
|
|
c = collar(P, cal, dput=0.20, dcall=0.20, tenor=14, dirs=dirs, prem_scale=10.0)["ret"]
|
|
ok_c = _drift(c) < _drift(rb)
|
|
esiti.append(ok_c)
|
|
print(f" premio x10 (deve perdere): drift {100*_drift(c):+.2f}%/a "
|
|
f"{'OK' if ok_c else 'FALLITO'}")
|
|
zc = collar(P, cal, dput=0.20, dcall=np.nan, tenor=14, dirs=dirs, zerocost=True)
|
|
prem_res = np.nan
|
|
ok_z = bool(len(zc.get("cicli", [])) > 0) and np.isfinite(_drift(zc["ret"]))
|
|
esiti.append(ok_z)
|
|
print(f" zero-cost costruibile e finito: drift {100*_drift(zc['ret']):+.2f}%/a "
|
|
f"{'OK' if ok_z else 'FALLITO'}")
|
|
if not all(esiti):
|
|
print(" 🚨 APPARATO GUASTO: il verdetto sotto NON e' leggibile (§46: un controllo "
|
|
"positivo rotto dichiara guasto l'apparato)")
|
|
|
|
# ---- VERDETTO A RUNTIME (N11) ------------------------------------------------------------
|
|
print("\n" + "=" * 96)
|
|
# il verdetto deve tenere conto della diagnostica VRP: il bordo vince, ma di che cosa?
|
|
dirs_f = gate_trend(P, True)
|
|
rb_f = collar(P, cal, dput=0.2, dcall=0.2, tenor=7, dirs=dirs_f, senza_opzioni=True)["ret"]
|
|
fw7 = (np.log(P["px"] / P["px"].shift(1))[::-1].rolling(7).std()[::-1]
|
|
* math.sqrt(DPY)).bfill().ffill()
|
|
Pf7 = P.copy()
|
|
Pf7["iv"] = fw7.reindex(P.index).values
|
|
ang_dvol = collar(P, cal, dput=0.02, dcall=0.50, tenor=7, dirs=dirs_f)["ret"]
|
|
ang_rv = collar(Pf7, cal, dput=0.02, dcall=0.50, tenor=7, dirs=dirs_f)["ret"]
|
|
quota_vrp = (_drift(ang_dvol) - _drift(ang_rv)) / max(_drift(ang_dvol), 1e-9)
|
|
|
|
if n_dd == 0:
|
|
v = "IL PAVIMENTO NON RIDUCE IL maxDD — cade come §46 e per lo stesso motivo"
|
|
elif n_win <= len(R) // 8:
|
|
v = (f"IL PAVIMENTO FUNZIONA (a differenza di §46: maxDD scende in {n_dd}/{len(R)}) MA IL "
|
|
f"TETTO LO PAGA TROPPO: solo {n_win}/{len(R)} celle battono il de-levering. "
|
|
f"E cio' che vince sul bordo NON e' il pavimento: e' il premio di varianza "
|
|
f"({100*quota_vrp:.0f}% del drift dell'angolo sparisce riprezzando alla vol vera) "
|
|
f"— cioe' il meccanismo di VRP01, non una scoperta, e §3 lo blocca in deploy")
|
|
else:
|
|
v = f"REGGE IL PRIMO GATE: {n_win}/{len(R)} celle battono il de-levering — serve M2/DSR"
|
|
print(f" VERDETTO (calcolato a runtime): {v}")
|
|
print("=" * 96)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|