import asyncio
import logging
from typing import Optional, Set
from async_lru import alru_cache
from brownie import chain
from y.constants import STABLECOINS, WRAPPED_GAS_COIN
from y.datatypes import Address, AnyAddressType
from y.prices.lending.aave import aave
from y.prices.lending.compound import CToken, compound
from y.prices.stable_swap.curve import curve
from y.prices.yearn import YearnInspiredVault, is_yearn_vault
from eth_portfolio.constants import BTC_LIKE, ETH_LIKE, INTL_STABLECOINS
logger = logging.getLogger(__name__)
OTHER_LONG_TERM_ASSETS: Set[Address] = {}.get(chain.id, set())
[docs]
async def get_token_bucket(token: AnyAddressType) -> str:
token = str(token)
try:
token = str(await _unwrap_token(token))
except ValueError as e:
if str(e).startswith("Source for") and str(e).endswith("has not been verified"):
return "Other short term assets"
raise
if _is_stable(token):
return "Cash & cash equivalents"
if token in ETH_LIKE:
return "ETH"
if token in BTC_LIKE:
return "BTC"
if token in OTHER_LONG_TERM_ASSETS:
return "Other long term assets"
return "Other short term assets"
@alru_cache(maxsize=None)
async def _unwrap_token(token) -> str:
"""
Unwraps the base
"""
if str(token) in {"ETH", "0xEeeeeEeeeEeEeeEeEeEeeEEEeeeeEeeeeeeeEEeE"}:
return token
if await is_yearn_vault(token, sync=False):
underlying = await YearnInspiredVault(token, asynchronous=True).underlying
return await _unwrap_token(underlying)
if curve and (pool := await curve.get_pool(token)):
pool_tokens = set(await asyncio.gather(*[_unwrap_token(coin) for coin in await pool.coins]))
if pool_bucket := _pool_bucket(pool_tokens):
return pool_bucket # type: ignore
if aave and await aave.is_atoken(token):
return str(await aave.underlying(token))
if compound and await compound.is_compound_market(token):
try:
return str(await CToken(token, asynchronous=True).underlying)
except AttributeError:
return WRAPPED_GAS_COIN
return token
def _pool_bucket(pool_tokens: set) -> Optional[str]:
logger.debug("Pool tokens: %s", pool_tokens)
if pool_tokens < BTC_LIKE:
return list(BTC_LIKE)[0]
if pool_tokens < ETH_LIKE:
return list(ETH_LIKE)[0]
if pool_tokens < STABLECOINS.keys():
return list(STABLECOINS.keys())[0]
return list(INTL_STABLECOINS)[0] if pool_tokens < INTL_STABLECOINS else None
def _is_stable(token: Address) -> bool:
return token in STABLECOINS or token in INTL_STABLECOINS