anton-roos_finance/scripts/market_quote.py

#!/usr/bin/env python3
"""
market_quote.py

Fetch the latest quote for:
- Stocks/ETFs/indices (via yfinance)
- FX pairs (via ExchangeRate-API open access)

Usage:
  python scripts/market_quote.py AAPL
  python scripts/market_quote.py ^GSPC
  python scripts/market_quote.py USD/ZAR
"""

from __future__ import annotations

import argparse
import json
import os
import re
import time
from dataclasses import dataclass
from typing import Any, Dict, Optional, Tuple

import requests

CACHE_DIR = os.path.join(".cache", "market-tracker")
os.makedirs(CACHE_DIR, exist_ok=True)

# Conservative defaults to reduce rate-limit pain.
DEFAULT_TTL_SECONDS_STOCKS = 60
DEFAULT_TTL_SECONDS_FX = 12 * 60 * 60  # open FX endpoint updates daily


@dataclass
class Quote:
    symbol: str
    kind: str  # "stock" | "fx"
    price: float
    currency: Optional[str]
    asof_unix: int
    source: str
    extra: Dict[str, Any]


def _cache_path(key: str) -> str:
    safe = re.sub(r"[^A-Za-z0-9_.-]+", "_", key)
    return os.path.join(CACHE_DIR, f"{safe}.json")


def _cache_get(key: str, ttl: int) -> Optional[Dict[str, Any]]:
    path = _cache_path(key)
    if not os.path.exists(path):
        return None
    try:
        with open(path, "r", encoding="utf-8") as f:
            payload = json.load(f)
        if int(time.time()) - int(payload.get("_cached_at", 0)) <= ttl:
            return payload
    except Exception:
        return None
    return None


def _cache_set(key: str, payload: Dict[str, Any]) -> None:
    payload["_cached_at"] = int(time.time())
    with open(_cache_path(key), "w", encoding="utf-8") as f:
        json.dump(payload, f, ensure_ascii=False, indent=2)


def _parse_fx_pair(s: str) -> Optional[Tuple[str, str]]:
    s = s.strip().upper()
    s = s.replace(" ", "")
    s = s.replace("-", "/")
    if "/" in s:
        parts = s.split("/")
        if len(parts) == 2 and len(parts[0]) == 3 and len(parts[1]) == 3:
            return parts[0], parts[1]
        return None
    # e.g. EURUSD
    if len(s) == 6 and s.isalpha():
        return s[:3], s[3:]
    return None


def _fetch_fx(base: str, quote: str) -> Quote:
    cache_key = f"fx_{base}_{quote}"
    cached = _cache_get(cache_key, DEFAULT_TTL_SECONDS_FX)
    if cached:
        return Quote(**cached)

    url = f"https://open.er-api.com/v6/latest/{base}"
    r = requests.get(url, timeout=20)
    r.raise_for_status()
    data = r.json()

    if data.get("result") != "success":
        raise RuntimeError(f"FX provider error: {data}")

    rates = data.get("rates") or {}
    if quote not in rates:
        raise RuntimeError(f"FX pair not supported by provider: {base}/{quote}")

    q = Quote(
        symbol=f"{base}/{quote}",
        kind="fx",
        price=float(rates[quote]),
        currency=quote,
        asof_unix=int(data.get("time_last_update_unix") or time.time()),
        source="ExchangeRate-API Open Access (open.er-api.com)",
        extra={
            "base": base,
            "quote": quote,
            "time_last_update_utc": data.get("time_last_update_utc"),
            "time_next_update_utc": data.get("time_next_update_utc"),
            "attribution_required": True,
            "provider": data.get("provider"),
            "documentation": data.get("documentation"),
        },
    )

    _cache_set(cache_key, q.__dict__)
    return q


def _fetch_stock(symbol: str) -> Quote:
    cache_key = f"stk_{symbol}"
    cached = _cache_get(cache_key, DEFAULT_TTL_SECONDS_STOCKS)
    if cached:
        return Quote(**cached)

    # Import inside to keep startup light if user only does FX.
    import yfinance as yf

    t = yf.Ticker(symbol)
    # Fast path: try fast_info first (often less fragile than scraping-heavy calls)
    price = None
    currency = None
    extra: Dict[str, Any] = {}

    try:
        fi = getattr(t, "fast_info", None)
        if fi:
            price = fi.get("lastPrice") or fi.get("last_price")
            currency = fi.get("currency")
            extra["exchange"] = fi.get("exchange")
            extra["timezone"] = fi.get("timezone")
    except Exception:
        pass

    # Fallback: use 1d history
    if price is None:
        try:
            hist = t.history(period="1d", interval="1m")
            if hist is not None and len(hist) > 0:
                price = float(hist["Close"].iloc[-1])
        except Exception as e:
            raise RuntimeError(f"Yahoo/yfinance fetch failed for {symbol}: {e}")

    if price is None:
        raise RuntimeError(f"Could not resolve price for symbol: {symbol}")

    # Try to enrich
    try:
        info = t.get_info()  # can be heavier / more likely to rate-limit
        if isinstance(info, dict):
            currency = currency or info.get("currency")
            extra["shortName"] = info.get("shortName") or info.get("longName")
            extra["marketState"] = info.get("marketState")
    except Exception:
        # Non-fatal: still return price
        pass

    q = Quote(
        symbol=symbol,
        kind="stock",
        price=float(price),
        currency=currency,
        asof_unix=int(time.time()),
        source="Yahoo Finance via yfinance (unofficial; best-effort)",
        extra=extra,
    )

    _cache_set(cache_key, q.__dict__)
    return q


def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("symbol", help="Ticker (AAPL, ^GSPC) or FX pair (USD/ZAR, EURUSD)")
    args = ap.parse_args()

    s = args.symbol.strip()

    fx = _parse_fx_pair(s)
    if fx:
        base, quote = fx
        q = _fetch_fx(base, quote)
    else:
        q = _fetch_stock(s)

    # Print as JSON so the calling agent can format nicely.
    print(json.dumps(q.__dict__, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()