1
0

Code quality improvements
All checks were successful
Python Ruff Code Quality / ruff (push) Successful in 22s
Run Python tests (through Pytest) / Test (push) Successful in 28s
Verify Python project can be installed, loaded and have version checked / Test (push) Successful in 23s

This commit is contained in:
Jon Michael Aanes 2024-11-01 11:30:18 +01:00
parent c459e16571
commit 836b3b2a35
Signed by: Jmaa
SSH Key Fingerprint: SHA256:Ab0GfHGCblESJx7JRE4fj4bFy/KRpeLhi41y4pF3sNA
7 changed files with 40 additions and 30 deletions

View File

@ -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

View File

@ -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()

View File

@ -94,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>
"""
@ -117,7 +114,10 @@ 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:
@ -197,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)

View File

@ -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)

View File

@ -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),
)

View File

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

View File

@ -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