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]

Get the price of a Belt LP token in USD.

This function retrieves the virtual price of a Belt LP token from its associated pool and converts it to a USD price.

Args:

token: The address of the Belt LP token. block: The block number at which to fetch the price. Defaults to the latest block.

Raises:

KeyError: If the token is not found in the POOLS dictionary.

Examples:
>>> await get_price("0x86aFa7ff694Ab8C985b79733745662760e454169")
UsdPrice(1.23456789)
>>> await get_price("0x9cb73F20164e399958261c289Eb5F9846f4D1404", block=12345678)
UsdPrice(1.23456789)
See Also:

is_belt_lp() to check if a token is a Belt LP token.

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]

Check if a token is a Belt LP token.

Parameters:

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

Returns:

True if the token is a Belt LP token, False otherwise.

Return type:

bool

Examples

>>> is_belt_lp("0x86aFa7ff694Ab8C985b79733745662760e454169")
True
>>> is_belt_lp("0x0000000000000000000000000000000000000000")
False

See also

POOLS for the list of recognized Belt LP tokens.

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

Examples

>>> loader = _Loader("0x1234567890abcdef1234567890abcdef12345678")
>>> await loader
True
__eq__(_ContractBase__o)

Return self==value.

Parameters:

_ContractBase__o (object)

Return type:

bool

__init__(self)[source]
Parameters:
__str__()

Return the contract address as a string.

Returns:

The contract address as a string.

Return type:

str

__build_name__: HiddenMethodDescriptor[Self, str]

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

Args:

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

Returns:

The property value.

The original docstring for get() is shown below:

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

Args:

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

Returns:

The property value.

Type:

