46 lines
1.5 KiB
Python
46 lines
1.5 KiB
Python
import datetime
|
|
import logging
|
|
from decimal import Decimal
|
|
|
|
import fin_defs
|
|
import kucoin.client
|
|
|
|
from .data import DepoGroup, DepoSingle
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class KucoinDepoFetcher:
|
|
def __init__(self, kucoin_key: str, kucoin_secret: str, kucoin_pass: str):
|
|
assert kucoin_key is not None, 'Missing kucoin_key'
|
|
assert kucoin_secret is not None, 'Missing kucoin_secret'
|
|
assert kucoin_pass is not None, 'Missing kucoin_pass'
|
|
self.client = kucoin.client.Client(
|
|
kucoin_key,
|
|
kucoin_secret,
|
|
kucoin_pass,
|
|
)
|
|
|
|
def get_depo(self) -> DepoGroup:
|
|
now = datetime.datetime.now(tz=datetime.UTC)
|
|
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()
|
|
],
|
|
)
|