market_data/providers.py

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,
    )