From 88a57bbc6c39ce6351e56b9c89afebf3f33eb300 Mon Sep 17 00:00:00 2001 From: Jon Michael Aanes Date: Fri, 19 Jul 2024 00:22:29 +0200 Subject: [PATCH] Add unified DepoFetcher interface --- fin_depo/data.py | 21 ++++++++++++++++++ fin_depo/defi_kraken.py | 6 ++++-- fin_depo/defi_kucoin.py | 12 +++++------ fin_depo/defi_partisia_blockchain.py | 11 +++++----- fin_depo/investbank_nordnet.py | 32 +++++++++++----------------- 5 files changed, 49 insertions(+), 33 deletions(-) diff --git a/fin_depo/data.py b/fin_depo/data.py index fc3ca96..4d68124 100644 --- a/fin_depo/data.py +++ b/fin_depo/data.py @@ -3,6 +3,7 @@ import dataclasses import datetime from collections.abc import Iterable, Mapping from decimal import Decimal +from typing import TypeVar import enforce_typing from fin_defs import Asset @@ -69,3 +70,23 @@ class DepoGroup(Depo): summed += nested_depo.get_amount_of_asset(asset) del nested_depo return summed + +T = TypeVar('T') + +class DepoFetcher(abc.ABC): + """Base `Depo` fetcher interface. + + Used to get depository information for specific websites/backends. + """ + + @abc.abstractmethod + def get_depo(self) -> Depo: + """Fetches the `Depo`s available for the fetcher, possibly as + a `DepoGroup`. + """ + + def assert_param(self, param_name: str, param_type: type[T], param_value: T) -> T: + if not isinstance(param_value, param_type): + msg = f'{self} expected {param_name} parameter of type {param_type}, but got: {param_value}' + raise TypeError(msg) + return param_value diff --git a/fin_depo/defi_kraken.py b/fin_depo/defi_kraken.py index acb77c9..d5335bd 100644 --- a/fin_depo/defi_kraken.py +++ b/fin_depo/defi_kraken.py @@ -7,12 +7,12 @@ from decimal import Decimal import fin_defs import krakenex -from .data import Depo, DepoSingle +from .data import Depo, DepoSingle, DepoFetcher logger = logging.getLogger(__name__) -class KrakenDepoFetcher: +class KrakenDepoFetcher(DepoFetcher): """Depository fetcher for [Kraken](https://www.kraken.com), the online crypto currency exchange. Requirements for use: @@ -27,6 +27,8 @@ class KrakenDepoFetcher: """ def __init__(self, kraken_key: str, kraken_secret: str): + self.assert_param('kraken_key', str, kraken_key) + self.assert_param('kraken_secret', str, kraken_secret) assert kraken_key is not None, 'Missing kraken_key' assert kraken_secret is not None, 'Missing kraken_secret' self.client = krakenex.API( diff --git a/fin_depo/defi_kucoin.py b/fin_depo/defi_kucoin.py index 90c566c..5d6b976 100644 --- a/fin_depo/defi_kucoin.py +++ b/fin_depo/defi_kucoin.py @@ -7,13 +7,13 @@ from decimal import Decimal import fin_defs import kucoin.client -from .data import DepoGroup, DepoSingle +from .data import DepoGroup, DepoSingle, DepoFetcher logger = logging.getLogger(__name__) -class KucoinDepoFetcher: - """Depository fetcher for [Kucoin](https://www.kucoin.com), the online crypto currency exchange. +class KucoinDepoFetcher(DepoFetcher): + """`Depo` fetcher for [Kucoin](https://www.kucoin.com), the online crypto currency exchange. Requirements for use: - Account on [Kucoin](https://www.kucoin.com). @@ -27,9 +27,9 @@ 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.assert_param('kucoin_key', str, kucoin_key) + self.assert_param('kucoin_secret', str, kucoin_secret) + self.assert_param('kucoin_pass', str, kucoin_pass) self.client = kucoin.client.Client( kucoin_key, kucoin_secret, diff --git a/fin_depo/defi_partisia_blockchain.py b/fin_depo/defi_partisia_blockchain.py index 3883d01..09b7aa4 100644 --- a/fin_depo/defi_partisia_blockchain.py +++ b/fin_depo/defi_partisia_blockchain.py @@ -2,6 +2,7 @@ import dataclasses import datetime +import requests import email.utils import json import logging @@ -12,7 +13,7 @@ import fin_defs import requests from frozendict import frozendict -from .data import DepoGroup, DepoSingle +from .data import DepoGroup, DepoSingle, DepoFetcher logger = logging.getLogger(__name__) @@ -138,7 +139,7 @@ BYOC_ASSETS = { } -class PartisiaBlockchainAccountDepoFetcher: +class PartisiaBlockchainAccountDepoFetcher(DepoFetcher): """Depository fetcher for individual [Partisia Blockchain](https://partisiablockchain.com/) accounts, including [MPC](https://partisiablockchain.gitlab.io/documentation/pbc-fundamentals/mpc-token-model-and-account-elements.html) and [Bring-Your-Own-Coin](https://partisiablockchain.gitlab.io/documentation/pbc-fundamentals/byoc/introduction-to-byoc.html). @@ -150,9 +151,9 @@ class PartisiaBlockchainAccountDepoFetcher: depo. """ - def __init__(self, session, blockchain_address: str): - assert session is not None, 'missing session' - assert blockchain_address is not None, 'missing blockchain_address' + def __init__(self, session: requests.Session, blockchain_address: str): + self.assert_param('session', requests.Session, session) + self.assert_param('blockchain_address', str, blockchain_address) self.client = PbcClient(session) self.blockchain_address = blockchain_address diff --git a/fin_depo/investbank_nordnet.py b/fin_depo/investbank_nordnet.py index 7123606..8cadd6e 100644 --- a/fin_depo/investbank_nordnet.py +++ b/fin_depo/investbank_nordnet.py @@ -11,12 +11,13 @@ I am grateful for his pioneering work. import datetime import logging +import requests from decimal import Decimal import fin_defs from frozendict import frozendict -from .data import Depo, DepoGroup, DepoSingle +from .data import Depo, DepoGroup, DepoSingle, DepoFetcher logger = logging.getLogger(__name__) @@ -41,7 +42,7 @@ def asset_from_instrument_json(json) -> fin_defs.Asset | None: EMPTY_DICT: dict[str,str | int] = frozendict() -class NordnetDepoFetcher: +class NordnetDepoFetcher(DepoFetcher): """Depository fetcher for [Nordnet](https://nordnet.dk), the online investment bank. Requirements for use: @@ -57,23 +58,14 @@ class NordnetDepoFetcher: `Depo`, with all of them nested under a "Nordnet" nested depository. """ - def __init__(self, session, username: str, password: str): - self.session = session - self.username = username - self.password = password + def __init__(self, session: requests.Session, username: str, password: str): + + self.session = self.assert_param('session', requests.Session, session) + self.username: str = self.assert_param('username', str, username) + self.password: str = self.assert_param('password', str, password) self.is_logged_in = False - if self.session is None: - msg = 'Session argument is missing' - raise TypeError(msg) - if self.username is not None: - msg = 'Username argument s missing' - raise TypeError(msg) - if self.password is not None: - msg = 'Password argument is missing' - raise TypeError(msg) - - def get_json(self, url: str, params: dict[str, str | int] = EMPTY_DICT)-> dict: + def _get_json(self, url: str, params: dict[str, str | int] = EMPTY_DICT)-> dict: if not url.startswith(API_ROOT): msg = f'Given url must be located below API ROOT: {url}' raise ValueError(msg) @@ -115,7 +107,7 @@ class NordnetDepoFetcher: tz=datetime.UTC, ) # TODO: Use info from json requests - json_accounts = self.get_json(API_ACCOUNTS) + json_accounts = self._get_json(API_ACCOUNTS) nested: list[Depo] = [] for json_account in json_accounts: @@ -125,7 +117,7 @@ class NordnetDepoFetcher: # Determine amount of usable currency stored on Nordnet if True: - json_account_info = self.get_json( + json_account_info = self._get_json( API_ACCOUNT_INFO.format(accid=account_id), ) asset = fin_defs.FiatCurrency( @@ -136,7 +128,7 @@ class NordnetDepoFetcher: del asset, amount # Determine positions on Nordnet - json_account_positions = self.get_json( + json_account_positions = self._get_json( API_ACCOUNT_POSITIONS.format(accid=account_id), ) for pos in json_account_positions: