112 lines
4.3 KiB
Python
112 lines
4.3 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
import time
|
|
from dataclasses import dataclass, field
|
|
from typing import Any
|
|
from urllib.parse import urlencode
|
|
from urllib.request import Request, urlopen
|
|
|
|
|
|
REST_TESTNET = "https://testnet.binancefuture.com"
|
|
REST_MAINNET = "https://fapi.binance.com"
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class BinanceUsdMAccountClient:
|
|
api_key: str
|
|
api_secret: str
|
|
testnet: bool = False
|
|
_response_cache: dict[str, tuple[float, Any]] = field(default_factory=dict, init=False, repr=False)
|
|
|
|
@property
|
|
def base_url(self) -> str:
|
|
return REST_TESTNET if self.testnet else REST_MAINNET
|
|
|
|
def _sign_params(self, params: dict[str, Any] | None = None) -> dict[str, Any]:
|
|
payload = dict(params or {})
|
|
payload["timestamp"] = int(time.time() * 1000)
|
|
query = urlencode(payload)
|
|
signature = hmac.new(self.api_secret.encode("utf-8"), query.encode("utf-8"), hashlib.sha256).hexdigest()
|
|
payload["signature"] = signature
|
|
return payload
|
|
|
|
def _get(self, path: str, params: dict[str, Any] | None = None, *, auth: bool = True) -> Any:
|
|
query_params = self._sign_params(params) if auth else dict(params or {})
|
|
url = f"{self.base_url}{path}?{urlencode(query_params)}"
|
|
request = Request(url, headers={"X-MBX-APIKEY": self.api_key})
|
|
with urlopen(request, timeout=15) as response:
|
|
return json.loads(response.read().decode("utf-8"))
|
|
|
|
def _post(self, path: str, params: dict[str, Any] | None = None, *, auth: bool = True) -> Any:
|
|
payload = self._sign_params(params) if auth else dict(params or {})
|
|
encoded = urlencode(payload).encode("utf-8")
|
|
url = f"{self.base_url}{path}"
|
|
request = Request(url, data=encoded, method="POST", headers={"X-MBX-APIKEY": self.api_key})
|
|
with urlopen(request, timeout=15) as response:
|
|
return json.loads(response.read().decode("utf-8"))
|
|
|
|
def _get_cached(self, key: str, ttl_seconds: float, loader) -> Any:
|
|
now = time.time()
|
|
cached = self._response_cache.get(key)
|
|
if cached is not None:
|
|
loaded_at, payload = cached
|
|
if now - loaded_at <= ttl_seconds:
|
|
return payload
|
|
payload = loader()
|
|
self._response_cache[key] = (now, payload)
|
|
return payload
|
|
|
|
def get_balance(self) -> list[dict[str, Any]]:
|
|
payload = self._get_cached("balance", 10.0, lambda: list(self._get("/fapi/v2/balance")))
|
|
return list(payload)
|
|
|
|
def get_position_risk(self) -> list[dict[str, Any]]:
|
|
payload = self._get_cached("position_risk", 5.0, lambda: list(self._get("/fapi/v2/positionRisk")))
|
|
return list(payload)
|
|
|
|
def get_exchange_info(self) -> dict[str, Any]:
|
|
payload = self._get_cached("exchange_info", 3600.0, lambda: dict(self._get("/fapi/v1/exchangeInfo", auth=False)))
|
|
return dict(payload)
|
|
|
|
def get_ticker_price(self, symbol: str | None = None) -> Any:
|
|
params = {"symbol": symbol} if symbol else None
|
|
key = f"ticker_price:{symbol or '*'}"
|
|
return self._get_cached(key, 5.0, lambda: self._get("/fapi/v1/ticker/price", params=params, auth=False))
|
|
|
|
def set_leverage(self, symbol: str, leverage: int) -> dict[str, Any]:
|
|
return dict(self._post("/fapi/v1/leverage", {"symbol": symbol, "leverage": max(1, int(leverage))}))
|
|
|
|
def place_market_order(
|
|
self,
|
|
*,
|
|
symbol: str,
|
|
side: str,
|
|
quantity: float,
|
|
reduce_only: bool = False,
|
|
client_order_id: str | None = None,
|
|
) -> dict[str, Any]:
|
|
params: dict[str, Any] = {
|
|
"symbol": symbol,
|
|
"side": side.upper(),
|
|
"type": "MARKET",
|
|
"quantity": self._format_decimal(quantity),
|
|
"newOrderRespType": "RESULT",
|
|
}
|
|
if reduce_only:
|
|
params["reduceOnly"] = "true"
|
|
if client_order_id:
|
|
clean = "".join(ch for ch in str(client_order_id) if ch.isalnum() or ch in "-_.")
|
|
if clean:
|
|
params["newClientOrderId"] = clean[:36]
|
|
response = dict(self._post("/fapi/v1/order", params))
|
|
self._response_cache.pop("balance", None)
|
|
self._response_cache.pop("position_risk", None)
|
|
return response
|
|
|
|
@staticmethod
|
|
def _format_decimal(value: float) -> str:
|
|
return ("%.12f" % float(value)).rstrip("0").rstrip(".")
|