1
0

Kraken account balance implemented
All checks were successful
Build container / Package-Python (push) Successful in 23s
Build container / Package-Container (push) Successful in 1m41s

This commit is contained in:
Jon Michael Aanes 2024-05-18 01:05:02 +02:00
parent 8a9f8bf5e8
commit 56a68b5119
Signed by: Jmaa
SSH Key Fingerprint: SHA256:Ab0GfHGCblESJx7JRE4fj4bFy/KRpeLhi41y4pF3sNA
4 changed files with 48 additions and 3 deletions

View File

@ -0,0 +1,44 @@
import dataclasses
import datetime
import logging
from collections.abc import Iterator, Mapping
from decimal import Decimal
import krakenex
from frozendict import frozendict
from personal_data.data import DeduplicateMode, Scraper
from .. import secrets
logger = logging.getLogger(__name__)
# TODO: Move these into secrets!
client = krakenex.API(
secrets.KRAKEN_KEY,
secrets.KRAKEN_SECRET,
)
@dataclasses.dataclass(frozen=True)
class KrakenAccountBalances(Scraper):
dataset_name = 'defi_kraken_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),
}
result = client.query_private('Balance')
for account, balance_str in result['result'].items():
key = f'balance.{account}'
balance = Decimal(balance_str)
balance = balance if balance != 0 else Decimal(0)
data_point[key] = data_point.get(key, Decimal(0)) + balance
del key, balance
yield frozendict(data_point)

View File

@ -14,9 +14,6 @@ from .. import secrets
logger = logging.getLogger(__name__)
HOSTNAME = 'api.kucoin.com'
URL_ACCOUNTS = 'https://{hostname}/api/v1/accounts'
# TODO: Move these into secrets!
client = kucoin.client.Client(
secrets.KUCOIN_KEY,

View File

@ -39,6 +39,10 @@ KUCOIN_KEY = load_secret('KUCOIN_KEY')
KUCOIN_SECRET = load_secret('KUCOIN_SECRET')
KUCOIN_PASS = load_secret('KUCOIN_PASS')
# KRAKEN
KRAKEN_KEY = load_secret('KRAKEN_KEY')
KRAKEN_SECRET = load_secret('KRAKEN_SECRET')
# Email configuration
MAILGUN_API_KEY = load_secret('MAILGUN_API_KEY')
MAILGUN_DOMAIN = load_secret('MAILGUN_DOMAIN')