dojo.observations

This module contains the observation objects emitted by the environments.

class GmxV2Observation(dojo.observations.base_observation.BaseObservation):

Contains observation logic for the Gmx agents and policies.

GmxV2Observation( market_venues: list[dojo.models.gmxV2.market.MarketVenue], backend: dojo.network.base_backend.BaseBackend)

Initialize the observation.

Parameters
  • backend: The backend to use.
def total_trader_pnl(self, address: str) -> decimal.Decimal:

Return the total PnL for a trader.

Parameters
  • address: The address of the trader to get the PnL for.
def get_market(self, market_key: str) -> dojo.models.gmxV2.market.Market:

Method to get market from GMX environment.

Parameters
  • market_key: The key identifying the market to retrieve.
def get_account_positions(self, account: str) -> list[dojo.models.gmxV2.position.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.

def get_token_price_by_token_symbol(self, token_symbol: str) -> decimal.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.

def get_token_price_by_token_address( self, token_address: str, price_type: dojo.observations.gmxV2.PriceType = <PriceType.MEAN: 'MEAN'>) -> decimal.Decimal:

Calculate the price of a token by address in USD.

Parameters
  • token_address: The address of the token to calculate the price for.
  • price_type: The type of price to calculate. Default is PriceType.MEAN.
Returns

The calculated price of the token as a Decimal.

def index_token_price(self, market_key: str) -> decimal.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.

Parameters
  • 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.

def index_token_price_min(self, market_key: str) -> decimal.Decimal:

Calculate the minimum price of the index token.

Parameters
  • 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.

def index_token_price_max(self, market_key: str) -> decimal.Decimal:

Calculate the maximum price of the index token.

Parameters
  • 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.

def long_token_price(self, market_key: str) -> decimal.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.

Parameters
  • 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.

def short_token_price(self, market_key: str) -> decimal.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.

Parameters
  • 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.

def get_position( self, position_key: hexbytes.main.HexBytes) -> dojo.models.gmxV2.position.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.

def get_market_token_price_for_traders( self, market_key: str, maximize: bool) -> tuple[decimal.Decimal, dojo.models.gmxV2.market_token.MarketPoolValueInfo]:

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

Parameters
  • market_key: The key identifying the market to retrieve the market token.
  • 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.

def get_market_token_price_for_deposit( self, market_key: str, maximize: bool) -> tuple[decimal.Decimal, dojo.models.gmxV2.market_token.MarketPoolValueInfo]:

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

Parameters
  • market_key: The key identifying the market to retrieve the market token.
  • 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.

def get_market_token_price_for_withdraw( self, market_key: str, maximize: bool) -> tuple[decimal.Decimal, dojo.models.gmxV2.market_token.MarketPoolValueInfo]:

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

Parameters
  • market_key: The key identifying the market to retrieve the market token.
  • 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.

def get_position_pnl_usd( self, position: dojo.models.gmxV2.position.Position) -> dojo.models.gmxV2.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.

def get_net_pnl(self, market_key: str, maximize: bool) -> decimal.Decimal:

Retrieve the net PnL.

Parameters
  • market_key: The key identifying the market to retrieve the net PnL for.
  • maximize: A boolean indicating whether to maximize the PnL.
Returns

The net PnL as a Decimal.

def get_pnl(self, market_key: str, is_long: bool, maximize: bool) -> decimal.Decimal:

Retrieve the PnL for a given position type.

Parameters
  • market_key: The key identifying the market to retrieve the PnL for.
  • 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.

def get_open_interest_with_pnl(self, market_key: str, is_long: bool, maximize: bool) -> decimal.Decimal:

Retrieve the open interest with PnL.

Parameters
  • market_key: The key identifying the market to retrieve the open interest for.
  • 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.

def get_market_info(self, market_key: str) -> dojo.models.gmxV2.market_info.MarketInfo:

Retrieve the market information.

Parameters
  • market_key: The key identifying the market to retrieve the information for.
Returns

A MarketInfo object containing detailed market information.

def get_deposit_amount_out( self, 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.

Parameters
  • market_key: The key identifying the market to retrieve the deposit amount for.
  • long_token_amount: The amount of long tokens to deposit.
  • short_token_amount: The amount of short tokens to deposit.
  • include_virtual_inventory_impact: A boolean indicating whether to include virtual inventory impact.
class UniswapV3Observation(dojo.observations.base_observation.BaseObservation):

Contains observation logic for the UniswapV3 agents and policies.

UniswapV3Observation(pools: list[str], backend: dojo.network.base_backend.BaseBackend)

Initialize the observation.

Parameters
  • pools: list of pools to use in the environment.
  • backend: The backend to use.
def pool_token_addresses(self, pool: str) -> tuple[str, str]:

Get the addresses of the tokens in a pool.

Parameters
  • pool: The pool name.
def pool_tokens(self, pool: str) -> tuple[str, str]:

Get the symbols of the tokens in a pool.

Parameters
  • pool: The pool name.
def tick_spacing(self, pool: str) -> int:

Get the tick spacing of a pool.

Parameters
  • pool: The pool name.
def active_tick_range(self, pool: str) -> tuple[int, int]:

Get the active tick range in a pool.

Parameters
  • pool: The pool name.
def slot0(self, pool: str) -> list[typing.Any]:

Get the contents of the pool's slot0.

Parameters
  • pool: The pool name.
def liquidity(self, pool: str) -> int:

Get the pool liquidity of the active range.

Parameters
  • pool: The pool name. /core/interfaces/pool/IUniswapV3PoolState#liquidity
def ticks( self, pool: str, tick: int) -> tuple[int, int, int, int, int, int, int, bool]:

Returns a tuple of various tick info.

Parameters
  • pool: The pool name.
  • tick: The tick index.

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

def pool_positions( self, pool: str, owner: str, tick_lower: int, tick_upper: int) -> dict[str, typing.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.

def nft_positions(self, token_id: int) -> dict[str, typing.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.

def lp_fees(self, token_ids: list[int]) -> dict[str, decimal.Decimal]:

Get the uncollected LP fees owed to an LP portfolio.

Parameters
  • token_ids: The token IDs of the LP NFTs in the portfolio.
def lp_quantities(self, token_ids: list[int]) -> dict[str, decimal.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.
def lp_portfolio(self, token_ids: list[int]) -> dict[str, decimal.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.
def tokens(self) -> list[str]:

Get all the tokens used in the observed pools.

def market_agent_tokens(self) -> list[str]:

Get all the tokens used in the observed pools by the market agent.

def token_addresses(self) -> list[str]:

Get all the token addresses used in the environment.

def tick_liquidities(self, pool: str, from_tick: int, to_tick: int) -> dict[str, typing.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,).

def pool_fee(self, pool: str) -> decimal.Decimal:

Get the pool's trade fee.

Parameters
  • pool: The name of the pool.
Returns

The pool's trade fee.

def protocol_fees(self, pool: str) -> tuple[decimal.Decimal, decimal.Decimal]:

Get the pool's protocol fees.

Parameters
  • pool: The name of the pool.
Returns

The pool's protocol fees.

def price(self, token: str, unit: str, pool: str) -> decimal.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.

class AAVEv3Observation(dojo.observations.base_observation.BaseObservation):

Contains observation logic for the AAVEv3 agents and policies.

AAVEv3Observation(backend: dojo.network.base_backend.BaseBackend)

Initialize the observation.

Parameters
  • backend: The backend to use.
@lru_cache
def tokens(self) -> list[typing.Any]:

Return all tokens on the Pool.

def market_agent_tokens(self) -> list[typing.Any]:

Return all tokens on the Pool specifically for market agent.

def balance(self, token_name: str, address: str) -> int:

ERC20 token balance.

Includes loands borrowed via AAVE.

Parameters
  • token_name: The token name.
  • address: The address of the account.
def get_asset_price(self, asset_name: str) -> int:

Get the price of an asset on AAVE in the Pools base currency.

Parameters
  • asset_name: ERC20 token symbol.
def get_asset_prices(self, asset_names: list[str]) -> list[int]:

Get the price of an asset on AAVE in the Pools base currency.

Parameters
  • asset_names: ERC20 token symbols.
def get_user_account_data_base(self, agent_address: str) -> dojo.observations.aaveV3.UserAccountData:

Get account data in the pools base currency.

Parameters
  • agent_address: The address of the account.
def get_user_account_data( self, agent_address: str, currency_name: str) -> dojo.observations.aaveV3.UserAccountData:

Get account data in the specified currency.

Parameters
  • agent_address: The address of the account.
  • currency_name: The currency symbol.
def events_last_blocks(self):

Get recently emitted events.

def get_reserve_data(self, reserve_symbol: str) -> dojo.observations.aaveV3.ReserveData:

Get the data of a reserve.

Parameters
  • reserve_symbol: The symbol of the reserve.