agent0.chainsync.db.hyperdrive.interface

Utilities for hyperdrive related postgres interactions.

Functions

`add_hyperdrive_addr_to_name`(→ None)	Add username mapping to postgres during agent initialization.
`get_hyperdrive_addr_to_name`(→ pandas.DataFrame)	Get all usermapping and returns as a pandas dataframe.
`add_trade_events`(→ None)	Add transfer events to the transfer events table.
`get_latest_block_number_from_trade_event`(→ int)	Get the latest block number based on the hyperdrive events table in the db.
`get_latest_block_number_from_positions_snapshot_table`(→ int)	Get the latest block number based on the positions snapshot table in the db.
`get_trade_events`(→ pandas.DataFrame)	Get all trade events and returns a pandas dataframe.
`get_current_positions`(→ pandas.DataFrame)	Gets all positions for a given wallet address.
`get_pool_config`(→ pandas.DataFrame)	Get all pool config and returns a pandas dataframe.
`add_pool_config`(→ None)	Add pool config to the pool config table if not exist.
`add_pool_infos`(→ None)	Add a pool info to the poolinfo table.
`add_checkpoint_info`(→ None)	Add checkpoint info to the checkpointinfo table if it doesn't exist.
`get_latest_block_number_from_pool_info_table`(→ int)	Get the latest block number based on the pool info table in the db.
`get_pool_info`(→ pandas.DataFrame)	Get all pool info and returns a pandas dataframe.
`get_latest_block_number_from_checkpoint_info_table`(→ int)	Get the latest block number based on the checkpoint info table in the db.
`get_checkpoint_info`(→ pandas.DataFrame)	Get all info associated with a given checkpoint.
`get_all_traders`(→ pandas.Series)	Get the list of all traders from the TradeEvent table.
`get_position_snapshot`(→ pandas.DataFrame)	Get all position snapshot data and returns a pandas dataframe.
`get_total_pnl_over_time`(→ pandas.DataFrame)	Aggregate pnl over time over all positions a wallet has.
`get_positions_over_time`(→ pandas.DataFrame)	Aggregate over token types over all position types.
`get_realized_value_over_time`(→ pandas.DataFrame)	Aggregate over realized value over all position types.

Module Contents

agent0.chainsync.db.hyperdrive.interface.add_hyperdrive_addr_to_name(name: str, hyperdrive_address: str, session: sqlalchemy.orm.Session, force_update: bool = False) → None

Add username mapping to postgres during agent initialization.

Parameters:

name (str) – The logical name to attach to the wallet address.
hyperdrive_address (str) – A hyperdrive address to map to the name.
session (Session) – The initialized session object.
force_update (bool) – If true and an existing mapping is found, will overwrite.

agent0.chainsync.db.hyperdrive.interface.get_hyperdrive_addr_to_name(session: sqlalchemy.orm.Session, hyperdrive_address: str | None = None) → pandas.DataFrame

Get all usermapping and returns as a pandas dataframe.

Parameters:

session (Session) – The initialized session object
hyperdrive_address (str | None, optional) – The hyperdrive address to filter the results on. Return all if None

Returns:

A DataFrame that consists of the queried pool config data

Return type:

DataFrame

agent0.chainsync.db.hyperdrive.interface.add_trade_events(transfer_events: list[agent0.chainsync.db.hyperdrive.schema.DBTradeEvent], session: sqlalchemy.orm.Session) → None

Add transfer events to the transfer events table.

This function is only used for injecting rows into the db. The actual ingestion happens via trade_events_to_db using dataframes.

Parameters:

transfer_events (list[HyperdriveTransferEvent]) – A list of HyperdriveTransferEvent objects to insert into postgres.
session (Session) – The initialized session object.

agent0.chainsync.db.hyperdrive.interface.get_latest_block_number_from_trade_event(session: sqlalchemy.orm.Session, hyperdrive_address: str | None, wallet_address: str | None) → int

Get the latest block number based on the hyperdrive events table in the db.

Parameters:

session (Session) – The initialized session object.
hyperdrive_address (str | None) – The hyperdrive address to filter the results on. Can be None to return latest block number regardless of pool.
wallet_address (str | None) – The wallet address to filter the results on. Can be None to return latest block number regardless of wallet.

