PythagorasGoal/scripts/research/alt/runs/OPT07.py

"""OPT07 — Collar Overlay
IDEA: Long spot + buy protective put + sell covered call (zero-ish cost collar).
  - Long 1 unit spot BTC/ETH
  - Sell OTM call at strike K_call = S * exp(+call_otm * sigma * sqrt(T))
  - Buy OTM put at strike K_put  = S * exp(-put_otm * sigma * sqrt(T))
  Net premium ≈ call premium received - put premium paid (can be near-zero or small debit/credit
  depending on the strikes chosen).

Goal: reduce drawdown vs buy&hold by capping upside (call) and flooring downside (put).
      Does this improve risk-adjusted return (Sharpe)?

Hypothesis: the vol risk premium means we receive more on the call than we pay for the put
(IV > RV historically), so the collar should produce a positive carry vs buying naked insurance.
In a crash the put activates and limits losses. Net effect should be improved Sharpe.

MODELED: premiums computed via Black-Scholes with DVOL as IV (no skew, no slippage on options).
DVOL history starts 2021-03 -> backtest from 2021-03 only.
CAVEAT: modeled, lead-only.

Grid (4 configs, 1 TF = 4 study_weights calls -> <=8 total backtests):
  1. Symmetric collar: call OTM=0.10, put OTM=0.10 (weekly)
  2. Tighter collar:   call OTM=0.05, put OTM=0.05 (weekly)
  3. Asymmetric:       call OTM=0.05, put OTM=0.10 (debit collar, more protection, less upside cap)
  4. Asymmetric:       call OTM=0.10, put OTM=0.05 (credit collar, less protection, more upside cap)

Style: study_weights (continuous position ~1x long + option overlay adjustments at settlement).
"""

import sys
sys.path.insert(0, "/opt/docker/PythagorasGoal/scripts/research/alt")
import altlib as al

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


# ── Black-Scholes call and put prices ────────────────────────────────────────
def bs_call(S: float, K: float, T: float, sigma: float, r: float = 0.0) -> float:
    """Black-Scholes call price. T in years. sigma annualized."""
    if T <= 0 or sigma <= 0 or S <= 0 or K <= 0:
        return 0.0
    d1 = (np.log(S / K) + (r + 0.5 * sigma**2) * T) / (sigma * np.sqrt(T))
    d2 = d1 - sigma * np.sqrt(T)
    return float(S * norm.cdf(d1) - K * np.exp(-r * T) * norm.cdf(d2))


def bs_put(S: float, K: float, T: float, sigma: float, r: float = 0.0) -> float:
    """Black-Scholes put price via put-call parity."""
    c = bs_call(S, K, T, sigma, r)
    return float(c - S + K * np.exp(-r * T))


# ── Collar P&L per settlement cycle ──────────────────────────────────────────
def collar_cycle_return(S_start: float, S_end: float,
                        K_call: float, K_put: float,
                        call_prem: float, put_cost: float) -> float:
    """
    Compute the net return of a collar for one option cycle.

    At initiation:
      - Receive call_prem (sell call)
      - Pay put_cost (buy put)
      Net option carry = call_prem - put_cost (per unit of spot, as fraction of S_start)

    At settlement:
      Spot P&L: S_end / S_start - 1
      Call settled: -max(0, S_end - K_call) / S_start   (we're short call)
      Put settled: +max(0, K_put - S_end) / S_start      (we're long put)

    Total: (S_end/S_start - 1)
           - max(0, S_end - K_call) / S_start
           + max(0, K_put - S_end) / S_start
           + (call_prem - put_cost) / S_start

    Which simplifies to the textbook collar:
      If S_end >= K_call: net = (K_call/S_start - 1) + carry   (upside capped)
      If S_end <= K_put:  net = (K_put/S_start - 1) + carry    (downside floored)
      Otherwise:          net = (S_end/S_start - 1) + carry
    """
    carry = (call_prem - put_cost) / S_start  # net option premium (positive = net credit)

    if S_end >= K_call:
        return (K_call / S_start - 1.0) + carry
    elif S_end <= K_put:
        return (K_put / S_start - 1.0) + carry
    else:
        return (S_end / S_start - 1.0) + carry


