k
103 lines
3.4 KiB
Python
103 lines
3.4 KiB
Python
import asyncio
|
|
import json
|
|
import logging
|
|
|
|
import httpx
|
|
import websockets
|
|
|
|
from app.config import settings
|
|
from app.services.price_store import price_store
|
|
from app.ws.manager import manager
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
ASSET_MAP = {
|
|
"btcusdt": "BTC",
|
|
"ethusdt": "ETH",
|
|
}
|
|
|
|
|
|
async def _process_message(raw: str):
|
|
try:
|
|
data = json.loads(raw)
|
|
# Combined stream wraps payload under "data"
|
|
payload = data.get("data", data)
|
|
if payload.get("e") != "kline":
|
|
return
|
|
|
|
kline = payload["k"]
|
|
symbol = kline["s"].lower()
|
|
asset = ASSET_MAP.get(symbol)
|
|
if asset is None:
|
|
return
|
|
|
|
candle = {
|
|
"time": kline["t"], # open time ms
|
|
"open": float(kline["o"]),
|
|
"high": float(kline["h"]),
|
|
"low": float(kline["l"]),
|
|
"close": float(kline["c"]),
|
|
"volume": float(kline["v"]),
|
|
}
|
|
price_store.update(asset, candle)
|
|
|
|
# Broadcast live price tick
|
|
await manager.broadcast({
|
|
"type": "price",
|
|
"asset": asset,
|
|
"price": candle["close"],
|
|
"time": candle["time"],
|
|
})
|
|
except Exception as exc:
|
|
logger.warning("Error processing Binance message: %s", exc)
|
|
|
|
|
|
async def fetch_historical(asset: str, symbol: str, interval: str = "1m", limit: int = 500):
|
|
"""Fetch historical klines from Binance REST API to pre-fill price_store."""
|
|
url = f"{settings.binance_rest_url}/api/v3/klines?symbol={symbol.upper()}&interval={interval}&limit={limit}"
|
|
try:
|
|
async with httpx.AsyncClient(timeout=15) as client:
|
|
resp = await client.get(url)
|
|
resp.raise_for_status()
|
|
for row in resp.json():
|
|
candle = {
|
|
"time": row[0], # open time ms
|
|
"open": float(row[1]),
|
|
"high": float(row[2]),
|
|
"low": float(row[3]),
|
|
"close": float(row[4]),
|
|
"volume": float(row[5]),
|
|
}
|
|
price_store.update(asset, candle)
|
|
logger.info("Loaded %d historical %s candles for %s", limit, interval, asset)
|
|
except Exception as exc:
|
|
logger.error("Failed to fetch historical data for %s: %s", asset, exc)
|
|
|
|
|
|
async def run_binance_ws():
|
|
"""Connect to Binance kline stream with exponential back-off on disconnect."""
|
|
# Pre-fill with historical data
|
|
await fetch_historical("BTC", "btcusdt", limit=500)
|
|
await fetch_historical("ETH", "ethusdt", limit=500)
|
|
|
|
backoff = 1
|
|
while True:
|
|
try:
|
|
logger.info("Connecting to Binance WebSocket: %s", settings.binance_ws_url)
|
|
async with websockets.connect(
|
|
settings.binance_ws_url,
|
|
ping_interval=20,
|
|
ping_timeout=10,
|
|
) as ws:
|
|
backoff = 1 # reset on successful connection
|
|
logger.info("Binance WebSocket connected.")
|
|
async for message in ws:
|
|
await _process_message(message)
|
|
except asyncio.CancelledError:
|
|
logger.info("Binance WebSocket task cancelled.")
|
|
return
|
|
except Exception as exc:
|
|
logger.error("Binance WebSocket error: %s. Reconnecting in %ds.", exc, backoff)
|
|
await asyncio.sleep(backoff)
|
|
backoff = min(backoff * 2, 60)
|