41 lines
1.1 KiB
Python
41 lines
1.1 KiB
Python
import datetime
|
|
import logging
|
|
from decimal import Decimal
|
|
|
|
import fin_defs
|
|
import krakenex
|
|
|
|
from .data import Depo, DepoSingle
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class KrakenDepoFetcher:
|
|
def __init__(self, kraken_key: str, kraken_secret: str):
|
|
assert kraken_key is not None, 'Missing kraken_key'
|
|
assert kraken_secret is not None, 'Missing kraken_secret'
|
|
self.client = krakenex.API(
|
|
kraken_key,
|
|
kraken_secret,
|
|
)
|
|
|
|
def get_depo(self) -> Depo:
|
|
now = datetime.datetime.now(tz=datetime.UTC)
|
|
|
|
result = self.client.query_private('Balance')
|
|
|
|
assets: dict[fin_defs.Asset, Decimal] = {}
|
|
for account, balance_str in result['result'].items():
|
|
account = account.removesuffix('.HOLD')
|
|
asset = fin_defs.WELL_KNOWN_SYMBOLS[account]
|
|
balance = Decimal(balance_str)
|
|
if balance != 0:
|
|
assets[asset] = assets.get(asset, Decimal(0)) + balance
|
|
del account, balance_str, asset, balance
|
|
|
|
return DepoSingle(
|
|
name='Kraken',
|
|
_assets=assets,
|
|
updated_time=now,
|
|
)
|