import dataclasses import datetime import logging from collections.abc import Iterator, Mapping from decimal import Decimal import kucoin.client from frozendict import frozendict from personal_data.data import DeduplicateMode, Scraper from .. import secrets logger = logging.getLogger(__name__) # TODO: Move these into secrets! client = kucoin.client.Client( secrets.KUCOIN_KEY, secrets.KUCOIN_SECRET, secrets.KUCOIN_PASS, ) @dataclasses.dataclass(frozen=True) class KucoinAccountBalances(Scraper): dataset_name = 'defi_kucoin_balance' deduplicate_mode = DeduplicateMode.ONLY_LATEST deduplicate_ignore_columns = ['account.update_time'] def scrape(self) -> Iterator[Mapping[str, object]]: data_point = { 'account.update_time': datetime.datetime.now(tz=datetime.UTC), } for account in client.get_accounts(): key = f"balance.{account['currency']}" data_point[key] = data_point.get(key, Decimal(0)) + Decimal( account['balance'], ) yield frozendict(data_point)