vault.vaultdb

Documentation for eth_defi.vault.vaultdb Python module.

Describe vault database pickle format.

Functions

get_pipeline_data_dir()	Return the pipeline data directory, configurable via PIPELINE_DATA_DIR.
has_good_fee_data(vault_row)	Check if the vault row has good fee data.
read_default_vault_prices()	Read the default raw vault prices database.

Classes

VaultDatabase	Metadata for all vaults across all chains we know about.
VaultRow	Vault info gathered during the vault discovery from the chain.

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)

__init__(rows=<factory>, leads=<factory>, last_scanned_block=<factory>)

Parameters

• rows (dict[eth_defi.vault.base.VaultSpec, eth_defi.vault.vaultdb.VaultRow]) –

• leads (dict[eth_defi.vault.base.VaultSpec, eth_defi.erc_4626.discovery_base.PotentialVaultMatch]) –

• last_scanned_block (dict[int, int]) –

Return type

None

get(key, default=None)

Get vault row by spec.

Parameters

key (eth_defi.vault.base.VaultSpec) –

Return type

Optional[eth_defi.vault.vaultdb.VaultRow]

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]

items()

Iterable human readable rows.

Return type

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

keys()

Iterable human readable vault (chain, address) tuples.

Return type

Iterable[eth_defi.vault.base.VaultSpec]

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 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 (Union[pathlib.Path, io.BufferedIOBase]) –

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.DataFrame

values()

Iterable human readable rows.

Return type

Iterable[eth_defi.vault.vaultdb.VaultRow]

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

Do an atomic write to avoid corrupted data.

Parameters

path (pathlib.Path) –

class VaultRow

Bases: TypedDict

Vault info gathered during the vault discovery from the chain.

• Avaulable as VaultDb pickle

• Human readable entry

• Machine readable data in _detection_data

__init__(*args, **kwargs)

__new__(**kwargs)

clear()

Remove all items from the dict.

copy()

Return a shallow copy of the dict.

fromkeys(value=None, /)

Create a new dictionary with keys from iterable and values set to value.

get(key, default=None, /)

Return the value for key if key is in the dictionary, else default.

items()

Return a set-like object providing a view on the dict’s items.

keys()

Return a set-like object providing a view on the dict’s keys.

pop(k[, d]) → v, remove specified key and return the corresponding value.

If the key is not found, return the default if given; otherwise, raise a KeyError.

popitem()

Remove and return a (key, value) pair as a 2-tuple.

Pairs are returned in LIFO (last-in, first-out) order. Raises KeyError if the dict is empty.

setdefault(key, default=None, /)

Insert key with a value of default if key is not in the dictionary.

Return the value for key if key is in the dictionary, else default.

update([E, ]**F) → None.  Update D from mapping/iterable E and F.

If E is present and has a .keys() method, then does: for k in E.keys(): D[k] = E[k] If E is present and lacks a .keys() method, then does: for k, v in E: D[k] = v In either case, this is followed by: for k in F: D[k] = F[k]

values()

Return an object providing a view on the dict’s values.

get_pipeline_data_dir()

Return the pipeline data directory, configurable via PIPELINE_DATA_DIR.

When the environment variable PIPELINE_DATA_DIR is set, all pipeline files (vault DB pickle, parquet, reader state, DuckDB, cycle state, lock) are stored under that directory. Otherwise falls back to the standard ~/.tradingstrategy/vaults location.

Returns

Absolute path to the pipeline data directory.

Return type

pathlib.Path

has_good_fee_data(vault_row)

Check if the vault row has good fee data.

Parameters

vault_row (eth_defi.vault.vaultdb.VaultRow) –

Return type

bool

read_default_vault_prices()

Read the default raw vault prices database.

• Use the default cleaned price data file

Return type

pandas.DataFrame