y.prices.dex.uniswap package

Submodules

y.prices.dex.uniswap.uniswap module

class y.prices.dex.uniswap.uniswap.UniswapMultiplexer[source]

Bases: ASyncGenericSingleton

__init__(asynchronous=False)[source]
Parameters:

asynchronous (bool)

Return type:

None

check_liquidity[source]
Parameters:
Return type:

int

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:
Return type:

UsdPrice | None

is_uniswap_pool[source]
Parameters:

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

Return type:

bool

routers_by_depth[source]

pool} ordered by liquidity depth, greatest to least

Type:

Returns a dict {router

Parameters:
Return type:

List[UniswapRouterV2]

y.prices.dex.uniswap.v1 module

class y.prices.dex.uniswap.v1.UniswapV1[source]

Bases: ASyncGenericBase

__init__(asynchronous=False)[source]
Parameters:

asynchronous (bool)

Return type:

None

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:
Return type:

int

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:
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 the ASyncIterable.

Return type:

AsyncIterator[T]

__await__()

Asynchronously iterate through the ASyncIterable and return all objects.

Returns:

A list of the objects yielded by the ASyncIterable.

Return type:

Generator[Any, Any, List[T]]

__del__()
Return type:

None

__init__(factory, label, asynchronous=False)[source]

Initializes the ASyncIterable with an async iterable.

Parameters:
__iter__()

Return an iterator that yields T objects from the ASyncIterable.

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).

_objects_thru(block)
Parameters:

block (int | None)

Return type:

AsyncIterator[T]

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:

ASyncIterator[_EventItem]

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 the ASyncIterable.

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:

ASyncIterator[_EventItem]

pools(to_block=None)[source]
Parameters:

to_block (int | None)

Return type:

AsyncIterator[UniswapV2Pool]

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 this ASyncIterable, but sorted.

Return type:

ASyncSorter[T]

classmethod wrap(wrapped)

Class method to wrap an AsyncIterable for backward compatibility.

Parameters:

wrapped (AsyncIterable[T])

Return type:

ASyncIterable[T]

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.

property is_asleep: bool
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.

Parameters:

_ContractBase__o (object)

Return type:

bool

__init__(router_address, *args, **kwargs)[source]
Parameters:
Return type:

None

__str__()[source]

Return str(self).

Return type:

str

pools_for_token(token_address, block=None, _ignore_pools=())[source]
Parameters:
Return type:

AsyncIterator[UniswapV2Pool]

__build_name__: HiddenMethodDescriptor[Self, str]

A HiddenMethodDescriptor for _ASyncPropertyDescriptorBase.get().

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

__pools__: HiddenMethodDescriptor[Self, List[UniswapV2Pool]]

A HiddenMethodDescriptor for _ASyncPropertyDescriptorBase.get().

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

address
asynchronous: bool = False
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:
Return type:

int

property contract: Contract
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:
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:
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:
Return type:

UniswapV2Pool | None

deploy_block: ASyncBoundMethod[Self, Any, int]
Parameters:

when_no_history_return_0 (bool)

Return type:

int

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:
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:
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:
Return type:

UsdPrice | None

get_quote[source]
Parameters:
Return type:

Tuple[int, int]

has_method
Parameters:

  • method (str)

  • return_response (bool)

Return type:

bool | Any

pools[source]
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:

int

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:

int

__eq__(_ContractBase__o)

Return self==value.

Parameters:

_ContractBase__o (object)

Return type:

bool

__init__(address, token0=None, token1=None, deploy_block=None, asynchronous=False)[source]
Parameters:
__str__()

Return str(self).

Return type:

str

__build_name__: HiddenMethodDescriptor[Self, str]

A HiddenMethodDescriptor for _ASyncPropertyDescriptorBase.get().

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

__decimals__: HiddenMethodDescriptor[Self, int]

A HiddenMethodDescriptor for _ASyncPropertyDescriptorBase.get().

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

__factory__: HiddenMethodDescriptor[Self, str | HexBytes | AnyAddress | EthAddress]

A HiddenMethodDescriptor for _ASyncPropertyDescriptorBase.get().

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

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

A HiddenMethodDescriptor for _ASyncPropertyDescriptorBase.get().

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

__symbol__: HiddenMethodDescriptor[Self, str]

A HiddenMethodDescriptor for _ASyncPropertyDescriptorBase.get().

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

__token0__: HiddenMethodDescriptor[Self, ERC20]

