dojo.environments

Environments package contains environment classes and logic.

class BaseEnvironment(abc.ABC, typing.Generic[~Action, ~Observation]):

The BaseEnvironment class defines the template for all environments.

BaseEnvironment( chain: dojo.common.constants.Chain, date_range: tuple[datetime.datetime, datetime.datetime], agents: list[dojo.agents.BaseAgent], backend_type: Literal['forked', 'local', 'live'] = 'forked', port: Optional[int] = None, token_data: Optional[dict[str, dict[datetime.datetime, decimal.Decimal]]] = None)

Initialize the environment.

Parameters

date_range: date range of the environment simulation.
agents: list of agents to use in the environment.
backend_type: type of backend to use, one of: ['forked', 'local', 'live'].
port: port to connect to the local chain. Defaults to find a free port.
token_data: optional dict containing token data to pass to each agent, needed if you want to call the agent wealth methods. The dictionary has format:
```
{
    "token_symbol": {
        datetime0: price0,
        datetime1: price1,
        ...
    },
    ...
}
```

Raises

ValueError: If backend_type not implemented or recognized

chain: dojo.common.constants.Chain

Get the blockchain for this environment.

def deploy_forked_contract(self, env_name: str, contract_name: str) -> None:

Sets up the contract on the forked network.

Parameters

env_name (str): Name of the environment, e.g. UniswapV3, AaveV3.
contract_name (str): Name of the contract.

def deploy_local_contract( self, protocol: str, name: str, args: list[typing.Any] = [], bytecode: Optional[str] = None) -> web3.contract.contract.Contract:

Sets up the contract on the local network.

Parameters

env_name (str): Name of the environment, e.g. UniswapV3, AaveV3.
contract_name (str): Name of the contract.
args (list, optional): Arguments to be passed to the constructor of the contract. Defaults to [].
method (str, optional): method to use for contract deployment:
- "standard": use the contract constructor
- "alternate": use the contract creation code. Defaults to "standard".

@abstractmethod

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

Returns

a list of contracts to approve for ERC20 transfers.

@abstractmethod

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

Deploy and register contracts on the local network.

Returns

a list of contracts to approve for ERC20 transfers.

@abstractmethod

def initialize_state(self) -> None:

Initialize the state of the environment.

def reset(self) -> ~Observation:

Reset the environment and agents before returning the observation object.

This method needs to be called before any simulation in order to setup the on- chain infrastructure.

def iter_block( self, *, batch_size: int = 0, status_bar: bool = False) -> Generator[int, NoneType, NoneType]:

Iterate over the simulation period one block at a time.

This also sets the date and block attributes of the environment.

Parameters

batch_size: parameter for ethereumetl.export_blocks_and_transactions see https://ethereum-etl.readthedocs.io/en/latest/commands/. If set to 0, will NOT get the date information for each block. :yield: block number.

Raises

Exception: batched no longer supported

def market_actions(self, agents_actions: list[~Action]) -> list[~Action]:

Predict the market actions of the current block.

Parameters

agents_actions: The actions from the policies of the other agents.

def step( self, actions: list[~Action]) -> tuple[~Observation, numpy.ndarray[typing.Any, numpy.dtype[numpy.generic]], numpy.ndarray[typing.Any, numpy.dtype[numpy.float64]], dict[int, dict[typing.Any, typing.Any]]]:

Execute a sequence of actions in the environment in list order.

Parameters

actions: in order sequence of actions to execute.

Returns

next observation.

Returns

array of rewards for each agent.

Returns

array of done flags for each agent.

Returns

dict of info for each agent.

class UniswapV3Env(dojo.environments.BaseEnvironment[dojo.actions.uniswapV3.UniswapV3Action, dojo.observations.uniswapV3.UniswapV3Observation]):

The UniswapV3Env environment models the UniswapV3 system dynamics.

It is responsible for updating both the protocol and agent states through time.

UniswapV3Env( chain: dojo.common.constants.Chain, agents: list[dojo.agents.BaseAgent], date_range: tuple[datetime.datetime, datetime.datetime], pools: list[str], market_impact: Literal['replay', 'no_market', 'replay_trades_only'] = 'replay', backend_type: Literal['forked', 'local', 'live'] = 'forked', dataloader: Union[dojo.dataloaders.base_loader.BaseLoader, Literal['auto'], NoneType] = 'auto', port: Optional[int] = None, token_data: Optional[dict[str, dict[datetime.datetime, decimal.Decimal]]] = None)

Instantiate a UniswapV3Env object.

Parameters

