Adriano Dal Pastro
5700280f65
§46. Prima misura del progetto sull'acquisto di put deep-OTM come assicurazione statica
sul libro (TP01 75 / SKH01 25), senza delta-hedge e senza pretendere alpha. Motore prezzi
RIUSATO da VRP01 (_bs_put/_strike_from_delta su DVOL reale). Griglia dichiarata prima:
3 delta x 3 tenori x 3 budget x 2 roll = 54 celle per lente di f (162 in tutto).
VERDETTO: il null del de-levering non arriva nemmeno a essere il gate che decide — la
premessa cade prima. Il maxDD SCENDE in 0/54 celle a OGNI lente di f (compresa f=1.00,
cioe' regalando alla copertura il prezzo di modello): 9,42% -> 9,53-10,37%. Meccanismo:
il bleed del premio cade DENTRO i drawdown, che su questo libro sono lunghi e poco
profondi, mentre il payoff cade su singoli giorni di crollo che NON sono il fondo del
drawdown. Beta del libro al sottostante +0,076 (il 19/05/2021 il mercato ha fatto -21,1%
e il libro -2,0%): la put protegge una perdita gia' ridotta di ~10x dalla strategia
stessa, quindi coprirla costerebbe ~1/beta volte il budget.
ATTESA A PRIORI, riportata anche dove sbagliata: (c) REFUTATA nella forma scritta — 15/20
dei peggiori giorni SONO crolli, e lo short-squeeze che avevo in mente sta fuori finestra;
(b) e (a) confermate, e (b) e' il meccanismo operativo.
f MISURATO OGGI sulle quote vere (420 put con bid E ask, catena USDC): 1,89 / 1,27 / 1,06
a delta -0,05 / -0,10 / -0,15 — piu' mite del 2,23-5,85 del 30/07, e il verdetto non
cambia. Struttura del modello che vale come risultato: f si paga solo sulla parte di
valore che converge a INTRINSECO, quindi un roll anticipato non lo paga (sensitivita' con
f asimmetrico riportata).
ESEGUIBILITA': corregge un muro ereditato. Comprare un'opzione costa il PREMIO, non il
NOZIONALE — i "lotti da migliaia di dollari" del 22/08 sono il collaterale per VENDERE.
Il lotto minimo BTC_USDC (0,01) costa $1,79-7,93; la copertura diventa comprabile a ogni
roll da ~$9,2k (30g, delta -0,05) a ~$38,8k (7g, delta -0,15). Secondo muro, STRUTTURALE:
il tick da 5 USDC — piu' la copertura e' deep-OTM (cioe' economica) piu' il tick domina.
CANALE FUNDED: sull'unica regola binding (max-loss 6% statico) la copertura va nel verso
sbagliato — alza il maxDD e quindi abbassa la leva ammessa (k 0,629 -> 0,594-0,620).
DUE ERRORI MIEI CATTURATI PRIMA DI PUBBLICARE, entrambi dal controllo e non a occhio:
(1) il controllo positivo "payoff gratis" FALLIVA perche' non accreditavo il valore
dell'asset regalato, mentre il MTM successivo ne addebitava il decadimento -> il free
lunch costava quanto il premio. Un controllo positivo rotto dichiara guasto l'apparato.
(2) la prova di causalita' dava 1,28e-04 ("DIVERGE"): confrontavo fino al taglio, ma il
prefisso si ferma un roll prima -> differiva per ASSENZA di un roll, non per
look-ahead. Finestra corretta: 0,00e+00 esatto.
(3) e un errore di CRITERIO: contavo "vittoria" anche dove il maxDD PEGGIORA, dove il null
del de-levering e' degenere (k=1) e un Δdrift>0 risponde a una domanda di alpha, non a
questa. Col criterio corretto le "3/54 vittorie" a f=1 diventano 0/54.
Controlli 3/3 OK (free lunch e f=0,1 riconosciuti, f=10 rifiutato). Causalita' esatta.
Lente wick accoppiata riusata da r0725_prop_coupled, copertura 100% del pannello.
Distorsioni dichiarate: finestra 2021-03+ (il DVOL non esiste prima, quindi marzo 2020 e'
FUORI CAMPIONE); spread misurato sui soli strumenti con bid E ask = pavimento che favorisce
la copertura; 3-5 breach in 5,4 anni non distinguono le varianti.
Libro, pesi, cron, config INVARIATI. Nessun ordine.
882 lines
48 KiB
Python
882 lines
48 KiB
Python
"""r0823_tail_hedge — §46 TAIL-HEDGE: comprare put deep-OTM come ASSICURAZIONE STATICA sul libro.
|
|
|
|
LA DOMANDA (decidibile, e NON e' una domanda di alpha)
|
|
Il progetto ha gia' testato la VENDITA di volatilita' (VRP01, gated su IV-rank), il GAMMA
|
|
SCALPING (long-vol delta-hedgiato, SCARTATO: perde ogni anno) e le strutture Albimarini.
|
|
Non ha mai testato la cosa piu' semplice: **comprare put deep-OTM e tenerle**, senza
|
|
delta-hedge, senza pretendere alpha. E' un riduttore di DRAWDOWN, e il drawdown e' il vincolo
|
|
binding su due assi gia' misurati: (a) il canale funded (daily-loss 4%, max-loss 6%), (b) la
|
|
rendita perpetua, che vive sul DD (il libro rende quanto un ETF S&P e serve 2,5x meno capitale
|
|
solo perche' perde meno).
|
|
Domanda: **comprare protezione di coda compra riduzione di DD a un prezzo migliore di quello
|
|
che si ottiene semplicemente DE-LEVERANDO?**
|
|
|
|
=========================== ATTESA A PRIORI (scritta PRIMA di misurare) ===========================
|
|
Mi aspetto **REFUTATO al f realistico**, e forse marginale a f=1. Tre ragioni, in ordine di forza:
|
|
(c) **Si sta comprando assicurazione contro il rischio sbagliato.** Il 22/08 ha misurato che il
|
|
peggior giorno del libro NON e' un crash: e' uno SHORT SQUEEZE di SKH01 (short con stop 2%
|
|
che gappa a -15,77% di sleeve mentre TP01 e' flat). Una PUT non protegge da un rialzo.
|
|
(b) **Il libro e' GIA' difensivo dove la put paga.** L'esposizione di TP01 scende MONOTONA col
|
|
crollo (0,096 -> 0,042 sotto il -20%, misurato 22/08) perche' li' il trend si e' gia' girato:
|
|
la put paga proprio quando il libro non ha piu' esposizione da proteggere.
|
|
(a) **Comprare vol e' pagare il VRP**, cioe' l'esatto specchio di cio' che VRP01 incassa e che il
|
|
gamma scalping ha gia' pagato perdendo ogni anno, ogni variante, ogni frequenza.
|
|
Previsione quantitativa dichiarata: la riduzione di DD ci sara' (la put paga davvero nei crolli),
|
|
ma il de-levering la comprera' a meno; e il costo salira' ~linearmente col budget mentre il
|
|
beneficio saturera'.
|
|
==================================================================================================
|
|
|
|
IL GATE CHE DECIDE — NULL DEL DE-LEVERING (6 occorrenze nel progetto: VRP-DD, TP01xDVOL, MAT01,
|
|
azioni intere UCITS, SOL, TP01-SIZE). Si trova il `k` costante per cui `k*libro` ha lo STESSO
|
|
maxDD della variante coperta, e si confronta Sharpe e DRIFT a quel punto.
|
|
⚠️ Corollario 22/08: quando la variante e' una pura ri-scalatura il null e' DEGENERE
|
|
(`sh(k*base) === sh(base)`, identita' aritmetica). Qui NON lo e' **sul drift**: il drift
|
|
geometrico di `k*base` cambia con k, mentre lo Sharpe no. Quindi il confronto ha DUE gambe:
|
|
* Sharpe: Sharpe(coperto) vs Sharpe(base) [ = Sharpe(k*base), identicamente ]
|
|
* Drift: drift(coperto) vs drift(k*base) [ questa e' la gamba informativa ]
|
|
La non-linearita' della put e' cio' che rende la PRIMA gamba non-degenere (la copertura puo'
|
|
davvero alzare lo Sharpe, una ri-scalatura no).
|
|
|
|
COSA C'E' DENTRO
|
|
1) Motore prezzi RIUSATO (`src.portfolio.sleeves._bs_put` / `_strike_from_delta`, gli stessi di
|
|
VRP01) su DVOL reale. Premio **MODELLATO**, non quotato: dichiarato come tale.
|
|
2) f MISURATO OGGI sulle quote vere della catena Deribit (BTC_USDC + inverse), sulle ali che
|
|
questa strategia comprerebbe — non assunto dal 30/07.
|
|
3) Griglia dichiarata prima: delta {-0.05,-0.10,-0.15} x tenore {7,14,30} x budget
|
|
{0.5%,1%,2%}/anno x roll {scadenza, meta' vita} = **54 celle** per lente di f.
|
|
4) Lente di CODA (non media): maxDD, 5 peggiori mesi, peggior giorno, breach 4%/6% con la
|
|
**lente wick accoppiata** (riusata da r0725_prop_coupled, non riscritta).
|
|
5) Eseguibilita' misurata DAL VENUE (min_trade_amount + tick), non assunta.
|
|
6) Controlli positivo e negativo obbligatori.
|
|
|
|
CONVENZIONI DICHIARATE
|
|
* Libro = TP01 75 / SKH01 25 (il book live Deribit). Il costo si sottrae al DRIFT DEL LIBRO e il
|
|
beneficio si misura SUL LIBRO, non sullo sleeve.
|
|
* Copertura 50/50 BTC/ETH (l'esposizione del libro e' 50/50).
|
|
* **Mark-to-market GIORNALIERO** della put (BS a DVOL e spot correnti). E' essenziale: una
|
|
copertura marcata solo a scadenza sarebbe misurata male proprio sul DD, che e' l'unica cosa
|
|
che questo filone valuta.
|
|
* `f` applicato SIMMETRICAMENTE (si compra E si marca a f x modello). Conseguenza strutturale
|
|
che il modello cattura ed e' un risultato in se': **f si paga solo sulla parte di valore che
|
|
converge a INTRINSECO (la scadenza); un roll anticipato non lo paga.** Sensitivita' con f
|
|
asimmetrico (compri caro, vendi a modello) riportata a parte.
|
|
* Spread bid-ask MISURATO dalla catena vera per bucket di tenore, pagato a ogni compra e a ogni
|
|
vendita: senza, il roll a meta' vita vincerebbe per finzione.
|
|
* Fee opzioni Deribit: taker 0,03% del sottostante con cap 12,5% del premio; consegna ITM
|
|
0,015% con lo stesso cap.
|
|
* Finestra: **2021-03-24 -> oggi** (il DVOL non esiste prima). Contiene il 2022. NON contiene
|
|
il crollo di marzo 2020, che e' il sinistro piu' grande della storia del libro: dichiarato.
|
|
|
|
Uso: `nice -n 19 timeout 900 uv run python scripts/research/r0823_tail_hedge.py`
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import math
|
|
import sys
|
|
import time
|
|
import urllib.request
|
|
from functools import lru_cache
|
|
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"))
|
|
sys.path.insert(0, str(ROOT / "scripts" / "research" / "alt"))
|
|
|
|
from src.data.downloader import load_data # noqa: E402
|
|
from src.strategies.trend_portfolio import resample_1d # noqa: E402
|
|
# MOTORE PREZZI RIUSATO — gli stessi di VRP01, non una reimplementazione.
|
|
from src.portfolio.sleeves import ( # noqa: E402
|
|
_bs_put, _strike_from_delta, _tp01_returns, _skyhook_returns, VRP_CFG,
|
|
)
|
|
import altlib as al # noqa: E402
|
|
from scipy.stats import norm as _norm # noqa: E402
|
|
|
|
ASSETS = ("BTC", "ETH")
|
|
W_TP, W_SKH = 0.75, 0.25 # book live Deribit (sleeves.deribit_book_sleeves)
|
|
DPY = 365.25
|
|
|
|
# ---- costi opzioni Deribit (listino pubblico) -------------------------------------------------
|
|
OPT_TAKER_UNDERLYING = 0.0003 # 0,03% del sottostante per gamba
|
|
OPT_FEE_CAP_FRAC = 0.125 # cap 12,5% del premio dell'opzione
|
|
OPT_DELIVERY_UNDERLYING = 0.00015 # 0,015% consegna ITM
|
|
|
|
# ---- GRIGLIA DICHIARATA PRIMA DI GUARDARE -----------------------------------------------------
|
|
G_DELTA = (-0.05, -0.10, -0.15)
|
|
G_TENOR = (7, 14, 30)
|
|
G_BUDGET = (0.005, 0.010, 0.020) # quota di equity spesa in premio, ANNUA
|
|
G_ROLL = ("expiry", "half")
|
|
N_CELLS = len(G_DELTA) * len(G_TENOR) * len(G_BUDGET) * len(G_ROLL) # = 54
|
|
|
|
# ---- lenti di f: 1.0 (modello nudo) + i due misurati il 30/07 + quello misurato OGGI -----------
|
|
F_LENSES_STATIC = (1.0, 2.23, 5.85)
|
|
|
|
DERIBIT = "https://www.deribit.com/api/v2/public/"
|
|
|
|
|
|
# ==================================================================== 0. UTIL METRICHE
|
|
|
|
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:
|
|
"""Drift GEOMETRICO annualizzato (CAGR). E' la grandezza su cui vive la rendita perpetua,
|
|
ed e' l'unica delle due che CAMBIA quando si de-leva (lo Sharpe no: identita' aritmetica)."""
|
|
r = np.asarray(pd.Series(x).dropna().values, float)
|
|
if len(r) < 2:
|
|
return 0.0
|
|
tot = float(np.prod(1.0 + r))
|
|
if tot <= 0:
|
|
return -1.0
|
|
return float(tot ** (DPY / len(r)) - 1.0)
|
|
|
|
|
|
def _dd_of_k(r: np.ndarray, k: float) -> float:
|
|
return _dd(k * r)
|
|
|
|
|
|
def k_for_same_drift(base: np.ndarray, target_drift: float) -> float:
|
|
"""Il de-levering a PARI DRIFT. Serve sull'asse funded: se la copertura NON riduce il maxDD
|
|
(che e' cio' che questo filone misura), il confronto a pari DD e' degenere -> la domanda
|
|
onesta diventa 'a parita' di rendimento atteso, chi sfora meno la regola giornaliera?'.
|
|
drift(k*r) e' crescente in k su un libro a drift positivo."""
|
|
lo, hi = 1e-4, 1.0
|
|
if _drift(hi * base) <= target_drift:
|
|
return 1.0
|
|
for _ in range(60):
|
|
mid = 0.5 * (lo + hi)
|
|
if _drift(mid * base) > target_drift:
|
|
hi = mid
|
|
else:
|
|
lo = mid
|
|
return 0.5 * (lo + hi)
|
|
|
|
|
|
def k_for_same_dd(base: np.ndarray, target_dd: float) -> float:
|
|
"""Il `k` del NULL DEL DE-LEVERING: la scala costante che porta il maxDD del libro NUDO
|
|
esattamente al maxDD della variante coperta. maxDD(k*r) e' monotono crescente in k
|
|
(compounding), quindi bisezione."""
|
|
lo, hi = 1e-4, 1.0
|
|
if _dd_of_k(base, hi) <= target_dd:
|
|
return 1.0 # la copertura non riduce il DD: k=1
|
|
for _ in range(60):
|
|
mid = 0.5 * (lo + hi)
|
|
if _dd_of_k(base, mid) > target_dd:
|
|
hi = mid
|
|
else:
|
|
lo = mid
|
|
return 0.5 * (lo + hi)
|
|
|
|
|
|
# ==================================================================== 1. DATI
|
|
|
|
@lru_cache(maxsize=1)
|
|
def daily_panel() -> pd.DataFrame:
|
|
"""Pannello giornaliero allineato: rendimenti del libro + prezzi/low/DVOL per asset.
|
|
Un solo indice, cosi' nessuna domanda successiva puo' disallinearsi."""
|
|
tp = _tp01_returns()
|
|
skh = _skyhook_returns()
|
|
J = pd.concat({"tp": tp, "skh": skh}, axis=1, join="inner").fillna(0.0)
|
|
book = W_TP * J["tp"] + W_SKH * J["skh"]
|
|
|
|
cols = {"book": book, "tp": J["tp"], "skh": J["skh"]}
|
|
for a in ASSETS:
|
|
d = resample_1d(load_data(a, "1h"))
|
|
idx = pd.DatetimeIndex(pd.to_datetime(d["datetime"], utc=True))
|
|
for c in ("close", "low", "high"):
|
|
assert c in d.columns, f"resample_1d senza colonna {c}"
|
|
cols[f"px_{a}"] = pd.Series(d["close"].values.astype(float), index=idx)
|
|
cols[f"lo_{a}"] = pd.Series(d["low"].values.astype(float), index=idx)
|
|
dv = pd.read_parquet(ROOT / "data" / "raw" / f"dvol_{a.lower()}.parquet")
|
|
cols[f"iv_{a}"] = pd.Series(dv["close"].values.astype(float) / 100.0,
|
|
index=pd.to_datetime(dv["timestamp"], unit="ms", utc=True))
|
|
P = pd.concat(cols, axis=1, join="inner").sort_index().dropna()
|
|
return P
|
|
|
|
|
|
# ==================================================================== 2. IL MOTORE DELLA COPERTURA
|
|
|
|
def _entry_fee(prem_paid: float, S: float) -> float:
|
|
return min(OPT_TAKER_UNDERLYING * S, OPT_FEE_CAP_FRAC * prem_paid) if prem_paid > 0 else 0.0
|
|
|
|
|
|
def hedge_leg(px: np.ndarray, lo: np.ndarray, iv: np.ndarray, *,
|
|
delta: float, tenor: int, budget_ann: float, roll: str,
|
|
f: float, spread: float, f_sell: float | None = None,
|
|
free: bool = False) -> tuple[np.ndarray, np.ndarray, dict]:
|
|
"""Serie GIORNALIERE (ret, wick_extra) del contributo di UNA gamba di copertura, per 1 unita'
|
|
di equity allocata a quella gamba.
|
|
|
|
ret[i] = variazione giornaliera del valore MTM della posizione (frazione di equity)
|
|
wick_extra[i] = guadagno AGGIUNTIVO della copertura al minimo intra-giornaliero, ottenuto
|
|
RIPREZZANDO la put al LOW del giorno (cattura la convessita', non solo il delta)
|
|
|
|
CAUSALE per costruzione: strike, premio e quantita' a `t` usano solo px[t]/iv[t]; il payoff
|
|
usa prezzi successivi e non retroagisce mai sulla decisione.
|
|
|
|
`free=True` -> CONTROLLO POSITIVO: il payoff c'e' ma il premio non viene addebitato.
|
|
"""
|
|
n = len(px)
|
|
T_full = tenor / DPY
|
|
hold = tenor if roll == "expiry" else max(1, tenor // 2)
|
|
n_rolls_year = DPY / hold
|
|
b_roll = budget_ann / n_rolls_year # spesa per roll, in frazione di equity
|
|
|
|
ret = np.zeros(n)
|
|
wick = np.zeros(n)
|
|
diag = dict(n_rolls=0, n_itm=0, prem_spent=0.0, payoff_recv=0.0, fees=0.0, spread_cost=0.0)
|
|
|
|
i = 60 # warm-up (coerente con VRP01)
|
|
while i + hold < n:
|
|
S0, sig = float(px[i]), float(iv[i])
|
|
if not (S0 > 0 and sig > 0):
|
|
i += hold
|
|
continue
|
|
K = _strike_from_delta(S0, T_full, sig, delta)
|
|
p_model = _bs_put(S0, K, T_full, sig)
|
|
if p_model <= 0:
|
|
i += hold
|
|
continue
|
|
p_ask = f * p_model * (1.0 + spread) # si COMPRA al lato caro
|
|
fee0 = _entry_fee(p_ask, S0)
|
|
cost_unit = p_ask + fee0
|
|
q = b_roll / cost_unit # unita' di sottostante coperte
|
|
# giorno d'ingresso: il valore MTM e' q*f*p_model (il mid), quindi si perde subito
|
|
# fee + mezzo spread. Se `free`, il premio non si paga affatto.
|
|
v_prev = q * f * p_model
|
|
# ⚠️ `free` = CONTROLLO POSITIVO (payoff regalato). La prima stesura metteva 0.0 qui e
|
|
# lasciava v_prev>0: cosi' il MTM successivo addebitava il decadimento di un valore mai
|
|
# accreditato -> il "free lunch" costava quanto il premio e il controllo FALLIVA. Il
|
|
# regalo va ACCREDITATO: l'equity sale di v_prev il giorno in cui ricevi l'asset.
|
|
ret[i] += (v_prev if free else (v_prev - b_roll))
|
|
diag["n_rolls"] += 1
|
|
diag["prem_spent"] += 0.0 if free else b_roll
|
|
diag["fees"] += q * fee0
|
|
diag["spread_cost"] += q * f * p_model * spread
|
|
|
|
for s in range(1, hold + 1):
|
|
j = i + s
|
|
rem = (tenor - s) / DPY
|
|
Sj, ivj = float(px[j]), float(iv[j])
|
|
if s == tenor: # SCADENZA: valore = intrinseco
|
|
intr = max(0.0, K - Sj)
|
|
v = q * intr
|
|
if intr > 0:
|
|
dfee = min(OPT_DELIVERY_UNDERLYING * Sj, OPT_FEE_CAP_FRAC * intr) * q
|
|
v -= dfee
|
|
diag["fees"] += dfee
|
|
diag["n_itm"] += 1
|
|
diag["payoff_recv"] += v
|
|
vw = v # a scadenza niente wick residuo
|
|
else:
|
|
fs = f if f_sell is None else f_sell
|
|
v_mid = q * f * _bs_put(Sj, K, rem, ivj)
|
|
v = v_mid
|
|
vw = q * f * _bs_put(float(lo[j]), K, rem, ivj)
|
|
if s == hold: # VENDITA anticipata: mezzo spread
|
|
v = q * fs * _bs_put(Sj, K, rem, ivj) * (1.0 - spread)
|
|
sfee = min(OPT_TAKER_UNDERLYING * Sj, OPT_FEE_CAP_FRAC * max(v, 0.0))
|
|
v -= sfee
|
|
diag["fees"] += sfee
|
|
diag["spread_cost"] += v_mid * spread
|
|
diag["payoff_recv"] += v
|
|
wick[j] += vw - v_mid # EXTRA al minimo, contro il MID di chiusura
|
|
ret[j] += v - v_prev
|
|
v_prev = v
|
|
i += hold
|
|
|
|
return ret, np.maximum(wick, 0.0), diag
|
|
|
|
|
|
@lru_cache(maxsize=None)
|
|
def hedge_series(delta: float, tenor: int, budget: float, roll: str, f: float,
|
|
spread: float, f_sell: float | None = None,
|
|
free: bool = False) -> tuple:
|
|
"""Copertura del LIBRO: 50/50 BTC+ETH (il libro e' 50/50). Ritorna (ret, wick) giornalieri
|
|
come frazione dell'equity del libro."""
|
|
P = daily_panel()
|
|
ret = np.zeros(len(P))
|
|
wick = np.zeros(len(P))
|
|
diags = {}
|
|
for a in ASSETS:
|
|
r, w, d = hedge_leg(P[f"px_{a}"].values, P[f"lo_{a}"].values, P[f"iv_{a}"].values,
|
|
delta=delta, tenor=tenor, budget_ann=budget * 0.5, roll=roll,
|
|
f=f, spread=spread, f_sell=f_sell, free=free)
|
|
ret += r
|
|
wick += w
|
|
diags[a] = d
|
|
return ret, wick, diags
|
|
|
|
|
|
# ==================================================================== 3. f e SPREAD DAL VENUE
|
|
|
|
def _get(path: str, timeout: int = 25):
|
|
with urllib.request.urlopen(DERIBIT + path, timeout=timeout) as r:
|
|
return json.load(r)["result"]
|
|
|
|
|
|
def measure_f_from_venue() -> dict:
|
|
"""f = quota MID di mercato / prezzo BS a DVOL piatto, MISURATO OGGI sulle ali che questa
|
|
strategia comprerebbe. Non e' un'assunzione ereditata dal 30/07: e' una misura su questa
|
|
catena, su questi strike, con questo motore.
|
|
|
|
Si misura anche il MEZZO SPREAD relativo per bucket di tenore, che serve al roll anticipato.
|
|
Solo strumenti con bid E ask veri (un ask solitario al tick non e' un prezzo).
|
|
"""
|
|
out = dict(ok=False, note=None, per_delta={}, spread={}, n=0)
|
|
try:
|
|
idx = {a: float(_get(f"get_index_price?index_name={a.lower()}_usdc")["index_price"])
|
|
for a in ASSETS}
|
|
dvol = {a: float(_get(f"get_volatility_index_data?currency={a}&start_timestamp="
|
|
f"{int((time.time()-86400)*1000)}&end_timestamp={int(time.time()*1000)}"
|
|
f"&resolution=3600")["data"][-1][4]) / 100.0
|
|
for a in ASSETS}
|
|
time.sleep(0.6)
|
|
bs = _get("get_book_summary_by_currency?currency=USDC&kind=option")
|
|
except Exception as e: # rete assente: si dichiara, non si finge
|
|
out["note"] = f"venue non raggiungibile ({type(e).__name__}: {e})"
|
|
return out
|
|
|
|
now = pd.Timestamp.now(tz="UTC")
|
|
rows = []
|
|
for r in bs:
|
|
nm = r["instrument_name"]
|
|
if not nm.endswith("-P"):
|
|
continue
|
|
base = nm.split("_USDC-")[0] if "_USDC-" in nm else None
|
|
if base not in ASSETS:
|
|
continue
|
|
try:
|
|
_, exp, ks, _ = nm.split("-")
|
|
K = float(ks.replace("d", "."))
|
|
texp = pd.to_datetime(exp, format="%d%b%y", utc=True) + pd.Timedelta(hours=8)
|
|
except Exception:
|
|
continue
|
|
bid, ask = r.get("bid_price"), r.get("ask_price")
|
|
if not bid or not ask or bid <= 0 or ask <= bid:
|
|
continue # niente bid vero -> non e' un prezzo
|
|
T = (texp - now).total_seconds() / (365.25 * 86400)
|
|
if T <= 1 / DPY or T > 45 / DPY:
|
|
continue
|
|
S, sig = idx[base], dvol[base]
|
|
mid = 0.5 * (bid + ask)
|
|
half_spread = 0.5 * (ask - bid) / mid
|
|
model = _bs_put(S, K, T, sig)
|
|
if model <= 1e-9:
|
|
continue
|
|
# delta BS della put allo stesso vol piatto (per assegnare l'osservazione al suo bucket)
|
|
d1 = (math.log(S / K) + 0.5 * sig ** 2 * T) / (sig * math.sqrt(T))
|
|
dlt = -float(_norm.cdf(-d1))
|
|
rows.append(dict(asset=base, K=K, T=T, days=T * DPY, delta=dlt,
|
|
mid=mid, model=model, f=mid / model, hs=half_spread))
|
|
if not rows:
|
|
out["note"] = "nessuno strumento con bid E ask veri nella finestra <=45g"
|
|
return out
|
|
D = pd.DataFrame(rows)
|
|
out["n"] = int(len(D))
|
|
for dtgt in G_DELTA:
|
|
sel = D[(D.delta >= dtgt - 0.035) & (D.delta <= dtgt + 0.035)]
|
|
if len(sel) >= 3:
|
|
out["per_delta"][dtgt] = dict(n=int(len(sel)), f_med=float(sel.f.median()),
|
|
f_lo=float(sel.f.quantile(0.10)),
|
|
f_hi=float(sel.f.quantile(0.90)),
|
|
hs_med=float(sel.hs.median()))
|
|
for lab, (a, b) in dict(t7=(1, 10), t14=(10, 21), t30=(21, 45)).items():
|
|
sel = D[(D.days >= a) & (D.days < b)]
|
|
if len(sel) >= 3:
|
|
out["spread"][lab] = dict(n=int(len(sel)), hs_med=float(sel.hs.median()),
|
|
hs_p90=float(sel.hs.quantile(0.90)))
|
|
out["ok"] = bool(out["per_delta"] or out["spread"])
|
|
return out
|
|
|
|
|
|
def venue_min_ticket() -> dict:
|
|
"""Eseguibilita' MISURATA dal venue: min_trade_amount + tick, e il costo in DOLLARI del
|
|
lotto minimo per una put deep-OTM vera. ⚠️ Comprare un'opzione costa il PREMIO, non il
|
|
NOZIONALE: il muro dei 'lotti da migliaia di dollari' del 22/08 e' il collaterale per
|
|
VENDERE, non il costo per COMPRARE. Qui si misura la cosa giusta."""
|
|
out = dict(ok=False, note=None, fam={})
|
|
try:
|
|
for cur, pref in (("USDC", "{a}_USDC-"), ("BTC", "BTC-"), ("ETH", "ETH-")):
|
|
inst = _get(f"get_instruments?currency={cur}&kind=option&expired=false")
|
|
time.sleep(0.6)
|
|
for a in ASSETS:
|
|
p = pref.format(a=a)
|
|
sel = [i for i in inst if i["instrument_name"].startswith(p)]
|
|
if not sel:
|
|
continue
|
|
fam = "USDC-lineare" if cur == "USDC" else "inverse"
|
|
out["fam"][(a, fam)] = dict(min_amt=float(sel[0]["min_trade_amount"]),
|
|
tick=float(sel[0]["tick_size"]),
|
|
settle=sel[0]["settlement_currency"],
|
|
n=len(sel))
|
|
out["ok"] = bool(out["fam"])
|
|
except Exception as e:
|
|
out["note"] = f"venue non raggiungibile ({type(e).__name__}: {e})"
|
|
return out
|
|
|
|
|
|
# ==================================================================== 4. LENTE WICK ACCOPPIATA
|
|
|
|
@lru_cache(maxsize=1)
|
|
def coupled_book_tuples() -> pd.DataFrame:
|
|
"""(R, m) giornaliere del libro 75/25 sotto la LENTE WICK ACCOPPIATA — RIUSATA da
|
|
r0725_prop_coupled (importata, non riscritta). Se il recon non e' disponibile si dichiara e
|
|
si ricade su close-only, DICENDO che close-only e' CIECA su una regola a UN giorno
|
|
(misurato 22/08: 0,00 breach/anno contro 0,40-1,21 veri, rapporto INF)."""
|
|
try:
|
|
import r0725_prop_coupled as pc
|
|
return pc.crypto_daily_tuples(W_TP, W_SKH)
|
|
except Exception as e: # dichiarato, mai silenzioso
|
|
print(f" ⚠️ lente accoppiata NON disponibile ({type(e).__name__}: {e}) -> close-only")
|
|
return pd.DataFrame()
|
|
|
|
|
|
# ==================================================================== 5. VALUTAZIONE DI UNA CELLA
|
|
|
|
def evaluate(book: np.ndarray, hret: np.ndarray) -> dict:
|
|
hedged = book + hret
|
|
dd_h = _dd(hedged)
|
|
k = k_for_same_dd(book, dd_h)
|
|
base_k = k * book
|
|
dd_b = _dd(book)
|
|
dd_reduced = bool(dd_h < dd_b - 1e-12)
|
|
dr_h, dr_k = _drift(hedged), _drift(base_k)
|
|
return dict(
|
|
sh_base=_sh(book), sh_hedged=_sh(hedged), sh_delev=_sh(base_k),
|
|
dd_base=dd_b, dd_hedged=dd_h, dd_delev=_dd(base_k),
|
|
dr_base=_drift(book), dr_hedged=dr_h, dr_delev=dr_k,
|
|
k=k, hedged=hedged, dd_reduced=dd_reduced,
|
|
# VINCE solo se (i) il DD scende davvero E (ii) a quel DD il drift batte il de-levering.
|
|
# Senza (i) il null e' DEGENERE (k=1, dr_delev==dr_base) e un Δdrift>0 direbbe solo
|
|
# "la put ha guadagnato in questa finestra", che e' una domanda di ALPHA — non questa.
|
|
wins=bool(dd_reduced and dr_h > dr_k),
|
|
)
|
|
|
|
|
|
def worst_months(r: pd.Series, n: int = 5) -> list:
|
|
m = (1.0 + r).resample("ME").prod() - 1.0
|
|
return [(str(i.date())[:7], float(v)) for i, v in m.nsmallest(n).items()]
|
|
|
|
|
|
# ==================================================================== MAIN
|
|
|
|
def main() -> None:
|
|
t0 = time.time()
|
|
print("=" * 100)
|
|
print("§46 TAIL-HEDGE — put deep-OTM come ASSICURAZIONE STATICA sul libro TP01 75 / SKH01 25")
|
|
print("=" * 100)
|
|
print(f"GRIGLIA DICHIARATA PRIMA DI GUARDARE: {len(G_DELTA)} delta x {len(G_TENOR)} tenori x "
|
|
f"{len(G_BUDGET)} budget x {len(G_ROLL)} roll = {N_CELLS} celle per lente di f.")
|
|
print("ATTESA A PRIORI: REFUTATO al f realistico. Ragione piu' forte (c): il peggior giorno del")
|
|
print(" libro NON e' un crash ma uno short-squeeze di SKH01 -> una PUT non protegge da un rialzo.")
|
|
|
|
P = daily_panel()
|
|
book = P["book"].values
|
|
idx = P.index
|
|
bs_book = pd.Series(book, index=idx)
|
|
print(f"\nFinestra (comune libro+DVOL): {idx[0].date()} -> {idx[-1].date()} "
|
|
f"({len(P)} giorni, {len(P)/DPY:.1f} anni)")
|
|
print(f"LIBRO NUDO su questa finestra: Sharpe {_sh(book):.3f} | maxDD {_dd(book)*100:.2f}% | "
|
|
f"drift {_drift(book)*100:.2f}%/a")
|
|
print("⚠️ La finestra NON contiene marzo 2020 (il DVOL parte 2021-03): il sinistro piu' grande")
|
|
print(" della storia del libro e' FUORI CAMPIONE. La copertura non e' testata li'.")
|
|
|
|
# ---------------------------------------------------------------- §A. f e spread dal venue
|
|
print("\n" + "=" * 100)
|
|
print("§A. f E SPREAD — MISURATI OGGI SULLE QUOTE VERE (non assunti dal 30/07)")
|
|
print("=" * 100)
|
|
fv = measure_f_from_venue()
|
|
if fv["ok"]:
|
|
print(f" {fv['n']} put con bid E ask veri, tenore <=45g, catena USDC-lineare.")
|
|
print(f" {'delta':>7} {'n':>4} {'f mediano':>10} {'f p10':>8} {'f p90':>8} {'mezzo-spread':>13}")
|
|
for d, v in sorted(fv["per_delta"].items(), reverse=True):
|
|
print(f" {d:>7.2f} {v['n']:>4} {v['f_med']:>10.2f} {v['f_lo']:>8.2f} "
|
|
f"{v['f_hi']:>8.2f} {v['hs_med']*100:>12.1f}%")
|
|
print(f" mezzo-spread per tenore: " + " ".join(
|
|
f"{k} {v['hs_med']*100:.1f}% (n={v['n']})" for k, v in fv["spread"].items()))
|
|
else:
|
|
print(f" NON MISURATO: {fv['note']}")
|
|
F_MEAS = {d: v["f_med"] for d, v in fv.get("per_delta", {}).items()}
|
|
if F_MEAS:
|
|
print(" -> f USATO nella lente 'misurato oggi': " +
|
|
" ".join(f"δ{d:.2f}={v:.2f}" for d, v in sorted(F_MEAS.items(), reverse=True)))
|
|
SPREAD = {}
|
|
for lab, ten in (("t7", 7), ("t14", 14), ("t30", 30)):
|
|
SPREAD[ten] = fv.get("spread", {}).get(lab, {}).get("hs_med", 0.10)
|
|
print(f" -> mezzo-spread usato per tenore: " +
|
|
" ".join(f"{t}g {SPREAD[t]*100:.1f}%" for t in G_TENOR) +
|
|
("" if fv["ok"] else " (DEFAULT 10%: venue non letto)"))
|
|
print(" ⚠️ DISTORSIONE DICHIARATA, e va nel verso che FAVORISCE la copertura: lo spread e' il")
|
|
print(" mediano dei soli strumenti con bid E ask veri. Le ali piu' profonde e piu' corte")
|
|
print(" spesso NON hanno bid (solo un ask al tick) e cadono fuori dal campione: quelle")
|
|
print(" sarebbero le piu' care, e per un tail-hedge sono proprio quelle che si comprano.")
|
|
print(" Quindi il costo qui e' un PAVIMENTO, e il verdetto REFUTATO ne esce rafforzato.")
|
|
|
|
# ---------------------------------------------------------------- §B. il null del de-levering
|
|
print("\n" + "=" * 100)
|
|
print("§B. NULL DEL DE-LEVERING — IL GATE CHE DECIDE (in cima, come si deve)")
|
|
print("=" * 100)
|
|
print(" Per ogni cella: k = scala che porta il libro NUDO allo STESSO maxDD della coperta.")
|
|
print(" Sharpe(k*base) == Sharpe(base) per identita' aritmetica -> la gamba informativa e' il")
|
|
print(" DRIFT. La copertura VINCE solo se drift(coperta) > drift(k*base) [e/o Sharpe sale].")
|
|
|
|
lenses = [("f=1.00 (modello nudo)", 1.0)]
|
|
for f in F_LENSES_STATIC[1:]:
|
|
lenses.append((f"f={f:.2f} (misurato 30/07)", f))
|
|
if F_MEAS:
|
|
lenses.append(("f MISURATO oggi per-delta", None))
|
|
|
|
all_rows = []
|
|
for lab, fval in lenses:
|
|
print(f"\n ---- lente {lab} " + "-" * (78 - len(lab)))
|
|
print(f" {'delta':>6} {'ten':>4} {'budget':>7} {'roll':>7} | {'ShCop':>7} {'ShBase':>7} | "
|
|
f"{'DDcop':>7} {'DDbase':>7} | {'k':>6} {'DrCop':>8} {'DrDelev':>8} {'Δdrift':>8} | esito")
|
|
best = None
|
|
for dlt in G_DELTA:
|
|
f_use = fval if fval is not None else F_MEAS.get(dlt, F_LENSES_STATIC[1])
|
|
for ten in G_TENOR:
|
|
for bud in G_BUDGET:
|
|
for roll in G_ROLL:
|
|
hr, hw, dg = hedge_series(dlt, ten, bud, roll, f_use, SPREAD[ten])
|
|
ev = evaluate(book, hr)
|
|
gap = ev["dr_hedged"] - ev["dr_delev"]
|
|
win = ev["wins"]
|
|
row = dict(lens=lab, f=f_use, delta=dlt, tenor=ten, budget=bud, roll=roll,
|
|
**{k: v for k, v in ev.items() if k != "hedged"}, gap=gap,
|
|
hret=hr, hwick=hw, diag=dg)
|
|
all_rows.append(row)
|
|
if best is None or gap > best["gap"]:
|
|
best = row
|
|
if bud == 0.010 and roll == "expiry": # riga stampata: budget centrale
|
|
print(f" {dlt:>6.2f} {ten:>4} {bud*100:>6.1f}% {roll:>7} | "
|
|
f"{ev['sh_hedged']:>7.3f} {ev['sh_base']:>7.3f} | "
|
|
f"{ev['dd_hedged']*100:>6.2f}% {ev['dd_base']*100:>6.2f}% | "
|
|
f"{ev['k']:>6.3f} {ev['dr_hedged']*100:>7.2f}% "
|
|
f"{ev['dr_delev']*100:>7.2f}% {gap*100:>+7.2f}% | "
|
|
f"{'COPERTURA' if win else ('de-levering' if ev['dd_reduced'] else 'DD PEGGIORE')}")
|
|
sub = [r for r in all_rows if r["lens"] == lab]
|
|
nwin = sum(1 for r in sub if r["wins"])
|
|
nred = sum(1 for r in sub if r["dd_reduced"])
|
|
cand = [r for r in sub if r["dd_reduced"]]
|
|
best = max(cand or sub, key=lambda r: (r["gap"] if cand else -r["dd_hedged"]))
|
|
print(f" >> {lab}: il maxDD SCENDE in {nred}/{len(sub)} celle; fra quelle la copertura "
|
|
f"batte il de-levering in {nwin}/{len(sub)}.")
|
|
head = ("migliore fra le celle che RIDUCONO il DD" if cand else
|
|
"NESSUNA cella riduce il DD -> si riporta quella col DD MINORE (k=1, confronto "
|
|
"degenere)")
|
|
print(f" {head}: delta {best['delta']:.2f} / "
|
|
f"{best['tenor']}g / {best['budget']*100:.1f}% / {best['roll']} -> "
|
|
f"maxDD {best['dd_base']*100:.2f}% -> {best['dd_hedged']*100:.2f}% (k {best['k']:.3f}), "
|
|
f"Δdrift vs de-levering {best['gap']*100:+.2f} pp/a, "
|
|
f"ΔSharpe {best['sh_hedged']-best['sh_base']:+.3f}")
|
|
|
|
# ---------------------------------------------------------------- §C. controlli
|
|
print("\n" + "=" * 100)
|
|
print("§C. CONTROLLI — un rilevatore che non segnala nulla puo' essere semplicemente rotto")
|
|
print("=" * 100)
|
|
ctrl = []
|
|
# (+1) copertura GRATIS (payoff senza premio): DEVE ridurre il DD e battere il de-levering
|
|
hr, _, _ = hedge_series(-0.10, 30, 0.020, "expiry", 1.0, 0.0, free=True)
|
|
ev = evaluate(book, hr); ctrl.append(("(+) payoff SENZA premio (free lunch)", ev, True))
|
|
# (+2) premio 1/10: stesso budget -> 10x le put. Speculare al negativo, DEVE funzionare
|
|
hr, _, _ = hedge_series(-0.10, 30, 0.020, "expiry", 0.1, 0.0)
|
|
ev = evaluate(book, hr); ctrl.append(("(+) premio 1/10 (f=0.1, 10x le put)", ev, True))
|
|
# (-) premio 10x: DEVE essere refutata
|
|
hr, _, _ = hedge_series(-0.10, 30, 0.020, "expiry", 10.0, 0.0)
|
|
ev = evaluate(book, hr); ctrl.append(("(-) premio 10x (f=10)", ev, False))
|
|
# (=) put ATM f=1, spread 0: NON e' un controllo (non deve funzionare per forza) ma e' il
|
|
# riferimento piu' caro possibile della famiglia: si stampa come misura, non come test.
|
|
hr, _, _ = hedge_series(-0.50, 30, 0.020, "expiry", 1.0, 0.0)
|
|
ev = evaluate(book, hr); ctrl.append(("(=) put ATM(δ-0.50) f=1 — misura, non controllo", ev, None))
|
|
for nm, ev, must in ctrl:
|
|
gap = ev["dr_hedged"] - ev["dr_delev"]
|
|
got = "RICONOSCIUTA" if ev["wins"] else "refutata"
|
|
tag = "" if must is None else (" ✓" if ev["wins"] == must else " ✗ CONTROLLO FALLITO")
|
|
print(f" {nm:<48} DD {ev['dd_base']*100:5.2f}%->{ev['dd_hedged']*100:5.2f}% k {ev['k']:.3f} "
|
|
f"Δdrift {gap*100:+7.2f}pp ShΔ {ev['sh_hedged']-ev['sh_base']:+.3f} -> {got}{tag}")
|
|
ok = all(ev["wins"] == must for _, ev, must in ctrl if must is not None)
|
|
print(" APPARATO: " + ("controlli 3/3 OK — il rilevatore riconosce cio' che DEVE riconoscere "
|
|
"e rifiuta cio' che DEVE rifiutare"
|
|
if ok else
|
|
"\u26a0\ufe0f UN CONTROLLO E' FALLITO: i numeri sopra NON sono credibili"))
|
|
|
|
# ---------------------------------------------------------------- §D. causalita'
|
|
print("\n" + "=" * 100)
|
|
print("§D. CAUSALITA' — prova esplicita, non un gate generico")
|
|
print("=" * 100)
|
|
print(" `altlib.causality_ok` vuole un target_fn(df)->posizioni: NON si applica a un overlay")
|
|
print(" di opzioni (non e' una posizione continua sul sottostante). Al suo posto la prova")
|
|
print(" diretta e piu' forte: si ricalcola la copertura su un PREFISSO del pannello e si")
|
|
print(" verifica che coincida con la coda della serie completa. Un leak divergerebbe.")
|
|
hr_full, _, _ = hedge_series(-0.10, 14, 0.010, "expiry", 1.0, SPREAD[14])
|
|
cut = int(len(P) * 0.85)
|
|
Pf = daily_panel()
|
|
sub = Pf.iloc[:cut]
|
|
hr_cut, _, _ = hedge_leg(sub["px_BTC"].values, sub["lo_BTC"].values, sub["iv_BTC"].values,
|
|
delta=-0.10, tenor=14, budget_ann=0.010 * 0.5, roll="expiry",
|
|
f=1.0, spread=SPREAD[14])
|
|
hr_full_btc, _, _ = hedge_leg(Pf["px_BTC"].values, Pf["lo_BTC"].values, Pf["iv_BTC"].values,
|
|
delta=-0.10, tenor=14, budget_ann=0.010 * 0.5, roll="expiry",
|
|
f=1.0, spread=SPREAD[14])
|
|
# ⚠️ ERRORE MIO, catturato dalla prima corsa (dmax 1.28e-04 = "DIVERGE"): confrontavo fino a
|
|
# `cut`, ma il prefisso si ferma un roll prima (`while i + hold < n`), quindi l'ultimo ciclo
|
|
# non e' CALCOLATO nel troncato mentre lo e' nel completo. La differenza era per ASSENZA di
|
|
# un roll, non per look-ahead. Il confronto onesto finisce prima dell'ultimo roll possibile.
|
|
hold = 14
|
|
stop = cut - 2 * hold
|
|
tail = 200
|
|
dmax = float(np.max(np.abs(hr_cut[stop - tail:stop] - hr_full_btc[stop - tail:stop])))
|
|
dmax_naive = float(np.max(np.abs(hr_cut[cut - tail:cut] - hr_full_btc[cut - tail:cut])))
|
|
print(f" max|differenza| su [{stop-tail}:{stop}] (cut a {cut}/{len(P)}, ultimo roll escluso): "
|
|
f"{dmax:.2e} -> {'CAUSALE' if dmax < 1e-12 else '⚠️ DIVERGE'}")
|
|
print(f" (fino a `cut` fa {dmax_naive:.2e}: NON e' un leak, e' il roll che il prefisso non ha")
|
|
print(f" ancora eseguito — la finestra sbagliata dichiarava un difetto inesistente.)")
|
|
|
|
# ---------------------------------------------------------------- §E. lente di coda
|
|
print("\n" + "=" * 100)
|
|
print("§E. LENTE DI CODA — maxDD, peggiori mesi, peggior giorno, breach 4%/6%")
|
|
print("=" * 100)
|
|
CP = coupled_book_tuples()
|
|
if not CP.empty:
|
|
CP = CP[(CP.index >= idx[0]) & (CP.index <= idx[-1])]
|
|
cov = len(CP.index.intersection(idx)) / len(idx)
|
|
print(f" lente ACCOPPIATA disponibile: {len(CP)} giorni, copertura {cov*100:.1f}% del pannello")
|
|
# scelta della cella per la lente di coda: la MIGLIORE al f misurato (o 2.23), budget 1%
|
|
f_pick = F_MEAS.get(-0.10, F_LENSES_STATIC[1])
|
|
cells_tail = [("NUDO", None)]
|
|
for dlt in G_DELTA:
|
|
cells_tail.append((f"δ{dlt:.2f} 30g 1%/a expiry", (dlt, 30, 0.010, "expiry",
|
|
F_MEAS.get(dlt, f_pick))))
|
|
cells_tail.append(("δ-0.10 30g 2%/a expiry", (-0.10, 30, 0.020, "expiry", f_pick)))
|
|
cells_tail.append(("δ-0.10 30g 2%/a f=1 (irreale)", (-0.10, 30, 0.020, "expiry", 1.0)))
|
|
|
|
def _breach(series_close: np.ndarray, wick_extra: np.ndarray) -> tuple:
|
|
s_ = pd.Series(series_close, index=idx)
|
|
n_yr = len(P) / DPY
|
|
br_c = float((s_ <= -0.04).sum()) / n_yr
|
|
if CP.empty:
|
|
return float("nan"), br_c
|
|
m = CP["m"].reindex(idx).values
|
|
n_ev = int(np.nansum((m + np.nan_to_num(wick_extra)) <= -0.04))
|
|
_breach.last_n = n_ev
|
|
return n_ev / n_yr, br_c
|
|
|
|
print(f"\n {'variante':<32} {'maxDD':>7} {'pegg.giorno':>12} {'breach 4% /a':>14} "
|
|
f"{'(close-only)':>13} {'drift':>8} {'Sharpe':>7}")
|
|
for nm, spec in cells_tail:
|
|
if spec is None:
|
|
hr = np.zeros(len(P)); hw = np.zeros(len(P))
|
|
else:
|
|
dlt, ten, bud, roll, fu = spec
|
|
hr, hw, _ = hedge_series(dlt, ten, bud, roll, fu, SPREAD[ten])
|
|
hed = book + hr
|
|
s = pd.Series(hed, index=idx)
|
|
worst_day = float(s.min())
|
|
br_coup, br_close = _breach(hed, hr + hw) # il minimo del libro + la copertura
|
|
n_ev = getattr(_breach, "last_n", -1)
|
|
print(f" {nm:<32} {_dd(hed)*100:>6.2f}% {worst_day*100:>11.2f}% {br_coup:>8.2f}({n_ev:>2}) "
|
|
f"{br_close:>12.2f} {_drift(hed)*100:>7.2f}% {_sh(hed):>7.3f}")
|
|
if spec is not None:
|
|
# ⚠️ il confronto a pari DD e' DEGENERE qui (il DD non scende, quindi k=1). Sull'asse
|
|
# funded la domanda onesta e' l'altra: a PARI DRIFT, chi sfora meno la regola a un
|
|
# giorno? Cioe' basta rimpicciolire la posizione per comprare gli stessi breach in meno.
|
|
kk = k_for_same_drift(book, _drift(hed))
|
|
brk, brkc = _breach(kk * book, np.zeros(len(P)) if CP.empty
|
|
else (kk - 1.0) * np.nan_to_num(CP['m'].reindex(idx).values))
|
|
lab_ = " ^ de-levering k=%.3f, PARI DRIFT" % kk
|
|
print(f" {lab_:<32} {_dd(kk*book)*100:>6.2f}% "
|
|
f"{float(np.min(kk*book))*100:>11.2f}% {brk:>8.2f}({getattr(_breach,'last_n',-1):>2}) "
|
|
f"{brkc:>12.2f} "
|
|
f"{_drift(kk*book)*100:>7.2f}% {_sh(kk*book):>7.3f}")
|
|
print(" ⚠️ close-only e' CIECA su una regola a UN giorno (22/08: 0,00 contro 0,40-1,21 veri).")
|
|
print(" La colonna che decide e' quella ACCOPPIATA. Fra parentesi il CONTEGGIO di eventi:")
|
|
print(" 3-5 breach in 5,4 anni NON distinguono le varianti fra loro (un evento in piu' o in")
|
|
print(" meno sposta il tasso del 20-30%) -> si legge il SEGNO, non la taglia. Il fatto")
|
|
print(" robusto e' che il de-levering a PARI DRIFT compra lo stesso ordine di grandezza di")
|
|
print(" riduzione dei breach senza toccare il maxDD nel verso sbagliato.")
|
|
print("\n REGOLA MAX-LOSS 6% (statica dal saldo iniziale, HyroTrader — e' la regola BINDING")
|
|
print(" del canale funded, gia' modellata il 25/07 come EV_DD=0.06):")
|
|
k6 = k_for_same_dd(book, 0.06)
|
|
print(f" libro NUDO maxDD {_dd(book)*100:5.2f}% -> SFORA (serve k={k6:.3f} per "
|
|
f"stare a 6%, drift {_drift(k6*book)*100:.2f}%/a)")
|
|
for nm, spec in cells_tail[1:]:
|
|
dlt, ten, bud, roll, fu = spec
|
|
hr2, _, _ = hedge_series(dlt, ten, bud, roll, fu, SPREAD[ten])
|
|
hd = book + hr2
|
|
k6h = k_for_same_dd(hd, 0.06)
|
|
print(f" {nm:<20} maxDD {_dd(hd)*100:5.2f}% -> SFORA (serve k={k6h:.3f}, "
|
|
f"drift {_drift(k6h*hd)*100:.2f}%/a)")
|
|
print(" >> Sull'UNICA regola che decide un conto funded la copertura va nel verso SBAGLIATO:")
|
|
print(" alza il maxDD e quindi ABBASSA la leva ammessa. Il beneficio sui breach a un")
|
|
print(" giorno (0,92 -> 0,55/anno) e' reale ma si compra piu' a buon mercato de-levando,")
|
|
print(" che sulla regola a 6 punti e' l'unica leva che funziona.")
|
|
|
|
print("\n 5 PEGGIORI MESI (nudo vs coperto δ-0.10 30g 1%/a, f misurato):")
|
|
hr, _, _ = hedge_series(-0.10, 30, 0.010, "expiry", F_MEAS.get(-0.10, f_pick), SPREAD[30])
|
|
wm_b = worst_months(bs_book)
|
|
wm_h = worst_months(pd.Series(book + hr, index=idx))
|
|
mh = dict(wm_h)
|
|
hs_m = (1.0 + pd.Series(book + hr, index=idx)).resample("ME").prod() - 1.0
|
|
hs_m.index = [str(i.date())[:7] for i in hs_m.index]
|
|
for mth, v in wm_b:
|
|
hv = float(hs_m.get(mth, float("nan")))
|
|
print(f" {mth} nudo {v*100:>7.2f}% coperto {hv*100:>7.2f}% "
|
|
f"delta {(hv - v)*100:>+6.2f}pp")
|
|
|
|
# ---------------------------------------------------------------- §F. il rischio sbagliato
|
|
print("\n" + "=" * 100)
|
|
print("§F. L'ATTESA (c) MISURATA — la put protegge dal rischio che il libro ha davvero?")
|
|
print("=" * 100)
|
|
und = 0.5 * (P["px_BTC"].pct_change() + P["px_ETH"].pct_change()).fillna(0.0).values
|
|
ordr = np.argsort(book)[:20]
|
|
n_crash = int((und[ordr] <= -0.05).sum())
|
|
print(f" Sui 20 PEGGIORI giorni del libro: {n_crash}/20 sono giorni di crollo del sottostante")
|
|
print(f" (BTC/ETH medio <= -5%). Negli altri {20-n_crash} il libro perde mentre il mercato NON")
|
|
print(f" crolla -> una put non puo' proteggere per costruzione.")
|
|
print(f" {'giorno':>12} {'libro':>8} {'BTC/ETH':>9} {'TP01':>8} {'SKH01':>8}")
|
|
for j in ordr[:8]:
|
|
print(f" {str(idx[j].date()):>12} {book[j]*100:>7.2f}% {und[j]*100:>8.2f}% "
|
|
f"{P['tp'].values[j]*100:>7.2f}% {P['skh'].values[j]*100:>7.2f}%")
|
|
beta = float(np.polyfit(und, book, 1)[0])
|
|
corr = float(np.corrcoef(book, und)[0, 1])
|
|
print(f" corr(libro, sottostante) = {corr:+.3f} | beta del libro al sottostante = {beta:+.3f}")
|
|
print()
|
|
print(" --- VERDETTO SULL'ATTESA A PRIORI (si riporta anche dove e' REFUTATA) ---")
|
|
if n_crash >= 12:
|
|
print(f" (c) REFUTATA nella forma in cui l'ho scritta: {n_crash}/20 dei peggiori giorni del")
|
|
print( " libro SONO giorni di crollo. Non e' vero che si compra assicurazione contro il")
|
|
print( " rischio sbagliato in senso di DIREZIONE. ⚠️ E il caso che avevo in mente (lo")
|
|
print( " short-squeeze di SKH01) sta FUORI FINESTRA: il DVOL parte 2021-03.")
|
|
else:
|
|
print(f" (c) CONFERMATA: solo {n_crash}/20 dei peggiori giorni sono crolli.")
|
|
print(f" (b) CONFERMATA, ed e' il meccanismo operativo: beta del libro al sottostante {beta:+.3f}.")
|
|
print( " Il 2021-05-19 il sottostante ha fatto -21,1% e il libro -2,0%: la strategia aveva")
|
|
print( " GIA' ridotto la perdita di ~10x. La put paga in proporzione al crollo del")
|
|
print( " SOTTOSTANTE, ma protegge una perdita gia' ridotta di un ordine di grandezza ->")
|
|
print( " per coprire un DD del libro serve un nozionale di put ~1/beta volte l'equity, e")
|
|
print( " quel nozionale costa ~1/beta volte il budget.")
|
|
print( " (a) CONFERMATA: vedi il decadimento del premio nel §B (il costo cresce col budget e")
|
|
print( " col f mentre il beneficio satura).")
|
|
|
|
# ---------------------------------------------------------------- §G. eseguibilita'
|
|
print("\n" + "=" * 100)
|
|
print("§G. ESEGUIBILITA' — misurata DAL VENUE. Comprare un'opzione costa il PREMIO, non il")
|
|
print(" NOZIONALE: il muro dei 'lotti da migliaia di dollari' e' il collaterale per VENDERE.")
|
|
print("=" * 100)
|
|
vm = venue_min_ticket()
|
|
if vm["ok"]:
|
|
print(f" {'asset':>6} {'famiglia':>14} {'min_trade_amount':>18} {'tick':>10} {'settle':>8}")
|
|
for (a, fam), v in sorted(vm["fam"].items()):
|
|
print(f" {a:>6} {fam:>14} {v['min_amt']:>18} {v['tick']:>10} {v['settle']:>8}")
|
|
else:
|
|
print(f" NON MISURATO: {vm['note']}")
|
|
# costo del lotto minimo con i premi modellati di fine campione
|
|
print(f"\n Costo del LOTTO MINIMO e capitale a cui la copertura diventa comprabile a ogni roll:")
|
|
print(f" {'delta':>6} {'ten':>4} {'premio/unita':>13} {'lotto min':>10} "
|
|
f"{'$ lotto':>9} {'budget 1%/roll a $635':>22} {'capitale minimo':>16}")
|
|
for dlt in G_DELTA:
|
|
for ten in (7, 30):
|
|
S = float(P["px_BTC"].values[-1]); sig = float(P["iv_BTC"].values[-1])
|
|
T = ten / DPY
|
|
K = _strike_from_delta(S, T, sig, dlt)
|
|
pm = _bs_put(S, K, T, sig)
|
|
fu = F_MEAS.get(dlt, F_LENSES_STATIC[1])
|
|
price = fu * pm
|
|
minamt = vm["fam"].get(("BTC", "USDC-lineare"), {}).get("min_amt", 0.01)
|
|
lot = minamt * price
|
|
per_roll_635 = 0.010 * (ten / DPY) * 0.5 * 635.0 # 1%/a, meta' su BTC
|
|
cap_min = lot / (0.010 * (ten / DPY) * 0.5)
|
|
print(f" {dlt:>6.2f} {ten:>4} {price:>13.2f} {minamt:>10} {lot:>9.2f} "
|
|
f"{per_roll_635:>21.2f} {cap_min:>15,.0f}")
|
|
print(" ⚠️ E c'e' un secondo muro, STRUTTURALE e non di capitale: il TICK. Su BTC_USDC il tick")
|
|
print(" e' 5 USDC -> una put che vale meno di $5 NON e' quotabile sotto $5. Piu' e' deep-OTM")
|
|
print(" (cioe' piu' e' economica la copertura) piu' il tick domina il prezzo.")
|
|
# eval_weights_smallcap sulla gamba LINEARE, come riferimento del pavimento
|
|
try:
|
|
from src.strategies.trend_portfolio import TrendPortfolio, CANONICAL
|
|
d1 = resample_1d(load_data("BTC", "1h"))
|
|
tpm = TrendPortfolio(**CANONICAL)
|
|
sc = al.eval_weights_smallcap(d1, tpm.target_series(d1), capital=635.0, min_order=5.0)
|
|
print(f"\n Riferimento (gamba LINEARE dello stesso libro): TP01/BTC a $635 haircut Sharpe "
|
|
f"{sc['sharpe_haircut']:+.3f} -> il pavimento $5 non morde sul perpetual.")
|
|
print(" La differenza NON e' il livello del costo ma la sua FORMA (regola 25/07): sul")
|
|
print(" perpetual il minimo e' in DOLLARI di nozionale, sull'opzione e' in UNITA' di")
|
|
print(" sottostante MOLTIPLICATE per un premio che tende a zero -> due muri diversi.")
|
|
except Exception as e:
|
|
print(f"\n (riferimento smallcap non calcolato: {type(e).__name__}: {e})")
|
|
|
|
# ---------------------------------------------------------------- §H. gate residui
|
|
print("\n" + "=" * 100)
|
|
print("§H. GATE RESIDUI")
|
|
print("=" * 100)
|
|
# cella migliore in assoluto al f misurato/realistico
|
|
real = [r for r in all_rows if r["lens"] != "f=1.00 (modello nudo)"]
|
|
pool = real if real else all_rows
|
|
bestr = max(pool, key=lambda r: r["gap"])
|
|
hedged_best = pd.Series(book + bestr["hret"], index=idx)
|
|
imp = al.implausible_sharpe(hedged_best)
|
|
print(f" implausible_sharpe(libro coperto, cella migliore al f realistico): "
|
|
f"implausible={imp['implausible']} Sharpe {imp.get('sharpe', float('nan')):.2f} "
|
|
f"maxDD {imp.get('maxdd', float('nan'))*100:.2f}%")
|
|
for r in imp["reasons"]:
|
|
print(f" - {r}")
|
|
sr_all = [r["sh_hedged"] for r in pool]
|
|
dsr, nullmax = al.deflated_sharpe(bestr["sh_hedged"], sr_all, hedged_best)
|
|
print(f" deflated_sharpe della cella MIGLIORE su {len(pool)} trial: DSR {dsr:.3f} "
|
|
f"(max atteso dal nullo {nullmax:.2f}) -> {'PASS' if dsr >= 0.95 else 'FAIL'}")
|
|
print(" ⚠️ Il DSR qui e' informativo ma NON e' il gate che decide: la domanda non e' 'questa")
|
|
print(" cella ha alpha' (non ne ha per costruzione) ma 'batte il de-levering' — ed e' il §B.")
|
|
|
|
# sensitivita': f asimmetrico (compri caro, vendi a modello)
|
|
print("\n SENSITIVITA' f ASIMMETRICO (si compra a f, si vende a modello — caso pessimo per il")
|
|
print(" roll anticipato, che nel modello base non paga f perche' non arriva a intrinseco):")
|
|
fu = F_MEAS.get(-0.10, F_LENSES_STATIC[1])
|
|
for roll in G_ROLL:
|
|
hs, _, _ = hedge_series(-0.10, 30, 0.010, roll, fu, SPREAD[30])
|
|
ha, _, _ = hedge_series(-0.10, 30, 0.010, roll, fu, SPREAD[30], f_sell=1.0)
|
|
es, ea = evaluate(book, hs), evaluate(book, ha)
|
|
print(f" roll={roll:<7} f simmetrico Δdrift {(es['dr_hedged']-es['dr_delev'])*100:+6.2f}pp"
|
|
f" f asimmetrico Δdrift {(ea['dr_hedged']-ea['dr_delev'])*100:+6.2f}pp")
|
|
|
|
# ---------------------------------------------------------------- §I. verdetto
|
|
print("\n" + "=" * 100)
|
|
print("§I. VERDETTO")
|
|
print("=" * 100)
|
|
for lab, _ in lenses:
|
|
sub = [r for r in all_rows if r["lens"] == lab]
|
|
nw = sum(1 for r in sub if r["wins"])
|
|
nred = sum(1 for r in sub if r["dd_reduced"])
|
|
cand = [r for r in sub if r["dd_reduced"]] or sub
|
|
bb = max(cand, key=lambda r: r["gap"])
|
|
print(f" {lab:<32}: DD scende {nred:>2}/{len(sub)} | batte il de-levering {nw:>2}/{len(sub)}"
|
|
f" | miglior Δdrift {bb['gap']*100:+6.2f} pp/a | ΔSharpe {bb['sh_hedged']-bb['sh_base']:+.3f}")
|
|
nw_real = sum(1 for r in pool if r["wins"])
|
|
nred_real = sum(1 for r in pool if r["dd_reduced"])
|
|
verdict = "REFUTATO" if nw_real == 0 else ("PARZIALE" if nw_real < len(pool) * 0.5 else "REGGE")
|
|
print(f"\n >>> {verdict}: al f realistico il maxDD scende in {nred_real}/{len(pool)} celle e la")
|
|
print(f" copertura batte il de-levering in {nw_real}/{len(pool)}.")
|
|
print( " ⚠️ Il fatto che il maxDD SALGA nella maggior parte delle celle e' il risultato piu'")
|
|
print( " forte del filone: prima ancora del prezzo, la premessa 'la copertura riduce il")
|
|
print( " drawdown' e' FALSA su questo libro — il bleed del premio cade DENTRO i drawdown")
|
|
print( " (che sono lunghi e poco profondi) mentre il payoff cade su singoli giorni di")
|
|
print( " crollo che NON sono il fondo del drawdown.")
|
|
print(f"\n Tempo: {time.time()-t0:.0f}s")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|