"""See `KucoinDepoFetcher` for documentation.""" import datetime import logging import time from decimal import Decimal import fin_defs import kucoin.client from .data import (DepoFetcher, DepoGroup, DepoSingle, TradeOrderDetails, WithdrawalDetails, DepositDetails) logger = logging.getLogger(__name__) def parse_asset_from_ticker(ticker: str) -> fin_defs.Asset: if ticker == 'KCS': return fin_defs.CryptoCurrency('KCS', coingecko_id='kucoin-shares') return fin_defs.WELL_KNOWN_SYMBOLS[ticker] class KucoinDepoFetcher(DepoFetcher): """`Depo` fetcher for [Kucoin](https://www.kucoin.com), the online crypto currency exchange. Requirements for use: - Account on [Kucoin](https://www.kucoin.com). - Have performed Know Your Customer (KYC) for your account. - Created API key from [settings menu](https://www.kucoin.com/account/api). API key must have the **General Permission**. Unless you are using the `place_market_order` functionality, the API key **should not have any additional permissions**. Employ principle of least priviledge. - Install [`python-kucoin`](https://python-kucoin.readthedocs.io/en/latest/) library. Depository structure: An upper level depo split for each of the sub-accounts (Funding, Trading, Margin, Futures...) """ def __init__( self, kucoin_key: str, kucoin_secret: str, kucoin_pass: str, allow_trades: bool = False, ): self.assert_param('kucoin_key', str, kucoin_key) self.assert_param('kucoin_secret', str, kucoin_secret) self.assert_param('kucoin_pass', str, kucoin_pass) self.allow_trades = allow_trades self.kucoin_client = kucoin.client.Client( kucoin_key, kucoin_secret, kucoin_pass, ) def get_depo(self) -> DepoGroup: # We would ideally get timestamp from request, # but this is fine for now. now = datetime.datetime.now(tz=datetime.UTC) # Assets are spread across account types, but we would like them # clustered into different depos. assets_by_account_type: dict[str, dict[fin_defs.Asset, Decimal]] = {} for account_data in self.kucoin_client.get_accounts(): asset = parse_asset_from_ticker(account_data['currency']) balance = Decimal(account_data['balance']) assets_for_account_type = assets_by_account_type.setdefault( account_data['type'], {}, ) assets_for_account_type[asset] = ( assets_for_account_type.get(asset, Decimal(0)) + balance ) del account_data, asset, balance, assets_for_account_type return DepoGroup( 'Kucoin', now, [ DepoSingle('Kucoin ' + account_type, now, assets) for account_type, assets in assets_by_account_type.items() ], ) def place_market_order( self, input_asset: fin_defs.Asset, input_amount: Decimal, output_asset: fin_defs.Asset, ) -> TradeOrderDetails: """Executes a market order through the Kucoin market. Will automatically determine the market based on the input and output assets. Requirements: - Fetcher must have been created with `allow_trades=True`. - API key used with fetcher must have **Spot Trading** permissions. - Assets must be on trading account. Assets on funding accounts or other accounts cannot be used. Note: - A fee will be paid to Kucoin, with the rate determined by your VIP level and the asset being traded. - The full `input_amount` may not be used. Inspect the resulting `TradeOrderDetails` to see how much of the `input_amount` have been used. References: - POST Market Order: - GET Market Order by id: """ # Check requirements if not self.allow_trades: msg = 'KucoinDepoFetcher.allow_trades is not enabled: Cannot make trades' raise PermissionError(msg) if fin_defs.USDT not in [input_asset, output_asset]: msg = 'Non-USDT Markets are not supported' raise NotImplementedError(msg) # Convert arguments to kucoin client arguments if input_asset == fin_defs.USDT: symbol: str = ( f'{output_asset.raw_short_name()}-{input_asset.raw_short_name()}' ) side: str = 'buy' size = None funds = str(input_amount) else: symbol = f'{input_asset.raw_short_name()}-{output_asset.raw_short_name()}' side = 'sell' size = str(input_amount) funds = None # Place order logger.info('Placing order: %s', str([symbol, side, size, funds])) response = self.kucoin_client.create_market_order( symbol=symbol, side=side, size=size, funds=funds, ) del symbol, side, size, funds, input_amount # Determine order details return self._get_order_details(response['orderId'], input_asset, output_asset) def _get_withdrawals(self) -> list[WithdrawalDetails]: raw_details = self.kucoin_client.get_withdrawals() withdrawals = [] for item in raw_details['items']: withdrawn_asset = parse_asset_from_ticker(item['currency']) withdrawn_amount = Decimal(item['amount']) fee_asset = withdrawn_asset # Assumed fee_amount = Decimal(item['fee']) executed_time=datetime.datetime.fromtimestamp( item['createdAt'] / 1000, tz=datetime.UTC, ) withdrawals.append(WithdrawalDetails(withdrawn_asset, withdrawn_amount, fee_asset, fee_amount, executed_time, item)) del item return withdrawals def _get_deposits(self) -> list[WithdrawalDetails]: raw_details = self.kucoin_client.get_deposits() deposits = [] for item in raw_details['items']: deposit_asset = parse_asset_from_ticker(item['currency']) deposit_amount = Decimal(item['amount']) fee_asset = deposit_asset # Assumed fee_amount = Decimal(item['fee']) executed_time=datetime.datetime.fromtimestamp( item['createdAt'] / 1000, tz=datetime.UTC, ) deposits.append(DepositDetails(deposit_asset, deposit_amount, fee_asset, fee_amount, executed_time, item)) del item return deposits def _get_order_details( self, order_id: str, input_asset: fin_defs.Asset, output_asset: fin_defs.Asset, ) -> TradeOrderDetails: """Determine the order details for the order with the given id. Retries the order a few times, as KuCoin might not have propagated the order through their systems. """ order_details = self._get_order_with_retries(order_id, num_retries=10) # Convert from kucoin results if input_asset == fin_defs.USDT: input_amount_final = Decimal(order_details['dealFunds']) output_amount_final = Decimal(order_details['dealSize']) else: output_amount_final = Decimal(order_details['dealFunds']) input_amount_final = Decimal(order_details['dealSize']) return TradeOrderDetails( input_asset=input_asset, input_amount=input_amount_final, output_asset=output_asset, output_amount=output_amount_final, fee_asset=parse_asset_from_ticker(order_details['feeCurrency']), fee_amount=Decimal(order_details['fee']), executed_time=datetime.datetime.fromtimestamp( order_details['createdAt'] / 1000, tz=datetime.UTC, ), order_id=order_id, raw_order_details=order_details, ) def _get_order_with_retries( self, order_id: str, *, num_retries: int, sleep_between_tries: float = 1.0, ) -> dict: """Get the order details from KuCoin backend. Retries the order a few times, as KuCoin might not have propagated the order through their systems since it was sent. """ for _ in range(num_retries): try: return self.kucoin_client.get_order(order_id) except kucoin.exceptions.KucoinAPIException: # noqa: PERF203 time.sleep(sleep_between_tries) return self.kucoin_client.get_order(order_id)