Spaces:
Sleeping
Sleeping
| """XAUUSD — live gold spot dashboard. FastAPI + Docker Space.""" | |
| from __future__ import annotations | |
| import asyncio | |
| import time | |
| from typing import Any, Optional | |
| import httpx | |
| from fastapi import FastAPI | |
| from fastapi.responses import JSONResponse | |
| from fastapi.staticfiles import StaticFiles | |
| CACHE_TTL = 30.0 # upstream asks clients to cache >= 30s | |
| REQUEST_TIMEOUT = 10.0 | |
| MAX_HISTORY = 480 # ~4h of 30s samples (in-memory, resets on restart) | |
| UPSTREAM = [ | |
| "https://xaus.com/api/v1/spot", | |
| "https://api.gold-api.com/price/XAU", | |
| ] | |
| app = FastAPI(title="XAUUSD", docs_url=None, redoc_url=None) | |
| _cache: dict[str, Any] = {"ts": 0.0, "payload": None} | |
| _history: list[dict[str, Any]] = [] | |
| _lock = asyncio.Lock() | |
| def _num(value: Any) -> Optional[float]: | |
| try: | |
| return float(value) | |
| except (TypeError, ValueError): | |
| return None | |
| def _parse(url: str, data: dict[str, Any]) -> Optional[dict[str, Any]]: | |
| """Normalise the different upstream shapes into one payload.""" | |
| price = _num(data.get("spot_usd_oz")) if "spot_usd_oz" in data else _num(data.get("price")) | |
| if price is None: | |
| return None | |
| per_gram = _num(data.get("per_gram_usd")) or price / 31.1034768 | |
| return { | |
| "price": price, | |
| "per_gram": per_gram, | |
| "per_kg": _num(data.get("per_kg_usd")) or per_gram * 1000.0, | |
| "silver": _num(data.get("silver_usd_oz")), | |
| "gold_silver_ratio": _num(data.get("gold_silver_ratio")), | |
| "xaut_ratio": _num(data.get("xaut_gold_ratio")), | |
| "paxg_ratio": _num(data.get("paxg_gold_ratio")), | |
| "btc_usd": _num(data.get("btc_usd")), | |
| "btc_gold_oz": _num(data.get("btc_gold_oz")), | |
| "source": str(data.get("source") or url.split("/")[2]), | |
| "upstream_updated_at": data.get("updated_at") or data.get("updatedAt"), | |
| } | |
| async def _fetch_upstream() -> Optional[dict[str, Any]]: | |
| async with httpx.AsyncClient(timeout=REQUEST_TIMEOUT, follow_redirects=True) as client: | |
| for url in UPSTREAM: | |
| try: | |
| r = await client.get(url, headers={"Accept": "application/json"}) | |
| r.raise_for_status() | |
| payload = _parse(url, r.json()) | |
| except Exception: | |
| continue | |
| if payload: | |
| return payload | |
| return None | |
| async def _get_price() -> tuple[Optional[dict[str, Any]], Optional[str]]: | |
| async with _lock: | |
| now = time.time() | |
| if _cache["payload"] and now - _cache["ts"] < CACHE_TTL: | |
| return _cache["payload"], None | |
| payload = await _fetch_upstream() | |
| if payload is None: | |
| if _cache["payload"]: # serve stale rather than nothing | |
| return _cache["payload"], "upstream unavailable — showing last known price" | |
| return None, "upstream unavailable" | |
| payload["fetched_at"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(now)) | |
| _cache.update(ts=now, payload=payload) | |
| _history.append({"t": now, "p": payload["price"]}) | |
| del _history[:-MAX_HISTORY] | |
| return payload, None | |
| async def api_price() -> JSONResponse: | |
| payload, warning = await _get_price() | |
| if payload is None: | |
| return JSONResponse({"ok": False, "error": warning}, status_code=503) | |
| body = dict(payload) | |
| body["ok"] = True | |
| body["warning"] = warning | |
| body["history"] = [round(point["p"], 2) for point in _history] | |
| return JSONResponse(body, headers={"Cache-Control": "no-store"}) | |
| async def health() -> dict[str, Any]: | |
| return {"status": "ok", "cached": bool(_cache["payload"])} | |
| # mount last so it doesn't shadow the API routes | |
| app.mount("/", StaticFiles(directory="static", html=True), name="static") |