Binance_Altcoin_Monitor/backend/app/binance.py

import asyncio
import logging
import time
from typing import Any

import httpx

from .config import settings
from .http_client import httpx_client_kwargs

logger = logging.getLogger(__name__)

# 418 = IP 被币安临时封禁（请求过快）；429 = 触发频率限制
_RATE_LIMIT_CODES = {418, 429}


class BinanceFuturesClient:
    def __init__(self) -> None:
        self.base = settings.binance_fapi_base.rstrip("/")
        self._symbols_cache: list[str] | None = None
        self._client: httpx.AsyncClient | None = None
        self._throttle_lock = asyncio.Lock()
        self._last_request_at: float = 0.0
        self._ban_until: float = 0.0

    async def _ensure_client(self) -> httpx.AsyncClient:
        if self._client is None or self._client.is_closed:
            limits = httpx.Limits(
                max_connections=max(2, settings.max_concurrency),
                max_keepalive_connections=max(2, settings.max_concurrency),
            )
            self._client = httpx.AsyncClient(
                timeout=30.0,
                limits=limits,
                **httpx_client_kwargs("binance"),
            )
        return self._client

    async def close(self) -> None:
        if self._client and not self._client.is_closed:
            await self._client.aclose()
            self._client = None

    async def _throttle(self) -> None:
        async with self._throttle_lock:
            now = time.monotonic()
            if now < self._ban_until:
                wait = self._ban_until - now
                logger.warning("Binance IP cooldown, sleeping %.0fs", wait)
                await asyncio.sleep(wait)
            gap = settings.request_interval_sec - (now - self._last_request_at)
            if gap > 0:
                await asyncio.sleep(gap)
            self._last_request_at = time.monotonic()

    async def _get(self, path: str, params: dict | None = None) -> Any:
        url = f"{self.base}{path}"
        last_err: Exception | None = None

        for attempt in range(1, settings.max_retries + 1):
            await self._throttle()
            try:
                client = await self._ensure_client()
                resp = await client.get(url, params=params or {})
                if resp.status_code in _RATE_LIMIT_CODES:
                    retry_after = int(
                        resp.headers.get("Retry-After", settings.ban_cooldown_sec)
                    )
                    retry_after = max(retry_after, settings.ban_cooldown_sec)
                    self._ban_until = time.monotonic() + retry_after
                    logger.warning(
                        "Binance HTTP %s on %s, cooldown %ss (attempt %d/%d)",
                        resp.status_code,
                        path,
                        retry_after,
                        attempt,
                        settings.max_retries,
                    )
                    last_err = httpx.HTTPStatusError(
                        f"{resp.status_code}",
                        request=resp.request,
                        response=resp,
                    )
                    await asyncio.sleep(retry_after)
                    continue
                resp.raise_for_status()
                return resp.json()
            except httpx.HTTPStatusError as e:
                last_err = e
                if e.response.status_code in _RATE_LIMIT_CODES:
                    continue
                raise
            except (httpx.ConnectError, httpx.ReadTimeout) as e:
                last_err = e
                logger.warning("Binance request error %s (attempt %d)", path, attempt)
                await asyncio.sleep(min(2 * attempt, 10))

        raise last_err or RuntimeError(f"Binance request failed: {path}")

    async def get_usdt_perpetual_symbols(self) -> list[str]:
        if self._symbols_cache:
            return self._symbols_cache
        info = await self._get("/fapi/v1/exchangeInfo")
        symbols = []
        for s in info.get("symbols", []):
            if (
                s.get("contractType") == "PERPETUAL"
                and s.get("quoteAsset") == "USDT"
                and s.get("status") == "TRADING"
            ):
                symbols.append(s["symbol"])
        self._symbols_cache = sorted(symbols)
        logger.info("Loaded %d USDT perpetual symbols", len(self._symbols_cache))
        return self._symbols_cache

    def clear_symbol_cache(self) -> None:
        self._symbols_cache = None

    async def get_24hr_tickers(self) -> list[dict]:
        """单次请求获取全市场 24h 行情（用于缩小 K 线拉取范围）。"""
        data = await self._get("/fapi/v1/ticker/24hr")
        return data if isinstance(data, list) else []

    async def get_klines(
        self,
        symbol: str,
        start_ms: int,
        end_ms: int,
        interval: str = "1h",
    ) -> list[list]:
        all_klines: list[list] = []
        cursor = start_ms
        while cursor < end_ms:
            batch = await self._get(
                "/fapi/v1/klines",
                {
                    "symbol": symbol,
                    "interval": interval,
                    "startTime": cursor,
                    "endTime": end_ms,
                    "limit": 1500,
                },
            )
            if not batch:
                break
            all_klines.extend(batch)
            last_open = int(batch[-1][0])
            next_cursor = last_open + 3600_000
            if next_cursor <= cursor:
                break
            cursor = next_cursor
            if len(batch) < 1500:
                break
        return all_klines

    async def get_price(self, symbol: str) -> float:
        data = await self._get("/fapi/v1/ticker/price", {"symbol": symbol})
        return float(data["price"])

    async def get_prices_batch(self, symbols: list[str]) -> dict[str, float]:
        tickers = await self._get("/fapi/v1/ticker/price")
        sym_set = set(symbols)
        return {t["symbol"]: float(t["price"]) for t in tickers if t["symbol"] in sym_set}


binance_client = BinanceFuturesClient()