y.prices.dex.uniswap package

Submodules

y.prices.dex.uniswap.uniswap module

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

Bases: ASyncGenericSingleton

A multiplexer for Uniswap routers that provides aggregated functionality across multiple Uniswap instances, including Uniswap V1, V2, V3, and certain protocols with similar interfaces like Solidly and Velodrome.

This class allows for seamless interaction with various decentralized exchange protocols, enabling price fetching and liquidity checks across different router implementations.

Examples

>>> multiplexer = UniswapMultiplexer(asynchronous=True)
>>> price = await multiplexer.get_price("0xTokenAddress", block=12345678)
>>> print(price)

See also

__init__(self)[source]
Parameters:

asynchronous (bool)

Return type:

None

check_liquidity[source]

Check the maximum liquidity for a token across all Uniswap instances.

This method checks the liquidity of the given token across all Uniswap instances and returns the maximum liquidity found.

Parameters:

  • token (str | HexBytes | AnyAddress | EthAddress) – The address of the token to check liquidity for.

  • block (int | BlockNumber) – The block number to query.

  • ignore_pools (optional) – A tuple of Pool objects to ignore when checking liquidity.

Return type:

int

Note

  • The method uses asyncio.gather to check liquidity across all Uniswap instances concurrently.

  • A semaphore is used to limit the number of concurrent checks to 100.

Examples

>>> multiplexer = UniswapMultiplexer(asynchronous=True)
>>> liquidity = await multiplexer.check_liquidity("0xTokenAddress", block=12345678)
>>> print(liquidity)
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

Examples

>>> multiplexer = UniswapMultiplexer(asynchronous=True)
>>> price = await multiplexer.get_price("0xTokenAddress", block=12345678)
>>> print(price)

See also

is_uniswap_pool[source]
Parameters:

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

Return type:

bool

routers_by_depth[source]

Get Uniswap routers sorted by liquidity depth for a given token.

This method checks the liquidity of the input token across all Uniswap instances and returns a list of routers sorted by their liquidity depth in descending order.

Parameters:

  • token_in (str | HexBytes | AnyAddress | EthAddress | Contract | int) – The address of the input token to check liquidity for.

  • block (optional) – The block number to query. Defaults to latest block.

  • ignore_pools (optional) – A tuple of Pool objects to ignore when checking liquidity.

Returns:

A list of UniswapRouterV2 objects sorted by liquidity depth in descending order.

Return type:

List[UniswapRouterV2]

Note

  • The method uses asyncio.gather to check liquidity across all Uniswap instances concurrently.

  • Routers with zero liquidity are excluded from the result.

Examples

>>> multiplexer = UniswapMultiplexer(asynchronous=True)
>>> routers = await multiplexer.routers_by_depth("0xTokenAddress", block=12345678)
>>> print(routers)

y.prices.dex.uniswap.v1 module

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

Bases: ASyncGenericBase

__init__(*, asynchronous=False)[source]

Initialize the UniswapV1 class.

Parameters:

asynchronous (bool) – If True, the class will operate in asynchronous mode.

Raises:

UnsupportedNetwork – If the current network is not Ethereum Mainnet.

Return type:

None

Examples

>>> uniswap_v1 = UniswapV1()
>>> uniswap_v1.factory
'0xc0a47dFe034B400B47bDaD5FecDa2621de6c4d95'
check_liquidity[source]

Check the liquidity of a token in its exchange.

Args:

token_address: The address of the token to check liquidity for. block: The block number at which to check liquidity. ignore_pools: A tuple of pools to ignore when checking liquidity.

Returns:

The liquidity of the token in its exchange.

Examples:
>>> uniswap_v1 = UniswapV1()
>>> liquidity = await uniswap_v1.check_liquidity("0xTokenAddress", 12345678)
>>> print(liquidity)
1000000
See Also:

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]

Get the exchange contract for a given token address.

Args:

token_address: The address of the token for which to find the exchange.

Returns:

The exchange contract if it exists, otherwise None.

