y.prices.dex.balancer package

Submodules

y.prices.dex.balancer.balancer module

class y.prices.dex.balancer.balancer.BalancerMultiplexer[source]

Bases: ASyncGenericBase

__init__(asynchronous=False)[source]
Parameters:

asynchronous (bool)

Return type:

None

__v1__: HiddenMethodDescriptor[Self, BalancerV1 | None]

A HiddenMethodDescriptor for _ASyncPropertyDescriptorBase.get().

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

__v2__: HiddenMethodDescriptor[Self, BalancerV2 | None]

A HiddenMethodDescriptor for _ASyncPropertyDescriptorBase.get().

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

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

A HiddenMethodDescriptor for _ASyncPropertyDescriptorBase.get().

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

get_pool_price[source]
Parameters:
Return type:

UsdPrice | None

get_price[source]

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

Parameters:
Return type:

UsdPrice | None

get_version[source]

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

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

Return type:

bool

v1[source]
v2[source]
versions[source]

y.prices.dex.balancer.v1 module

class y.prices.dex.balancer.v1.BalancerV1[source]

Bases: BalancerABC[BalancerV1Pool]

__init__(asynchronous=False)[source]
Parameters:

asynchronous (bool)

Return type:

None

__orig_bases__ = (y.prices.dex.balancer._abc.BalancerABC[y.prices.dex.balancer.v1.BalancerV1Pool],)
__parameters__ = ()
check_liquidity[source]
Parameters:
Return type:

int

check_liquidity_against[source]
Parameters:
Return type:

int | None

get_pool_price
Parameters:
Return type:

UsdPrice

get_some_output[source]
Parameters:
Return type:

Tuple[EthAddress, int] | None

get_token_price[source]
Parameters:
Return type:

UsdPrice | None

is_pool

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

Bases: BalancerPool

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, asynchronous=False, _deploy_block=None)
Parameters:
Return type:

None

__str__()

Return str(self).

Return type:

str

__build_name__: HiddenMethodDescriptor[Self, str]

A HiddenMethodDescriptor for _ASyncPropertyDescriptorBase.get().

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

__decimals__: HiddenMethodDescriptor[Self, int]

A HiddenMethodDescriptor for _ASyncPropertyDescriptorBase.get().

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

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

A HiddenMethodDescriptor for _ASyncPropertyDescriptorBase.get().

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

__symbol__: HiddenMethodDescriptor[Self, str]

A HiddenMethodDescriptor for _ASyncPropertyDescriptorBase.get().

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

__tokens__: HiddenMethodDescriptor[Self, List[ERC20]]

A HiddenMethodDescriptor for _ASyncPropertyDescriptorBase.get().

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

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

int

balance_of_readable
Parameters:
Return type:

float

build_name
check_liquidity[source]

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

Parameters:
Return type:

int

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

when_no_history_return_0 (bool)

Return type:

int

get_balance[source]
Parameters:
Return type:

Decimal

get_balances[source]
Parameters:

block (int | BlockNumber | None)

Return type:

Dict[ERC20, Decimal]

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

get_tvl[source]
Parameters:
Return type:

UsdValue | None

has_method
Parameters:

  • method (str)

  • return_response (bool)

Return type:

bool | Any

name
price
Parameters:
Return type:

UsdPrice | None

scale
symbol
tokens[source]
total_supply
Parameters:

block (int | BlockNumber | None)

Return type:

int

total_supply_readable
Parameters:

block (int | BlockNumber | None)

Return type:

float

y.prices.dex.balancer.v2 module

class y.prices.dex.balancer.v2.BalancerEvents[source]

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

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

obj_type

alias of _EventItem

__aiter__()

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

Return type:

AsyncIterator[T]

__await__()

Asynchronously iterate through the ASyncIterable and return all objects.

Returns:

A list of the objects yielded by the ASyncIterable.

Return type:

Generator[Any, Any, List[T]]

__del__()
Return type:

None

__init__(vault, *args, asynchronous=False, **kwargs)[source]

Initializes the ASyncIterable with an async iterable.

Parameters:
__iter__()

Return an iterator that yields T objects from the ASyncIterable.

Return type:

Iterator[T]

classmethod __subclasshook__(C)

Abstract classes can override this to customize issubclass().

This is invoked early on by abc.ABCMeta.__subclasscheck__(). It should return True, False or NotImplemented. If it returns NotImplemented, the normal algorithm is used. Otherwise, it overrides the normal algorithm (and the outcome is cached).

_objects_thru(block)
Parameters:

block (int | None)

Return type:

AsyncIterator[T]

events(to_block)

Get events up to a given block.

Parameters:

to_block (int) – The ending block to fetch events to.

Yields:

A decoded event.