# ── Build collar target array ─────────────────────────────────────────────────
def build_collar_target(close: np.ndarray, sigma_ann: np.ndarray,
                        call_otm: float, put_otm: float,
                        roll_bars: int, T_years: float) -> np.ndarray:
    """
    Build a synthetic 'effective position' array for the collar strategy.

    At each bar i, target[i] is held during bar i+1.
    On settlement bars: effective position encodes the full cycle's collar P&L.
    On non-settlement bars (mid-cycle): position = 1.0 (pure spot, no adjustment yet).

    Settlement bar technique (same as OPT01):
      target[i-1] * r_spot[i] ≈ cc_return for the cycle
      For multi-bar cycles: option_adj = collar_r - cycle_spot_r is applied at settlement.
    """
    n = len(close)
    target = np.ones(n)  # default: long spot

    # Find first bar with valid DVOL
    first_valid = np.where(np.isfinite(sigma_ann) & (sigma_ann > 0))[0]
    if len(first_valid) == 0:
        return target
    start_bar = int(first_valid[0])

    r_spot = al.simple_returns(close)

    # Initialize first collar at start_bar
    S0 = close[start_bar]
    sig0 = sigma_ann[start_bar]

    option_K_call = None
    option_K_put = None
    call_prem = 0.0
    put_cost = 0.0
    cycle_start_bar = start_bar
    cycle_start_price = S0

    if sig0 > 0 and np.isfinite(sig0):
        K_call = S0 * np.exp(call_otm * sig0 * np.sqrt(T_years))
        K_put  = S0 * np.exp(-put_otm * sig0 * np.sqrt(T_years))
        option_K_call = K_call
        option_K_put  = K_put
        call_prem = bs_call(S0, K_call, T_years, sig0)
        put_cost  = bs_put(S0, K_put,  T_years, sig0)

    for i in range(start_bar + 1, n):
        bars_in_cycle = i - cycle_start_bar

        if option_K_call is None or option_K_put is None:
            # No active collar -> pure spot
            target[i - 1] = 1.0
            # Try to re-initialize
            sig_i = sigma_ann[i]
            if np.isfinite(sig_i) and sig_i > 0:
                S_i = close[i]
                K_call = S_i * np.exp(call_otm * sig_i * np.sqrt(T_years))
                K_put  = S_i * np.exp(-put_otm * sig_i * np.sqrt(T_years))
                option_K_call = K_call
                option_K_put  = K_put
                call_prem = bs_call(S_i, K_call, T_years, sig_i)
                put_cost  = bs_put(S_i, K_put,  T_years, sig_i)
                cycle_start_bar = i
                cycle_start_price = S_i
            continue

        if bars_in_cycle >= roll_bars:
            # Settlement bar: compute collar payoff for the full cycle
            S_end   = close[i]
            S_start = cycle_start_price

            collar_r = collar_cycle_return(
                S_start, S_end,
                option_K_call, option_K_put,
                call_prem, put_cost
            )
            cycle_spot_r = S_end / S_start - 1.0

            # Encode the option adjustment on the settlement bar
            r_i = r_spot[i]
            option_adj = collar_r - cycle_spot_r  # premium carry ± cap/floor adjustments

            if abs(r_i) > 1e-10:
                target[i - 1] = 1.0 + option_adj / r_i
            else:
                # r_spot[i] ≈ 0: no spot movement on settlement bar -> just carry position=1
                # (option_adj can't be embedded cleanly, but it's typically small)
                target[i - 1] = 1.0

            # Roll new collar
            sig_new = sigma_ann[i]
            if np.isfinite(sig_new) and sig_new > 0:
                K_call_new = S_end * np.exp(call_otm * sig_new * np.sqrt(T_years))
                K_put_new  = S_end * np.exp(-put_otm * sig_new * np.sqrt(T_years))
                option_K_call = K_call_new
                option_K_put  = K_put_new
                call_prem = bs_call(S_end, K_call_new, T_years, sig_new)
                put_cost  = bs_put(S_end, K_put_new,  T_years, sig_new)
            else:
                option_K_call = None
                option_K_put  = None
                call_prem = 0.0
                put_cost  = 0.0

            cycle_start_bar = i
            cycle_start_price = S_end
        else:
            # Mid-cycle: hold spot (position=1, no adjustment)
            target[i - 1] = 1.0

    target = np.nan_to_num(target, nan=1.0)
    # Clip extreme values (guard against division artifacts when r_spot ≈ 0)
    target = np.clip(target, -5.0, 5.0)
    return target