_ASyncPropertyDescriptorBase.get(self, instance

Type:

I, owner

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

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

EthAddress

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

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.

Examples

>>> loader = _EventsLoader("0x1234567890abcdef1234567890abcdef12345678")
>>> await loader.loaded
True
class y.prices.stable_swap.curve.AddressProviderEvents[source]

Bases: CurveEvents

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

Example

>>> my_object = AddressProviderEvents(...)
>>> all_contents = await my_object
>>> isinstance(all_contents, list)
True
>>> isinstance(all_contents[0], _EventItem)
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__(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)

Examples

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

Return an iterator that yields T objects from the ASyncIterable.

Note

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

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

Return type:

Iterator[T]

_objects_thru(block)
Parameters:

block (int | None)

Return type:

AsyncIterator[T]

events(to_block)

Get events up to a given block.

Parameters:

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

Yields:

A decoded event.

Return type:

ASyncIterator[_EventItem]

Examples

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

Filters the T objects yielded by the ASyncIterable based on a function.

Parameters:

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

Returns:

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

Return type:

ASyncFilter[T]

logs(to_block)

Get logs up to a given block.

Parameters:

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

Yields:

A raw log.

Return type:

ASyncIterator[Log]

Examples

>>> log_filter = LogFilter(addresses=["0x1234..."], topics=["0x5678..."])
>>> logs = log_filter.logs(1000100)
>>> print(logs)
objects(to_block)

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

Parameters:

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

Returns:

An ASyncIterator that yields all included events.

Return type:

ASyncIterator[_EventItem]

Examples

>>> processed_events = ProcessedEvents(addresses=["0x1234..."], topics=["0x5678..."])
>>> async for event in processed_events.objects(1000100):
...     print(event)
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
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 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 _EventItem will be returned.

Example

>>> my_object = CurveEvents(...)
>>> all_contents = await my_object
>>> isinstance(all_contents, list)
True
>>> isinstance(all_contents[0], _EventItem)
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__(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)

Examples

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

Return an iterator that yields T objects from the ASyncIterable.

Note

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

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

Return type:

Iterator[T]

_objects_thru(block)
Parameters:

block (int | None)

Return type:

AsyncIterator[T]

events(to_block)

Get events up to a given block.

Parameters:

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

Yields:

A decoded event.

Return type:

ASyncIterator[_EventItem]

Examples

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

Filters the T objects yielded by the ASyncIterable based on a function.

Parameters:

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

Returns:

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

Return type:

ASyncFilter[T]

logs(to_block)

Get logs up to a given block.

Parameters:

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

Yields:

A raw log.

Return type:

ASyncIterator[Log]

Examples

>>> log_filter = LogFilter(addresses=["0x1234..."], topics=["0x5678..."])
>>> logs = log_filter.logs(1000100)
>>> print(logs)
objects(to_block)

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

Parameters:

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

Returns:

An ASyncIterator that yields all included events.

Return type:

ASyncIterator[_EventItem]

Examples

>>> processed_events = ProcessedEvents(addresses=["0x1234..."], topics=["0x5678..."])
>>> async for event in processed_events.objects(1000100):
...     print(event)
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
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.stable_swap.curve.CurvePool[source]

Bases: ERC20

Represents a Curve pool.

This class provides methods to interact with Curve pools, including fetching pool information, calculating token exchanges, and retrieving liquidity data.

Note

This class inherits from ERC20 for convenience, but a Curve pool is not always an ERC20 token. This inheritance is used to leverage existing functionality for token interactions.

- :meth:`coins`

Fetches the coins in the pool.

- :meth:`get_dy`

Calculates the amount of output tokens for a given input amount, simulating a token exchange.

- :meth:`get_balances`

Retrieves the balances of tokens in the pool.

- :meth:`get_tvl`

Retrieves the total value locked in the pool.

Examples

>>> pool = CurvePool("0x1234567890abcdef1234567890abcdef12345678")
>>> await pool.coins
[<ERC20 TKN1 '0x...'>, <ERC20 TKN2 '0x...'>]
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:
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

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

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

__decimals__: HiddenMethodDescriptor[Self, int]

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

Args:

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

Returns:

The property value.

The original docstring for get() is shown below:

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

Args:

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

Returns:

The property value.

Type:

_ASyncPropertyDescriptorBase.get(self, instance

Type:

I, owner

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

__factory__: HiddenMethodDescriptor[Self, Contract]

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

Args:

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

Returns:

The property value.

The original docstring for get() is shown below:

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

Args:

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

Returns:

The property value.

Type:

_ASyncPropertyDescriptorBase.get(self, instance

Type:

I, owner

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

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

__name__: HiddenMethodDescriptor[Self, str] = 'CurvePool'
__num_coins__: 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

__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

address: Address

The contract address of the token.

asynchronous: bool = False
balance_of

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

Parameters:
Returns:

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

Return type:

int

Examples

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

float

build_name

Get the contract’s build name.

Returns:

The contract’s build name.

Examples

>>> contract = ContractBase("0x1234567890abcdef1234567890abcdef12345678")
>>> await contract.build_name
'MyContract'
check_liquidity[source]

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.

Returns:

A list of ERC20 tokens representing the coins in the pool.

Examples

>>> pool = CurvePool("0x1234567890abcdef1234567890abcdef12345678")
>>> await pool.coins
[<ERC20 TKN1 '0x...'>, <ERC20 TKN2 '0x...'>]
coins_decimals[source]
property contract: Contract
decimals

The number of decimal places for the token.

Returns:

The number of decimal places for the token.

Examples

>>> token = ERC20("0x1234567890abcdef1234567890abcdef12345678")
>>> await token.decimals
18
deploy_block: ASyncBoundMethod[Self, Any, int]

Get the block number when the contract was deployed.

Parameters:

when_no_history_return_0 (bool) – If True, return 0 when no history is found instead of raising an exception.

Returns:

The block number when the contract was deployed, or 0 if when_no_history_return_0 is True and the deploy block cannot be determined.

Return type:

int

Examples

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

See also

  • contract_creation_block_async()

factory[source]
get_balances[source]

balance} of liquidity in the pool.

Args:

block: The block number to query. Defaults to the latest block. skip_cache: If True, skip using the cache while fetching balance data.

Returns:

A list of WeiBalance objects representing the balances of tokens in the pool.

Examples:
>>> pool = CurvePool("0x1234567890abcdef1234567890abcdef12345678")
>>> await pool.get_balances()
[<WeiBalance token=<ERC20 TKN1 '0x...'> balance=1000000000000000000 block=None>, ...]

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:

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

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

Returns:

The total value locked in the pool as a UsdValue, or None if the TVL cannot be determined.

Return type:

UsdValue | None

Examples

>>> pool = CurvePool("0x1234567890abcdef1234567890abcdef12345678")
>>> await pool.get_tvl()
UsdValue(1234567.89)
get_underlying_coins[source]
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

name

The token’s name.

Returns:

The token’s name.

Examples

>>> token = ERC20("0x1234567890abcdef1234567890abcdef12345678")
>>> await token.name
'TokenName'
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

Examples

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

See also

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'
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
class y.prices.stable_swap.curve.CurveRegistry[source]

Bases: ASyncGenericSingleton

__init__(self)[source]
Parameters:

asynchronous (bool)

Return type:

None

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

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

__registry__: HiddenMethodDescriptor[Self, Contract]

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

Args:

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

Returns:

The property value.

The original docstring for get() is shown below:

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

Args:

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

Returns:

The property value.

Type:

_ASyncPropertyDescriptorBase.get(self, instance

Type:

I, owner

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

check_liquidity[source]
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) – The address or contract of the pool.

Returns:

The Contract representing the factory.

Return type:

Contract

Examples

>>> factory = await curve.get_factory("0x1234567890abcdef1234567890abcdef12345678")
>>> print(factory)
<Contract '0x...'>
get_pool[source]

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

Args:

token: The address of the LP token.

Returns:

The CurvePool associated with the LP token.

Examples:
>>> pool = await curve.get_pool("0x1234567890abcdef1234567890abcdef12345678")
>>> print(pool)
<CurvePool '0x...'>

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

Examples

>>> loader = _Loader("0x1234567890abcdef1234567890abcdef12345678")
>>> await loader
True
__eq__(_ContractBase__o)

Return self==value.

Parameters:

_ContractBase__o (object)

Return type:

bool

__init__(self)
Parameters:
__str__()

Return the contract address as a string.

Returns:

The contract address as a string.

Return type:

str

__build_name__: HiddenMethodDescriptor[Self, str]

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

Args:

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

Returns:

The property value.

The original docstring for get() is shown below:

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

Args:

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

Returns:

The property value.

Type:

_ASyncPropertyDescriptorBase.get(self, instance

Type:

I, owner

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

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

i (int)

Return type:

EthAddress

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

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.

Examples

>>> loader = _Loader("0x1234567890abcdef1234567890abcdef12345678")
>>> await loader.loaded
True
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]]

Examples

>>> loader = _Loader("0x1234567890abcdef1234567890abcdef12345678")
>>> await loader
True
__eq__(_ContractBase__o)

Return self==value.

Parameters:

_ContractBase__o (object)

Return type:

bool

__init__(self)[source]
Parameters:
__str__()

Return the contract address as a string.

Returns:

The contract address as a string.

Return type:

str

__build_name__: HiddenMethodDescriptor[Self, str]

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

Args:

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

Returns:

The property value.

The original docstring for get() is shown below:

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

Args:

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

Returns:

The property value.

Type:

_ASyncPropertyDescriptorBase.get(self, instance

Type:

I, owner

Parameters:

  • instance (I)

  • owner (Type[I] | None)

Return type:

T

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

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

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.

Examples

>>> loader = _EventsLoader("0x1234567890abcdef1234567890abcdef12345678")
>>> await loader.loaded
True
class y.prices.stable_swap.curve.RegistryEvents[source]

Bases: CurveEvents

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

Example

>>> my_object = RegistryEvents(...)
>>> all_contents = await my_object
>>> isinstance(all_contents, list)
True
>>> isinstance(all_contents[0], _EventItem)
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__(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)

Examples

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

Return an iterator that yields T objects from the ASyncIterable.

Note

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

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

Return type:

Iterator[T]

_objects_thru(block)
Parameters:

block (int | None)

Return type:

AsyncIterator[T]

events(to_block)

Get events up to a given block.

Parameters:

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

Yields:

A decoded event.

Return type:

ASyncIterator[_EventItem]

Examples

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

Filters the T objects yielded by the ASyncIterable based on a function.

Parameters:

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

Returns:

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

Return type:

ASyncFilter[T]

logs(to_block)

Get logs up to a given block.

Parameters:

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

Yields:

A raw log.

Return type:

ASyncIterator[Log]

Examples

>>> log_filter = LogFilter(addresses=["0x1234..."], topics=["0x5678..."])
>>> logs = log_filter.logs(1000100)
>>> print(logs)
objects(to_block)

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

Parameters:

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

Returns:

An ASyncIterator that yields all included events.

Return type:

ASyncIterator[_EventItem]

Examples

>>> processed_events = ProcessedEvents(addresses=["0x1234..."], topics=["0x5678..."])
>>> async for event in processed_events.objects(1000100):
...     print(event)
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
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 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]

Get the USD price of a token at a specific block.

This function calculates the price of a token in USD by determining the total value locked (TVL) and dividing it by the total supply of the token.

Args:

token_address: The address of the token to get the price for. block: The block number to query. Defaults to the latest block. skip_cache: If True, skip using the cache while fetching price data.

Returns:

The price of the token in USD.

Examples:
>>> await get_price("0x1234567890abcdef1234567890abcdef12345678")
UsdPrice(1.23)
See Also:

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]

