y.prices.dex.balancer package

Submodules

y.prices.dex.balancer.balancer module

class y.prices.dex.balancer.balancer.BalancerMultiplexer

Bases: ASyncGenericBase

A multiplexer for interacting with different versions of Balancer pools.

This class provides methods to determine if a token is a Balancer pool, retrieve pool prices, and get token prices across different Balancer versions.

Examples

Initialize the multiplexer:

>>> multiplexer = BalancerMultiplexer(asynchronous=True)

Check if a token is a Balancer pool:

>>> is_pool = await multiplexer.is_balancer_pool(token_address)

Get the price of a token:

>>> price = await multiplexer.get_price(token_address, block=12345678)

See also

• BalancerV1

• BalancerV2

__init__(*, asynchronous=False)

Initialize the BalancerMultiplexer.

Parameters:

asynchronous (bool) – Whether to operate in asynchronous mode.

Return type:

None

__v1__: HiddenMethodDescriptor[Self, BalancerV1 | None]

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

__v2__: HiddenMethodDescriptor[Self, BalancerV2 | None]

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

__versions__: HiddenMethodDescriptor[Self, List[BalancerV1 | BalancerV2]]

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

get_pool_price

Get the price of a Balancer pool.

Parameters:

• token_address (str | HexBytes | AnyAddress | EthAddress | Contract | int) – The address of the pool token.

• block (int | BlockNumber | None) – The block number to query the price at.

• skip_cache (bool) – Whether to skip the cache.

Returns:

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

Return type:

UsdPrice | None

Examples

>>> price = await multiplexer.get_pool_price(token_address, block=12345678)

get_price

Get the price of a token using Balancer pools.

Args:

token_address: The address of the token. block: The block number to query the price at. skip_cache: Whether to skip the cache.

Returns:

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

Examples:

>>> price = await multiplexer.get_price(token_address, block=12345678)

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_address (str | HexBytes | AnyAddress | EthAddress | Contract | int)

• block (int | BlockNumber | None)

• skip_cache (bool)

Return type:

UsdPrice | None

get_version

Determine the Balancer version for a given token address.

Args:

token_address: The address of the token.

Returns:

The Balancer version instance.

Raises:

exceptions.TokenError: If the token is not a Balancer pool.

Examples:

>>> version = await multiplexer.get_version(token_address)

Since get_version 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)

Return type:

BalancerABC

is_balancer_pool

Check if a given token address is a Balancer pool.

Parameters:

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

Returns:

True if the token is a Balancer pool, otherwise False.

Return type:

bool

Examples

>>> is_pool = await multiplexer.is_balancer_pool(token_address)

v1

Get the Balancer V1 instance.

Returns:

An instance of BalancerV1 if available, otherwise None.

Examples

>>> v1 = await multiplexer.v1

v2

Get the Balancer V2 instance.

Returns:

An instance of BalancerV2 if available, otherwise None.

Examples

>>> v2 = await multiplexer.v2

versions

Get the available Balancer versions.

Returns:

A list of available Balancer versions.

Examples

>>> versions = await multiplexer.versions

y.prices.dex.balancer.v1 module

class y.prices.dex.balancer.v1.BalancerV1

Bases: BalancerABC[BalancerV1Pool]

A Balancer V1 instance.

__init__(*, asynchronous=False)

Initialize a BalancerV1 instance.

Parameters:

asynchronous (bool) – Whether to use asynchronous operations.

Return type:

None

Examples

>>> balancer = BalancerV1(asynchronous=True)

check_liquidity

Check the liquidity of a token in the pool.

Parameters:

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

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

• ignore_pools (Tuple[UniswapV2Pool | CurvePool, ...]) – A tuple of pools to ignore.

Returns:

The liquidity of the token in the pool.

Return type:

int

Examples

>>> balancer = BalancerV1(asynchronous=True)
>>> await balancer.check_liquidity("0xabcdefabcdefabcdefabcdefabcdefabcdef", 12345678)
1000

check_liquidity_against

Check the liquidity of a token against another token in the pool.

Parameters:

• token_in (str | HexBytes | AnyAddress | EthAddress | Contract) – The input token address or contract.

