k
4ffcb442fe
- New migrations for daily_budget, active_window, and bottrade snapshot - Add trumpstruth scraper and price_impact_monitor service - Expand bot_engine, hyperliquid, recovery, and tp_sl_monitor logic - Update API/schemas/models for new features Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
111 lines
3.8 KiB
Python
111 lines
3.8 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)
|
|
|
|
# TP/SL check on every tick (cheap no-op when no trades are being watched)
|
|
from app.services.tp_sl_monitor import on_price_tick
|
|
on_price_tick(asset, candle["close"])
|
|
|
|
# Price-impact peak tracker (updates rolling max for open post windows)
|
|
from app.services.price_impact_monitor import on_price_tick as pi_tick
|
|
pi_tick(asset, candle["high"], candle["low"], candle["close"])
|
|
|
|
# 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)
|