Check if a given token address is an EPS rewards pool.

This function verifies if the specified token address has the methods associated with an EPS rewards pool.

Args:

token_address: The address of the token to check.

Returns:

True if the token address is an EPS rewards pool, False otherwise.

Examples:
>>> await is_eps_rewards_pool("0x1234567890abcdef1234567890abcdef12345678")
True
See Also:

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]

Get the USD price of the Froyo pool token on the Fantom network.

This function retrieves the virtual price of the Froyo pool token if the provided token address matches the Froyo pool address on the Fantom network. The price is returned as a y.datatypes.UsdPrice object.

Args:

token: The token address to get the price for. block: The block number to query the price at, defaults to the latest block.

Returns:

The USD price of the Froyo pool token, or None if the token is not the Froyo pool.

Examples:
>>> await get_price("0x83E5f18Da720119fF363cF63417628eB0e9fd523")
UsdPrice(1.23456789)
>>> await get_price("0x0000000000000000000000000000000000000000")
None
See Also:

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]

Check if a given token is the Froyo token on the Fantom network.

Parameters:

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

Returns:

True if the token is the Froyo token on Fantom, False otherwise.

Return type:

bool

Examples

>>> is_froyo("0x4f85Bbf3B0265DCEd4Ec72ebD0358ccCf190F1B3")
True
>>> is_froyo("0x0000000000000000000000000000000000000000")
False

