''' https://github.com/Kucoin/kucoin-universal-sdk/blob/main/sdk/python/README.md ''' from credentials import get_api_key from kucoin_universal_sdk.api.client import DefaultClient from kucoin_universal_sdk.model.client_option import ClientOptionBuilder from kucoin_universal_sdk.model.constants import GLOBAL_API_ENDPOINT, GLOBAL_FUTURES_API_ENDPOINT, GLOBAL_BROKER_API_ENDPOINT from kucoin_universal_sdk.model.transport_option import TransportOptionBuilder from kucoin_universal_sdk.generate.earn.earn.model_get_savings_products_req import GetSavingsProductsReqBuilder from kucoin_universal_sdk.generate.earn.earn.model_get_account_holding_req import GetAccountHoldingReqBuilder from kucoin_universal_sdk.generate.earn.earn.model_purchase_req import PurchaseReqBuilder from kucoin_universal_sdk.generate.earn.earn.model_redeem_req import RedeemReqBuilder from kucoin_universal_sdk.generate.account.account.model_get_spot_account_list_req import GetSpotAccountListReqBuilder class kucoin_earn: def __init__(self): api_key, api_secret, api_passphrase = get_api_key("kucoin") http_transport_option = ( TransportOptionBuilder() .set_keep_alive(True) .set_max_pool_size(10) .set_max_connection_per_pool(10) .build() ) client_option = ( ClientOptionBuilder() .set_key(api_key) .set_secret(api_secret) .set_passphrase(api_passphrase) .set_spot_endpoint(GLOBAL_API_ENDPOINT) .set_futures_endpoint(GLOBAL_FUTURES_API_ENDPOINT) .set_broker_endpoint(GLOBAL_BROKER_API_ENDPOINT) .set_transport_option(http_transport_option) .build() ) self.client = DefaultClient(client_option) kucoin_rest_service = self.client.rest_service() self.account_api = kucoin_rest_service.get_account_service().get_account_api self.earn_api = kucoin_rest_service.get_earn_service().get_earn_api def __str__(self): return "kucoin" def get_trading_balance(self, coin): ''' Args: coin (str): The coin to get the balance of Returns: dict: The free available balance of the coin in the trading account (or the equivalent account in the exchange) ''' request = GetSpotAccountListReqBuilder().set_currency(coin).set_type("trade").build() response = self.account_api().get_spot_account_list(request) balance = response.to_dict()["data"][0]["available"] return {coin: balance} def get_available_products(self, coin): request = GetSavingsProductsReqBuilder().set_currency(coin).build() response = self.earn_api().get_savings_products(request) return {"coin": response.to_dict()["common_response"]["data"][0]["currency"], "product_id": response.to_dict()["common_response"]["data"][0]["id"], "rate": response.to_dict()["common_response"]["data"][0]["returnRate"], "min_amount": response.to_dict()["common_response"]["data"][0]["userLowerLimit"], "isActive": response.to_dict()["common_response"]["data"][0]["status"]=="ONGOING"} def subscribe_product(self, product_id, amount, auto_subscribe=False, source_account="SPOT"): ''' source_account: "TRADE" or "MAIN" auto_subscribe is ignored here response: {'common_response': {'code': '200000', 'data': {'orderId': '2987632', 'orderTxId': '6681886'}, 'rate_limit': {'limit': 2000, 'remaining': 1995, 'reset': 29901}}, 'orderId': '2987632', 'orderTxId': '6681886'} ''' if source_account.upper() in ["SPOT","TRADE","ALL"]: source="TRADE" elif source_account.upper() in ["FUND","MAIN"]: source="MAIN" #In Binance SPOT is TRADE and MAIN is FUND else: return {"Error": "Invalid source_account. Values should be TRADE or MAIN for kucoin, SPOT, FUND or ALL for binance"} request = PurchaseReqBuilder().set_product_id(product_id).set_amount(str(amount)).set_account_type(source).build() response = self.earn_api().purchase(request).to_dict() response_dict = response["common_response"]["data"] if response_dict!={}: return {"Success": "", "orderId": response["orderId"], "txId": response["orderTxId"], "amount": amount } else: return {"Error": response} def redeem_product(self, order_id, amount="0", source_account="SPOT"): ''' source_account is ignored unless orderID=ETH2: "TRADE" or "MAIN" auto_subscribe is ignored here returns {'common_response': {'code': '200000', 'data': {'orderTxId': '6681887', 'deliverTime': 1736027626000, 'status': 'SUCCESS', 'amount': '20'}, 'rate_limit': {'limit': 2000, 'remaining': 1990, 'reset': 9770}}, 'orderTxId': '6681887', 'deliverTime': 1736027626000, 'status': , 'amount': '20'} ''' if source_account in ["SPOT","TRADE","ALL"]: source="TRADE" elif source_account in ["FUND","MAIN"]: source="MAIN" #In Binance SPOT is TRADE and MAIN is FUND else: return {"Error": "Invalid source_account. Values should be TRADE or MAIN for kucoin, SPOT, FUND or ALL for binance"} request = RedeemReqBuilder().set_order_id(order_id).set_amount(str(amount)).set_from_account_type(source).build() response = self.earn_api().redeem(request).to_dict() response_dict = response["common_response"]["data"] if response_dict["status"]=="SUCCESS": return {"Success": "", "orderId": "", "txId": response["orderTxId"], "amount": response["amount"] } else: return {"Error": response} def get_position(self, coin): ''' Return {'common_response': {'code': '200000', 'data': {'totalNum': 1, 'items': [{'orderId': '2987632', 'productId': '2152', 'productCategory': 'DEMAND', 'productType': 'DEMAND', 'currency': 'USDT', 'incomeCurrency': 'USDT', 'returnRate': '0.04767484', 'holdAmount': '20', 'redeemedAmount': '0', 'redeemingAmount': '0', 'lockStartTime': 1641806718000, 'lockEndTime': None, 'purchaseTime': 1736027283000, 'redeemPeriod': 0, 'status': 'LOCKED', 'earlyRedeemSupported': 0}], 'currentPage': 1, 'pageSize': 15, 'totalPage': 1}, 'rate_limit': {'limit': 2000, 'remaining': 1995, 'reset': 16550}}, 'totalNum': 1, 'items': [{'orderId': '2987632', 'productId': '2152', 'productCategory': 'DEMAND', 'productType': 'DEMAND', 'currency': 'USDT', 'incomeCurrency': 'USDT', 'returnRate': '0.04767484', 'holdAmount': '20', 'redeemedAmount': '0', 'redeemingAmount': '0', 'lockStartTime': 1641806718000, 'purchaseTime': 1736027283000, 'redeemPeriod': 0, 'status': , 'earlyRedeemSupported': }], 'currentPage': 1, 'pageSize': 15, 'totalPage': 1} ''' request = GetAccountHoldingReqBuilder().set_currency(coin).build() response = self.earn_api().get_account_holding(request).to_dict() if len(response["items"])==0: return {} elif len(response["items"])==1: return {"asset": response["items"][0]["currency"], "positionId": response["items"][0]["orderId"], "productId": response["items"][0]["productId"], "amount": response["items"][0]["holdAmount"], "rate": response["items"][0]["returnRate"]} else: return {"Error": response} def get_account(self,coin,**kwargs): request = GetAccountHoldingReqBuilder().build() response = self.earn_api().get_account_holding(request) return response.to_dict() def get_personal_left_quota(self, *args, **kwargs): return {"Error": "N/A"} def get_subscription_record(self, **kwargs): return {"Error": "N/A"} def get_redemption_record(self, **kwargs): return {"Error": "N/A"} def get_rewards_history(self, type="ALL", **kwargs): return {"Error": "N/A"} def get_subscription_preview(self, product_id, amount, **kwargs): return {"Error": "N/A"}