fin_depo/__init__.py

@@ -43,27 +43,20 @@ implementing this functionality. Exposes the same data as the home page.
 
 - [ ] Investment Bank: Saxo Bank OpenAPI
 - [ ] Personal Bank: Personal Bank Account (Open Banking) Maybe use AIIA?
-- [ ] Partisia Blockchain: Implement sharding routing correctly.
 """
 
-BACKEND_MODULE_NAMES = [
-    'defi_kraken',
-    'defi_kucoin',
-    'investbank_nordnet',
-    'defi_partisia_blockchain',
-]
-
 __all__ = [
     'defi_kraken',
     'defi_kucoin',
-    'investbank_nordnet',
     'defi_partisia_blockchain',
+    'investbank_nordnet',
     '__version__',
     'data',
     'static',
 ]
 
 import importlib
+import inspect
 import logging
 import sys
 
@@ -75,12 +68,10 @@ from ._version import __version__
 
 logger = logging.getLogger(__name__)
 
+CURRENT_MODULE = sys.modules[__name__]
 
-def load_backend(root_module, name: str) -> object | None:
-    """Initializes the backend with the given name.
 
-    Module is assigned as an attribute to the root module.
-    """
+def load_backend(name: str) -> object | None:
     try:
         imported_module = importlib.import_module(f'{__name__}.{name}')
     except Exception:
@@ -89,12 +80,19 @@ def load_backend(root_module, name: str) -> object | None:
             name,
         )
         return None
-    setattr(root_module, name, imported_module)
+    setattr(CURRENT_MODULE, name, imported_module)
     return imported_module
 
 
 # Import modules
 
-for m in BACKEND_MODULE_NAMES:
-    load_backend(sys.modules[__name__], m)
+backend_modules = [
+    'defi_kraken',
+    'defi_kucoin',
+    'investbank_nordnet',
+    'defi_partisia_blockchain',
+]
+for m in backend_modules:
+    load_backend(m)
     del m
+del backend_modules, CURRENT_MODULE

fin_depo/data.py

@@ -45,9 +45,7 @@ class DepoSingle(Depo):
     _assets: Mapping[Asset, Decimal]
 
     def __post_init__(self):
-        if None in self._assets:
-            msg = 'DepoSingle must not containg a None Asset key'
-            raise ValueError(msg)
+        assert None not in self._assets
 
     def assets(self) -> Iterable[Asset]:
         return self._assets
@@ -64,9 +62,7 @@ class DepoGroup(Depo):
     nested: list[Depo]
 
     def __post_init__(self):
-        if None in self.nested:
-            msg = 'DepoGroup must not containg an None depository'
-            raise ValueError(msg)
+        assert None not in self.nested
 
     def assets(self) -> Iterable[Asset]:
         assets: set[Asset] = set()

fin_depo/defi_kraken.py

@@ -55,7 +55,6 @@ class KrakenDepoFetcher(DepoFetcher):
             updated_time=now,
         )
 
-
 def parse_asset_from_ticker(ticker: str) -> fin_defs.Asset:
     account = ticker.removesuffix('.HOLD')
     if account == 'ZEUR':

fin_depo/defi_kucoin.py

@@ -12,7 +12,6 @@ from .data import DepoFetcher, DepoGroup, DepoSingle, TradeOrderDetails
 
 logger = logging.getLogger(__name__)
 
-
 def parse_asset_from_ticker(ticker: str) -> fin_defs.Asset:
     if ticker == 'KCS':
         return fin_defs.CryptoCurrency('KCS', coingecko_id='kucoin-shares')
@@ -94,19 +93,22 @@ class KucoinDepoFetcher(DepoFetcher):
         assets.
 
         Requirements:
+
         - Fetcher must have been created with `allow_trades=True`.
         - API key used with fetcher must have **Spot Trading** permissions.
         - Assets must be on trading account. Assets on funding accounts or
           other accounts cannot be used.
 
         Note:
-        - A fee will be paid to Kucoin, with the rate determined by your VIP
+
+        - A fee will be paid to Kucoin, with the rate determined by your WIP
           level and the asset being traded.
         - The full `input_amount` may not be used. Inspect the resulting
           `TradeOrderDetails` to see how much of the `input_amount` have been
           used.
 
         References:
+
         - POST Market Order: <https://www.kucoin.com/docs/rest/spot-trading/orders/place-order>
         - GET Market Order by id: <https://www.kucoin.com/docs/rest/spot-trading/orders/get-order-details-by-orderid>
         """
@@ -114,16 +116,11 @@ class KucoinDepoFetcher(DepoFetcher):
         if not self.allow_trades:
             msg = 'KucoinDepoFetcher.allow_trades is not enabled: Cannot make trades'
             raise PermissionError(msg)
-
-        if fin_defs.USDT not in [input_asset, output_asset]:
-            msg = 'Non-USDT Markets are not supported'
-            raise NotImplementedError(msg)
+        assert fin_defs.USDT in [input_asset, output_asset], 'USDT markets only for now'
 
         # Convert arguments to kucoin client arguments
         if input_asset == fin_defs.USDT:
-            symbol: str = (
-                f'{output_asset.raw_short_name()}-{input_asset.raw_short_name()}'
-            )
+            symbol: str = f'{output_asset.raw_short_name()}-{input_asset.raw_short_name()}'
             side: str = 'buy'
             size = None
             funds = str(input_amount)
@@ -197,6 +194,6 @@ class KucoinDepoFetcher(DepoFetcher):
         for _ in range(num_retries):
             try:
                 return self.kucoin_client.get_order(order_id)
-            except kucoin.exceptions.KucoinAPIException:  # noqa: PERF203
+            except kucoin.exceptions.KucoinAPIException as e:  # noqa
                 time.sleep(sleep_between_tries)
         return self.kucoin_client.get_order(order_id)

fin_depo/defi_partisia_blockchain.py

@@ -33,15 +33,13 @@ MPC_DECIMALS = 10000
 
 
 def shard_id_for_address(address: str) -> str:
-    # Very rough implementation
-    if address is None:
-        msg = 'Address must not be None'
-        raise TypeError(msg)
+    assert address is not None, 'missing address'
     if address.endswith('a'):
-        return 'shards/Shard0/'
-    if address.endswith('2'):
-        return 'shards/Shard1/'
-    return 'shards/Shard2/'
+        return 'shards/Shard0/'  # TODO
+    elif address.endswith('2'):
+        return 'shards/Shard1/'  # TODO
+    else:
+        return 'shards/Shard2/'  # TODO
 
 
 @dataclasses.dataclass(frozen=True)

fin_depo/static.py

@@ -48,5 +48,5 @@ class AggregateDepoFetcher(DepoFetcher):
         return DepoGroup(
             name=self.name,
             nested=[fetcher.get_depo() for fetcher in self.aggregated],
-            updated_time=datetime.datetime.now(tz=datetime.UTC),
+            updated_time=datetime.datetime.now(tz=datetime.UTC),  # TODO
         )

test/__init__.py

@@ -1 +0,0 @@
-"""Tests for fin_depo."""

test/test_nordnet.py

@@ -14,8 +14,7 @@ needs_secrets = pytest.mark.skipif(
 def print_pretty(depo: fin_depo.data.Depo):
     for asset in depo.assets():
         amount = depo.get_amount_of_asset(asset)
-        str_asset = str(asset)
-        print(f'{str_asset:15} : {amount:10}')
+        print(f'{str(asset):15} : {amount:10}')
         del asset, amount