133 lines
3.8 KiB
Python
133 lines
3.8 KiB
Python
import abc
|
|
import dataclasses
|
|
import datetime
|
|
from collections.abc import Iterable, Mapping
|
|
from decimal import Decimal
|
|
from typing import TypeVar
|
|
|
|
import enforce_typing
|
|
from fin_defs import Asset
|
|
|
|
|
|
@enforce_typing.enforce_types
|
|
@dataclasses.dataclass(frozen=True)
|
|
class Depo(abc.ABC):
|
|
"""A depository tracking some amount of assets.
|
|
|
|
Depo can either be DepoSingle, which is the base layer of the depository
|
|
structure, or nested in DepoGroup, which allows for a complex hierarcy of
|
|
depositories.
|
|
|
|
The depository structure exposed by each backend depends upon the logical
|
|
structure of the relevant service and the API of this service.
|
|
"""
|
|
|
|
name: str
|
|
updated_time: datetime.datetime
|
|
|
|
@abc.abstractmethod
|
|
def assets(self) -> Iterable[Asset]:
|
|
"""Returns the different assets managed by this depo."""
|
|
|
|
@abc.abstractmethod
|
|
def get_amount_of_asset(self, asset: Asset) -> Decimal:
|
|
"""Returns the amount of owned assets for all nested depos.
|
|
|
|
Must return 0 if depo does not contain the given asset.
|
|
"""
|
|
|
|
|
|
@enforce_typing.enforce_types
|
|
@dataclasses.dataclass(frozen=True)
|
|
class DepoSingle(Depo):
|
|
"""Base level of depository."""
|
|
|
|
_assets: Mapping[Asset, Decimal]
|
|
|
|
def __post_init__(self):
|
|
assert None not in self._assets
|
|
|
|
def assets(self) -> Iterable[Asset]:
|
|
return self._assets
|
|
|
|
def get_amount_of_asset(self, asset: Asset) -> Decimal:
|
|
return self._assets.get(asset, Decimal(0))
|
|
|
|
|
|
@enforce_typing.enforce_types
|
|
@dataclasses.dataclass(frozen=True)
|
|
class DepoGroup(Depo):
|
|
"""Nested depository."""
|
|
|
|
nested: list[Depo]
|
|
|
|
def __post_init__(self):
|
|
assert None not in self.nested
|
|
|
|
def assets(self) -> Iterable[Asset]:
|
|
assets: set[Asset] = set()
|
|
for nested_depo in self.nested:
|
|
assets.update(nested_depo.assets())
|
|
return assets
|
|
|
|
def get_amount_of_asset(self, asset: Asset) -> Decimal:
|
|
summed: Decimal = Decimal(0)
|
|
for nested_depo in self.nested:
|
|
summed += nested_depo.get_amount_of_asset(asset)
|
|
del nested_depo
|
|
return summed
|
|
|
|
|
|
T = TypeVar('T')
|
|
|
|
|
|
class DepoFetcher(abc.ABC):
|
|
"""Base `Depo` fetcher interface.
|
|
|
|
Used to get depository information for specific websites/backends.
|
|
"""
|
|
|
|
@abc.abstractmethod
|
|
def get_depo(self) -> Depo:
|
|
"""Fetches the `Depo`s available for the fetcher, possibly as
|
|
a `DepoGroup`.
|
|
"""
|
|
|
|
def assert_param(self, param_name: str, param_type: type[T], param_value: T) -> T:
|
|
if not isinstance(param_value, param_type):
|
|
msg = f'{self} expected {param_name} parameter of type {param_type}, but got: {param_value}'
|
|
raise TypeError(msg)
|
|
return param_value
|
|
|
|
|
|
@enforce_typing.enforce_types
|
|
@dataclasses.dataclass(frozen=True)
|
|
class TradeOrderDetails:
|
|
"""Information about an executed trade.
|
|
|
|
Includes both well-structured data about the order, and unstructured data
|
|
from the backend. The unstructured data might be needed for tax purposes.
|
|
|
|
Fields:
|
|
- `input_asset`: Asset that have been sold
|
|
- `input_amount`: Amount of asset that has been sold
|
|
- `output_asset`: Asset that have been bought
|
|
- `output_amount`: Amount of asset that have been bought.
|
|
- `fee_asset`: Asset used to pay fee.
|
|
- `fee_amount`: Amount of asset that have been used to pay fee.
|
|
- `executed_time`: Point in time when order was executed.
|
|
- `order_id`: Unique identifier of order. Determined by backend.
|
|
- `raw_order_details`: For storing arbitrary unstructured data from backend.
|
|
"""
|
|
|
|
input_asset: Asset
|
|
input_amount: Decimal
|
|
output_asset: Asset
|
|
output_amount: Decimal
|
|
fee_asset: Asset
|
|
fee_amount: Decimal
|
|
executed_time: datetime.datetime
|
|
|
|
order_id: object
|
|
raw_order_details: object
|