y.prices.tokenized_fund package

Submodules

y.prices.tokenized_fund.basketdao module

ASyncFunctiony.prices.tokenized_fund.basketdao.get_price(address: eth_typing.evm.ChecksumAddress, block: Union[int, eth_typing.evm.BlockNumber, NoneType] = None, skip_cache: bool = EnvironmentVariable[bool](name=`YPRICEMAGIC_SKIP_CACHE`, default_value=False, using_default=True)) → y.datatypes.UsdPrice[source]

Get the price of a BasketDAO token in USD by calculating the total value of its assets.

Args:

address: The address of the BasketDAO token. block (optional): The block number to query. Defaults to None (latest). skip_cache (optional): Whether to bypass the disk cache. Defaults to ENVS.SKIP_CACHE.

Example:
>>> price = get_price("0x0ac58Df435D3dC9F6e079B2C5F358A4b7e861B69")
>>> print(price)
1.05
Notes:

This function does not explicitly raise any exceptions such as a PriceError. Underlying errors during asset data retrieval will propagate, so callers should handle exceptions as needed.

See Also:

Since get_price is an ASyncFunctionSyncDefault, you can optionally pass sync=False or asynchronous=True to force it to return a coroutine. Without either kwarg, it will run synchronously.

Parameters:
Return type:

UsdPrice

ASyncFunctiony.prices.tokenized_fund.basketdao.is_basketdao_index(address: eth_typing.evm.ChecksumAddress) → bool[source]

Check if the given token is a BasketDAO token by attempting to call the getAssetsAndBalances method.

Args:

address: The address of the token to check.

Returns:

True if the token is a BasketDAO token, False otherwise.

Example:
>>> is_bd = is_basketdao_index("0x0ac58Df435D3dC9F6e079B2C5F358A4b7e861B69")
>>> print(is_bd)
True
Notes:

Any exceptions (e.g. ContractLogicError or ValueError) encountered during the call are caught, causing the function to return False.

See Also:

  • get_price() for fetching the price of a BasketDAO token.

Since is_basketdao_index is an ASyncFunctionSyncDefault, you can optionally pass sync=False or asynchronous=True to force it to return a coroutine. Without either kwarg, it will run synchronously.

Parameters:

address (ChecksumAddress)

Return type:

bool

y.prices.tokenized_fund.gelato module

ASyncFunctiony.prices.tokenized_fund.gelato.get_price(token: Union[str, hexbytes.main.HexBytes, ~AnyAddress, evmspec.data._main.Address, brownie.convert.datatypes.EthAddress, brownie.network.contract.Contract, int], block: Union[int, eth_typing.evm.BlockNumber, NoneType] = None, skip_cache: bool = EnvironmentVariable[bool](name=`YPRICEMAGIC_SKIP_CACHE`, default_value=False, using_default=True)) → y.datatypes.UsdPrice[source]

Calculate the price of a Gelato pool token in USD.

This function calculates the price of a Gelato pool token by retrieving the balances, scales, and prices of the pool’s underlying assets (token0 and token1), calculating their total value, and dividing by the total supply of the pool token.

Args:

token: The address of the token to price. block: The block number at which to get the price. Defaults to None. skip_cache: Whether to skip the cache. Defaults to ENVS.SKIP_CACHE.

Example:
>>> get_price("0x1234567890abcdef1234567890abcdef12345678")
123.45
See Also:

Since get_price is an ASyncFunctionSyncDefault, you can optionally pass sync=False or asynchronous=True to force it to return a coroutine. Without either kwarg, it will run synchronously.

Parameters:
Return type:

UsdPrice

ASyncFunctiony.prices.tokenized_fund.gelato.is_gelato_pool(token_address: str | hexbytes.main.HexBytes | AnyAddress | evmspec.data._main.Address | brownie.convert.datatypes.EthAddress | brownie.network.contract.Contract | int) → bool[source]

Check if a given token address is a Gelato pool.

Args:

token_address: The address of the token to check.

Returns:

True if the token is a Gelato pool, False otherwise.

