Code quality improvements

fin_depo/__init__.py

@@ -43,20 +43,27 @@ 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',
-    'defi_partisia_blockchain',
     'investbank_nordnet',
+    'defi_partisia_blockchain',
     '__version__',
     'data',
     'static',
 ]
 
 import importlib
-import inspect
 import logging
 import sys
 
@@ -68,10 +75,12 @@ 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.
 
-def load_backend(name: str) -> object | None:
+    Module is assigned as an attribute to the root module.
+    """
     try:
         imported_module = importlib.import_module(f'{__name__}.{name}')
     except Exception:
@@ -80,19 +89,12 @@ def load_backend(name: str) -> object | None:
             name,
         )
         return None
-    setattr(CURRENT_MODULE, name, imported_module)
+    setattr(root_module, name, imported_module)
     return imported_module
 
 
 # Import modules
 
-backend_modules = [
-    'defi_kraken',
-    'defi_kucoin',
-    'investbank_nordnet',
-    'defi_partisia_blockchain',
-]
-for m in backend_modules:
-    load_backend(m)
+for m in BACKEND_MODULE_NAMES:
+    load_backend(sys.modules[__name__], m)
     del m
-del backend_modules, CURRENT_MODULE

fin_depo/data.py

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

fin_depo/defi_kraken.py

@@ -55,6 +55,7 @@ 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,6 +12,7 @@ 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')
@@ -93,22 +94,19 @@ 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 WIP
+        - A fee will be paid to Kucoin, with the rate determined by your VIP
           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>
         """
@@ -116,11 +114,16 @@ class KucoinDepoFetcher(DepoFetcher):
         if not self.allow_trades:
             msg = 'KucoinDepoFetcher.allow_trades is not enabled: Cannot make trades'
             raise PermissionError(msg)
-        assert fin_defs.USDT in [input_asset, output_asset], 'USDT markets only for now'
+
+        if fin_defs.USDT not in [input_asset, output_asset]:
+            msg = 'Non-USDT Markets are not supported'
+            raise NotImplementedError(msg)
 
         # 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)
@@ -194,6 +197,6 @@ class KucoinDepoFetcher(DepoFetcher):
         for _ in range(num_retries):
             try:
                 return self.kucoin_client.get_order(order_id)
-            except kucoin.exceptions.KucoinAPIException as e:  # noqa
+            except kucoin.exceptions.KucoinAPIException:  # noqa: PERF203
                 time.sleep(sleep_between_tries)
         return self.kucoin_client.get_order(order_id)

fin_depo/defi_partisia_blockchain.py

@@ -33,13 +33,15 @@ MPC_DECIMALS = 10000
 
 
 def shard_id_for_address(address: str) -> str:
-    assert address is not None, 'missing address'
+    # Very rough implementation
+    if address is None:
+        msg = 'Address must not be None'
+        raise TypeError(msg)
     if address.endswith('a'):
-        return 'shards/Shard0/'  # TODO
-    elif address.endswith('2'):
-        return 'shards/Shard1/'  # TODO
-    else:
-        return 'shards/Shard2/'  # TODO
+        return 'shards/Shard0/'
+    if address.endswith('2'):
+        return 'shards/Shard1/'
+    return 'shards/Shard2/'
 
 
 @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),  # TODO
+            updated_time=datetime.datetime.now(tz=datetime.UTC),
         )

test/__init__.py

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

test/test_nordnet.py

@@ -14,7 +14,8 @@ 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)
-        print(f'{str(asset):15} : {amount:10}')
+        str_asset = str(asset)
+        print(f'{str_asset:15} : {amount:10}')
         del asset, amount