dojo.observations

This module contains the observation objects emitted by the environments.

class BaseObservation(abc.ABC):

Base observation class for all environments.

BaseObservation(backend: dojo.network.BaseBackend)

Initialize the observation.

signals: dict[str, float]

Return signals.

def add_signal(self, name: str, value: Union[float, int, decimal.Decimal]) -> None:

Add a singal.

It will be logged and displayed in the dashboard.

def add_block_fees(self, fees: list[typing.Optional[int]]) -> None:

Add the gas prices in a block to the internal fees datastructure.

@abstractmethod
def tokens(self) -> list[str]:

Return a list of token symbols in the observation.

def token_decimals(self, token: str) -> int:

Return the number of decimals of the token.

Parameters
  • token: token symbol.
def token_symbol(self, address: str) -> str:

Return the token symbol.

Parameters
  • address: token address.
def fee_history(self, block_count: int = 1) -> list[list[typing.Optional[int]]]:

Return the fee history of the chain.

Parameters
  • block_count: number of blocks to return.
class GmxV2Observation(dojo.observations.BaseObservation):

Contains observation logic for the Gmx agents and policies.

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

Initialize the observation.

Parameters
  • backend: The backend to use.
def tokens(self) -> list[str]:

Return all tokens for provided markets.

def set_oracle_prices( self, oracle_prices: collections.defaultdict[int, dict[str, typing.Any]]) -> None:

Method to set prices from GMX environment.

def set_market_dict(self, market_dict: dict[str, dojo.models.gmxV2.market.Market]) -> None:

Method to set market dict (market_key -> Market) from GMX environment.

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 index_token_price(self) -> 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.

Returns

The calculated price of the index token as a Decimal.

def long_token_price(self) -> 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.

Returns

The calculated price of the long token as a Decimal.

def short_token_price(self) -> 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.

Returns

The calculated price of the short token as a Decimal.

def get_position(self, position_key: str) -> 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, maximize: bool) -> tuple[decimal.Decimal, dojo.models.gmxV2.market_token.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.

def get_market_token_price_for_deposit( self, maximize: bool) -> tuple[decimal.Decimal, dojo.models.gmxV2.market_token.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.

def get_market_token_price_for_withdraw( self, maximize: bool) -> tuple[decimal.Decimal, dojo.models.gmxV2.market_token.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.

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, maximize: bool) -> decimal.Decimal:

Retrieve the net PnL.

Parameters
  • maximize: A boolean indicating whether to maximize the PnL.
Returns

The net PnL as a Decimal.

def get_pnl(self, is_long: bool, maximize: bool) -> decimal.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.

def get_open_interest_with_pnl(self, is_long: bool, maximize: bool) -> decimal.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.

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

Retrieve the market information.

Returns

A MarketInfo object containing detailed market information.

class UniswapV3Observation(dojo.observations.BaseObservation):

Contains observation logic for the UniswapV3 agents and policies.

UniswapV3Observation(pools: list[str], backend: dojo.network.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.

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

def liquidity(self, pool: str) -> int:

Get the pool liquidity of the active range.

Ref https://docs.uniswap.org/contracts/v3/reference/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.

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_total_potential_tokens_on_withdrawal(self, token_ids: list[int]) -> dict[str, decimal.Decimal]:

Get the total token that would be received on withdrawing some positions.

This is equivalent to lp_fees()+lp_quantities() for each position.

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 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.BaseObservation):

Contains observation logic for the AAVEv3 agents and policies.

AAVEv3Observation(backend: dojo.network.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 balance(self, token_name: str, address: str) -> int:

ERC20 token balance.

Includes loands borrowed via AAVE.

def get_asset_price(self, asset_name: str) -> int:

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

def get_asset_prices(self, asset_names: list[str]) -> list[int]:

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

def get_user_account_data_base(self, agent_address: str) -> dojo.observations.aaveV3.UserAccountData:

Get account data in the pools base currency.

def get_user_account_data( self, agent_address: str, currency_name: str) -> dojo.observations.aaveV3.UserAccountData:

Get account data in the specified currency.

def events_last_blocks(self):

Get recently emitted events.