grvt.vault_scanner

Documentation for eth_defi.grvt.vault_scanner Python module.

GRVT vault scanner with DuckDB storage.

This module provides functionality for scanning all known GRVT vaults and storing historical snapshots in a DuckDB database for tracking TVL and other metrics over time.

Vault discovery uses the public GraphQL API at https://edge.grvt.io/query (which includes per-vault fee data), enriched with live data from the market data API.

Example usage:

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

db = scan_vaults()
print(db.get_latest_snapshots())
db.close()

Functions

scan_vaults([session, db_path, timeout, ...])

Scan all GRVT vaults and store snapshots in DuckDB.

Classes

VaultSnapshot

A point-in-time snapshot of a GRVT vault's state.

VaultSnapshotDatabase

DuckDB database for storing GRVT vault snapshots over time.

GRVT_VAULT_METADATA_DATABASE = PosixPath('/home/runner/.tradingstrategy/grvt/vaults.duckdb')

Default path for GRVT vault metadata database

class VaultSnapshot

Bases: object

A point-in-time snapshot of a GRVT vault’s state.

Contains the key metrics we want to track over time for each vault.

snapshot_timestamp: datetime.datetime

When this snapshot was taken

vault_id: str

Vault string ID on the GRVT platform (e.g. VLT:xxx)

chain_vault_id: int

Numeric on-chain vault ID

name: str

Vault display name

tvl: float | None

Total Value Locked in USDT

share_price: float | None

Current share price

apr: float | None

Annualised percentage return

investor_count: int | None

Number of investors in the vault

__init__(snapshot_timestamp, vault_id, chain_vault_id, name, tvl, share_price, apr=None, investor_count=None)
Parameters
Return type

None

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.

__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

scan_vaults(session=None, db_path=PosixPath('/home/runner/.tradingstrategy/grvt/vaults.duckdb'), timeout=30.0, only_discoverable=True)

Scan all GRVT vaults and store snapshots in DuckDB.

Discovers vaults via the public GraphQL API (includes per-vault fees), enriched with live data from the market data API. No authentication required.

Example:

from eth_defi.grvt.vault_scanner import scan_vaults

db = scan_vaults()
df = db.get_latest_snapshots()
print(f"Scanned {len(df)} vaults")
db.close()
Parameters
  • session (requests.sessions.Session | None) – HTTP session. If None, one is created via create_grvt_session().

  • db_path (pathlib.Path) – Path to the DuckDB database file.

  • timeout (float) – HTTP request timeout in seconds.

  • only_discoverable (bool) – If True, only scan vaults marked as discoverable.

Returns

VaultSnapshotDatabase instance with the newly inserted snapshots.

Return type

eth_defi.grvt.vault_scanner.VaultSnapshotDatabase