trumpsignal-backend/app/services/binance.py

import asyncio
import json
import logging
from datetime import datetime, timezone
from typing import Optional

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

# Liveness signal for /api/health/deep. Updated on every processed kline tick.
# A stale value means the price feed is down → TP/SL/trailing stops are NOT
# firing for live trades, which is a real-money incident, not a soft warning.
last_tick_at: Optional[datetime] = None

# symbol (lowercase Binance ticker) → asset (HL perp name, uppercase).
#
# WHY THIS MATTERS: tp_sl_monitor.on_price_tick(asset, price) is the sole
# mechanism that fires stop-loss, take-profit, and trailing-stop for live
# trades. If an asset is missing here, every trade on that asset silently
# loses all TP/SL protection — only the asyncio max-hold timer remains.
#
# Trump's AI picks target_asset from the full HL perp universe.  The assets
# below cover the realistic set (BTC/ETH always, SOL/TRUMP/BNB/DOGE/LINK/AAVE
# for direct-named or thematic posts).  HL-native assets (HYPE, PURR) are NOT
# on Binance; those require a separate HL price feed (see BUG-08 note in CLAUDE.md).
# Until that feed is added, HYPE trades fall back to max-hold only.
ASSET_MAP: dict[str, str] = {
    "btcusdt":  "BTC",
    "ethusdt":  "ETH",
    "solusdt":  "SOL",
    "trumpusdt": "TRUMP",
    "bnbusdt":  "BNB",
    "dogeusdt": "DOGE",
    "linkusdt": "LINK",
    "aaveusdt": "AAVE",
}

# Build the combined-stream WS URL from ASSET_MAP so the two are always in sync.
# Format: wss://...?streams=sym1@kline_1m/sym2@kline_1m/...
_BINANCE_WS_BASE = "wss://data-stream.binance.vision/stream"
BINANCE_WS_URL = (
    _BINANCE_WS_BASE
    + "?streams="
    + "/".join(f"{sym}@kline_1m" for sym in ASSET_MAP)
)


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)

        global last_tick_at
        last_tick_at = datetime.now(timezone.utc)

        # 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.

    Binance routinely cycles connections every ~24h, and intermediate proxies
    can drop without sending a close frame — both surface here as exceptions.
    The reconnect must be fast (sub-second) so price ticks aren't visibly
    stuck on the dashboard during a hiccup.

    Resilience defenses:
      * ping_interval/ping_timeout — peer-of-record liveness check (websockets lib)
      * close_timeout — bound how long we wait for a clean close handshake
      * open_timeout — bound the initial connect handshake so a hung TCP doesn't
        permanently park this task
      * max_queue — drop frames if downstream can't keep up rather than back
        the WS receive buffer up indefinitely
    """
    # Pre-fill historical candles for every tracked asset so price_store has
    # a warm baseline before the first live tick arrives.
    await asyncio.gather(*(
        fetch_historical(asset, symbol, limit=500)
        for symbol, asset in ASSET_MAP.items()
    ), return_exceptions=True)

    backoff = 1
    while True:
        try:
            logger.info("Connecting to Binance WebSocket (%d assets): %s",
                        len(ASSET_MAP), BINANCE_WS_URL)
            async with websockets.connect(
                BINANCE_WS_URL,
                ping_interval=20,
                ping_timeout=10,
                close_timeout=5,
                open_timeout=10,
                max_queue=128,
            ) as ws:
                backoff = 1  # reset on successful connection
                logger.info("Binance WebSocket connected.")
                async for message in ws:
                    await _process_message(message)
            # If we exit the `async with` without an exception, the server
            # closed normally — reconnect immediately rather than backing off.
            logger.info("Binance WebSocket closed cleanly, reconnecting immediately.")
        except asyncio.CancelledError:
            logger.info("Binance WebSocket task cancelled.")
            return
        except Exception as exc:
            # Use exception() once so a TRACE-back-with-context is logged the
            # first time something interesting happens; subsequent same-error
            # retries log compactly to avoid drowning the log.
            logger.warning(
                "Binance WebSocket error: %s (%s). Reconnecting in %ds.",
                type(exc).__name__, exc, backoff,
            )
            await asyncio.sleep(backoff)
            backoff = min(backoff * 2, 60)