Examples:
>>> uniswap_v1 = UniswapV1()
>>> exchange = await uniswap_v1.get_exchange("0xTokenAddress")
>>> print(exchange)
<Contract '0xExchangeAddress'>
See Also:

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]

Get the price of a token in USD.

Parameters:
Returns:

The price of the token in USD, or None if the price cannot be determined.

Return type:

UsdPrice | None

Examples

>>> uniswap_v1 = UniswapV1()
>>> price = await uniswap_v1.get_price("0xTokenAddress", 12345678)
>>> print(price)
1.23

See also

y.prices.dex.uniswap.v2 module

class y.prices.dex.uniswap.v2.PoolsFromEvents[source]

Bases: ProcessedEvents[UniswapV2Pool]

When awaited, a list of all UniswapV2Pool will be returned.

Example

>>> my_object = PoolsFromEvents(...)
>>> all_contents = await my_object
>>> isinstance(all_contents, list)
True
>>> isinstance(all_contents[0], UniswapV2Pool)
True
obj_type

alias of _EventItem

__aiter__(self) → AsyncIterator[T]

Return an async iterator that yields T objects from the ASyncIterable.

Return type:

AsyncIterator[T]

__await__(self) → Generator[Any, Any, List[T]]

Asynchronously iterate through the ASyncIterable and return all T objects.

Returns:

A list of the T objects yielded by the ASyncIterable.

Return type:

Generator[Any, Any, List[T]]

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

Initialize a LogFilter instance.

Parameters:

  • addresses – List of contract addresses to fetch logs from.

  • topics – List of event topics to filter logs by.

  • from_block – The starting block to fetch logs from.

  • chunk_size – The number of blocks to fetch in each chunk.

  • chunks_per_batch – The number of chunks to fetch in each batch.

  • semaphore – A semaphore for limiting concurrent requests.

  • executor – An executor for running tasks asynchronously.

  • is_reusable – Whether the filter is reusable.

  • verbose – Verbosity level for logging.

  • factory (str | HexBytes | AnyAddress | EthAddress | Contract | int)

  • label (str)

  • asynchronous (bool)

Examples

>>> log_filter = LogFilter(addresses=["0x1234..."], topics=["0x5678..."])
>>> logs = log_filter.logs(1000100)
>>> print(logs)
__iter__(self) → Iterator[T]

Return an iterator that yields T objects from the ASyncIterable.

Note

Synchronous iteration leverages ASyncIterator, which uses asyncio.BaseEventLoop.run_until_complete() to fetch items. ASyncIterator.__next__() raises a SyncModeInAsyncContextError if the event loop is already running.

If you encounter a SyncModeInAsyncContextError, you are likely working in an async codebase and should consider asynchronous iteration using __aiter__() and __anext__() instead.

Return type:

Iterator[T]

_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]

Examples

>>> events = Events(addresses=["0x1234..."], topics=["0x5678..."])
>>> async for event in events.events(1000100):
...     print(event)
filter(self, function: ViewFn[T]) → 'ASyncFilter[T]'

Filters the T objects yielded by 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 T objects from the ASyncIterable.

Return type:

ASyncFilter[T]

logs(to_block)

Get logs up to a given block.

Parameters:

to_block (int | None) – The ending block to fetch logs to.

Yields:

A raw log.

Return type:

ASyncIterator[Log]

Examples

>>> log_filter = LogFilter(addresses=["0x1234..."], topics=["0x5678..."])
>>> logs = log_filter.logs(1000100)
>>> print(logs)
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]

Examples

>>> processed_events = ProcessedEvents(addresses=["0x1234..."], topics=["0x5678..."])
>>> async for event in processed_events.objects(1000100):
...     print(event)
pools(to_block=None)[source]
Parameters:

to_block (int | None)

Return type:

AsyncIterator[UniswapV2Pool]

sort(self, *, key: SortKey[T] = None, reverse: bool = False) → 'ASyncSorter[T]'

Sort the T objects yielded by 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 T objects yielded from this ASyncIterable, but sorted.

Return type:

ASyncSorter[T]

classmethod wrap(cls, wrapped: AsyncIterable[T]) → 'ASyncIterable[T]'

Class method to wrap an AsyncIterable for backward compatibility.

Parameters:

wrapped (AsyncIterable[T])

Return type:

ASyncIterable[T]

PairCreated = '0x0d3648bd0f6ba80134a33ba9275ac585d9d315f0ad8355cddefde31afa28d0e9'
__wrapped__
addresses
asynchronous
property bulk_insert: Callable[[List[Log]], Awaitable[None]]

Get the function for bulk inserting logs into the database.

Returns:

A function for bulk inserting logs.

Examples

>>> log_filter = LogFilter(addresses=["0x1234..."], topics=["0x5678..."])
>>> await log_filter.bulk_insert(logs)
property cache: LogCache
property executor: AsyncThreadPoolExecutor
from_block
property insert_to_db: Callable[[Log], 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 T objects.

Returns:

A list of the T 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__(self)[source]
Parameters:
Return type:

None

__str__()[source]

Return the contract address as a string.

Returns:

The contract address as a string.

Return type:

str

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

AsyncIterator[UniswapV2Pool]

__build_name__: HiddenMethodDescriptor[Self, str]

Optional[Type[I]] = None) -> T Asynchronously retrieves the property value.

Args:

instance: The instance from which the property is accessed. owner: The owner class of the property.

Returns:

The property value.

The original docstring for get() is shown below:

_ASyncPropertyDescriptorBase.get(self, instance: I, owner: Optional[Type[I]] = None) -> T Asynchronously retrieves the property value.

Args:

instance: The instance from which the property is accessed. owner: The owner class of the property.

Returns:

The property value.

Type:

_ASyncPropertyDescriptorBase.get(self, instance

Type:

I, owner

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

__pools__: HiddenMethodDescriptor[Self, List[UniswapV2Pool]]

Optional[Type[I]] = None) -> T Asynchronously retrieves the property value.

Args:

instance: The instance from which the property is accessed. owner: The owner class of the property.

Returns:

The property value.

The original docstring for get() is shown below:

_ASyncPropertyDescriptorBase.get(self, instance: I, owner: Optional[Type[I]] = None) -> T Asynchronously retrieves the property value.

Args:

instance: The instance from which the property is accessed. owner: The owner class of the property.

Returns:

The property value.

Type:

_ASyncPropertyDescriptorBase.get(self, instance

Type:

I, owner

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

address
all_pools_for[source]

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

Return type:

Dict[UniswapV2Pool, str | HexBytes | AnyAddress | EthAddress]

asynchronous: bool = False
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[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]

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:

int

Examples

>>> contract = ContractBase("0x1234567890abcdef1234567890abcdef12345678")
>>> await contract.deploy_block()
1234567

See also

  • contract_creation_block_async()

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]
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

Check if the contract has a specific method.

Parameters:

  • method (str) – The name of the method to check for.

  • return_response (bool) – If True, return the response of the method call instead of a boolean.

Returns:

A boolean indicating whether the contract has the method, or the response of the method call if return_response is True.

Return type:

bool | Any

Examples

>>> contract = ContractBase("0x1234567890abcdef1234567890abcdef12345678")
>>> await contract.has_method("name()")
True

See also

pools[source]
class y.prices.dex.uniswap.v2.UniswapV2Pool[source]

Bases: ERC20

Represents a liquidity pool from Uniswap V2 or its forks.

This class provides methods to interact with and retrieve information about a Uniswap V2 or fork liquidity pool, such as token reserves, total value locked (TVL), and price calculations.

Examples

>>> pool = UniswapV2Pool("0xAddress")
>>> reserves = await pool.reserves()
>>> price = await pool.get_price()
>>> tvl = await pool.tvl()

See also

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__(self)[source]
Parameters:
__str__()

Return the contract address as a string.

Returns:

The contract address as a string.

Return type:

str

__build_name__: HiddenMethodDescriptor[Self, str]

Optional[Type[I]] = None) -> T Asynchronously retrieves the property value.

Args:

instance: The instance from which the property is accessed. owner: The owner class of the property.

Returns:

The property value.

The original docstring for get() is shown below:

_ASyncPropertyDescriptorBase.get(self, instance: I, owner: Optional[Type[I]] = None) -> T Asynchronously retrieves the property value.

Args:

instance: The instance from which the property is accessed. owner: The owner class of the property.

Returns:

The property value.

Type:

_ASyncPropertyDescriptorBase.get(self, instance

Type:

I, owner

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

__decimals__: HiddenMethodDescriptor[Self, int]

Optional[Type[I]] = None) -> T Asynchronously retrieves the property value.

Args:

instance: The instance from which the property is accessed. owner: The owner class of the property.

Returns:

The property value.

The original docstring for get() is shown below:

_ASyncPropertyDescriptorBase.get(self, instance: I, owner: Optional[Type[I]] = None) -> T Asynchronously retrieves the property value.

Args:

instance: The instance from which the property is accessed. owner: The owner class of the property.

Returns:

The property value.

Type:

_ASyncPropertyDescriptorBase.get(self, instance

Type:

I, owner

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

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

Optional[Type[I]] = None) -> T Asynchronously retrieves the property value.

Args:

instance: The instance from which the property is accessed. owner: The owner class of the property.

Returns:

The property value.

The original docstring for get() is shown below:

_ASyncPropertyDescriptorBase.get(self, instance: I, owner: Optional[Type[I]] = None) -> T Asynchronously retrieves the property value.

Args:

instance: The instance from which the property is accessed. owner: The owner class of the property.

Returns:

The property value.

Type:

_ASyncPropertyDescriptorBase.get(self, instance

Type:

I, owner

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

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

Optional[Type[I]] = None) -> T Asynchronously retrieves the property value.

Args:

instance: The instance from which the property is accessed. owner: The owner class of the property.

Returns:

The property value.

The original docstring for get() is shown below:

_ASyncPropertyDescriptorBase.get(self, instance: I, owner: Optional[Type[I]] = None) -> T Asynchronously retrieves the property value.

Args:

instance: The instance from which the property is accessed. owner: The owner class of the property.

Returns:

The property value.

Type:

_ASyncPropertyDescriptorBase.get(self, instance

Type:

I, owner

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

__symbol__: HiddenMethodDescriptor[Self, str]

Optional[Type[I]] = None) -> T Asynchronously retrieves the property value.

Args:

instance: The instance from which the property is accessed. owner: The owner class of the property.

Returns:

The property value.

The original docstring for get() is shown below:

_ASyncPropertyDescriptorBase.get(self, instance: I, owner: Optional[Type[I]] = None) -> T Asynchronously retrieves the property value.

Args:

instance: The instance from which the property is accessed. owner: The owner class of the property.

Returns:

The property value.

Type:

_ASyncPropertyDescriptorBase.get(self, instance

Type:

I, owner

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

__token0__: HiddenMethodDescriptor[Self, ERC20]

Optional[Type[I]] = None) -> T Asynchronously retrieves the property value.

Args:

instance: The instance from which the property is accessed. owner: The owner class of the property.

Returns:

The property value.

The original docstring for get() is shown below:

_ASyncPropertyDescriptorBase.get(self, instance: I, owner: Optional[Type[I]] = None) -> T Asynchronously retrieves the property value.

Args:

instance: The instance from which the property is accessed. owner: The owner class of the property.

Returns:

The property value.

Type:

_ASyncPropertyDescriptorBase.get(self, instance

Type:

I, owner

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

__token1__: HiddenMethodDescriptor[Self, ERC20]

Optional[Type[I]] = None) -> T Asynchronously retrieves the property value.

Args:

instance: The instance from which the property is accessed. owner: The owner class of the property.

Returns:

The property value.

The original docstring for get() is shown below:

_ASyncPropertyDescriptorBase.get(self, instance: I, owner: Optional[Type[I]] = None) -> T Asynchronously retrieves the property value.

Args:

instance: The instance from which the property is accessed. owner: The owner class of the property.

Returns:

The property value.

Type:

_ASyncPropertyDescriptorBase.get(self, instance

Type:

I, owner

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

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

Optional[Type[I]] = None) -> T Asynchronously retrieves the property value.

Args:

instance: The instance from which the property is accessed. owner: The owner class of the property.

Returns:

The property value.

The original docstring for get() is shown below:

_ASyncPropertyDescriptorBase.get(self, instance: I, owner: Optional[Type[I]] = None) -> T Asynchronously retrieves the property value.

Args:

instance: The instance from which the property is accessed. owner: The owner class of the property.

Returns:

The property value.

Type:

_ASyncPropertyDescriptorBase.get(self, instance

Type:

I, owner

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

address: Address

The contract address of the token.

asynchronous: bool = False
balance_of

Query the balance of the token for a given address at a specific block.

Parameters:
Returns:

The balance of the token held by address at block block.

Return type:

int

Examples

>>> token = ERC20("0x1234567890abcdef1234567890abcdef12345678")
>>> await token.balance_of("0xabcdefabcdefabcdefabcdefabcdefabcdef")
500000000000000000000
balance_of_readable
Parameters:
Return type:

float

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

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

int

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

int

Examples

>>> contract = ContractBase("0x1234567890abcdef1234567890abcdef12345678")
>>> await contract.deploy_block()
1234567

See also

  • contract_creation_block_async()

factory[source]
get_price[source]

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

Check if the contract has a specific method.

Parameters:

  • method (str) – The name of the method to check for.

  • return_response (bool) – If True, return the response of the method call instead of a boolean.

Returns:

A boolean indicating whether the contract has the method, or the response of the method call if return_response is True.

Return type:

bool | Any

Examples

>>> contract = ContractBase("0x1234567890abcdef1234567890abcdef12345678")
>>> await contract.has_method("name()")
True

See also

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

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 yPriceMagicError on failure.

  • skip_cache (bool) – If True, skip using the cache while fetching price data.

  • ignore_pools (Tuple[UniswapV2Pool | CurvePool, ...]) – An optional tuple of pools to ignore when calculating the price.

Returns:

The price of the token in USD, or None if return_None_on_failure is True and the price cannot be retrieved.

Raises:

yPriceMagicError – If return_None_on_failure is False and the price cannot be retrieved.

Return type:

UsdPrice | None

Examples

>>> token = ERC20("0x1234567890abcdef1234567890abcdef12345678")
>>> await token.price()
1.23

See also

reserves[source]
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[source]
token1[source]
tokens[source]
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:

int

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:

float

Examples

>>> token = ERC20("0x1234567890abcdef1234567890abcdef12345678")
>>> await token.total_supply_readable()
1000.0
tvl[source]

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

y.prices.dex.uniswap.v2_forks module

Key components: - UNISWAPS: A dictionary mapping network names to dictionaries of Uniswap V2 fork protocols, including their factory and router addresses. - FACTORY_TO_ROUTER, ROUTER_TO_FACTORY: Mappings between factory and router addresses. - FACTORY_TO_PROTOCOL, ROUTER_TO_PROTOCOL: Mappings from addresses to protocol names. - SPECIAL_PATHS: A dictionary of special token paths for specific protocols and networks.

The module also includes a function for retrieving special token paths for a given protocol associated with a router address.

y.prices.dex.uniswap.v2_forks.special_paths(router_address)[source]

Retrieves special token paths for a given protocol associated with a router address.

Parameters:

router_address (str) – The address of the router.

Returns:

A dictionary of special paths (token in -> swap path) for the protocol associated with the given router.

Return type:

Dict[str, Dict[str | HexBytes | AnyAddress | EthAddress, List[str | HexBytes | AnyAddress | EthAddress]]]

Note

This function is cached using lru_cache to improve performance for repeated calls.

Examples

>>> special_paths("0x7a250d5630B4cF539739dF2C5dAcb4c659F2488D")
{
    "0xEF69B5697f2Fb0345cC680210fD39b593a2f9684": (
        "0xEF69B5697f2Fb0345cC680210fD39b593a2f9684",
        "0x6B3595068778DD592e39A122f4f5a5cF09C90fE2",
        "0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2",
        "0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48",
    ),
    ...
}

See also

  • SPECIAL_PATHS

y.prices.dex.uniswap.v3 module

class y.prices.dex.uniswap.v3.UniV3Pools[source]

Bases: ProcessedEvents[UniswapV3Pool]

Represents a collection of Uniswap V3 Pools.

When awaited, a list of all UniswapV3Pool will be returned.

Example

>>> my_object = UniV3Pools(...)
>>> all_contents = await my_object
>>> isinstance(all_contents, list)
True
>>> isinstance(all_contents[0], UniswapV3Pool)
True
obj_type

alias of _EventItem

__aiter__(self) → AsyncIterator[T]

Return an async iterator that yields T objects from the ASyncIterable.

Return type:

AsyncIterator[T]

__await__(self) → Generator[Any, Any, List[T]]

Asynchronously iterate through the ASyncIterable and return all T objects.

Returns:

A list of the T objects yielded by the ASyncIterable.

Return type:

Generator[Any, Any, List[T]]

__init__(factory, asynchronous=False)[source]

Initialize a UniV3Pools instance.

Parameters:

  • factory (Contract) – The factory contract for Uniswap V3.

  • asynchronous (bool) – Whether to use asynchronous operations.

Examples

>>> factory_contract = Contract(...)
>>> pools = UniV3Pools(factory_contract, asynchronous=True)
__iter__(self) → Iterator[T]

Return an iterator that yields T objects from the ASyncIterable.

Note

Synchronous iteration leverages ASyncIterator, which uses asyncio.BaseEventLoop.run_until_complete() to fetch items. ASyncIterator.__next__() raises a SyncModeInAsyncContextError if the event loop is already running.

If you encounter a SyncModeInAsyncContextError, you are likely working in an async codebase and should consider asynchronous iteration using __aiter__() and __anext__() instead.

Return type:

Iterator[T]

_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]

Examples

>>> events = Events(addresses=["0x1234..."], topics=["0x5678..."])
>>> async for event in events.events(1000100):
...     print(event)
filter(self, function: ViewFn[T]) → 'ASyncFilter[T]'

Filters the T objects yielded by 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 T objects from the ASyncIterable.

Return type:

ASyncFilter[T]

logs(to_block)

Get logs up to a given block.

Parameters:

to_block (int | None) – The ending block to fetch logs to.

Yields:

A raw log.

Return type:

ASyncIterator[Log]

Examples

>>> log_filter = LogFilter(addresses=["0x1234..."], topics=["0x5678..."])
>>> logs = log_filter.logs(1000100)
>>> print(logs)
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]

Examples

>>> processed_events = ProcessedEvents(addresses=["0x1234..."], topics=["0x5678..."])
>>> async for event in processed_events.objects(1000100):
...     print(event)
sort(self, *, key: SortKey[T] = None, reverse: bool = False) → 'ASyncSorter[T]'

Sort the T objects yielded by 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 T objects yielded from this ASyncIterable, but sorted.

Return type:

ASyncSorter[T]

classmethod wrap(cls, wrapped: AsyncIterable[T]) → 'ASyncIterable[T]'

Class method to wrap an AsyncIterable for backward compatibility.

Parameters:

wrapped (AsyncIterable[T])

Return type:

ASyncIterable[T]

__wrapped__
addresses
asynchronous
property bulk_insert: Callable[[List[Log]], Awaitable[None]]

Get the function for bulk inserting logs into the database.

Returns:

A function for bulk inserting logs.

Examples

>>> log_filter = LogFilter(addresses=["0x1234..."], topics=["0x5678..."])
>>> await log_filter.bulk_insert(logs)
property cache: LogCache
property executor: AsyncThreadPoolExecutor
from_block
property insert_to_db: Callable[[Log], 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 T objects.

Returns:

A list of the T objects yielded by the ASyncIterable.

property semaphore: BlockSemaphore
to_block
topics
class y.prices.dex.uniswap.v3.UniswapV3[source]

Bases: ASyncGenericSingleton

Represents the Uniswap V3 protocol.

__contains__(asset)[source]

Check if an asset is part of the Uniswap V3 protocol.

Parameters:

asset – The asset to check.

Returns:

True if the asset is part of the protocol, False otherwise.

Return type:

bool

Examples

>>> uniswap_v3 = UniswapV3(...)
>>> "0xAssetAddress" in uniswap_v3
True
__init__(asynchronous=True)[source]

Initialize a UniswapV3 instance.

Parameters:

asynchronous (bool) – Whether to use asynchronous operations.

Raises:

UnsupportedNetwork – If Uniswap V3 is not supported on the current network.

Return type:

None

Examples

>>> uniswap_v3 = UniswapV3(asynchronous=True)
pools_for_token(token, block)[source]

Get the pools for a specific token.

Parameters:
Yields:

Uniswap V3 pools for the token.

Return type:

AsyncIterator[UniswapV3Pool]

Examples

>>> uniswap_v3 = UniswapV3(...)
>>> async for pool in uniswap_v3.pools_for_token("0xTokenAddress", 1234567):
...     print(pool)
__factory__: HiddenMethodDescriptor[Self, Contract]

Optional[Type[I]] = None) -> T Asynchronously retrieves the property value.

Args:

instance: The instance from which the property is accessed. owner: The owner class of the property.

Returns:

The property value.

The original docstring for get() is shown below:

_ASyncPropertyDescriptorBase.get(self, instance: I, owner: Optional[Type[I]] = None) -> T Asynchronously retrieves the property value.

Args:

instance: The instance from which the property is accessed. owner: The owner class of the property.

Returns:

The property value.

Type:

_ASyncPropertyDescriptorBase.get(self, instance

Type:

I, owner

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

__pools__: HiddenMethodDescriptor[Self, UniV3Pools]

Optional[Type[I]] = None) -> T Asynchronously retrieves the property value.

Args:

instance: The instance from which the property is accessed. owner: The owner class of the property.

Returns:

The property value.

The original docstring for get() is shown below:

_ASyncPropertyDescriptorBase.get(self, instance: I, owner: Optional[Type[I]] = None) -> T Asynchronously retrieves the property value.

Args:

instance: The instance from which the property is accessed. owner: The owner class of the property.

Returns:

The property value.

Type:

_ASyncPropertyDescriptorBase.get(self, instance

Type:

I, owner

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

__quoter__: HiddenMethodDescriptor[Self, Contract]

Optional[Type[I]] = None) -> T Asynchronously retrieves the property value.

Args:

instance: The instance from which the property is accessed. owner: The owner class of the property.

Returns:

The property value.

The original docstring for get() is shown below:

_ASyncPropertyDescriptorBase.get(self, instance: I, owner: Optional[Type[I]] = None) -> T Asynchronously retrieves the property value.

Args:

instance: The instance from which the property is accessed. owner: The owner class of the property.

Returns:

The property value.

Type:

_ASyncPropertyDescriptorBase.get(self, instance

Type:

I, owner

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

check_liquidity[source]

Check the liquidity of a token in the Uniswap V3 protocol.

Args:

token: The address of the token. block: The block number to check liquidity at. ignore_pools: Pools to ignore.

Returns:

The liquidity of the token.

Examples:
>>> uniswap_v3 = UniswapV3(...)
>>> liquidity = await uniswap_v3.check_liquidity("0xTokenAddress", 1234567)

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 the factory contract for the Uniswap V3 protocol.

Returns:

The factory contract.

Examples

>>> uniswap_v3 = UniswapV3(...)
>>> factory_contract = await uniswap_v3.factory
get_price[source]

Get the price of a token in USD.

Args:

token: The address of the token. block: The block number to get the price at. ignore_pools: Pools to ignore (unused). skip_cache: Whether to skip cache (unused).

Returns:

The price of the token in USD, or None if not available.

Examples:
>>> uniswap_v3 = UniswapV3(...)
>>> price = await uniswap_v3.get_price("0xTokenAddress", 1234567)

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: CythonEvent

Get the loaded event for the Uniswap V3 instance.

Returns:

The loaded event.

Examples

>>> uniswap_v3 = UniswapV3(...)
>>> loaded_event = uniswap_v3.loaded
pools[source]

Get the list of Uniswap V3 pools.

Returns:

A list of Uniswap V3 pools.

Examples

>>> uniswap_v3 = UniswapV3(...)
>>> pools = await uniswap_v3.pools
quoter[source]

Get the quoter contract for the Uniswap V3 protocol.

Returns:

The quoter contract.

Examples

>>> uniswap_v3 = UniswapV3(...)
>>> quoter_contract = await uniswap_v3.quoter
class y.prices.dex.uniswap.v3.UniswapV3Pool[source]

Bases: ContractBase

Represents a Uniswap V3 Pool.

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]

Check the liquidity of the token out for a given token in.

Args:

token_in: The address of the token in. block: The block number to check liquidity at.

Returns:

The liquidity of the token out, or None if not available.

Examples:
>>> pool = UniswapV3Pool(...)
>>> liquidity = await pool._check_liquidity_token_out("0xToken0Address", 1234567)

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]

Check if a token is part of the pool.

Parameters:

token (str | HexBytes | AnyAddress | EthAddress) – The address of the token to check.

Returns:

True if the token is part of the pool, False otherwise.

Return type:

bool

Examples

>>> pool = UniswapV3Pool(...)
>>> "0xToken0Address" in pool
True
__eq__(_ContractBase__o)

Return self==value.

Parameters:

_ContractBase__o (object)

Return type:

bool

__getitem__(token)[source]

Get the ERC20 token object for a given token address.

Parameters:

token (str | HexBytes | AnyAddress | EthAddress) – The address of the token.

Returns:

The ERC20 token object.

Raises:

TokenNotFound – If the token is not part of the pool.

Return type:

ERC20

Examples

>>> pool = UniswapV3Pool(...)
>>> token = pool["0xToken0Address"]
__init__(address, token0, token1, tick_spacing, fee, deploy_block, asynchronous=False)[source]

Initialize a UniswapV3Pool instance.

Parameters:
Return type:

None

Examples

>>> pool = UniswapV3Pool(
...     address="0x1234567890abcdef1234567890abcdef12345678",
...     token0="0xToken0Address",
...     token1="0xToken1Address",
...     tick_spacing=60,
...     fee=3000,
...     deploy_block=1234567,
...     asynchronous=True
... )
__str__()

Return the contract address as a string.

Returns:

The contract address as a string.

Return type:

str

__build_name__: HiddenMethodDescriptor[Self, str]

Optional[Type[I]] = None) -> T Asynchronously retrieves the property value.

Args:

instance: The instance from which the property is accessed. owner: The owner class of the property.

Returns:

The property value.

The original docstring for get() is shown below:

_ASyncPropertyDescriptorBase.get(self, instance: I, owner: Optional[Type[I]] = None) -> T Asynchronously retrieves the property value.

Args:

instance: The instance from which the property is accessed. owner: The owner class of the property.

Returns:

The property value.

Type:

_ASyncPropertyDescriptorBase.get(self, instance

Type:

I, owner

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

__contains_cache__: Dict[str | HexBytes | AnyAddress | EthAddress, Dict[ChecksumAddress, bool]] = {}
address
asynchronous: bool = False
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[source]

Check the liquidity of a token in the pool at a specific block.

Args:

token: The address of the token. block: The block number to check liquidity at.

Returns:

The liquidity of the token in the pool, or None if not available.

Examples:
>>> pool = UniswapV3Pool(...)
>>> liquidity = await pool.check_liquidity("0xToken0Address", 1234567)

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]

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:

int

Examples

>>> contract = ContractBase("0x1234567890abcdef1234567890abcdef12345678")
>>> await contract.deploy_block()
1234567

See also

  • contract_creation_block_async()

fee
has_method

Check if the contract has a specific method.

Parameters:

  • method (str) – The name of the method to check for.

  • return_response (bool) – If True, return the response of the method call instead of a boolean.

Returns:

A boolean indicating whether the contract has the method, or the response of the method call if return_response is True.

Return type:

bool | Any

Examples

>>> contract = ContractBase("0x1234567890abcdef1234567890abcdef12345678")
>>> await contract.has_method("name()")
True

See also

tick_spacing
token0
token1

Module contents