diff --git a/fin_depo/__init__.py b/fin_depo/__init__.py index cfe5d8d..e7aa364 100644 --- a/fin_depo/__init__.py +++ b/fin_depo/__init__.py @@ -52,6 +52,7 @@ __all__ = [ 'investbank_nordnet', '__version__', 'data', + 'static', ] from . import ( @@ -60,5 +61,6 @@ from . import ( defi_kucoin, defi_partisia_blockchain, investbank_nordnet, + static, ) from ._version import __version__ diff --git a/fin_depo/static.py b/fin_depo/static.py new file mode 100644 index 0000000..a0927dd --- /dev/null +++ b/fin_depo/static.py @@ -0,0 +1,49 @@ +"""Static depository fetcher. + +This module does not represent an API or other integration. This module +contains aggregators and static depositories. +""" + +import datetime +import logging +from decimal import Decimal + +import fin_defs +import krakenex +import dataclasses + +from .data import Depo, DepoFetcher, DepoSingle, DepoGroup + +logger = logging.getLogger(__name__) + +@dataclasses.dataclass(frozen=True) +class StaticDepoFetcher(DepoFetcher): + """Depository "fetcher" that doesn't do any fetching. + + Is given a static set of assets and will always return these directly. + """ + + name: str + depo_assets: dict[fin_defs.Asset,Decimal] + last_updated: datetime.datetime = dataclasses.field(default_factory=lambda: datetime.datetime.now(tz=datetime.UTC)) + + def get_depo(self) -> DepoSingle: + return DepoSingle( + name=self.name, + _assets=self.depo_assets, + updated_time=self.last_updated, + ) + +@dataclasses.dataclass(frozen=True) +class AggregateDepoFetcher(DepoFetcher): + """Depository "fetcher" that delegates to the aggregated fetchers.""" + + name: str + aggregated: list[DepoFetcher] + + def get_depo(self) -> DepoGroup: + return DepoGroup( + name=self.name, + nested=[fetcher.get_depo() for fetcher in self.aggregated], + updated_time=datetime.datetime.now(tz=datetime.UTC), # TODO + ) diff --git a/test/test_static.py b/test/test_static.py new file mode 100644 index 0000000..a6cc532 --- /dev/null +++ b/test/test_static.py @@ -0,0 +1,16 @@ +import fin_defs + +import fin_depo +from decimal import Decimal + +def test_get_depo(): + fetcher = fin_depo.static.StaticDepoFetcher( + 'Test', {fin_defs.BTC: Decimal(1000), fin_defs.USD: Decimal(2000)} + ) + + depo = fetcher.get_depo() + + # Check layout + assert isinstance(depo, fin_depo.data.DepoSingle) + assert depo.get_amount_of_asset(fin_defs.BTC) == 1000 + assert depo.get_amount_of_asset(fin_defs.USD) == 2000