• token_out (str | HexBytes | AnyAddress | EthAddress | Contract) – The output token address or contract.

• scale (int) – The scale factor.

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

Returns:

The total output amount, or None if it cannot be determined.

Return type:

int | None

Examples

>>> balancer = BalancerV1(asynchronous=True)
>>> await balancer.check_liquidity_against("0xabcdefabcdefabcdefabcdefabcdefabcdef", "0x1234567890abcdef1234567890abcdef12345678")
1000

get_pool_price

Get the price of a Balancer pool.

Parameters:

• pool_address (str | HexBytes | AnyAddress | EthAddress | Contract | int) – The address of the pool.

• block (int | BlockNumber | None) – The block number at which to get the price. If None, uses the latest block.

• skip_cache (bool) – If True, bypasses ypricemagic’s local caching mechanisms and forces a fresh calculation.

Returns:

The price of the pool as a UsdPrice object.

Return type:

UsdPrice

Examples

>>> price = await balancer.get_pool_price("0xPoolAddress")
>>> print(price)
100.0

get_some_output

Get some output for a given input token.

Parameters:

• token_in (str | HexBytes | AnyAddress | EthAddress | Contract) – The input token address or contract.

• scale (int) – The scale factor.

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

Returns:

A tuple containing the output token address and the total output amount, or None if it cannot be determined.

Return type:

Tuple[EthAddress, int] | None

Examples

>>> balancer = BalancerV1(asynchronous=True)
>>> await balancer.get_some_output("0xabcdefabcdefabcdefabcdefabcdefabcdef")
('0x1234567890abcdef1234567890abcdef12345678', 1000)

get_token_price

Get the price of a token in the pool.

Parameters:

• token_address (str | HexBytes | AnyAddress | EthAddress | Contract) – The token address or contract.

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

• skip_cache (bool) – Whether to skip the cache.

Returns:

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

Return type:

UsdPrice | None

Examples

>>> balancer = BalancerV1(asynchronous=True)
>>> await balancer.get_token_price("0xabcdefabcdefabcdefabcdefabcdefabcdef")
1.23

See also

• UsdPrice

is_pool

Check if the given token address is a Balancer pool.

Args:

token_address: The address of the token to check.

Returns:

True if the token is a Balancer pool, False otherwise.

Examples:

>>> is_pool = await balancer.is_pool("0xTokenAddress")
>>> print(is_pool)
True

Since is_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)

Return type:

bool

class y.prices.dex.balancer.v1.BalancerV1Pool

Bases: BalancerPool

A Balancer V1 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__(self)

Parameters:

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

• asynchronous (bool)

• _deploy_block (int | None)

Return type:

None

__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

__name__: HiddenMethodDescriptor[Self, str] = 'BalancerV1Pool'

__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

