import json import logging import os from datetime import datetime import websockets from fastapi import FastAPI, WebSocket, WebSocketDisconnect from fastapi.responses import FileResponse, HTMLResponse from fastapi.staticfiles import StaticFiles logging.basicConfig(level=logging.INFO) logger = logging.getLogger("btcprice") BTCDATA_URL = os.environ.get("BTCDATA_URL", "http://localhost:4100") 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...

""" @app.get("/", response_class=HTMLResponse) async def home(): return HOME_HTML # --- Health endpoint --- @app.get("/api/health") async def health(): return {"status": "up"} # --- 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 = "" html = ( '

' '

' f'

${da_fmt(",.2f", p)}

' ) 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)