fin-depo/fin_depo/defi_kucoin.py

"""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: <https://www.kucoin.com/docs/rest/spot-trading/orders/place-order>
        - GET Market Order by id: <https://www.kucoin.com/docs/rest/spot-trading/orders/get-order-details-by-orderid>
        """
        # 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)