""" Tabdeal Futures API client (Binance-compatible FAPI). Handles auth, signing, and all REST calls needed by the bot. """ import hashlib import hmac import time import logging from typing import Any, Optional from urllib.parse import urlencode import requests from tabdeal_bot.config.settings import ExchangeConfig logger = logging.getLogger(__name__) class TabdealClient: def __init__(self, cfg: ExchangeConfig): self.cfg = cfg self.session = requests.Session() self.session.headers.update({ "X-MBX-APIKEY": cfg.api_key, "Content-Type": "application/x-www-form-urlencoded", }) # ------------------------------------------------------------------ # # Signing helpers # # ------------------------------------------------------------------ # def _timestamp(self) -> int: return int(time.time() * 1000) def _sign(self, params: dict) -> dict: params["timestamp"] = self._timestamp() params["recvWindow"] = self.cfg.recv_window query = urlencode(params) sig = hmac.new( self.cfg.api_secret.encode(), query.encode(), hashlib.sha256 ).hexdigest() params["signature"] = sig return params # ------------------------------------------------------------------ # # Low-level request # # ------------------------------------------------------------------ # def _request( self, method: str, path: str, params: Optional[dict] = None, signed: bool = False, data: Optional[dict] = None, ) -> Any: url = self.cfg.base_url + path p = dict(params or {}) d = dict(data or {}) if signed: if method.upper() in ("GET", "DELETE"): p = self._sign(p) else: d = self._sign(d) try: resp = self.session.request( method, url, params=p if method.upper() in ("GET", "DELETE") else None, data=d if method.upper() not in ("GET", "DELETE") else None, timeout=10, ) resp.raise_for_status() return resp.json() except requests.exceptions.HTTPError as e: logger.error("HTTP %s %s → %s: %s", method, path, e.response.status_code, e.response.text) raise except requests.exceptions.RequestException as e: logger.error("Request failed %s %s: %s", method, path, e) raise # ------------------------------------------------------------------ # # Market data # # ------------------------------------------------------------------ # def ping(self) -> bool: try: self._request("GET", "/r/fapi/v1/ping") return True except Exception: return False def get_server_time(self) -> int: return self._request("GET", "/r/fapi/v1/time")["serverTime"] def get_exchange_info(self, symbol: Optional[str] = None) -> dict: p = {} if symbol: p["symbol"] = symbol return self._request("GET", "/r/fapi/v1/exchangeInfo", params=p) def get_klines(self, symbol: str, interval: str, limit: int = 100) -> list: """ Fetch OHLCV candles. Tabdeal mirrors Binance FAPI — endpoint may be /r/fapi/v1/klines or /fapi/v1/klines. Returns list of [open_time, open, high, low, close, volume, ...]. """ p = {"symbol": symbol, "interval": interval, "limit": limit} try: return self._request("GET", "/r/fapi/v1/klines", params=p) except Exception: # fallback path without /r/ prefix return self._request("GET", "/fapi/v1/klines", params=p) def get_depth(self, symbol: str, limit: int = 20) -> dict: return self._request("GET", "/r/fapi/v1/depth", params={"symbol": symbol, "limit": limit}) # ------------------------------------------------------------------ # # Account # # ------------------------------------------------------------------ # def get_balance(self) -> list: return self._request("GET", "/r/fapi/v3/balance", signed=True) def get_account(self) -> dict: return self._request("GET", "/r/fapi/v3/account", signed=True) def get_usdt_balance(self) -> float: balances = self.get_balance() for b in balances: if b.get("asset") == "USDT": return float(b.get("availableBalance", 0)) return 0.0 # ------------------------------------------------------------------ # # Leverage # # ------------------------------------------------------------------ # def set_leverage(self, symbol: str, leverage: int) -> dict: return self._request( "POST", "/fapi/v1/leverage", data={"symbol": symbol, "leverage": leverage}, signed=True, ) def get_leverage(self, symbol: str) -> int: result = self._request("GET", "/r/fapi/v1/leverage", params={"symbol": symbol}, signed=True) if isinstance(result, list): for r in result: if r.get("symbol") == symbol: return int(r.get("leverage", 1)) return int(result.get("leverage", 1)) # ------------------------------------------------------------------ # # Positions # # ------------------------------------------------------------------ # def get_position(self, symbol: str) -> Optional[dict]: """Return the open position for a symbol, or None.""" positions = self._request("GET", "/r/fapi/v3/positionRisk", params={"symbol": symbol}, signed=True) if isinstance(positions, list): for p in positions: if abs(float(p.get("positionAmt", 0))) > 0: return p return None return None if abs(float(positions.get("positionAmt", 0))) == 0 else positions def get_all_positions(self) -> list: return self._request("GET", "/r/fapi/v3/positionRisk", signed=True) def close_position(self, symbol: str) -> dict: """Close open position with a market order.""" return self._request("DELETE", "/fapi/v1/position", data={"symbol": symbol}, signed=True) # ------------------------------------------------------------------ # # Orders # # ------------------------------------------------------------------ # def place_order( self, symbol: str, side: str, # BUY or SELL order_type: str, # LIMIT or MARKET quantity: float, price: Optional[float] = None, time_in_force: str = "GTC", reduce_only: bool = False, client_order_id: Optional[str] = None, ) -> dict: d: dict = { "symbol": symbol, "side": side, "type": order_type, "quantity": quantity, } if order_type == "LIMIT": d["price"] = price d["timeInForce"] = time_in_force if reduce_only: d["reduceOnly"] = "true" if client_order_id: d["newClientOrderId"] = client_order_id return self._request("POST", "/fapi/v1/order", data=d, signed=True) def set_sl_tp( self, position_id: str, symbol: Optional[str] = None, sl_price: Optional[float] = None, tp_price: Optional[float] = None, working_type: str = "MARK_PRICE", ) -> dict: d: dict = {"positionId": position_id, "workingType": working_type} if symbol: d["symbol"] = symbol if sl_price is not None: d["slPrice"] = sl_price if tp_price is not None: d["tpPrice"] = tp_price return self._request("POST", "/fapi/v1/positionSlTp", data=d, signed=True) def get_open_orders(self, symbol: Optional[str] = None) -> list: p = {} if symbol: p["symbol"] = symbol return self._request("GET", "/r/fapi/v1/openOrders", params=p, signed=True) def cancel_order(self, symbol: str, order_id: int) -> dict: return self._request( "DELETE", "/fapi/v1/order", data={"symbol": symbol, "orderId": order_id}, signed=True, ) def get_order(self, symbol: str, order_id: int) -> dict: return self._request( "GET", "/r/fapi/v1/order", params={"symbol": symbol, "orderId": order_id}, signed=True, )