PythagorasGoal/scripts/research/options_vrp_managed.py

"""VRP01 + GESTIONE ATTIVA (test del doc 'strategia-credit-spread-eth', 2026-06-20).

Innesta sul put credit spread di VRP01 le regole di gestione intra-trade del documento:
  - profit-take 50% del credito
  - stop-loss stretto 1.5x il credito (debito di chiusura)
  - VOL-STOP: chiudi se DVOL sale >=10 punti dall'apertura (regola crypto-specifica, NUOVA)
  - delta-exit: chiudi se |delta| dello short put >= 0.30 (niente rolling/difesa)
  - time-stop 7 DTE
Confronto A/B ONESTO sugli STESSI ingressi gated (VRP>0 + IV-rank>0.30) e dati certificati:
  BASE = hold-to-expiry (come VRP01)   vs   MANAGED = stesso trade con la gestione attiva.
Il MTM giornaliero dello spread usa BS sul path certificato + DVOL reale (causale: decisione al
giorno j con dati <= j). CAVEAT invariato: premio MODELLATO su DVOL ATM (no skew), nessun fill di
stress reale -> LEAD, non deploy. Qui misuriamo solo SE la gestione attiva taglia la coda.

    uv run python scripts/research/options_vrp_managed.py
"""
from __future__ import annotations
import sys
from pathlib import Path

import numpy as np
import pandas as pd
from scipy.stats import norm

sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
from src.data.downloader import load_data
from src.strategies.trend_portfolio import resample_1d
from src.portfolio.portfolio import to_daily, metrics, HOLDOUT
from src.portfolio.sleeves import _bs_put, _strike_from_delta, VRP_CFG, _HL_DIR

CFG = dict(VRP_CFG)  # short_delta -0.28, long_delta -0.10, f 1.0, gate_ivr 0.30, crash_skip 0.90, fee_frac 0.125


def _put_delta_mag(S, K, T, sig):
    if T <= 0 or sig <= 0:
        return 1.0 if S < K else 0.0
    d1 = (np.log(S / K) + 0.5 * sig ** 2 * T) / (sig * np.sqrt(T))
    return float(norm.cdf(-d1))            # |delta| dello short put (=N(-d1))


def simulate(asset: str, tenor_d: int, mode: str = "hte"):
    """mode: 'hte' hold-to-expiry | 'full' tutte le regole | 'volstop' solo vol-stop DVOL+10 (+PT50).
    Ritorna (serie rendimenti per-trade indicizzata alla data di uscita, dict conteggio exit)."""
    manage = mode != "hte"
    full = mode == "full"
    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()
    px = J["px"].values
    dvf = J["dvol"].values / 100.0
    idx = J.index
    n = len(px)
    tn = tenor_d
    f, fee = CFG["f"], CFG["fee_frac"]
    rets, exits = {}, {}
    i = 60
    while i + tn < n:
        S0, sig0 = px[i], dvf[i]
        # --- gates d'ingresso identici a VRP01 (causali) ---
        skip = False
        if i >= 31:
            rv = np.std(np.diff(np.log(px[i - 30:i + 1]))) * np.sqrt(365.25)
            if (sig0 - rv) <= 0:                                   # VRP>0
                skip = True
        if not skip and i >= 60:
            ivr = float((dvf[:i] < dvf[i]).mean())                # IV-rank espandente causale
            if ivr < CFG["gate_ivr"] or ivr > CFG["crash_skip"]:
                skip = True
        if skip:
            i += tn
            continue
        T0 = tn / 365.25
        Ks = _strike_from_delta(S0, T0, sig0, CFG["short_delta"])
        Kl = _strike_from_delta(S0, T0, sig0, CFG["long_delta"])
        net_prem = (_bs_put(S0, Ks, T0, sig0) - _bs_put(S0, Kl, T0, sig0)) * f
        if net_prem <= 0:
            i += tn
            continue
        reason, pnl, exit_j = None, None, i + tn
        if manage:
            for j in range(i + 1, i + tn):                        # giorni STRETTAMENTE prima della scadenza
                Trem = (i + tn - j) / 365.25
                Sj, sigj = px[j], dvf[j]
                sval = _bs_put(Sj, Ks, Trem, sigj) - _bs_put(Sj, Kl, Trem, sigj)   # MTM dello spread
                if sval <= 0.5 * net_prem:
                    reason, pnl, exit_j = "PT50", net_prem - sval, j; break
                if (sigj - sig0) >= 0.10:                          # VOL-STOP (la regola crypto nuova del doc)
                    reason, pnl, exit_j = "VOLSTOP", net_prem - sval, j; break
                if full and sval >= 1.5 * net_prem:
                    reason, pnl, exit_j = "SL150", net_prem - sval, j; break
                if full and _put_delta_mag(Sj, Ks, Trem, sigj) >= 0.30:
                    reason, pnl, exit_j = "DELTA", net_prem - sval, j; break
                if full and (i + tn - j) <= 7:
                    reason, pnl, exit_j = "TIME7", net_prem - sval, j; break
        if reason is None:                                        # scadenza
            S1 = px[i + tn]
            payoff = max(0.0, Ks - S1) - max(0.0, Kl - S1)
            pnl, reason, exit_j = net_prem - payoff, "expiry", i + tn
        pnl -= fee * abs(net_prem)                                # fee d'ingresso (su entrambe le gambe via net_prem)
        if reason != "expiry":
            pnl -= fee * abs(net_prem)                            # fee di chiusura anticipata (ricompro lo spread)
        rets[idx[exit_j]] = pnl / Ks
        exits[reason] = exits.get(reason, 0) + 1
        i += tn
    return pd.Series(rets).sort_index(), exits


def daily(series):
    if series.empty:
        return series
    days = pd.date_range(series.index.min().normalize(), series.index.max().normalize(), freq="1D", tz="UTC")
    out = pd.Series(0.0, index=days)
    out.loc[series.index.normalize()] = series.values
    return out


def report(label, perTrade):
    dl = to_daily(daily(perTrade))
    m = metrics(dl)
    mh = metrics(dl[dl.index >= HOLDOUT])
    wins = float((perTrade > 0).mean()) * 100
    worst = float(perTrade.min()) * 100
    print(f"  {label:<22s} n={len(perTrade):>3d} win={wins:>4.0f}% ret={m['ret']*100:>+6.0f}% "
          f"Sh={m['sharpe']:>5.2f} DD={m['maxdd']*100:>4.1f}% HOLD Sh={mh['sharpe']:>+5.2f} "
          f"worst-trade={worst:>+5.1f}%")
    return dl


def main():
    print("=" * 100)
    print("  VRP01 hold-to-expiry  vs  GESTIONE ATTIVA (vol-stop DVOL+10, SL 1.5x, PT50, delta-exit, 7DTE)")
    print("  Stessi ingressi gated (VRP>0 + IV-rank>0.30), dati certificati, premio MODELLATO su DVOL (no skew)")
    print("=" * 100)
    combos = {}
    for asset in ("ETH", "BTC"):
        print(f"\n--- {asset} ---")
        report("VRP01 live (7d HtE)", simulate(asset, 7, "hte")[0])     # riferimento live
        # confronto equo a tenor 14 (range del doc), STESSI ingressi
        b14, _ = simulate(asset, 14, "hte")
        v14, exv = simulate(asset, 14, "volstop")                       # SOLO vol-stop (la regola nuova)
        m14, exm = simulate(asset, 14, "full")                          # tutte le regole del doc
        report("14d hold-to-expiry", b14)
        report("14d +vol-stop only", v14); print(f"     exit volstop: {exv}")
        report("14d FULL managed", m14); print(f"     exit full: {exm}")
        combos[asset] = dict(base14=daily(b14), vol14=daily(v14), man14=daily(m14))

    # combo 50/50 BTC+ETH (come lo sleeve VRP01) — il confronto che conta per il portafoglio
    print("\n--- COMBO 50/50 BTC+ETH (sleeve-level) ---")
    for tag, key in (("14d hold-to-expiry", "base14"), ("14d +vol-stop only", "vol14"), ("14d FULL managed", "man14")):
        J = pd.concat({"B": combos["BTC"][key], "E": combos["ETH"][key]}, axis=1, join="outer").fillna(0.0)
        combo = to_daily(0.5 * J["B"] + 0.5 * J["E"])
        m, mh = metrics(combo), metrics(combo[combo.index >= HOLDOUT])
        print(f"  {tag:<22s} Sh={m['sharpe']:>5.2f} DD={m['maxdd']*100:>4.1f}% ret={m['ret']*100:>+6.0f}% "
              f"HOLD Sh={mh['sharpe']:>+5.2f}")
    print("\n  Lettura: la gestione attiva VALE se taglia maxDD e worst-trade SENZA distruggere Sharpe/ritorno.")
    print("  Caveat invariato: premio modellato su DVOL ATM (no skew) + nessun fill di stress reale -> LEAD, non deploy.")


if __name__ == "__main__":
    main()