y.prices.stable_swap package

Submodules

y.prices.stable_swap.belt module

ASyncFunctiony.prices.stable_swap.belt.get_price(token: str | hexbytes.main.HexBytes | AnyAddress | brownie.convert.datatypes.EthAddress | brownie.network.contract.Contract | int, block: int | eth_typing.evm.BlockNumber | NoneType = None) → y.datatypes.UsdPrice[source]

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

Parameters:
Return type:

UsdPrice

y.prices.stable_swap.belt.is_belt_lp(token)[source]
Parameters:

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

Return type:

bool

y.prices.stable_swap.curve module

class y.prices.stable_swap.curve.AddressProvider[source]

Bases: _CurveEventsLoader

__await__()

Returns True once the _Loader has loaded all relevant data thru the current block.

Returns:

A generator that yields True once the _Loader has loaded all relevant data thru the current block.

Return type:

Generator[Any, None, Literal[True]]

__eq__(_ContractBase__o)

Return self==value.

Parameters:

_ContractBase__o (object)

Return type:

bool

__init__(address, asynchronous=False)[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]

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:

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.

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.

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

when_no_history_return_0 (bool)

Return type:

int

get_registry[source]
Return type:

EthAddress

has_method
Parameters:

  • method (str)

  • return_response (bool)

Return type:

bool | Any

identifiers
property loaded: Awaitable[Literal[True]]

Returns True once the _Loader has loaded all relevant data thru the current block.

Returns:

An awaitable that resolves to True once the _Loader has loaded all relevant data thru the current block.

class y.prices.stable_swap.curve.AddressProviderEvents[source]

Bases: CurveEvents

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

__init__(base)

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.

  • base (_LT)

__iter__()

Return an iterator that yields T objects from the ASyncIterable.

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]

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)

Get logs up to a given block.

Parameters:

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

Yields:

A decoded event log.

Return type:

ASyncIterator[Log]

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]

__wrapped__: AsyncIterable[T]

The wrapped async iterable object.

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

property cache: LogCache
property executor: _AsyncExecutorMixin
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 objects.

Returns:

A list of the objects yielded by the ASyncIterable.

property provider: AddressProvider
property semaphore: BlockSemaphore
to_block
topics
class y.prices.stable_swap.curve.CurveEvents[source]

Bases: ProcessedEvents[_EventItem]

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

__init__(base)[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.

  • base (_LT)

__iter__()

Return an iterator that yields T objects from the ASyncIterable.

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]

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)

Get logs up to a given block.

Parameters:

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

Yields:

A decoded event log.

Return type:

ASyncIterator[Log]

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]

__wrapped__: AsyncIterable[T]

The wrapped async iterable object.

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

property cache: LogCache
property executor: _AsyncExecutorMixin
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 objects.

Returns:

A list of the objects yielded by the ASyncIterable.

property semaphore: BlockSemaphore
to_block
topics
class y.prices.stable_swap.curve.CurvePool[source]

Bases: ERC20

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 the contract address as a string.

Returns:

The contract address as a string.

Return type:

str

__build_name__: HiddenMethodDescriptor[Self, str]

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:

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.

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

__coins__: HiddenMethodDescriptor[Self, List[ERC20]]

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:

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.

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

__coins_decimals__: HiddenMethodDescriptor[Self, List[int]]

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:

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.

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

__decimals__: HiddenMethodDescriptor[Self, int]

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:

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.

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

__factory__: HiddenMethodDescriptor[Self, Contract]

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:

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.

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

__get_underlying_coins__: HiddenMethodDescriptor[Self, List[ERC20]]

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:

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.

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

__name__: HiddenMethodDescriptor[Self, str] = 'CurvePool'
__num_coins__: HiddenMethodDescriptor[Self, int]

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:

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.

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

__scale__: HiddenMethodDescriptor[Self, int]

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:

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.

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

