ApexMetricsDatabase

Documentation for eth_defi.apex.metrics.ApexMetricsDatabase Python class.

class ApexMetricsDatabase

Bases: object

Single-owner DuckDB database for ApeX vault metadata and time series.

The three tables use application-enforced logical keys and deliberately have no primary or unique constraints. This avoids DuckDB ART indexes on affected Python 3.14/macOS ARM64 environments.

vault_metadata columns contain one current record per vault_id. vault_prices contains actual naive UTC timestamps, source DOUBLE values and the source discriminator. history_sync tracks response and canonical retained-history bounds separately.

Parameters

path – File-backed DuckDB path.

Open or create an owner-thread ApeX metrics database.

The parent directory is created as needed, automatic WAL checkpoints are disabled and the forward-compatible schema is initialised.

Parameters

path – File-backed DuckDB path.

Methods summary

__init__([path])	Open or create an owner-thread ApeX metrics database.
apply_history_success(vault_id, points, ...)	Atomically append/correct one history and update its sync state.
apply_ranking(vaults, observed_at, *, ...)	Atomically store ranking metadata, observations and lifecycle state.
checkpoint()	Run one explicit file-backed database checkpoint.
close()	Close the owner-thread database connection.
get_history_sync()	Return history maintenance state ordered by vault ID.
get_price_count()	Return the number of retained ApeX price observations.
get_vault_metadata()	Return current metadata ordered by vault ID.
get_vault_prices([vault_id])	Return actual-timestamp price rows.
record_history_error(vault_id, error, ...)	Record one retryable history API failure without touching prices.
select_history_candidates(present_vault_ids, ...)	Select histories due under the independent maintenance gate.

__init__(path=PosixPath('/home/runner/.tradingstrategy/vaults/apex-vaults.duckdb'))

Open or create an owner-thread ApeX metrics database.

The parent directory is created as needed, automatic WAL checkpoints are disabled and the forward-compatible schema is initialised.

Parameters

path (pathlib.Path) – File-backed DuckDB path.

Return type

None

apply_ranking(vaults, observed_at, *, manage_disappearance)

Atomically store ranking metadata, observations and lifecycle state.

Every supplied vault is present and selected. An unfiltered scan passes manage_disappearance=True so previously known absent vaults start a missing generation. Any supplied present vault clears its old missing generation, including during targeted scans.

Parameters

• vaults (tuple[eth_defi.apex.vault.ApexVaultSummary, ...]) – Selected ranking records from the stabilised second pass.

• observed_at (datetime.datetime) – Common naive UTC ranking observation timestamp.

• manage_disappearance (bool) – Whether absent known vaults should be marked missing.

Returns

None.

Return type

None

select_history_candidates(present_vault_ids, now, *, mode, refresh_interval, include_missing)

Select histories due under the independent maintenance gate.

Present, terminal and disappeared vault generations use separate persisted success markers so an empty response remains retryable where finalisation requires data.

Parameters

• present_vault_ids (set[str]) – IDs present in the selected stabilised ranking.

• now (datetime.datetime) – Naive UTC eligibility timestamp.

• mode (Literal['incremental', 'refresh', 'none']) – Incremental, forced refresh or disabled history mode.

• refresh_interval (datetime.timedelta) – Positive non-terminal history refresh interval.

• include_missing (bool) – Whether disappeared vault generations may be selected.

Returns

Sorted due vault IDs.

Return type

tuple[str, …]

apply_history_success(vault_id, points, attempted_at)

Atomically append/correct one history and update its sync state.

Only timestamps returned by the source are replaced. Omitted existing timestamps remain intact, and the canonical retained range is updated in the same transaction as attempt and lifecycle state.

Parameters

• vault_id (str) – Existing ApeX platform vault ID.

• points (tuple[eth_defi.apex.vault.ApexHistoryPoint, ...]) – Fully parsed source history, possibly empty.

• attempted_at (datetime.datetime) – Naive UTC completion timestamp.

Returns

None.

Return type

None

record_history_error(vault_id, error, attempted_at)

Record one retryable history API failure without touching prices.

The isolated state transaction preserves every retained price row and non-empty response diagnostic.

Parameters

• vault_id (str) – Existing ApeX platform vault ID.

• error (str) – Human-readable bounded API failure.

• attempted_at (datetime.datetime) – Naive UTC failure timestamp.

Returns

None.

Return type

None

checkpoint()

Run one explicit file-backed database checkpoint.

Automatic WAL checkpoints are disabled for this database, so callers invoke this once after a complete successful scan.

Returns

None.

Return type

None

close()

Close the owner-thread database connection.

The same thread that created the database must close it after all writes and the final checkpoint have completed.

Returns

None.

Return type

None

get_vault_metadata()

Return current metadata ordered by vault ID.

Returns

Dataframe containing all vault_metadata columns.

Return type

pandas.DataFrame

get_price_count()

Return the number of retained ApeX price observations.

This scalar query avoids materialising the complete price history when an orchestration caller only needs a progress metric.

Returns

Number of rows in vault_prices.

Return type

int

get_vault_prices(vault_id=None)

Return actual-timestamp price rows.

Parameters

vault_id (Optional[str]) – Optional platform vault identifier.

Returns

Dataframe containing all vault_prices columns.

Return type

pandas.DataFrame

get_history_sync()

Return history maintenance state ordered by vault ID.

Returns

Dataframe containing all history_sync columns.

Return type

pandas.DataFrame