A HiddenMethodDescriptor for _ASyncPropertyDescriptorBase.get().

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

__token1__: HiddenMethodDescriptor[Self, ERC20]

A HiddenMethodDescriptor for _ASyncPropertyDescriptorBase.get().

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

__tokens__: HiddenMethodDescriptor[Self, Tuple[ERC20, ERC20]]

A HiddenMethodDescriptor for _ASyncPropertyDescriptorBase.get().

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

address
asynchronous: bool = False
balance_of
Parameters:
Return type:

int

balance_of_readable
Parameters:
Return type:

float

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:
Return type:

int

property contract: Contract
decimals
deploy_block: ASyncBoundMethod[Self, Any, int]
Parameters:

when_no_history_return_0 (bool)

Return type:

int

factory[source]
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:
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:

ERC20

has_method
Parameters:

  • method (str)

  • return_response (bool)

Return type:

bool | Any

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:

bool

name
price
Parameters:
Return type:

UsdPrice | None

reserves[source]
Parameters:

block (int | BlockNumber | None)

Return type:

Tuple[WeiBalance, WeiBalance] | None

scale
symbol
token0[source]
token1[source]
tokens[source]
total_supply
Parameters:

block (int | BlockNumber | None)

Return type:

int

total_supply_readable
Parameters:

block (int | BlockNumber | None)

Return type:

float

tvl[source]
Parameters:
Return type:

Decimal | None

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 the ASyncIterable.

Return type:

AsyncIterator[T]

__await__()

Asynchronously iterate through the ASyncIterable and return all objects.

Returns:

A list of the objects yielded by the ASyncIterable.

Return type:

Generator[Any, Any, List[T]]

__del__()
Return type:

None

__init__(factory, asynchronous=False)[source]

Initializes the ASyncIterable with an async iterable.

Parameters:
__iter__()

Return an iterator that yields T objects from the ASyncIterable.

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).

_objects_thru(block)
Parameters:

block (int | None)

Return type:

AsyncIterator[T]

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:

ASyncIterator[_EventItem]

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 the ASyncIterable.

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:

ASyncIterator[_EventItem]

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 this ASyncIterable, but sorted.

Return type:

ASyncSorter[T]

classmethod wrap(wrapped)

Class method to wrap an AsyncIterable for backward compatibility.

Parameters:

wrapped (AsyncIterable[T])

Return type:

ASyncIterable[T]

__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.

property is_asleep: bool
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

__contains__(asset)[source]
Return type:

bool

__init__(asynchronous=True)[source]
Parameters:

asynchronous (bool)

Return type:

None

pools_for_token(token, block)[source]
Parameters:
Return type:

AsyncIterator[UniswapV3Pool]

__factory__: HiddenMethodDescriptor[Self, Contract]

A HiddenMethodDescriptor for _ASyncPropertyDescriptorBase.get().

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

__pools__: HiddenMethodDescriptor[Self, UniV3Pools]

A HiddenMethodDescriptor for _ASyncPropertyDescriptorBase.get().

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

__quoter__: HiddenMethodDescriptor[Self, Contract]

A HiddenMethodDescriptor for _ASyncPropertyDescriptorBase.get().

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

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:
Return type:

int

factory[source]
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:
Return type:

UsdPrice | None

property loaded: Event
pools[source]
quoter[source]
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:
Return type:

int | None

__contains__(token)[source]
Parameters:

token (str | HexBytes | AnyAddress | EthAddress)

Return type:

bool

__eq__(_ContractBase__o)

Return self==value.

Parameters:

_ContractBase__o (object)

Return type:

bool

__getitem__(token)[source]
Parameters:

token (str | HexBytes | AnyAddress | EthAddress)

Return type:

ERC20

__init__(address, token0, token1, tick_spacing, fee, deploy_block, asynchronous=False)[source]
Parameters:
Return type:

None

__str__()

Return str(self).

Return type:

str

__build_name__: HiddenMethodDescriptor[Self, str]

A HiddenMethodDescriptor for _ASyncPropertyDescriptorBase.get().

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

address
asynchronous: bool = False
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:
Return type:

int | None

property contract: Contract
deploy_block: ASyncBoundMethod[Self, Any, int]
Parameters:

when_no_history_return_0 (bool)

Return type:

int

fee
has_method
Parameters:

  • method (str)

  • return_response (bool)

Return type:

bool | Any

tick_spacing
token0
token1

Module contents