y.prices.dex.uniswap package
Submodules
y.prices.dex.uniswap.uniswap module
- class y.prices.dex.uniswap.uniswap.UniswapMultiplexer[source]
Bases:
ASyncGenericSingleton
- check_liquidity[source]
- Parameters:
token (str | HexBytes | AnyAddress | EthAddress)
block (int | BlockNumber)
ignore_pools (Tuple[UniswapV2Pool | CurvePool, ...])
- Return type:
- get_price[source]
Calculate a price based on Uniswap Router quote for selling one token_in. Always finds the deepest swap path for token_in.
- Parameters:
token_in (str | HexBytes | AnyAddress | EthAddress | Contract | int)
block (int | BlockNumber | None)
ignore_pools (Tuple[UniswapV2Pool | CurvePool, ...])
skip_cache (bool)
- Return type:
UsdPrice | None
- is_uniswap_pool[source]
- Parameters:
token_address (str | HexBytes | AnyAddress | EthAddress | Contract | int)
- Return type:
- routers_by_depth[source]
pool} ordered by liquidity depth, greatest to least
- Type:
Returns a dict {router
- Parameters:
token_in (str | HexBytes | AnyAddress | EthAddress | Contract | int)
block (int | BlockNumber | None)
ignore_pools (Tuple[UniswapV2Pool | CurvePool, ...])
- Return type:
y.prices.dex.uniswap.v1 module
- class y.prices.dex.uniswap.v1.UniswapV1[source]
Bases:
ASyncGenericBase
- check_liquidity[source]
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_address (str | HexBytes | AnyAddress | EthAddress)
block (int | BlockNumber)
ignore_pools (Tuple[UniswapV2Pool | CurvePool, ...])
- Return type:
- factory = '0xc0a47dFe034B400B47bDaD5FecDa2621de6c4d95'
- get_exchange[source]
Since get_exchange 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)
- Return type:
Contract | None
- get_price[source]
- Parameters:
token_address (str | HexBytes | AnyAddress | EthAddress)
block (int | BlockNumber | None)
ignore_pools (Tuple[UniswapV2Pool | CurvePool, ...])
skip_cache (bool)
- Return type:
UsdPrice | None
y.prices.dex.uniswap.v2 module
- class y.prices.dex.uniswap.v2.PoolsFromEvents[source]
Bases:
ProcessedEvents
[UniswapV2Pool
]When awaited, a list of all elements will be returned.
- obj_type
alias of
_EventItem
- __aiter__()
Return an async iterator that yields
T
objects from theASyncIterable
.- Return type:
- __await__()
Asynchronously iterate through the
ASyncIterable
and return all objects.
- __del__()
- Return type:
None
- __init__(factory, label, asynchronous=False)[source]
Initializes the ASyncIterable with an async iterable.
- Parameters:
factory (str | HexBytes | AnyAddress | EthAddress | Contract | int)
label (str)
asynchronous (bool)
- __iter__()
Return an iterator that yields
T
objects from theASyncIterable
.- Return type:
Iterator[T]
- classmethod __subclasshook__(C)
Abstract classes can override this to customize issubclass().
This is invoked early on by abc.ABCMeta.__subclasscheck__(). It should return True, False or NotImplemented. If it returns NotImplemented, the normal algorithm is used. Otherwise, it overrides the normal algorithm (and the outcome is cached).
- events(to_block)
Get events up to a given block.
- Parameters:
to_block (int) – The ending block to fetch events to.
- Yields:
A decoded event.
- Return type:
- filter(function)
Filters the contents of the
ASyncIterable
based on a function.- Parameters:
function (Callable[[T], Awaitable[bool]] | Callable[[T], bool]) – A function that returns a boolean that indicates if an item should be included in the filtered result. Can be sync or async.
- Returns:
An instance of
ASyncFilter
that yields the filtered objects from theASyncIterable
.- Return type:
ASyncFilter[T]
- logs(to_block)
- Parameters:
to_block (int | None)
- Return type:
ASyncIterator[LogReceipt]
- objects(to_block)
Get an
ASyncIterator
that yields all events up to a given block.- Parameters:
to_block (int) – The ending block to fetch events to.
- Returns:
An
ASyncIterator
that yields all included events.- Return type:
- sort(*, key=None, reverse=False)
Sort the contents of the
ASyncIterable
.- Parameters:
key (optional) – A function of one argument that is used to extract a comparison key from each list element. If None, the elements themselves will be sorted. Defaults to None.
reverse (optional) – If True, the yielded elements will be sorted in reverse order. Defaults to False.
- Returns:
An instance of
ASyncSorter
that will yield the objects yielded from thisASyncIterable
, but sorted.- Return type:
ASyncSorter[T]
- classmethod wrap(wrapped)
Class method to wrap an AsyncIterable for backward compatibility.
- Parameters:
wrapped (AsyncIterable[T])
- Return type:
- PairCreated = '0x0d3648bd0f6ba80134a33ba9275ac585d9d315f0ad8355cddefde31afa28d0e9'
- __async_property__
- __orig_bases__ = (y.utils.events.ProcessedEvents[y.prices.dex.uniswap.v2.UniswapV2Pool],)
- __parameters__ = ()
- __wrapped__: AsyncIterable[T]
The wrapped async iterable object.
- addresses
- asynchronous
- property bulk_insert: Callable[[List[LogReceipt]], Awaitable[None]]
Get the function for bulk inserting logs into the database.
- Returns:
A function for bulk inserting logs.
- property cache: LogCache
- property executor: _AsyncExecutorMixin
- from_block
- property insert_to_db: Callable[[LogReceipt], None]
Get the function for inserting logs into the database.
- Raises:
NotImplementedError – If this method is not implemented in the subclass.
- is_reusable
- label
- property materialized: List[T]
Synchronously iterate through the
ASyncIterable
and return all objects.- Returns:
A list of the objects yielded by the
ASyncIterable
.
- property semaphore: BlockSemaphore
- to_block
- topics
- class y.prices.dex.uniswap.v2.UniswapRouterV2[source]
Bases:
ContractBase
- __eq__(_ContractBase__o)
Return self==value.
- __init__(router_address, *args, **kwargs)[source]
- 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=())[source]
- 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[source]
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[source]
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[source]
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[source]
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[source]
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[source]
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[source]
- 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:
- class y.prices.dex.uniswap.v2.UniswapV2Pool[source]
Bases:
ERC20
Represents a Uniswap V2 liquidity pool.
- 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)[source]
- 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[source]
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
- get_price[source]
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[source]
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[source]
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[source]
- Parameters:
block (int | BlockNumber | None)
- Return type:
Tuple[WeiBalance, WeiBalance] | None
- scale
- symbol
- total_supply
- Parameters:
block (int | BlockNumber | None)
- Return type:
- total_supply_readable
- Parameters:
block (int | BlockNumber | None)
- Return type:
y.prices.dex.uniswap.v2_forks module
- y.prices.dex.uniswap.v2_forks.special_paths(router_address)[source]
- Parameters:
router_address (str)
- Return type:
Dict[str, Dict[str | HexBytes | AnyAddress | EthAddress, List[str | HexBytes | AnyAddress | EthAddress]]]
y.prices.dex.uniswap.v3 module
- class y.prices.dex.uniswap.v3.UniV3Pools[source]
Bases:
ProcessedEvents
[UniswapV3Pool
]When awaited, a list of all elements will be returned.
- obj_type
alias of
_EventItem
- __aiter__()
Return an async iterator that yields
T
objects from theASyncIterable
.- Return type:
- __await__()
Asynchronously iterate through the
ASyncIterable
and return all objects.
- __del__()
- Return type:
None
- __init__(factory, asynchronous=False)[source]
Initializes the ASyncIterable with an async iterable.
- __iter__()
Return an iterator that yields
T
objects from theASyncIterable
.- Return type:
Iterator[T]
- classmethod __subclasshook__(C)
Abstract classes can override this to customize issubclass().
This is invoked early on by abc.ABCMeta.__subclasscheck__(). It should return True, False or NotImplemented. If it returns NotImplemented, the normal algorithm is used. Otherwise, it overrides the normal algorithm (and the outcome is cached).
- events(to_block)
Get events up to a given block.
- Parameters:
to_block (int) – The ending block to fetch events to.
- Yields:
A decoded event.
- Return type:
- filter(function)
Filters the contents of the
ASyncIterable
based on a function.- Parameters:
function (Callable[[T], Awaitable[bool]] | Callable[[T], bool]) – A function that returns a boolean that indicates if an item should be included in the filtered result. Can be sync or async.
- Returns:
An instance of
ASyncFilter
that yields the filtered objects from theASyncIterable
.- Return type:
ASyncFilter[T]
- logs(to_block)
- Parameters:
to_block (int | None)
- Return type:
ASyncIterator[LogReceipt]
- objects(to_block)
Get an
ASyncIterator
that yields all events up to a given block.- Parameters:
to_block (int) – The ending block to fetch events to.
- Returns:
An
ASyncIterator
that yields all included events.- Return type:
- sort(*, key=None, reverse=False)
Sort the contents of the
ASyncIterable
.- Parameters:
key (optional) – A function of one argument that is used to extract a comparison key from each list element. If None, the elements themselves will be sorted. Defaults to None.
reverse (optional) – If True, the yielded elements will be sorted in reverse order. Defaults to False.
- Returns:
An instance of
ASyncSorter
that will yield the objects yielded from thisASyncIterable
, but sorted.- Return type:
ASyncSorter[T]
- classmethod wrap(wrapped)
Class method to wrap an AsyncIterable for backward compatibility.
- Parameters:
wrapped (AsyncIterable[T])
- Return type:
- __async_property__
- __orig_bases__ = (y.utils.events.ProcessedEvents[y.prices.dex.uniswap.v3.UniswapV3Pool],)
- __parameters__ = ()
- __wrapped__: AsyncIterable[T]
The wrapped async iterable object.
- addresses
- asynchronous
- property bulk_insert: Callable[[List[LogReceipt]], Awaitable[None]]
Get the function for bulk inserting logs into the database.
- Returns:
A function for bulk inserting logs.
- property cache: LogCache
- property executor: _AsyncExecutorMixin
- from_block
- property insert_to_db: Callable[[LogReceipt], None]
Get the function for inserting logs into the database.
- Raises:
NotImplementedError – If this method is not implemented in the subclass.
- is_reusable
- property materialized: List[T]
Synchronously iterate through the
ASyncIterable
and return all objects.- Returns:
A list of the objects yielded by the
ASyncIterable
.
- property semaphore: BlockSemaphore
- to_block
- topics
- class y.prices.dex.uniswap.v3.UniswapV3[source]
Bases:
ASyncGenericSingleton
- pools_for_token(token, block)[source]
- Parameters:
token (str | HexBytes | AnyAddress | EthAddress)
block (int | BlockNumber)
- Return type:
- __factory__: HiddenMethodDescriptor[Self, Contract]
A
HiddenMethodDescriptor
for_ASyncPropertyDescriptorBase.get()
.
- __pools__: HiddenMethodDescriptor[Self, UniV3Pools]
A
HiddenMethodDescriptor
for_ASyncPropertyDescriptorBase.get()
.
- __quoter__: HiddenMethodDescriptor[Self, Contract]
A
HiddenMethodDescriptor
for_ASyncPropertyDescriptorBase.get()
.
- check_liquidity[source]
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)
ignore_pools (Tuple[UniswapV2Pool | CurvePool, ...])
- Return type:
- get_price[source]
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 (str | HexBytes | AnyAddress | EthAddress)
block (int | BlockNumber | None)
ignore_pools (Tuple[UniswapV2Pool | CurvePool, ...])
skip_cache (bool)
- Return type:
UsdPrice | None
- class y.prices.dex.uniswap.v3.UniswapV3Pool[source]
Bases:
ContractBase
- ASyncFunction_check_liquidity_token_out(token_in: str | hexbytes.main.HexBytes | AnyAddress | brownie.convert.datatypes.EthAddress | brownie.network.contract.Contract | int, block: int | eth_typing.evm.BlockNumber) int | None [source]
Since _check_liquidity_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 | Contract | int)
block (int | BlockNumber)
- Return type:
int | None
- __contains__(token)[source]
- Parameters:
token (str | HexBytes | AnyAddress | EthAddress)
- Return type:
- __eq__(_ContractBase__o)
Return self==value.
- __getitem__(token)[source]
- Parameters:
token (str | HexBytes | AnyAddress | EthAddress)
- Return type:
- __init__(address, token0, token1, tick_spacing, fee, deploy_block, asynchronous=False)[source]
- Parameters:
address (str | HexBytes | AnyAddress | EthAddress)
token0 (str | HexBytes | AnyAddress | EthAddress)
token1 (str | HexBytes | AnyAddress | EthAddress)
tick_spacing (int)
fee (int)
deploy_block (int)
asynchronous (bool)
- Return type:
None
- __build_name__: HiddenMethodDescriptor[Self, str]
A
HiddenMethodDescriptor
for_ASyncPropertyDescriptorBase.get()
.
- address
- build_name
- check_liquidity[source]
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 | Contract | int)
block (int | BlockNumber)
- Return type:
int | None
- deploy_block: ASyncBoundMethod[Self, Any, int]
- fee
- tick_spacing
- token0
- token1