Returns:

The latest block number in the hyperdrive_events table.

Return type:

int

agent0.chainsync.db.hyperdrive.interface.get_latest_block_number_from_positions_snapshot_table(session: sqlalchemy.orm.Session, wallet_addr: str | None, hyperdrive_address: str | None) → int

Get the latest block number based on the positions snapshot table in the db.

Parameters:

session (Session) – The initialized session object.
wallet_addr (str | None) – The wallet address to filter the results on. Can be None to return latest block number regardless of wallet.
hyperdrive_address (str | None) – The hyperdrive address to filter the results on. Can be None to return latest block number regardless of pool.

Returns:

The latest block number in the hyperdrive_events table.

Return type:

int

agent0.chainsync.db.hyperdrive.interface.get_trade_events(session: sqlalchemy.orm.Session, wallet_address: str | list[str] | None = None, hyperdrive_address: str | list[str] | None = None, all_token_deltas: bool = True, sort_ascending: bool = True, query_limit: int | None = None, coerce_float=False) → pandas.DataFrame

Get all trade events and returns a pandas dataframe.

Parameters:

session (Session) – The initialized db session object.
wallet_address (str | list[str] | None, optional) – The wallet address(es) to filter the results on. Return all if None.
hyperdrive_address (str | list[str] | None, optional) – The hyperdrive address(es) to filter the results on. Returns all if None.
all_token_deltas (bool, optional) – When removing liquidity that results in withdrawal shares, the events table returns two entries for this transaction to keep track of token deltas (one for lp tokens and one for withdrawal shares). If this flag is True, will return all entries in the table, which is useful for calculating token positions. If False, will drop the duplicate withdrawal share entry (useful for returning a ticker). Defaults to True.
sort_ascending (bool, optional) – If True, will sort events in ascending block order. Otherwise, will sort in descending order. Defaults to True.
query_limit (int | None, optional) – The number of rows to return. Defaults to return all rows.
coerce_float (bool, optional) – If True, will return floats in dataframe. Otherwise, will return fixed point Decimal. Defaults to False.

Returns:

A DataFrame that consists of the queried trade events data.

Return type:

DataFrame

agent0.chainsync.db.hyperdrive.interface.get_current_positions(session: sqlalchemy.orm.Session, wallet_addr: str | None = None, hyperdrive_address: str | None = None, query_block: int | None = None, show_closed_positions: bool = False, coerce_float=False) → pandas.DataFrame

Gets all positions for a given wallet address.

Parameters:

session (Session) – The initialized db session object.
wallet_addr (str) – The wallet address to filter the results on.
hyperdrive_address (str | None, optional) – The hyperdrive address to filter the results on. Returns all if None.
query_block (int | None, optional) – The block to get positions for. query_block integers matches python slicing notation, e.g., list[:3], list[:-3].
show_closed_positions (bool, optional) – Whether to show positions closed positions (i.e., positions with zero balance). Defaults to False. When False, will only return currently open positions. Useful for gathering currently open positions. When True, will also return any closed positions. Useful for calculating overall pnl of all positions.
coerce_float (bool) – If True, will coerce all numeric columns to float.

Returns:

A DataFrame that consists of the queried pool info data.

Return type:

DataFrame

agent0.chainsync.db.hyperdrive.interface.get_pool_config(session: sqlalchemy.orm.Session, hyperdrive_address: str | None = None, coerce_float=False) → pandas.DataFrame

Get all pool config and returns a pandas dataframe.

Parameters:

session (Session) – The initialized session object.
hyperdrive_address (str | None, optional) – The contract_address to filter the results on. Return all if None. Defaults to returning all.
coerce_float (bool, optional) – If True, will coerce all numeric columns to float. Defaults to False.

Returns:

A DataFrame that consists of the queried pool config data.

Return type:

DataFrame

agent0.chainsync.db.hyperdrive.interface.add_pool_config(pool_config: agent0.chainsync.db.hyperdrive.schema.DBPoolConfig, session: sqlalchemy.orm.Session) → None

Add pool config to the pool config table if not exist.

