dojo.observations package¶

This module contains the observation objects emitted by the environments.

class dojo.observations.AAVEv3Observation(backend: BaseBackend)¶

Bases: BaseObservation

Contains observation logic for the AAVEv3 agents and policies.

balance(token_name: str, address: str) → int¶

ERC20 token balance.

Includes loands borrowed via AAVE.

events_last_blocks()¶: Get recently emitted events.

get_asset_price(asset_name: str) → int¶: Get the price of an asset on AAVE in the Pools base currency.

get_asset_prices(asset_names: list[str]) → list[int]¶: Get the price of an asset on AAVE in the Pools base currency.

get_reserve_data(reserve_symbol: str) → ReserveData¶: Get the data of a reserve.

get_user_account_data(agent_address: str, currency_name: str) → UserAccountData¶: Get account data in the specified currency.

get_user_account_data_base(agent_address: str) → UserAccountData¶: Get account data in the pools base currency.

market_agent_tokens() → list[Any]¶: Return all tokens on the Pool specifically for market agent.

tokens() → list[Any]¶: Return all tokens on the Pool.

class dojo.observations.GmxV2Observation(market_venues: list[MarketVenue], backend: BaseBackend)¶

Bases: BaseObservation

Contains observation logic for the Gmx agents and policies.

get_account_positions(account: str) → list[Position]¶

Get the positions of a given account.

Parameters:: account – The account address to get positions for.
Returns:: A list of Position objects representing the account’s positions.

get_deposit_amount_out(market_key: str, long_token_amount: int, short_token_amount: int, include_virtual_inventory_impact: bool) → Any¶: Get minimum number of GM tokens when depositing.

get_market(market_key: str) → Market¶: Method to get market from GMX environment.

get_market_info(market_key: str) → MarketInfo¶

Retrieve the market information.

Returns:: A MarketInfo object containing detailed market information.

get_market_token_price_for_deposit(market_key: str, maximize: bool) → tuple[Decimal, MarketPoolValueInfo]¶

Retrieve the market token price and market pool value info for deposits.

Parameters:: maximize – A boolean indicating whether to maximize the price.
Returns:: A tuple containing the calculated GM token value as a Decimal and a MarketPoolValueInfo object with detailed market pool value information.

get_market_token_price_for_traders(market_key: str, maximize: bool) → tuple[Decimal, MarketPoolValueInfo]¶

Retrieve the market token price and market pool value info for traders.

Parameters:: maximize – A boolean indicating whether to maximize the price.
Returns:: A tuple containing the calculated GM token value as a Decimal and a MarketPoolValueInfo object with detailed market pool value information.

get_market_token_price_for_withdraw(market_key: str, maximize: bool) → tuple[Decimal, MarketPoolValueInfo]¶

Retrieve the market token price and market pool value info for withdrawals.

Parameters:: maximize – A boolean indicating whether to maximize the price.
Returns:: A tuple containing the calculated GM token value as a Decimal and a MarketPoolValueInfo object with detailed market pool value information.

get_net_pnl(market_key: str, maximize: bool) → Decimal¶

Retrieve the net PnL.

Parameters:: maximize – A boolean indicating whether to maximize the PnL.
Returns:: The net PnL as a Decimal.

get_open_interest_with_pnl(market_key: str, is_long: bool, maximize: bool) → Decimal¶

Retrieve the open interest with PnL.

Parameters:

is_long – A boolean indicating whether the position is long.
maximize – A boolean indicating whether to maximize the PnL.

Returns:

The open interest with PnL as a Decimal.

get_pnl(market_key: str, is_long: bool, maximize: bool) → Decimal¶

Retrieve the PnL for a given position type.

Parameters:

is_long – A boolean indicating whether the position is long.
maximize – A boolean indicating whether to maximize the PnL.

Returns:

The PnL as a Decimal.

get_position(position_key: HexBytes) → Position¶

Retrieve a position based on the given position key.

This method interacts with the backend to call the getPosition function from the Reader contract, using the provided position key. The data retrieved is then used to create and return a Position object.

Parameters:: position_key – The key identifying the position to retrieve.
Returns:: A Position object representing the retrieved position.

get_position_pnl_usd(position: Position) → PositionPnl¶

Retrieve the position PnL in USD.

Parameters:: position – The position object containing details of the position.
Returns:: A PositionPnl object containing the PnL details in USD.

get_token_price_by_token_address(token_address: str, price_type: PriceType = PriceType.MEAN) → Decimal¶

Calculate the price of a token by address in USD.

Parameters:: token_address – The address of the token to calculate the price for.
Returns:: The calculated price of the token as a Decimal.

get_token_price_by_token_symbol(token_symbol: str) → Decimal¶

Calculate the price of a token by symbol in USD.

Parameters:: token_symbol – The symbol of the token to calculate the price for.
Returns:: The calculated price of the token as a Decimal.

index_token_price(market_key: str) → Decimal¶

Calculate the price of the index token.

This method retrieves the market information and oracle prices for the index token, then calculates the average price using the minimum and maximum prices provided by the oracle. The price is scaled according to the token’s decimals and the oracle precision.

Param:: market_key: The key identifying the market to retrieve the index token price for. In the form of index_token:long_token:short_token, Example: WETH:WETH:USDC.
Returns:: The calculated price of the index token as a Decimal.

index_token_price_max(market_key: str) → Decimal¶

Calculate the maximum price of the index token.

Param:: market_key: The key identifying the market to retrieve the index token price for. In the form of index_token:long_token:short_token, Example: WETH:WETH:USDC.
Returns:: The calculated price of the index token as a Decimal.

index_token_price_min(market_key: str) → Decimal¶

Calculate the minimum price of the index token.

Param:: market_key: The key identifying the market to retrieve the index token price for. In the form of index_token:long_token:short_token, Example: WETH:WETH:USDC.
Returns:: The calculated price of the index token as a Decimal.

long_token_price(market_key: str) → Decimal¶

Calculate the price of the long token.

This method retrieves the market information and oracle prices for the long token, then calculates the average price using the minimum and maximum prices provided by the oracle. The price is scaled according to the token’s decimals and the oracle precision.

Param:: market_key: The key identifying the market to retrieve the index token price for. In the form of index_token:long_token:short_token, Example: WETH:WETH:USDC.
Returns:: The calculated price of the long token as a Decimal.

market_agent_tokens() → list[str]¶: Return all tokens for all markets.

set_market_dict(market_dict: dict[str, Market]) → None¶: Method to set market dict (market_key -> Market) from GMX environment.

set_oracle_prices(oracle_prices: defaultdict[int, dict[str, Any]]) → None¶: Method to set prices from GMX environment.

short_token_price(market_key: str) → Decimal¶

Calculate the price of the short token.

This method retrieves the market information and oracle prices for the short token, then calculates the average price using the minimum and maximum prices provided by the oracle. The price is scaled according to the token’s decimals and the oracle precision.

Param:: market_key: The key identifying the market to retrieve the index token price for. In the form of index_token:long_token:short_token, Example: WETH:WETH:USDC.
Returns:: The calculated price of the short token as a Decimal.

tokens() → list[str]¶: Return all tokens for all markets.

total_trader_pnl(address: str) → Decimal¶: Return the total PnL for a trader.

class dojo.observations.UniswapV3Observation(pools: list[str], backend: BaseBackend)¶

Bases: BaseObservation

Contains observation logic for the UniswapV3 agents and policies.

active_tick_range(pool: str) → tuple[int, int]¶

Get the active tick range in a pool.

Parameters:: pool – The pool name.

liquidity(pool: str) → int¶

Get the pool liquidity of the active range.

Ref https://docs.uniswap.org/contracts/v3/reference/core/interfaces/pool/IUniswapV3PoolState#liquidity

lp_fees(token_ids: list[int]) → dict[str, Decimal]¶

Get the uncollected LP fees owed to an LP portfolio.

Parameters:: token_ids – The token IDs of the LP NFTs in the portfolio.

