PythagorasGoal/Old/scripts/waste/W21_vrp_real_dvol.py

"""S2-11: VRP con DVOL REALE — unico test valido.
Solo 90 giorni di dati, ma REALI.
Confronta DVOL (IV reale Deribit) vs RV realizzata.
"""
from __future__ import annotations
import sys
sys.path.insert(0, ".")

import numpy as np
import pandas as pd
from src.data.downloader import load_data

FEE_ROUNDTRIP = 0.0052
INITIAL = 1000


def rv_ann(close, window):
    lr = np.diff(np.log(np.where(close == 0, 1e-10, close)))
    r = np.full(len(close), np.nan)
    for i in range(window, len(lr)):
        r[i + 1] = np.std(lr[i - window : i]) * np.sqrt(24 * 365)
    return r


def straddle_prem(iv_pct, dte_h):
    """iv_pct è la IV in decimale (es 0.50 = 50%)."""
    if iv_pct <= 0 or dte_h <= 0:
        return 0
    return iv_pct * np.sqrt(dte_h / (24 * 365)) * 0.8


for asset in ["ETH", "BTC"]:
    print(f"\n{'='*70}")
    print(f"  {asset} — VRP CON DVOL REALE (90 giorni)")
    print(f"{'='*70}")

    df_price = load_data(asset, "1h")
    df_dvol = pd.read_parquet(f"data/raw/{asset.lower()}_dvol.parquet")

    close = df_price["close"].values
    ts_price = df_price["timestamp"].values
    n = len(close)

    dvol_ts = df_dvol["timestamp"].values
    dvol_vals = df_dvol["dvol"].values / 100  # converti a decimale

    rv_24 = rv_ann(close, 24)
    rv_48 = rv_ann(close, 48)

    # Allinea DVOL ai candles 1h (DVOL è giornaliero)
    dvol_aligned = np.full(n, np.nan)
    for j in range(len(dvol_ts)):
        mask = (ts_price >= dvol_ts[j]) & (ts_price < dvol_ts[j] + 86400000)
        dvol_aligned[mask] = dvol_vals[j]

    valid_count = np.sum(~np.isnan(dvol_aligned))
    print(f"  Candele con DVOL reale: {valid_count}")
    print(f"  DVOL range: {np.nanmin(dvol_aligned)*100:.1f}% — {np.nanmax(dvol_aligned)*100:.1f}%")

    # Analisi IV vs RV reale
    iv_rv_ratios = []
    for i in range(n):
        if np.isnan(dvol_aligned[i]) or np.isnan(rv_24[i]) or rv_24[i] <= 0:
            continue
        iv_rv_ratios.append(dvol_aligned[i] / rv_24[i])

    if iv_rv_ratios:
        print(f"\n  IV/RV ratio REALE:")
        print(f"    Mean:   {np.mean(iv_rv_ratios):.3f}")
        print(f"    Median: {np.median(iv_rv_ratios):.3f}")
        print(f"    Min:    {np.min(iv_rv_ratios):.3f}")
        print(f"    Max:    {np.max(iv_rv_ratios):.3f}")
        print(f"    >1 (VRP+): {sum(1 for r in iv_rv_ratios if r > 1)/len(iv_rv_ratios)*100:.0f}% del tempo")

    # Backtest VRP reale
    for dte in [24, 48]:
        print(f"\n  --- DTE={dte}h ---")
        capital = float(INITIAL)
        trades = []
        daily_done = set()

        for i in range(100, n - dte):
            if np.isnan(dvol_aligned[i]) or np.isnan(rv_24[i]):
                continue

            ts_dt = pd.Timestamp(ts_price[i], unit="ms", tz="UTC")
            if ts_dt.hour != 8:
                continue

            day = ts_dt.strftime("%Y-%m-%d")
            if day in daily_done:
                continue

            iv = dvol_aligned[i]
            rv = rv_24[i]

            # Filtro regime: skip se RV > IV (no premium)
            if rv > iv:
                continue

            prem = straddle_prem(iv, dte)
            spot = close[i]
            exit_idx = min(i + dte, n - 1)
            actual_move = abs(close[exit_idx] - spot) / spot

            pos_pct = 0.10
            if actual_move <= prem:
                raw = (prem - actual_move) * pos_pct
            else:
                raw = -(actual_move - prem) * pos_pct
                raw = max(raw, -pos_pct * 0.05)

            net = raw - FEE_ROUNDTRIP * pos_pct
            capital += capital * net
            capital = max(capital, 10)

            trades.append({
                "day": day,
                "iv": iv * 100,
                "rv": rv * 100,
                "premium": prem * 100,
                "move": actual_move * 100,
                "pnl": net * capital,
                "win": raw > 0,
            })
            daily_done.add(day)

        if not trades:
            print("    Nessun trade!")
            continue

        wins = sum(1 for t in trades if t["win"])
        acc = wins / len(trades) * 100
        ret = (capital - INITIAL) / INITIAL * 100
        avg_iv = np.mean([t["iv"] for t in trades])
        avg_rv = np.mean([t["rv"] for t in trades])
        avg_prem = np.mean([t["premium"] for t in trades])
        avg_move = np.mean([t["move"] for t in trades])

        print(f"    Trades:   {len(trades)}")
        print(f"    Accuracy: {acc:.1f}%")
        print(f"    Return:   {ret:+.1f}%")
        print(f"    Capital:  €{capital:.0f}")
        print(f"    Avg IV:   {avg_iv:.1f}%")
        print(f"    Avg RV:   {avg_rv:.1f}%")
        print(f"    Avg Prem: {avg_prem:.2f}%")
        print(f"    Avg Move: {avg_move:.2f}%")
        print(f"    IV > Move (win condition): {sum(1 for t in trades if t['premium'] > t['move'])/len(trades)*100:.0f}%")

        # Worst trade
        worst = min(trades, key=lambda t: t["pnl"])
        print(f"    Worst:    day={worst['day']} iv={worst['iv']:.1f}% move={worst['move']:.2f}%")