date_range: date range of the environment simulation.
agents: list of agents to use in the environment.
pools: list of pool names to use in the environment.
market_impact: market impact model to use, one of: ['replay', 'no_market']. Defaults to 'replay'.
backend_type: type of backend to use, one of: ['forked', 'local', 'live']. Defaults to 'forked'.
- 'forked': forks a live network to initialize the environment state at the given date range. This mode requires more communication with an archive node and is more unpredictable when funding agents with tokens, but it is faster to intialize state, especially at later start dates.
- 'local': deploys contracts locally to initialize the environment state at the given date range. This mode requires less communication with an archive node and is guaranteed stable when funding agents with tokens, but it can be slower to initialize state, especially at later start dates.
dataloader: optional dataloader instance to use for loading events data. You'll need to provide this if you want to run a simulation with pools that are not listed in the docs https://compasslabs.github.io/docs/tutorial/environments/UniswapV3#supported-pools.
port: port to connect to the local chain. Defaults to find a free port.
token_data: optional dict containing token price data to pass to each agent, needed if you want to call the agent wealth methods. The dictionary has format:
```
{
    "token_symbol": {
        datetime0: price0,
        datetime1: price1,
        ...
    },
    ...
}
```

Raises

ValueError: if the market impact param is not one given by the UniswapV3MarketModelType enum.

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

Deploy and register contracts on the local network.

Returns

a list of contracts to approve for ERC20 transfers.

Raises

ValueError: If a pool has not yet been deployed at the requested start_block.

def initialize_state(self) -> None:

Initialize the state of the environment.

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

Returns

a list of contracts to approve for ERC20 transfers.

class AAVEv3Env(dojo.environments.BaseEnvironment[dojo.actions.base_action.BaseAction[dojo.observations.aaveV3.AAVEv3Observation], dojo.observations.aaveV3.AAVEv3Observation]):

The AAVEv3Env environment models the AAVEv3Env system dynamics.

It is responsible for updating both the protocol and agent states through time.

AAVEv3Env( chain: dojo.common.constants.Chain, agents: list[dojo.agents.BaseAgent], date_range: tuple[datetime.datetime, datetime.datetime], backend_type: Literal['forked', 'local', 'live'] = 'forked', port: Optional[int] = None, token_data: Optional[dict[str, dict[datetime.datetime, decimal.Decimal]]] = None, market_impact: Literal['default'] = 'default')

Instantiate a AAVEv3Env object.

Parameters

date_range: date range of the environment simulation.
agents: list of agents to use in the environment.
backend_type: type of backend to use, one of: ['forked', 'local', 'live']. Defaults to 'forked'.
- 'forked': forks a live network to initialize the environment state at the given date range. This mode requires more communication with an archive node and is more unpredictable when funding agents with tokens, but it is faster to intialize state, especially at later start dates.
- 'local': deploys contracts locally to initialize the environment state at the given date range. This mode requires less communication with an archive node and is guaranteed stable when funding agents with tokens, but it can be slower to initialize state, especially at later start dates.
port: port to connect to the local chain. Defaults to find a free port.
token_data: optional dict containing token price data to pass to each agent, needed if you want to call the agent wealth methods. The dictionary has format:
```
{
    "token_symbol": {
        datetime0: price0,
        datetime1: price1,
        ...
    },
    ...
}
```

Raises

ValueError: If backend_type is unsupported.
ValueError: If the market impact model is unsupported.
NotImplementedError: If the environment is compiled.

def initialize_state(self) -> None:

Initialize state by supplying tokens to the Aave reserves.

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

Setup contracts for a forked env.

def initReserves(self, treasury_address: str) -> None:

Initialize reserves.

def configureReserves(self) -> None:

Configure reserves.

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

Setup local version of the contracts.

class GmxV2Env(dojo.environments.BaseEnvironment[dojo.actions.gmxV2.base_gmx_action.BaseGmxAction, dojo.observations.gmxV2.GmxV2Observation]):

The GmxV2Env models the gmx v2 system dynamics.

GmxV2Env( chain: dojo.common.constants.Chain, agents: list[dojo.agents.BaseAgent], date_range: tuple[datetime.datetime, datetime.datetime], market_venues: list[dojo.models.gmxV2.market.MarketVenue], market_impact: Literal['replay', 'no_market'] = 'no_market', backend_type: Literal['forked', 'local', 'live'] = 'forked', port: Optional[int] = None, token_data: Optional[dict[str, dict[datetime.datetime, decimal.Decimal]]] = None, execution_delay: int = 10)

Instantiate a GmxV2Env object.

Args: chain: blockchain to use in the environment agents: list of agents to use in the environment. date_range: date range of the environment simulation. market_venues: list of market venues to trade on. market_impact: market impact model to use, one of: ['replay', 'no_market']. Defaults to 'replay'. backend_type: type of backend to use, one of: ['forked', 'local', 'live']. Defaults to 'forked'. - 'forked': forks a live network to initialize the environment state at the given date range. This mode requires more communication with an archive node and is more unpredictable when funding agents with tokens, but it is faster to intialize state, especially at later start dates. - 'local': deploys contracts locally to initialize the environment state at the given date range. This mode requires less communication with an archive node and is guaranteed stable when funding agents with tokens, but it can be slower to initialize state, especially at later start dates. port: port to connect to the local chain. Defaults to find a free port. token_data: optional dict containing token price data to pass to each agent. execution_delay: delay in blocks to wait before executing actions. Defaults to 10.

noqa: DAR401

def initialize_state(self) -> None:

Initialise state.

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

Deploy contracts when running local backend.

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