"""Thin wrapper around the official Binance Spot connector.""" from __future__ import annotations from collections.abc import Callable from typing import Any from requests.exceptions import RequestException, SSLError # type: ignore[import-untyped] class SpotBinanceClient: def __init__( self, *, api_key: str, api_secret: str, base_url: str, recv_window: int, client: Any | None = None, ) -> None: self.recv_window = recv_window if client is not None: self._client = client return try: from binance.spot import Spot except ModuleNotFoundError as exc: # pragma: no cover raise RuntimeError("binance-connector is not installed") from exc self._client = Spot(api_key=api_key, api_secret=api_secret, base_url=base_url) def _call(self, operation: str, func: Callable[..., Any], *args: Any, **kwargs: Any) -> Any: try: return func(*args, **kwargs) except SSLError as exc: raise RuntimeError( "Binance Spot request failed because TLS certificate verification failed. " "This usually means the local Python trust store is incomplete or a proxy is intercepting HTTPS. " "Update the local CA trust chain or configure the host environment with the correct corporate/root CA." ) from exc except RequestException as exc: raise RuntimeError(f"Binance Spot request failed during {operation}: {exc}") from exc def account_info(self) -> dict[str, Any]: return self._call("account info", self._client.account, recvWindow=self.recv_window) # type: ignore[no-any-return] def exchange_info(self, symbol: str | None = None) -> dict[str, Any]: kwargs: dict[str, Any] = {} if symbol: kwargs["symbol"] = symbol return self._call("exchange info", self._client.exchange_info, **kwargs) # type: ignore[no-any-return] def ticker_24h(self, symbols: list[str] | None = None) -> list[dict[str, Any]]: if not symbols: response = self._call("24h ticker", self._client.ticker_24hr) elif len(symbols) == 1: response = self._call("24h ticker", self._client.ticker_24hr, symbol=symbols[0]) else: response = self._call("24h ticker", self._client.ticker_24hr, symbols=symbols) return response if isinstance(response, list) else [response] # type: ignore[no-any-return] def ticker_price(self, symbols: list[str] | None = None) -> list[dict[str, Any]]: if not symbols: response = self._call("ticker price", self._client.ticker_price) elif len(symbols) == 1: response = self._call("ticker price", self._client.ticker_price, symbol=symbols[0]) else: response = self._call("ticker price", self._client.ticker_price, symbols=symbols) return response if isinstance(response, list) else [response] # type: ignore[no-any-return] def klines(self, symbol: str, interval: str, limit: int) -> list[list[Any]]: return self._call("klines", self._client.klines, symbol=symbol, interval=interval, limit=limit) # type: ignore[no-any-return] def new_order(self, **kwargs: Any) -> dict[str, Any]: kwargs.setdefault("recvWindow", self.recv_window) return self._call("new order", self._client.new_order, **kwargs) # type: ignore[no-any-return]