VaultSnapshotDatabase

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

class VaultSnapshotDatabase

Bases: object

DuckDB database for storing GRVT vault snapshots over time.

Stores point-in-time snapshots of vault metrics including TVL and share price. Each snapshot is keyed by timestamp and vault ID.

Example:

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

db = VaultSnapshotDatabase(Path("vaults.duckdb"))
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_latest_snapshots`()	Get the most recent snapshot for each vault.
`get_snapshot_timestamps`()	Get all unique snapshot timestamps in the database.
`get_vault_count`()	Get number of unique vaults in the database.
`get_vault_history`(vault_id)	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.grvt.vault_scanner.VaultSnapshot) – VaultSnapshot to insert.

insert_snapshots(snapshots)

Bulk insert vault snapshots into the database.

Parameters: snapshots (list[eth_defi.grvt.vault_scanner.VaultSnapshot]) – List 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 ID.
Return type: pandas.DataFrame

get_vault_history(vault_id)

Get all snapshots for a specific vault.

Parameters: vault_id (str) – The vault string ID to query.
Returns: DataFrame with all snapshots for the vault, ordered by 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 IDs.
Return type: int

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