__tokens__: HiddenMethodDescriptor[Self, List[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:

• address (str | HexBytes | AnyAddress | EthAddress | Contract | int) – The address to query.

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

Returns:

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

Return type:

int

Examples

>>> token = ERC20("0x1234567890abcdef1234567890abcdef12345678")
>>> await token.balance_of("0xabcdefabcdefabcdefabcdefabcdefabcdef")
500000000000000000000

balance_of_readable

Parameters:

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

• block (int | BlockNumber | None)

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

Check the liquidity of a specific token in the pool.

Args:

token: The token address. block: The block number to query.

Returns:

The liquidity of the token in the pool.

Examples:

>>> pool = BalancerV1Pool("0x1234567890abcdef1234567890abcdef12345678")
>>> await pool.check_liquidity("0xabcdefabcdefabcdefabcdefabcdefabcdef")
1000

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:

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

get_balance

Get the balance of a specific token in the pool.

Parameters:

• token (str | HexBytes | AnyAddress | EthAddress | Contract | int) – The token address.

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

Returns:

The balance of the token in the pool.

Return type:

Decimal

Examples

>>> pool = BalancerV1Pool("0x1234567890abcdef1234567890abcdef12345678")
>>> await pool.get_balance("0xabcdefabcdefabcdefabcdefabcdefabcdef")
Decimal('1000')

get_balances

Get the balances of tokens in the pool.

Parameters:

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

Returns:

A dictionary mapping ERC20 tokens to their balances.

Return type:

Dict[ERC20, Decimal]

Examples

>>> pool = BalancerV1Pool("0x1234567890abcdef1234567890abcdef12345678")
>>> await pool.get_balances()
{<ERC20 TKN '0x1234567890abcdef1234567890abcdef12345678'>: Decimal('1000'), ...}

See also

• ERC20

get_pool_price

Calculate the price of the liquidity pool token.

This method calculates the price of a single liquidity pool token by dividing the total value locked (TVL) by the total supply of the pool tokens.

Parameters:

• block (int | BlockNumber | None) – The block number at which to calculate the price. If None, uses the latest block.

• skip_cache (bool) – If True, bypasses ypricemagic’s local caching mechanisms and forces a fresh calculation.

Returns:

The price of a single liquidity pool token as a UsdPrice object.

Return type:

UsdPrice | None

Examples

>>> class MyLiquidityPool(LiquidityPool):
...     async def get_tvl(self, block=None, skip_cache=False):
...         return UsdValue(1000000)
...
>>> pool = MyLiquidityPool("0xAddress")
>>> price = await pool.get_pool_price()
>>> print(price)
1000000.0

See also

• get_tvl()

get_tvl

Get the total value locked (TVL) in the pool.

Parameters:

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

• skip_cache (bool) – Whether to skip the cache.

Returns:

The total value locked in the pool, or None if it cannot be determined.

Return type:

UsdValue | None

Examples

>>> pool = BalancerV1Pool("0x1234567890abcdef1234567890abcdef12345678")
>>> await pool.get_tvl()
123456.78

See also

• UsdValue

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

• has_method()

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

• y.prices.magic.get_price()

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'

tokens

Get the list of tokens in the pool.

Returns:

A list of ERC20 tokens in the pool.

Examples

>>> pool = BalancerV1Pool("0x1234567890abcdef1234567890abcdef12345678")
>>> await pool.tokens
[<ERC20 TKN '0x1234567890abcdef1234567890abcdef12345678'>, ...]

See also

• ERC20

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

y.prices.dex.balancer.v2 module

class y.prices.dex.balancer.v2.BalancerEvents

Bases: ProcessedEvents[Tuple[HexBytes, EthAddress, int | BlockNumber]]

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

Example

>>> my_object = BalancerEvents(...)
>>> all_contents = await my_object
>>> isinstance(all_contents, list)
True
>>> isinstance(all_contents[0], Tuple)
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__(vault, *args, asynchronous=False, **kwargs)

Initialize a BalancerEvents instance.

Parameters:

• vault (BalancerV2Vault) – The associated BalancerV2Vault.

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

Examples

>>> events = BalancerEvents(vault)

__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__: AsyncIterable[T]

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.balancer.v2.BalancerV2

Bases: BalancerABC[BalancerV2Pool]

__init__(*, asynchronous=False)

Initialize a BalancerV2 instance.

Parameters:

asynchronous (bool) – Whether to use asynchronous operations.

Return type:

None

Examples

>>> balancer = BalancerV2(asynchronous=True)

deepest_pool_for

Find the deepest pool for a specific token.

Parameters:

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

• block (int | BlockNumber | None) – The block number to query. Defaults to the latest block.

Returns:

The BalancerV2Pool with the deepest liquidity for the specified token, or None if not found.

Return type:

BalancerV2Pool | None

Examples

>>> deepest_pool = await balancer.deepest_pool_for("0xTokenAddress")

get_pool_price

Get the price of a Balancer pool.

Parameters:

• pool_address (str | HexBytes | AnyAddress | EthAddress | Contract | int) – The address of the pool.

• block (int | BlockNumber | None) – The block number at which to get the price. If None, uses the latest block.

• skip_cache (bool) – If True, bypasses ypricemagic’s local caching mechanisms and forces a fresh calculation.

Returns:

The price of the pool as a UsdPrice object.

Return type:

UsdPrice

Examples

>>> price = await balancer.get_pool_price("0xPoolAddress")
>>> print(price)
100.0

get_token_price

Get the price of a specific token in USD.

Parameters:

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

• block (int | BlockNumber | None) – The block number to query. Defaults to the latest block.

• skip_cache (bool) – Whether to skip the cache.

Returns:

The price of the specified token in USD.

Return type:

UsdPrice

Examples

>>> price = await balancer.get_token_price("0xTokenAddress")

is_pool

Check if the given token address is a Balancer pool.

Args:

token_address: The address of the token to check.

Returns:

True if the token is a Balancer pool, False otherwise.

Examples:

>>> is_pool = await balancer.is_pool("0xTokenAddress")
>>> print(is_pool)
True

Since is_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)

Return type:

bool

class y.prices.dex.balancer.v2.BalancerV2Pool

Bases: BalancerPool

A pool from Balancer Protocol v2

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, *, id=None, specialization=None, vault=None, asynchronous=False, _deploy_block=None)

