1
0

Add unified DepoFetcher interface

This commit is contained in:
Jon Michael Aanes 2024-07-19 00:22:29 +02:00
parent 4e8214a948
commit 88a57bbc6c
Signed by: Jmaa
SSH Key Fingerprint: SHA256:Ab0GfHGCblESJx7JRE4fj4bFy/KRpeLhi41y4pF3sNA
5 changed files with 49 additions and 33 deletions

View File

@ -3,6 +3,7 @@ import dataclasses
import datetime import datetime
from collections.abc import Iterable, Mapping from collections.abc import Iterable, Mapping
from decimal import Decimal from decimal import Decimal
from typing import TypeVar
import enforce_typing import enforce_typing
from fin_defs import Asset from fin_defs import Asset
@ -69,3 +70,23 @@ class DepoGroup(Depo):
summed += nested_depo.get_amount_of_asset(asset) summed += nested_depo.get_amount_of_asset(asset)
del nested_depo del nested_depo
return summed 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

View File

@ -7,12 +7,12 @@ from decimal import Decimal
import fin_defs import fin_defs
import krakenex import krakenex
from .data import Depo, DepoSingle from .data import Depo, DepoSingle, DepoFetcher
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
class KrakenDepoFetcher: class KrakenDepoFetcher(DepoFetcher):
"""Depository fetcher for [Kraken](https://www.kraken.com), the online crypto currency exchange. """Depository fetcher for [Kraken](https://www.kraken.com), the online crypto currency exchange.
Requirements for use: Requirements for use:
@ -27,6 +27,8 @@ class KrakenDepoFetcher:
""" """
def __init__(self, kraken_key: str, kraken_secret: str): 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_key is not None, 'Missing kraken_key'
assert kraken_secret is not None, 'Missing kraken_secret' assert kraken_secret is not None, 'Missing kraken_secret'
self.client = krakenex.API( self.client = krakenex.API(

View File

@ -7,13 +7,13 @@ from decimal import Decimal
import fin_defs import fin_defs
import kucoin.client import kucoin.client
from .data import DepoGroup, DepoSingle from .data import DepoGroup, DepoSingle, DepoFetcher
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
class KucoinDepoFetcher: class KucoinDepoFetcher(DepoFetcher):
"""Depository fetcher for [Kucoin](https://www.kucoin.com), the online crypto currency exchange. """`Depo` fetcher for [Kucoin](https://www.kucoin.com), the online crypto currency exchange.
Requirements for use: Requirements for use:
- Account on [Kucoin](https://www.kucoin.com). - 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): def __init__(self, kucoin_key: str, kucoin_secret: str, kucoin_pass: str):
assert kucoin_key is not None, 'Missing kucoin_key' self.assert_param('kucoin_key', str, kucoin_key)
assert kucoin_secret is not None, 'Missing kucoin_secret' self.assert_param('kucoin_secret', str, kucoin_secret)
assert kucoin_pass is not None, 'Missing kucoin_pass' self.assert_param('kucoin_pass', str, kucoin_pass)
self.client = kucoin.client.Client( self.client = kucoin.client.Client(
kucoin_key, kucoin_key,
kucoin_secret, kucoin_secret,

View File

@ -2,6 +2,7 @@
import dataclasses import dataclasses
import datetime import datetime
import requests
import email.utils import email.utils
import json import json
import logging import logging
@ -12,7 +13,7 @@ import fin_defs
import requests import requests
from frozendict import frozendict from frozendict import frozendict
from .data import DepoGroup, DepoSingle from .data import DepoGroup, DepoSingle, DepoFetcher
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@ -138,7 +139,7 @@ BYOC_ASSETS = {
} }
class PartisiaBlockchainAccountDepoFetcher: class PartisiaBlockchainAccountDepoFetcher(DepoFetcher):
"""Depository fetcher for individual [Partisia """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 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). [Bring-Your-Own-Coin](https://partisiablockchain.gitlab.io/documentation/pbc-fundamentals/byoc/introduction-to-byoc.html).
@ -150,9 +151,9 @@ class PartisiaBlockchainAccountDepoFetcher:
depo. depo.
""" """
def __init__(self, session, blockchain_address: str): def __init__(self, session: requests.Session, blockchain_address: str):
assert session is not None, 'missing session' self.assert_param('session', requests.Session, session)
assert blockchain_address is not None, 'missing blockchain_address' self.assert_param('blockchain_address', str, blockchain_address)
self.client = PbcClient(session) self.client = PbcClient(session)
self.blockchain_address = blockchain_address self.blockchain_address = blockchain_address

View File

@ -11,12 +11,13 @@ I am grateful for his pioneering work.
import datetime import datetime
import logging import logging
import requests
from decimal import Decimal from decimal import Decimal
import fin_defs import fin_defs
from frozendict import frozendict from frozendict import frozendict
from .data import Depo, DepoGroup, DepoSingle from .data import Depo, DepoGroup, DepoSingle, DepoFetcher
logger = logging.getLogger(__name__) 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() EMPTY_DICT: dict[str,str | int] = frozendict()
class NordnetDepoFetcher: class NordnetDepoFetcher(DepoFetcher):
"""Depository fetcher for [Nordnet](https://nordnet.dk), the online investment bank. """Depository fetcher for [Nordnet](https://nordnet.dk), the online investment bank.
Requirements for use: Requirements for use:
@ -57,23 +58,14 @@ class NordnetDepoFetcher:
`Depo`, with all of them nested under a "Nordnet" nested depository. `Depo`, with all of them nested under a "Nordnet" nested depository.
""" """
def __init__(self, session, username: str, password: str): def __init__(self, session: requests.Session, username: str, password: str):
self.session = session
self.username = username self.session = self.assert_param('session', requests.Session, session)
self.password = password self.username: str = self.assert_param('username', str, username)
self.password: str = self.assert_param('password', str, password)
self.is_logged_in = False self.is_logged_in = False
if self.session is None: def _get_json(self, url: str, params: dict[str, str | int] = EMPTY_DICT)-> dict:
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:
if not url.startswith(API_ROOT): if not url.startswith(API_ROOT):
msg = f'Given url must be located below API ROOT: {url}' msg = f'Given url must be located below API ROOT: {url}'
raise ValueError(msg) raise ValueError(msg)
@ -115,7 +107,7 @@ class NordnetDepoFetcher:
tz=datetime.UTC, tz=datetime.UTC,
) # TODO: Use info from json requests ) # TODO: Use info from json requests
json_accounts = self.get_json(API_ACCOUNTS) json_accounts = self._get_json(API_ACCOUNTS)
nested: list[Depo] = [] nested: list[Depo] = []
for json_account in json_accounts: for json_account in json_accounts:
@ -125,7 +117,7 @@ class NordnetDepoFetcher:
# Determine amount of usable currency stored on Nordnet # Determine amount of usable currency stored on Nordnet
if True: if True:
json_account_info = self.get_json( json_account_info = self._get_json(
API_ACCOUNT_INFO.format(accid=account_id), API_ACCOUNT_INFO.format(accid=account_id),
) )
asset = fin_defs.FiatCurrency( asset = fin_defs.FiatCurrency(
@ -136,7 +128,7 @@ class NordnetDepoFetcher:
del asset, amount del asset, amount
# Determine positions on Nordnet # Determine positions on Nordnet
json_account_positions = self.get_json( json_account_positions = self._get_json(
API_ACCOUNT_POSITIONS.format(accid=account_id), API_ACCOUNT_POSITIONS.format(accid=account_id),
) )
for pos in json_account_positions: for pos in json_account_positions: