VaultDatabase

Documentation for eth_defi.vault.vaultdb.VaultDatabase Python class.

class VaultDatabase

Bases: object

Metadata for all vaults across all chains we know about.

  • The pickle format dump for vault-db.pickle

  • Does not include share price/TVL data, only metadata

Manually checking the contents:

from eth_defi.vault.vaultdb import VaultDatabase
from eth_defi.vault.base import VaultSpec

vault_db = VaultDatabase.read()
# Choose D2 HYPE++ on Arbitrum
spec = VaultSpec(chain_id=42161, vault_address="0x75288264FDFEA8ce68e6D852696aB1cE2f3E5004")
vault_metadata = vault_db.rows[spec]
print(vault_metadata)

Attributes summary

rows

Correctly detected vaults.

leads

(chain id, address) -> vault info mapping for ongoing scans on which we are still unsure

last_scanned_block

Chain id -> block number.

Methods summary

__init__([rows, leads, last_scanned_block])

get(key[, default])

Get vault row by spec.

get_chain_start_block(chain_id[, ...])

Get the block to start scanning from for a given chain id.

get_existing_leads_by_chain(chain_id)

Get existing leads for a given chain id.

get_lead_count()

items()

Iterable human readable rows.

keys()

Iterable human readable vault (chain, address) tuples.

limit_to_single_vault(vault_spec)

Limit results to a single vault.

read([path])

Load the picked file.

to_dataframe(rows)

Convert selected vault rows to a Pandas DataFrame for diagnostics.

update_leads_and_rows(chain_id, ...)

values()

Iterable human readable rows.

write([path])

Do an atomic write to avoid corrupted data.

rows: dict[eth_defi.vault.base.VaultSpec, eth_defi.vault.vaultdb.VaultRow]

Correctly detected vaults.

(chain id, address) -> vault info mapping for detected vaults

leads: dict[eth_defi.vault.base.VaultSpec, eth_defi.erc_4626.discovery_base.PotentialVaultMatch]

(chain id, address) -> vault info mapping for ongoing scans on which we are still unsure

Persistent to continue scan

last_scanned_block: dict[int, int]

Chain id -> block number.

static read(path=PosixPath('/home/runner/.tradingstrategy/vaults/vault-metadata-db.pickle'))

Load the picked file.

Example:

from eth_defi.vault.vaultdb import VaultDatabase

# Load from the default location on local filesystem
vault_db = VaultDatabase.read()

print(f"We have data for {vault_db.get_lead_count()} potential vaults")
Parameters

path (pathlib.Path | io.BufferedIOBase) –

Return type

eth_defi.vault.vaultdb.VaultDatabase

write(path=PosixPath('/home/runner/.tradingstrategy/vaults/vault-metadata-db.pickle'))

Do an atomic write to avoid corrupted data.

Parameters

path (pathlib.Path) –

get_chain_start_block(chain_id, default_start_block=1)

Get the block to start scanning from for a given chain id.

  • Used to for iterative scanning

Parameters

chain_id (int) –

Return type

int

get_existing_leads_by_chain(chain_id)

Get existing leads for a given chain id.

  • Used to for iterative scanning

Parameters

chain_id (int) –

Return type

dict[eth_typing.evm.HexAddress, eth_defi.erc_4626.discovery_base.PotentialVaultMatch]

limit_to_single_vault(vault_spec)

Limit results to a single vault.

Used for diagnostics.

Parameters

vault_spec (eth_defi.vault.base.VaultSpec) –

Return type

eth_defi.vault.vaultdb.VaultDatabase

static to_dataframe(rows)

Convert selected vault rows to a Pandas DataFrame for diagnostics.

Parameters

rows (Iterable[eth_defi.vault.vaultdb.VaultRow]) –

Return type

pandas.core.frame.DataFrame

keys()

Iterable human readable vault (chain, address) tuples.

Return type

Iterable[eth_defi.vault.base.VaultSpec]

values()

Iterable human readable rows.

Return type

Iterable[eth_defi.vault.vaultdb.VaultRow]

items()

Iterable human readable rows.

Return type

Iterable[tuple[eth_typing.evm.HexAddress, eth_defi.vault.vaultdb.VaultRow]]

get(key, default=None)

Get vault row by spec.

Parameters

key (eth_defi.vault.base.VaultSpec) –

Return type

eth_defi.vault.vaultdb.VaultRow | None

__init__(rows=<factory>, leads=<factory>, last_scanned_block=<factory>)
Parameters
Return type

None