Initialize a BalancerV2Pool instance.

Parameters:

• address (str | HexBytes | AnyAddress | EthAddress | Contract | int) – The address of the pool.

• id (HexBytes | None) – The ID of the pool.

• specialization (PoolSpecialization | None) – The specialization of the pool.

• vault (BalancerV2Vault | None) – The associated BalancerV2Vault.

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

• _deploy_block (int | BlockNumber | None) – The block number when the pool was deployed.

Examples

>>> pool = BalancerV2Pool("0xPoolAddress")

__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

__id__: HiddenMethodDescriptor[Self, PoolId]

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] = 'BalancerV2Pool'

__pool_type__: HiddenMethodDescriptor[Self, PoolSpecialization | None]

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

__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

__vault__: HiddenMethodDescriptor[Self, BalancerV2Vault | None]

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:

• address (str | HexBytes | AnyAddress | EthAddress | Contract | int) – The address to query.

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

Returns:

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

Return type:

int

Examples

>>> token = ERC20("0x1234567890abcdef1234567890abcdef12345678")
>>> await token.balance_of("0xabcdefabcdefabcdefabcdefabcdefabcdef")
500000000000000000000

balance_of_readable

Parameters:

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

• block (int | BlockNumber | None)

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'

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

get_balance

Get the balance of a specific token in the pool.

Parameters:

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

• block (int | BlockNumber | None) – The block number to query. Defaults to the latest block.

• skip_cache (bool) – Whether to skip the cache.

Returns:

The WeiBalance of the specified token in the pool, or None if not found.

Return type:

WeiBalance | None

Examples

>>> balance = await pool.get_balance("0xTokenAddress")

get_balances

Get the balances of tokens in the pool.

Args:

block: The block number to query. Defaults to the latest block. skip_cache: Whether to skip the cache.

Returns:

A dictionary mapping ERC20 tokens to their WeiBalance in the pool.

Examples:

>>> balances = await pool.get_balances()

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

Dict[ERC20, WeiBalance]

get_pool_price

Calculate the price of the liquidity pool token.

This method calculates the price of a single liquidity pool token by dividing the total value locked (TVL) by the total supply of the pool tokens.

Parameters:

• block (int | BlockNumber | None) – The block number at which to calculate the price. If None, uses the latest block.

• skip_cache (bool) – If True, bypasses ypricemagic’s local caching mechanisms and forces a fresh calculation.

Returns:

The price of a single liquidity pool token as a UsdPrice object.

Return type:

UsdPrice | None

Examples

>>> class MyLiquidityPool(LiquidityPool):
...     async def get_tvl(self, block=None, skip_cache=False):
...         return UsdValue(1000000)
...
>>> pool = MyLiquidityPool("0xAddress")
>>> price = await pool.get_pool_price()
>>> print(price)
1000000.0

See also

• get_tvl()

get_token_price

Get the price of a specific token in the pool in USD.

Parameters:

• token_address (str | HexBytes | AnyAddress | EthAddress | Contract | int) – The address of the token.

• block (int | BlockNumber | None) – The block number to query. Defaults to the latest block.

• skip_cache (bool) – Whether to skip the cache.

