Added static and aggregator depo fetchers
All checks were successful
Test Python / Test (push) Successful in 26s
All checks were successful
Test Python / Test (push) Successful in 26s
This commit is contained in:
parent
177ce6a888
commit
66ad96af16
|
@ -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__
|
||||
|
|
49
fin_depo/static.py
Normal file
49
fin_depo/static.py
Normal file
|
@ -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
|
||||
)
|
16
test/test_static.py
Normal file
16
test/test_static.py
Normal file
|
@ -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
|
Loading…
Reference in New Issue
Block a user