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

index_token_to_market_venue: dict[str, dojo.models.gmxV2.market.MarketVenue]

Return a mapping from index token to market venue.

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.