Bot Integration with KuCoin API

KuCoin API Bot Integration

KuCoin provides REST and WebSocket APIs for spot, margin, and futures (KuCoin Futures). Key feature: for WebSocket connections, KuCoin requires obtaining an endpoint via REST request — you cannot simply connect to a static URL.

Authentication

import hmac
import hashlib
import base64
import time
import json
import httpx

class KuCoinClient:
    BASE_URL = "https://api.kucoin.com"
    FUTURES_URL = "https://api-futures.kucoin.com"

    def __init__(self, api_key: str, api_secret: str, passphrase: str):
        self.api_key = api_key
        self.api_secret = api_secret
        # KuCoin v2 signature: passphrase is also signed
        self.passphrase = base64.b64encode(
            hmac.new(api_secret.encode(), passphrase.encode(), hashlib.sha256).digest()
        ).decode()

    def _sign(self, timestamp: str, method: str, endpoint: str, body: str = "") -> str:
        str_to_sign = timestamp + method.upper() + endpoint + body
        return base64.b64encode(
            hmac.new(self.api_secret.encode(), str_to_sign.encode(), hashlib.sha256).digest()
        ).decode()

    def _headers(self, method: str, endpoint: str, body: str = "") -> dict:
        timestamp = str(int(time.time() * 1000))
        return {
            "KC-API-KEY": self.api_key,
            "KC-API-SIGN": self._sign(timestamp, method, endpoint, body),
            "KC-API-TIMESTAMP": timestamp,
            "KC-API-PASSPHRASE": self.passphrase,
            "KC-API-KEY-VERSION": "2",
            "Content-Type": "application/json"
        }

Trading Operations

async def place_order(
    self,
    symbol: str,       # 'BTC-USDT'
    side: str,         # 'buy' or 'sell'
    order_type: str,   # 'limit' or 'market'
    size: str,         # for limit/market by base
    price: str = None,
    funds: str = None  # for market buy by quote
) -> dict:
    endpoint = "/api/v1/orders"
    payload = {
        "clientOid": str(int(time.time() * 1000)),  # unique client ID
        "symbol": symbol,
        "side": side,
        "type": order_type
    }

    if order_type == "limit":
        payload["size"] = size
        payload["price"] = price
    elif side == "buy" and funds:
        payload["funds"] = funds  # buy for $X USDT
    else:
        payload["size"] = size

    body = json.dumps(payload)
    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"{self.BASE_URL}{endpoint}",
            content=body,
            headers=self._headers("POST", endpoint, body)
        )
    result = response.json()

    if result.get("code") != "200000":
        raise KuCoinError(f"Order error: {result.get('msg')}")
    return result["data"]

async def get_accounts(self, currency: str = None) -> list:
    endpoint = "/api/v1/accounts"
    if currency:
        endpoint += f"?currency={currency}"

    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"{self.BASE_URL}{endpoint}",
            headers=self._headers("GET", endpoint)
        )
    return response.json().get("data", [])

WebSocket: Obtaining Endpoint

KuCoin does not publish a static WebSocket URL — you must request it:

async def get_ws_endpoint(self, private: bool = False) -> dict:
    endpoint = "/api/v1/bullet-private" if private else "/api/v1/bullet-public"
    method = "POST" if private else "POST"

    headers = self._headers(method, endpoint) if private else {"Content-Type": "application/json"}

    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"{self.BASE_URL}{endpoint}",
            headers=headers
        )
    data = response.json()["data"]

    # Form URL with token
    server = data["instanceServers"][0]
    token = data["token"]
    ws_url = f"{server['endpoint']}?token={token}&connectId={int(time.time()*1000)}"
    ping_interval = server["pingInterval"] / 1000  # in seconds

    return {"url": ws_url, "ping_interval": ping_interval}

async def subscribe_ticker(self, symbols: list[str]):
    ws_data = await self.get_ws_endpoint(private=False)

    async with websockets.connect(ws_data["url"]) as ws:
        # Subscribe
        await ws.send(json.dumps({
            "id": str(int(time.time() * 1000)),
            "type": "subscribe",
            "topic": f"/market/ticker:{','.join(symbols)}",
            "privateChannel": False,
            "response": True
        }))

        # Ping on schedule
        async def keepalive():
            while True:
                await asyncio.sleep(ws_data["ping_interval"])
                await ws.send(json.dumps({"id": "ping", "type": "ping"}))

        asyncio.create_task(keepalive())

        async for message in ws:
            data = json.loads(message)
            if data["type"] == "message" and "data" in data:
                await self.on_ticker(data["data"])

KuCoin API returns code: "200000" on success (not HTTP status). Check exactly this field. Official SDK: pip install kucoin-python. Sandbox: https://openapi-sandbox.kucoin.com — complete production mirror.