y.prices package

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

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

has_feed[source]

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

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 elements will be returned.

obj_type: alias of _EventItem

__aiter__()

Returns an async iterator for the wrapped async iterable.

Return type:: AsyncIterator[T]

__await__()

Asynchronously iterates through all contents of Self and returns a list containing the results.

Return type:: Generator[Any, Any, List[T]]

__del__()

Return type:: None

__init__(addresses, topics, asynchronous=True)[source]

Initializes the ASyncIterable with an async iterable.

Parameters:: asynchronous (bool)

__iter__()

Returns an iterator for the wrapped async iterable.

Return type:: Iterator[T]

classmethod __subclasshook__(C)

Abstract classes can override this to customize issubclass().

This is invoked early on by abc.ABCMeta.__subclasscheck__(). It should return True, False or NotImplemented. If it returns NotImplemented, the normal algorithm is used. Otherwise, it overrides the normal algorithm (and the outcome is cached).

_objects_thru(block)

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

events(to_block)

Parameters:: to_block (int)
Return type:: ASyncIterator[_EventItem]

filter(function)

Filters the contents of the async iterable 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 will yield the filtered objects yielded from this instance.
Return type:: ASyncFilter[T]

logs(to_block)

Parameters:: to_block (int | None)
Return type:: ASyncIterator[LogReceipt]

objects(to_block)

Parameters:: to_block (int)
Return type:: ASyncIterator[_EventItem]

sort(*, key=None, reverse=False)

Sorts the contents of the {cls}.

Parameters:

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

Returns:

An instance of ASyncSorter that will yield the objects yielded from this {cls}, but sorted.

Return type:

ASyncSorter[T]

classmethod wrap(wrapped)

Class method to wrap an AsyncIterable for backward compatibility.

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

__async_property__

__orig_bases__ = (y.utils.events.ProcessedEvents[y.prices.chainlink.Feed],)

__parameters__ = ()

__wrapped__: AsyncIterable[T]: The wrapped async iterable object.

addresses

asynchronous

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

property cache: LogCache

property executor: _AsyncExecutorMixin

from_block

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

property is_asleep: bool

is_reusable

property materialized: List[T]: Iterates through all contents of Self and returns a list containing the results.

property semaphore: BlockSemaphore

to_block

topics

y.prices.convex module

ASyncFunctiony.prices.convex.get_price(token_address: 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]>) → 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 (EthAddress)
block (int | BlockNumber | None)
skip_cache (bool)

Return type:

UsdPrice

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

Parameters:: token_address (EthAddress)
Return type:: bool

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__(address, asynchronous=False, _deploy_block=None)

Parameters:

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

Return type:

None

__str__()

Return str(self).

Return type:: str

__build_name__: HiddenMethodDescriptor[Self, str]

Parameters:

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

Return type:

T

__contract__: HiddenMethodDescriptor[Self, Contract]

Parameters:

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

Return type:

T

__diesel_token__: HiddenMethodDescriptor[Self, ERC20]

Parameters:

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

Return type:

T

__underlying__: HiddenMethodDescriptor[Self, ERC20]

Parameters:

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

Return type:

T

address

asynchronous: bool = False

build_name

contract[source]

deploy_block: ASyncBoundMethod[Self, Any, int]

Parameters:: when_no_history_return_0 (bool)
Return type:: int

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

Parameters:

method (str)
return_response (bool)

Return type:

bool | Any

underlying[source]

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

Bases: ASyncGenericBase

__init__(asynchronous=False)[source]

Parameters:: asynchronous (bool)

__registry__

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

Don’t pass an int like 123 into token_address please, that’s just silly. - ypricemagic accepts ints to allow you to pass y.get_price(0x0bc529c00C6401aEF6D220BE8C6Ea1667F6Ad93e)

so you can save yourself some keystrokes while testing in a console

(as opposed to y.get_price(“0x0bc529c00C6401aEF6D220BE8C6Ea1667F6Ad93e”))

