y.prices.dex package
Subpackages
- y.prices.dex.balancer package
- Submodules
- y.prices.dex.balancer.balancer module
BalancerMultiplexer
BalancerMultiplexer.__init__()
BalancerMultiplexer.__v1__
BalancerMultiplexer.__v2__
BalancerMultiplexer.__versions__
BalancerMultiplexer.get_pool_price
BalancerMultiplexer.get_price
BalancerMultiplexer.get_version
BalancerMultiplexer.is_balancer_pool
BalancerMultiplexer.v1
BalancerMultiplexer.v2
BalancerMultiplexer.versions
- y.prices.dex.balancer.v1 module
BalancerV1
BalancerV1Pool
BalancerV1Pool._decimals
BalancerV1Pool._scale
BalancerV1Pool.__eq__()
BalancerV1Pool.__init__()
BalancerV1Pool.__str__()
BalancerV1Pool.__build_name__
BalancerV1Pool.__decimals__
BalancerV1Pool.__name__
BalancerV1Pool.__scale__
BalancerV1Pool.__symbol__
BalancerV1Pool.__tokens__
BalancerV1Pool.address
BalancerV1Pool.asynchronous
BalancerV1Pool.balance_of
BalancerV1Pool.balance_of_readable
BalancerV1Pool.build_name
BalancerV1Pool.check_liquidity
BalancerV1Pool.contract
BalancerV1Pool.decimals
BalancerV1Pool.deploy_block
BalancerV1Pool.get_balance
BalancerV1Pool.get_balances
BalancerV1Pool.get_pool_price
BalancerV1Pool.get_tvl
BalancerV1Pool.has_method
BalancerV1Pool.name
BalancerV1Pool.price
BalancerV1Pool.scale
BalancerV1Pool.symbol
BalancerV1Pool.tokens
BalancerV1Pool.total_supply
BalancerV1Pool.total_supply_readable
- y.prices.dex.balancer.v2 module
BalancerEvents
BalancerEvents.obj_type
BalancerEvents.__aiter__()
BalancerEvents.__await__()
BalancerEvents.__del__()
BalancerEvents.__init__()
BalancerEvents.__iter__()
BalancerEvents.__subclasshook__()
BalancerEvents._objects_thru()
BalancerEvents.events()
BalancerEvents.filter()
BalancerEvents.logs()
BalancerEvents.objects()
BalancerEvents.sort()
BalancerEvents.wrap()
BalancerEvents.__async_property__
BalancerEvents.__orig_bases__
BalancerEvents.__parameters__
BalancerEvents.__wrapped__
BalancerEvents.addresses
BalancerEvents.asynchronous
BalancerEvents.bulk_insert
BalancerEvents.cache
BalancerEvents.executor
BalancerEvents.from_block
BalancerEvents.insert_to_db
BalancerEvents.is_asleep
BalancerEvents.is_reusable
BalancerEvents.materialized
BalancerEvents.semaphore
BalancerEvents.threads
BalancerEvents.to_block
BalancerEvents.topics
BalancerV2
BalancerV2Pool
BalancerV2Pool._decimals
BalancerV2Pool._scale
BalancerV2Pool.__eq__()
BalancerV2Pool.__init__()
BalancerV2Pool.__str__()
BalancerV2Pool.__build_name__
BalancerV2Pool.__decimals__
BalancerV2Pool.__id__
BalancerV2Pool.__name__
BalancerV2Pool.__pool_type__
BalancerV2Pool.__scale__
BalancerV2Pool.__symbol__
BalancerV2Pool.__vault__
BalancerV2Pool.address
BalancerV2Pool.asynchronous
BalancerV2Pool.balance_of
BalancerV2Pool.balance_of_readable
BalancerV2Pool.build_name
BalancerV2Pool.contract
BalancerV2Pool.decimals
BalancerV2Pool.deploy_block
BalancerV2Pool.get_balance
BalancerV2Pool.get_balances
BalancerV2Pool.get_pool_price
BalancerV2Pool.get_token_price
BalancerV2Pool.get_tvl
BalancerV2Pool.has_method
BalancerV2Pool.id
BalancerV2Pool.name
BalancerV2Pool.pool_type
BalancerV2Pool.price
BalancerV2Pool.scale
BalancerV2Pool.symbol
BalancerV2Pool.tokens
BalancerV2Pool.total_supply
BalancerV2Pool.total_supply_readable
BalancerV2Pool.vault
BalancerV2Pool.weights
BalancerV2Vault
BalancerV2Vault.__eq__()
BalancerV2Vault.__init__()
BalancerV2Vault.__str__()
BalancerV2Vault.pools()
BalancerV2Vault.pools_for_token()
BalancerV2Vault.__build_name__
BalancerV2Vault.address
BalancerV2Vault.asynchronous
BalancerV2Vault.build_name
BalancerV2Vault.contract
BalancerV2Vault.deepest_pool_for
BalancerV2Vault.deploy_block
BalancerV2Vault.get_pool_info
BalancerV2Vault.get_pool_tokens
BalancerV2Vault.has_method
PoolSpecialization
PoolSpecialization.__abs__()
PoolSpecialization.__add__()
PoolSpecialization.__and__()
PoolSpecialization.__bool__()
PoolSpecialization.__ceil__()
PoolSpecialization.__divmod__()
PoolSpecialization.__eq__()
PoolSpecialization.__float__()
PoolSpecialization.__floor__()
PoolSpecialization.__floordiv__()
PoolSpecialization.__format__()
PoolSpecialization.__ge__()
PoolSpecialization.__getattribute__()
PoolSpecialization.__getnewargs__()
PoolSpecialization.__gt__()
PoolSpecialization.__index__()
PoolSpecialization.__int__()
PoolSpecialization.__invert__()
PoolSpecialization.__le__()
PoolSpecialization.__lshift__()
PoolSpecialization.__lt__()
PoolSpecialization.__mod__()
PoolSpecialization.__mul__()
PoolSpecialization.__ne__()
PoolSpecialization.__neg__()
PoolSpecialization.__or__()
PoolSpecialization.__pos__()
PoolSpecialization.__pow__()
PoolSpecialization.__radd__()
PoolSpecialization.__rand__()
PoolSpecialization.__rdivmod__()
PoolSpecialization.__rfloordiv__()
PoolSpecialization.__rlshift__()
PoolSpecialization.__rmod__()
PoolSpecialization.__rmul__()
PoolSpecialization.__ror__()
PoolSpecialization.__round__()
PoolSpecialization.__rpow__()
PoolSpecialization.__rrshift__()
PoolSpecialization.__rshift__()
PoolSpecialization.__rsub__()
PoolSpecialization.__rtruediv__()
PoolSpecialization.__rxor__()
PoolSpecialization.__sizeof__()
PoolSpecialization.__sub__()
PoolSpecialization.__truediv__()
PoolSpecialization.__trunc__()
PoolSpecialization.__xor__()
PoolSpecialization.as_integer_ratio()
PoolSpecialization.bit_count()
PoolSpecialization.bit_length()
PoolSpecialization.conjugate()
PoolSpecialization.from_bytes()
PoolSpecialization.to_bytes()
PoolSpecialization.with_immutable_tokens()
PoolSpecialization.ComposableStablePool
PoolSpecialization.CronV1Pool
PoolSpecialization.WeightedPool
PoolSpecialization.WeightedPool2Tokens
PoolSpecialization.denominator
PoolSpecialization.imag
PoolSpecialization.numerator
PoolSpecialization.real
- 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
PoolsFromEvents
PoolsFromEvents.obj_type
PoolsFromEvents.__aiter__()
PoolsFromEvents.__await__()
PoolsFromEvents.__del__()
PoolsFromEvents.__init__()
PoolsFromEvents.__iter__()
PoolsFromEvents.__subclasshook__()
PoolsFromEvents._objects_thru()
PoolsFromEvents.events()
PoolsFromEvents.filter()
PoolsFromEvents.logs()
PoolsFromEvents.objects()
PoolsFromEvents.pools()
PoolsFromEvents.sort()
PoolsFromEvents.wrap()
PoolsFromEvents.PairCreated
PoolsFromEvents.__async_property__
PoolsFromEvents.__orig_bases__
PoolsFromEvents.__parameters__
PoolsFromEvents.__wrapped__
PoolsFromEvents.addresses
PoolsFromEvents.asynchronous
PoolsFromEvents.bulk_insert
PoolsFromEvents.cache
PoolsFromEvents.executor
PoolsFromEvents.from_block
PoolsFromEvents.insert_to_db
PoolsFromEvents.is_asleep
PoolsFromEvents.is_reusable
PoolsFromEvents.label
PoolsFromEvents.materialized
PoolsFromEvents.semaphore
PoolsFromEvents.to_block
PoolsFromEvents.topics
UniswapRouterV2
UniswapRouterV2.__eq__()
UniswapRouterV2.__init__()
UniswapRouterV2.__str__()
UniswapRouterV2.pools_for_token()
UniswapRouterV2.__build_name__
UniswapRouterV2.__pools__
UniswapRouterV2.address
UniswapRouterV2.asynchronous
UniswapRouterV2.build_name
UniswapRouterV2.check_liquidity
UniswapRouterV2.contract
UniswapRouterV2.deepest_pool
UniswapRouterV2.deepest_pool_for
UniswapRouterV2.deepest_stable_pool
UniswapRouterV2.deploy_block
UniswapRouterV2.get_path_to_stables
UniswapRouterV2.get_pools_for
UniswapRouterV2.get_price
UniswapRouterV2.get_quote
UniswapRouterV2.has_method
UniswapRouterV2.pools
UniswapV2Pool
UniswapV2Pool._decimals
UniswapV2Pool._scale
UniswapV2Pool.__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.address
UniswapV2Pool.asynchronous
UniswapV2Pool.balance_of
UniswapV2Pool.balance_of_readable
UniswapV2Pool.build_name
UniswapV2Pool.check_liquidity
UniswapV2Pool.contract
UniswapV2Pool.decimals
UniswapV2Pool.deploy_block
UniswapV2Pool.factory
UniswapV2Pool.get_price
UniswapV2Pool.get_reserves
UniswapV2Pool.get_token_out
UniswapV2Pool.has_method
UniswapV2Pool.is_uniswap_pool
UniswapV2Pool.name
UniswapV2Pool.price
UniswapV2Pool.reserves
UniswapV2Pool.scale
UniswapV2Pool.symbol
UniswapV2Pool.token0
UniswapV2Pool.token1
UniswapV2Pool.tokens
UniswapV2Pool.total_supply
UniswapV2Pool.total_supply_readable
UniswapV2Pool.tvl
- y.prices.dex.uniswap.v2_forks module
- y.prices.dex.uniswap.v3 module
UniV3Pools
UniV3Pools.obj_type
UniV3Pools.__aiter__()
UniV3Pools.__await__()
UniV3Pools.__del__()
UniV3Pools.__init__()
UniV3Pools.__iter__()
UniV3Pools.__subclasshook__()
UniV3Pools._objects_thru()
UniV3Pools.events()
UniV3Pools.filter()
UniV3Pools.logs()
UniV3Pools.objects()
UniV3Pools.sort()
UniV3Pools.wrap()
UniV3Pools.__async_property__
UniV3Pools.__orig_bases__
UniV3Pools.__parameters__
UniV3Pools.__wrapped__
UniV3Pools.addresses
UniV3Pools.asynchronous
UniV3Pools.bulk_insert
UniV3Pools.cache
UniV3Pools.executor
UniV3Pools.from_block
UniV3Pools.insert_to_db
UniV3Pools.is_asleep
UniV3Pools.is_reusable
UniV3Pools.materialized
UniV3Pools.semaphore
UniV3Pools.to_block
UniV3Pools.topics
UniswapV3
UniswapV3Pool
UniswapV3Pool._check_liquidity_token_out
UniswapV3Pool.__contains__()
UniswapV3Pool.__eq__()
UniswapV3Pool.__getitem__()
UniswapV3Pool.__init__()
UniswapV3Pool.__str__()
UniswapV3Pool.__build_name__
UniswapV3Pool.address
UniswapV3Pool.asynchronous
UniswapV3Pool.build_name
UniswapV3Pool.check_liquidity
UniswapV3Pool.contract
UniswapV3Pool.deploy_block
UniswapV3Pool.fee
UniswapV3Pool.has_method
UniswapV3Pool.tick_spacing
UniswapV3Pool.token0
UniswapV3Pool.token1
- Module contents
Submodules
y.prices.dex.genericamm module
- class y.prices.dex.genericamm.GenericAmm[source]
Bases:
ASyncGenericBase
A 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 (str | HexBytes | AnyAddress | EthAddress | Contract | int) – 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 price of the LP token in USD.
- Return type:
Example
>>> amm = GenericAmm(asynchronous=True) >>> address = "0x6B175474E89094C44Da98b954EedeAC495271d0F" >>> price = await amm.get_price(address) >>> print(price) 1.0
- 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...))
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 (str | HexBytes | AnyAddress | EthAddress | Contract | int)
- Return type:
- get_tvl[source]
Get the Total Value Locked (TVL) in the AMM pool.
- Parameters:
lp_token_address (str | HexBytes | AnyAddress | EthAddress | Contract | int) – 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
- 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 | 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
GenericAmm
with asynchronous mode enabled.
y.prices.dex.mooniswap module
- ASyncFunctiony.prices.dex.mooniswap.get_pool_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 the given Mooniswap pool token.
- Args:
token: The address of the pool token. block (optional): The block number to get the price at. Defaults to latest block. skip_cache (optional): Whether to skip the cache. Defaults to
ENVS.SKIP_CACHE
.- Returns:
The price of the pool token in USD.
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 | EthAddress | Contract | int)
block (int | BlockNumber | None)
skip_cache (bool)
- Return type:
- ASyncFunctiony.prices.dex.mooniswap.is_mooniswap_pool(token: str | hexbytes.main.HexBytes | AnyAddress | brownie.convert.datatypes.EthAddress | brownie.network.contract.Contract | int) bool [source]
Check if the given token address is a Mooniswap pool.
- Args:
token: The address of the token to check.
- Returns:
True if the token is a Mooniswap pool, False otherwise.
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 | EthAddress | Contract | int)
- Return type:
y.prices.dex.solidly module
- class y.prices.dex.solidly.SolidlyPool[source]
Bases:
UniswapV2Pool
- 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, token0=None, token1=None, deploy_block=None, asynchronous=False)
- Parameters:
address (str | HexBytes | AnyAddress | EthAddress | Contract | int)
token0 (str | HexBytes | AnyAddress | EthAddress | None)
token1 (str | HexBytes | AnyAddress | EthAddress | None)
deploy_block (int | None)
asynchronous (bool)
- __build_name__: HiddenMethodDescriptor[Self, str]
A
HiddenMethodDescriptor
for_ASyncPropertyDescriptorBase.get()
.
- __decimals__: HiddenMethodDescriptor[Self, int]
A
HiddenMethodDescriptor
for_ASyncPropertyDescriptorBase.get()
.
- __factory__: HiddenMethodDescriptor[Self, str | HexBytes | AnyAddress | EthAddress]
A
HiddenMethodDescriptor
for_ASyncPropertyDescriptorBase.get()
.
- __scale__: HiddenMethodDescriptor[Self, int]
A
HiddenMethodDescriptor
for_ASyncPropertyDescriptorBase.get()
.
- __symbol__: HiddenMethodDescriptor[Self, str]
A
HiddenMethodDescriptor
for_ASyncPropertyDescriptorBase.get()
.
- __token0__: HiddenMethodDescriptor[Self, ERC20]
A
HiddenMethodDescriptor
for_ASyncPropertyDescriptorBase.get()
.
- __token1__: HiddenMethodDescriptor[Self, ERC20]
A
HiddenMethodDescriptor
for_ASyncPropertyDescriptorBase.get()
.
- __tokens__: HiddenMethodDescriptor[Self, Tuple[ERC20, ERC20]]
A
HiddenMethodDescriptor
for_ASyncPropertyDescriptorBase.get()
.
- address
- balance_of
- Parameters:
address (str | HexBytes | AnyAddress | EthAddress | Contract | int)
block (int | BlockNumber | None)
- Return type:
- balance_of_readable
- Parameters:
address (str | HexBytes | AnyAddress | EthAddress | Contract | int)
block (int | BlockNumber | None)
- Return type:
- build_name
- 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.
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 | EthAddress)
block (int | BlockNumber)
- Return type:
- decimals
- factory
- get_price
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 | EthAddress)
- Return type:
- 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:
- name
- price
- Parameters:
block (int | BlockNumber | None)
return_None_on_failure (bool)
skip_cache (bool)
ignore_pools (Tuple[UniswapV2Pool | CurvePool, ...])
- Return type:
UsdPrice | None
- reserves
- Parameters:
block (int | BlockNumber | None)
- Return type:
Tuple[WeiBalance, WeiBalance] | None
- scale
- symbol
- token0
- token1
- tokens
- total_supply
- Parameters:
block (int | BlockNumber | None)
- Return type:
- total_supply_readable
- Parameters:
block (int | BlockNumber | None)
- Return type:
- tvl
- Parameters:
block (int | BlockNumber | None)
skip_cache (bool)
- Return type:
Decimal | None
- class y.prices.dex.solidly.SolidlyRouter[source]
Bases:
SolidlyRouterBase
- __eq__(_ContractBase__o)
Return self==value.
- __init__(router_address, *args, **kwargs)
- Parameters:
router_address (str | HexBytes | AnyAddress | EthAddress | Contract | int)
args (Any)
kwargs (Any)
- Return type:
None
- pools_for_token(token_address, block=None, _ignore_pools=())
- Parameters:
token_address (str | HexBytes | AnyAddress | EthAddress)
block (int | BlockNumber | None)
_ignore_pools (Tuple[UniswapV2Pool, ...])
- Return type:
- __build_name__: HiddenMethodDescriptor[Self, str]
A
HiddenMethodDescriptor
for_ASyncPropertyDescriptorBase.get()
.
- __pools__: HiddenMethodDescriptor[Self, List[UniswapV2Pool]]
A
HiddenMethodDescriptor
for_ASyncPropertyDescriptorBase.get()
.
- address
- build_name
- 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 | 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 | 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 | EthAddress)
block (int | BlockNumber)
- Return type:
Tuple[str | HexBytes | AnyAddress | 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 | EthAddress | Contract | int)
block (int | BlockNumber | None)
_ignore_pools (Tuple[UniswapV2Pool, ...])
- Return type:
UniswapV2Pool | None
- 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 | EthAddress | Contract | int)
block (int | BlockNumber | None)
_loop_count (int)
_ignore_pools (Tuple[UniswapV2Pool, ...])
- Return type:
List[str | HexBytes | AnyAddress | EthAddress | Contract]
- get_pool[source]
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 | EthAddress)
output_token (str | HexBytes | AnyAddress | EthAddress)
stable (bool)
block (int | BlockNumber)
- Return type:
SolidlyPool | None
- get_pools_for
Since get_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 | EthAddress)
block (int | BlockNumber | None)
- Return type:
Dict[UniswapV2Pool, str | HexBytes | AnyAddress | 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 | EthAddress)
block (int | BlockNumber | None)
token_out (str | HexBytes | AnyAddress | EthAddress)
paired_against (str | HexBytes | AnyAddress | EthAddress)
skip_cache (bool)
ignore_pools (Tuple[UniswapV2Pool | CurvePool, ...])
- Return type:
UsdPrice | None
- get_quote
- Parameters:
amount_in (int)
path (List[str | HexBytes | AnyAddress | EthAddress | Contract])
block (int | BlockNumber | None)
- Return type:
- get_routes_from_path[source]
- Parameters:
path (List[str | HexBytes | AnyAddress | EthAddress | Contract])
block (int | BlockNumber)
- Return type:
List[Tuple[str | HexBytes | AnyAddress | EthAddress, str | HexBytes | AnyAddress | EthAddress, bool]]
- pair_for[source]
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 | EthAddress)
output_token (str | HexBytes | AnyAddress | EthAddress)
stable (bool)
- Return type:
str | HexBytes | AnyAddress | EthAddress
- pools
- class y.prices.dex.solidly.SolidlyRouterBase[source]
Bases:
UniswapRouterV2
Solidly is a modified fork of Uni V2. The uniswap_multiplexer is the entrypoint for pricing using this object.
- __eq__(_ContractBase__o)
Return self==value.
- __init__(router_address, *args, **kwargs)
- Parameters:
router_address (str | HexBytes | AnyAddress | EthAddress | Contract | int)
args (Any)
kwargs (Any)
- Return type:
None
- pools_for_token(token_address, block=None, _ignore_pools=())
- Parameters:
token_address (str | HexBytes | AnyAddress | EthAddress)
block (int | BlockNumber | None)
_ignore_pools (Tuple[UniswapV2Pool, ...])
- Return type:
- __build_name__: HiddenMethodDescriptor[Self, str]
A
HiddenMethodDescriptor
for_ASyncPropertyDescriptorBase.get()
.
- __pools__: HiddenMethodDescriptor[Self, List[UniswapV2Pool]]
A
HiddenMethodDescriptor
for_ASyncPropertyDescriptorBase.get()
.
- address
- build_name
- 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 | 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 | 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 | EthAddress)
block (int | BlockNumber)
- Return type:
Tuple[str | HexBytes | AnyAddress | 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 | EthAddress | Contract | int)
block (int | BlockNumber | None)
_ignore_pools (Tuple[UniswapV2Pool, ...])
- Return type:
UniswapV2Pool | None
- 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 | EthAddress | Contract | int)
block (int | BlockNumber | None)
_loop_count (int)
_ignore_pools (Tuple[UniswapV2Pool, ...])
- Return type:
List[str | HexBytes | AnyAddress | EthAddress | Contract]
- get_pools_for
Since get_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 | EthAddress)
block (int | BlockNumber | None)
- Return type:
Dict[UniswapV2Pool, str | HexBytes | AnyAddress | 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 | EthAddress)
block (int | BlockNumber | None)
token_out (str | HexBytes | AnyAddress | EthAddress)
paired_against (str | HexBytes | AnyAddress | EthAddress)
skip_cache (bool)
ignore_pools (Tuple[UniswapV2Pool | CurvePool, ...])
- Return type:
UsdPrice | None
- get_quote[source]
- Parameters:
amount_in (int)
path (List[str | HexBytes | AnyAddress | EthAddress | Contract])
block (int | BlockNumber | None)
- Return type:
- pools
y.prices.dex.velodrome module
- class y.prices.dex.velodrome.VelodromePool[source]
Bases:
UniswapV2Pool
- 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, token0=None, token1=None, stable=None, deploy_block=None, asynchronous=False)[source]
- Parameters:
address (str | HexBytes | AnyAddress | EthAddress | Contract | int)
token0 (str | HexBytes | AnyAddress | EthAddress | Contract | int | None)
token1 (str | HexBytes | AnyAddress | EthAddress | Contract | int | None)
stable (bool | None)
deploy_block (int | None)
asynchronous (bool)
- __build_name__: HiddenMethodDescriptor[Self, str]
A
HiddenMethodDescriptor
for_ASyncPropertyDescriptorBase.get()
.
- __decimals__: HiddenMethodDescriptor[Self, int]
A
HiddenMethodDescriptor
for_ASyncPropertyDescriptorBase.get()
.
- __factory__: HiddenMethodDescriptor[Self, str | HexBytes | AnyAddress | EthAddress]
A
HiddenMethodDescriptor
for_ASyncPropertyDescriptorBase.get()
.
- __scale__: HiddenMethodDescriptor[Self, int]
A
HiddenMethodDescriptor
for_ASyncPropertyDescriptorBase.get()
.
- __symbol__: HiddenMethodDescriptor[Self, str]
A
HiddenMethodDescriptor
for_ASyncPropertyDescriptorBase.get()
.
- __token0__: HiddenMethodDescriptor[Self, ERC20]
A
HiddenMethodDescriptor
for_ASyncPropertyDescriptorBase.get()
.
- __token1__: HiddenMethodDescriptor[Self, ERC20]
A
HiddenMethodDescriptor
for_ASyncPropertyDescriptorBase.get()
.
- __tokens__: HiddenMethodDescriptor[Self, Tuple[ERC20, ERC20]]
A
HiddenMethodDescriptor
for_ASyncPropertyDescriptorBase.get()
.
- address
- balance_of
- Parameters:
address (str | HexBytes | AnyAddress | EthAddress | Contract | int)
block (int | BlockNumber | None)
- Return type:
- balance_of_readable
- Parameters:
address (str | HexBytes | AnyAddress | EthAddress | Contract | int)
block (int | BlockNumber | None)
- Return type:
- build_name
- 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.
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 | EthAddress)
block (int | BlockNumber)
- Return type:
- decimals
- factory
- get_price
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 | EthAddress)
- Return type:
- is_stable
- 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:
- name
- price
- Parameters:
block (int | BlockNumber | None)
return_None_on_failure (bool)
skip_cache (bool)
ignore_pools (Tuple[UniswapV2Pool | CurvePool, ...])
- Return type:
UsdPrice | None
- reserves
- Parameters:
block (int | BlockNumber | None)
- Return type:
Tuple[WeiBalance, WeiBalance] | None
- scale
- symbol
- token0
- token1
- tokens
- total_supply
- Parameters:
block (int | BlockNumber | None)
- Return type:
- total_supply_readable
- Parameters:
block (int | BlockNumber | None)
- Return type:
- tvl
- Parameters:
block (int | BlockNumber | None)
skip_cache (bool)
- Return type:
Decimal | None
- class y.prices.dex.velodrome.VelodromeRouterV2[source]
Bases:
SolidlyRouterBase
- __eq__(_ContractBase__o)
Return self==value.
- pools_for_token(token_address, block=None, _ignore_pools=())
- Parameters:
token_address (str | HexBytes | AnyAddress | EthAddress)
block (int | BlockNumber | None)
_ignore_pools (Tuple[UniswapV2Pool, ...])
- Return type:
- __build_name__: HiddenMethodDescriptor[Self, str]
A
HiddenMethodDescriptor
for_ASyncPropertyDescriptorBase.get()
.
- __pools__: HiddenMethodDescriptor[Self, Set[VelodromePool]]
A
HiddenMethodDescriptor
for_ASyncPropertyDescriptorBase.get()
.
- address
- build_name
- 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 | 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 | 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 | EthAddress)
block (int | BlockNumber)
- Return type:
Tuple[str | HexBytes | AnyAddress | 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 | EthAddress | Contract | int)
block (int | BlockNumber | None)
_ignore_pools (Tuple[UniswapV2Pool, ...])
- Return type:
UniswapV2Pool | None
- 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 | EthAddress | Contract | int)
block (int | BlockNumber | None)
_loop_count (int)
_ignore_pools (Tuple[UniswapV2Pool, ...])
- Return type:
List[str | HexBytes | AnyAddress | EthAddress | Contract]
- get_pool[source]
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 | EthAddress)
output_token (str | HexBytes | AnyAddress | EthAddress)
stable (bool)
block (int | BlockNumber)
- Return type:
UniswapV2Pool | None
- get_pools_for
Since get_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 | EthAddress)
block (int | BlockNumber | None)
- Return type:
Dict[UniswapV2Pool, str | HexBytes | AnyAddress | 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 | EthAddress)
block (int | BlockNumber | None)
token_out (str | HexBytes | AnyAddress | EthAddress)
paired_against (str | HexBytes | AnyAddress | EthAddress)
skip_cache (bool)
ignore_pools (Tuple[UniswapV2Pool | CurvePool, ...])
- Return type:
UsdPrice | None
- get_quote
- Parameters:
amount_in (int)
path (List[str | HexBytes | AnyAddress | EthAddress | Contract])
block (int | BlockNumber | None)
- Return type:
- get_routes_from_path[source]
- Parameters:
path (List[str | HexBytes | AnyAddress | EthAddress | Contract])
block (int | BlockNumber)
- Return type:
List[Tuple[str | HexBytes | AnyAddress | EthAddress, str | HexBytes | AnyAddress | EthAddress, bool]]
- pool_for[source]
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 | EthAddress)
output_token (str | HexBytes | AnyAddress | EthAddress)
stable (bool)
- Return type:
str | HexBytes | AnyAddress | EthAddress | None
- async y.prices.dex.velodrome.is_contract(pool_address)[source]
- Parameters:
pool_address (str | HexBytes | AnyAddress | EthAddress)
- Return type: