src/option_implied_lab/oipd_adapter.py

from __future__ import annotations

from dataclasses import dataclass
from datetime import datetime
from typing import Any

import pandas as pd


@dataclass(frozen=True)
class OIPDConfig:
    risk_free_rate: float = 0.04
    method: str = "svi"
    pricing_engine: str = "black76"
    price_method: str = "mid"
    max_staleness_days: int = 3


@dataclass(frozen=True)
class OIPDDistributionResult:
    density: pd.DataFrame
    metadata: dict[str, Any]


def prepare_oipd_chain(options: pd.DataFrame, expiry: datetime) -> pd.DataFrame:
    s = options[pd.to_datetime(options["expiry"]) == pd.to_datetime(expiry)].copy()
    if s.empty:
        raise ValueError(f"No option rows for expiry={expiry}")

    s["option_type"] = (
        s["option_type"]
        .astype(str)
        .str.lower()
        .map({"call": "C", "put": "P"})
        .fillna(s["option_type"])
    )
    s["last_price"] = s["mid"].astype(float)
    s["expiry"] = pd.to_datetime(s["expiry"]).dt.tz_localize(None)

    needed_cols = ["strike", "option_type", "bid", "ask", "last_price", "expiry"]
    for col in ["bid", "ask"]:
        if col not in s.columns:
            s[col] = pd.NA
    out = s[needed_cols].copy()
    out = out.dropna(subset=["strike", "last_price", "option_type"]).reset_index(
        drop=True
    )
    if out.empty:
        raise ValueError("No valid rows after OIPD chain preparation")
    return out


def fit_oipd_distribution(
    chain: pd.DataFrame,
    spot: float,
    valuation_time: datetime,
    config: OIPDConfig,
) -> OIPDDistributionResult:
    try:
        from oipd import MarketInputs, VolCurve
    except ImportError as exc:
        raise RuntimeError("oipd is not installed. Run: uv sync --extra v2") from exc

    market = MarketInputs(
        risk_free_rate=float(config.risk_free_rate),
        valuation_date=valuation_time.date(),
        underlying_price=float(spot),
    )
    vc = VolCurve(
        method=config.method,
        pricing_engine=config.pricing_engine,
        price_method=config.price_method,
        max_staleness_days=int(config.max_staleness_days),
    )
    vc.fit(chain, market)
    prob = vc.implied_distribution()
    df = prob.density_results()
    density = df.rename(columns={"price": "strike", "pdf": "density"})[
        ["strike", "density", "cdf"]
    ].copy()

    metadata = {
        "method": config.method,
        "pricing_engine": config.pricing_engine,
        "price_method": config.price_method,
        "diagnostics": vc.diagnostics,
    }
    return OIPDDistributionResult(density=density, metadata=metadata)