VaultSnapshotDatabase

Documentation for eth_defi.hyperliquid.vault_scanner.VaultSnapshotDatabase Python class.

class VaultSnapshotDatabase

Bases: object

DuckDB database for storing Hyperliquid vault snapshots over time.

Stores point-in-time snapshots of vault metrics including TVL, PnL, APR, and follower count. Each snapshot is keyed by timestamp and vault address.

Example:

from pathlib import Path
from eth_defi.hyperliquid.vault_scanner import VaultSnapshotDatabase

db = VaultSnapshotDatabase(Path("vaults.duckdb"))

# Query recent snapshots
df = db.get_latest_snapshots()
print(df)

db.close()

Initialise the database connection.

Parameters: path – Path to the DuckDB file. Parent directories will be created if needed.

Methods summary

`__init__`(path)	Initialise the database connection.
`close`()	Close the database connection.
`get_count`()	Get total number of snapshot records in the database.
`get_disabled_vault_addresses`()	Get vault addresses that have scan_disabled_reason set in their latest snapshot.
`get_latest_snapshots`()	Get the most recent snapshot for each vault.
`get_snapshot_timestamps`()	Get all unique snapshot timestamps in the database.
`get_snapshots_at_time`(timestamp)	Get all vault snapshots at a specific timestamp.
`get_vault_count`()	Get number of unique vaults in the database.
`get_vault_history`(vault_address)	Get all snapshots for a specific vault.
`insert_snapshot`(snapshot)	Insert a single vault snapshot into the database.
`insert_snapshots`(snapshots)	Bulk insert vault snapshots into the database.
`is_closed`()	Check if the database connection is closed.
`save`()	Force a checkpoint to ensure data is written to disk.

__init__(path)

Initialise the database connection.

Parameters: path (pathlib.Path) – Path to the DuckDB file. Parent directories will be created if needed.

insert_snapshot(snapshot)

Insert a single vault snapshot into the database.

Parameters: snapshot (eth_defi.hyperliquid.vault_scanner.VaultSnapshot) – VaultSnapshot to insert

insert_snapshots(snapshots)

Bulk insert vault snapshots into the database.

Parameters: snapshots (Iterator[eth_defi.hyperliquid.vault_scanner.VaultSnapshot]) – Iterator of VaultSnapshot objects to insert

get_latest_snapshots()

Get the most recent snapshot for each vault.

Returns: DataFrame with the latest snapshot for each vault address
Return type: pandas.DataFrame

get_vault_history(vault_address)

Get all snapshots for a specific vault.

Parameters: vault_address (eth_typing.evm.HexAddress) – The vault’s blockchain address
Returns: DataFrame with all snapshots for the vault, ordered by timestamp
Return type: pandas.DataFrame

get_snapshots_at_time(timestamp)

Get all vault snapshots at a specific timestamp.

Parameters: timestamp (datetime.datetime) – The snapshot timestamp to query
Returns: DataFrame with all vault snapshots at that timestamp
Return type: pandas.DataFrame

get_snapshot_timestamps()

Get all unique snapshot timestamps in the database.

Returns: List of snapshot timestamps, ordered from oldest to newest
Return type: list[datetime.datetime]

get_count()

Get total number of snapshot records in the database.

Returns: Total count of snapshot records
Return type: int

get_vault_count()

Get number of unique vaults in the database.

Returns: Count of unique vault addresses
Return type: int

get_disabled_vault_addresses()

Get vault addresses that have scan_disabled_reason set in their latest snapshot.

Returns: Set of vault addresses that should be skipped during scanning
Return type: set[eth_typing.evm.HexAddress]

save(): Force a checkpoint to ensure data is written to disk.

close(): Close the database connection.

is_closed()

Check if the database connection is closed.

Return type: bool