import asyncio import json import logging import os import time 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 (30m)

""" @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-half-hour") async def kraken_half_hour(): return await _proxy_btcdata_json("/api/kraken-half-hour") @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 --- # htmx's WebSocket extension reconnects a browser socket only after an abnormal # close (1006/1011/1012/1013). A normal 1000 close means "the server is done", # so the extension never reconnects and the panel stays frozen until the visitor # reloads. A btcdata restart must therefore never reach the browser as a normal # close: the proxy reconnects upstream behind the still-open browser socket, and # only if that keeps failing does it close with 1012 (Service Restart) and hand # recovery to the page's own reconnect. BROWSER_RETRY_CLOSE_CODE = 1012 UPSTREAM_OPEN_TIMEOUT_SECONDS = 5.0 UPSTREAM_RETRY_DELAYS_SECONDS = (0.5, 1.0, 2.0, 4.0, 8.0, 8.0) UPSTREAM_STABLE_SECONDS = 30.0 def _btcdata_ws_url(path: str) -> str: return BTCDATA_URL.replace("http", "ws", 1) + path async def _await_browser_close(browser_ws: WebSocket) -> None: """Return as soon as the browser side goes away. The browser sends nothing, but reading is the only way to notice that it left while the proxy is waiting on a dead upstream rather than sending. """ try: while True: if (await browser_ws.receive())["type"] == "websocket.disconnect": return except (WebSocketDisconnect, RuntimeError): return async def _forward_upstream(browser_ws: WebSocket, upstream_url: str, transform) -> None: """Forward one upstream connection to the browser until that connection ends.""" async with websockets.connect( upstream_url, ping_interval=None, open_timeout=UPSTREAM_OPEN_TIMEOUT_SECONDS ) as upstream: unrenderable = 0 async for message in upstream: if not message: continue try: html = transform(message) except Exception: # One unexpected payload must not take down a working page. Log # the first per connection, then stay quiet about the rest. unrenderable += 1 logger.log( logging.WARNING if unrenderable == 1 else logging.DEBUG, "Discarding an upstream message that could not be rendered", exc_info=True, ) continue await browser_ws.send_text(html) async def _ws_proxy(browser_ws: WebSocket, upstream_path: str, transform) -> None: await browser_ws.accept() upstream_url = _btcdata_ws_url(upstream_path) logger.info("Browser connected, proxying to %s", upstream_url) browser_closed = asyncio.create_task(_await_browser_close(browser_ws)) attempt = 0 try: while True: started = time.monotonic() forwarding = asyncio.create_task( _forward_upstream(browser_ws, upstream_url, transform) ) done, _ = await asyncio.wait( {forwarding, browser_closed}, return_when=asyncio.FIRST_COMPLETED ) if browser_closed in done: # The browser left; a healthy upstream would otherwise keep this # task forwarding into a socket nobody is reading. forwarding.cancel() await asyncio.gather(forwarding, return_exceptions=True) logger.info("Browser disconnected from %s", upstream_url) return try: forwarding.result() logger.info("Upstream %s closed; reconnecting", upstream_url) except Exception: logger.info("Upstream %s unavailable; reconnecting", upstream_url) logger.debug("Upstream connection error", exc_info=True) if time.monotonic() - started >= UPSTREAM_STABLE_SECONDS: # A connection that worked for a while means a restart, not a # persistently broken upstream: start the schedule over. attempt = 0 if attempt >= len(UPSTREAM_RETRY_DELAYS_SECONDS): logger.warning( "Upstream %s still unavailable; asking the browser to reconnect", upstream_url, ) return await asyncio.wait({browser_closed}, timeout=UPSTREAM_RETRY_DELAYS_SECONDS[attempt]) attempt += 1 if browser_closed.done(): logger.info("Browser disconnected from %s", upstream_url) return finally: browser_closed.cancel() await asyncio.gather(browser_closed, return_exceptions=True) try: # Every server-side exit, expected or not, closes with a code the # page retries; a browser that already left ignores it. await browser_ws.close(code=BROWSER_RETRY_CLOSE_CODE) 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)