__symbol__: HiddenMethodDescriptor[Self, str]

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:

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.

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

balance_of_readable
Parameters:
Return type:

float

build_name

Get the contract’s build name.

Returns:

The contract’s 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

coins[source]

Get coins of pool.

coins_decimals[source]
property contract: Contract
decimals

The number of decimal places for the token.

Returns:

The number of decimal places for the token.

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

when_no_history_return_0 (bool)

Return type:

int

factory[source]
get_balances[source]

balance} of liquidity in the pool.

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.

Type:

Get {token

Parameters:
Return type:

List[WeiBalance]

get_coin_index[source]

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

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

Return type:

int

get_dy[source]
Parameters:
Return type:

WeiBalance | None

get_tvl[source]

Get total value in Curve pool.

Parameters:
Return type:

UsdValue | None

get_underlying_coins[source]
has_method
Parameters:

  • method (str)

  • return_response (bool)

Return type:

bool | Any

name

The token’s name.

Returns:

The token’s name.

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

scale

Get the scaling factor for the token.

Returns:

The scaling factor for the token.

symbol

The token’s symbol.

Returns:

The token’s symbol.

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

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

class y.prices.stable_swap.curve.CurveRegistry[source]

Bases: ASyncGenericSingleton

__init__(asynchronous=False)[source]
Parameters:

asynchronous (bool)

Return type:

None

__coin_to_pools__: HiddenMethodDescriptor[Self, Dict[str, List[CurvePool]]]

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:

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.

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

__registry__: HiddenMethodDescriptor[Self, Contract]

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:

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.

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

check_liquidity[source]
Parameters:
Return type:

int

coin_to_pools[source]
get_factory[source]

Get metapool factory that has spawned a pool.

Parameters:

pool (str | HexBytes | AnyAddress | EthAddress | Contract)

Return type:

Contract

get_pool[source]

Get Curve pool (swap) address by LP token address. Supports factory pools.

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

Return type:

CurvePool

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:

float | None

get_price_for_underlying[source]

Since get_price_for_underlying 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 identifiers: List[EthAddress]
load_all[source]
Return type:

None

registry[source]
class y.prices.stable_swap.curve.Factory[source]

Bases: _Loader

__await__()

Returns True once the _Loader has loaded all relevant data thru the current block.

Returns:

A generator that yields True once the _Loader has loaded all relevant data thru the current block.

Return type:

Generator[Any, None, Literal[True]]

__eq__(_ContractBase__o)

Return self==value.

Parameters:

_ContractBase__o (object)

Return type:

bool

__init__(address, asynchronous=False)
Parameters:
__str__()

Return the contract address as a string.

Returns:

The contract address as a string.

Return type:

str

__build_name__: HiddenMethodDescriptor[Self, str]

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:

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.

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.

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

when_no_history_return_0 (bool)

Return type:

int

get_pool[source]
Parameters:

i (int)

Return type:

EthAddress

has_method
Parameters:

  • method (str)

  • return_response (bool)

Return type:

bool | Any

property loaded: Awaitable[Literal[True]]

Returns True once the _Loader has loaded all relevant data thru the current block.

Returns:

An awaitable that resolves to True once the _Loader has loaded all relevant data thru the current block.

pool_count[source]
Parameters:

block (int | None)

Return type:

int

read_pools[source]
Return type:

List[EthAddress]

class y.prices.stable_swap.curve.Ids[source]

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.

CryptoPool_Factory = 6
CryptoSwap_Registry = 5
Cryptopool_Factory = 7
CurveStableswapFactoryNG = 12
Curve_Tricrypto_Factory = 11
Exchanges = 2
Fee_Distributor = 4
Main_Registry = 0
Metapool_Factory = 3
PoolInfo_Getters = 1
crvUSD_Plain_Pools = 10
crvUSD_Plain_Pools_deprecated_1 = 8
crvUSD_Plain_Pools_deprecated_2 = 9
denominator

the denominator of a rational number in lowest terms

numerator

the numerator of a rational number in lowest terms

class y.prices.stable_swap.curve.Registry[source]

Bases: _CurveEventsLoader

__await__()

Returns True once the _Loader has loaded all relevant data thru the current block.

Returns:

A generator that yields True once the _Loader has loaded all relevant data thru the current block.

Return type:

Generator[Any, None, Literal[True]]

__eq__(_ContractBase__o)

Return self==value.

Parameters:

_ContractBase__o (object)

Return type:

bool

__init__(address, curve, asynchronous=False)[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]

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:

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.

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.

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

when_no_history_return_0 (bool)

Return type:

int

has_method
Parameters:

  • method (str)

  • return_response (bool)

Return type:

bool | Any

property loaded: Awaitable[Literal[True]]

Returns True once the _Loader has loaded all relevant data thru the current block.

Returns:

An awaitable that resolves to True once the _Loader has loaded all relevant data thru the current block.

class y.prices.stable_swap.curve.RegistryEvents[source]

Bases: CurveEvents

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

__init__(base)[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.

  • base (_LT)

__iter__()

Return an iterator that yields T objects from the ASyncIterable.

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]

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)

