y.prices.dex package
Subpackages
- y.prices.dex.balancer package
- Submodules
- y.prices.dex.balancer.balancer module
BalancerMultiplexerBalancerMultiplexer.__init__()BalancerMultiplexer.__v1__BalancerMultiplexer.__v2__BalancerMultiplexer.__versions__BalancerMultiplexer.get_pool_priceBalancerMultiplexer.get_priceBalancerMultiplexer.get_versionBalancerMultiplexer.is_balancer_poolBalancerMultiplexer.v1BalancerMultiplexer.v2BalancerMultiplexer.versions
- y.prices.dex.balancer.v1 module
BalancerV1BalancerV1PoolBalancerV1Pool.__eq__()BalancerV1Pool.__init__()BalancerV1Pool.__str__()BalancerV1Pool.__build_name__BalancerV1Pool.__decimals__BalancerV1Pool.__name__BalancerV1Pool.__scale__BalancerV1Pool.__symbol__BalancerV1Pool.__tokens__BalancerV1Pool.addressBalancerV1Pool.asynchronousBalancerV1Pool.balance_ofBalancerV1Pool.balance_of_readableBalancerV1Pool.build_nameBalancerV1Pool.check_liquidityBalancerV1Pool.contractBalancerV1Pool.decimalsBalancerV1Pool.deploy_blockBalancerV1Pool.get_balanceBalancerV1Pool.get_balancesBalancerV1Pool.get_pool_priceBalancerV1Pool.get_tvlBalancerV1Pool.has_methodBalancerV1Pool.nameBalancerV1Pool.priceBalancerV1Pool.scaleBalancerV1Pool.symbolBalancerV1Pool.tokensBalancerV1Pool.total_supplyBalancerV1Pool.total_supply_readable
- y.prices.dex.balancer.v2 module
BalancerEventsBalancerEvents.__origin__BalancerEvents.obj_typeBalancerEvents.__aiter__()BalancerEvents.__await__()BalancerEvents.__init__()BalancerEvents.__iter__()BalancerEvents._objects_thru()BalancerEvents.events()BalancerEvents.filter()BalancerEvents.logs()BalancerEvents.objects()BalancerEvents.sort()BalancerEvents.wrap()BalancerEvents.__args__BalancerEvents.__pyx_vtable__BalancerEvents.__wrapped__BalancerEvents.addressesBalancerEvents.asynchronousBalancerEvents.bulk_insertBalancerEvents.cacheBalancerEvents.executorBalancerEvents.from_blockBalancerEvents.insert_to_dbBalancerEvents.is_reusableBalancerEvents.materializedBalancerEvents.semaphoreBalancerEvents.to_blockBalancerEvents.topics
BalancerV2BalancerV2PoolBalancerV2Pool.__eq__()BalancerV2Pool.__init__()BalancerV2Pool.__str__()BalancerV2Pool.__build_name__BalancerV2Pool.__decimals__BalancerV2Pool.__id__BalancerV2Pool.__name__BalancerV2Pool.__pool_type__BalancerV2Pool.__scale__BalancerV2Pool.__symbol__BalancerV2Pool.__vault__BalancerV2Pool.addressBalancerV2Pool.asynchronousBalancerV2Pool.balance_ofBalancerV2Pool.balance_of_readableBalancerV2Pool.build_nameBalancerV2Pool.contractBalancerV2Pool.decimalsBalancerV2Pool.deploy_blockBalancerV2Pool.get_balanceBalancerV2Pool.get_balancesBalancerV2Pool.get_pool_priceBalancerV2Pool.get_token_priceBalancerV2Pool.get_tvlBalancerV2Pool.has_methodBalancerV2Pool.idBalancerV2Pool.nameBalancerV2Pool.pool_typeBalancerV2Pool.priceBalancerV2Pool.scaleBalancerV2Pool.symbolBalancerV2Pool.tokensBalancerV2Pool.total_supplyBalancerV2Pool.total_supply_readableBalancerV2Pool.vaultBalancerV2Pool.weights
BalancerV2VaultBalancerV2Vault.__eq__()BalancerV2Vault.__init__()BalancerV2Vault.__str__()BalancerV2Vault.pools()BalancerV2Vault.pools_for_token()BalancerV2Vault.__build_name__BalancerV2Vault.addressBalancerV2Vault.asynchronousBalancerV2Vault.build_nameBalancerV2Vault.contractBalancerV2Vault.deepest_pool_forBalancerV2Vault.deploy_blockBalancerV2Vault.get_pool_infoBalancerV2Vault.get_pool_tokensBalancerV2Vault.has_method
PoolSpecializationPoolSpecialization.__add__()PoolSpecialization.__and__()PoolSpecialization.__bool__()PoolSpecialization.__eq__()PoolSpecialization.__floordiv__()PoolSpecialization.__ge__()PoolSpecialization.__getattribute__()PoolSpecialization.__gt__()PoolSpecialization.__index__()PoolSpecialization.__invert__()PoolSpecialization.__le__()PoolSpecialization.__lshift__()PoolSpecialization.__lt__()PoolSpecialization.__mul__()PoolSpecialization.__or__()PoolSpecialization.__pow__()PoolSpecialization.__radd__()PoolSpecialization.__rand__()PoolSpecialization.__rlshift__()PoolSpecialization.__ror__()PoolSpecialization.__rrshift__()PoolSpecialization.__rshift__()PoolSpecialization.__rxor__()PoolSpecialization.__sizeof__()PoolSpecialization.__sub__()PoolSpecialization.__truediv__()PoolSpecialization.__xor__()PoolSpecialization.from_bytes()PoolSpecialization.to_bytes()PoolSpecialization.with_immutable_tokens()PoolSpecialization.ComposableStablePoolPoolSpecialization.CronV1PoolPoolSpecialization.WeightedPoolPoolSpecialization.WeightedPool2TokensPoolSpecialization.denominatorPoolSpecialization.numerator
- Module contents
- y.prices.dex.uniswap package
- Submodules
- y.prices.dex.uniswap.uniswap module
- y.prices.dex.uniswap.v1 module
- y.prices.dex.uniswap.v2 module
PoolsFromEventsPoolsFromEvents.__origin__PoolsFromEvents.obj_typePoolsFromEvents.__aiter__()PoolsFromEvents.__await__()PoolsFromEvents.__init__()PoolsFromEvents.__iter__()PoolsFromEvents._objects_thru()PoolsFromEvents.events()PoolsFromEvents.filter()PoolsFromEvents.logs()PoolsFromEvents.objects()PoolsFromEvents.pools()PoolsFromEvents.sort()PoolsFromEvents.wrap()PoolsFromEvents.PairCreatedPoolsFromEvents.__args__PoolsFromEvents.__pyx_vtable__PoolsFromEvents.__wrapped__PoolsFromEvents.addressesPoolsFromEvents.asynchronousPoolsFromEvents.bulk_insertPoolsFromEvents.cachePoolsFromEvents.executorPoolsFromEvents.from_blockPoolsFromEvents.insert_to_dbPoolsFromEvents.is_reusablePoolsFromEvents.labelPoolsFromEvents.materializedPoolsFromEvents.semaphorePoolsFromEvents.to_blockPoolsFromEvents.topics
UniswapRouterV2UniswapRouterV2.__eq__()UniswapRouterV2.__init__()UniswapRouterV2.__str__()UniswapRouterV2.pools_for_token()UniswapRouterV2.__build_name__UniswapRouterV2.__pools__UniswapRouterV2.addressUniswapRouterV2.all_pools_forUniswapRouterV2.asynchronousUniswapRouterV2.build_nameUniswapRouterV2.check_liquidityUniswapRouterV2.contractUniswapRouterV2.deepest_poolUniswapRouterV2.deepest_pool_forUniswapRouterV2.deepest_stable_poolUniswapRouterV2.deploy_blockUniswapRouterV2.get_path_to_stablesUniswapRouterV2.get_pools_forUniswapRouterV2.get_priceUniswapRouterV2.get_quoteUniswapRouterV2.has_methodUniswapRouterV2.pools
UniswapV2PoolUniswapV2Pool.__eq__()UniswapV2Pool.__init__()UniswapV2Pool.__str__()UniswapV2Pool.__build_name__UniswapV2Pool.__decimals__UniswapV2Pool.__factory__UniswapV2Pool.__name__UniswapV2Pool.__scale__UniswapV2Pool.__symbol__UniswapV2Pool.__token0__UniswapV2Pool.__token1__UniswapV2Pool.__tokens__UniswapV2Pool.addressUniswapV2Pool.asynchronousUniswapV2Pool.balance_ofUniswapV2Pool.balance_of_readableUniswapV2Pool.build_nameUniswapV2Pool.check_liquidityUniswapV2Pool.contractUniswapV2Pool.decimalsUniswapV2Pool.deploy_blockUniswapV2Pool.factoryUniswapV2Pool.get_priceUniswapV2Pool.get_reservesUniswapV2Pool.get_token_outUniswapV2Pool.has_methodUniswapV2Pool.is_uniswap_poolUniswapV2Pool.nameUniswapV2Pool.priceUniswapV2Pool.reservesUniswapV2Pool.scaleUniswapV2Pool.symbolUniswapV2Pool.token0UniswapV2Pool.token1UniswapV2Pool.tokensUniswapV2Pool.total_supplyUniswapV2Pool.total_supply_readableUniswapV2Pool.tvl
log_debug()log_liquidity()
- y.prices.dex.uniswap.v2_forks module
- y.prices.dex.uniswap.v3 module
SlipstreamPoolsSlipstreamPools.__origin__SlipstreamPools.obj_typeSlipstreamPools.__aiter__()SlipstreamPools.__await__()SlipstreamPools.__init__()SlipstreamPools.__iter__()SlipstreamPools._objects_thru()SlipstreamPools.events()SlipstreamPools.filter()SlipstreamPools.logs()SlipstreamPools.objects()SlipstreamPools.sort()SlipstreamPools.wrap()SlipstreamPools.__args__SlipstreamPools.__pyx_vtable__SlipstreamPools.__wrapped__SlipstreamPools.addressesSlipstreamPools.asynchronousSlipstreamPools.bulk_insertSlipstreamPools.cacheSlipstreamPools.executorSlipstreamPools.from_blockSlipstreamPools.insert_to_dbSlipstreamPools.is_reusableSlipstreamPools.materializedSlipstreamPools.semaphoreSlipstreamPools.to_blockSlipstreamPools.topics
UniV3PoolsUniV3Pools.__origin__UniV3Pools.obj_typeUniV3Pools.__aiter__()UniV3Pools.__await__()UniV3Pools.__init__()UniV3Pools.__iter__()UniV3Pools._objects_thru()UniV3Pools.events()UniV3Pools.filter()UniV3Pools.logs()UniV3Pools.objects()UniV3Pools.sort()UniV3Pools.wrap()UniV3Pools.__args__UniV3Pools.__pyx_vtable__UniV3Pools.__wrapped__UniV3Pools.addressesUniV3Pools.asynchronousUniV3Pools.bulk_insertUniV3Pools.cacheUniV3Pools.executorUniV3Pools.from_blockUniV3Pools.insert_to_dbUniV3Pools.is_reusableUniV3Pools.materializedUniV3Pools.semaphoreUniV3Pools.to_blockUniV3Pools.topics
UniswapV3UniswapV3PoolUniswapV3Pool._check_liquidity_token_outUniswapV3Pool.__contains__()UniswapV3Pool.__eq__()UniswapV3Pool.__getitem__()UniswapV3Pool.__init__()UniswapV3Pool.__str__()UniswapV3Pool.__build_name__UniswapV3Pool.addressUniswapV3Pool.asynchronousUniswapV3Pool.build_nameUniswapV3Pool.check_liquidityUniswapV3Pool.contractUniswapV3Pool.deploy_blockUniswapV3Pool.feeUniswapV3Pool.has_methodUniswapV3Pool.tick_spacingUniswapV3Pool.token0UniswapV3Pool.token1
log_liquidity()
- Module contents
Submodules
y.prices.dex.genericamm module
- final class y.prices.dex.genericamm.GenericAmm[source]
Bases:
ASyncGenericBaseA class for handling generic Automated Market Maker (AMM) Liquidity Pool (LP) tokens.
This class provides methods to interact with and price generic AMM LP token contracts.
- __init__(*, asynchronous=False)[source]
Initialize the GenericAmm instance.
- Parameters:
asynchronous (optional) – Whether methods will return coroutines by default. Defaults to False.
- Return type:
None
- get_price[source]
Get the price of the LP token in USD.
- Parameters:
lp_token_address – The address of the LP token.
block (optional) – The block number to query. Defaults to None (latest).
skip_cache (optional) – Whether to skip cache. Defaults to
ENVS.SKIP_CACHE.lp_token (str | HexBytes | AnyAddress | Address | EthAddress | Contract | int)
- Returns:
The price of the LP token in USD.
- Return type:
Example
>>> amm = GenericAmm(asynchronous=True) >>> address = "0x6B175474E89094C44Da98b954EedeAC495271d0F" >>> price = await amm.get_price(address) >>> print(price) 0.5
See also
ERC20.total_supply_readable()
- get_tokens[source]
Get the tokens in the AMM pool.
- Args:
lp_token_address: The address of the LP token.
- Returns:
A tuple containing the two ERC20 tokens in the pool.
- Example:
>>> amm = GenericAmm(asynchronous=True) >>> address = "0x6B175474E89094C44Da98b954EedeAC495271d0F" >>> tokens = await amm.get_tokens(address) >>> print(tokens) (ERC20(0x123...), ERC20(0x456...))
- See Also:
ERC20
Since get_tokens 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:
lp_token_address (ChecksumAddress)
- Return type:
- get_tvl[source]
Get the Total Value Locked (TVL) in the AMM pool.
- Parameters:
lp_token_address (ChecksumAddress) – The address of the LP 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 Total Value Locked in USD.
- Return type:
Example
>>> amm = GenericAmm(asynchronous=True) >>> address = "0x6B175474E89094C44Da98b954EedeAC495271d0F" >>> tvl = await amm.get_tvl(address) >>> print(tvl) 1000000.0
See also
- async y.prices.dex.genericamm.is_generic_amm(lp_token_address)[source]
Check if the given address is a generic AMM LP token.
- Parameters:
lp_token_address (str | HexBytes | AnyAddress | Address | EthAddress | Contract | int) – The address to check.
- Returns:
True if the address is a generic AMM LP token, False otherwise.
- Return type:
Example
>>> address = "0x6B175474E89094C44Da98b954EedeAC495271d0F" >>> is_amm = await is_generic_amm(address) >>> print(is_amm) False
- y.prices.dex.genericamm.generic_amm = <y.prices.dex.genericamm.GenericAmm object>
A global instance of
GenericAmmwith asynchronous mode enabled.
y.prices.dex.mooniswap module
- ASyncFunctiony.prices.dex.mooniswap.get_pool_price(token: Union[str, faster_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 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
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:
Since get_pool_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 | Address | EthAddress | Contract | int)
block (int | BlockNumber | None)
skip_cache (bool)
- Return type:
- ASyncFunctiony.prices.dex.mooniswap.is_mooniswap_pool(token: str | faster_hexbytes.main.HexBytes | AnyAddress | evmspec.data._main.Address | brownie.convert.datatypes.EthAddress | brownie.network.contract.Contract | int) bool[source]
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:
Since is_mooniswap_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 (str | HexBytes | AnyAddress | Address | EthAddress | Contract | int)
- Return type:
y.prices.dex.solidly module
- class y.prices.dex.solidly.SolidlyPool[source]
Bases:
UniswapV2PoolRepresents a liquidity pool in the Solidly protocol.
This class inherits from
UniswapV2Pooland provides the same interface for interacting with liquidity pools.See also
- __eq__(_ContractBase__o)
Return self==value.
- __init__(self)
- Parameters:
address (str | HexBytes | AnyAddress | Address | EthAddress | Contract | int)
token0 (str | HexBytes | AnyAddress | Address | EthAddress | None)
token1 (str | HexBytes | AnyAddress | Address | EthAddress | None)
deploy_block (int | None)
asynchronous (bool)
- __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.
- Parameters:
- Returns:
The property value.
- Return type:
T
The original docstring for
get()is shown below:_ASyncPropertyDescriptorBase.get(self, instance: I, owner: Optional[Type[I]] = None) -> T
Asynchronously retrieves the property value.
- __decimals__: HiddenMethodDescriptor[Self, int]
Optional[Type[I]] = None) -> T
Asynchronously retrieves the property value.
- Parameters:
- Returns:
The property value.
- Return type:
T
The original docstring for
get()is shown below:_ASyncPropertyDescriptorBase.get(self, instance: I, owner: Optional[Type[I]] = None) -> T
Asynchronously retrieves the property value.
- __factory__: HiddenMethodDescriptor[Self, str | HexBytes | AnyAddress | Address | EthAddress]
Optional[Type[I]] = None) -> T
Asynchronously retrieves the property value.
- Parameters:
- Returns:
The property value.
- Return type:
T
The original docstring for
get()is shown below:_ASyncPropertyDescriptorBase.get(self, instance: I, owner: Optional[Type[I]] = None) -> T
Asynchronously retrieves the property value.
- __scale__: HiddenMethodDescriptor[Self, int]
Optional[Type[I]] = None) -> T
Asynchronously retrieves the property value.
- Parameters:
- Returns:
The property value.
- Return type:
T
The original docstring for
get()is shown below:_ASyncPropertyDescriptorBase.get(self, instance: I, owner: Optional[Type[I]] = None) -> T
Asynchronously retrieves the property value.
- __symbol__: HiddenMethodDescriptor[Self, str]
Optional[Type[I]] = None) -> T
Asynchronously retrieves the property value.
- Parameters:
- Returns:
The property value.
- Return type:
T
The original docstring for
get()is shown below:_ASyncPropertyDescriptorBase.get(self, instance: I, owner: Optional[Type[I]] = None) -> T
Asynchronously retrieves the property value.
- __token0__: HiddenMethodDescriptor[Self, ERC20]
Optional[Type[I]] = None) -> T
Asynchronously retrieves the property value.
- Parameters:
- Returns:
The property value.
- Return type:
T
The original docstring for
get()is shown below:_ASyncPropertyDescriptorBase.get(self, instance: I, owner: Optional[Type[I]] = None) -> T
Asynchronously retrieves the property value.
- __token1__: HiddenMethodDescriptor[Self, ERC20]
Optional[Type[I]] = None) -> T
Asynchronously retrieves the property value.
- Parameters:
- Returns:
The property value.
- Return type:
T
The original docstring for
get()is shown below:_ASyncPropertyDescriptorBase.get(self, instance: I, owner: Optional[Type[I]] = None) -> T
Asynchronously retrieves the property value.
- __tokens__: HiddenMethodDescriptor[Self, Tuple[ERC20, ERC20]]
Optional[Type[I]] = None) -> T
Asynchronously retrieves the property value.
- Parameters:
- Returns:
The property value.
- Return type:
T
The original docstring for
get()is shown below:_ASyncPropertyDescriptorBase.get(self, instance: I, owner: Optional[Type[I]] = None) -> T
Asynchronously retrieves the property value.
- address: ChecksumAddress
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 | Address | 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 | Address | EthAddress | Contract | int)
block (int | BlockNumber | None)
- Return type:
- build_name
Get the contract’s build name.
- Returns:
The contract’s build name.
Examples
>>> contract = ContractBase("0x1234567890abcdef1234567890abcdef12345678") >>> await contract.build_name 'MyContract'
- check_liquidity
Check the liquidity of a specific token in the pool at a given block.
- Args:
token: The address of the token to check. block: The block number to query.
- Returns:
The liquidity of the token in the pool.
- Raises:
TokenNotFound: If the token is not one of the two tokens in the liquidity pool.
- Examples:
>>> pool = UniswapV2Pool("0xAddress") >>> liquidity = await pool.check_liquidity("0xTokenAddress", 12345678) >>> print(liquidity)
Since check_liquidity 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:
token (str | HexBytes | AnyAddress | Address | EthAddress)
block (int | BlockNumber)
- 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()
- factory
- get_price
Calculate the price of the pool’s liquidity token in USD.
This method calculates the price of the pool’s liquidity token by dividing the total value locked (TVL) by the total supply of the liquidity token.
- Args:
block: The block number to query. Defaults to the latest block. skip_cache: If True, skip using the cache while fetching price data.
- Examples:
>>> pool = UniswapV2Pool("0xAddress") >>> price = await pool.get_price() >>> print(price)
- See Also:
Since get_price 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)
skip_cache (bool)
- Return type:
UsdPrice | None
- get_reserves
- get_token_out
Since get_token_out 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:
token_in (str | HexBytes | AnyAddress | Address | EthAddress)
- Return type:
- 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
- is_uniswap_pool
Check if this contract is a valid liquidity pool for Uniswap V2 or one of its forks.
- Parameters:
block (optional) – The block number to query. Defaults to latest block.
- Returns:
True if the contract is a valid Uniswap V2 pool, False otherwise.
- Return type:
Examples
>>> pool = UniswapV2Pool("0xAddress") >>> is_valid = await pool.is_uniswap_pool() >>> print(is_valid)
- 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
yPriceMagicErroron 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
- reserves
- Parameters:
block (int | BlockNumber | None)
- Return type:
Tuple[WeiBalance, WeiBalance] | None
- 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'
- token0
- token1
- tokens
- 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
- tvl
Calculate the total value locked (TVL) in the pool in USD.
This method calculates the TVL by summing the USD value of the reserves of the two tokens in the pool.
- 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:
Decimal | None
Examples
>>> pool = UniswapV2Pool("0xAddress") >>> tvl = await pool.tvl() >>> print(tvl)
See also
- class y.prices.dex.solidly.SolidlyRouter[source]
Bases:
SolidlyRouterBaseA router for interacting with the Solidly protocol.
This class extends
SolidlyRouterBaseto provide additional functionality specific to the Solidly protocol, such as determining the appropriate pool for a token pair and calculating swap routes.See also
- __eq__(_ContractBase__o)
Return self==value.
- __init__(self)
- Parameters:
router_address (str | HexBytes | AnyAddress | Address | EthAddress | Contract | int)
asynchronous (bool)
- Return type:
None
- __str__()
Return the contract address as a string.
- Returns:
The contract address as a string.
- Return type:
- pools_for_token(token_address, block=None, _ignore_pools=())
- Parameters:
token_address (str | HexBytes | AnyAddress | Address | EthAddress)
block (int | BlockNumber | None)
_ignore_pools (Tuple[UniswapV2Pool, ...])
- Return type:
- __build_name__: HiddenMethodDescriptor[Self, str]
Optional[Type[I]] = None) -> T
Asynchronously retrieves the property value.
- Parameters:
- Returns:
The property value.
- Return type:
T
The original docstring for
get()is shown below:_ASyncPropertyDescriptorBase.get(self, instance: I, owner: Optional[Type[I]] = None) -> T
Asynchronously retrieves the property value.
- __pools__: HiddenMethodDescriptor[Self, List[UniswapV2Pool]]
Optional[Type[I]] = None) -> T
Asynchronously retrieves the property value.
- Parameters:
- Returns:
The property value.
- Return type:
T
The original docstring for
get()is shown below:_ASyncPropertyDescriptorBase.get(self, instance: I, owner: Optional[Type[I]] = None) -> T
Asynchronously retrieves the property value.
- address
The address of the contract.
- all_pools_for
Since all_pools_for 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:
token_in (str | HexBytes | AnyAddress | Address | EthAddress)
- Return type:
Dict[UniswapV2Pool, str | HexBytes | AnyAddress | Address | EthAddress]
- build_name
Get the contract’s build name.
- Returns:
The contract’s build name.
Examples
>>> contract = ContractBase("0x1234567890abcdef1234567890abcdef12345678") >>> await contract.build_name 'MyContract'
- check_liquidity
Since check_liquidity 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:
token (str | HexBytes | AnyAddress | Address | EthAddress)
block (int | BlockNumber)
- Return type:
- deepest_pool
returns the deepest pool for token_address at block, excluding pools in _ignore_pools
Since deepest_pool 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:
token_address (str | HexBytes | AnyAddress | Address | EthAddress | Contract | int)
block (int | BlockNumber | None)
_ignore_pools (Tuple[UniswapV2Pool, ...])
- Return type:
UniswapV2Pool | None
- deepest_pool_for
Since deepest_pool_for 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:
token (str | HexBytes | AnyAddress | Address | EthAddress)
block (int | BlockNumber)
- Return type:
Tuple[str | HexBytes | AnyAddress | Address | EthAddress, int]
- deepest_stable_pool
returns the deepest pool for token_address at block which has token_address paired with a stablecoin, excluding pools in _ignore_pools
Since deepest_stable_pool 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:
token_address (str | HexBytes | AnyAddress | Address | EthAddress | Contract | int)
block (int | BlockNumber | None)
_ignore_pools (Tuple[UniswapV2Pool, ...])
- Return type:
UniswapV2Pool | None
- 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_path_to_stables
Since get_path_to_stables 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:
token (str | HexBytes | AnyAddress | Address | EthAddress | Contract | int)
block (int | BlockNumber | None)
_loop_count (int)
_ignore_pools (Tuple[UniswapV2Pool, ...])
- Return type:
List[str | HexBytes | AnyAddress | Address | EthAddress | Contract]
- get_pool[source]
Get the pool object for a given token pair.
This method returns a
SolidlyPoolobject representing the pool that contains the specified input and output tokens, and indicates whether the pool is stable.- Args:
input_token: The address of the input token. output_token: The address of the output token. stable: A boolean indicating whether to look for a stable pool. block: The block number to query.
- Returns:
A
SolidlyPoolobject if the pool exists, otherwise None.- Examples:
>>> router = SolidlyRouter("0xRouterAddress") >>> pool = await router.get_pool("0xTokenA", "0xTokenB", True, 12345678) >>> print(pool)
Since get_pool 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:
input_token (str | HexBytes | AnyAddress | Address | EthAddress)
output_token (str | HexBytes | AnyAddress | Address | EthAddress)
stable (bool)
block (int | BlockNumber)
- Return type:
SolidlyPool | None
- get_pools_for
- Parameters:
token_in (str | HexBytes | AnyAddress | Address | EthAddress)
block (int | BlockNumber | None)
- Return type:
Dict[UniswapV2Pool, str | HexBytes | AnyAddress | Address | EthAddress]
- get_price
- Calculate a price based on Uniswap Router quote for selling one token_in.
Always uses intermediate WETH pair if [token_in,weth,token_out] swap path available.
Since get_price 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:
token_in (str | HexBytes | AnyAddress | Address | EthAddress)
block (int | BlockNumber | None)
token_out (str | HexBytes | AnyAddress | Address | EthAddress)
paired_against (str | HexBytes | AnyAddress | Address | EthAddress)
skip_cache (bool)
ignore_pools (Tuple[UniswapV2Pool | CurvePool, ...])
- Return type:
UsdPrice | None
- get_quote
Get a price quote for a given input amount and swap path.
This method calculates the output amount for a given input amount and swap path by interacting with the Solidly contract.
- Parameters:
amount_in (int) – The amount of input tokens.
path (List[str | HexBytes | AnyAddress | Address | EthAddress | Contract]) – The swap path as a list of token addresses.
block (int | BlockNumber | None) – The block number to query. Defaults to the latest block.
- Returns:
A tuple containing the output amount and the path used.
- Raises:
Exception – If the call reverts for reasons other than insufficient
input amount or liquidity. –
- Return type:
Examples
>>> router = SolidlyRouterBase("0xRouterAddress") >>> quote = await router.get_quote(1000, ["0xTokenA", "0xTokenB"]) >>> print(quote)
- get_routes_from_path[source]
Determine the swap routes from a given path.
This method calculates the swap routes for a given path by checking for available stable and unstable pools and selecting the deepest pool.
- Parameters:
path (List[str | HexBytes | AnyAddress | Address | EthAddress | Contract]) – The swap path as a list of token addresses.
block (int | BlockNumber) – The block number to query.
- Returns:
A list of tuples, each containing the input token, output token, and a boolean indicating whether the pool is stable.
- Raises:
ValueError – If no pool is found for a token pair.
- Return type:
List[Tuple[str | HexBytes | AnyAddress | Address | EthAddress, str | HexBytes | AnyAddress | Address | EthAddress, bool]]
Examples
>>> router = SolidlyRouter("0xRouterAddress") >>> routes = await router.get_routes_from_path(["0xTokenA", "0xTokenB"], 12345678) >>> print(routes)
- 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
- pair_for[source]
Get the address of the pool for a given token pair.
This method returns the address of the pool that contains the specified input and output tokens, and indicates whether the pool is stable.
- Args:
input_token: The address of the input token. output_token: The address of the output token. stable: A boolean indicating whether to look for a stable pool.
- Returns:
The address of the pool.
- Examples:
>>> router = SolidlyRouter("0xRouterAddress") >>> pool_address = await router.pair_for("0xTokenA", "0xTokenB", True) >>> print(pool_address)
Since pair_for 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:
input_token (str | HexBytes | AnyAddress | Address | EthAddress)
output_token (str | HexBytes | AnyAddress | Address | EthAddress)
stable (bool)
- Return type:
str | HexBytes | AnyAddress | Address | EthAddress
- pools
- class y.prices.dex.solidly.SolidlyRouterBase[source]
Bases:
UniswapRouterV2Solidly is a modified fork of Uni V2. The uniswap_multiplexer is the entrypoint for pricing using this object.
This class provides methods to interact with the Solidly protocol, which is based on the Uniswap V2 model. It includes functionality to get price quotes and determine routes for token swaps.
See also
- __eq__(_ContractBase__o)
Return self==value.
- __init__(self)
- Parameters:
router_address (str | HexBytes | AnyAddress | Address | EthAddress | Contract | int)
asynchronous (bool)
- Return type:
None
- __str__()
Return the contract address as a string.
- Returns:
The contract address as a string.
- Return type:
- pools_for_token(token_address, block=None, _ignore_pools=())
- Parameters:
token_address (str | HexBytes | AnyAddress | Address | EthAddress)
block (int | BlockNumber | None)
_ignore_pools (Tuple[UniswapV2Pool, ...])
- Return type:
- __build_name__: HiddenMethodDescriptor[Self, str]
Optional[Type[I]] = None) -> T
Asynchronously retrieves the property value.
- Parameters:
- Returns:
The property value.
- Return type:
T
The original docstring for
get()is shown below:_ASyncPropertyDescriptorBase.get(self, instance: I, owner: Optional[Type[I]] = None) -> T
Asynchronously retrieves the property value.
- __pools__: HiddenMethodDescriptor[Self, List[UniswapV2Pool]]
Optional[Type[I]] = None) -> T
Asynchronously retrieves the property value.
- Parameters:
- Returns:
The property value.
- Return type:
T
The original docstring for
get()is shown below:_ASyncPropertyDescriptorBase.get(self, instance: I, owner: Optional[Type[I]] = None) -> T
Asynchronously retrieves the property value.
- address
The address of the contract.
- all_pools_for
Since all_pools_for 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:
token_in (str | HexBytes | AnyAddress | Address | EthAddress)
- Return type:
Dict[UniswapV2Pool, str | HexBytes | AnyAddress | Address | EthAddress]
- build_name
Get the contract’s build name.
- Returns:
The contract’s build name.
Examples
>>> contract = ContractBase("0x1234567890abcdef1234567890abcdef12345678") >>> await contract.build_name 'MyContract'
- check_liquidity
Since check_liquidity 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:
token (str | HexBytes | AnyAddress | Address | EthAddress)
block (int | BlockNumber)
- Return type:
- deepest_pool
returns the deepest pool for token_address at block, excluding pools in _ignore_pools
Since deepest_pool 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:
token_address (str | HexBytes | AnyAddress | Address | EthAddress | Contract | int)
block (int | BlockNumber | None)
_ignore_pools (Tuple[UniswapV2Pool, ...])
- Return type:
UniswapV2Pool | None
- deepest_pool_for
Since deepest_pool_for 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:
token (str | HexBytes | AnyAddress | Address | EthAddress)
block (int | BlockNumber)
- Return type:
Tuple[str | HexBytes | AnyAddress | Address | EthAddress, int]
- deepest_stable_pool
returns the deepest pool for token_address at block which has token_address paired with a stablecoin, excluding pools in _ignore_pools
Since deepest_stable_pool 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:
token_address (str | HexBytes | AnyAddress | Address | EthAddress | Contract | int)
block (int | BlockNumber | None)
_ignore_pools (Tuple[UniswapV2Pool, ...])
- Return type:
UniswapV2Pool | None
- 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_path_to_stables
Since get_path_to_stables 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:
token (str | HexBytes | AnyAddress | Address | EthAddress | Contract | int)
block (int | BlockNumber | None)
_loop_count (int)
_ignore_pools (Tuple[UniswapV2Pool, ...])
- Return type:
List[str | HexBytes | AnyAddress | Address | EthAddress | Contract]
- get_pools_for
- Parameters:
token_in (str | HexBytes | AnyAddress | Address | EthAddress)
block (int | BlockNumber | None)
- Return type:
Dict[UniswapV2Pool, str | HexBytes | AnyAddress | Address | EthAddress]
- get_price
- Calculate a price based on Uniswap Router quote for selling one token_in.
Always uses intermediate WETH pair if [token_in,weth,token_out] swap path available.
Since get_price 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:
token_in (str | HexBytes | AnyAddress | Address | EthAddress)
block (int | BlockNumber | None)
token_out (str | HexBytes | AnyAddress | Address | EthAddress)
paired_against (str | HexBytes | AnyAddress | Address | EthAddress)
skip_cache (bool)
ignore_pools (Tuple[UniswapV2Pool | CurvePool, ...])
- Return type:
UsdPrice | None
- get_quote[source]
Get a price quote for a given input amount and swap path.
This method calculates the output amount for a given input amount and swap path by interacting with the Solidly contract.
- Parameters:
amount_in (int) – The amount of input tokens.
path (List[str | HexBytes | AnyAddress | Address | EthAddress | Contract]) – The swap path as a list of token addresses.
block (int | BlockNumber | None) – The block number to query. Defaults to the latest block.
- Returns:
A tuple containing the output amount and the path used.
- Raises:
Exception – If the call reverts for reasons other than insufficient
input amount or liquidity. –
- Return type:
Examples
>>> router = SolidlyRouterBase("0xRouterAddress") >>> quote = await router.get_quote(1000, ["0xTokenA", "0xTokenB"]) >>> print(quote)
- 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
- pools
y.prices.dex.velodrome module
- class y.prices.dex.velodrome.VelodromePool[source]
Bases:
UniswapV2Pool- __eq__(_ContractBase__o)
Return self==value.
- __init__(address, token0=None, token1=None, stable=None, deploy_block=None, *, asynchronous=False)[source]
Initialize a
VelodromePoolinstance.- Parameters:
address (str | HexBytes | AnyAddress | Address | EthAddress | Contract | int) – The address of the pool.
token0 (str | HexBytes | AnyAddress | Address | EthAddress | Contract | int | None) – The address of the first token in the pool.
token1 (str | HexBytes | AnyAddress | Address | EthAddress | Contract | int | None) – The address of the second token in the pool.
stable (bool | None) – Indicates if the pool is stable.
deploy_block (int | None) – The block number at which the pool was deployed.
asynchronous (bool) – Whether to use asynchronous operations.
Examples
>>> pool = VelodromePool("0xPoolAddress", "0xToken0", "0xToken1", True, 12345678) >>> print(pool.is_stable) True
See also
UniswapV2Pool
- __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.
- Parameters:
- Returns:
The property value.
- Return type:
T
The original docstring for
get()is shown below:_ASyncPropertyDescriptorBase.get(self, instance: I, owner: Optional[Type[I]] = None) -> T
Asynchronously retrieves the property value.
- __decimals__: HiddenMethodDescriptor[Self, int]
Optional[Type[I]] = None) -> T
Asynchronously retrieves the property value.
- Parameters:
- Returns:
The property value.
- Return type:
T
The original docstring for
get()is shown below:_ASyncPropertyDescriptorBase.get(self, instance: I, owner: Optional[Type[I]] = None) -> T
Asynchronously retrieves the property value.
- __factory__: HiddenMethodDescriptor[Self, str | HexBytes | AnyAddress | Address | EthAddress]
Optional[Type[I]] = None) -> T
Asynchronously retrieves the property value.
- Parameters:
- Returns:
The property value.
- Return type:
T
The original docstring for
get()is shown below:_ASyncPropertyDescriptorBase.get(self, instance: I, owner: Optional[Type[I]] = None) -> T
Asynchronously retrieves the property value.
- __scale__: HiddenMethodDescriptor[Self, int]
Optional[Type[I]] = None) -> T
Asynchronously retrieves the property value.
- Parameters:
- Returns:
The property value.
- Return type:
T
The original docstring for
get()is shown below:_ASyncPropertyDescriptorBase.get(self, instance: I, owner: Optional[Type[I]] = None) -> T
Asynchronously retrieves the property value.
- __symbol__: HiddenMethodDescriptor[Self, str]
Optional[Type[I]] = None) -> T
Asynchronously retrieves the property value.
- Parameters:
- Returns:
The property value.
- Return type:
T
The original docstring for
get()is shown below:_ASyncPropertyDescriptorBase.get(self, instance: I, owner: Optional[Type[I]] = None) -> T
Asynchronously retrieves the property value.
- __token0__: HiddenMethodDescriptor[Self, ERC20]
Optional[Type[I]] = None) -> T
Asynchronously retrieves the property value.
- Parameters:
- Returns:
The property value.
- Return type:
T
The original docstring for
get()is shown below:_ASyncPropertyDescriptorBase.get(self, instance: I, owner: Optional[Type[I]] = None) -> T
Asynchronously retrieves the property value.
- __token1__: HiddenMethodDescriptor[Self, ERC20]
Optional[Type[I]] = None) -> T
Asynchronously retrieves the property value.
- Parameters:
- Returns:
The property value.
- Return type:
T
The original docstring for
get()is shown below:_ASyncPropertyDescriptorBase.get(self, instance: I, owner: Optional[Type[I]] = None) -> T
Asynchronously retrieves the property value.
- __tokens__: HiddenMethodDescriptor[Self, Tuple[ERC20, ERC20]]
Optional[Type[I]] = None) -> T
Asynchronously retrieves the property value.
- Parameters:
- Returns:
The property value.
- Return type:
T
The original docstring for
get()is shown below:_ASyncPropertyDescriptorBase.get(self, instance: I, owner: Optional[Type[I]] = None) -> T
Asynchronously retrieves the property value.
- address: ChecksumAddress
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 | Address | 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 | Address | EthAddress | Contract | int)
block (int | BlockNumber | None)
- Return type:
- build_name
Get the contract’s build name.
- Returns:
The contract’s build name.
Examples
>>> contract = ContractBase("0x1234567890abcdef1234567890abcdef12345678") >>> await contract.build_name 'MyContract'
- check_liquidity
Check the liquidity of a specific token in the pool at a given block.
- Args:
token: The address of the token to check. block: The block number to query.
- Returns:
The liquidity of the token in the pool.
- Raises:
TokenNotFound: If the token is not one of the two tokens in the liquidity pool.
- Examples:
>>> pool = UniswapV2Pool("0xAddress") >>> liquidity = await pool.check_liquidity("0xTokenAddress", 12345678) >>> print(liquidity)
Since check_liquidity 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:
token (str | HexBytes | AnyAddress | Address | EthAddress)
block (int | BlockNumber)
- 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()
- factory
- get_price
Calculate the price of the pool’s liquidity token in USD.
This method calculates the price of the pool’s liquidity token by dividing the total value locked (TVL) by the total supply of the liquidity token.
- Args:
block: The block number to query. Defaults to the latest block. skip_cache: If True, skip using the cache while fetching price data.
- Examples:
>>> pool = UniswapV2Pool("0xAddress") >>> price = await pool.get_price() >>> print(price)
- See Also:
Since get_price 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)
skip_cache (bool)
- Return type:
UsdPrice | None
- get_reserves
- get_token_out
Since get_token_out 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:
token_in (str | HexBytes | AnyAddress | Address | EthAddress)
- Return type:
- 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
- is_stable
Indicates if the pool is stable, as opposed to volatile.
- is_uniswap_pool
Check if this contract is a valid liquidity pool for Uniswap V2 or one of its forks.
- Parameters:
block (optional) – The block number to query. Defaults to latest block.
- Returns:
True if the contract is a valid Uniswap V2 pool, False otherwise.
- Return type:
Examples
>>> pool = UniswapV2Pool("0xAddress") >>> is_valid = await pool.is_uniswap_pool() >>> print(is_valid)
- 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
yPriceMagicErroron 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
- reserves
- Parameters:
block (int | BlockNumber | None)
- Return type:
Tuple[WeiBalance, WeiBalance] | None
- 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'
- token0
- token1
- tokens
- 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
- tvl
Calculate the total value locked (TVL) in the pool in USD.
This method calculates the TVL by summing the USD value of the reserves of the two tokens in the pool.
- 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:
Decimal | None
Examples
>>> pool = UniswapV2Pool("0xAddress") >>> tvl = await pool.tvl() >>> print(tvl)
See also
- class y.prices.dex.velodrome.VelodromeRouterV2[source]
Bases:
SolidlyRouterBase- __eq__(_ContractBase__o)
Return self==value.
- __init__(*args, **kwargs)[source]
Initialize a
VelodromeRouterV2instance.This class is a specialized router for Velodrome V2, inheriting from
SolidlyRouterBase.- Parameters:
*args – Variable length argument list.
**kwargs – Arbitrary keyword arguments.
- Return type:
None
Examples
>>> router = VelodromeRouterV2() >>> print(router.default_factory) 0xF1046053aa5682b4F9a81b5481394DA16BE5FF5a
See also
SolidlyRouterBase
- __str__()
Return the contract address as a string.
- Returns:
The contract address as a string.
- Return type:
- pools_for_token(token_address, block=None, _ignore_pools=())
- Parameters:
token_address (str | HexBytes | AnyAddress | Address | EthAddress)
block (int | BlockNumber | None)
_ignore_pools (Tuple[UniswapV2Pool, ...])
- Return type:
- __build_name__: HiddenMethodDescriptor[Self, str]
Optional[Type[I]] = None) -> T
Asynchronously retrieves the property value.
- Parameters:
- Returns:
The property value.
- Return type:
T
The original docstring for
get()is shown below:_ASyncPropertyDescriptorBase.get(self, instance: I, owner: Optional[Type[I]] = None) -> T
Asynchronously retrieves the property value.
- __pools__: HiddenMethodDescriptor[Self, Set[VelodromePool]]
Optional[Type[I]] = None) -> T
Asynchronously retrieves the property value.
- Parameters:
- Returns:
The property value.
- Return type:
T
The original docstring for
get()is shown below:_ASyncPropertyDescriptorBase.get(self, instance: I, owner: Optional[Type[I]] = None) -> T
Asynchronously retrieves the property value.
- address
The address of the contract.
- all_pools_for
Since all_pools_for 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:
token_in (str | HexBytes | AnyAddress | Address | EthAddress)
- Return type:
Dict[UniswapV2Pool, str | HexBytes | AnyAddress | Address | EthAddress]
- build_name
Get the contract’s build name.
- Returns:
The contract’s build name.
Examples
>>> contract = ContractBase("0x1234567890abcdef1234567890abcdef12345678") >>> await contract.build_name 'MyContract'
- check_liquidity
Since check_liquidity 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:
token (str | HexBytes | AnyAddress | Address | EthAddress)
block (int | BlockNumber)
- Return type:
- deepest_pool
returns the deepest pool for token_address at block, excluding pools in _ignore_pools
Since deepest_pool 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:
token_address (str | HexBytes | AnyAddress | Address | EthAddress | Contract | int)
block (int | BlockNumber | None)
_ignore_pools (Tuple[UniswapV2Pool, ...])
- Return type:
UniswapV2Pool | None
- deepest_pool_for
Since deepest_pool_for 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:
token (str | HexBytes | AnyAddress | Address | EthAddress)
block (int | BlockNumber)
- Return type:
Tuple[str | HexBytes | AnyAddress | Address | EthAddress, int]
- deepest_stable_pool
returns the deepest pool for token_address at block which has token_address paired with a stablecoin, excluding pools in _ignore_pools
Since deepest_stable_pool 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:
token_address (str | HexBytes | AnyAddress | Address | EthAddress | Contract | int)
block (int | BlockNumber | None)
_ignore_pools (Tuple[UniswapV2Pool, ...])
- Return type:
UniswapV2Pool | None
- default_factory
The default factory address for the current network.
- 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_path_to_stables
Since get_path_to_stables 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:
token (str | HexBytes | AnyAddress | Address | EthAddress | Contract | int)
block (int | BlockNumber | None)
_loop_count (int)
_ignore_pools (Tuple[UniswapV2Pool, ...])
- Return type:
List[str | HexBytes | AnyAddress | Address | EthAddress | Contract]
- get_pool[source]
Get the
VelodromePoolinstance for a given pair of tokens and stability preference.- Args:
input_token: The address of the input token. output_token: The address of the output token. stable: Indicates if a stable pool is preferred. block: The block number to consider.
- Returns:
A
VelodromePoolinstance if the pool exists and the address is a contract, otherwise None.- Examples:
>>> router = VelodromeRouterV2() >>> pool = await router.get_pool("0xTokenA", "0xTokenB", True, 12345678) >>> print(pool) <VelodromePool instance>
- See Also:
Since get_pool 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:
input_token (str | HexBytes | AnyAddress | Address | EthAddress)
output_token (str | HexBytes | AnyAddress | Address | EthAddress)
stable (bool)
block (int | BlockNumber)
- Return type:
VelodromePool | None
- get_pools_for
- Parameters:
token_in (str | HexBytes | AnyAddress | Address | EthAddress)
block (int | BlockNumber | None)
- Return type:
Dict[UniswapV2Pool, str | HexBytes | AnyAddress | Address | EthAddress]
- get_price
- Calculate a price based on Uniswap Router quote for selling one token_in.
Always uses intermediate WETH pair if [token_in,weth,token_out] swap path available.
Since get_price 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:
token_in (str | HexBytes | AnyAddress | Address | EthAddress)
block (int | BlockNumber | None)
token_out (str | HexBytes | AnyAddress | Address | EthAddress)
paired_against (str | HexBytes | AnyAddress | Address | EthAddress)
skip_cache (bool)
ignore_pools (Tuple[UniswapV2Pool | CurvePool, ...])
- Return type:
UsdPrice | None
- get_quote
Get a price quote for a given input amount and swap path.
This method calculates the output amount for a given input amount and swap path by interacting with the Solidly contract.
- Parameters:
amount_in (int) – The amount of input tokens.
path (List[str | HexBytes | AnyAddress | Address | EthAddress | Contract]) – The swap path as a list of token addresses.
block (int | BlockNumber | None) – The block number to query. Defaults to the latest block.
- Returns:
A tuple containing the output amount and the path used.
- Raises:
Exception – If the call reverts for reasons other than insufficient
input amount or liquidity. –
- Return type:
Examples
>>> router = SolidlyRouterBase("0xRouterAddress") >>> quote = await router.get_quote(1000, ["0xTokenA", "0xTokenB"]) >>> print(quote)
- get_routes_from_path[source]
Get the routes for a given path of tokens.
- Parameters:
path (List[str | HexBytes | AnyAddress | Address | EthAddress | Contract]) – A list of token addresses representing the path.
block (int | BlockNumber) – The block number to consider.
- Returns:
A list of tuples, each containing the input token, output token, and stability preference.
- Raises:
NoReservesError – If no route is available for the given path.
- Return type:
List[Tuple[str | HexBytes | AnyAddress | Address | EthAddress, str | HexBytes | AnyAddress | Address | EthAddress, bool]]
Examples
>>> router = VelodromeRouterV2() >>> routes = await router.get_routes_from_path(["0xTokenA", "0xTokenB"], 12345678) >>> print(routes) [("0xTokenA", "0xTokenB", True)]
See also
- 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
- pool_for[source]
Get the pool address for a given pair of tokens and stability preference.
- Args:
input_token: The address of the input token. output_token: The address of the output token. stable: Indicates if a stable pool is preferred.
- Returns:
The address of the pool if it exists, otherwise None.
- Examples:
>>> router = VelodromeRouterV2() >>> pool_address = await router.pool_for("0xTokenA", "0xTokenB", True) >>> print(pool_address) 0xPoolAddress
- See Also:
Since pool_for 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:
input_token (str | HexBytes | AnyAddress | Address | EthAddress)
output_token (str | HexBytes | AnyAddress | Address | EthAddress)
stable (bool)
- Return type:
str | HexBytes | AnyAddress | Address | EthAddress | None
- pools[source]
Fetch all Velodrome pools.
This method retrieves all pools for the Velodrome protocol on the current network.
- Returns:
A set of
VelodromePoolinstances.
Examples
>>> router = VelodromeRouterV2() >>> pools = await router.pools >>> print(len(pools)) 42
See also
- async y.prices.dex.velodrome.is_contract(pool_address)[source]
Check if a given address is a contract.
- Parameters:
pool_address (str | HexBytes | AnyAddress | Address | EthAddress) – The address to check.
- Returns:
True if the address is a contract, otherwise False.
- Return type:
Examples
>>> result = await is_contract("0xPoolAddress") >>> print(result) True