"""Datastructures for `fin-depo`.""" 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, 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 @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: AssetAmount output: AssetAmount fee: AssetAmount executed_time: datetime.datetime order_id: object raw_order_details: object @enforce_typing.enforce_types @dataclasses.dataclass(frozen=True) class WithdrawalDetails: withdrawn: AssetAmount fee: AssetAmount executed_time: datetime.datetime raw_details: object @enforce_typing.enforce_types @dataclasses.dataclass(frozen=True) class DepositDetails: deposit: AssetAmount fee: AssetAmount executed_time: datetime.datetime raw_details: object @enforce_typing.enforce_types @dataclasses.dataclass(frozen=True) class DoubleRegister: input: AssetAmount | None output: AssetAmount | None executed_time: datetime.datetime