BlockTimestampDatabase

Documentation for eth_defi.event_reader.timestamp_cache.BlockTimestampDatabase Python class.

class BlockTimestampDatabase

Bases: object

Mapping of chain ID -> block number -> timestamp using DuckDB.

• Internal storage: DuckDB on-disk database (or in-memory).

• Efficient selective loading and upserting

• One second precision for disk space and speed savings

Modern high-throughput chains, including Monad, can produce multiple blocks during one Unix-timestamp second. Block timestamps are therefore a one-to-many mapping from timestamp to block number: equal timestamp values are expected and must not be used as unique block or observation IDs. One second precision is sufficient for this project’s historical scans, so we deliberately preserve the shared timestamp instead of inventing higher-resolution values.

For usage see eth_defi.event_reader.multicall_timestamp.fetch_block_timestamps_multiprocess_auto_backend

Initialize the database connection.

Parameters

path – Path to the DuckDB file. Use ‘:memory:’ for transient storage.

Methods summary

__init__(chain_id, path)	Initialize the database connection.
close()	Release duckdb resources.
create(chain_id, path)	Create an in-memory instance.
find_gaps()	Find all gaps in the block timestamp database.
get_count()
get_database_file_chain(chain_id[, path])	Get the default database file path for a given chain ID.
get_first_and_last_block()	Get the first and last block numbers we have for a given chain ID.
get_first_block()	Get the first block number we have for a given chain ID.
get_last_block()	Get the last block number we have for a given chain ID.
get_missing_block_numbers(block_numbers)	Return requested block numbers absent from this cache.
get_slicer()
import_chain_data(chain_id, data)	Import data from raw dictionary format to the database.
is_closed()	Check if the database connection is closed.
load(chain_id, path)	Load the database from disk.
query(start_block, end_block)	Get timestamps for a single chain in an inclusive block range.
save()	Force a checkpoint.
to_series()	Get timestamps for a single chain.
transform_time_values(series)	Post-process our raw values from the database to actual time format.}

__init__(chain_id, path)

Initialize the database connection.

Parameters

• path (pathlib.Path) – Path to the DuckDB file. Use ‘:memory:’ for transient storage.

• chain_id (int) –

import_chain_data(chain_id, data)

Import data from raw dictionary format to the database.

• Uses an upsert strategy (ON CONFLICT REPLACE) to ensure latest data is kept.

Parameters

• chain_id (int) – Chain ID for the data being imported.

• data (Union[dict[int, datetime.datetime], pandas.Series]) –

  Mapping of block number (int) to timestamp (datetime).

  Give block number -> unix timestamp pd.Series for max speed.

static get_database_file_chain(chain_id, path=PosixPath('/home/runner/.tradingstrategy/block-timestamp'))

Get the default database file path for a given chain ID.

Parameters

chain_id (int) –

Return type

pathlib.Path

static load(chain_id, path)

Load the database from disk.

Parameters

• chain_id (int) –

• path (pathlib.Path) –

Return type

eth_defi.event_reader.timestamp_cache.BlockTimestampDatabase

static create(chain_id, path)

Create an in-memory instance.

Parameters

• chain_id (int) –

• path (pathlib.Path) –

Return type

eth_defi.event_reader.timestamp_cache.BlockTimestampDatabase

save()

Force a checkpoint.

Note: DuckDB usually auto-commits. If moving from :memory: to disk, we need to copy.

get_first_and_last_block()

Get the first and last block numbers we have for a given chain ID.

Returns

0,0 if no data

Return type

tuple[int, int]

get_first_block()

Get the first block number we have for a given chain ID.

Returns

0 if no data

Return type

int

get_last_block()

Get the last block number we have for a given chain ID.

Returns

0 if no data

Return type

int

to_series()

Get timestamps for a single chain.

Returns a Pandas Series to maintain compatibility with the original API.

Returns

Pandas series block number (int) -> block timestamp (pd.Timestamp)

Return type

Optional[pandas.Series]

query(start_block, end_block)

Get timestamps for a single chain in an inclusive block range.

Returns a Pandas Series to maintain compatibility with the original API.

Parameters

• chain_id – EVM chain id

• start_block (int) – Inclusive start block

• end_block (int) – Inclusive end block

Returns

Pandas series block number (int) -> block timestamp (pd.Timestamp)

Return type

pandas.Series

transform_time_values(series)

Post-process our raw values from the database to actual time format.}

Parameters

series (pandas.Series) – Pandas Series with datetime values

Returns

Pandas Series with integer unix timestamps (seconds)

Return type

pandas.Series

get_missing_block_numbers(block_numbers)

Return requested block numbers absent from this cache.

The lookup is performed as a DuckDB anti-join so sparse historical scans do not need to materialise a multi-million-row cache in Pandas.

Parameters

block_numbers (collections.abc.Iterable[int]) – Exact EVM block numbers needed by a caller.

Returns

Missing block numbers in ascending order.

Return type

list[int]

find_gaps()

Find all gaps in the block timestamp database.

Uses LEAD window function for efficient gap boundary detection without materialising the full expected block range.

Returns

List of (gap_start, gap_end, gap_size) tuples. gap_start is the last present block before the gap, gap_end is the first present block after the gap, gap_size is the number of missing blocks.

Return type

list[tuple[int, int, int]]

close()

Release duckdb resources.

is_closed()

Check if the database connection is closed.

Return type

bool