lp_portfolio(token_ids: list[int]) → dict[str, Decimal]¶

Get the token quantities and uncollected fees invested in an LP portfolio.

Parameters:: token_ids – The token IDs of the LP NFTs in the portfolio.

lp_quantities(token_ids: list[int]) → dict[str, Decimal]¶

Get the total real token quantities invested in an LP portfolio.

Parameters:: token_ids – The token IDs of the LP NFTs in the portfolio.

market_agent_tokens() → list[str]¶: Get all the tokens used in the observed pools by the market agent.

nft_positions(token_id: int) → dict[str, Any]¶

Get LP position data of an LP NFT.

Parameters:: token_id – The token ID of the LP NFT.
Returns:: A dictionary containing the LP position data, with keys: - token0: The symbol of the token0 of the pool. - token1: The symbol of the token1 of the pool. - fee: The fee of the pool. - pool: The name of the pool. - tick_range: The tick range of the LP position. - liquidity: The liquidity of the LP position. - real_quantities: The real token quantities of the LP position. - uncollected_fees: The uncollected fees of the LP position.

pool_fee(pool: str) → Decimal¶

Get the pool’s trade fee.

Parameters:: pool – The name of the pool.
Returns:: The pool’s trade fee.

pool_positions(pool: str, owner: str, tick_lower: int, tick_upper: int) → dict[str, Any]¶

Get the pool position of an owner in the specified tick range.

Parameters:

pool – The pool name.
owner – The owner address.
tick_lower – The lower tick of the position.
tick_upper – The upper tick of the position.

Returns:

A dictionary containing the position data, with keys: - token0: The symbol of the token0 of the pool. - token1: The symbol of the token1 of the pool. - pool: The name of the pool. - owner: The address of the owner. - tick_range: The tick range of the LP position. - liquidity: The liquidity of the position. - uncollected_fees: The uncollected fees of the position. - uncollected_fees_adjusted: The uncollected fees of the position adjusted by the token decimals.

pool_token_addresses(pool: str) → tuple[str, str]¶

Get the addresses of the tokens in a pool.

Parameters:: pool – The pool name.

pool_tokens(pool: str) → tuple[str, str]¶

Get the symbols of the tokens in a pool.

Parameters:: pool – The pool name.

price(token: str, unit: str, pool: str) → Decimal¶

Get the price of a token in a pool in the units of another token.

Parameters:

token – token symbol.
unit – unit token symbol.
pool – pool name.

Returns:

current price of the token in the pool.

protocol_fees(pool: str) → tuple[Decimal, Decimal]¶

Get the pool’s protocol fees.

Parameters:: pool – The name of the pool.
Returns:: The pool’s protocol fees.

slot0(pool: str) → list[Any]¶

Get the contents of the pool’s slot0.

Ref https://docs.uniswap.org/contracts/v3/reference/core/interfaces/pool/IUniswapV3PoolState#slot0

tick_liquidities(pool: str, from_tick: int, to_tick: int) → dict[str, Any]¶

Get the pool liquidities related information across ticks.

Parameters:

pool – The name of the pool.
from_tick – The bottom tick of the range (included).
to_tick – The top tick of the range (included).

Returns:

A dict containing: - current_tick_idx: The index of the currently active tick in the returned arrays (int). - real_quantities: The tick indexed real token quantities of the pool (num_ticks, 2). - liquidities: The tick indexed liquidities of the pool (num_ticks,). - lower_ticks: The tick indexed lower ticks of the pool (num_ticks,).

tick_spacing(pool: str) → int¶

Get the tick spacing of a pool.

Parameters:: pool – The pool name.

ticks(pool: str, tick: int) → tuple[int, int, int, int, int, int, int, bool]¶

Returns a tuple of various tick info.

The return format is: int128 liquidityNet uint256 feeGrowthOutside0X128 uint256 feeGrowthOutside1X128 int56 tickCumulativeOutside uint160 secondsPerLiquidityOutsideX128 uint32 secondsOutside bool initialized

Ref https://docs.uniswap.org/contracts/v3/reference/core/interfaces/pool/IUniswapV3PoolState#ticks # noqa: E501

token_addresses() → list[str]¶: Get all the token addresses used in the environment.

tokens() → list[str]¶: Get all the tokens used in the observed pools.

Submodules¶

Base observation class for all environments.

Observation logic for UniswapV3.

class dojo.observations.uniswapV3.UniswapV3Observation(pools: list[str], backend: BaseBackend)¶

Bases: BaseObservation

Contains observation logic for the UniswapV3 agents and policies.

active_tick_range(pool: str) → tuple[int, int]¶

Get the active tick range in a pool.

Parameters:: pool – The pool name.

liquidity(pool: str) → int¶

Get the pool liquidity of the active range.

Ref https://docs.uniswap.org/contracts/v3/reference/core/interfaces/pool/IUniswapV3PoolState#liquidity

lp_fees(token_ids: list[int]) → dict[str, Decimal]¶

Get the uncollected LP fees owed to an LP portfolio.

Parameters:: token_ids – The token IDs of the LP NFTs in the portfolio.

lp_portfolio(token_ids: list[int]) → dict[str, Decimal]¶

Get the token quantities and uncollected fees invested in an LP portfolio.

Parameters:: token_ids – The token IDs of the LP NFTs in the portfolio.

lp_quantities(token_ids: list[int]) → dict[str, Decimal]¶

Get the total real token quantities invested in an LP portfolio.

Parameters:: token_ids – The token IDs of the LP NFTs in the portfolio.

market_agent_tokens() → list[str]¶: Get all the tokens used in the observed pools by the market agent.

nft_positions(token_id: int) → dict[str, Any]¶

Get LP position data of an LP NFT.

Parameters:: token_id – The token ID of the LP NFT.
Returns:: A dictionary containing the LP position data, with keys: - token0: The symbol of the token0 of the pool. - token1: The symbol of the token1 of the pool. - fee: The fee of the pool. - pool: The name of the pool. - tick_range: The tick range of the LP position. - liquidity: The liquidity of the LP position. - real_quantities: The real token quantities of the LP position. - uncollected_fees: The uncollected fees of the LP position.

pool_fee(pool: str) → Decimal¶

Get the pool’s trade fee.

Parameters:: pool – The name of the pool.
Returns:: The pool’s trade fee.

pool_positions(pool: str, owner: str, tick_lower: int, tick_upper: int) → dict[str, Any]¶

Get the pool position of an owner in the specified tick range.

Parameters:

pool – The pool name.
owner – The owner address.
tick_lower – The lower tick of the position.
tick_upper – The upper tick of the position.

Returns:

pool_token_addresses(pool: str) → tuple[str, str]¶

Get the addresses of the tokens in a pool.

Parameters:: pool – The pool name.

pool_tokens(pool: str) → tuple[str, str]¶

Get the symbols of the tokens in a pool.

Parameters:: pool – The pool name.

price(token: str, unit: str, pool: str) → Decimal¶

Get the price of a token in a pool in the units of another token.

Parameters:

token – token symbol.
unit – unit token symbol.
pool – pool name.

Returns:

current price of the token in the pool.

protocol_fees(pool: str) → tuple[Decimal, Decimal]¶

Get the pool’s protocol fees.

Parameters:: pool – The name of the pool.
Returns:: The pool’s protocol fees.

slot0(pool: str) → list[Any]¶

Get the contents of the pool’s slot0.

Ref https://docs.uniswap.org/contracts/v3/reference/core/interfaces/pool/IUniswapV3PoolState#slot0

tick_liquidities(pool: str, from_tick: int, to_tick: int) → dict[str, Any]¶

Get the pool liquidities related information across ticks.

Parameters:

pool – The name of the pool.
from_tick – The bottom tick of the range (included).
to_tick – The top tick of the range (included).

Returns:

tick_spacing(pool: str) → int¶

Get the tick spacing of a pool.

Parameters:: pool – The pool name.

ticks(pool: str, tick: int) → tuple[int, int, int, int, int, int, int, bool]¶

