mango-explorer/mango/valuation.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://markets/) support is available at:
#   [Docs](https://docs.markets/)
#   [Discord](https://discord.gg/67jySBhxrg)
#   [Twitter](https://twitter.com/mangomarkets)
#   [Github](https://github.com/blockworks-foundation)
#   [Email](mailto:hello@blockworks.foundation)

import typing

from datetime import datetime
from decimal import Decimal
from solana.publickey import PublicKey

from .account import Account
from .accountinstrumentvalues import AccountInstrumentValues
from .cache import Cache
from .context import Context
from .group import Group
from .instrumentvalue import InstrumentValue
from .openorders import OpenOrders
from .token import Instrument, Token, SolToken


class TokenValuation:
    def __init__(self, raw_token_value: InstrumentValue, price_token_value: InstrumentValue,
                 value_token_value: InstrumentValue) -> None:
        self.raw: InstrumentValue = raw_token_value
        self.price: InstrumentValue = price_token_value
        self.value: InstrumentValue = value_token_value

    @staticmethod
    def from_json_dict(context: Context, json: typing.Dict[str, typing.Any]) -> "TokenValuation":
        symbol: str = json["symbol"]
        value_currency: str = json["valueCurrency"]
        balance: Decimal = Decimal(json["balance"])
        price: Decimal = Decimal(json["price"])
        value: Decimal = Decimal(json["value"])
        token = context.instrument_lookup.find_by_symbol(symbol)
        if token is None:
            raise Exception(f"Could not find token for symbol: {symbol}")
        currency_token = context.instrument_lookup.find_by_symbol(value_currency)
        if currency_token is None:
            raise Exception(f"Could not find token for currency symbol: {value_currency}")
        raw_token_value: InstrumentValue = InstrumentValue(token, balance)
        price_token_value: InstrumentValue = InstrumentValue(currency_token, price)
        value_token_value: InstrumentValue = InstrumentValue(currency_token, value)
        return TokenValuation(raw_token_value, price_token_value, value_token_value)

    @staticmethod
    def from_token_balance(context: Context, group: Group, cache: Cache, token_balance: InstrumentValue) -> "TokenValuation":
        token_to_lookup: Instrument = token_balance.token
        if token_balance.token == SolToken:
            token_to_lookup = context.instrument_lookup.find_by_symbol_or_raise("SOL")
        cached_token_price: InstrumentValue = group.token_price_from_cache(cache, token_to_lookup)
        balance_value: InstrumentValue = token_balance * cached_token_price
        return TokenValuation(token_balance, cached_token_price, balance_value)

    @staticmethod
    def all_from_wallet(context: Context, group: Group, cache: Cache, address: PublicKey) -> typing.Sequence["TokenValuation"]:
        balances: typing.List[InstrumentValue] = []
        sol_balance = context.client.get_balance(address)
        balances += [InstrumentValue(SolToken, sol_balance)]

        for slot_token_bank in group.tokens:
            if isinstance(slot_token_bank.token, Token):
                balance = InstrumentValue.fetch_total_value(context, address, slot_token_bank.token)
                balances += [balance]

        wallet_tokens: typing.List[TokenValuation] = []
        for balance in balances:
            if balance.value != 0:
                wallet_tokens += [TokenValuation.from_token_balance(context, group, cache, balance)]

        return wallet_tokens

    def to_json_dict(self) -> typing.Dict[str, typing.Any]:
        return {
            "symbol": self.raw.token.symbol,
            "valueCurrency": self.price.token.symbol,
            "balance": f"{self.raw.value:.8f}",
            "price": f"{self.price.value:.8f}",
            "value": f"{self.value.value:.8f}",
        }

    def __str__(self) -> str:
        return f"   {self.raw:<45} worth {self.value}"