Return type:

ASyncIterator[_EventItem]

filter(function)

Filters the contents of the ASyncIterable based on a function.

Parameters:

function (Callable[[T], Awaitable[bool]] | Callable[[T], bool]) – A function that returns a boolean that indicates if an item should be included in the filtered result. Can be sync or async.

Returns:

An instance of ASyncFilter that yields the filtered objects from the ASyncIterable.

Return type:

ASyncFilter[T]

logs(to_block)
Parameters:

to_block (int | None)

Return type:

ASyncIterator[LogReceipt]

objects(to_block)

Get an ASyncIterator that yields all events up to a given block.

Parameters:

to_block (int) – The ending block to fetch events to.

Returns:

An ASyncIterator that yields all included events.

Return type:

ASyncIterator[_EventItem]

sort(*, key=None, reverse=False)

Sort the contents of the ASyncIterable.

Parameters:

  • key (optional) – A function of one argument that is used to extract a comparison key from each list element. If None, the elements themselves will be sorted. Defaults to None.

  • reverse (optional) – If True, the yielded elements will be sorted in reverse order. Defaults to False.

Returns:

An instance of ASyncSorter that will yield the objects yielded from this ASyncIterable, but sorted.

Return type:

ASyncSorter[T]

classmethod wrap(wrapped)

Class method to wrap an AsyncIterable for backward compatibility.

Parameters:

wrapped (AsyncIterable[T])

Return type:

ASyncIterable[T]

__async_property__
__orig_bases__ = (y.utils.events.ProcessedEvents[typing.Tuple[hexbytes.main.HexBytes, brownie.convert.datatypes.EthAddress, typing.Union[int, eth_typing.evm.BlockNumber]]],)
__parameters__ = ()
__wrapped__: AsyncIterable[T]

The wrapped async iterable object.

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

Get the function for bulk inserting logs into the database.

Returns:

A function for bulk inserting logs.

property cache: LogCache
property executor: _AsyncExecutorMixin
from_block
property insert_to_db: Callable[[LogReceipt], None]

Get the function for inserting logs into the database.

Raises:

NotImplementedError – If this method is not implemented in the subclass.

property is_asleep: bool
is_reusable
property materialized: List[T]

Synchronously iterate through the ASyncIterable and return all objects.

Returns:

A list of the objects yielded by the ASyncIterable.

property semaphore: BlockSemaphore
threads = <PruningThreadPoolExecutor object at 0x7ff5f41c5760 [0/6 threads]>
to_block
topics
class y.prices.dex.balancer.v2.BalancerV2[source]

Bases: BalancerABC[BalancerV2Pool]

__init__(asynchronous=False)[source]
Parameters:

asynchronous (bool)

Return type:

None

__orig_bases__ = (y.prices.dex.balancer._abc.BalancerABC[y.prices.dex.balancer.v2.BalancerV2Pool],)
__parameters__ = ()
deepest_pool_for[source]
Parameters:
Return type:

BalancerV2Pool | None

get_pool_price
Parameters:
Return type:

UsdPrice

get_token_price[source]
Parameters:
Return type:

UsdPrice

is_pool

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

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

Return str(self).

Return type:

str

__build_name__: HiddenMethodDescriptor[Self, str]

A HiddenMethodDescriptor for _ASyncPropertyDescriptorBase.get().

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

__decimals__: HiddenMethodDescriptor[Self, int]

A HiddenMethodDescriptor for _ASyncPropertyDescriptorBase.get().

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

__id__: HiddenMethodDescriptor[Self, PoolId]

A HiddenMethodDescriptor for _ASyncPropertyDescriptorBase.get().

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

__name__: HiddenMethodDescriptor[Self, str] = 'BalancerV2Pool'
__pool_type__: HiddenMethodDescriptor[Self, PoolSpecialization | None]

A HiddenMethodDescriptor for _ASyncPropertyDescriptorBase.get().

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

__scale__: HiddenMethodDescriptor[Self, int]

A HiddenMethodDescriptor for _ASyncPropertyDescriptorBase.get().

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

__symbol__: HiddenMethodDescriptor[Self, str]

A HiddenMethodDescriptor for _ASyncPropertyDescriptorBase.get().

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

__vault__: HiddenMethodDescriptor[Self, BalancerV2Vault | None]

A HiddenMethodDescriptor for _ASyncPropertyDescriptorBase.get().

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

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

int

balance_of_readable
Parameters:
Return type:

float

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

when_no_history_return_0 (bool)

Return type:

int

get_balance[source]
Parameters:
Return type:

WeiBalance | None

get_balances[source]

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

get_token_price[source]
Parameters:
Return type:

UsdPrice | None

get_tvl[source]
Parameters:
Return type:

UsdValue | None

has_method
Parameters:

  • method (str)

  • return_response (bool)

Return type:

bool | Any

id[source]
name
pool_type[source]
price
Parameters:
Return type:

UsdPrice | None

scale
symbol
tokens[source]

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

Tuple[ERC20, …]

total_supply
Parameters:

block (int | BlockNumber | None)

Return type:

int

total_supply_readable
Parameters:

block (int | BlockNumber | None)

Return type:

float

vault[source]
weights[source]

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

Bases: ContractBase

__eq__(_ContractBase__o)

Return self==value.

Parameters:

_ContractBase__o (object)

Return type:

bool

__init__(address, asynchronous=False)[source]
Parameters:
Return type:

None

__str__()

Return str(self).

Return type:

str

pools(block=None)[source]
Parameters:

block (int | BlockNumber | None)

Return type:

AsyncIterator[BalancerV2Pool]

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

AsyncIterator[BalancerV2Pool]

__build_name__: HiddenMethodDescriptor[Self, str]

A HiddenMethodDescriptor for _ASyncPropertyDescriptorBase.get().

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

address
asynchronous: bool = False
build_name
property contract: Contract
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:

BalancerV2Pool

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

when_no_history_return_0 (bool)

Return type:

int

get_pool_info[source]

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

List[Tuple]

get_pool_tokens[source]

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

  • method (str)

  • return_response (bool)

Return type:

bool | Any

class y.prices.dex.balancer.v2.PoolSpecialization[source]

Bases: IntEnum

An enumeration.

__abs__()

abs(self)

__add__(value, /)

Return self+value.

__and__(value, /)

Return self&value.

__bool__()

True if self else False

__ceil__()

Ceiling of an Integral returns itself.

__divmod__(value, /)

Return divmod(self, value).

__eq__(value, /)

Return self==value.

__float__()

float(self)

__floor__()

Flooring an Integral returns itself.

__floordiv__(value, /)

Return self//value.

__format__(format_spec)

Returns format using actual value type unless __str__ has been overridden.

__ge__(value, /)

Return self>=value.

__getattribute__(name, /)

Return getattr(self, name).

__getnewargs__()
__gt__(value, /)

Return self>value.

__index__()

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

__int__()

int(self)

__invert__()

~self

__le__(value, /)

Return self<=value.

__lshift__(value, /)

Return self<<value.

__lt__(value, /)

Return self<value.

__mod__(value, /)

Return self%value.

__mul__(value, /)

Return self*value.

__ne__(value, /)

Return self!=value.

__neg__()

-self

__or__(value, /)

Return self|value.

__pos__()

+self

__pow__(value, mod=None, /)

Return pow(self, value, mod).

__radd__(value, /)

Return value+self.

__rand__(value, /)

Return value&self.

__rdivmod__(value, /)

Return divmod(value, self).

__rfloordiv__(value, /)

Return value//self.

__rlshift__(value, /)

Return value<<self.

__rmod__(value, /)

Return value%self.

__rmul__(value, /)

Return value*self.

__ror__(value, /)

Return value|self.

__round__()

Rounding an Integral returns itself.

Rounding with an ndigits argument also returns an integer.

__rpow__(value, mod=None, /)

Return pow(value, self, mod).

__rrshift__(value, /)

Return value>>self.

__rshift__(value, /)

Return self>>value.

__rsub__(value, /)

Return value-self.

__rtruediv__(value, /)

Return value/self.

__rxor__(value, /)

Return value^self.

__sizeof__()

Returns size in memory, in bytes.

__sub__(value, /)

Return self-value.

__truediv__(value, /)

Return self/value.

__trunc__()

Truncating an Integral returns itself.

__xor__(value, /)

Return self^value.

as_integer_ratio()

Return integer ratio.

Return a pair of integers, whose ratio is exactly equal to the original int and with a positive denominator.

>>> (10).as_integer_ratio()
(10, 1)
>>> (-10).as_integer_ratio()
(-10, 1)
>>> (0).as_integer_ratio()
(0, 1)
bit_count()

Number of ones in the binary representation of the absolute value of self.

Also known as the population count.

>>> bin(13)
'0b1101'
>>> (13).bit_count()
3
bit_length()

Number of bits necessary to represent self in binary.

>>> bin(37)
'0b100101'
>>> (37).bit_length()
6
conjugate()

Returns self, the complex conjugate of any int.

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

List[PoolSpecialization]

ComposableStablePool = 0
CronV1Pool = -1
WeightedPool = 1
WeightedPool2Tokens = 2
denominator

the denominator of a rational number in lowest terms

imag

the imaginary part of a complex number

numerator

the numerator of a rational number in lowest terms

real

the real part of a complex number

Module contents