69 lines
2.5 KiB
Python
69 lines
2.5 KiB
Python
"""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()
|
|
],
|
|
)
|