# ── Per-asset runner (wraps study_weights) ────────────────────────────────────
def run_collar(call_otm: float, put_otm: float, roll_days: int = 7,
               tfs: tuple = ("1d",)) -> dict:
    """Run collar study for one config. Returns report dict."""
    name = f"OPT07-COLLAR-C{int(call_otm*100)}P{int(put_otm*100)}-roll{roll_days}d"
    T_years = roll_days / 365.25

    cells = []
    for tf in tfs:
        per_asset = {}
        fee_ok_all = True
        for asset in al.CERTIFIED:
            df = al.get(asset, tf)
            sigma_ann = al.dvol(df, asset) / 100.0
            roll_bars = roll_days  # 1d tf: 1 bar = 1 day

            tgt = build_collar_target(
                df["close"].values.astype(float),
                sigma_ann,
                call_otm=call_otm,
                put_otm=put_otm,
                roll_bars=roll_bars,
                T_years=T_years
            )

            base = al.eval_weights(df, tgt, fee_side=al.FEE_SIDE)
            sweep = {
                f"{2*f*100:.2f}%RT": al.eval_weights(df, tgt, fee_side=f)["full"]["sharpe"]
                for f in al.FEE_SWEEP
            }
            fee_ok = sweep.get("0.20%RT", -9) > 0
            fee_ok_all = fee_ok_all and fee_ok
            per_asset[asset] = dict(
                full=base["full"], holdout=base["holdout"],
                tim=base["time_in_market"],
                turnover=base["turnover_per_year"],
                fee_sweep=sweep, yearly=base["yearly"]
            )

        min_full = min(per_asset[a]["full"]["sharpe"] for a in al.CERTIFIED)
        min_hold = min(per_asset[a]["holdout"].get("sharpe", 0.0) for a in al.CERTIFIED)
        cells.append(dict(
            tf=tf, per_asset=per_asset,
            min_asset_full_sharpe=round(min_full, 3),
            min_asset_holdout_sharpe=round(min_hold, 3),
            full_sharpe=round(float(np.mean([per_asset[a]["full"]["sharpe"] for a in al.CERTIFIED])), 3),
            fee_survives=fee_ok_all
        ))

    verdict = al._verdict(cells)
    return dict(name=name, kind="weights", cells=cells, verdict=verdict)


# ── Main: small grid ──────────────────────────────────────────────────────────
if __name__ == "__main__":
    # Grid: 4 configs x 1 TF = 4 study calls = 8 total asset backtests (fine for 2 CPUs)
    CONFIGS = [
        # (call_otm, put_otm, roll_days, description)
        (0.10, 0.10, 7,  "symmetric 10%/10% weekly"),
        (0.05, 0.05, 7,  "tight 5%/5% weekly"),
        (0.05, 0.10, 7,  "debit collar: call 5% / put 10% -> more downside protection"),
        (0.10, 0.05, 7,  "credit collar: call 10% / put 5% -> less protection, net credit"),
    ]

    print("OPT07 Collar Overlay — MODELED on DVOL (lead-only, from 2021-03)")
    print("Long spot + sell OTM call + buy OTM put (zero-ish cost collar)")
    print()

    best_rep = None
    best_score = -999.0

    for call_otm, put_otm, roll_days, desc in CONFIGS:
        print(f"--- {desc} (call_otm={call_otm}, put_otm={put_otm}, roll={roll_days}d) ---")
        rep = run_collar(call_otm=call_otm, put_otm=put_otm, roll_days=roll_days, tfs=("1d",))
        print(al.fmt(rep))
        score = rep["verdict"].get("best_holdout_sharpe", -9)
        if score > best_score:
            best_score = score
            best_rep = rep
        print()

    print("=" * 60)
    print("BEST CONFIG:")
    print(al.fmt(best_rep))
    print()
    print("JSON:", al.as_json(best_rep))