Spaces:
Runtime error
Runtime error
File size: 4,839 Bytes
8f1601b | 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 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 | from __future__ import annotations
from datetime import date, datetime
from typing import Any
import pandas as pd
import yfinance as yf
from .schemas import OptionChain, OptionContract, UnderlyingQuote
def none_if_nan(value: Any) -> Any:
if pd.isna(value):
return None
return value
def to_float(value: Any) -> float | None:
value = none_if_nan(value)
return float(value) if value is not None else None
def to_int(value: Any) -> int | None:
value = none_if_nan(value)
return int(value) if value is not None else None
def get_price_history(
symbol: str,
period: str = "1y",
interval: str = "1d",
start: str | None = None,
end: str | None = None,
) -> pd.DataFrame:
ticker = yf.Ticker(symbol.strip().upper())
return ticker.history(period=period, interval=interval, start=start, end=end)
def get_current_quote(symbol: str) -> UnderlyingQuote:
symbol = symbol.strip().upper()
ticker = yf.Ticker(symbol)
data = ticker.history(period="1d", interval="1m")
if not data.empty:
latest_row = data.iloc[-1]
return UnderlyingQuote(
symbol=symbol,
current_price=float(latest_row["Close"]),
open=float(latest_row["Open"]),
high=float(latest_row["High"]),
low=float(latest_row["Low"]),
volume=int(latest_row["Volume"]),
timestamp=str(data.index[-1]),
data_type="intraday_1m",
)
info = ticker.info
current_price = (
info.get("regularMarketPrice")
or info.get("previousClose")
or info.get("ask")
or info.get("bid")
)
return UnderlyingQuote(
symbol=symbol,
current_price=float(current_price) if current_price else None,
open=to_float(info.get("regularMarketOpen") or info.get("open")),
high=to_float(info.get("regularMarketDayHigh") or info.get("dayHigh")),
low=to_float(info.get("regularMarketDayLow") or info.get("dayLow")),
volume=to_int(info.get("regularMarketVolume") or info.get("volume")),
timestamp=datetime.utcnow().isoformat(timespec="seconds"),
data_type="cached_info",
short_name=info.get("shortName", ""),
)
def list_option_expirations(symbol: str) -> list[str]:
ticker = yf.Ticker(symbol.strip().upper())
return list(ticker.options or [])
def liquidity_warnings(row: pd.Series) -> list[str]:
warnings = []
bid = to_float(row.get("bid"))
ask = to_float(row.get("ask"))
volume = to_int(row.get("volume")) or 0
open_interest = to_int(row.get("openInterest")) or 0
if bid is None or ask is None or bid <= 0 or ask <= 0:
warnings.append("missing_or_zero_bid_ask")
elif ask > 0 and (ask - bid) / ask > 0.25:
warnings.append("wide_bid_ask_spread")
if volume <= 0:
warnings.append("zero_volume")
if open_interest <= 0:
warnings.append("zero_open_interest")
return warnings
def row_to_contract(row: pd.Series, option_type: str, expiration: str) -> OptionContract:
bid = to_float(row.get("bid"))
ask = to_float(row.get("ask"))
mid = (bid + ask) / 2 if bid is not None and ask is not None and bid > 0 and ask > 0 else None
days_to_expiration = max((date.fromisoformat(expiration) - date.today()).days, 0)
return OptionContract(
contract_symbol=str(row.get("contractSymbol", "")),
option_type=option_type,
expiration=expiration,
strike=float(row.get("strike")),
bid=bid,
ask=ask,
mid=mid,
last_price=to_float(row.get("lastPrice")),
volume=to_int(row.get("volume")),
open_interest=to_int(row.get("openInterest")),
implied_volatility=to_float(row.get("impliedVolatility")),
in_the_money=bool(row.get("inTheMoney", False)),
days_to_expiration=days_to_expiration,
liquidity_warnings=liquidity_warnings(row),
)
def get_option_chain(symbol: str, expiration: str | None = None) -> OptionChain:
symbol = symbol.strip().upper()
ticker = yf.Ticker(symbol)
expirations = list(ticker.options or [])
if not expirations:
raise ValueError(f"No option expirations found for {symbol}.")
expiration = expiration or expirations[0]
if expiration not in expirations:
raise ValueError(f"Expiration {expiration} is not available for {symbol}.")
chain = ticker.option_chain(expiration)
quote = get_current_quote(symbol)
calls = [row_to_contract(row, "call", expiration) for _, row in chain.calls.iterrows()]
puts = [row_to_contract(row, "put", expiration) for _, row in chain.puts.iterrows()]
return OptionChain(
symbol=symbol,
expiration=expiration,
underlying_price=quote.current_price,
calls=calls,
puts=puts,
)
|