Get logs up to a given block.

Parameters:

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

Yields:

A decoded event log.

Return type:

ASyncIterator[Log]

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]

__wrapped__: AsyncIterable[T]

The wrapped async iterable object.

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

property cache: LogCache
property executor: _AsyncExecutorMixin
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 objects.

Returns:

A list of the objects yielded by the ASyncIterable.

property registry: Registry
property semaphore: BlockSemaphore
to_block
topics
y.prices.stable_swap.curve.ADDRESS_PROVIDER = '0x0000000022D53366457F9d5E68Ec105046FC4383'

Curve’s address provider contract on all chains.

y.prices.stable_swap.curve.DED_POOLS = {'0x8301AE4fc9c624d1D396cbDAa1ed877821D7C511': '0xEd4064f376cB8d68F770FB1Ff088a3d0F3FF5c4d'}

The on chain registry no longer returns the lp token address for these dead pools, so we need to provide it manually.

y.prices.stable_swap.ellipsis module

ASyncFunctiony.prices.stable_swap.ellipsis.get_price(token_address: Union[str, hexbytes.main.HexBytes, ~AnyAddress, brownie.convert.datatypes.EthAddress, brownie.network.contract.Contract], block: Union[int, eth_typing.evm.BlockNumber, NoneType] = None, skip_cache: bool = <EnvironmentVariable[name=`YPRICEMAGIC_SKIP_CACHE`, type=bool, default_value=False, current_value=False, using_default=True]>) → y.datatypes.UsdPrice[source]

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

Parameters:
Return type:

UsdPrice

ASyncFunctiony.prices.stable_swap.ellipsis.is_eps_rewards_pool(token_address: str | hexbytes.main.HexBytes | AnyAddress | brownie.convert.datatypes.EthAddress | brownie.network.contract.Contract | int) → bool[source]

Since is_eps_rewards_pool is an ASyncFunctionSyncDefault, you can optionally pass sync=False or asynchronous=True to force it to return a coroutine. Without either kwarg, it will run synchronously.

Parameters:

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

Return type:

bool

y.prices.stable_swap.froyo module

ASyncFunctiony.prices.stable_swap.froyo.get_price(token: str | hexbytes.main.HexBytes | AnyAddress | brownie.convert.datatypes.EthAddress | brownie.network.contract.Contract | int, block: int | eth_typing.evm.BlockNumber | NoneType = None) → y.datatypes.UsdPrice | None[source]

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

Parameters:
Return type:

UsdPrice | None

y.prices.stable_swap.froyo.is_froyo(token)[source]
Parameters:

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

Return type:

bool

y.prices.stable_swap.mstablefeederpool module

