Create main.py

main.py

@@ -0,0 +1,110 @@
+"""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
+
+
+@app.get("/api/price")
+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"})
+
+
+@app.get("/health")
+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")