Returns:

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

Return type:

UsdPrice | None

Examples

>>> price = await pool.get_token_price("0xTokenAddress")

get_tvl

Get the total value locked (TVL) in the pool in USD.

Parameters:

• block (int | BlockNumber | None) – The block number to query. Defaults to the latest block.

• skip_cache (bool) – Whether to skip the cache.

Returns:

The TVL in USD, or None if it cannot be determined.

Return type:

UsdValue | None

Examples

>>> tvl = await pool.get_tvl()

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

• has_method()

id

Get the ID of the pool.

Returns:

The pool ID.

Examples

>>> pool_id = await pool.id

name

The token’s name.

Returns:

The token’s name.

Examples

>>> token = ERC20("0x1234567890abcdef1234567890abcdef12345678")
>>> await token.name
'TokenName'

pool_type

Get the type of the pool.

Returns:

The pool type as a PoolSpecialization or an integer.

Examples

>>> pool_type = await pool.pool_type

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

• y.prices.magic.get_price()

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'

tokens

Get the tokens in the pool.

Args:

block: The block number to query. Defaults to the latest block. skip_cache: Whether to skip the cache.

Returns:

A tuple of ERC20 tokens in the pool.

Examples:

>>> tokens = await pool.tokens()

Since tokens is an ASyncFunctionAsyncDefault, you can optionally pass sync=True or asynchronous=False to force it to run synchronously and return a value. Without either kwarg, it will return a coroutine for you to await.

Parameters:

• block (int | BlockNumber | None)

• skip_cache (bool)

Return type:

Tuple[ERC20, …]

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

vault

Get the associated Balancer V2 Vault.

Returns:

The associated BalancerV2Vault, or None if not found.

Examples

>>> vault = await pool.vault

weights

Get the weights of tokens in the pool.

Args:

block: The block number to query. Defaults to the latest block.

Returns:

A list of weights for the tokens in the pool.

Examples:

>>> weights = await pool.weights()

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

List[int]

class y.prices.dex.balancer.v2.BalancerV2Vault

Bases: ContractBase

__eq__(_ContractBase__o)

Return self==value.

Parameters:

_ContractBase__o (object)

Return type:

bool

__init__(address, *, asynchronous=False)

Initialize a BalancerV2Vault instance.

Parameters:

• address (str | HexBytes | AnyAddress | EthAddress | Contract | int) – The address of the Balancer V2 Vault.

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

Return type:

None

Examples

>>> vault = BalancerV2Vault("0xBA12222222228d8Ba445958a75a0704d566BF2C8")

__str__()

Return the contract address as a string.

Returns:

The contract address as a string.

Return type:

str

pools(block=None)

Asynchronously iterate over Balancer V2 pools.

Parameters:

block (int | BlockNumber | None) – The block number to query. Defaults to the latest block.

Yields:

Instances of BalancerV2Pool.

Return type:

AsyncIterator[BalancerV2Pool]

Examples

>>> async for pool in vault.pools():
...     print(pool)

pools_for_token(token, block=None)

Asynchronously iterate over Balancer V2 pools containing a specific token.

Parameters:

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

• block (int | BlockNumber | None) – The block number to query. Defaults to the latest block.

Yields:

Instances of BalancerV2Pool containing the specified token.

Return type:

AsyncIterator[BalancerV2Pool]

Examples

>>> async for pool in vault.pools_for_token("0xTokenAddress"):
...     print(pool)

__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

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'

property contract: Contract

deepest_pool_for

Find the deepest pool for a specific token.

Args:

token_address: The address of the token. block: The block number to query. Defaults to the latest block.

Returns:

The BalancerV2Pool with the deepest liquidity for the specified token.

Examples:

>>> deepest_pool = await vault.deepest_pool_for("0xTokenAddress")

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_address (str | HexBytes | AnyAddress | EthAddress)

• block (int | BlockNumber | None)

Return type:

BalancerV2Pool

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_pool_info

Get information for multiple pools.

Args:

poolids: A tuple of pool IDs. block: The block number to query. Defaults to the latest block.

Returns:

A list of tuples containing pool information.

