y.prices.utils package

Submodules

y.prices.utils.buckets module

ASyncFunctiony.prices.utils.buckets.check_bucket(token: str | hexbytes.main.HexBytes | AnyAddress | evmspec.data._main.Address | brownie.convert.datatypes.EthAddress | brownie.network.contract.Contract | int) → str

Determine and return the category or “bucket” of a given token.

This function classifies a token by performing a set of tests in the following order:

1. It first attempts to retrieve a cached bucket from the database via y._db.utils.token.get_bucket().

2. Next, it applies simple string-based comparisons defined in the string_matchers dictionary.

3. If no bucket is determined, it concurrently executes a set of asynchronous “calls-only” tests for further classification. The function will return immediately once any of these tests confirms the token’s membership in a bucket.

4. If none of the concurrent tests succeed, it falls back to a series of sequential asynchronous checks that involve both contract initializations and blockchain calls.

Note

The fallback sequential tests are executed in order. However, the first two fallback checks (for “solidex” and “uni or uni-like lp”) are evaluated independently; as a result, if both conditions return true, the latter bucket (“uni or uni-like lp”) will overwrite the earlier assignment from “solidex”.

The final determined bucket is stored in the database using db.set_bucket().

Args:

token: The token address to classify.

Examples:

Basic usage with a stable USD token:

>>> bucket = check_bucket("0x6B175474E89094C44Da98b954EedeAC495271d0F")
>>> print(bucket)
stable usd

Example triggering a concurrent asynchronous test:

>>> bucket = check_bucket("0xSomeLPTokenAddress")
>>> print(bucket)
popsicle

Example where multiple fallback conditions are met (note that if both “solidex” and “uni or uni-like lp” tests pass, the latter overwrites the former):

>>> bucket = check_bucket("0xAnotherTokenAddress")
>>> print(bucket)
uni or uni-like lp

See Also:

y.convert.to_address_async() y.utils.logging.get_price_logger() _check_bucket_helper() (Helper used for asynchronous bucket checks)

Since check_bucket 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 | Address | EthAddress | Contract | int)

Return type:

str

y.prices.utils.debug module

y.prices.utils.debug.debug_price(token, block=None)

Debug token price retrieval.

Parameters:

• token (str) – The address of the token to debug.

• block (Optional[int]) – The block number at which to retrieve the token price. If not specified, uses the current block height.

Returns:

The price of the token at the specified block.

Return type:

float

Raises:

ValueError – If token is not provided.

y.prices.utils.sense_check module

y.prices.utils.sense_check.sense_check()

y.prices.utils.ypriceapi module

exception y.prices.utils.ypriceapi.BadResponse

Bases: Exception

Exception raised for bad responses from ypriceAPI.

y.prices.utils.ypriceapi.announce_beta()

Announce the beta version of ypriceAPI.

This function logs the beta announcement message, which includes information about the benefits of using ypriceAPI and testimonials from users.

Examples

>>> announce_beta()
ypriceAPI is now in beta!
...

Return type:

None

y.prices.utils.ypriceapi.chain_supported(chainid)

Check if a chain is supported by ypriceAPI.

Parameters:

chainid (int) – The chain ID to check.

Returns:

True if the chain is supported, False otherwise.

Return type:

bool

Examples

>>> await chain_supported(1)
True
>>> await chain_supported(9999)
False

See also

get_chains() for retrieving the list of supported chains.

y.prices.utils.ypriceapi.get_chains()

Get the supported chains from ypriceAPI.

Returns:

A dictionary mapping chain IDs to chain names.

Return type:

Dict[int, str]

Examples

>>> chains = await get_chains()
>>> chains[1]
'Ethereum Mainnet'

async y.prices.utils.ypriceapi.get_price(token, block)

Get the price of a token from ypriceAPI.

Parameters:

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

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

Returns:

The price of the token in USD as an instance of UsdPrice, or None if the price cannot be retrieved.

Return type:

UsdPrice | None

This function will return None if the authorization headers are not present or if the current time is less than the resume time.

Examples

>>> price = await get_price("0xTokenAddress", 12345678)
>>> print(price)
$1234.56780000

Raises:

• ClientConnectorSSLError – If there is an SSL error during the request.

• asyncio.TimeoutError – If the request times out.

• ClientError – If there is a client error during the request.

Parameters:

• token (str | HexBytes | AnyAddress | Address | EthAddress)

• block (int | BlockNumber | None)

Return type:

UsdPrice | None

See also

UsdPrice for the price representation.

y.prices.utils.ypriceapi.get_retry_header(x)

y.prices.utils.ypriceapi.get_session()

Get an aiohttp ClientSession for ypriceAPI.

This session is configured with the ypriceAPI URL, headers, and timeout.

Returns:

An aiohttp ClientSession object.

Return type:

ClientSession

Examples

>>> session = await get_session()

async y.prices.utils.ypriceapi.read_response(response, token=None, block=None)

Read and process the response from ypriceAPI.

Parameters:

• response (ClientResponse) – The aiohttp ClientResponse object.

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

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

Returns:

The parsed response data, or None if the response is not valid.

Raises:

BadResponse – If the response content type is invalid.

Return type:

Any | None

Examples

>>> response = await session.get("/get_price/1/0xTokenAddress?block=12345678")
>>> data = await read_response(response)

Module contents