Verify pool config if it does exist.

Parameters:

pool_config (PoolConfig) – A PoolConfig object to insert into postgres.
session (Session) – The initialized session object.

agent0.chainsync.db.hyperdrive.interface.add_pool_infos(pool_infos: list[agent0.chainsync.db.hyperdrive.schema.DBPoolInfo], session: sqlalchemy.orm.Session) → None

Add a pool info to the poolinfo table.

Parameters:

pool_infos (list[PoolInfo]) – A list of PoolInfo objects to insert into postgres.
session (Session) – The initialized session object.

agent0.chainsync.db.hyperdrive.interface.add_checkpoint_info(checkpoint_info: agent0.chainsync.db.hyperdrive.schema.DBCheckpointInfo, session: sqlalchemy.orm.Session) → None

Add checkpoint info to the checkpointinfo table if it doesn’t exist.

This function is only used for injecting rows into the db. The actual ingestion happens via checkpoint_events_to_db using dataframes.

Parameters:

checkpoint_info (CheckpointInfo) – A CheckpointInfo object to insert into postgres.
session (Session) – The initialized session object.

agent0.chainsync.db.hyperdrive.interface.get_latest_block_number_from_pool_info_table(session: sqlalchemy.orm.Session, hyperdrive_address: str | None = None) → int

Get the latest block number based on the pool info table in the db.

Parameters:

session (Session) – The initialized session object.
hyperdrive_address (str | None) – The hyperdrive address to filter the results on. Can be None to return latest block number regardless of pool.

Returns:

The latest block number in the poolinfo table.

Return type:

int

agent0.chainsync.db.hyperdrive.interface.get_pool_info(session: sqlalchemy.orm.Session, hyperdrive_address: str | None = None, start_block: int | None = None, end_block: int | None = None, coerce_float=False) → pandas.DataFrame

Get all pool info and returns a pandas dataframe.

Parameters:

session (Session) – The initialized session object.
hyperdrive_address (str | None, optional) – The hyperdrive address to filter the query on. Return all if None.
start_block (int | None, optional) – The starting block to filter the query on. start_block integers matches python slicing notation, e.g., list[:3], list[:-3].
end_block (int | None, optional) – The ending block to filter the query on. end_block integers matches python slicing notation, e.g., list[:3], list[:-3].
coerce_float (bool, optional) – If true, will return floats in dataframe. Otherwise, will return fixed point Decimal.

Returns:

A DataFrame that consists of the queried pool info data.

Return type:

DataFrame

agent0.chainsync.db.hyperdrive.interface.get_latest_block_number_from_checkpoint_info_table(session: sqlalchemy.orm.Session, hyperdrive_address: str | None) → int

Get the latest block number based on the checkpoint info table in the db.

Parameters:

session (Session) – The initialized session object.
hyperdrive_address (str | None) – The hyperdrive pool address to filter the query on.

Returns:

The latest block number in the poolinfo table.

Return type:

int

agent0.chainsync.db.hyperdrive.interface.get_checkpoint_info(session: sqlalchemy.orm.Session, hyperdrive_address: str | None = None, checkpoint_time: int | None = None, coerce_float=False) → pandas.DataFrame

Get all info associated with a given checkpoint.

Parameters:

session (Session) – The initialized session object.
hyperdrive_address (str | None, optional) – The hyperdrive pool address to filter the query on. Defaults to returning all checkpoint infos.
checkpoint_time (int | None, optional) – The checkpoint time to filter the query on. Defaults to returning all checkpoint infos.
coerce_float (bool, optional) – If True, will return floats in dataframe. Otherwise, will return fixed point Decimal. Defaults to False

Returns:

A DataFrame that consists of the queried checkpoint info.

Return type:

DataFrame

agent0.chainsync.db.hyperdrive.interface.get_all_traders(session: sqlalchemy.orm.Session, hyperdrive_address: str | None = None) → pandas.Series

Get the list of all traders from the TradeEvent table.

Parameters:

session (Session) – The initialized session object.
hyperdrive_address (str | None, optional) – The hyperdrive pool address to filter the query on. Defaults to returning all traders.

Returns:

A list of addresses that have made a trade.

Return type:

pd.Series

agent0.chainsync.db.hyperdrive.interface.get_position_snapshot(session: sqlalchemy.orm.Session, hyperdrive_address: str | list[str] | None = None, latest_entry: bool = False, start_block: int | None = None, end_block: int | None = None, wallet_address: list[str] | str | None = None, coerce_float=False) → pandas.DataFrame

Get all position snapshot data and returns a pandas dataframe.

Parameters:

session (Session) – The initialized session object.
hyperdrive_address (str | list[str] | None, optional) – The hyperdrive pool address(es) to filter the query on. Defaults to returning all position snapshots.
latest_entry (bool, optional) – If true, will return the latest entry for every pool. Defaults to False.
start_block (int | None, optional) – The starting block to filter the query on. start_block integers matches python slicing notation, e.g., list[:3], list[:-3]. Defaults to first entry. Not used if latest_entry is True.
end_block (int | None, optional) – The ending block to filter the query on. end_block integers matches python slicing notation, e.g., list[:3], list[:-3]. Defaults to last entry. Not used if latest_entry is True.
wallet_address (list[str] | None, optional) – The wallet addresses to filter the query on. Returns all if None.
coerce_float (bool, optional) – If True, will return floats in dataframe. Otherwise, will return fixed point Decimal. Defaults to False.

Returns:

A DataFrame that consists of the queried pool info data.

Return type:

DataFrame

agent0.chainsync.db.hyperdrive.interface.get_total_pnl_over_time(session: sqlalchemy.orm.Session, start_block: int | None = None, end_block: int | None = None, wallet_address: list[str] | None = None, coerce_float=False) → pandas.DataFrame

Aggregate pnl over time over all positions a wallet has.

Parameters:

session (Session) – The initialized session object.
start_block (int | None, optional) – The starting block to filter the query on. start_block integers matches python slicing notation, e.g., list[:3], list[:-3].
end_block (int | None, optional) – The ending block to filter the query on. end_block integers matches python slicing notation, e.g., list[:3], list[:-3].
wallet_address (list[str] | None, optional) – The wallet addresses to filter the query on. Returns all if None.
coerce_float (bool) – If true, will return floats in dataframe. Otherwise, will return fixed point Decimal.

Returns:

A DataFrame that consists of the queried pool info data.

Return type:

DataFrame

agent0.chainsync.db.hyperdrive.interface.get_positions_over_time(session: sqlalchemy.orm.Session, start_block: int | None = None, end_block: int | None = None, wallet_address: list[str] | None = None, coerce_float=False) → pandas.DataFrame

Aggregate over token types over all position types.

Parameters:

session (Session) – The initialized session object.
start_block (int | None, optional) – The starting block to filter the query on. start_block integers matches python slicing notation, e.g., list[:3], list[:-3].
end_block (int | None, optional) – The ending block to filter the query on. end_block integers matches python slicing notation, e.g., list[:3], list[:-3].
wallet_address (list[str] | None, optional) – The wallet addresses to filter the query on. Returns all if None.
coerce_float (bool) – If true, will return floats in dataframe. Otherwise, will return fixed point Decimal.

Returns:

A DataFrame that consists of the queried pool info data.

Return type:

DataFrame

agent0.chainsync.db.hyperdrive.interface.get_realized_value_over_time(session: sqlalchemy.orm.Session, start_block: int | None = None, end_block: int | None = None, wallet_address: list[str] | None = None, coerce_float=False) → pandas.DataFrame

Aggregate over realized value over all position types.

Parameters:

session (Session) – The initialized session object.
start_block (int | None, optional) – The starting block to filter the query on. start_block integers matches python slicing notation, e.g., list[:3], list[:-3].
end_block (int | None, optional) – The ending block to filter the query on. end_block integers matches python slicing notation, e.g., list[:3], list[:-3].
wallet_address (list[str] | None, optional) – The wallet addresses to filter the query on. Returns all if None.
coerce_float (bool) – If true, will return floats in dataframe. Otherwise, will return fixed point Decimal.

Returns:

A DataFrame that consists of the queried pool info data.

Return type:

DataFrame