vault.settlement_data

Documentation for eth_defi.vault.settlement_data Python module.

Vault settlement event storage and price-row annotation helpers.

This module stores sparse asynchronous vault settlement events in a small DuckDB database. The settlement data is intentionally kept separate from the price parquets: raw price rows remain scanner snapshots, while settlement events are merged into the cleaned in-memory price DataFrame during the cleaning pipeline.

Functions

annotate_prices_with_vault_settlements(...)	Annotate cleaned price rows with settlement timestamps.
checkpoint_vault_settlement_database_if_exists([path])	Checkpoint the settlement DuckDB database if it exists.
get_default_vault_settlement_database_path()	Return the default vault settlement DuckDB path.
load_vault_settlements([path])	Load settlement rows from DuckDB if the database exists.
merge_vault_settlements_into_cleaned_prices(...)	Merge settlement timestamps into the cleaned price DataFrame.

Classes

VaultSettlement	One asynchronous vault settlement transaction.
VaultSettlementDatabase	Mini DuckDB interface for vault settlement events.

get_default_vault_settlement_database_path()

Return the default vault settlement DuckDB path.

The path follows the active pipeline data directory so test and production runs can redirect all vault artefacts with PIPELINE_DATA_DIR.

Returns

DuckDB file path.

Return type

pathlib.Path

class VaultSettlement

Bases: object

One asynchronous vault settlement transaction.

Parameters

• chain_id – EVM chain id.

• address – Vault address.

• block_number – Settlement block number.

• protocol – Protocol name, e.g. "Lagoon Finance".

• block_hash – Settlement block hash.

• timestamp – Naive UTC block timestamp.

• tx_hash – Settlement transaction hash.

• event_name – Protocol event name, e.g. "SettleDeposit".

• inserted_at – Naive UTC timestamp when this row was inserted. None uses current time during database insert.

as_db_tuple(inserted_at)

Convert to a DuckDB insert tuple.

Parameters

inserted_at (datetime.datetime) – Insert timestamp to use when this event does not already specify one.

Returns

Tuple matching the vault_settlements table schema.

Return type

tuple

__init__(chain_id, address, block_number, protocol, block_hash, timestamp, tx_hash, event_name='', inserted_at=None)

Parameters

• chain_id (int) –

• address (Union[eth_typing.evm.HexAddress, str]) –

• block_number (int) –

• protocol (str) –

• block_hash (Union[hexbytes.main.HexBytes, str]) –

• timestamp (datetime.datetime) –

• tx_hash (Union[hexbytes.main.HexBytes, str]) –

• event_name (str) –

• inserted_at (Optional[datetime.datetime]) –

Return type

None

class VaultSettlementDatabase

Bases: object

Mini DuckDB interface for vault settlement events.

The schema has no primary key because DuckDB ART indexes have caused stability issues elsewhere in this project. Idempotence is handled with a delete-then-insert transaction keyed by (chain_id, address, tx_hash, event_name). Different transactions in the same block and different settlement events in the same transaction are separate rows.

Parameters

path – DuckDB database path.

Open the database and create the schema if needed.

Parameters

path – DuckDB file path.

__init__(path)

Open the database and create the schema if needed.

Parameters

path (pathlib.Path) – DuckDB file path.

close()

Close the database connection.

Return type

None

save()

Checkpoint the database.

Return type

None

upsert_settlements(settlements)

Insert settlement rows idempotently.

Existing rows with the same (chain_id, address, tx_hash, event_name) are replaced. This lets protocol readers rescan overlapping block ranges without collapsing different settlement transactions in the same block or different event logs in the same transaction.

Parameters

settlements (list[eth_defi.vault.settlement_data.VaultSettlement]) – Settlement rows to store.

Returns

Number of rows inserted.

Return type

int

get_settlements(chain_id=None, address=None, protocol=None)

Read settlement rows as a DataFrame.

Parameters

