data.py

"""Market data access via Yahoo Finance's chart API.

Hits ``query1.finance.yahoo.com/v8/finance/chart`` directly using ``curl_cffi``
with browser impersonation. Two reasons for this approach over the ``yfinance``
high-level API:

  * ``yfinance``'s cookie bootstrap targets ``fc.yahoo.com``, which fails in
    environments where that host is DNS-blocked.
  * Yahoo fingerprints the TLS handshake and returns HTTP 429 to plain Python
    HTTP clients. ``curl_cffi`` impersonates Chrome's TLS signature, which is
    what ``yfinance`` itself uses internally.

Yahoo intraday history limits:
  1m -> ~7 days     2m/5m/15m/30m -> ~60 days     1h -> ~730 days
"""
from __future__ import annotations

import time

import pandas as pd
from curl_cffi import requests
from curl_cffi.requests.exceptions import RequestException

_CHART_URL = "https://query1.finance.yahoo.com/v8/finance/chart/{symbol}"
_OHLCV = ["open", "high", "low", "close", "volume"]

# Yahoo rejects (HTTP 422) windows wider than these per intraday interval.
_MAX_DAYS = {"1m": 7, "2m": 59, "5m": 59, "15m": 59,
             "30m": 59, "60m": 729, "90m": 59, "1h": 729}


def _period_to_seconds(period: str) -> int:
    """Convert a period string like '60d', '3mo', '1y' to seconds."""
    period = period.strip().lower()
    for suffix, mult in (("mo", 2_592_000), ("d", 86_400),
                          ("w", 604_800), ("y", 31_536_000)):
        if period.endswith(suffix):
            return int(float(period[: -len(suffix)]) * mult)
    raise ValueError(f"unrecognized period: {period!r} (use e.g. 5d, 60d, 3mo, 1y)")


def _fetch(symbol: str, interval: str, period: str,
           prepost: bool = False, retries: int = 4) -> dict:
    """Call the Yahoo chart API, retrying through transient errors / HTTP 429."""
    now = int(time.time())
    span = _period_to_seconds(period)
    if interval in _MAX_DAYS:
        span = min(span, _MAX_DAYS[interval] * 86_400)
    params = {
        "interval": interval,
        "period1": now - span,
        "period2": now,
        "includePrePost": "true" if prepost else "false",
    }
    last_error = "unknown error"
    for attempt in range(retries):
        try:
            resp = requests.get(_CHART_URL.format(symbol=symbol), params=params,
                                impersonate="chrome", timeout=15)
            if resp.status_code == 429:
                last_error = "rate limited by Yahoo (HTTP 429)"
                time.sleep(1.5 * (attempt + 1))
                continue
            resp.raise_for_status()
            return resp.json()
        except RequestException as exc:
            last_error = str(exc)
            time.sleep(1.0 * (attempt + 1))
    raise ConnectionError(f"Yahoo Finance request failed for {symbol}: {last_error}")


def _to_frame(payload: dict) -> pd.DataFrame:
    """Turn a Yahoo chart payload into a clean, tz-aware OHLCV frame."""
    chart = payload.get("chart", {})
    if chart.get("error"):
        raise ValueError(f"Yahoo Finance error: {chart['error']}")
    result = chart.get("result")
    if not result:
        raise ValueError("no data returned from Yahoo Finance (bad ticker?)")

    res = result[0]
    timestamps = res.get("timestamp")
    if not timestamps:
        raise ValueError("no data returned from Yahoo Finance (empty date range)")

    quote = res["indicators"]["quote"][0]
    tz = res.get("meta", {}).get("exchangeTimezoneName", "UTC")
    df = pd.DataFrame(
        {col: quote.get(col) for col in _OHLCV},
        index=pd.to_datetime(timestamps, unit="s", utc=True),
    )
    df.index = df.index.tz_convert(tz)
    df.index.name = "timestamp"
    return df.dropna()


def get_intraday(ticker: str, interval: str = "5m", period: str = "60d") -> pd.DataFrame:
    """Fetch intraday OHLCV bars for a single ticker (regular session only)."""
    return _to_frame(_fetch(ticker.upper(), interval, period, prepost=False))


def get_daily(ticker: str, period: str = "1y") -> pd.DataFrame:
    """Fetch daily OHLCV bars for a single ticker."""
    return _to_frame(_fetch(ticker.upper(), "1d", period))


def latest_price(ticker: str) -> float:
    """Most recent traded price for a ticker."""
    df = get_intraday(ticker, interval="1m", period="1d")
    return float(df["close"].iloc[-1])