apex.metrics

Documentation for eth_defi.apex.metrics Python module.

DuckDB persistence and scan orchestration for ApeX vault metrics.

The reader stores exact source timestamps without resampling. Historical rows are append-and-correct: returned timestamps may be corrected, but timestamps omitted by a later bounded response are never pruned.

Functions

run_scan(session_pool, database, *[, ...])

Run one complete ApeX ranking observation and due history maintenance.

Classes

`ApexConfigurationFetchResult`	Immutable worker result for one public vault configuration request.
`ApexHistoryFetchResult`	Immutable worker result for one history request.
`ApexMetricsDatabase`	Single-owner DuckDB database for ApeX vault metadata and time series.
`ApexScanResult`	Summary of one completed ApeX scan.

class ApexHistoryFetchResult

Bases: object

Immutable worker result for one history request.

vault_id: str: Platform vault identifier.

points: Optional[tuple[eth_defi.apex.vault.ApexHistoryPoint, ...]]: Parsed history, or None on failure.

error: Optional[str]: Error text on failure.

__init__(vault_id, points, error)

Parameters

vault_id (str) –
points (Optional[tuple[eth_defi.apex.vault.ApexHistoryPoint, ...]]) –
error (Optional[str]) –

Return type

None

class ApexConfigurationFetchResult

Bases: object

Immutable worker result for one public vault configuration request.

vault_id: str: Platform vault identifier.

configuration: Optional[eth_defi.apex.vault.ApexVaultConfiguration]: Parsed vault configuration, or None on failure.

error: Optional[str]: Error text on failure.

__init__(vault_id, configuration, error)

Parameters

vault_id (str) –
configuration (Optional[eth_defi.apex.vault.ApexVaultConfiguration]) –
error (Optional[str]) –

Return type

None

class ApexScanResult

Bases: object

Summary of one completed ApeX scan.

observed_at: datetime.datetime: Common ranking observation timestamp.

discovered_vaults: int: Complete source vault count before any target filter.

selected_vaults: int: Vault count selected for ranking storage.

attempted_histories: int: Histories selected by the independent maintenance gate.

successful_histories: int: Successful history responses, including empty responses.

failed_histories: int: Failed history responses.

__init__(observed_at, discovered_vaults, selected_vaults, attempted_histories, successful_histories, failed_histories)

Parameters

observed_at (datetime.datetime) –
discovered_vaults (int) –
selected_vaults (int) –
attempted_histories (int) –
successful_histories (int) –
failed_histories (int) –

Return type

None

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.

__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, redemption_delays=None)

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.
redemption_delays (Optional[dict[str, datetime.timedelta]]) – Fresh public redemption delays keyed by platform vault ID. When a configuration read fails, an existing value is retained instead of replacing verified metadata with an absent value.

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, …]

select_configuration_candidates(vaults, official_vault_ids)

Select vaults whose public redemption delay still needs refreshing.

Official liquidity-provider vaults have no lock-up according to ApeX’s Protocol Vault announcement, so they are excluded. A terminal vault’s configuration cannot affect future subscriptions, so its verified value is retained without another request. Active ranked vaults remain eligible because ApeX may change their per-vault delay.

Parameters

vaults (tuple[eth_defi.apex.vault.ApexVaultSummary, ...]) – Current selected ranking records.
official_vault_ids (set[str]) – IDs sourced from ApeX’s official liquidity-provider endpoint.

Returns

Platform vault IDs requiring a configuration request.

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

run_scan(session_pool, database, *, vault_ids=None, max_workers=8, history_mode='incremental', history_refresh_interval=datetime.timedelta(days=1), ranking_timeout=300.0, history_timeout=120.0)

Run one complete ApeX ranking observation and due history maintenance.

Ranking is always fetched and validated in full before a target filter is applied. The command scheduler owns ranking cadence; this function records all selected non-terminal vaults whenever called. Historical maintenance is independently gated by persisted success timestamps.

Parameters

session_pool (eth_defi.apex.session.ApexSessionPool) – Configured worker-local HTTP session pool.
database (eth_defi.apex.metrics.ApexMetricsDatabase) – Already-open owner-thread database.
vault_ids (Optional[tuple[str, ...]]) – Optional exact targeted IDs.
max_workers (int) – Threaded history reader count.
history_mode (Literal['incremental', 'refresh', 'none']) – incremental, refresh or none.
history_refresh_interval (datetime.timedelta) – Positive independent historical refresh cadence.
ranking_timeout (float) – Whole two-pass ranking deadline in seconds.
history_timeout (float) – Per-vault history and configuration deadline in seconds.

Returns

Typed completed-scan summary.

Return type

eth_defi.apex.metrics.ApexScanResult