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) @dataclasses.dataclass(frozen=True) class KucoinDepositAddresses(Scraper): dataset_name = 'defi_kucoin_deposit_address' deduplicate_mode = DeduplicateMode.ONLY_LATEST deduplicate_ignore_columns = ['account.update_time'] def scrape(self) -> Iterator[Mapping[str, object]]: data_point = {} addresses = client.get_deposit_address('MPC') data_point['account.num_deposit_addresses'] = len(addresses) data_point['account.update_time'] = datetime.datetime.now(tz=datetime.UTC) for k, v in addresses[-1].items(): data_point[f'deposit.{k}'] = v yield frozendict(data_point)