• chain_id (Optional[int]) – Optional chain filter.

• address (Optional[Union[eth_typing.evm.HexAddress, str]]) – Optional vault address filter.

• protocol (Optional[str]) – Optional protocol filter.

Returns

DataFrame sorted by chain_id, address, timestamp and tx_hash.

Return type

pandas.DataFrame

get_settlement_count()

Return the number of stored settlement rows.

Return type

int

get_latest_block_number(chain_id, address)

Return the latest stored settlement block for a vault.

Parameters

• chain_id (int) – Chain id.

• address (Union[eth_typing.evm.HexAddress, str]) – Vault address.

Returns

Latest block number, or None if no settlement is stored.

Return type

Optional[int]

get_latest_scanned_block_number(chain_id, address)

Return the latest successfully scanned settlement block for a vault.

Settlement events are sparse, so the event table cannot tell whether a vault has no new events or whether it has never been scanned. This scan-state table records successful empty scans as well.

Parameters

• chain_id (int) – Chain id.

• address (Union[eth_typing.evm.HexAddress, str]) – Vault address.

Returns

Latest scanned block number, or None if no scan state exists.

Return type

Optional[int]

upsert_scan_state(scan_states)

Update settlement scan watermarks.

The stored watermark never moves backwards. Forced historical backfills may scan old ranges, but they must not erase knowledge of newer ranges that already completed successfully.

Parameters

scan_states (list[tuple[int, eth_typing.evm.HexAddress | str, int]]) – Tuples of (chain_id, address, last_scanned_block).

Returns

Number of distinct vault scan-state rows updated.

Return type

int

load_vault_settlements(path=None)

Load settlement rows from DuckDB if the database exists.

Parameters

path (Optional[pathlib.Path]) – Optional DuckDB path. None resolves to the default pipeline path.

Returns

Settlement DataFrame. Empty when the database does not exist.

Return type

pandas.DataFrame

checkpoint_vault_settlement_database_if_exists(path=None)

Checkpoint the settlement DuckDB database if it exists.

DuckDB can keep recent writes in a WAL file while a connection is open. Backup and export code copies only vault-settlements.duckdb, so make sure the main database file is self-contained before copying it.

Parameters

path (Optional[pathlib.Path]) – Optional DuckDB path. None resolves to the default pipeline path.

Returns

True if an existing database was checkpointed, False if there was no database file.

Return type

bool

annotate_prices_with_vault_settlements(prices_df, settlements_df)

Annotate cleaned price rows with settlement timestamps.

For each (chain, address) group, a price row receives the latest settlement timestamp in (previous_price_timestamp, current_timestamp]. The first row receives the latest settlement timestamp up to its timestamp.

The production cleaning pipeline calls this after row-level cleaning, so raw price parquet rows remain settlement-free.

Parameters

• prices_df (pandas.DataFrame) – Cleaned vault prices DataFrame. Must contain chain, address and timestamp columns, or use a datetime index for timestamps.

• settlements_df (pandas.DataFrame) – Settlement DataFrame from VaultSettlementDatabase.

Returns

Copy of prices_df with vault_settlement_at populated.

Return type

pandas.DataFrame

merge_vault_settlements_into_cleaned_prices(cleaned_prices_df, settlement_db_path=None)

Merge settlement timestamps into the cleaned price DataFrame.

The raw vault-prices-1h.parquet file is not rewritten with settlement markers. Settlement events are stored in vault-settlements.duckdb and applied here to the cleaned in-memory price frame before cleaned-vault-prices-1h.parquet is written.

Parameters

• cleaned_prices_df (pandas.DataFrame) – Cleaned scanner price DataFrame produced by the vault price cleaning pipeline.

• settlement_db_path (Optional[pathlib.Path]) – Optional settlement DuckDB path.

Returns

Cleaned price DataFrame with vault_settlement_at added.

Return type

pandas.DataFrame