Spaces:
Sleeping
Sleeping
File size: 2,580 Bytes
da46dbf | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 | 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)
|