agent0.chainsync.exec.analyze_data

Script to format on-chain hyperdrive pool, config, and transaction data post-processing.

Functions

analyze_data([start_block, interfaces, rpc_uri, ...])

Execute the data acquisition pipeline.

get_latest_data_block(→ int)

Gets the latest block the data pipeline has written

Module Contents

agent0.chainsync.exec.analyze_data.analyze_data(start_block: int = 0, interfaces: list[agent0.ethpy.hyperdrive.HyperdriveReadInterface] | None = None, rpc_uri: str | None = None, hyperdrive_addresses: list[eth_typing.ChecksumAddress] | dict[str, eth_typing.ChecksumAddress] | None = None, db_session: sqlalchemy.orm.Session | None = None, postgres_config: agent0.chainsync.PostgresConfig | None = None, calc_pnl: bool = True, backfill: bool = False, backfill_sample_period: int | None = None, backfill_progress_bar: bool = False)

Execute the data acquisition pipeline.

Parameters:

  • start_block (int, optional) – The starting block to run analysis on for backfilling.

  • interfaces (list[HyperdriveReadInterface] | None, optional) – A collection of Hyperdrive interface objects, each connected to a pool. If not set, will initialize one based on rpc_uri and hyperdrive_address.

  • rpc_uri (str, optional) – The URI for the web3 provider to initialize the interface with. Not used if an interface is provided.

  • hyperdrive_addresses (list[ChecksumAddress] | dict[str, ChecksumAddress] | None, optional) – A collection of Hyperdrive address, each pointing to an initialized pool. Can also be the output of get_hyperdrive_addresses_from_registry, which is a dictionary keyed by a logical name and a value of a hyperdrive address. Not used if a list of interfaces is provided.

  • db_session (Session | None) – Session object for connecting to db. If None, will initialize a new session based on .env.

  • postgres_config (PostgresConfig | None = None,) – PostgresConfig for connecting to db. If none, will set from .env.

  • calc_pnl (bool) – Whether to calculate pnl. Defaults to True.

  • backfill (bool, optional) – If true, will fill in missing pool info data for every backfill_sample_period blocks. Defaults to True.

  • backfill_sample_period (int | None, optional) – The sample frequency when backfilling. If None, will backfill every block.

  • backfill_progress_bar (bool, optional) – If true, will show a progress bar when backfilling. Defaults to False.

agent0.chainsync.exec.analyze_data.get_latest_data_block(db_session: sqlalchemy.orm.Session) → int

Gets the latest block the data pipeline has written Since there are multiple tables that analysis reads from, we query the latest block from all read tables and select the minimum block from the list.

Parameters:

db_session (Session) – The initialized db session.

Returns:

The latest block number from the PoolInfo table.

Return type:

int