from __future__ import annotations import hashlib import hmac import time from decimal import Decimal from collections.abc import Callable from typing import Any from urllib.parse import urlencode import httpx from .audit_log import AuditLogger from .models import AccountSnapshot, Balance, OrderSide, SymbolMetadata TESTNET_REST_BASE_URL = "https://testnet.binance.vision/api" def sign_query(query_string: str, api_secret: str) -> str: return hmac.new(api_secret.encode("utf-8"), query_string.encode("utf-8"), hashlib.sha256).hexdigest() class BinanceSpotClient: def __init__( self, api_key: str, api_secret: str, base_url: str = TESTNET_REST_BASE_URL, base_urls: tuple[str, ...] | None = None, http_client: httpx.Client | None = None, recv_window: int = 5000, proxy_url: str | None = None, audit_logger: AuditLogger | None = None, endpoint_probe_timeout: float = 5, endpoint_probe_clock: Callable[[], float] | None = None, ) -> None: self.api_key = api_key self.api_secret = api_secret self.base_urls = tuple(item.rstrip("/") for item in (base_urls or (base_url,))) self.recv_window = recv_window self._client = http_client or httpx.Client(proxy=proxy_url, timeout=10, trust_env=False) self.audit_logger = audit_logger or AuditLogger() self._clock = endpoint_probe_clock or time.time self._server_time_offset_ms = 0 self.base_url = self._select_base_url(endpoint_probe_timeout) def account(self, timestamp_ms: int | None = None) -> AccountSnapshot: payload = self._signed_request("GET", "/v3/account", {}, timestamp_ms) return self.parse_account(payload) def exchange_info(self, symbol: str) -> SymbolMetadata: payload = self._public_request("GET", "/v3/exchangeInfo", {"symbol": symbol.upper()}) symbol_info = payload["symbols"][0] filters = {item["filterType"]: item for item in symbol_info["filters"]} lot_size = filters.get("LOT_SIZE", {}) price_filter = filters.get("PRICE_FILTER", {}) min_notional_filter = filters.get("MIN_NOTIONAL") or filters.get("NOTIONAL") or {} min_notional = Decimal(min_notional_filter.get("minNotional", "0")) return SymbolMetadata( symbol=symbol_info["symbol"], base_asset=symbol_info["baseAsset"], quote_asset=symbol_info["quoteAsset"], min_notional=min_notional, min_qty=Decimal(lot_size.get("minQty", "0")), step_size=Decimal(lot_size.get("stepSize", "0")), tick_size=Decimal(price_filter.get("tickSize", "0")), min_price=Decimal(price_filter.get("minPrice", "0")), ) def symbol_order_book_ticker(self, symbol: str) -> dict: return self._public_request("GET", "/v3/ticker/bookTicker", {"symbol": symbol.upper()}) def klines(self, symbol: str, interval: str = "1h", limit: int = 6) -> list: return self._public_request( "GET", "/v3/klines", {"symbol": symbol.upper(), "interval": interval, "limit": str(limit)}, ) def depth(self, symbol: str, limit: int = 20) -> dict: return self._public_request("GET", "/v3/depth", {"symbol": symbol.upper(), "limit": str(limit)}) def aggregate_trades(self, symbol: str, limit: int = 500) -> list: return self._public_request("GET", "/v3/aggTrades", {"symbol": symbol.upper(), "limit": str(limit)}) def ticker_24hr(self, symbol: str) -> dict: return self._public_request("GET", "/v3/ticker/24hr", {"symbol": symbol.upper()}) def avg_price(self, symbol: str) -> dict: return self._public_request("GET", "/v3/avgPrice", {"symbol": symbol.upper()}) def open_orders(self, symbol: str, timestamp_ms: int | None = None) -> list[dict]: payload = self._signed_request("GET", "/v3/openOrders", {"symbol": symbol.upper()}, timestamp_ms) if not isinstance(payload, list): raise ValueError("open orders response must be a list") return payload def my_trades( self, symbol: str, start_time_ms: int | None = None, from_id: int | None = None, limit: int = 1000, timestamp_ms: int | None = None, ) -> list[dict]: params = { "symbol": symbol.upper(), "limit": str(limit), } if start_time_ms is not None: params["startTime"] = str(start_time_ms) if from_id is not None: params["fromId"] = str(from_id) payload = self._signed_request("GET", "/v3/myTrades", params, timestamp_ms) if not isinstance(payload, list): raise ValueError("my trades response must be a list") return payload def cancel_order(self, symbol: str, orig_client_order_id: str, timestamp_ms: int | None = None) -> dict: params = { "symbol": symbol.upper(), "origClientOrderId": orig_client_order_id, } return self._signed_request("DELETE", "/v3/order", params, timestamp_ms) def test_market_buy_order( self, symbol: str, quote_amount: Decimal, client_order_id: str, timestamp_ms: int | None = None, ) -> dict: params = { "symbol": symbol.upper(), "side": OrderSide.BUY.value, "type": "MARKET", "quoteOrderQty": format(quote_amount, "f"), "newClientOrderId": client_order_id, } return self._signed_request("POST", "/v3/order/test", params, timestamp_ms) def create_market_sell_quantity( self, symbol: str, quantity: Decimal, client_order_id: str, timestamp_ms: int | None = None, ) -> dict: params = { "symbol": symbol.upper(), "side": OrderSide.SELL.value, "type": "MARKET", "quantity": format(quantity, "f"), "newClientOrderId": client_order_id, } return self._signed_request("POST", "/v3/order", params, timestamp_ms) def create_stop_loss_limit_sell( self, symbol: str, quantity: Decimal, stop_price: Decimal, limit_price: Decimal, client_order_id: str, timestamp_ms: int | None = None, ) -> dict: params = { "symbol": symbol.upper(), "side": OrderSide.SELL.value, "type": "STOP_LOSS_LIMIT", "timeInForce": "GTC", "quantity": format(quantity, "f"), "stopPrice": format(stop_price, "f"), "price": format(limit_price, "f"), "newOrderRespType": "RESULT", "newClientOrderId": client_order_id, } return self._signed_request("POST", "/v3/order", params, timestamp_ms) def create_oco_sell( self, symbol: str, quantity: Decimal, take_profit_price: Decimal, stop_price: Decimal, stop_limit_price: Decimal, list_client_order_id: str, take_profit_client_order_id: str, stop_client_order_id: str, timestamp_ms: int | None = None, ) -> dict: params = { "symbol": symbol.upper(), "side": OrderSide.SELL.value, "quantity": format(quantity, "f"), "aboveType": "LIMIT_MAKER", "abovePrice": format(take_profit_price, "f"), "aboveClientOrderId": take_profit_client_order_id, "belowType": "STOP_LOSS_LIMIT", "belowStopPrice": format(stop_price, "f"), "belowPrice": format(stop_limit_price, "f"), "belowTimeInForce": "GTC", "belowClientOrderId": stop_client_order_id, "listClientOrderId": list_client_order_id, "newOrderRespType": "RESULT", } return self._signed_request("POST", "/v3/orderList/oco", params, timestamp_ms) def create_market_order( self, symbol: str, side: OrderSide, quote_amount: Decimal, client_order_id: str, timestamp_ms: int | None = None, ) -> dict: params = { "symbol": symbol.upper(), "side": side.value, "type": "MARKET", "quoteOrderQty": format(quote_amount, "f"), "newClientOrderId": client_order_id, } return self._signed_request("POST", "/v3/order", params, timestamp_ms) def _select_base_url(self, endpoint_probe_timeout: float) -> str: if len(self.base_urls) == 1: return self.base_urls[0] best_url: str | None = None best_elapsed: float | None = None best_server_time_offset_ms = 0 for base_url in self.base_urls: start = self._clock() try: response = self._client.get( f"{base_url}/v3/time", timeout=endpoint_probe_timeout, ) response.raise_for_status() payload = response.json() except Exception: continue elapsed = self._clock() - start if best_elapsed is None or elapsed < best_elapsed: best_url = base_url best_elapsed = elapsed best_server_time_offset_ms = int(payload["serverTime"]) - int(self._clock() * 1000) self._server_time_offset_ms = best_server_time_offset_ms return best_url or self.base_urls[0] def _signed_request( self, method: str, path: str, params: dict[str, str], timestamp_ms: int | None = None, ) -> Any: headers = {"X-MBX-APIKEY": self.api_key} request_base_urls = self._request_base_urls(method) for index, base_url in enumerate(request_base_urls): signed_params = self._signed_params(params, timestamp_ms=timestamp_ms, base_url=base_url) request_kwargs = {"params": signed_params} if method.upper() == "GET" else {"data": signed_params} try: response = self._client.request(method, f"{base_url}{path}", headers=headers, **request_kwargs) response.raise_for_status() payload = response.json() except httpx.TransportError as exc: self._audit_api_call(method, path, signed_params, headers=headers, signed=True, error=exc) if self._should_retry_transport_error(method, index, request_base_urls): continue raise except Exception as exc: self._audit_api_call(method, path, signed_params, headers=headers, signed=True, error=exc) raise self.base_url = base_url self._audit_api_call( method, path, signed_params, headers=headers, signed=True, status_code=response.status_code, payload=payload, ) return payload raise RuntimeError("unreachable signed request retry state") def _signed_params(self, params: dict[str, str], *, timestamp_ms: int | None, base_url: str) -> dict[str, str]: signed_params = dict(params) signed_params["recvWindow"] = str(self.recv_window) signed_params["timestamp"] = str( timestamp_ms if timestamp_ms is not None else self._signed_timestamp_ms(base_url) ) query = urlencode(signed_params) signed_params["signature"] = sign_query(query, self.api_secret) return signed_params def _signed_timestamp_ms(self, base_url: str | None = None) -> int: resolved_base_url = base_url or self.base_url try: response = self._client.get(f"{resolved_base_url}/v3/time") response.raise_for_status() return int(response.json()["serverTime"]) except Exception: return int(self._clock() * 1000) + self._server_time_offset_ms def _public_request(self, method: str, path: str, params: dict[str, str]) -> Any: request_base_urls = self._request_base_urls(method) for index, base_url in enumerate(request_base_urls): try: response = self._client.request(method, f"{base_url}{path}", params=params) response.raise_for_status() payload = response.json() except httpx.TransportError as exc: self._audit_api_call(method, path, params, signed=False, error=exc) if self._should_retry_transport_error(method, index, request_base_urls): continue raise except Exception as exc: self._audit_api_call(method, path, params, signed=False, error=exc) raise self.base_url = base_url self._audit_api_call(method, path, params, signed=False, status_code=response.status_code, payload=payload) return payload raise RuntimeError("unreachable public request retry state") def _request_base_urls(self, method: str) -> tuple[str, ...]: if method.upper() != "GET": return (self.base_url,) return (self.base_url, *(base_url for base_url in self.base_urls if base_url != self.base_url)) def _should_retry_transport_error(self, method: str, index: int, request_base_urls: tuple[str, ...]) -> bool: return method.upper() == "GET" and index < len(request_base_urls) - 1 def _audit_api_call( self, method: str, path: str, params: dict[str, str], *, signed: bool, headers: dict[str, str] | None = None, status_code: int | None = None, payload: Any | None = None, error: Exception | None = None, ) -> None: try: self.audit_logger.log( "binance.api", params={ "method": method.upper(), "path": path, "signed": signed, "params": params, **({"headers": headers} if headers else {}), }, result=( None if error is not None else { "status_code": status_code, "payload": payload, } ), error=error, level="ERROR" if error is not None else "INFO", ) except Exception: pass @staticmethod def parse_account(payload: dict) -> AccountSnapshot: balances = { row["asset"].upper(): Balance( asset=row["asset"], free=Decimal(row["free"]), locked=Decimal(row.get("locked", "0")), ) for row in payload.get("balances", []) } return AccountSnapshot(balances)