fin-depo/fin_depo/defi_kucoin.py

"""See `KucoinDepoFetcher` for documentation."""

import datetime
import logging
from decimal import Decimal

import fin_defs
import kucoin.client

from .data import DepoFetcher, DepoGroup, DepoSingle

logger = logging.getLogger(__name__)


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**, and **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):
        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.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.client.get_accounts():
            asset = fin_defs.WELL_KNOWN_SYMBOLS[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()
            ],
        )