mango-explorer/mango/instrumentvalue.py

# # ⚠ Warning
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT
# LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN
# NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
# SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
#
# [🥭 Mango Markets](https://mango.markets/) support is available at:
#   [Docs](https://docs.mango.markets/)
#   [Discord](https://discord.gg/67jySBhxrg)
#   [Twitter](https://twitter.com/mangomarkets)
#   [Github](https://github.com/blockworks-foundation)
#   [Email](mailto:hello@blockworks.foundation)


import logging
import numbers
import typing

from decimal import Decimal
from solana.publickey import PublicKey
from solana.rpc.types import TokenAccountOpts

from .context import Context
from .output import output
from .tokens import Instrument, Token


def _decimal_from_number(value: numbers.Number) -> Decimal:
    # Decimal constructor can only handle these Number types:
    # Union[Decimal, float, str, Tuple[int, Sequence[int], int]]
    if isinstance(value, Decimal):
        return value
    if isinstance(value, int):
        return Decimal(value)
    if isinstance(value, float):
        return Decimal(value)
    if isinstance(value, str):
        return Decimal(value)
    raise Exception(f"Cannot handle conversion of {value} to Decimal.")


# # 🥭 InstrumentValue class
#
# The `InstrumentValue` class is a simple way of keeping a token and value together, and
# displaying them nicely consistently.
#
class InstrumentValue:
    def __init__(self, token: Instrument, value: Decimal) -> None:
        self._logger: logging.Logger = logging.getLogger(self.__class__.__name__)
        self.token: Instrument = token
        self.value: Decimal = value
        if not isinstance(self.value, Decimal):
            raise Exception(f"Value is {type(self.value)}, not Decimal: {self.value}")

    def shift_to_native(self) -> "InstrumentValue":
        new_value = self.token.shift_to_native(self.value)
        return InstrumentValue(self.token, new_value)

    @staticmethod
    def fetch_total_value_or_none(
        context: Context, account_public_key: PublicKey, token: Token
    ) -> typing.Optional["InstrumentValue"]:
        opts = TokenAccountOpts(mint=token.mint)

        token_accounts = context.client.get_token_accounts_by_owner(
            account_public_key, opts
        )
        if len(token_accounts) == 0:
            return None

        total_value = Decimal(0)
        for token_account in token_accounts:
            token_balance: Decimal = context.client.get_token_account_balance(
                token_account["pubkey"]
            )
            total_value += token_balance

        return InstrumentValue(token, total_value)

    @staticmethod
    def fetch_total_value(
        context: Context, account_public_key: PublicKey, token: Token
    ) -> "InstrumentValue":
        value = InstrumentValue.fetch_total_value_or_none(
            context, account_public_key, token
        )
        if value is None:
            return InstrumentValue(token, Decimal(0))
        return value

    @staticmethod
    def report(
        values: typing.Sequence["InstrumentValue"],
        reporter: typing.Callable[[str], None] = output,
    ) -> None:
        for value in values:
            reporter(f"{value.value:>18,.8f} {value.token.name}")

    @staticmethod
    def find_by_symbol(
        values: typing.Sequence[typing.Optional["InstrumentValue"]], symbol: str
    ) -> "InstrumentValue":
        found = [
            value
            for value in values
            if value is not None
            and value.token is not None
            and value.token.symbol_matches(symbol)
        ]
        if len(found) == 0:
            raise Exception(f"Token '{symbol}' not found in token values: {values}")

        if len(found) > 1:
            raise Exception(
                f"Token '{symbol}' matched multiple tokens in values: {values}"
            )

        return found[0]

    @staticmethod
    def find_by_token(
        values: typing.Sequence[typing.Optional["InstrumentValue"]], token: Instrument
    ) -> "InstrumentValue":
        return InstrumentValue.find_by_symbol(values, token.symbol)

    @staticmethod
    def changes(
        before: typing.Sequence["InstrumentValue"],
        after: typing.Sequence["InstrumentValue"],
    ) -> typing.Sequence["InstrumentValue"]:
        changes: typing.List[InstrumentValue] = []
        for before_balance in before:
            after_balance = InstrumentValue.find_by_token(after, before_balance.token)
            result = InstrumentValue(
                before_balance.token, after_balance.value - before_balance.value
            )
            changes += [result]

        return changes

    def __add__(self, token_value_to_add: "InstrumentValue") -> "InstrumentValue":
        if self.token != token_value_to_add.token:
            raise Exception(
                f"Cannot add InstrumentValues from different tokens ({self.token} and {token_value_to_add.token})."
            )
        return InstrumentValue(self.token, self.value + token_value_to_add.value)

    def __sub__(self, token_value_to_subtract: "InstrumentValue") -> "InstrumentValue":
        if self.token != token_value_to_subtract.token:
            raise Exception(
                f"Cannot subtract InstrumentValues from different tokens ({self.token} and {token_value_to_subtract.token})."
            )
        return InstrumentValue(self.token, self.value - token_value_to_subtract.value)

    def __mul__(self, token_value_to_multiply: "InstrumentValue") -> "InstrumentValue":
        # Multiplying by another InstrumentValue is assumed to be a token value multiplied by a token price.
        # The result should be denominated in the currency of the price.
        return InstrumentValue(
            token_value_to_multiply.token, self.value * token_value_to_multiply.value
        )

    def __lt__(self, other: typing.Any) -> bool:
        if isinstance(other, numbers.Number):
            return self.value < _decimal_from_number(other)

        if not isinstance(other, InstrumentValue):
            return NotImplemented

        if self.token != other.token:
            raise Exception(
                f"Cannot compare token values when one token is {self.token.symbol} and the other is {other.token.symbol}."
            )
        return self.value < other.value

    def __gt__(self, other: typing.Any) -> bool:
        if isinstance(other, numbers.Number):
            return self.value > _decimal_from_number(other)

        if not isinstance(other, InstrumentValue):
            return NotImplemented

        if self.token != other.token:
            raise Exception(
                f"Cannot compare token values when one token is {self.token.symbol} and the other is {other.token.symbol}."
            )
        return self.value > other.value

    def __eq__(self, other: typing.Any) -> bool:
        if (
            isinstance(other, InstrumentValue)
            and self.token == other.token
            and self.value == other.value
        ):
            return True
        return False

    def __format__(self, format_spec: str) -> str:
        return format(str(self), format_spec)

    def __str__(self) -> str:
        name = "« Un-Named Instrument »"
        if self.token and self.token.name:
            name = self.token.name
        return f"« InstrumentValue: {self.value:>18,.8f} {name} »"

    def __repr__(self) -> str:
        return f"{self}"