dojo.observations package

This module contains the observation objects emitted by the environments.

class dojo.observations.BaseObs(backend: BaseBackend)

Bases: ABC

Base observation class for all environments.

add_signal(name: str, value: float)

TODO.

property signals

Return signals.

token_decimals(token: str) → int

Return the number of decimals of the token.

Parameters:

token – token symbol.

token_symbol(address: str) → str

Return the token symbol.

Parameters:

address – token address.

abstract tokens() → List[str]

Return a list of token symbols in the observation.

class dojo.observations.UniV3Obs(pools: List[str], backend: BaseBackend)

Bases: BaseObs

The UniV3Obs contains observation logic for the agents and policies.

POOL_SIZE = 2
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.

nft_positions(token_id: int) → dict

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

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

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

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.

class dojo.observations.base_observation.BaseObs(backend: BaseBackend)

Bases: ABC

Base observation class for all environments.

add_signal(name: str, value: float)

TODO.

property signals

Return signals.

token_decimals(token: str) → int

Return the number of decimals of the token.

Parameters:

token – token symbol.

token_symbol(address: str) → str

Return the token symbol.

Parameters:

address – token address.

abstract tokens() → List[str]

Return a list of token symbols in the observation.

Observation logic for UniswapV3.

class dojo.observations.uniswapV3.UniV3Obs(pools: List[str], backend: BaseBackend)

Bases: BaseObs

The UniV3Obs contains observation logic for the agents and policies.

POOL_SIZE = 2
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.

nft_positions(token_id: int) → dict

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

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

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

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.

dojo.observations.uniswapV3.batch_get_real_quantities(sqrt_priceX96: int, liquidities: ndarray, lower_ticks: ndarray) → ndarray

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.batch_get_virtual_quantities(real_quantities: ndarray, liquidities: ndarray, lower_ticks: ndarray) → ndarray

Get the virtual quantities of a list of liquidities 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:

  • real_quantities – The real token quantities of the liquidities.

  • liquidities – The liquidities to find the quantities of.

  • lower_ticks – The lower ticks of the tick ranges, should have length of liquidities + 1.

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.

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.