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[name=`YPRICEMAGIC_SKIP_CACHE`, type=bool, default_value=False, current_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
.- Raises:
PriceError: If unable to fetch the price.
- Example:
>>> price = get_price("0x0ac58Df435D3dC9F6e079B2C5F358A4b7e861B69") >>> print(price) 1.05
- See Also:
is_basketdao_index()
for checking if a token is a BasketDAO token.
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:
address (ChecksumAddress)
block (int | BlockNumber | None)
skip_cache (bool)
- Return type:
- 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
- Raises:
ContractLogicError: If the contract logic fails during the call. ValueError: If the call returns an unexpected value.
- 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:
y.prices.tokenized_fund.gelato module
- ASyncFunctiony.prices.tokenized_fund.gelato.get_price(token: Union[str, hexbytes.main.HexBytes, ~AnyAddress, brownie.convert.datatypes.EthAddress, brownie.network.contract.Contract, int], block: Union[int, eth_typing.evm.BlockNumber, NoneType] = None, skip_cache: bool = <EnvironmentVariable[name=`YPRICEMAGIC_SKIP_CACHE`, type=bool, default_value=False, current_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:
token (str | HexBytes | AnyAddress | EthAddress | Contract | int)
block (int | BlockNumber | None)
skip_cache (bool)
- Return type:
- ASyncFunctiony.prices.tokenized_fund.gelato.is_gelato_pool(token_address: str | hexbytes.main.HexBytes | AnyAddress | 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 | EthAddress | Contract | int)
- Return type:
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:
bpool (str | HexBytes | AnyAddress | EthAddress) – The address of the Balancer pool.
token (ERC20) – The
ERC20
token to query.block (int | BlockNumber | None) – The block number to query the balance at. Defaults to the latest block.
- Return type:
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:
pie_address (str | HexBytes | AnyAddress | EthAddress) – The address of the PieDAO token.
block (int | BlockNumber | None) – The block number to query the pool at. Defaults to the latest block.
- Returns:
The address of the Balancer pool, or the PieDAO address if no pool is found.
- Return type:
str | HexBytes | AnyAddress | EthAddress
Example
>>> get_bpool("0x1234567890abcdef1234567890abcdef12345678") '0xBpoolAddress'
- ASyncFunctiony.prices.tokenized_fund.piedao.get_price(pie: Union[str, hexbytes.main.HexBytes, ~AnyAddress, brownie.convert.datatypes.EthAddress, brownie.network.contract.Contract, int], block: Union[int, eth_typing.evm.BlockNumber, NoneType] = None, skip_cache: bool = <EnvironmentVariable[name=`YPRICEMAGIC_SKIP_CACHE`, type=bool, default_value=False, current_value=False, using_default=True]>) y.datatypes.UsdPrice [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:
ERC20
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:
pie (str | HexBytes | AnyAddress | EthAddress | Contract | int)
block (int | BlockNumber | None)
skip_cache (bool)
- Return type:
- async y.prices.tokenized_fund.piedao.get_tokens(pie_address, block=None)[source]
Get the list of tokens in a PieDAO token.
- Parameters:
pie_address (str | HexBytes | AnyAddress | EthAddress) – The address of the PieDAO token.
block (int | BlockNumber | None) – The block number to query the tokens at. Defaults to the latest block.
- Returns:
A list of
ERC20
objects representing the tokens in the PieDAO token.- Return type:
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[name=`YPRICEMAGIC_SKIP_CACHE`, type=bool, default_value=False, current_value=False, using_default=True]>)[source]
Get the total value locked (TVL) in a PieDAO token in USD.
- Parameters:
pie_address (str | HexBytes | AnyAddress | 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:
Example
>>> get_tvl("0x1234567890abcdef1234567890abcdef12345678") 1000000.00
See also
- async y.prices.tokenized_fund.piedao.get_value(bpool, token, block=None, skip_cache=<EnvironmentVariable[name=`YPRICEMAGIC_SKIP_CACHE`, type=bool, default_value=False, current_value=False, using_default=True]>)[source]
Get the USD value of a token in a Balancer pool.
- Parameters:
bpool (str | HexBytes | AnyAddress | 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:
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 | 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 | EthAddress | Contract | int)
- Return type:
y.prices.tokenized_fund.reserve module
- ASyncFunctiony.prices.tokenized_fund.reserve.get_price(token_address: Union[str, hexbytes.main.HexBytes, ~AnyAddress, brownie.convert.datatypes.EthAddress], block: Union[int, eth_typing.evm.BlockNumber, NoneType] = None, skip_cache: bool = <EnvironmentVariable[name=`YPRICEMAGIC_SKIP_CACHE`, type=bool, default_value=False, current_value=False, using_default=True]>) decimal.Decimal [source]
Get the price of a Reserve Protocol R-token in USD.
- Args:
token_address: The address of the R-token. block (optional): The block number to query. Defaults to None (latest). skip_cache (optional): Whether to skip cache. Defaults to
ENVS.SKIP_CACHE
.- Returns:
The price of the R-token in USD.
- Raises:
TypeError: If the token is not a valid R-token. Exception: If unable to calculate the price.
- Example:
>>> address = "0xaCeeD87BD5754c3d714F3Bd43a9B7B0C9250ab0D" # RSV token >>> price = await get_price(address, sync=False) >>> print(price) 1.00
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:
token_address (str | HexBytes | AnyAddress | EthAddress)
block (int | BlockNumber | None)
skip_cache (bool)
- Return type:
- ASyncFunctiony.prices.tokenized_fund.reserve.is_rtoken(token_address: str | hexbytes.main.HexBytes | AnyAddress | brownie.convert.datatypes.EthAddress) bool [source]
Check if the 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) >>> is_r = is_rtoken(address) >>> print(is_r) False
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 | EthAddress)
- Return type:
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.
- ASyncFunction_decimals(block: int | eth_typing.evm.BlockNumber | NoneType = None) int
used to fetch decimals at specific block
Since _decimals 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:
- ASyncFunction_scale(block: int | eth_typing.evm.BlockNumber | NoneType = None) int
Since _scale 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:
- __eq__(_ContractBase__o)
Return self==value.
- __init__(address, *, asynchronous=False)[source]
Initialize a TokenSet instance.
- Parameters:
address (str | HexBytes | AnyAddress | EthAddress | Contract | int) – The address of the TokenSet.
asynchronous (bool) – If True, the instance will be used asynchronously.
Examples
>>> tokenset = TokenSet("0x1234567890abcdef1234567890abcdef12345678")
- __str__()
Return the contract address as a string.
- Returns:
The contract address as a string.
- Return type:
- __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.
- __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.
- __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.
- __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.
- address: str | HexBytes | AnyAddress | EthAddress
The contract address of the token.
- balance_of
Query the balance of the token for a given address at a specific block.
- Parameters:
address (str | HexBytes | AnyAddress | EthAddress | Contract | int) – The address to query.
block (optional) – The block number to query. Defaults to latest block.
- Returns:
The balance of the token held by address at block block.
- Return type:
Examples
>>> token = ERC20("0x1234567890abcdef1234567890abcdef12345678") >>> await token.balance_of("0xabcdefabcdefabcdefabcdefabcdefabcdef") 500000000000000000000
- balance_of_readable
- Parameters:
address (str | HexBytes | AnyAddress | EthAddress | Contract | int)
block (int | BlockNumber | None)
- Return type:
- 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:
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:
- 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:
Examples
>>> contract = ContractBase("0x1234567890abcdef1234567890abcdef12345678") >>> await contract.deploy_block() 1234567
See also
contract_creation_block_async()
- get_price[source]
Get the price of the TokenSet in USD.
This method calculates the price of the TokenSet by summing the USD values of its components and dividing 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:
Examples
>>> tokenset = TokenSet("0x1234567890abcdef1234567890abcdef12345678") >>> price = tokenset.get_price()
- has_method
Check if the contract has a specific method.
- Parameters:
- Returns:
A boolean indicating whether the contract has the method, or the response of the method call if return_response is True.
- Return type:
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:
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:
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, brownie.convert.datatypes.EthAddress, brownie.network.contract.Contract, int], block: Union[int, eth_typing.evm.BlockNumber, NoneType] = None, skip_cache: bool = <EnvironmentVariable[name=`YPRICEMAGIC_SKIP_CACHE`, type=bool, default_value=False, current_value=False, using_default=True]>) y.datatypes.UsdPrice [source]
Get the price of a TokenSet in USD.
This function retrieves the price of a TokenSet by creating an instance of the
TokenSet
class and calling its get_price method.- 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:
>>> get_price("0x1234567890abcdef1234567890abcdef12345678") 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:
token (str | HexBytes | AnyAddress | EthAddress | Contract | int)
block (int | BlockNumber | None)
skip_cache (bool)
- Return type:
- ASyncFunctiony.prices.tokenized_fund.tokensets.is_token_set(token: str | hexbytes.main.HexBytes | AnyAddress | 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 | EthAddress | Contract | int)
- Return type: