Initial commit with translated description

scripts/market_quote.py

@@ -0,0 +1,207 @@
+#!/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()