When get_price is unable to find a price: - If silent == True, ypricemagic will print an error message using standard python logging - If silent == False, ypricemagic will not log any error - If fail_to_None == True, ypricemagic will return None - If fail_to_None == False, ypricemagic will raise a PriceError

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]: A more optimized way to fetch prices for multiple assets at the same block.

# NOTE silent kwarg is currently disabled. In every case: - if silent == True, tqdm will not be used - if silent == False, tqdm will be used

When get_prices is unable to find a price: - if fail_to_None == True, ypricemagic will return None for that token - if fail_to_None == False, ypricemagic will raise a yPriceMagicError

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.

y.prices.magic.map_prices(token_addresses: Iterable[str | HexBytes | AnyAddress | EthAddress | Contract | int], block: int | BlockNumber, *, fail_to_None: Literal[True], skip_cache: bool = ENVS.SKIP_CACHE, silent: bool = False) → TaskMapping[str | HexBytes | AnyAddress | EthAddress | Contract | int, UsdPrice | None][source]
y.prices.magic.map_prices(token_addresses: Iterable[str | HexBytes | AnyAddress | EthAddress | Contract | int], block: int | BlockNumber, *, fail_to_None: bool = False, skip_cache: bool = ENVS.SKIP_CACHE, silent: bool = False) → TaskMapping[str | HexBytes | AnyAddress | EthAddress | Contract | int, UsdPrice]

y.prices.one_to_one module

ASyncFunctiony.prices.one_to_one.get_price(token_address: 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]>) → 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 (EthAddress)
block (int | BlockNumber | None)
skip_cache (bool)

Return type:

UsdPrice

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

Parameters:: token_address (EthAddress)
Return type:: bool

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

Parameters:: lp_token (str | HexBytes | AnyAddress | EthAddress)
Return type:: Tuple[str, str, str]

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

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

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

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

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

__init__(asynchronous=False)[source]

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

__address_resolver__: HiddenMethodDescriptor[Self, Contract]

Parameters:

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

Return type:

T

__synths__

Parameters:

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

Return type:

T

address_resolver[source]

get_address[source]

Get contract from Synthetix registry.: See also https://docs.synthetix.io/addresses/

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

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:: HexString | None

get_price[source]

Get a price of a synth in dollars.

Parameters:

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

Return type:

UsdPrice | None

is_synth[source]

returns True if a token is a synth, False if not

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

synths[source]: Get target addresses of all synths.

y.prices.synthetix.encode_bytes(s)

y.prices.yearn module

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

Bases: ERC20

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

used to fetch decimals at specific block

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

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

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

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

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

__eq__(_ContractBase__o)

Return self==value.

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

__init__(address, asynchronous=False, _deploy_block=None)

Parameters:

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

Return type:

None

__str__()

Return str(self).

Return type:: str

__build_name__: HiddenMethodDescriptor[Self, str]

Parameters:

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

Return type:

T

__decimals__: HiddenMethodDescriptor[Self, int]

Parameters:

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

Return type:

T

__name__: HiddenMethodDescriptor[Self, str] = 'YearnInspiredVault'

__scale__: HiddenMethodDescriptor[Self, int]

Parameters:

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

Return type:

T

__symbol__: HiddenMethodDescriptor[Self, str]

Parameters:

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

Return type:

T

__underlying__: HiddenMethodDescriptor[Self, ERC20]

Parameters:

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

Return type:

T

address

asynchronous: bool = False

balance_of

Parameters:

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

Return type:

int

balance_of_readable

Parameters:

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

Return type:

float

build_name

property contract: Contract

decimals

deploy_block: ASyncBoundMethod[Self, Any, int]

Parameters:: when_no_history_return_0 (bool)
Return type:: int

has_method

Parameters:

method (str)
return_response (bool)

Return type:

bool | Any

name

price[source]

Parameters:

block (int | BlockNumber | None)
ignore_pools (Tuple[UniswapV2Pool | CurvePool, ...])
skip_cache (bool)

Return type:

UsdPrice

scale

share_price[source]

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

symbol

total_supply

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

total_supply_readable

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

underlying[source]

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