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 [source]
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:
It first attempts to retrieve a cached bucket from the database via
y._db.utils.token.get_bucket()
.Next, it applies simple string-based comparisons defined in the
string_matchers
dictionary.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.
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:
y.prices.utils.sense_check module
- y.prices.utils.sense_check.sense_check()
y.prices.utils.ypriceapi module
- exception y.prices.utils.ypriceapi.BadResponse[source]
Bases:
Exception
Exception raised for bad responses from ypriceAPI.
- y.prices.utils.ypriceapi.announce_beta()[source]
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)[source]
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:
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()[source]
Get the supported chains from ypriceAPI.
Examples
>>> chains = await get_chains() >>> chains[1] 'Ethereum Mainnet'
- async y.prices.utils.ypriceapi.get_price(token, block)[source]
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()[source]
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)[source]
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)