class AccountValuation:
    def __init__(self, name: str, address: PublicKey, tokens: typing.Sequence[TokenValuation]) -> None:
        self.name: str = name
        self.address: PublicKey = address
        self.tokens: typing.Sequence[TokenValuation] = tokens

    @property
    def value(self) -> InstrumentValue:
        return sum((t.value for t in self.tokens[1:]), start=self.tokens[0].value)

    @staticmethod
    def from_json_dict(context: Context, json: typing.Dict[str, typing.Any]) -> "AccountValuation":
        name: str = json["name"]
        address: PublicKey = PublicKey(json["address"])
        tokens: typing.List[TokenValuation] = []
        for token_dict in json["tokens"]:
            token_valuation: TokenValuation = TokenValuation.from_json_dict(context, token_dict)
            tokens += [token_valuation]

        return AccountValuation(name, address, tokens)

    @staticmethod
    def from_account(context: Context, group: Group, account: Account, cache: Cache) -> "AccountValuation":
        open_orders: typing.Dict[str, OpenOrders] = account.load_all_spot_open_orders(context)
        token_values: typing.List[TokenValuation] = []
        for asset in account.base_slots:
            if (asset.net_value.value != 0) or ((asset.perp_account is not None) and not asset.perp_account.empty):
                report: AccountInstrumentValues = AccountInstrumentValues.from_account_basket_base_token(
                    asset, open_orders, group)
                asset_valuation = TokenValuation.from_token_balance(context, group, cache, report.net_value)
                token_values += [asset_valuation]

        quote_valuation = TokenValuation.from_token_balance(context, group, cache, account.shared_quote.net_value)
        token_values += [quote_valuation]

        return AccountValuation(account.info, account.address, token_values)

    def to_json_dict(self) -> typing.Dict[str, typing.Any]:
        value: InstrumentValue = self.value
        return {
            "name": self.name,
            "address": f"{self.address}",
            "value": f"{value.value:.8f}",
            "valueCurrency": value.token.symbol,
            "tokens": list([tok.to_json_dict() for tok in self.tokens])
        }


class Valuation:
    def __init__(self, timestamp: datetime, address: PublicKey, wallet_tokens: typing.Sequence[TokenValuation], accounts: typing.Sequence[AccountValuation]):
        self.timestamp: datetime = timestamp
        self.address: PublicKey = address
        self.wallet_tokens: typing.Sequence[TokenValuation] = wallet_tokens
        self.accounts: typing.Sequence[AccountValuation] = accounts

    @property
    def value(self) -> InstrumentValue:
        wallet_tokens_value: InstrumentValue = sum(
            (t.value for t in self.wallet_tokens[1:]), start=self.wallet_tokens[0].value)
        return sum((acc.value for acc in self.accounts), start=wallet_tokens_value)

    @staticmethod
    def from_json_dict(context: Context, json: typing.Dict[str, typing.Any]) -> "Valuation":
        timestamp: datetime = datetime.fromisoformat(json["timestamp"])
        address: PublicKey = PublicKey(json["address"])
        wallet_tokens: typing.List[TokenValuation] = []
        for token_dict in json["wallet"]["tokens"]:
            token_valuation: TokenValuation = TokenValuation.from_json_dict(context, token_dict)
            wallet_tokens += [token_valuation]

        accounts: typing.List[AccountValuation] = []
        for account_dict in json["accounts"]:
            account_valuation: AccountValuation = AccountValuation.from_json_dict(context, account_dict)
            accounts += [account_valuation]

        return Valuation(timestamp, address, wallet_tokens, accounts)

    @staticmethod
    def from_wallet(context: Context, group: Group, cache: Cache, address: PublicKey) -> "Valuation":
        spl_tokens = TokenValuation.all_from_wallet(context, group, cache, address)

        mango_accounts = Account.load_all_for_owner(context, address, group)
        account_valuations = []
        for account in mango_accounts:
            account_valuations += [AccountValuation.from_account(context, group, account, cache)]

        return Valuation(datetime.now(), address, spl_tokens, account_valuations)

    def to_json_dict(self) -> typing.Dict[str, typing.Any]:
        value: InstrumentValue = self.value
        wallet_value: InstrumentValue = sum(
            (t.value for t in self.wallet_tokens[1:]), start=self.wallet_tokens[0].value)
        return {
            "timestamp": self.timestamp.isoformat(),
            "address": f"{self.address}",
            "value": f"{value.value:.8f}",
            "valueCurrency": value.token.symbol,
            "wallet": {
                "value": f"{wallet_value.value:.8f}",
                "valueCurrency": wallet_value.token.symbol,
                "tokens": list([tok.to_json_dict() for tok in self.wallet_tokens])
            },
            "accounts": list([acc.to_json_dict() for acc in self.accounts])
        }

    def __str__(self) -> str:
        address: str = f"{self.address}:"
        wallet_total: InstrumentValue = sum(
            (t.value for t in self.wallet_tokens[1:]), start=self.wallet_tokens[0].value)
        accounts: typing.List[str] = []
        for account in self.accounts:
            account_tokens: str = "\n        ".join([f"{item}" for item in account.tokens])
            accounts += [f"""Account '{account.name}' (total: {account.value}):
        {account_tokens}"""]

        accounts_tokens: str = "\n        ".join(accounts)
        wallet_tokens: str = "\n        ".join([f"{item}" for item in self.wallet_tokens])
        return f"""« Valuation of {address:<47} {self.value}
    Wallet Tokens (total: {wallet_total}):
        {wallet_tokens}
    {accounts_tokens}
»"""