Example:
>>> is_gelato_pool("0x1234567890abcdef1234567890abcdef12345678")
True
See Also:

Since is_gelato_pool is an ASyncFunctionSyncDefault, you can optionally pass sync=False or asynchronous=True to force it to return a coroutine. Without either kwarg, it will run synchronously.

Parameters:

token_address (str | HexBytes | AnyAddress | Address | EthAddress | Contract | int)

Return type:

bool

y.prices.tokenized_fund.piedao module

async y.prices.tokenized_fund.piedao.get_balance(bpool, token, block=None)[source]

Get the balance of a token in a Balancer pool.

Parameters:
Return type:

Decimal

Example

>>> get_balance("0xBpoolAddress", ERC20("0xTokenAddress"))
1000.0
async y.prices.tokenized_fund.piedao.get_bpool(pie_address, block=None)[source]

Get the Balancer pool address for a PieDAO token.

Parameters:
Returns:

The address of the Balancer pool, or the PieDAO address if no pool is found.

Return type:

str | HexBytes | AnyAddress | Address | EthAddress

Example

>>> get_bpool("0x1234567890abcdef1234567890abcdef12345678")
'0xBpoolAddress'
ASyncFunctiony.prices.tokenized_fund.piedao.get_price(pie: Union[str, hexbytes.main.HexBytes, ~AnyAddress, evmspec.data._main.Address, brownie.convert.datatypes.EthAddress, brownie.network.contract.Contract, int], block: Union[int, eth_typing.evm.BlockNumber, NoneType] = None, skip_cache: bool = EnvironmentVariable[bool](name=`YPRICEMAGIC_SKIP_CACHE`, default_value=False, using_default=True)) → y.datatypes.UsdPrice | None[source]

Get the price of a PieDAO token in USD.

Args:

pie: The address of the PieDAO token. block: The block number to query the price at. Defaults to the latest block. skip_cache: Whether to skip the cache when fetching the price. Defaults to the value of ENVS.SKIP_CACHE.

Example:
>>> get_price("0x1234567890abcdef1234567890abcdef12345678")
123.45
See Also:

Since get_price is an ASyncFunctionSyncDefault, you can optionally pass sync=False or asynchronous=True to force it to return a coroutine. Without either kwarg, it will run synchronously.

Parameters:
Return type:

UsdPrice | None

async y.prices.tokenized_fund.piedao.get_tokens(pie_address, block=None)[source]

Get the list of tokens in a PieDAO token.

Parameters:
Returns:

A list of ERC20 objects representing the tokens in the PieDAO token.

Return type:

List[ERC20] | None

Example

>>> get_tokens("0x1234567890abcdef1234567890abcdef12345678")
[ERC20('0xTokenAddress1'), ERC20('0xTokenAddress2')]

Note

This function retrieves token addresses using a multicall and then creates ERC20 instances from those addresses.

async y.prices.tokenized_fund.piedao.get_tvl(pie_address, block=None, skip_cache=EnvironmentVariable[bool](name=`YPRICEMAGIC_SKIP_CACHE`, default_value=False, using_default=True))[source]

Get the total value locked (TVL) in a PieDAO token in USD.

Parameters:

  • pie_address (str | HexBytes | AnyAddress | Address | EthAddress) – The address of the PieDAO token.

  • block (int | BlockNumber | None) – The block number to query the TVL at. Defaults to the latest block.

  • skip_cache (bool) – Whether to skip the cache when fetching the TVL. Defaults to the value of ENVS.SKIP_CACHE.

Return type:

UsdValue | None

Example

>>> get_tvl("0x1234567890abcdef1234567890abcdef12345678")
1000000.00

See also

async y.prices.tokenized_fund.piedao.get_value(bpool, token, block=None, skip_cache=EnvironmentVariable[bool](name=`YPRICEMAGIC_SKIP_CACHE`, default_value=False, using_default=True))[source]

Get the USD value of a token in a Balancer pool.

