agent0.core.hyperdrive.interactive.local_chain

The chain objects that encapsulates a chain.

Classes

LocalChain

Launches a local anvil chain in a subprocess, along with a postgres container.

Module Contents

class agent0.core.hyperdrive.interactive.local_chain.LocalChain(config: Config | None = None, fork_uri: str | None = None, fork_block_number: int | None = None)

Bases: agent0.core.hyperdrive.interactive.chain.Chain

Launches a local anvil chain in a subprocess, along with a postgres container.

class Config

Bases: agent0.core.hyperdrive.interactive.chain.Chain.Config

The configuration for the local chain object.

verbose: bool = False

If True, will print underlying subprocess output to stdout. Defaults to suppressing output.

dashboard_port: int | None = None

The URL port for the deployed dashboard. Defaults to finding an open port.

block_time: int | None = None

If None, mines per transaction. Otherwise mines every block_time seconds.

block_timestamp_interval: int | None = 12

Number of seconds to advance time for every mined block. Uses real time if None.

chain_host: str | None = None

The host to bind for the anvil chain. Defaults to 127.0.0.1.

chain_id: int | None = None

The chain ID for the local anvil chain.

chain_port: int = 10000

The port to bind for the anvil chain. Will fail if this port is being used.

chain_genesis_timestamp: int | None = None

The genesis timestamp (in epoch seconds) for the anvil chain. If None, uses the current time.

transaction_block_keeper: int = 10000

The number of blocks to keep transaction records for. Undocumented in Anvil, we’re being optimistic here.

snapshot_dir: str | None = None

The directory where the snapshot will be stored. Defaults to .interactive_state/snapshot/chain_<chain_port>.

load_rng_on_snapshot: bool = True

If True, loading a snapshot also loads the RNG state of the underlying policy. This results in the same RNG state as when the snapshot was taken. If False, will use the existing RNG state before load. Defaults to False.

manual_database_sync: bool = False

If True, depends on the user to sync the database against the chain by calling pool.sync_database(). NOTE if this is True, the caller must call pool.sync_database() before any references to getting the user’s wallet, including any policy actions that require a wallet. Otherwise, the wallet may be out of date. Use this at your own risk.

crash_log_ticker: bool = False

Whether to log the trade ticker in crash reports. Defaults to False.

config = None
dashboard_subprocess: subprocess.Popen | None = None
cleanup()

Kills the subprocess in this class’ destructor.

get_deployer_account_private_key() → str

Get the private key of the deployer account.

Returns:

The private key for the deployer account.

Return type:

str

get_deployer_account() → eth_account.signers.local.LocalAccount

Get the public key of the deployer account.

Returns:

The public key for the deployer account.

Return type:

src

get_deployer_address() → str

Get the public key of the deployer account.

Returns:

The public key for the deployer account.

Return type:

src

property is_local_chain: bool

Returns if this object is a local chain.

mine_blocks(num_blocks: int = 1) → None

Advance time for this chain using the anvil_mine RPC call.

This function mines the specified amount of blocks with the chain config specified time between blocks.

Note

This advances the chain for all pool connected to this chain.

Parameters:

num_blocks (int) – The amount of blocks to advance. Defaults to 1.

advance_time(time_delta: int | datetime.timedelta, create_checkpoints: bool = True) → dict[agent0.core.hyperdrive.interactive.local_hyperdrive.LocalHyperdrive, list[hyperdrivetypes.CreateCheckpointEventFP]]

Advance time for this chain using the evm_mine RPC call.

This function looks at the timestamp of the current block, then mines a block explicitly setting the timestamp to the current block timestamp + time_delta.

If create_checkpoints is True, it will also create intermediate checkpoints when advancing time.

Note

This advances the chain for all pool connected to this chain.

Note

This function is a best effort attempt at being exact in advancing time, but the final result may be off by a second.

Parameters:

  • time_delta (int | timedelta) – The amount of time to advance. Can either be a datetime.timedelta object or an integer in seconds.

  • create_checkpoints (bool, optional) – If set to true, will create intermediate checkpoints between advance times. Defaults to True.

Returns:

Returns a dictionary keyed by the interactive hyperdrive object, with a value of a list of emitted CreateCheckpoint events called from advancing time.

Return type:

dict[InteractiveHyperdrive, list[CreateCheckpoint]]

save_state(save_dir: pathlib.Path | None = None, save_prefix: str | None = None) → pathlib.Path

Saves the interactive state using the anvil_dumpState RPC call. Saving/loading state can be done across chains.

