""" CoinSwitch PRO — HFT (low-latency futures + options) trading client (Python). Single-file client. Every HFT REST endpoint is a method. The HFT response envelope is `{retCode, retMsg, result, retExtInfo, time}` — always check `retCode == 0` before trusting `result`. HFT base URL: https://dma.coinswitch.co/ Requirements: pip install cryptography requests Usage: from coinswitch_hft import HFTClient h = HFTClient(api_key="", secret_key="") print(h.server_time()) # {retCode: 0, ...} print(h.tickers(symbol="BTCUSDT")) h.transfer_funds(direction="IN", amount=100, quote_asset="USDT") h.set_leverage(symbol="BTCUSDT", buy_leverage=2, sell_leverage=2) h.place_order( symbol="BTCUSDT", side="Buy", order_type="Limit", qty="0.001", price="50000", position_idx=1, order_link_id="my-order-1", ) """ from __future__ import annotations import json import time import urllib.parse import uuid from typing import Any, Optional import requests from cryptography.hazmat.primitives.asymmetric import ed25519 BASE_URL = "https://dma.coinswitch.co" class HFTClient: def __init__( self, api_key: str, secret_key: str, base_url: str = BASE_URL, request_timeout: float = 5.0, ) -> None: self.api_key = api_key self.secret_key = secret_key self.base_url = base_url.rstrip("/") self.timeout = request_timeout self._secret = ed25519.Ed25519PrivateKey.from_private_bytes( bytes.fromhex(secret_key) ) self._session = requests.Session() # ------------------------------------------------------------------ # Internals # ------------------------------------------------------------------ def _sign(self, method: str, path: str, params: Optional[dict] = None): method = method.upper() if params: sep = "&" if "?" in path else "?" path = path + sep + urllib.parse.urlencode(params) decoded_path = urllib.parse.unquote_plus(path) epoch = str(int(time.time() * 1000)) message = method + decoded_path + epoch signature = self._secret.sign(message.encode("utf-8")).hex() headers = { "Content-Type": "application/json", "X-AUTH-APIKEY": self.api_key, "X-AUTH-SIGNATURE": signature, "X-AUTH-EPOCH": epoch, # mandatory on HFT } return headers, decoded_path def _request(self, method: str, path: str, *, params: Optional[dict] = None, body: Optional[dict] = None) -> Any: headers, decoded_path = self._sign(method, path, params) kwargs: dict = {"headers": headers, "timeout": self.timeout} if body is not None: kwargs["data"] = json.dumps(body) r = self._session.request(method.upper(), self.base_url + decoded_path, **kwargs) r.raise_for_status() return r.json() # ------------------------------------------------------------------ # Market data # ------------------------------------------------------------------ def server_time(self) -> Any: """GET /v5/market/time. The HFT proxy requires auth headers even on nominally-public endpoints.""" return self._request("GET", "/v5/market/time") def instruments_info(self, *, category: str = "linear", symbol: Optional[str] = None, limit: Optional[int] = None) -> Any: params: dict = {"category": category} if symbol: params["symbol"] = symbol if limit is not None: params["limit"] = limit return self._request("GET", "/v5/market/instruments-info", params=params) def tickers(self, *, category: str = "linear", symbol: Optional[str] = None) -> Any: params: dict = {"category": category} if symbol: params["symbol"] = symbol return self._request("GET", "/v5/market/tickers", params=params) def order_book(self, *, category: str = "linear", symbol: str, limit: Optional[int] = None) -> Any: params: dict = {"category": category, "symbol": symbol} if limit is not None: params["limit"] = limit return self._request("GET", "/v5/market/orderbook", params=params) # ------------------------------------------------------------------ # Orders # ------------------------------------------------------------------ def place_order( self, *, symbol: str, side: str, order_type: str, qty: str, price: Optional[str] = None, position_idx: int, time_in_force: str = "GTC", order_link_id: Optional[str] = None, category: str = "linear", ) -> Any: body: dict = { "category": category, "symbol": symbol, "side": side, "orderType": order_type, "qty": qty, "positionIdx": position_idx, "timeInForce": time_in_force, } if price is not None: body["price"] = price body["orderLinkId"] = order_link_id or str(uuid.uuid4()) return self._request("POST", "/v5/order/create", body=body) def order_realtime( self, *, category: str = "linear", symbol: Optional[str] = None, order_id: Optional[str] = None, order_link_id: Optional[str] = None, cursor: Optional[str] = None, ) -> Any: params: dict = {"category": category} if symbol: params["symbol"] = symbol if order_id: params["orderId"] = order_id if order_link_id: params["orderLinkId"] = order_link_id if cursor: params["cursor"] = cursor return self._request("GET", "/v5/order/realtime", params=params) def execution_list( self, *, category: str = "linear", symbol: Optional[str] = None, order_id: Optional[str] = None, order_link_id: Optional[str] = None, cursor: Optional[str] = None, ) -> Any: params: dict = {"category": category} if symbol: params["symbol"] = symbol if order_id: params["orderId"] = order_id if order_link_id: params["orderLinkId"] = order_link_id if cursor: params["cursor"] = cursor return self._request("GET", "/v5/execution/list", params=params) # ------------------------------------------------------------------ # Positions # ------------------------------------------------------------------ def positions(self, *, category: str = "linear", symbol: Optional[str] = None, cursor: Optional[str] = None) -> Any: params: dict = {"category": category} if symbol: params["symbol"] = symbol if cursor: params["cursor"] = cursor return self._request("GET", "/v5/position/list", params=params) def set_leverage(self, *, symbol: str, buy_leverage: float, sell_leverage: float, category: str = "linear") -> Any: return self._request( "POST", "/v5/position/set-leverage", body={ "category": category, "symbol": symbol, "buyLeverage": str(buy_leverage), "sellLeverage": str(sell_leverage), }, ) def switch_position_mode(self, *, mode: int, coin: str = "USDT", category: str = "linear") -> Any: """mode: 0 = one-way, 3 = hedge.""" return self._request( "POST", "/v5/position/switch-mode", body={"category": category, "coin": coin, "mode": mode}, ) # ------------------------------------------------------------------ # Account # ------------------------------------------------------------------ def transfer_funds( self, *, direction: str, amount: float, quote_asset: str = "USDT", client_txn_id: Optional[str] = None, ) -> Any: """ Move funds between your CoinSwitch wallet and your HFT trading account. direction: "IN" or "OUT". The first transfer locks the account's settlement currency (USDT or INR); subsequent transfers must use the same currency. """ body = { "direction": direction, "amount": amount, "quote_asset": quote_asset, "client_txn_id": client_txn_id or str(uuid.uuid4()), } return self._request("POST", "/dma/api/v1/funds/transfer", body=body) # ------------------------------------------------------------------ # WebSocket signature for private NATS streams # ------------------------------------------------------------------ def socket_signature(self, *, expires_in_seconds: int = 300) -> Any: """ Returns {api_key, expires, signature} — pass these directly to the upstream NATS WebSocket auth message. Do not log or persist. """ return self._request( "GET", "/dma/api/v1/socket/signature", params={"expires_in_seconds": expires_in_seconds}, )