ASyncFunctiony.prices.stable_swap.mstablefeederpool.get_price(token: Union[str, hexbytes.main.HexBytes, ~AnyAddress, brownie.convert.datatypes.EthAddress, brownie.network.contract.Contract, int], block: Union[int, eth_typing.evm.BlockNumber, NoneType] = None, skip_cache: bool = <EnvironmentVariable[name=`YPRICEMAGIC_SKIP_CACHE`, type=bool, default_value=False, current_value=False, using_default=True]>) → y.datatypes.UsdPrice[source]

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

Parameters:
Return type:

UsdPrice

ASyncFunctiony.prices.stable_swap.mstablefeederpool.is_mstable_feeder_pool(address: str | hexbytes.main.HexBytes | AnyAddress | brownie.convert.datatypes.EthAddress | brownie.network.contract.Contract | int) → bool[source]

Since is_mstable_feeder_pool is an ASyncFunctionSyncDefault, you can optionally pass sync=False or asynchronous=True to force it to return a coroutine. Without either kwarg, it will run synchronously.

Parameters:

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

Return type:

bool

y.prices.stable_swap.saddle module

ASyncFunctiony.prices.stable_swap.saddle.get_pool(token_address: str | hexbytes.main.HexBytes | AnyAddress | brownie.convert.datatypes.EthAddress | brownie.network.contract.Contract | int) → str | hexbytes.main.HexBytes | AnyAddress | brownie.convert.datatypes.EthAddress[source]

Since get_pool is an ASyncFunctionSyncDefault, you can optionally pass sync=False or asynchronous=True to force it to return a coroutine. Without either kwarg, it will run synchronously.

Parameters:

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

Return type:

str | HexBytes | AnyAddress | EthAddress

ASyncFunctiony.prices.stable_swap.saddle.get_price(token_address: str | hexbytes.main.HexBytes | AnyAddress | brownie.convert.datatypes.EthAddress | brownie.network.contract.Contract, block: int | eth_typing.evm.BlockNumber | NoneType = None) → y.datatypes.UsdPrice[source]

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

Parameters:
Return type:

UsdPrice

ASyncFunctiony.prices.stable_swap.saddle.get_tokens(token_address: str | hexbytes.main.HexBytes | AnyAddress | brownie.convert.datatypes.EthAddress | brownie.network.contract.Contract | int, block: int | eth_typing.evm.BlockNumber | NoneType = None) → List[y.classes.common.ERC20][source]

Since get_tokens is an ASyncFunctionSyncDefault, you can optionally pass sync=False or asynchronous=True to force it to return a coroutine. Without either kwarg, it will run synchronously.

Parameters:
Return type:

List[ERC20]

ASyncFunctiony.prices.stable_swap.saddle.get_tvl(token_address: Union[str, hexbytes.main.HexBytes, ~AnyAddress, brownie.convert.datatypes.EthAddress, brownie.network.contract.Contract, int], block: Union[int, eth_typing.evm.BlockNumber, NoneType] = None, skip_cache: bool = <EnvironmentVariable[name=`YPRICEMAGIC_SKIP_CACHE`, type=bool, default_value=False, current_value=False, using_default=True]>) → y.datatypes.UsdValue[source]

Since get_tvl is an ASyncFunctionSyncDefault, you can optionally pass sync=False or asynchronous=True to force it to return a coroutine. Without either kwarg, it will run synchronously.

Parameters:
Return type:

UsdValue

ASyncFunctiony.prices.stable_swap.saddle.is_saddle_lp(token_address: str | hexbytes.main.HexBytes | AnyAddress | brownie.convert.datatypes.EthAddress | brownie.network.contract.Contract | int) → bool[source]

Since is_saddle_lp is an ASyncFunctionSyncDefault, you can optionally pass sync=False or asynchronous=True to force it to return a coroutine. Without either kwarg, it will run synchronously.

Parameters:

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

Return type:

bool

Module contents