y.prices package

Subpackages

Submodules

y.prices.band module

class y.prices.band.Band[source]

Bases: ASyncGenericSingleton

A class to interact with the Band Protocol oracle on supported networks.

This class provides methods to check asset support and fetch prices using the Band Protocol oracle.

Raises:: UnsupportedNetwork – If the Band Protocol is not supported on the current network.

Examples

>>> band = Band(asynchronous=True)
>>> "0xaf319E5789945197e365E7f7fbFc56B130523B33" in band
True

See also

a_sync.ASyncGenericSingleton

__contains__(asset)[source]

Check if an asset is supported by the Band Protocol on the current network.

Parameters:: asset (str | HexBytes | AnyAddress | EthAddress | Contract) – The asset address or contract to check.
Return type:: bool

Examples

>>> band = Band(asynchronous=True)
>>> "0xaf319E5789945197e365E7f7fbFc56B130523B33" in band
True

__init__(*, asynchronous=False)[source]

Initialize the Band class.

Parameters:: asynchronous (bool) – Whether to use asynchronous operations.
Raises:: UnsupportedNetwork – If the Band Protocol is not supported on the current network.
Return type:: None

Examples

>>> band = Band(asynchronous=True)

__oracle__: HiddenMethodDescriptor[Self, Contract]

Optional[Type[I]] = None) -> T Asynchronously retrieves 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.

Type:

_ASyncPropertyDescriptorBase.get(self, instance

Type:

I, owner

Parameters:

instance (I)
owner (Type[I] | None)

Return type:

T

get_price[source]

Get the price of an asset in terms of USDC using the Band Protocol oracle.

Parameters:

asset (str | HexBytes | AnyAddress | EthAddress) – The asset address to get the price for.
block (int | BlockNumber | None) – The block number to query the price at. Defaults to the latest block.

Returns:

The price of the asset in terms of USDC, or None if the price cannot be fetched.

Return type:

float | None

Examples

>>> band = Band(asynchronous=True)
>>> price = await band.get_price("0xaf319E5789945197e365E7f7fbFc56B130523B33")
>>> print(price)
1.0

oracle[source]

Get the Band Protocol oracle contract for the current network.

Examples

>>> band = Band(asynchronous=True)
>>> oracle = await band.oracle

y.prices.chainlink module

class y.prices.chainlink.Chainlink[source]

Bases: ASyncGenericBase

__init__(*, asynchronous=True)[source]

Initialize the Chainlink class.

Parameters:: asynchronous (bool) – Whether to use asynchronous operations.
Raises:: UnsupportedNetwork – If Chainlink is not supported on the current network.
Return type:: None

Examples

>>> chainlink = Chainlink(asynchronous=True)

get_feed[source]

Get the feed for a specific asset.

Since get_feed 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:: asset (str | HexBytes | AnyAddress | EthAddress)
Return type:: Feed | None

get_price[source]

Get the price of an asset at a specific block.

If the block is not specified, it uses the latest block.

Parameters:

asset (str | HexBytes | AnyAddress | EthAddress | Contract | int) – The address of the asset.
block (int | BlockNumber | None) – The block number to get the price for.

Return type:

UsdPrice | None

Examples

>>> price = await chainlink.get_price("0xAsset", 12345678)
>>> print(price)

Returns:

The price of the asset in USD, or None if the price cannot be fetched.

Parameters:

asset (str | HexBytes | AnyAddress | EthAddress | Contract | int)
block (int | BlockNumber | None)

Return type:

UsdPrice | None

See also

get_feed()

has_feed[source]

Check if a feed exists for a specific asset.

Parameters:: asset (str | HexBytes | AnyAddress | EthAddress | Contract | int) – The address of the asset.
Return type:: bool

Examples

>>> has_feed = await chainlink.has_feed("0xAsset")
>>> print(has_feed)

Returns:: True if a feed exists for the specified asset, False otherwise.
Parameters:: asset (str | HexBytes | AnyAddress | EthAddress | Contract | int)
Return type:: bool

class y.prices.chainlink.Feed[source]

Bases: object

__init__(address, asset, start_block=0, *, asynchronous=False)[source]

Parameters:

address (str | HexBytes | AnyAddress | EthAddress | Contract | int)
asset (str | HexBytes | AnyAddress | EthAddress | Contract | int)
start_block (int)
asynchronous (bool)

async decimals()[source]

Return type:: int

async get_price(block)[source]

Get the price of the asset at a specific block.

If the feed is stale, it returns None.

Parameters:: block (int) – The block number to get the price for.
Return type:: UsdPrice | None

Examples

>>> feed = Feed("0xAddress", "0xAsset")
>>> price = await feed.get_price(12345678)
>>> print(price)

Returns:: The price of the asset in USD, or None if the feed is stale.
Parameters:: block (int)
Return type:: UsdPrice | None

address

asset

property contract: Contract

latest_answer

latest_timestamp

scale = <async_lru._LRUCacheWrapperInstanceMethod object>[source]

Return type:: int | None

start_block

class y.prices.chainlink.FeedsFromEvents[source]

Bases: ProcessedEvents[Feed]

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

Example

>>> my_object = FeedsFromEvents(...)
>>> all_contents = await my_object
>>> isinstance(all_contents, list)
True
>>> isinstance(all_contents[0], Feed)
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__(addresses, topics, *, asynchronous=True)[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.
asynchronous (bool)

Examples

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

__iter__(self) → Iterator[T]

Return an iterator that yields T objects from the ASyncIterable.

Note

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

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

Return type:: Iterator[T]

_objects_thru(block)

Parameters:: block (int | None)
Return type:: AsyncIterator[T]

events(to_block)

Get events up to a given block.

Parameters:: to_block (int) – The ending block to fetch events to.
Yields:: A decoded event.
Return type:: ASyncIterator[_EventItem]

Examples

>>> events = Events(addresses=["0x1234..."], topics=["0x5678..."])
>>> async for event in events.events(1000100):
...     print(event)

filter(self, function: ViewFn[T]) → 'ASyncFilter[T]'

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

Parameters:: function (Callable[[T], Awaitable[bool]] | Callable[[T], bool]) – A function that returns a boolean that indicates if an item should be included in the filtered result. Can be sync or async.
Returns:: An instance of ASyncFilter that yields the filtered T objects from the ASyncIterable.
Return type:: ASyncFilter[T]

logs(to_block)

Get logs up to a given block.

Parameters:: to_block (int | None) – The ending block to fetch logs to.
Yields:: A raw log.
Return type:: ASyncIterator[Log]

Examples

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

objects(to_block)

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

Parameters:: to_block (int) – The ending block to fetch events to.
Returns:: An ASyncIterator that yields all included events.
Return type:: ASyncIterator[_EventItem]

Examples

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

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

Sort the T objects yielded by the ASyncIterable.

Parameters:

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

Returns:

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

Return type:

ASyncSorter[T]

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

Class method to wrap an AsyncIterable for backward compatibility.

Parameters:: wrapped (AsyncIterable[T])
Return type:: ASyncIterable[T]

__wrapped__

addresses

asynchronous

property bulk_insert: Callable[[List[Log]], Awaitable[None]]

Get the function for bulk inserting logs into the database.

Returns:: A function for bulk inserting logs.

Examples

>>> log_filter = LogFilter(addresses=["0x1234..."], topics=["0x5678..."])
>>> await log_filter.bulk_insert(logs)

property cache: LogCache

property executor: AsyncThreadPoolExecutor

from_block

property insert_to_db: Callable[[Log], None]

Get the function for inserting logs into the database.

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

property is_asleep: bool

is_reusable

property materialized: List[T]

Synchronously iterate through the ASyncIterable and return all T objects.

Returns:: A list of the T objects yielded by the ASyncIterable.

property semaphore: BlockSemaphore

to_block

topics

y.prices.convex module

ASyncFunctiony.prices.convex.get_price(token_address: eth_typing.evm.ChecksumAddress, 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:

token_address (ChecksumAddress)
block (int | BlockNumber | None)
skip_cache (bool)

Return type:

UsdPrice

y.prices.convex.is_convex_lp(token_address)[source]

Check if a given token address is a Convex LP token.

Parameters:: token_address (ChecksumAddress) – The address of the token to check.
Returns:: True if the token is a Convex LP token, False otherwise.
Return type:: bool

Examples

>>> is_convex_lp("0x30D9410ED1D5DA1F6C8391af5338C93ab8d4035C")
True

>>> is_convex_lp("0x0000000000000000000000000000000000000000")
False

y.prices.gearbox module

class y.prices.gearbox.DieselPool[source]

Bases: ContractBase

__eq__(_ContractBase__o)

Return self==value.

Parameters:: _ContractBase__o (object)
Return type:: bool

__init__(self)

Parameters:

address (str | HexBytes | AnyAddress | EthAddress | Contract | int)
asynchronous (bool)
_deploy_block (int | None)

Return type:

None

__str__()

Return the contract address as a string.

Returns:: The contract address as a string.
Return type:: str

__build_name__: HiddenMethodDescriptor[Self, str]

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

The original docstring for get() is shown below:

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

Type:

_ASyncPropertyDescriptorBase.get(self, instance

Type:

I, owner

Parameters:

instance (I)
owner (Type[I] | None)

Return type:

T

__contract__: HiddenMethodDescriptor[Self, Contract]

Optional[Type[I]] = None) -> T Asynchronously retrieves 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.

Type:

_ASyncPropertyDescriptorBase.get(self, instance

Type:

I, owner

Parameters:

instance (I)
owner (Type[I] | None)

Return type:

T

__diesel_token__: HiddenMethodDescriptor[Self, ERC20]

Optional[Type[I]] = None) -> T Asynchronously retrieves 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.

Type:

_ASyncPropertyDescriptorBase.get(self, instance

Type:

I, owner

Parameters:

instance (I)
owner (Type[I] | None)

Return type:

T

__underlying__: HiddenMethodDescriptor[Self, ERC20]

Optional[Type[I]] = None) -> T Asynchronously retrieves 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.

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'

contract[source]

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

diesel_token[source]

exchange_rate[source]

Parameters:: block (int | BlockNumber)
Return type:: Decimal

get_price[source]

Parameters:

block (int | BlockNumber)
skip_cache (bool)

Return type:

Decimal

has_method

Check if the contract has a specific method.

Parameters:

method (str) – The name of the method to check for.
return_response (bool) – If True, return the response of the method call instead of a boolean.

Returns:

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

Return type:

bool | Any

Examples

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

See also

has_method()

underlying[source]

class y.prices.gearbox.Gearbox[source]

Bases: ASyncGenericBase

__init__(*, asynchronous=False)[source]

Initialize a Gearbox instance.

Raises:: UnsupportedNetwork – If the current network is not Ethereum Mainnet.
Parameters:: asynchronous (bool) – Whether to use asynchronous operations. Defaults to False.

Example

>>> from y.prices.gearbox import Gearbox
>>> gearbox = Gearbox(asynchronous=True)
>>> # This will raise an UnsupportedNetwork exception if not on Mainnet

__registry__

Optional[Type[I]] = None) -> T Asynchronously retrieves 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.

Type:

_ASyncPropertyDescriptorBase.get(self, instance

Type:

I, owner

Parameters:

instance (I)
owner (Type[I] | None)

Return type:

T

diesel_tokens[source]

Return type:: Dict[ERC20, DieselPool]

get_price[source]

Parameters:

token (str | HexBytes | AnyAddress | EthAddress)
block (int | BlockNumber)
skip_cache (bool)

Return type:

Decimal

is_diesel_token[source]

Parameters:: token (str | HexBytes | AnyAddress | EthAddress)
Return type:: bool

pools[source]

Since pools 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.

Return type:: List[DieselPool]

registry[source]

y.prices.magic module

ASyncFunctiony.prices.magic.get_price(token_address: AnyAddressType, block: Block | None = None, *, fail_to_None: Literal[True], skip_cache: bool = ENVS.SKIP_CACHE, ignore_pools: Tuple[Pool, ...] = (), silent: bool = False) → UsdPrice | None[source]

ASyncFunctiony.prices.magic.get_price(token_address: AnyAddressType, block: Block | None = None, *, fail_to_None: bool = False, skip_cache: bool = ENVS.SKIP_CACHE, ignore_pools: Tuple[Pool, ...] = (), silent: bool = False) → UsdPrice

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.

ASyncFunctiony.prices.magic.get_prices(token_addresses: Iterable[AnyAddressType], block: Block | None = None, *, fail_to_None: Literal[True], skip_cache: bool = ENVS.SKIP_CACHE, silent: bool = False) → List[UsdPrice | None][source]

ASyncFunctiony.prices.magic.get_prices(token_addresses: Iterable[AnyAddressType], block: Block | None = None, *, fail_to_None: bool = False, skip_cache: bool = ENVS.SKIP_CACHE, silent: bool = False) → List[UsdPrice]

Since get_prices 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.

Map token addresses to their prices asynchronously.

Parameters:

token_addresses – An iterable of token addresses to price.
block (optional) – The block number at which to get the prices. Defaults to latest block.
fail_to_None (optional) – If True, map to None for tokens whose price couldn’t be determined. Default False.
skip_cache (optional) – If True, bypass the cache and fetch prices directly. Defaults to ENVS.SKIP_CACHE.
silent (optional) – If True, suppress error logging. Default False.

Returns:

An a_sync.TaskMapping object mapping token addresses to their prices.

y.prices.one_to_one module

ASyncFunctiony.prices.one_to_one.get_price(token_address: eth_typing.evm.ChecksumAddress, 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:

token_address (ChecksumAddress)
block (int | BlockNumber | None)
skip_cache (bool)

Return type:

UsdPrice

y.prices.one_to_one.is_one_to_one_token(token_address)[source]

Check if a token address is a one-to-one token.

Parameters:: token_address (ChecksumAddress) – The address of the token to check.
Returns:: True if the token is a one-to-one token, False otherwise.
Return type:: bool

Examples

>>> is_one_to_one_token("0x4da27a545c0c5B758a6BA100e3a049001de870f5")
True

>>> is_one_to_one_token("0x0000000000000000000000000000000000000000")
False

y.prices.pendle module

ASyncFunctiony.prices.pendle.get_lp_price(token: Union[str, hexbytes.main.HexBytes, ~AnyAddress, brownie.convert.datatypes.EthAddress], block: Union[int, eth_typing.evm.BlockNumber] = None, skip_cache: bool = <EnvironmentVariable[name=`YPRICEMAGIC_SKIP_CACHE`, type=bool, default_value=False, current_value=False, using_default=True]>) → decimal.Decimal[source]

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

token (str | HexBytes | AnyAddress | EthAddress)
block (int | BlockNumber)
skip_cache (bool)

Return type:

Decimal

y.prices.pendle.get_tokens(lp_token)[source]

Retrieves the addresses of the tokens in a Pendle LP token.

This function is cached to improve performance for repeated calls.

Parameters:: lp_token (str | HexBytes | AnyAddress | EthAddress) – The address of the Pendle LP token.
Returns:: A tuple containing the addresses of the SY token, PT token, and YT token.
Return type:: Tuple[str, str, str]

Example

>>> tokens = await get_tokens("0x1b92b5242301ce4a8c73cc3ef0d6dee33a3a5b23")
>>> print(tokens)
('0x...', '0x...', '0x...')  # Addresses of SY, PT, and YT tokens

Note

The function returns the result of the readTokens method from the Pendle LP contract, which provides the addresses of the underlying tokens.

See also

is_pendle_lp() for checking if a token is a Pendle LP token.

ASyncFunctiony.prices.pendle.is_pendle_lp(token: str | hexbytes.main.HexBytes | AnyAddress | brownie.convert.datatypes.EthAddress) → bool[source]

Since is_pendle_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 (str | HexBytes | AnyAddress | EthAddress)
Return type:: bool

y.prices.popsicle module

ASyncFunctiony.prices.popsicle.get_balances(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]>, _async_balance_objects: bool = False) → Tuple[y.classes.common.WeiBalance, y.classes.common.WeiBalance] | None[source]

Since get_balances 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 (str | HexBytes | AnyAddress | EthAddress | Contract | int)
block (int | BlockNumber | None)
skip_cache (bool)
_async_balance_objects (bool)

Return type:

Tuple[WeiBalance, WeiBalance] | None

ASyncFunctiony.prices.popsicle.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 | 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:

token (str | HexBytes | AnyAddress | EthAddress | Contract | int)
block (int | BlockNumber | None)
skip_cache (bool)

Return type:

UsdPrice | None

ASyncFunctiony.prices.popsicle.get_tvl(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.UsdValue | None[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:

token (str | HexBytes | AnyAddress | EthAddress | Contract | int)
block (int | BlockNumber | None)
skip_cache (bool)

Return type:

UsdValue | None

Since is_popsicle_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

y.prices.rkp3r module

async y.prices.rkp3r.get_discount(block=None)[source]

Retrieve the discount for the RKP3R token from the contract.

Parameters:: block (int | BlockNumber | None) – The block number to query the discount at.
Returns:: The discount as a percentage.
Return type:: Decimal

Examples

>>> await get_discount()
Decimal('5.0')

See also

y.contracts.Contract

ASyncFunctiony.prices.rkp3r.get_price(address: Union[str, hexbytes.main.HexBytes, ~AnyAddress, brownie.convert.datatypes.EthAddress], 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]>) → decimal.Decimal[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:

address (str | HexBytes | AnyAddress | EthAddress)
block (int | BlockNumber | None)
skip_cache (bool)

Return type:

Decimal

y.prices.rkp3r.is_rkp3r(address)[source]

Check if the given address is the RKP3R token on the Ethereum Mainnet.

Parameters:: address (str | HexBytes | AnyAddress | EthAddress) – The address to check.
Returns:: True if the address is RKP3R on Mainnet, False otherwise.
Return type:: bool

Examples

>>> is_rkp3r("0xEdB67Ee1B171c4eC66E6c10EC43EDBbA20FaE8e9")
True

>>> is_rkp3r("0x1cEB5cB57C4D4E2b2433641b95Dd330A33185A44")
False

y.prices.solidex module

ASyncFunctiony.prices.solidex.get_price(token: Union[str, hexbytes.main.HexBytes, ~AnyAddress, brownie.convert.datatypes.EthAddress, brownie.network.contract.Contract, int], block: Optional[int] = None, skip_cache: bool = <EnvironmentVariable[name=`YPRICEMAGIC_SKIP_CACHE`, type=bool, default_value=False, current_value=False, using_default=True]>)[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:

token (str | HexBytes | AnyAddress | EthAddress | Contract | int)
block (int | None)
skip_cache (bool)

Since is_solidex_deposit 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 (str | HexBytes | AnyAddress | EthAddress | Contract | int)
Return type:: bool

y.prices.synthetix module

class y.prices.synthetix.Synthetix[source]

Bases: ASyncGenericSingleton

A class to interact with the Synthetix protocol.

This class provides methods to interact with the Synthetix protocol, allowing users to get contract addresses, synths, and prices.

Raises:: UnsupportedNetwork – If the Synthetix protocol is not supported on the current network.

Examples

>>> synthetix = Synthetix(asynchronous=True)
>>> address = await synthetix.get_address("ProxyERC20")
>>> print(address)
<Contract object at 0x...>

__init__(self)[source]

Parameters:: asynchronous (bool)
Return type:: None

__address_resolver__: HiddenMethodDescriptor[Self, Contract]

Optional[Type[I]] = None) -> T Asynchronously retrieves 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.

Type:

_ASyncPropertyDescriptorBase.get(self, instance

Type:

I, owner

Parameters:

instance (I)
owner (Type[I] | None)

Return type:

T

__synths__

Optional[Type[I]] = None) -> T Asynchronously retrieves 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.

Type:

_ASyncPropertyDescriptorBase.get(self, instance

Type:

I, owner

Parameters:

instance (I)
owner (Type[I] | None)

Return type:

T

address_resolver[source]

Get the address resolver contract.

Returns:: The address resolver contract.

Examples

>>> resolver = await synthetix.address_resolver
>>> print(resolver)
<Contract object at 0x...>

get_address[source]

Get contract from Synthetix registry.

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

name (str)
block (int | BlockNumber)

Return type:

Contract

get_currency_key[source]

Get the currency key for a given token.

Since get_currency_key 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:: HexStr | None

get_price[source]

Get the price of a synth in dollars.

Parameters:

token (str | HexBytes | AnyAddress | EthAddress | Contract | int) – The token address to get the price for.
block (int | BlockNumber | None) – The block number to query at. Defaults to the latest block.

Returns:

The price of the synth in USD, or None if the price is stale.

Raises:

Exception – If an unexpected error occurs during the price retrieval.

Return type:

UsdPrice | None

Examples

>>> price = await synthetix.get_price("0x...")
>>> print(price)
1.23

See also

get_currency_key()

is_synth[source]

Check if a token is a synth.

Parameters:: token (str | HexBytes | AnyAddress | EthAddress | Contract | int) – The token address to check.
Returns:: True if the token is a synth, False if not.
Raises:: Exception – If an unexpected error occurs during the check.
Return type:: bool

Examples

>>> is_synth = await synthetix.is_synth("0x...")
>>> print(is_synth)
True

See also

get_currency_key()

synths[source]

Get target addresses of all synths.

Returns:: A list of target addresses for all synths.

Examples

>>> synths = await synthetix.synths
>>> print(synths)
['0x...', '0x...', ...]

y.prices.synthetix.encode_bytes(s)

y.prices.yearn module

class y.prices.yearn.YearnInspiredVault[source]

Bases: ERC20

Represents a vault token from Yearn or a similar protocol.

This class extends ERC20 and provides methods to interact with vaults, including fetching the underlying asset, share price, and token price.

ASyncFunction_decimals(block: int | eth_typing.evm.BlockNumber | NoneType = None) → int

used to fetch decimals at specific block

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

Parameters:: block (int | BlockNumber | None)
Return type:: int

ASyncFunction_scale(block: int | eth_typing.evm.BlockNumber | NoneType = None) → int

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

Parameters:: block (int | BlockNumber | None)
Return type:: int

__eq__(_ContractBase__o)

Return self==value.

Parameters:: _ContractBase__o (object)
Return type:: bool

__init__(self)

Parameters:

address (str | HexBytes | AnyAddress | EthAddress | Contract | int)
asynchronous (bool)
_deploy_block (int | None)

Return type:

None

__str__()

Return the contract address as a string.

Returns:: The contract address as a string.
Return type:: str

__build_name__: HiddenMethodDescriptor[Self, str]

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

The original docstring for get() is shown below:

_ASyncPropertyDescriptorBase.get(self, instance: I, owner: Optional[Type[I]] = None) -> T Asynchronously retrieves 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.

The original docstring for get() is shown below:

_ASyncPropertyDescriptorBase.get(self, instance: I, owner: Optional[Type[I]] = None) -> T Asynchronously retrieves 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] = 'YearnInspiredVault'

__scale__: HiddenMethodDescriptor[Self, int]

Optional[Type[I]] = None) -> T Asynchronously retrieves 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.

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.

The original docstring for get() is shown below:

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

Type:

_ASyncPropertyDescriptorBase.get(self, instance

Type:

I, owner

Parameters:

instance (I)
owner (Type[I] | None)

Return type:

T

__underlying__: HiddenMethodDescriptor[Self, ERC20]

Optional[Type[I]] = None) -> T Asynchronously retrieves 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.

Type:

_ASyncPropertyDescriptorBase.get(self, instance

Type:

I, owner

Parameters:

instance (I)
owner (Type[I] | None)

Return type:

T

address: Address: The contract address of the token.

asynchronous: bool = False

balance_of

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

Parameters:

address (str | HexBytes | AnyAddress | EthAddress | Contract | int) – The address to query.
block (optional) – The block number to query. Defaults to latest block.

Returns:

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

Return type:

int

Examples

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

balance_of_readable

Parameters:

address (str | HexBytes | AnyAddress | EthAddress | Contract | int)
block (int | BlockNumber | None)

Return type:

float

build_name

Get the contract’s build name.

Returns:: The contract’s build name.

Examples

>>> contract = ContractBase("0x1234567890abcdef1234567890abcdef12345678")
>>> await contract.build_name
'MyContract'

property contract: Contract

decimals

The number of decimal places for the token.

Returns:: The number of decimal places for the token.

Examples

>>> token = ERC20("0x1234567890abcdef1234567890abcdef12345678")
>>> await token.decimals
18

deploy_block: ASyncBoundMethod[Self, Any, int]

Get the block number when the contract was deployed.

Parameters:: when_no_history_return_0 (bool) – If True, return 0 when no history is found instead of raising an exception.
Returns:: The block number when the contract was deployed, or 0 if when_no_history_return_0 is True and the deploy block cannot be determined.
Return type:: int

Examples

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

See also

contract_creation_block_async()

has_method

Check if the contract has a specific method.

Parameters:

method (str) – The name of the method to check for.
return_response (bool) – If True, return the response of the method call instead of a boolean.

Returns:

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

Return type:

bool | Any

Examples

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

See also

has_method()

name

The token’s name.

Returns:: The token’s name.

Examples

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

price[source]

Calculates the USD price of the vault token.

Parameters:

block (optional) – The block number to query. Defaults to the latest block.
ignore_pools (Tuple[UniswapV2Pool | CurvePool, ...]) – Pools to ignore when calculating the price.
skip_cache (bool) – Whether to skip the cache when fetching prices.

Returns:

The USD price of the vault token.

Return type:

UsdPrice

Example

>>> vault = YearnInspiredVault("0x5f18C75AbDAe578b483E5F43f12a39cF75b973a9")  # yvUSDC
>>> price = vault.price(block=14_000_000)
>>> print(f"{price:.6f}")
1.096431  # The price of yvUSDC in USD

scale

Get the scaling factor for the token.

Returns:: The scaling factor for the token.

Examples

>>> token = ERC20("0x1234567890abcdef1234567890abcdef12345678")
>>> await token.scale
1000000000000000000

share_price[source]

Calculates the share price of the vault.

Parameters:: block (optional) – The block number to query. Defaults to the latest block.
Returns:: The share price of the vault, or None if the vault’s total supply is zero.
Raises:: CantFetchParam – If the share price cannot be fetched or calculated.
Return type:: Decimal | None

Example

>>> vault = YearnInspiredVault("0x5f18C75AbDAe578b483E5F43f12a39cF75b973a9")  # yvUSDC
>>> share_price = await vault.share_price(block=14_000_000)
>>> print(f"{share_price:.6f}")
1.096431

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

underlying[source]

Fetches the underlying asset of the vault.

Returns:: The underlying ERC20 token.
Raises:: CantFetchParam – If the underlying asset cannot be determined.

Special Cases:

Arbitrum USDL: For the specific address 0x57c7E0D43C05bCe429ce030132Ca40F6FA5839d7 on Arbitrum, it uses the ‘usdl()’ method to fetch the underlying token.
Beefy Vaults: For certain Beefy vaults (BeefyVaultV6Matic and BeefyVenusVaultBNB), it uses ‘wmatic()’ or ‘wbnb()’ methods respectively.
Reaper Vaults: For certain Reaper vaults, it checks for a ‘lendPlatform()’ method and then queries the ‘underlying()’ method on the lend platform contract.

Example

>>> vault = YearnInspiredVault("0x5f18C75AbDAe578b483E5F43f12a39cF75b973a9")  # yvUSDC
>>> underlying = vault.underlying()
>>> underlying.address
"0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48"  # USDC address

ASyncFunctiony.prices.yearn.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]>, ignore_pools: Tuple[Union[ForwardRef('UniswapV2Pool'), ForwardRef('CurvePool')], ...] = ()) → 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:

token (str | HexBytes | AnyAddress | EthAddress | Contract | int)
block (int | BlockNumber | None)
skip_cache (bool)
ignore_pools (Tuple[UniswapV2Pool | CurvePool, ...])

Return type:

UsdPrice

Since is_yearn_vault 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 (str | HexBytes | AnyAddress | EthAddress | Contract | int)
Return type:: bool

y.prices.yearn.share_price_methods = ('pricePerShare()(uint)', 'getPricePerShare()(uint)', 'getPricePerFullShare()(uint)', 'getSharesToUnderlying()(uint)', 'exchangeRate()(uint)'): List of methods which might be used to get the share price of a vault.

y.prices.yearn.underlying_methods = ('token()(address)', 'underlying()(address)', 'native()(address)', 'want()(address)', 'input()(address)', 'asset()(address)', 'wmatic()(address)', 'wbnb()(address)', 'based()(address)'): List of methods which might be used to get the underlying asset of a vault.

y.prices package

Subpackages

Submodules

y.prices.band module

y.prices.chainlink module

y.prices.convex module

y.prices.gearbox module

y.prices.magic module

y.prices.one_to_one module

y.prices.pendle module

y.prices.popsicle module

y.prices.rkp3r module

y.prices.solidex module

y.prices.synthetix module

y.prices.yearn module

Module contents