Source code for tradeexecutor.state.generic_position

"""Base class for trading and reserve positions."""

from _decimal import Decimal
from abc import abstractmethod, ABC
from typing import Iterable, Tuple

from tradeexecutor.state.balance_update import BalanceUpdate
from tradeexecutor.state.identifier import AssetIdentifier
from tradeexecutor.state.types import USDollarAmount

class BalanceUpdateEventAlreadyAdded(Exception):
    """Tries to include the same balance update event twice.

    See :py:meth:`GenericPosition.add_balance_update_event`

[docs]class GenericPosition(ABC): """Base class for trading and reserve positions. Implements common method all positions need to implement. TODO: How to define generic `balance_updates` mapping. See also - :py:class:`tradeexecutor.state.state.position.TradingPosition` - :py:class:`tradeexecutor.state.state.reserve.ReservePosition` """
[docs] @abstractmethod def get_human_readable_name(self) -> str: """How to refer this position in log output."""
[docs] @abstractmethod def get_quantity(self) -> Decimal: """Get the number of tokens held in this position."""
[docs] @abstractmethod def calculate_quantity_usd_value(self, quantity: Decimal) -> USDollarAmount: """Price token amount in this position in US dollar. An estimation. Use whatever latest exchange rate that is appropriate. :return: Dollar amount """
[docs] @abstractmethod def get_base_token_balance_update_quantity(self) -> Decimal: """Get quantity of all balance updates for this position. Balance update events are - Deposits - Redemptions - Accounting corrections ... but not trades. :return: What's the total value of non-trade events affecting the balances of this position. """
[docs] @abstractmethod def get_balance_update_events(self) -> Iterable[BalanceUpdate]: """Iterate over all balance update events. Balance updates describe external events affecting the balance of this position: the update was not triggered by the trade executor itself. - Deposits - Redemptions - Account corrections - **Trades** are not included here """
[docs] @abstractmethod def add_balance_update_event(self, event: BalanceUpdate): """Include a new balance update event :raise BalanceUpdateEventAlreadyAdded: In the case of a duplicate and event id is already used. """
[docs] @abstractmethod def get_held_assets(self) -> Iterable[Tuple[AssetIdentifier, Decimal]]: """Get all assets on-chain wallet should hold for this position. - A position may hold multiple on-chain assets - Multiple positions can share the same on-chain asset - Any on-chain accrued interest is included in ``quantity`` :return: Iterable (asset, quantity) tuples """