Parameters:

  • bpool (str | HexBytes | AnyAddress | Address | EthAddress) – The address of the Balancer pool.

  • token (ERC20) – The ERC20 token to query.

  • block (int | BlockNumber | None) – The block number to query the value at. Defaults to the latest block.

  • skip_cache (bool) – Whether to skip the cache when fetching the value. Defaults to the value of ENVS.SKIP_CACHE.

Return type:

UsdValue

Example

>>> get_value("0xBpoolAddress", ERC20("0xTokenAddress"))
500.0

Note

This function calculates the value by multiplying the token balance in the pool by its price.

See also

ASyncFunctiony.prices.tokenized_fund.piedao.is_pie(token: str | hexbytes.main.HexBytes | AnyAddress | evmspec.data._main.Address | brownie.convert.datatypes.EthAddress | brownie.network.contract.Contract | int) → bool[source]

Check if a given token is a PieDAO token.

Args:

token: The address of the token to check.

Returns:

True if the token is a PieDAO token, False otherwise.

Example:
>>> is_pie("0x1234567890abcdef1234567890abcdef12345678")
True

Since is_pie is an ASyncFunctionSyncDefault, you can optionally pass sync=False or asynchronous=True to force it to return a coroutine. Without either kwarg, it will run synchronously.

Parameters:

token (str | HexBytes | AnyAddress | Address | EthAddress | Contract | int)

Return type:

bool

y.prices.tokenized_fund.reserve module

ASyncFunctiony.prices.tokenized_fund.reserve.get_price(token_address: Union[str, hexbytes.main.HexBytes, ~AnyAddress, evmspec.data._main.Address, brownie.convert.datatypes.EthAddress], block: Union[int, eth_typing.evm.BlockNumber, NoneType] = None, skip_cache: bool = EnvironmentVariable[bool](name=`YPRICEMAGIC_SKIP_CACHE`, default_value=False, using_default=True)) → decimal.Decimal[source]

Retrieve the USD price of a Reserve Protocol R-token.

This function obtains the basket handler associated with the R-token by calling its “main” and “basketHandler” methods. It then calls the basket handler’s price method to retrieve a low and a high price value. The function computes the integer average of these two price values and scales the result by 1e18 to convert it to a USD price. If the call to the basket handler’s price method reverts, the function returns None instead of raising an exception.

Args:

token_address: The address of the R-token. block: The block number to query. Defaults to the latest block if not specified. skip_cache: If True, skip using the cache. Defaults to SKIP_CACHE.

Returns:

The USD price of the R-token as computed from the basket handler, or None if the price cannot be determined due to a remote call failure (i.e. a Revert).

Raises:

TypeError: If the token is not a valid R-token.

Example:
>>> address = "0xaCeeD87BD5754c3d714F3Bd43a9B7B0C9250ab0D"  # RSV token
>>> price = await get_price(address, sync=False)
>>> if price is None:
...     print("Price calculation failed due to a Revert exception")
... else:
...     print(f"Calculated price: {price}")
>>> # In a synchronous context:
>>> price = get_price(address)
>>> print(price)
See Also:

is_rtoken()

Since get_price is an ASyncFunctionSyncDefault, you can optionally pass sync=False or asynchronous=True to force it to return a coroutine. Without either kwarg, it will run synchronously.

Parameters:
Return type:

Decimal

ASyncFunctiony.prices.tokenized_fund.reserve.is_rtoken(token_address: str | hexbytes.main.HexBytes | AnyAddress | evmspec.data._main.Address | brownie.convert.datatypes.EthAddress) → bool[source]

Determine whether a given token is a Reserve Protocol R-token.

Args:

token_address: The address of the token to check.

Returns:

True if the token is a Reserve Protocol R-token, False otherwise.

Example:
>>> address = "0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48"  # USDC (not an R-token)
>>> result = is_rtoken(address)
>>> print(result)
False
See Also:

get_price()

Since is_rtoken is an ASyncFunctionSyncDefault, you can optionally pass sync=False or asynchronous=True to force it to return a coroutine. Without either kwarg, it will run synchronously.