Parameters:

  • save_dir (Path, optional) – The directory to save the state to. Defaults to {Config.save_state_dir}/{save_prefix}_{current_time}/, where Config.save_state_dir defaults to ./.interactive_state/.

  • save_prefix (str, optional) – If save_dir wasn’t provided, prepends an optional prefix to the time suffix for this state.

Returns:

The path to the saved state.

Return type:

str

abstract load_state(load_dir: pathlib.Path) → None

Loads the interactive state from the save_state function. Saving/loading state can be done across chains.

Note

This feature is currently unavailable. There are issues around load_state, namely:

  • Anvil load state doesn’t load the block number and timestamp.

  • Anvil load state only loads the current state, not all previous states.

  • There exists an issue with the underlying yield contract, as there is a last_updated var that gets saved, but anvil doesn’t load the original timestamp, so the yield contract throws an error. (May be able to solve if we’re able to solve issue 1 to correctly load the block number and previous states.)

  • To load the state in another chain, we need this function to load all original objects created from the saved chain, e.g., interactive_hyperdrive and all agents they contain, and return them from this function.

Parameters:

load_dir (Path) – The directory to load the state from.

save_snapshot() → None

Saves a snapshot using the evm_snapshot RPC call. The chain can store one snapshot at a time, saving another snapshot overwrites the previous snapshot. Saving/loading snapshot only persist on the same chain, not across chains.

load_snapshot() → None

Loads the previous snapshot using the evm_revert RPC call. Can load the snapshot multiple times.

Note

Saving/loading snapshot only persist on the same chain, not across chains.

Note

Loading a snapshot affects any agent’s active pools and policies, and will revert both to the state of the saved snapshot.

init_agent(private_key: str | None = None, public_address: str | None = None, pool: agent0.core.hyperdrive.interactive.hyperdrive.Hyperdrive | None = None, policy: Type[agent0.core.hyperdrive.policies.HyperdriveBasePolicy] | None = None, policy_config: agent0.core.hyperdrive.policies.HyperdriveBasePolicy.Config | None = None, name: str | None = None, base: fixedpointmath.FixedPoint | None = None, eth: fixedpointmath.FixedPoint | None = None) → agent0.core.hyperdrive.interactive.local_hyperdrive_agent.LocalHyperdriveAgent

Initializes an agent with initial funding and a logical name.

Parameters:

  • private_key (str, optional) – The private key of the associated account. Default is auto-generated.

  • public_address (str | None, optional) – Must be None for local agents.

  • pool (LocalHyperdrive, optional) – An optional pool to set as the active pool.

  • policy (HyperdrivePolicy, optional) – An optional policy to set as the active policy.

  • policy_config (HyperdrivePolicy, optional) – The configuration for the attached policy.

  • name (str, optional) – The name of the agent. Defaults to the wallet address.

  • base (FixedPoint | None, optional) – The amount of base to fund the agent with. Defaults to 0. If a private key is provided then the base amount is added to their previous balance.

  • eth (FixedPoint | None, optional) – The amount of ETH to fund the agent with. Defaults to 0. If a private key is provided then the eth amount is added to their previous balance.

Returns:

The agent object for a user to execute trades with.

Return type:

LocalHyperdriveAgent

get_dashboard_iframe(width: int = 1000, height: int = 800) → IPython.display.IFrame

Embeds the streamlit dashboard into a Jupyter notebook as an IFrame.

Note

The interactive hyperdrive script must be in a paused state (before cleanup) for the dashboard to connect with the underlying database, otherwise cleanup and/or the main thread executed will kill the streamlit server. Passing blocking=True will block execution of the main script in this function until a keypress is registered.

Parameters:

  • width (int) – Width, in pixels, of the IFrame. Defaults to 1000.

  • height (int) – Height, in pixels, of the IFrame. Defaults to 800.

Returns:

A dashboard IFrame that can be shown in a Jupyter notebook with the display command.

Return type:

IFrame

run_dashboard(blocking: bool = False) → None

Runs the streamlit dashboard in a subprocess connected to interactive hyperdrive.

Note

The interactive hyperdrive script must be in a paused state (before cleanup) for the dashboard to connect with the underlying database, otherwise cleanup and/or the main thread executed will kill the streamlit server. Passing blocking=True will block execution of the main script in this function until a keypress is registered.

Parameters:

blocking (bool) – If True, will block execution of the main script in this function until a keypress is registered. When in blocking mode, the server will be killed upon return of control to caller. If False, will clean up subprocess in cleanup.