Returns a tuple of various tick info.

Ref https://docs.uniswap.org/contracts/v3/reference/core/interfaces/pool/IUniswapV3PoolState#ticks # noqa: E501

token_addresses() → list[str]¶: Get all the token addresses used in the environment.

tokens() → list[str]¶: Get all the tokens used in the observed pools.

dojo.observations.uniswapV3.batch_get_real_quantities(sqrt_priceX96: int, liquidities: list[int], lower_ticks: list[int]) → list[tuple[int, int]]¶

Returns real token quantities liquidities in the pool in their tick ranges.

Each liquidity is indexed by the lower tick of its range. For example: - lower_ticks = [0, 2, 4] - liquidities = [100, 200]

The first liquidity is in the range [0, 2] and the second liquidity is in the range [2, 4].

Parameters:

sqrt_priceX96 – The square root of the pool price left shifted by 96 bits.
lower_ticks – The lower ticks of the tick ranges, should have length of liquidities + 1.
liquidities – The liquidities to find the quantities of.

dojo.observations.uniswapV3.get_liquidity(sqrt_priceX96: int, real_quantities: tuple[int, int], tick_lower: int, tick_upper: int) → int¶

Get the liquidity of a position in a pool.

https://github.com/Uniswap/v3-periphery/blob/main/contracts/libraries/LiquidityAmounts.sol#L56

Parameters:

real_quantities – The real token quantities of the position.
tick_lower – The lower tick of the position.
tick_upper – The upper tick of the position.

dojo.observations.uniswapV3.get_liquidity0(lower_sqrt_priceX96: int, uppper_sqrt_priceX96: int, real_quantity0: int) → int¶

Get liqudity given real quantities of asset 0.

https://github.com/Uniswap/v3-periphery/blob/main/contracts/libraries/LiquidityAmounts.sol#L23

Parameters:

lower_sqrt_priceX96 – The lower sqrt price of the position.
uppper_sqrt_priceX96 – The upper sqrt price of the position.
real_quantity0 – The real quantity of asset 0.

dojo.observations.uniswapV3.get_liquidity1(lower_sqrt_priceX96: int, uppper_sqrt_priceX96: int, real_quantity1: int) → int¶

Get liqudity given real quantities of asset 1.

https://github.com/Uniswap/v3-periphery/blob/main/contracts/libraries/LiquidityAmounts.sol#L39

Parameters:

lower_sqrt_priceX96 – The lower sqrt price of the position.
uppper_sqrt_priceX96 – The upper sqrt price of the position.
real_quantity1 – The real quantity of asset 1.

dojo.observations.uniswapV3.get_pool_fee(backend: BaseBackend, pool: str) → int¶

Get the pool’s trade fee.

Runs the fee() function of the pool contract.

dojo.observations.uniswapV3.get_quantity0_delta(lower_sqrt_priceX96: int, upper_sqrt_priceX96: int, liquidity_change: int) → int¶: GetAmount0Delta as in UniswapV3Pool.

dojo.observations.uniswapV3.get_quantity1_delta(lower_sqrt_priceX96: int, upper_sqrt_priceX96: int, liquidity_change: int) → int¶: GetAmount1Delta as in UniswapV3Pool.

dojo.observations.uniswapV3.get_real_quantities(sqrt_priceX96: int, liquidity: int, tick_lower: int, tick_upper: int, round_up: bool = True) → tuple[int, int]¶

Get the real token quantities of a liquidity in the pool in its tick range.

Parameters:

sqrt_priceX96 – The square root of the pool price left shifted by 96 bits.
tick_lower – The lower tick index of the tick range.
tick_upper – The upper tick index of the tick range.
liquidity – The liquidity to find the quantities of.

dojo.observations.uniswapV3.get_real_quantity0(liquidity: int, lower_sqrt_priceX96: int, upper_sqrt_priceX96: int, round_up: bool) → int¶

Get the real quantity of token 0 in a liquidity position.

Parameters:

liquidity – The liquidity of the position.
lower_sqrt_priceX96 – The sqrt_priceX96 of the lower tick of the position.
upper_sqrt_priceX96 – The sqrt_priceX96 of the upper tick of the position.
round_up – Whether to round up the result.

dojo.observations.uniswapV3.get_real_quantity1(liquidity: int, lower_sqrt_priceX96: int, upper_sqrt_priceX96: int, round_up: bool) → int¶

Get the real quantity of token 1 in a liquidity position.

Parameters:

liquidity – The liquidity of the position.
lower_sqrt_priceX96 – The sqrt_priceX96 of the lower tick of the position.
upper_sqrt_priceX96 – The sqrt_priceX96 of the upper tick of the position.
round_up – Whether to round up the result.

dojo.observations.uniswapV3.get_virtual_quantities(real_quantities: tuple[int, int], liquidity: int, lower_tick: int, upper_tick: int) → tuple[int, int]¶

Get the virtual quantities of a liquidity position bounded by a tick range.

Parameters:

real_quantities – The real token quantities of the liquidity position.
liquidity – The liquidity of the position.
lower_tick – The lower tick of the tick range.
upper_tick – The upper tick of the tick range.

dojo.observations.uniswapV3.price_to_active_tick(price: Decimal, tick_spacing: int, decimals: tuple[int, int]) → int¶

Convert a price to the nearest active tick.

Parameters:

price – The price.
decimals – The decimals of the token pair.

Returns:

The nearest active tick.

dojo.observations.uniswapV3.price_to_sqrt_priceX96(price: Decimal) → int¶

Convert a price to a sqrt price shifted by 96 bits.

Parameters:: price – The price.

dojo.observations.uniswapV3.price_to_tick(price: Decimal, decimals: tuple[int, int]) → int¶

Convert a price to a tick as log_1.0001(price).

Parameters:

price – The price.
decimals – The decimals of the token pair.

Returns:

The tick.

dojo.observations.uniswapV3.price_to_tick_range(price: Decimal, tick_spacing: int, decimals: tuple[int, int]) → tuple[int, int]¶

Get the unit tick range of a price.

Parameters:

price – The price.
tick_spacing – The pool’s tick spacing.
decimals – The decimals of the token pair.

Returns:

The tick range.

dojo.observations.uniswapV3.price_wrt_range(sqrt_priceX96: int, tick_lower: int, tick_upper: int) → int¶

Check whether a price is below, within, or above a tick range.

Parameters:

sqrt_priceX96 – The square root of the pool price.
tick_lower – The lower tick of the range.
tick_upper – The upper tick of the range.

Returns:

0 if the price is within the range, 1 if it is below, 2 if it is above.

Raises:

Exception – never

dojo.observations.uniswapV3.sqrt_priceX96_to_price(sqrt_priceX96: int, decimals: tuple[int, int]) → Decimal¶

Convert a sqrt price shifted by 96 bits to a human-readable price.

Parameters:

sqrt_priceX96 – sqrt price shifted by 96 bits.
decimals – The decimals of the token pair.

dojo.observations.uniswapV3.sqrt_priceX96_to_tick(sqrt_priceX96: int) → int¶

Convert a sqrt price shifted by 96 bits to a tick.

Effectively implements log_1.0001(sqrt_priceX96 ** 2 / 2 ** 192)

Parameters:: sqrt_priceX96 – sqrt price shifted by 96 bits.

dojo.observations.uniswapV3.sqrt_priceX96_to_tick_range(sqrt_priceX96: int, tick_spacing: int) → tuple[int, int]¶

Get the unit tick range of a sqrt price.

Parameters:

sqrt_priceX96 – The sqrt price.
tick_spacing – The pool’s tick spacing.

Returns:

The tick range.

dojo.observations.uniswapV3.tick_to_sqrt_priceX96(tick: int) → int¶

Convert a tick to a sqrt price shifted by 96 bits.

Effectively implements sqrt(1.0001^tick) * 2^96.

Parameters:: tick – The tick to get the sqrt price of.
Returns:: The sqrt price of the tick shifted by 96 bits.