Examples:

>>> pool_info = await vault.get_pool_info((pool_id1, pool_id2))

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

• poolids (Tuple[HexBytes, ...])

• block (int | BlockNumber | None)

Return type:

List[Tuple]

get_pool_tokens

Get the tokens and balances for a specific pool.

Args:

pool_id: The ID of the pool. block: The block number to query. Defaults to the latest block.

Returns:

A tuple containing the tokens and their balances.

Examples:

>>> tokens, balances = await vault.get_pool_tokens(pool_id)

Since get_pool_tokens is an ASyncFunctionAsyncDefault, you can optionally pass sync=True or asynchronous=False to force it to run synchronously and return a value. Without either kwarg, it will return a coroutine for you to await.

Parameters:

• pool_id (HexBytes)

• block (int | BlockNumber | None)

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

• has_method()

class y.prices.dex.balancer.v2.PoolSpecialization

Bases: IntEnum

An enumeration.

__add__(value, /)

Return self+value.

__and__(value, /)

Return self&value.

__bool__()

True if self else False

__eq__(value, /)

Return self==value.

__floordiv__(value, /)

Return self//value.

__ge__(value, /)

Return self>=value.

__getattribute__(name, /)

Return getattr(self, name).

__gt__(value, /)

Return self>value.

__index__()

Return self converted to an integer, if self is suitable for use as an index into a list.

__invert__()

~self

__le__(value, /)

Return self<=value.

__lshift__(value, /)

Return self<<value.

__lt__(value, /)

Return self<value.

__mul__(value, /)

Return self*value.

__or__(value, /)

Return self|value.

__pow__(value, mod=None, /)

Return pow(self, value, mod).

__radd__(value, /)

Return value+self.

__rand__(value, /)

Return value&self.

__rlshift__(value, /)

Return value<<self.

__ror__(value, /)

Return value|self.

__rrshift__(value, /)

Return value>>self.

__rshift__(value, /)

Return self>>value.

__rxor__(value, /)

Return value^self.

__sizeof__()

Returns size in memory, in bytes.

__sub__(value, /)

Return self-value.

__truediv__(value, /)

Return self/value.

__xor__(value, /)

Return self^value.

from_bytes(byteorder, *, signed=False)

Return the integer represented by the given array of bytes.

bytes

Holds the array of bytes to convert. The argument must either support the buffer protocol or be an iterable object producing bytes. Bytes and bytearray are examples of built-in objects that support the buffer protocol.

byteorder

The byte order used to represent the integer. If byteorder is ‘big’, the most significant byte is at the beginning of the byte array. If byteorder is ‘little’, the most significant byte is at the end of the byte array. To request the native byte order of the host system, use `sys.byteorder’ as the byte order value.

signed

Indicates whether two’s complement is used to represent the integer.

to_bytes(length, byteorder, *, signed=False)

Return an array of bytes representing an integer.

length

Length of bytes object to use. An OverflowError is raised if the integer is not representable with the given number of bytes.

byteorder

The byte order used to represent the integer. If byteorder is ‘big’, the most significant byte is at the beginning of the byte array. If byteorder is ‘little’, the most significant byte is at the end of the byte array. To request the native byte order of the host system, use `sys.byteorder’ as the byte order value.

signed

Determines whether two’s complement is used to represent the integer. If signed is False and a negative integer is given, an OverflowError is raised.

static with_immutable_tokens()

Get a list of pool specializations with immutable tokens.

Returns:

A list of PoolSpecialization enums representing pools with immutable tokens.

Return type:

List[PoolSpecialization]

Examples

>>> PoolSpecialization.with_immutable_tokens()
[<PoolSpecialization.ComposableStablePool: 0>, <PoolSpecialization.WeightedPool: 1>, <PoolSpecialization.WeightedPool2Tokens: 2>, <PoolSpecialization.CronV1Pool: -1>]

ComposableStablePool = 0

CronV1Pool = -1

WeightedPool = 1

WeightedPool2Tokens = 2

denominator

the denominator of a rational number in lowest terms

numerator

the numerator of a rational number in lowest terms

Module contents