fin_depo/data.py

@@ -3,7 +3,6 @@ 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
@@ -70,25 +69,3 @@ 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

fin_depo/defi_kraken.py

@@ -7,12 +7,12 @@ from decimal import Decimal
 import fin_defs
 import krakenex
 
-from .data import Depo, DepoFetcher, DepoSingle
+from .data import Depo, DepoSingle
 
 logger = logging.getLogger(__name__)
 
 
-class KrakenDepoFetcher(DepoFetcher):
+class KrakenDepoFetcher:
     """Depository fetcher for [Kraken](https://www.kraken.com), the online crypto currency exchange.
 
     Requirements for use:
@@ -22,14 +22,13 @@ class KrakenDepoFetcher(DepoFetcher):
       settings](https://pro.kraken.com/app/settings/api).
       API key must have the **Query Funds Permission**, and **should not have
       any additional permissions**. Employ principle of least priviledge.
-    - Install [`krakenex`](https://pypi.org/project/krakenex/) library.
 
     Depository structure: A `DepoSingle`. No nesting.
     """
 
     def __init__(self, kraken_key: str, kraken_secret: str):
-        self.assert_param('kraken_key', str, kraken_key)
+        assert kraken_key is not None, 'Missing kraken_key'
-        self.assert_param('kraken_secret', str, kraken_secret)
+        assert kraken_secret is not None, 'Missing kraken_secret'
         self.client = krakenex.API(
             kraken_key,
             kraken_secret,
@@ -41,13 +40,13 @@ class KrakenDepoFetcher(DepoFetcher):
         result = self.client.query_private('Balance')
 
         assets: dict[fin_defs.Asset, Decimal] = {}
-        for account_raw, balance_str in result['result'].items():
+        for account, balance_str in result['result'].items():
-            account = account_raw.removesuffix('.HOLD')
+            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, account_raw, balance_str, asset, balance
+            del account, balance_str, asset, balance
 
         return DepoSingle(
             name='Kraken',

fin_depo/defi_kucoin.py

@@ -7,13 +7,13 @@ from decimal import Decimal
 import fin_defs
 import kucoin.client
 
-from .data import DepoFetcher, DepoGroup, DepoSingle
+from .data import DepoGroup, DepoSingle
 
 logger = logging.getLogger(__name__)
 
 
-class KucoinDepoFetcher(DepoFetcher):
+class KucoinDepoFetcher:
-    """`Depo` fetcher for [Kucoin](https://www.kucoin.com), the online crypto currency exchange.
+    """Depository fetcher for [Kucoin](https://www.kucoin.com), the online crypto currency exchange.
 
     Requirements for use:
     - Account on [Kucoin](https://www.kucoin.com).
@@ -21,16 +21,15 @@ class KucoinDepoFetcher(DepoFetcher):
     - Created API key from [settings menu](https://www.kucoin.com/account/api).
       API key must have the **General Permission**, and **should not have
       any additional permissions**. Employ principle of least priviledge.
-    - Install [`python-kucoin`](https://python-kucoin.readthedocs.io/en/latest/) library.
 
     Depository structure: An upper level depo split for each of the
     sub-accounts (Funding, Trading, Margin, Futures...)
     """
 
     def __init__(self, kucoin_key: str, kucoin_secret: str, kucoin_pass: str):
-        self.assert_param('kucoin_key', str, kucoin_key)
+        assert kucoin_key is not None, 'Missing kucoin_key'
-        self.assert_param('kucoin_secret', str, kucoin_secret)
+        assert kucoin_secret is not None, 'Missing kucoin_secret'
-        self.assert_param('kucoin_pass', str, kucoin_pass)
+        assert kucoin_pass is not None, 'Missing kucoin_pass'
         self.client = kucoin.client.Client(
             kucoin_key,
             kucoin_secret,

fin_depo/defi_partisia_blockchain.py

@@ -12,7 +12,7 @@ import fin_defs
 import requests
 from frozendict import frozendict
 
-from .data import DepoFetcher, DepoGroup, DepoSingle
+from .data import DepoGroup, DepoSingle
 
 logger = logging.getLogger(__name__)
 
@@ -138,7 +138,7 @@ BYOC_ASSETS = {
 }
 
 
-class PartisiaBlockchainAccountDepoFetcher(DepoFetcher):
+class PartisiaBlockchainAccountDepoFetcher:
     """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 +150,9 @@ class PartisiaBlockchainAccountDepoFetcher(DepoFetcher):
     depo.
     """
 
-    def __init__(self, session: requests.Session, blockchain_address: str):
+    def __init__(self, session, blockchain_address: str):
-        self.assert_param('session', requests.Session, session)
+        assert session is not None, 'missing session'
-        self.assert_param('blockchain_address', str, blockchain_address)
+        assert blockchain_address is not None, 'missing blockchain_address'
 
         self.client = PbcClient(session)
         self.blockchain_address = blockchain_address
@@ -166,7 +166,7 @@ class PartisiaBlockchainAccountDepoFetcher(DepoFetcher):
             {fin_defs.MPC: balances.mpc},
         )
         depo_byoc = DepoSingle(
-            'Bring Your Own Coin',
+            'BYOC',
             balances.update_time,
             {BYOC_ASSETS[k]: v for k, v in balances.byoc.items()},
         )

fin_depo/investbank_nordnet.py

@@ -14,10 +14,9 @@ import logging
 from decimal import Decimal
 
 import fin_defs
-import requests
 from frozendict import frozendict
 
-from .data import Depo, DepoFetcher, DepoGroup, DepoSingle
+from .data import Depo, DepoGroup, DepoSingle
 
 logger = logging.getLogger(__name__)
 
@@ -40,11 +39,9 @@ def asset_from_instrument_json(json) -> fin_defs.Asset | None:
     exchange_id = json['tradables'][0]['mic']
     return fin_defs.Stock(symbol, fin_defs.EXCHANGES_BY_IDS[exchange_id])
 
-
 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:
@@ -60,13 +57,23 @@ class NordnetDepoFetcher(DepoFetcher):
     `Depo`, with all of them nested under a "Nordnet" nested depository.
     """
 
-    def __init__(self, session: requests.Session, username: str, password: str):
+    def __init__(self, session, username: str, password: str):
-        self.session = self.assert_param('session', requests.Session, session)
+        self.session = session
-        self.username: str = self.assert_param('username', str, username)
+        self.username = username
-        self.password: str = self.assert_param('password', str, password)
+        self.password = password
         self.is_logged_in = False
 
-    def _get_json(self, url: str, params: dict[str, str | int] = EMPTY_DICT) -> dict:
+        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:
         if not url.startswith(API_ROOT):
             msg = f'Given url must be located below API ROOT: {url}'
             raise ValueError(msg)
@@ -108,7 +115,7 @@ class NordnetDepoFetcher(DepoFetcher):
             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:
@@ -118,7 +125,7 @@ class NordnetDepoFetcher(DepoFetcher):
 
             # 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(
@@ -129,7 +136,7 @@ class NordnetDepoFetcher(DepoFetcher):
                 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:

test/test_nordnet.py

@@ -14,11 +14,11 @@ needs_secrets = pytest.mark.skipif(
 @needs_secrets
 def test_get_depo():
     session = requests.Session()
-    fetcher = investbank_nordnet.NordnetDepoFetcher(
+    nordnet = investbank_nordnet.NordnetDepoFetcher(
         session,
         secrets.NORDNET_USERNAME,
         secrets.NORDNET_PASSWORD,
     )
 
-    depo = fetcher.get_depo()
+    depo = nordnet.get_depo()
     assert depo is not None

test/test_partisia_blockchain.py

@@ -1,23 +0,0 @@
-import fin_defs
-import requests
-
-from fin_depo import defi_partisia_blockchain
-
-
-def test_get_depo():
-    session = requests.Session()
-    fetcher = defi_partisia_blockchain.PartisiaBlockchainAccountDepoFetcher(
-        session,
-        '00f3a778fbcf663e4f0d47b41aa4ec51e0f9d39d60',
-    )
-
-    depo = fetcher.get_depo()
-    assert depo is not None
-    assert depo.name == 'Partisia Blockchain'
-    assert depo.updated_time is not None
-    assert depo.nested[0].name == 'Native'
-    assert depo.nested[0].updated_time == depo.updated_time
-    assert depo.nested[1].name == 'Bring Your Own Coin'
-    assert depo.nested[1].updated_time == depo.updated_time
-
-    assert depo.get_amount_of_asset(fin_defs.MPC)