Spaces:
Sleeping
Sleeping
File size: 1,929 Bytes
b6d53e2 | 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 | from __future__ import annotations
from datetime import datetime, timezone
import pandas as pd
import yfinance as yf
from .types import OptionSnapshot
def _normalize_chain(
df: pd.DataFrame, option_type: str, expiry: pd.Timestamp, spot: float
) -> pd.DataFrame:
out = df.copy()
out["option_type"] = option_type
out["expiry"] = pd.to_datetime(expiry).tz_localize(None)
out["mid"] = (out["bid"].fillna(0.0) + out["ask"].fillna(0.0)) / 2.0
out["volume"] = out.get("volume", 0.0)
out["openInterest"] = out.get("openInterest", 0.0)
out = out[
[
"expiry",
"option_type",
"strike",
"bid",
"ask",
"mid",
"volume",
"openInterest",
]
].copy()
out = out.dropna(subset=["strike", "mid"])
out = out[out["strike"] > 0].copy()
out["moneyness"] = out["strike"] / float(spot)
return out
def fetch_option_snapshot(ticker: str, max_expiries: int = 2) -> OptionSnapshot:
tk = yf.Ticker(ticker)
hist = tk.history(period="1d")
if hist.empty:
raise ValueError(f"No price history for ticker {ticker}")
spot = float(hist["Close"].iloc[-1])
expiries = tk.options[:max_expiries]
if not expiries:
raise ValueError(f"No option expiries for ticker {ticker}")
rows = []
for expiry_str in expiries:
chain = tk.option_chain(expiry_str)
expiry = pd.to_datetime(expiry_str)
rows.append(_normalize_chain(chain.calls, "call", expiry, spot))
rows.append(_normalize_chain(chain.puts, "put", expiry, spot))
options = pd.concat(rows, ignore_index=True)
options = options.sort_values(["expiry", "option_type", "strike"]).reset_index(
drop=True
)
return OptionSnapshot(
ticker=ticker,
snapshot_time=datetime.now(timezone.utc),
spot=spot,
options=options,
)
|