"""Datastructures for `fin-depo`.""" import abc import dataclassabc 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, AssetAmount @enforce_typing.enforce_types @dataclasses.dataclass(frozen=True) class Depo(abc.ABC): """A depository tracking several 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): 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 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): 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() 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 class DoubleRegister(abc.ABC): """Information about some movement of assets.""" @property @abc.abstractmethod def input(self) -> AssetAmount | None: """How much and what kind of asset have been placed into the depository.""" raise NotImplementedError @property @abc.abstractmethod def output(self) -> AssetAmount | None: """How much and what kind of asset have been removed from the depository.""" raise NotImplementedError @property @abc.abstractmethod def fee(self) -> AssetAmount | None: """How much and what kind of asset was spent as a service fee.""" raise NotImplementedError @property @abc.abstractmethod def executed_time(self) -> datetime.datetime: """The precise point in time when the asset transaction was executed.""" raise NotImplementedError @enforce_typing.enforce_types @dataclassabc.dataclassabc(frozen=True) class TradeOrderDetails(DoubleRegister): """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: AssetAmount output: AssetAmount fee: AssetAmount executed_time: datetime.datetime order_id: object raw_order_details: object @enforce_typing.enforce_types @dataclassabc.dataclassabc(frozen=True) class WithdrawalDetails(DoubleRegister): """Information about a withdrawal of assets from a specific service. Withdraw assets may appear in another depository as a `DepositDetails`, but these cannot be automatically linked by `fin-depo`. Includes both well-structured data about the order, and unstructured data from the backend. The unstructured data might be needed for tax purposes. """ withdrawn: AssetAmount fee: AssetAmount executed_time: datetime.datetime raw_details: object @property def input(self) -> AssetAmount: return self.withdrawn @property def output(self) -> None: return None @enforce_typing.enforce_types @dataclassabc.dataclassabc(frozen=True) class DepositDetails(DoubleRegister): """Information about a deposit of assets to a specific service. Deposited assets may appear in another depository as a `WithdrawalDetails`, but these cannot be automatically linked by `fin-depo`. Includes both well-structured data about the order, and unstructured data from the backend. The unstructured data might be needed for tax purposes. """ deposit: AssetAmount fee: AssetAmount executed_time: datetime.datetime raw_details: object @property def input(self) -> None: return None @property def output(self) -> AssetAmount: return self.deposit