""" KuCoin REST API client for the executor. Covers symbol metadata fetch, HF order-test endpoint for paper mode, real order placement for live mode, and private WebSocket token acquisition required by ws_client.py. """ import aiohttp import base64 import hmac import hashlib import json import time import uuid import structlog from decimal import Decimal from typing import Optional logger = structlog.get_logger().bind(component="executor-kucoin") KUCoin_SYMBOLS_URL = "https://api.kucoin.com/api/v1/symbols" KUCoin_ORDER_TEST_URL = "https://api.kucoin.com/api/v1/hf/orders/test" KUCoin_ORDER_PLACE_URL = "https://api.kucoin.com/api/v1/hf/orders" KUCoin_SERVER_TIME_URL = "https://api.kucoin.com/api/v1/time" KUCoin_BULLET_PRIVATE_URL = "https://api.kucoin.com/api/v1/bullet-private" def _signRequest(timestamp: str, method: str, path: str, secret: str, data: str = "") -> str: """ Compute the HMAC-SHA256 signature for a KuCoin API request. Parameters ---------- timestamp : str Millisecond Unix timestamp. method : str HTTP method (e.g. "POST"). path : str API endpoint path (e.g. "/api/v1/hf/orders/test"). secret : str API secret as returned by the user. data : str Request body. Empty string for GET requests. Returns ------- str Base64-encoded HMAC-SHA256 digest suitable for the KC-API-SIGN header. """ # KuCoin signs timestamp + method + path + body concatenated. message = timestamp + method + path + data mac = hmac.new(secret.encode("utf-8"), message.encode("utf-8"), hashlib.sha256) return base64.b64encode(mac.digest()).decode("utf-8") def _encryptPassphrase(passphrase: str, secret: str) -> str: """ Encrypt the API passphrase for KuCoin. KuCoin requires the passphrase to be encrypted with HMAC-SHA256 using the API secret as the key, then base64-encoded. Returns ------- str Base64-encoded encrypted passphrase for the KC-API-PASSPHRASE header. """ mac = hmac.new(secret.encode("utf-8"), passphrase.encode("utf-8"), hashlib.sha256) return base64.b64encode(mac.digest()).decode("utf-8") class SymbolMeta: """ Cached metadata for a single KuCoin trading pair. Fetched once at startup from /api/v1/symbols and held in memory for the lifetime of the process. """ def __init__( self, symbol: str, base: str, quote: str, base_min_size: Decimal, quote_min_size: Decimal, base_increment: Decimal, quote_increment: Decimal, min_funds: Decimal, taker_fee: Decimal, maker_fee: Decimal, incr_min_size: Decimal, base_precision: int, quote_precision: int, ) -> None: self.symbol = symbol self.base = base self.quote = quote self.base_min_size = base_min_size self.quote_min_size = quote_min_size self.base_increment = base_increment self.quote_increment = quote_increment self.min_funds = min_funds self.taker_fee = taker_fee self.maker_fee = maker_fee self.incr_min_size = incr_min_size self.base_precision = base_precision self.quote_precision = quote_precision def _cost_to_precision(amount: Decimal, quote_increment: Decimal) -> Decimal: """ Round a quote-currency amount *up* to the nearest valid increment. KuCoin requires quote amounts to be multiples of quote_increment. When computing the minimum order size in quote currency we therefore round up so we never underestimate the minimum. """ if quote_increment <= 0: return amount mult = Decimal("1") / quote_increment return (amount * mult).to_integral_value(rounding="ROUND_UP") / mult def _amount_to_precision( amount: Decimal, base_increment: Decimal, round_down: bool = False, ) -> Decimal: """ Round a base-currency amount to the nearest valid increment. Use round_down=True for sell orders so the resulting size never exceeds the available balance. Use round_down=False (default) for buy orders and minimum-size computations where rounding up is safe. """ if base_increment <= 0: return amount mult = Decimal("1") / base_increment rounding = "ROUND_DOWN" if round_down else "ROUND_UP" return (amount * mult).to_integral_value(rounding=rounding) / mult class KuCoinAPI: """ Thin client for KuCoin REST API calls needed by the executor. Handles HMAC signing, symbol metadata caching, order placement (live) and order validation (paper), plus private token fetch for the WebSocket client. """ def __init__(self, api_key: str = "", api_secret: str = "", api_passphrase: str = "") -> None: self._api_key = api_key self._api_secret = api_secret self._api_passphrase = api_passphrase # Pre-encrypt the passphrase at init time so we don't recompute per request. self._encrypted_passphrase = _encryptPassphrase(api_passphrase, api_secret) if api_passphrase and api_secret else "" self._symbols: dict[str, SymbolMeta] = {} self._log = logger async def fetch_symbols(self) -> None: """ Fetch all trading pair metadata from KuCoin and populate _symbols. Logs a warning for any symbol that fails to parse and skips it. """ async with aiohttp.ClientSession() as session: async with session.get(KUCoin_SYMBOLS_URL) as resp: resp.raise_for_status() payload = await resp.json() for item in payload.get("data", []): symbol = item.get("symbol", "") base = item.get("baseCurrency", "") quote = item.get("quoteCurrency", "") if not all([symbol, base, quote]): continue raw_base_min = item.get("baseMinSize") or "0" raw_quote_min = item.get("quoteMinSize") or "0" raw_base_inc = item.get("baseIncrement") or "0" raw_quote_inc = item.get("quoteIncrement") or "0" raw_min_funds = item.get("minFunds") or "0" raw_maker = item.get("makerFeeRate") or "0" raw_taker = item.get("takerFeeRate") or "0" raw_incr_min = item.get("incrMinSize") or "0" raw_base_prec = item.get("basePrecision") or "0" raw_quote_prec = item.get("quotePrecision") or "0" try: base_min_size = Decimal(raw_base_min) quote_min_size = Decimal(raw_quote_min) base_increment = Decimal(raw_base_inc) quote_increment = Decimal(raw_quote_inc) min_funds = Decimal(raw_min_funds) taker_fee = Decimal(raw_taker) maker_fee = Decimal(raw_maker) incr_min_size = Decimal(raw_incr_min) base_precision = int(raw_base_prec) if raw_base_prec else 0 quote_precision = int(raw_quote_prec) if raw_quote_prec else 0 except Exception as e: self._log.warning("symbol_field_parse_error", symbol=symbol, error=str(e), baseMinSize=repr(raw_base_min), quoteMinSize=repr(raw_quote_min), baseIncrement=repr(raw_base_inc), quoteIncrement=repr(raw_quote_inc), minFunds=repr(raw_min_funds), makerFeeRate=repr(raw_maker), takerFeeRate=repr(raw_taker), incrMinSize=repr(raw_incr_min), basePrecision=repr(raw_base_prec), quotePrecision=repr(raw_quote_prec)) continue self._symbols[symbol] = SymbolMeta( symbol=symbol, base=base, quote=quote, base_min_size=base_min_size, quote_min_size=quote_min_size, base_increment=base_increment, quote_increment=quote_increment, min_funds=min_funds, taker_fee=taker_fee, maker_fee=maker_fee, incr_min_size=incr_min_size, base_precision=base_precision, quote_precision=quote_precision, ) self._log.info("symbols_loaded", count=len(self._symbols)) def get_symbol_meta(self, symbol: str) -> Optional[SymbolMeta]: """Return cached SymbolMeta for a pair, or None if not yet loaded.""" return self._symbols.get(symbol) async def order_test( self, session: aiohttp.ClientSession, symbol: str, side: str, order_type: str, price: Optional[Decimal] = None, size: Optional[Decimal] = None, funds: Optional[Decimal] = None, ) -> tuple[bool, str, Optional[str]]: """ Validate an order against KuCoin's paper-trading endpoint. In paper mode the executor uses this instead of a real order so that fill simulation can proceed without risking capital. Parameters ---------- session : aiohttp.ClientSession Shared session for connection pooling. symbol : str KuCoin symbol e.g. "BTC-USDT". side : str "buy" or "sell". order_type : str Order type string (e.g. "market"). price, size, funds : Optional[Decimal] Order parameters; at least one must be provided. Returns ------- tuple[bool, str, Optional[str]] (success, error_message, order_id). order_id is populated on success. """ timestamp = str(int(time.time() * 1000)) path = "/api/v1/hf/orders/test" body: dict = { "symbol": symbol, "type": order_type, "side": side, "clientOid": f"paper-{timestamp}-{uuid.uuid4().hex[:8]}", } if price is not None: body["price"] = str(price) if size is not None: body["size"] = str(size) if funds is not None: body["funds"] = str(funds) raw_body = json.dumps(body, separators=(",", ":")) sign = _signRequest(timestamp, "POST", path, self._api_secret, raw_body) headers = { "KC-API-TIMESTAMP": timestamp, "KC-API-SIGN": sign, "KC-API-KEY": self._api_key, "KC-API-PASSPHRASE": self._encrypted_passphrase, "KC-API-SIGN-TYPE": "2", "KC-API-KEY-VERSION": "3", "Content-Type": "application/json", } try: async with session.post( KUCoin_ORDER_TEST_URL, data=raw_body.encode(), headers=headers, ) as resp: text = await resp.text() if resp.status == 200: data = json.loads(text) order_id = data.get("data", {}).get("orderId", "") return True, "", order_id else: data = json.loads(text) code = data.get("code", "") msg = data.get("msg", text) return False, f"{code}: {msg}", None except (json.JSONDecodeError, ValueError) as e: return False, f"invalid_response: {e}", None except (aiohttp.ClientError, OSError) as e: return False, f"http_error: {e}", None async def order_place( self, session: aiohttp.ClientSession, symbol: str, side: str, order_type: str, price: Optional[Decimal] = None, size: Optional[Decimal] = None, funds: Optional[Decimal] = None, client_oid: str = "", ) -> tuple[bool, str, Optional[str]]: """ Place a real market order on KuCoin. Parameters ---------- session : aiohttp.ClientSession Shared session for connection pooling. symbol : str KuCoin symbol e.g. "BTC-USDT". side : str "buy" or "sell". order_type : str Order type string (e.g. "market"). price, size, funds : Optional[Decimal] Order parameters; at least one must be provided. client_oid : str Client-order ID for matching with WS fill events. Returns ------- tuple[bool, str, Optional[str]] (success, error_message, order_id). order_id is populated on success. """ timestamp = str(int(time.time() * 1000)) path = "/api/v1/hf/orders" body: dict = { "symbol": symbol, "type": order_type, "side": side, "clientOid": client_oid or f"live-{timestamp}-{uuid.uuid4().hex[:8]}", } if price is not None: body["price"] = str(price) if size is not None: body["size"] = str(size) if funds is not None: body["funds"] = str(funds) raw_body = json.dumps(body, separators=(",", ":")) sign = _signRequest(timestamp, "POST", path, self._api_secret, raw_body) headers = { "KC-API-TIMESTAMP": timestamp, "KC-API-SIGN": sign, "KC-API-KEY": self._api_key, "KC-API-PASSPHRASE": self._encrypted_passphrase, "KC-API-SIGN-TYPE": "2", "KC-API-KEY-VERSION": "3", "Content-Type": "application/json", } try: async with session.post( KUCoin_ORDER_PLACE_URL, data=raw_body.encode(), headers=headers, ) as resp: text = await resp.text() self._log.info("order_place_raw_response", status=resp.status, response=text[:300]) data = json.loads(text) if resp.status == 200 and data.get("code") == "200000": order_id = data.get("data", {}).get("orderId", "") return True, "", order_id else: code = data.get("code", resp.status) msg = data.get("msg", text) return False, f"{code}: {msg}", None except (json.JSONDecodeError, ValueError) as e: return False, f"invalid_response: {e}", None except (aiohttp.ClientError, OSError) as e: return False, f"http_error: {e}", None async def get_private_token(self, session: aiohttp.ClientSession) -> Optional[dict]: """ Request a private WebSocket token from KuCoin. Returns the full data dict from the bullet-private response, containing 'token', 'instanceServers' with endpoint, pingInterval, pingTimeout. Returns None on error. """ timestamp = str(int(time.time() * 1000)) path = "/api/v1/bullet-private" sign = _signRequest(timestamp, "POST", path, self._api_secret, "") headers = { "KC-API-TIMESTAMP": timestamp, "KC-API-SIGN": sign, "KC-API-KEY": self._api_key, "KC-API-PASSPHRASE": self._encrypted_passphrase, "KC-API-SIGN-TYPE": "2", "KC-API-KEY-VERSION": "3", "Content-Type": "application/json", } try: async with session.post( KUCoin_BULLET_PRIVATE_URL, data=b"", headers=headers, ) as resp: if resp.status == 200: data = await resp.json() return data.get("data") else: text = await resp.text() self._log.error("private_token_failed", status=resp.status, response=text[:200]) return None except (aiohttp.ClientError, OSError) as e: self._log.error("private_token_http_error", error=str(e)) return None async def warmup_session(self, session: aiohttp.ClientSession) -> None: """Warm up the authenticated HTTP connection pool with a minimal GET.""" timestamp = str(int(time.time() * 1000)) path = "/api/v1/accounts" sign = _signRequest(timestamp, "GET", path, self._api_secret, "") headers = { "KC-API-TIMESTAMP": timestamp, "KC-API-SIGN": sign, "KC-API-KEY": self._api_key, "KC-API-PASSPHRASE": self._encrypted_passphrase, "KC-API-SIGN-TYPE": "2", "KC-API-KEY-VERSION": "3", } try: async with session.get( f"https://api.kucoin.com{path}", headers=headers, ) as resp: await resp.read() self._log.info("session_warmed", status=resp.status) except Exception as e: self._log.warning("session_warmup_failed", error=str(e))