Parameters:

token_address (str | HexBytes | AnyAddress | Address | EthAddress)

Return type:

bool

y.prices.tokenized_fund.tokensets module

class y.prices.tokenized_fund.tokensets.TokenSet[source]

Bases: ERC20

Represents a TokenSet, a type of ERC20 token with additional functionality.

A TokenSet is a specialized ERC20 token that can contain multiple components, each with its own balance. This class provides methods to interact with and retrieve information about TokenSets.

__eq__(_ContractBase__o)

Return self==value.

Parameters:

_ContractBase__o (object)

Return type:

bool

__init__(address, *, asynchronous=False)[source]

Initialize a TokenSet instance.

Parameters:

Examples

>>> tokenset = TokenSet("0x1234567890abcdef1234567890abcdef12345678")
__str__()

Return the contract address as a string.

Returns:

The contract address as a string.

Return type:

str

__build_name__: HiddenMethodDescriptor[Self, str]

Optional[Type[I]] = None) -> T Asynchronously retrieves the property value.

Args:

instance: The instance from which the property is accessed. owner: The owner class of the property.

Returns:

The property value.

The original docstring for get() is shown below:

_ASyncPropertyDescriptorBase.get(self, instance: I, owner: Optional[Type[I]] = None) -> T Asynchronously retrieves the property value.

Args:

instance: The instance from which the property is accessed. owner: The owner class of the property.

Returns:

The property value.

Type:

_ASyncPropertyDescriptorBase.get(self, instance

Type:

I, owner

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

__decimals__: HiddenMethodDescriptor[Self, int]

Optional[Type[I]] = None) -> T Asynchronously retrieves the property value.

Args:

instance: The instance from which the property is accessed. owner: The owner class of the property.

Returns:

The property value.

The original docstring for get() is shown below:

_ASyncPropertyDescriptorBase.get(self, instance: I, owner: Optional[Type[I]] = None) -> T Asynchronously retrieves the property value.

Args:

instance: The instance from which the property is accessed. owner: The owner class of the property.

Returns:

The property value.

Type:

_ASyncPropertyDescriptorBase.get(self, instance

Type:

I, owner

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

__name__: HiddenMethodDescriptor[Self, str] = 'TokenSet'
__scale__: HiddenMethodDescriptor[Self, int]

Optional[Type[I]] = None) -> T Asynchronously retrieves the property value.

Args:

instance: The instance from which the property is accessed. owner: The owner class of the property.

Returns:

The property value.

The original docstring for get() is shown below:

_ASyncPropertyDescriptorBase.get(self, instance: I, owner: Optional[Type[I]] = None) -> T Asynchronously retrieves the property value.

Args:

instance: The instance from which the property is accessed. owner: The owner class of the property.

Returns:

The property value.

Type:

_ASyncPropertyDescriptorBase.get(self, instance

Type:

I, owner

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

__symbol__: HiddenMethodDescriptor[Self, str]

Optional[Type[I]] = None) -> T Asynchronously retrieves the property value.

Args:

instance: The instance from which the property is accessed. owner: The owner class of the property.

Returns:

The property value.

The original docstring for get() is shown below:

_ASyncPropertyDescriptorBase.get(self, instance: I, owner: Optional[Type[I]] = None) -> T Asynchronously retrieves the property value.

Args:

instance: The instance from which the property is accessed. owner: The owner class of the property.

Returns:

The property value.

Type:

_ASyncPropertyDescriptorBase.get(self, instance

Type:

I, owner

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

address: ChecksumAddress

The contract address of the token.

asynchronous: bool = False

A boolean indicating whether the instance will be used asynchronously.

balance_of

Query the balance of the token for a given address at a specific block.

Parameters:
Returns:

The balance of the token held by address at block block.

Return type:

int

Examples

>>> token = ERC20("0x1234567890abcdef1234567890abcdef12345678")
>>> await token.balance_of("0xabcdefabcdefabcdefabcdefabcdefabcdef")
500000000000000000000
balance_of_readable
Parameters:
Return type:

Decimal

balances[source]

Get the balances of the components in the TokenSet.

This method retrieves the balances of the components in the TokenSet by first checking if the contract has the getUnits method. If getUnits is available, it is used to determine the balances. If not, and if the contract has the getTotalComponentRealUnits method, it falls back to using that method.

Parameters:

  • block (int | BlockNumber | None) – The block number to query. Defaults to the latest block.

  • skip_cache (bool) – If True, skip using the cache while fetching balance data.

Returns:

A list of WeiBalance instances representing the balances of the components.

Return type:

List[WeiBalance]

Examples

>>> tokenset = TokenSet("0x1234567890abcdef1234567890abcdef12345678")
>>> balances = tokenset.balances()

See also

  • WeiBalance

build_name

Get the contract’s build name.

Returns:

The contract’s build name.

Examples

>>> contract = ContractBase("0x1234567890abcdef1234567890abcdef12345678")
>>> await contract.build_name
'MyContract'
components[source]

Get the components of the TokenSet.

Args:

block: The block number to query. Defaults to the latest block.

Returns:

A list of ERC20 instances representing the components of the TokenSet.

Examples:
>>> tokenset = TokenSet("0x1234567890abcdef1234567890abcdef12345678")
>>> components = tokenset.components()

Since components is an ASyncFunctionAsyncDefault, you can optionally pass sync=True or asynchronous=False to force it to run synchronously and return a value. Without either kwarg, it will return a coroutine for you to await.

Parameters:

block (int | BlockNumber | None)

Return type:

List[ERC20]

property contract: Contract
decimals

The number of decimal places for the token.

Returns:

The number of decimal places for the token.

Examples

>>> token = ERC20("0x1234567890abcdef1234567890abcdef12345678")
>>> await token.decimals
18
deploy_block: ASyncBoundMethod[Self, Any, int]

Get the block number when the contract was deployed.

Parameters:

when_no_history_return_0 (bool) – If True, return 0 when no history is found instead of raising an exception.

Returns:

The block number when the contract was deployed, or 0 if when_no_history_return_0 is True and the deploy block cannot be determined.

Return type:

int

Examples

>>> contract = ContractBase("0x1234567890abcdef1234567890abcdef12345678")
>>> await contract.deploy_block()
1234567

See also

  • contract_creation_block_async()

get_price[source]

Get the USD price of the TokenSet.

This method computes the price of the TokenSet using one of two approaches depending on the underlying contract’s available methods: 1. If the contract has a getUnits method: The price is calculated by fetching the balances of

the components, converting them to their USD values, summing these values to obtain the total value (TVL), and then dividing the TVL by the total supply of the TokenSet.

  1. If the contract has a getTotalComponentRealUnits method: The price is computed as the direct sum of the USD values of its components without normalization by the total supply.

Parameters:

  • block (int | BlockNumber | None) – The block number to query. Defaults to the latest block.

  • skip_cache (bool) – If True, skip using the cache while fetching price data.

Return type:

UsdPrice

Examples

>>> tokenset = TokenSet("0x1234567890abcdef1234567890abcdef12345678")
>>> price = tokenset.get_price()
>>> print(price)
UsdPrice(123.45)

See also

has_method

Check if the contract has a specific method.

Parameters:

  • method (str) – The name of the method to check for.

  • return_response (bool) – If True, return the response of the method call instead of a boolean.

Returns:

A boolean indicating whether the contract has the method, or the response of the method call if return_response is True.

Return type:

bool | Any

Examples

>>> contract = ContractBase("0x1234567890abcdef1234567890abcdef12345678")
>>> await contract.has_method("name()")
True

See also

name

The token’s name.

Returns:

The token’s name.

Examples

>>> token = ERC20("0x1234567890abcdef1234567890abcdef12345678")
>>> await token.name
'TokenName'
price

Get the price of the token in USD.

Parameters:

  • block (optional) – The block number to query. Defaults to latest block.

  • return_None_on_failure (bool) – If True, return None instead of raising a yPriceMagicError on failure.

  • skip_cache (bool) – If True, skip using the cache while fetching price data.

  • ignore_pools (Tuple[UniswapV2Pool | CurvePool, ...]) – An optional tuple of pools to ignore when calculating the price.

Returns:

The price of the token in USD, or None if return_None_on_failure is True and the price cannot be retrieved.

Raises:

yPriceMagicError – If return_None_on_failure is False and the price cannot be retrieved.

Return type:

UsdPrice | None

Examples

>>> token = ERC20("0x1234567890abcdef1234567890abcdef12345678")
>>> await token.price()
1.23

See also

scale

Get the scaling factor for the token.

Returns:

The scaling factor for the token.

Examples

>>> token = ERC20("0x1234567890abcdef1234567890abcdef12345678")
>>> await token.scale
1000000000000000000
symbol

The token’s symbol.

Returns:

The token’s symbol.

Examples

>>> token = ERC20("0x1234567890abcdef1234567890abcdef12345678")
>>> await token.symbol
'TKN'
total_supply

Get the total supply of the token.

Parameters:

block (optional) – The block number to query. Defaults to latest block.

Returns:

The total supply of the token.

Return type:

int

Examples

>>> token = ERC20("0x1234567890abcdef1234567890abcdef12345678")
>>> await token.total_supply()
1000000000000000000000
total_supply_readable

Get the total supply of the token scaled to a human-readable decimal.

Parameters:

block (optional) – The block number to query.

Returns:

The total supply of the token scaled to a decimal.

Return type:

float

Examples

>>> token = ERC20("0x1234567890abcdef1234567890abcdef12345678")
>>> await token.total_supply_readable()
1000.0
ASyncFunctiony.prices.tokenized_fund.tokensets.get_price(token: Union[str, hexbytes.main.HexBytes, ~AnyAddress, evmspec.data._main.Address, brownie.convert.datatypes.EthAddress, brownie.network.contract.Contract, int], block: Union[int, eth_typing.evm.BlockNumber, NoneType] = None, skip_cache: bool = EnvironmentVariable[bool](name=`YPRICEMAGIC_SKIP_CACHE`, default_value=False, using_default=True)) → y.datatypes.UsdPrice[source]

Get the USD price of a TokenSet.

This function retrieves the price of a TokenSet by instantiating a TokenSet and invoking its get_price() method. The computation method depends on the available contract methods: if the underlying contract has a getUnits method, the price is calculated as the total USD value of its components divided by the total supply. Otherwise, if the contract has a getTotalComponentRealUnits method, the price is computed as the direct sum of the USD values of its components.

Args:

token: The address of the TokenSet. block: The block number to query. Defaults to the latest block. skip_cache: If True, skip using the cache while fetching price data.

Examples:
>>> price = get_price("0x1234567890abcdef1234567890abcdef12345678")
>>> print(price)
UsdPrice(123.45)
See Also:

Since get_price is an ASyncFunctionSyncDefault, you can optionally pass sync=False or asynchronous=True to force it to return a coroutine. Without either kwarg, it will run synchronously.

Parameters:
Return type:

UsdPrice

ASyncFunctiony.prices.tokenized_fund.tokensets.is_token_set(token: str | hexbytes.main.HexBytes | AnyAddress | evmspec.data._main.Address | brownie.convert.datatypes.EthAddress | brownie.network.contract.Contract | int) → bool[source]

Check if a given token is a TokenSet.

A TokenSet is identified by the presence of specific methods in its contract.

Args:

token: The address of the token to check.

Returns:

True if the token is a TokenSet, False otherwise.

Examples:
>>> is_token_set("0x1234567890abcdef1234567890abcdef12345678")
True
See Also:

Since is_token_set is an ASyncFunctionSyncDefault, you can optionally pass sync=False or asynchronous=True to force it to return a coroutine. Without either kwarg, it will run synchronously.

Parameters:

token (str | HexBytes | AnyAddress | Address | EthAddress | Contract | int)

Return type:

bool

Module contents