import json import logging import os from datetime import datetime import httpx import websockets from fastapi import FastAPI, WebSocket, WebSocketDisconnect from fastapi.responses import FileResponse, HTMLResponse, JSONResponse from fastapi.staticfiles import StaticFiles logging.basicConfig(level=logging.INFO) logger = logging.getLogger("btcprice") BTCDATA_URL = os.environ.get("BTCDATA_URL", "http://localhost:4100").rstrip("/") APP_VERSION = os.environ.get("APP_VERSION", "development") READINESS_TIMEOUT_SECONDS = 3.0 app = FastAPI() app.mount("/css", StaticFiles(directory="resources/public/css"), name="css") @app.get("/favicon.ico", include_in_schema=False) async def favicon(): return FileResponse("resources/public/favicon.svg", media_type="image/svg+xml") # --- HTML page --- HOME_HTML = """\ BTC Price

BTC Price

Connecting...

Loading Strike rates...

1h Candles (1m)

24h Candles (1h)

""" @app.get("/", response_class=HTMLResponse) async def home(): return HOME_HTML # --- Health endpoint --- @app.get("/api/health") async def health(): return {"status": "up"} @app.get("/api/ready") async def ready(): """Report readiness only while the btcdata dependency is healthy.""" url = f"{BTCDATA_URL}/api/health" try: async with httpx.AsyncClient() as client: response = await client.get(url, timeout=READINESS_TIMEOUT_SECONDS) response.raise_for_status() except httpx.HTTPError: logger.warning("btcdata readiness check failed for %s", url) return JSONResponse(content={"status": "not ready"}, status_code=503) return {"status": "ready"} @app.get("/api/version") async def version(): return {"version": APP_VERSION} async def _proxy_btcdata_json(path: str) -> JSONResponse: url = f"{BTCDATA_URL}{path}" try: async with httpx.AsyncClient() as client: resp = await client.get(url, timeout=10) return JSONResponse(content=resp.json(), status_code=resp.status_code) except httpx.RequestError as exc: logger.warning("Unable to connect to btcdata at %s: %s", url, exc) return JSONResponse( content={ "error": "btcdata unavailable", "detail": f"Could not connect to {BTCDATA_URL}. Start btcdata first or set BTCDATA_URL.", }, status_code=503, ) except json.JSONDecodeError: logger.warning("btcdata returned non-JSON response from %s", url) return JSONResponse( content={"error": "invalid btcdata response"}, status_code=502, ) @app.get("/api/kraken-hour") async def kraken_hour(): return await _proxy_btcdata_json("/api/kraken-hour") @app.get("/api/kraken-minute") async def kraken_minute(): return await _proxy_btcdata_json("/api/kraken-minute") # --- Formatting helpers (Danish locale) --- def da_fmt(fmt: str, val: float) -> str: """Format a number using Danish locale conventions.""" if fmt == ",.2f": # e.g. 12.345,67 int_part = int(val) frac = abs(val - int_part) int_str = f"{int_part:,}".replace(",", ".") return f"{int_str},{frac:.2f}"[:-1].split(",")[0] + "," + f"{frac:.2f}"[2:] if fmt == ",.0f": return f"{int(round(val)):,}".replace(",", ".") if fmt == ".4f": return f"{val:.4f}".replace(".", ",") if fmt == ",d": return f"{int(val):,}".replace(",", ".") return str(val) # --- Price transform --- def price_json_to_html(text: str) -> str: data = json.loads(text) p = float(data["price"]) pp = float(data["prev_price"]) if data.get("prev_price") is not None else None if pp is None: color = "text-gray-900" elif p > pp: color = "text-green-600" elif p < pp: color = "text-red-600" else: color = "text-gray-900" recorded_at = data.get("recorded_at", "") if recorded_at: dt = datetime.fromisoformat(recorded_at.replace("Z", "+00:00")) local_dt = dt.astimezone() ts = local_dt.strftime("%d-%m %H:%M:%S") else: ts = "" formatted_price = f'${da_fmt(",.2f", p)}' html = ( f'

' '

' f'

{formatted_price}

' ) if data.get("price_eur") is not None: p_eur = float(data["price_eur"]) p_dkk = float(data["price_dkk"]) eur_rate = p_eur / p dkk_rate = p_dkk / p html += ( '

' f'\u20ac{da_fmt(",.0f", p_eur)}' f'{da_fmt(",.0f", p_dkk)} kr' "

" '

' f'EUR/USD {da_fmt(".4f", eur_rate)}' f'DKK/USD {da_fmt(".4f", dkk_rate)}' "

" ) html += f'

Updated {ts}

' html += "

" return html # --- Strike transform --- STRIKE_PAIRS = [ ("BTC/USDT", "$"), ("BTC/EUR", "\u20ac"), ("USDT/EUR", ""), ("EUR/BTC", ""), ] def format_strike_amount(prefix: str, amount) -> str: if amount is None: return "\u2014" v = float(amount) if prefix in ("$", "\u20ac"): return f'{prefix}{da_fmt(",.2f", v)}' return str(amount) def strike_json_to_html(text: str) -> str: rates = json.loads(text) sats = rates.get("quote-sats") html = ( '

' '

' ) if sats is not None: html += ( '

' f'{da_fmt(",d", sats)} sats

' ) html += '

' for pair, prefix in STRIKE_PAIRS: amount = rates.get(pair) html += ( '

' f'{pair}' f'{format_strike_amount(prefix, amount)}' "

" ) html += "

" return html # --- WebSocket proxy --- def _btcdata_ws_url(path: str) -> str: return BTCDATA_URL.replace("http", "ws", 1) + path async def _ws_proxy(browser_ws: WebSocket, upstream_path: str, transform): await browser_ws.accept() upstream_url = _btcdata_ws_url(upstream_path) logger.info("Browser connected, proxying to %s", upstream_url) try: async with websockets.connect(upstream_url, ping_interval=None) as upstream: async for message in upstream: if not message: continue try: html = transform(message) await browser_ws.send_text(html) except WebSocketDisconnect: break except Exception: logger.debug("Error forwarding to browser", exc_info=True) break except WebSocketDisconnect: pass except Exception: logger.debug("Upstream connection error", exc_info=True) finally: try: await browser_ws.close() except Exception: pass @app.websocket("/ws/price") async def ws_price(websocket: WebSocket): await _ws_proxy(websocket, "/api/price/ws", price_json_to_html) @app.websocket("/ws/strike") async def ws_strike(websocket: WebSocket): await _ws_proxy(websocket, "/api/strike/ws", strike_json_to_html)