"""See `KucoinDepoFetcher` for documentation.""" import datetime import logging import time from collections.abc import Iterator from decimal import Decimal import fin_defs import kucoin.client from fin_defs import AssetAmount from .data import ( DepoFetcher, DepoGroup, DepoSingle, DepositDetails, DoubleRegister, TradeOrderDetails, WithdrawalDetails, ) 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] def order_from_json(order_details: dict): (asset, base_asset) = [ parse_asset_from_ticker(a) for a in order_details['symbol'].split('-') ] # Convert from kucoin results if order_details['side'] == 'buy': input_asset, output_asset = base_asset, asset input_amount_final = Decimal(order_details['dealFunds']) output_amount_final = Decimal(order_details['dealSize']) else: input_asset, output_asset = asset, base_asset output_amount_final = Decimal(order_details['dealFunds']) input_amount_final = Decimal(order_details['dealSize']) fee_asset = parse_asset_from_ticker(order_details['feeCurrency']) return TradeOrderDetails( input=AssetAmount(input_asset, input_amount_final), output=AssetAmount(output_asset, output_amount_final), fee=AssetAmount(fee_asset, Decimal(order_details['fee'])), executed_time=datetime.datetime.fromtimestamp( order_details['createdAt'] / 1000, tz=datetime.UTC, ), order_id=order_details['id'], raw_order_details=order_details, ) 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_: fin_defs.AssetAmount, 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_ # Determine order details return self._get_order_details(response['orderId']) 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( AssetAmount(withdrawn_asset, withdrawn_amount), AssetAmount(fee_asset, fee_amount), executed_time, item, ), ) del item return withdrawals def _get_deposits(self) -> list[DepositDetails]: 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( AssetAmount(deposit_asset, deposit_amount), AssetAmount(fee_asset, fee_amount), executed_time, item, ), ) del item return deposits def _get_historic_spot_orders(self) -> Iterator[TradeOrderDetails]: end_time = datetime.datetime.now(tz=datetime.UTC) for _weeks_back in range(20): end_time = end_time - datetime.timedelta(days=7) timestamp = int(end_time.timestamp() * 1000) raw_details = self.kucoin_client.get_orders(end=timestamp) yield from (order_from_json(item) for item in raw_details['items']) del _weeks_back, raw_details def _get_double_registers(self) -> list[DoubleRegister]: double_registers: list[DoubleRegister] = [] double_registers += self._get_deposits() double_registers += self._get_withdrawals() double_registers += self._get_historic_spot_orders() double_registers.sort(key=lambda x: x.executed_time) return double_registers def _get_order_details( self, order_id: str, ) -> 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) return order_from_json(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)