XerberusDatabase

Documentation for eth_defi.xerberus.database.XerberusDatabase Python class.

class XerberusDatabase

Bases: object

DuckDB database for Xerberus registry and vault risk scores.

Thread safety: all operations are protected by _db_lock.

Example:

from pathlib import Path
from eth_defi.xerberus.database import XerberusDatabase

db = XerberusDatabase(Path("/tmp/xerberus.duckdb"))
db.save()
db.close()

Initialise the database connection.

Parameters

  • path – Path to the DuckDB file. Parent directories are created if needed (unless read_only).

  • read_only – Open existing database read-only (export path). Requires the file to exist.

Methods summary

__init__(path, *[, read_only])

Initialise the database connection.

close()

Close the database connection.

get_entity_counts()

Return simple table row counts.

get_latest_pool_score(chain_id, address[, ...])

Get the latest registry pool score for a vault.

get_latest_protocol_by_id(entity_id[, ...])

Get the latest registry protocol snapshot by Xerberus id.

get_latest_registry_entities([entity_type, ...])

Return the latest snapshot per (entity_type, entity_id).

get_latest_vault_list([max_age_days])

Return the latest vault-list row per (chain_id, address).

get_report_url(chain_id, address)

Return a cached report URL for (chain_id, address) if successful.

insert_registry_snapshot_batch(entities, ...)

Insert a batch of registry entity snapshots.

insert_vault_list_snapshot_batch(platform, ...)

Insert vault list rows for one platform.

save()

Force a checkpoint to persist data to disk.

update_sync_state(data_type[, meta])

Update sync watermark for a data type.

upsert_report_url(chain_id, address, ...[, ...])

Store or replace a report URL for (chain_id, address).

upsert_score_daily_points(points)

Upsert daily scores with DELETE + INSERT on (entity_type, entity_id, day).
__init__(path, *, read_only=False)

Initialise the database connection.

Parameters

  • path (pathlib.Path) – Path to the DuckDB file. Parent directories are created if needed (unless read_only).

  • read_only (bool) – Open existing database read-only (export path). Requires the file to exist.

close()

Close the database connection.

Return type

None

save()

Force a checkpoint to persist data to disk.

Return type

None

insert_registry_snapshot_batch(entities, fetched_at)

Insert a batch of registry entity snapshots.

Normalises addresses to lowercase. Skips pool rows missing chainId or address with a warning.

Parameters

  • entities (list[dict]) – Raw entity dicts from /registry/scores.

  • fetched_at (datetime.datetime) – Poll timestamp (naive UTC).

Returns

Number of rows inserted.

Return type

int

insert_vault_list_snapshot_batch(platform, vaults, fetched_at)

Insert vault list rows for one platform.

Parameters

  • platform (str) – Platform slug.

  • vaults (list[dict]) – Raw vault dicts from /vault/list.

  • fetched_at (datetime.datetime) – Poll timestamp.

Returns

Number of rows inserted.

Return type

int

upsert_score_daily_points(points)

Upsert daily scores with DELETE + INSERT on (entity_type, entity_id, day).

Each point dict must contain: entity_type, entity_id, day (date or datetime), score, fetched_at, and optionally chain_id / address.

Parameters

points (list[dict]) – Daily score rows for the current scan.

Returns

Number of rows written.

Return type

int

upsert_report_url(chain_id, address, report_url, fetched_at, entity_id=None, error=None)

Store or replace a report URL for (chain_id, address).

Parameters

  • chain_id (int) – EIP-155 chain id.

  • address (str) – Vault address (will be lowercased).

  • report_url (Optional[str]) – Deep link URL, or None on error.

  • fetched_at (datetime.datetime) – Fetch timestamp.

  • entity_id (Optional[str]) – Registry pool id when known.

  • error (Optional[str]) – Error message when fetch failed.

Return type

None

update_sync_state(data_type, meta=None)

Update sync watermark for a data type.

Parameters

  • data_type (str) – e.g. registry, vault_list:morpho.

  • meta (Optional[dict]) – Optional metadata dict serialised as JSON.

Return type

None

get_latest_registry_entities(entity_type=None, max_age_days=None)

Return the latest snapshot per (entity_type, entity_id).

Parameters

  • entity_type (Optional[str]) – Optional filter (pool, protocol, …).

  • max_age_days (Optional[int]) – Optional age filter applied in Python after the SQL latest-row join.

Returns

DataFrame of latest entity rows.

Return type

pandas.DataFrame

get_latest_vault_list(max_age_days=None)

Return the latest vault-list row per (chain_id, address).

Parameters

max_age_days (Optional[int]) – Optional age filter.

Returns

DataFrame of latest vault list rows.

Return type

pandas.DataFrame

get_latest_pool_score(chain_id, address, max_age_days=30)

Get the latest registry pool score for a vault.

Parameters

  • chain_id (int) – EIP-155 chain id.

  • address (str) – Vault address (case-insensitive).

  • max_age_days (Optional[int]) – Max age filter; None disables.

Returns

Dict of column values, or None.

Return type

Optional[dict]

get_latest_protocol_by_id(entity_id, max_age_days=30)

Get the latest registry protocol snapshot by Xerberus id.

Parameters

  • entity_id (str) – Xerberus protocol entity id.

  • max_age_days (Optional[int]) – Max age filter.

Returns

Dict of column values, or None.

Return type

Optional[dict]

get_report_url(chain_id, address)

Return a cached report URL for (chain_id, address) if successful.

Parameters

  • chain_id (int) – EIP-155 chain id.

  • address (str) – Vault address.

Returns

URL string or None.

Return type

Optional[str]

get_entity_counts()

Return simple table row counts.

Returns

Dict of count metrics.

Return type

dict[str, int]