"""R0730 VRP-PT — profit-take al 50% del credito su VRP01, e cosa ne resta a $3.000. DOMANDA. VRP01 come codificato tiene il credit spread FINO A SCADENZA: in `sleeves._vrp_weekly_asset` il payoff e' calcolato su `S1 = px[i + tn]` e non esiste nessuna gestione infra-settimana. La misura del 30/07 sulle quote REALI ha trovato che un profit-take al 50% del credito sarebbe scattato in **13 settimane su 15**, e che il peggior mark infra-settimana e' mediana -6.9% / minimo -81.7% su trade finiti in utile. Questo script chiede se quella gestione MIGLIORA lo sleeve, e cosa sopravvive a $3.000. L'idea non e' nuova ed e' la regola 1 del motore di cerbero-bite (`debit <= 50% credito`), progetto dismesso il 30/07. Qui non si eredita il suo giudizio — non ne ha uno: 59 decisioni, tutte no_entry, 0 posizioni. Si misura. COSA E' GIA' STATO RIFIUTATO, e non si rifa'. L'onda 03/07 (`r0703_vrpimp_ddoverlay`) ha testato 4 overlay di protezione DD — exit su spike DVOL, **stop-loss MTM a 1.5/2/3x il credito**, ala di coda, cooldown — e li ha REFUTATI 4/4 col null del de-levering. Il profit-take NON era fra quelli: e' un'uscita sul lato del PROFITTO, non del rischio. E' l'unico grado di liberta' rimasto non misurato su questo sleeve. METODO. (1) Replica bit-exact di `_vrp_weekly_asset` (assert max|diff| == 0). Senza questa, ogni delta misurato dopo e' rumore di implementazione. (2) Marks giornalieri: dentro la settimana lo spread e' riprezzato con lo STESSO pricer dell'ingresso (BS su DVOL corrente, T residuo). Chiudo quando `debit <= (1 - pt) * credito`. (3) NULL DEL DE-LEVERING, obbligatorio: un claim "meno drawdown" parte REFUTED finche' non batte il canonico semplicemente ridotto di size. Confronto a vol-matched E a DD-matched. (4) Fee REALI per gamba invece della convenzione forfettaria dello sleeve: `min(0.03% del sottostante, 12.5% del premio dell'opzione)`, cioe' il listino Deribit vero. Applicata IDENTICA a canonico e variante: un'uscita anticipata paga 2 gambe in piu', ed e' il costo principale di questa idea. (5) $3.000: contratti INTERI (ETH min_trade_amount=1 -> 1 ETH), BTC fuori (min 0.1 BTC = ~$6.500 di nozionale per gamba), collaterale = strike corto (convenzione cash-secured dello sleeve). f. Il fattore di prezzo misurato il 30/07 (f=0.73 sulle quote reali) e' applicato UNIFORMEMENTE al valore dello spread, all'ingresso e all'uscita: se il modello sbaglia il prezzo di quella struttura, lo sbaglia anche quando la si ricompra. Conseguenza utile: il TRIGGER del profit-take e' invariante in f (`f*debit <= 0.5*f*credito`), quindi f scala il PnL ma non cambia QUANDO si esce. uv run python scripts/research/r0730_vrp_profit_take.py uv run python scripts/research/r0730_vrp_profit_take.py --capital 3000 """ from __future__ import annotations import argparse import sys from pathlib import Path import numpy as np import pandas as pd PROJECT_ROOT = Path(__file__).resolve().parents[2] sys.path.insert(0, str(PROJECT_ROOT)) from src.data.downloader import load_data # noqa: E402 from src.portfolio.portfolio import metrics # noqa: E402 from src.portfolio.sleeves import ( # noqa: E402 VRP_CFG, _bs_put, _strike_from_delta, _vrp_weekly_asset, _HL_DIR, ) from src.strategies.trend_portfolio import resample_1d # noqa: E402 HOLDOUT = pd.Timestamp("2025-01-01", tz="UTC") ASSETS = ("BTC", "ETH") PT_GRID = (0.25, 0.50, 0.75) F_GRID = (1.00, 0.80, 0.73) TENOR_GRID = (7, 10, 14, 18, 21, 28) # 18 = la scadenza di cerbero-bite # listino Deribit opzioni, letto da public/get_instruments il 2026-07-30 FEE_UNDERLYING = 0.0003 # 0.03% del sottostante per contratto, per gamba FEE_PREMIUM_CAP = 0.125 # ...ma mai piu' del 12.5% del premio dell'opzione MIN_TRADE = {"BTC": 0.1, "ETH": 1.0} # --------------------------------------------------------------------------- # dati: identici a quelli del sleeve # --------------------------------------------------------------------------- def series(asset: str): df = resample_1d(load_data(asset, "1h")) s = pd.Series(df["close"].values.astype(float), index=pd.to_datetime(df["datetime"])) if s.index.tz is None: s.index = s.index.tz_localize("UTC") dv = pd.read_parquet(_HL_DIR / f"dvol_{asset.lower()}.parquet") d = pd.Series(dv["close"].values.astype(float), index=pd.to_datetime(dv["timestamp"], unit="ms", utc=True)) J = pd.concat({"px": s, "dvol": d}, axis=1, join="inner").sort_index().dropna() return J["px"].values, J["dvol"].values / 100.0, J.index def leg_fee(spot: float, premium: float) -> float: """Fee Deribit di UNA gamba: 0.03% del sottostante, cap al 12.5% del premio.""" return min(FEE_UNDERLYING * spot, FEE_PREMIUM_CAP * max(premium, 0.0)) # --------------------------------------------------------------------------- # il motore: un solo grado di liberta' per volta # --------------------------------------------------------------------------- def run_asset(asset: str, pt: float | None = None, f: float = 1.0, real_fee: bool = True, redeploy: bool = False, tenor: int | None = None): """Ritorna (serie settimanale, diagnostica). `pt=None` = canonico (tieni a scadenza). `real_fee=False` riproduce la convenzione forfettaria dello sleeve (per la replica bit-exact). `redeploy=True` rientra subito dopo un'uscita anticipata invece di aspettare la cadenza settimanale (strategia DIVERSA, non un overlay). """ px, dvf, idx = series(asset) n = len(px) cfg = VRP_CFG tn = int(tenor if tenor is not None else cfg["tenor_d"]) T0 = tn / 365.25 rets: dict[pd.Timestamp, float] = {} n_trades = n_early = 0 days_held: list[int] = [] worst_mark: list[float] = [] i = 60 while i + tn < n: S0 = px[i] sig = dvf[i] skip = False if cfg["gate_vrp"] and i >= 31: # VRP>0: DVOL > RV30 causale rv = np.std(np.diff(np.log(px[i - 30:i + 1]))) * np.sqrt(365.25) if (sig - rv) <= 0: skip = True if not skip and (cfg["gate_ivr"] > 0 or cfg["crash_skip"] < 1.0) and i >= 60: ivr = float((dvf[:i] < dvf[i]).mean()) # IV-rank espandente causale if cfg["gate_ivr"] > 0 and ivr < cfg["gate_ivr"]: skip = True if cfg["crash_skip"] < 1.0 and ivr > cfg["crash_skip"]: skip = True if skip: rets[idx[i + tn]] = 0.0 i += tn continue Ks = _strike_from_delta(S0, T0, sig, cfg["short_delta"]) Kl = _strike_from_delta(S0, T0, sig, cfg["long_delta"]) ps0, pl0 = _bs_put(S0, Ks, T0, sig), _bs_put(S0, Kl, T0, sig) credit = (ps0 - pl0) * f # --- fee d'ingresso --- if real_fee: fee = leg_fee(S0, ps0 * f) + leg_fee(S0, pl0 * f) else: fee = 0.0 # forfait applicato a valle # --- cammino infra-settimana --- exit_j = i + tn # default: scadenza early = False mk_worst = 0.0 if pt is not None: for j in range(i + 1, i + tn): Trem = (i + tn - j) / 365.25 Sj, sj = px[j], dvf[j] debit = (_bs_put(Sj, Ks, Trem, sj) - _bs_put(Sj, Kl, Trem, sj)) * f mk_worst = min(mk_worst, (credit - debit) / Ks) if debit <= (1.0 - pt) * credit: exit_j = j early = True break if early: Trem = (i + tn - exit_j) / 365.25 Sj, sj = px[exit_j], dvf[exit_j] psx, plx = _bs_put(Sj, Ks, Trem, sj), _bs_put(Sj, Kl, Trem, sj) cost = (psx - plx) * f if real_fee: fee += leg_fee(Sj, psx * f) + leg_fee(Sj, plx * f) pnl = credit - cost - fee else: S1 = px[i + tn] payoff = max(0.0, Ks - S1) - max(0.0, Kl - S1) if real_fee: # a scadenza: fee di regolamento solo sulle gambe ITM (0.015% del sottostante) if Ks > S1: fee += 0.5 * FEE_UNDERLYING * S1 if Kl > S1: fee += 0.5 * FEE_UNDERLYING * S1 pnl = credit - payoff - fee if not real_fee: pnl -= cfg["fee_frac"] * abs(credit) # convenzione dello sleeve rets[idx[exit_j]] = rets.get(idx[exit_j], 0.0) + pnl / Ks n_trades += 1 n_early += int(early) days_held.append(exit_j - i) worst_mark.append(mk_worst) i = (exit_j + 1) if (redeploy and early) else (i + tn) diag = dict(trades=n_trades, early=n_early, early_pct=100.0 * n_early / max(n_trades, 1), days_held=float(np.mean(days_held)) if days_held else float("nan"), worst_mark=float(np.median(worst_mark)) if worst_mark else float("nan")) return pd.Series(rets).sort_index(), diag def paired_trades(asset: str, pt: float, f: float = 1.0) -> pd.DataFrame: """Per OGNI ingresso, l'esito hold-to-expiry E quello col profit-take: unico confronto valido, perche' le due varianti escono in date diverse.""" px, dvf, idx = series(asset) n = len(px) cfg = VRP_CFG tn = cfg["tenor_d"] T0 = tn / 365.25 recs = [] i = 60 while i + tn < n: S0, sig = px[i], dvf[i] skip = False if cfg["gate_vrp"] and i >= 31: rv = np.std(np.diff(np.log(px[i - 30:i + 1]))) * np.sqrt(365.25) if (sig - rv) <= 0: skip = True if not skip and i >= 60: ivr = float((dvf[:i] < dvf[i]).mean()) if ivr < cfg["gate_ivr"] or ivr > cfg["crash_skip"]: skip = True if skip: i += tn continue Ks = _strike_from_delta(S0, T0, sig, cfg["short_delta"]) Kl = _strike_from_delta(S0, T0, sig, cfg["long_delta"]) credit = (_bs_put(S0, Ks, T0, sig) - _bs_put(S0, Kl, T0, sig)) * f ej = None for j in range(i + 1, i + tn): Trem = (i + tn - j) / 365.25 debit = (_bs_put(px[j], Ks, Trem, dvf[j]) - _bs_put(px[j], Kl, Trem, dvf[j])) * f if debit <= (1.0 - pt) * credit: ej = j break S1 = px[i + tn] hold = (credit - (max(0.0, Ks - S1) - max(0.0, Kl - S1))) / Ks if ej is None: ptv, early = hold, False else: Trem = (i + tn - ej) / 365.25 cost = (_bs_put(px[ej], Ks, Trem, dvf[ej]) - _bs_put(px[ej], Kl, Trem, dvf[ej])) * f ptv, early = (credit - cost) / Ks, True recs.append((hold, ptv, early)) i += tn return pd.DataFrame(recs, columns=["hold", "pt", "early"]) def combo(pt=None, f=1.0, real_fee=True, redeploy=False, tenor=None): """Book 50/50 BTC+ETH su griglia giornaliera, come `_vrp_combo_returns`.""" out, diags = {}, {} for a in ASSETS: out[a], diags[a] = run_asset(a, pt=pt, f=f, real_fee=real_fee, redeploy=redeploy, tenor=tenor) wk = pd.concat({"B": out["BTC"], "E": out["ETH"]}, axis=1, join="inner").mean(axis=1).sort_index() if wk.empty: return wk, diags days = pd.date_range(wk.index.min().normalize(), wk.index.max().normalize(), freq="1D", tz="UTC") daily = pd.Series(0.0, index=days) daily.loc[wk.index.normalize()] = wk.values return daily, diags # --------------------------------------------------------------------------- # metriche e null # --------------------------------------------------------------------------- def stats(daily: pd.Series) -> dict: m = metrics(daily) h = metrics(daily[daily.index >= HOLDOUT]) return dict(sh=m["sharpe"], dd=m["maxdd"], cagr=m["cagr"], sh_h=h["sharpe"], worst=float(daily.min())) def delever_null(canon: pd.Series, variant: pd.Series, match: str) -> dict: """Il canonico ridotto di size fa lo stesso lavoro? `match` in {'vol','dd'}.""" sv, sc = stats(variant), stats(canon) if match == "vol": k = variant.std() / canon.std() if canon.std() > 0 else 1.0 else: # DD-matched, per bisezione lo, hi = 0.01, 3.0 for _ in range(60): k = 0.5 * (lo + hi) if stats(canon * k)["dd"] < sv["dd"]: lo = k else: hi = k k = 0.5 * (lo + hi) sk = stats(canon * k) return dict(k=k, sh_null=sk["sh"], dd_null=sk["dd"], sh_var=sv["sh"], dd_var=sv["dd"], d_sh=sv["sh"] - sk["sh"], batte=(sv["sh"] > sk["sh"])) # --------------------------------------------------------------------------- # eseguibilita' a capitale dato # --------------------------------------------------------------------------- def executability(capital: float) -> dict: """Quanti contratti INTERI entrano, per asset, con collaterale = strike corto.""" rows = {} for a in ASSETS: px, dvf, idx = series(a) S0, sig = px[-1], dvf[-1] T0 = VRP_CFG["tenor_d"] / 365.25 Ks = _strike_from_delta(S0, T0, sig, VRP_CFG["short_delta"]) Kl = _strike_from_delta(S0, T0, sig, VRP_CFG["long_delta"]) unit = MIN_TRADE[a] collat = Ks * unit # cash-secured, per lotto minimo ps, pl = _bs_put(S0, Ks, T0, sig), _bs_put(S0, Kl, T0, sig) credit = (ps - pl) * unit fee_rt = 2 * (leg_fee(S0, ps) + leg_fee(S0, pl)) * unit # ingresso + uscita anticipata rows[a] = dict(spot=S0, Ks=Ks, Kl=Kl, unit=unit, collat=collat, lots=int(capital // collat), credit=credit, fee_rt=fee_rt, fee_pct_credit=100.0 * fee_rt / credit if credit > 0 else float("nan"), width=(Ks - Kl) * unit) return rows # --------------------------------------------------------------------------- def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("--capital", type=float, default=3000.0) args = ap.parse_args() print("=" * 92) print(" R0730 VRP-PT — profit-take al 50% del credito su VRP01") print("=" * 92) # ---- (0) replica bit-exact ------------------------------------------------ print("\n (0) REPLICA del sleeve (senza questa, ogni delta dopo e' rumore)") for a in ASSETS: mine, _ = run_asset(a, pt=None, f=VRP_CFG["f"], real_fee=False) ref = _vrp_weekly_asset(a) j = pd.concat({"m": mine, "r": ref}, axis=1, join="inner") d = float((j["m"] - j["r"]).abs().max()) n_only = len(mine) + len(ref) - 2 * len(j) flag = "OK" if (d == 0.0 and n_only == 0) else "⚠ DIVERGE" print(f" {a}: max|diff| = {d:.2e} date non condivise = {n_only} {flag}") if d != 0.0 or n_only != 0: print(" REPLICA FALLITA — mi fermo, il confronto non sarebbe valido.") return 1 # ---- (1) griglia profit-take ---------------------------------------------- print(f"\n (1) PROFIT-TAKE — fee REALI Deribit per gamba, f={F_GRID[0]:.2f}" f" (trigger invariante in f)") canon, dcan = combo(pt=None, f=1.0) sc = stats(canon) print(f" {'variante':<22}{'ShFULL':>8}{'ShHOLD':>8}{'maxDD':>8}{'CAGR':>8}" f"{'peggior sett':>14}{'usciti prima':>14}{'giorni':>8}") print(f" {'canonico (scadenza)':<22}{sc['sh']:>8.2f}{sc['sh_h']:>8.2f}" f"{sc['dd']*100:>7.1f}%{sc['cagr']*100:>7.1f}%{sc['worst']*100:>13.2f}%" f"{'—':>14}{7.0:>8.1f}") variants = {} for pt in PT_GRID: v, dg = combo(pt=pt, f=1.0) variants[pt] = v s = stats(v) early = np.mean([dg[a]["early_pct"] for a in ASSETS]) held = np.mean([dg[a]["days_held"] for a in ASSETS]) print(f" {'PT ' + str(int(pt*100)) + '% del credito':<22}{s['sh']:>8.2f}{s['sh_h']:>8.2f}" f"{s['dd']*100:>7.1f}%{s['cagr']*100:>7.1f}%{s['worst']*100:>13.2f}%" f"{early:>13.0f}%{held:>8.1f}") # ---- (2) null del de-levering --------------------------------------------- print("\n (2) NULL DEL DE-LEVERING — il canonico ridotto di size fa lo stesso lavoro?") print(" (regola del progetto: un claim 'meno DD' parte REFUTED finche' non lo batte)") for pt in PT_GRID: for match in ("vol", "dd"): r = delever_null(canon, variants[pt], match) verdict = "batte il null" if r["batte"] else "REFUTED" print(f" PT {int(pt*100):>2}% match {match:>3}: k={r['k']:.3f} " f"Sh null {r['sh_null']:>5.2f} vs variante {r['sh_var']:>5.2f} " f"(Δ {r['d_sh']:+.2f}) DD {r['dd_null']*100:.1f}% vs {r['dd_var']*100:.1f}% {verdict}") # ---- (3) sensibilita' a f -------------------------------------------------- print("\n (3) BANDA f — f scala il PnL ma NON il trigger (f*debit <= 0.5*f*credito)") print(f" {'f':>6}{'Sh canonico':>14}{'Sh PT50':>10}{'ΔSh':>8}" f"{'DD canonico':>14}{'DD PT50':>10}") for f in F_GRID: c, _ = combo(pt=None, f=f) v, _ = combo(pt=0.50, f=f) s0, s1 = stats(c), stats(v) print(f" {f:>6.2f}{s0['sh']:>14.2f}{s1['sh']:>10.2f}{s1['sh']-s0['sh']:>+8.2f}" f"{s0['dd']*100:>13.1f}%{s1['dd']*100:>9.1f}%") # ---- (4) redeploy ---------------------------------------------------------- print("\n (4) RIENTRO IMMEDIATO dopo l'uscita (strategia DIVERSA, non un overlay)") rd, drd = combo(pt=0.50, f=1.0, redeploy=True) s = stats(rd) tr = sum(drd[a]["trades"] for a in ASSETS) tr0 = sum(dcan[a]["trades"] for a in ASSETS) print(f" PT50 + rientro: Sh {s['sh']:.2f} ShHOLD {s['sh_h']:.2f} DD {s['dd']*100:.1f}%" f" CAGR {s['cagr']*100:.1f}% trade {tr} (canonico {tr0}, +{100*(tr/tr0-1):.0f}%)") # ---- (4-bis) confronto APPAIATO per ingresso: il meccanismo --------------- print("\n (4-bis) MECCANISMO — confronto appaiato PER INGRESSO") print(" (le serie sono indicizzate sulla data di USCITA, che il PT cambia: un inner-join") print(" fra le due confronterebbe solo le settimane senza uscita anticipata)") for a in ASSETS: R = paired_trades(a, pt=0.50) W, L = R[R.hold > 0], R[R.hold <= 0] print(f" {a}: {len(R)} trade, uscita anticipata {100*R.early.mean():.0f}%" f" — scatta sul {100*W.early.mean():.0f}% dei VINCENTI e sul {100*L.early.mean():.0f}% dei PERDENTI") print(f" vincenti {W.hold.mean()*100:+.3f}% -> {W.pt.mean()*100:+.3f}% ({100*(W.pt.mean()/W.hold.mean()-1):+.0f}%)" f" perdenti {L.hold.mean()*100:+.3f}% -> {L.pt.mean()*100:+.3f}% ({100*(L.pt.mean()/L.hold.mean()-1):+.0f}%)") # ---- (4-ter) ipotesi: funziona a scadenze piu' lunghe? -------------------- print("\n (4-ter) IPOTESI DA VERIFICARE: su 7g chi tocca +50% e' chi sarebbe scaduto") print(" senza valore, quindi il PT dovrebbe pagare a scadenze LUNGHE (bite usava 18g).") print(f" {'tenore':>8}{'Sh hold':>10}{'Sh PT50':>10}{'ΔSh':>8}{'DD hold':>10}{'DD PT50':>10}{'usciti':>9}") for tn in TENOR_GRID: c, _ = combo(pt=None, f=1.0, tenor=tn) v, dg = combo(pt=0.50, f=1.0, tenor=tn) s0, s1 = stats(c), stats(v) early = np.mean([dg[a]["early_pct"] for a in ASSETS]) print(f" {tn:>6}g{s0['sh']:>10.2f}{s1['sh']:>10.2f}{s1['sh']-s0['sh']:>+8.2f}" f"{s0['dd']*100:>9.1f}%{s1['dd']*100:>9.1f}%{early:>8.0f}%") print(" -> IPOTESI REFUTATA se ΔSh resta negativo ovunque. ⚠ La colonna 'Sh hold' NON e'") print(" un risultato: 6 celle scelte sul campione pieno; per crederci serve") print(" study_family_honest + deflated-Sharpe, non questo sweep.") # ---- (5) $3.000 ------------------------------------------------------------- print(f"\n (5) ESEGUIBILITA' A ${args.capital:,.0f} — contratti INTERI, collaterale = strike corto") ex = executability(args.capital) for a, r in ex.items(): stato = "ESEGUIBILE" if r["lots"] >= 1 else "FUORI" print(f" {a}: spot ${r['spot']:,.0f} lotto minimo {r['unit']} " f"collaterale/lotto ${r['collat']:,.0f} -> {r['lots']} lotti {stato}") print(f" credito ${r['credit']:.2f}/lotto · larghezza ${r['width']:.2f} · " f"fee round-trip ${r['fee_rt']:.2f} = {r['fee_pct_credit']:.1f}% del credito") tot = sum(r["lots"] for r in ex.values()) print(f"\n Il sleeve e' 50/50 BTC+ETH: a ${args.capital:,.0f} " f"{'diventa ETH-only' if ex['BTC']['lots'] == 0 and ex['ETH']['lots'] > 0 else 'non e collocabile'}.") print(f" Al PESO DI BOOK (12%) l'allocazione sarebbe ${args.capital*0.12:,.0f} = " f"{int(args.capital*0.12 // ex['ETH']['collat'])} lotti ETH.") if tot == 0: print(" -> a questo capitale lo sleeve NON e' eseguibile in nessuna forma.") print() return 0 if __name__ == "__main__": raise SystemExit(main())