See also

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]

Get the price of an mStable Feeder Pool token in USD.

This function calculates the price of a given mStable Feeder Pool token by fetching the price ratio and the underlying mAsset price, then combining them to get the USD price.

Args:

token: The token address to get the price for. block: The block number to query. Defaults to the latest block. skip_cache: If True, skip using the cache while fetching price data.

Examples:
>>> await get_price("0x1234567890abcdef1234567890abcdef12345678")
UsdPrice(1.23)
Returns:

The price of the token in USD.

See Also:

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]

Check if a given address is an mStable Feeder Pool.

This function verifies if the contract at the specified address implements the methods required for an mStable Feeder Pool, specifically getPrice()((uint256,uint256)) and mAsset()(address).

Args:

address: The address to check.

Examples:
>>> await is_mstable_feeder_pool("0x1234567890abcdef1234567890abcdef12345678")
True
Returns:

True if the address is an mStable Feeder Pool, False otherwise.

See Also:

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]

Retrieve the pool address for a given Saddle LP token.

This function checks for specific token addresses on the Ethereum Mainnet and returns hardcoded pool addresses for those tokens. If the token is not one of the known tokens, it attempts to find a pool by checking if the token has a swap() method.

Args:

token_address: The address of the token to find the pool for.

Returns:

The address of the pool if found, otherwise None.

Examples:
>>> await get_pool("0xc9da65931ABf0Ed1b74Ce5ad8c041C4220940368")
'0xa6018520EAACC06C30fF2e1B3ee2c7c22e64196a'
See Also:

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]

Calculate the price of a Saddle LP token in USD.

This function calculates the price by dividing the total value locked (TVL) by the total supply of the token.

Args:

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

Examples:
>>> await get_price("0x1234567890abcdef1234567890abcdef12345678")
UsdPrice(1.2345)
See Also:

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]

Retrieve the list of tokens in a Saddle LP token pool.

This function retrieves the pool and then uses multicall to get the tokens in the pool.

Args:

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

Examples:
>>> await get_tokens("0x1234567890abcdef1234567890abcdef12345678")
[ERC20('0xToken1'), ERC20('0xToken2')]
See Also:

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]

Calculate the total value locked (TVL) in a Saddle LP token pool.

This function retrieves the pool and tokens concurrently, then uses multicall to get the balances. It calculates the TVL by summing the value of each token in the pool.

Args:

token_address: The address of the token to get the TVL for. block: The block number to query. Defaults to the latest block. skip_cache: If True, skip using the cache while fetching price data.

Examples:
>>> await get_tvl("0x1234567890abcdef1234567890abcdef12345678")
UsdValue(123456.78)
See Also:

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]

Determine if a given token is a Saddle LP token.

This function checks if the token has specific methods that are characteristic of Saddle LP tokens.

Args:

token_address: The address of the token to check.

Returns:

True if the token is a Saddle LP token, False otherwise.

Examples:
>>> await is_saddle_lp("0x1234567890abcdef1234567890abcdef12345678")
False
See Also:

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