Source code for y.prices.dex.mooniswap

import logging
from decimal import Decimal
from typing import Optional

import dank_mids
from a_sync import a_sync, cgather
from brownie import ZERO_ADDRESS, chain

from y import ENVIRONMENT_VARIABLES as ENVS
from y import convert
from y.classes.common import ERC20
from y.constants import weth
from y.contracts import Contract
from y.datatypes import AnyAddressType, Block, UsdPrice
from y.prices import magic
from y.utils import gather_methods

logger = logging.getLogger(__name__)

if chain.id == 1:
    router = Contract("0xbAF9A5d4b0052359326A6CDAb54BABAa3a3A9643")
    gas_coin = weth
elif chain.id == 56:
    router = Contract("0xD41B24bbA51fAc0E4827b6F94C0D6DDeB183cD64")
    gas_coin = Contract("0xbb4CdB9CBd36B01bD1cBaEBF2De08d9173bc095c")  # wbnb
else:
    router = None
    gas_coin = None


[docs] @a_sync(default="sync", ram_cache_ttl=5 * 60) async def is_mooniswap_pool(token: AnyAddressType) -> bool: """ Check if the given token address is a Mooniswap pool. This function converts the token address and uses the router contract's `isPool` method to determine if the token is a Mooniswap pool. Note that if the active chain does not configure the Mooniswap router (i.e. chain IDs other than 1 and 56), the function always returns False. Args: token: The address of the token to check. Returns: True if the token is a Mooniswap pool and the router is configured; otherwise, False. Examples: >>> is_mooniswap_pool("0x1234567890abcdef1234567890abcdef12345678") True >>> is_mooniswap_pool("0xabcdefabcdefabcdefabcdefabcdefabcdef") False See Also: :func:`~y.prices.magic.get_price` """ address = await convert.to_address_async(token) return False if router is None else await router.isPool.coroutine(address)
[docs] @a_sync(default="sync") async def get_pool_price( token: AnyAddressType, block: Optional[Block] = None, skip_cache: bool = ENVS.SKIP_CACHE, ) -> UsdPrice: """ Get the price of the given Mooniswap pool token. This function retrieves the underlying token addresses with the pool's `token0` and `token1` methods, and then obtains the corresponding token balances and USD prices via asynchronous calls. It computes the total USD value of the pool by multiplying each token balance with its USD price, sums these values, and divides by the total supply of pool tokens. Although the return type is declared as UsdPrice, the function returns a Decimal value. Args: token: The address of the pool token. block: The block number at which to retrieve the data. Defaults to the latest block. skip_cache: Whether to skip using the cache for price retrieval. Defaults to :obj:`ENVS.SKIP_CACHE`. Returns: The computed USD price of the pool token as a Decimal. Note that this differs from the documented UsdPrice type. Examples: >>> price = get_pool_price("0x1234567890abcdef1234567890abcdef12345678") >>> print(price) 1.2345 >>> price = get_pool_price("0xabcdefabcdefabcdefabcdefabcdefabcdef", block=12345678) >>> print(price) 0.9876 See Also: :func:`~y.prices.magic.get_price` """ address = await convert.to_address_async(token) token0, token1 = await gather_methods(address, ("token0", "token1")) bal0, bal1, price0, price1, total_supply = await cgather( ( dank_mids.eth.get_balance(address, block_identifier=block) if token0 == ZERO_ADDRESS else ERC20(token0, asynchronous=True).balance_of_readable(address, block) ), ( dank_mids.eth.get_balance(address, block_identifier=block) if token1 == ZERO_ADDRESS else ERC20(token1, asynchronous=True).balance_of_readable(address, block) ), ( magic.get_price(gas_coin, block, skip_cache=skip_cache, sync=False) if token0 == ZERO_ADDRESS else magic.get_price(token0, block, skip_cache=skip_cache, sync=False) ), ( magic.get_price(gas_coin, block, skip_cache=skip_cache, sync=False) if token1 == ZERO_ADDRESS else magic.get_price(token1, block, skip_cache=skip_cache, sync=False) ), ERC20(address, asynchronous=True).total_supply_readable(block), ) if token0 == ZERO_ADDRESS: bal0 = Decimal(bal0) / 10**18 elif token1 == ZERO_ADDRESS: bal1 = Decimal(bal1) / 10**18 totalVal = bal0 * Decimal(price0) + bal1 * Decimal(price1) return totalVal / Decimal(total_supply)