provider.rpcdb

Documentation for eth_defi.provider.rpcdb Python module.

JSON-RPC request accounting and DuckDB persistence.

A caller creates one RPCRequestStats for a phase, passes it to Web3 providers, and persists the aggregate with RPCUsageDatabase.

Module Attributes

`RPC_TRACKING_DATABASE_PATH_ENV`	Environment variable overriding `DEFAULT_RPC_TRACKING_DATABASE`.
`ZERO_CALL_MARKER`	Marker value used to preserve a completed zero-call scan iteration.

Functions

`format_rpc_usage_report`(database, chain, ...)	Format current-cycle and daily JSON-RPC usage for one chain.
`normalise_rpc_error`(error)	Convert heterogeneous provider failures to stable aggregation values.
`resolve_rpc_tracking_database_path`()	Resolve the shared JSON-RPC tracking DuckDB path.

Classes

`RPCRequestStats`	Thread-safe, pickle-safe physical JSON-RPC request counters.
`RPCUsageDatabase`	Append-only DuckDB storage for JSON-RPC scan accounting.

DEFAULT_RPC_TRACKING_DATABASE = PosixPath('/home/runner/.tradingstrategy/rpc-tracking.duckdb'): Default shared DuckDB path for JSON-RPC request accounting.

RPC_TRACKING_DATABASE_PATH_ENV = 'RPC_TRACKING_DATABASE_PATH': Environment variable overriding DEFAULT_RPC_TRACKING_DATABASE.

ZERO_CALL_MARKER = 'none': Marker value used to preserve a completed zero-call scan iteration.

resolve_rpc_tracking_database_path()

Resolve the shared JSON-RPC tracking DuckDB path.

The resolver follows the same convention as other repository DuckDB modules: a module-level default below ~/.tradingstrategy and an environment-variable override with user-home expansion.

Returns: Expanded path from RPC_TRACKING_DATABASE_PATH or the default path.
Return type: pathlib.Path

normalise_rpc_error(error)

Convert heterogeneous provider failures to stable aggregation values.

JSON-RPC response dictionaries use their numeric error code. HTTP failures use http_<status>. Other Python failures use the concrete exception class name. The provider’s original error message is retained.

Parameters: error (Union[BaseException, dict[str, Any]]) – A JSON-RPC error dictionary or an exception raised by the provider.
Returns: (error_code, error_message) suitable for RPCRequestStats.record_error().
Return type: tuple[str, str]

class RPCRequestStats

Bases: object

Thread-safe, pickle-safe physical JSON-RPC request counters.

calls is keyed by (rpc_provider_domain, api_call) and errors by (rpc_provider_domain, error_code, error_message). The lock is excluded from pickle state and recreated in subprocesses.

calls: collections.Counter[tuple[str, str]]: Provider-domain and JSON-RPC method counts.

errors: collections.Counter[tuple[str, str, str]]: Provider-domain, error-code and error-message counts.

record_call(rpc_provider_domain, api_call, count=1)

Record physical JSON-RPC request attempts.

Parameters

rpc_provider_domain (str) – Provider hostname, optionally including a non-default port.
api_call (str) – JSON-RPC method name such as eth_call.
count (int) – Positive number of attempts to add.

Return type

None

record_error(rpc_provider_domain, error_code, error_message, count=1)

Record JSON-RPC request failures.

Parameters

rpc_provider_domain (str) – Provider hostname, optionally including a non-default port.
error_code (str) – Stable JSON-RPC, HTTP, or exception-class error code.
error_message (str) – Provider error message.
count (int) – Positive number of matching failures to add.

Return type

None

merge(other)

Merge another worker or phase aggregate exactly once.

Parameters: other (eth_defi.provider.rpcdb.RPCRequestStats) – Detached task statistics to add to this accumulator.
Return type: None

export()

Take a detached copy of both counter mappings.

Returns: Copied call and error counters safe to iterate without holding the accumulator lock.
Return type: tuple[collections.Counter[tuple[str, str]], collections.Counter[tuple[str, str, str]]]

__init__(calls=<factory>, errors=<factory>)

Parameters

calls (collections.Counter[tuple[str, str]]) –
errors (collections.Counter[tuple[str, str, str]]) –

Return type

None

class RPCUsageDatabase

Bases: object

Append-only DuckDB storage for JSON-RPC scan accounting.

One connection belongs to one externally serialised writer. Callers sharing a database across processes must hold their pipeline lock for the complete connection lifetime.

Open the tracking database and initialise its fixed schema.

Parameters: path – DuckDB file path. Parent directories are created automatically.

__init__(path)

Open the tracking database and initialise its fixed schema.

Parameters: path (pathlib.Path) – DuckDB file path. Parent directories are created automatically.
Return type: None

allocate_cycle()

Allocate the next persistent cycle number.

Allocation uses both tables and must be called while the external pipeline-writer lock is held. A crash before the first row is inserted may reuse the unpersisted number on the next invocation.

Returns: Next positive cycle number.
Return type: int

record_scan(chain, phase, cycle_started, cycle_number, stats, items_scanned)

Append one completed scan-attempt aggregate atomically.

Call and error rows are committed in the same transaction. An empty call aggregate writes a zero-count marker so the scan iteration and its item count remain visible. Unknown item counts on early failures should be passed as zero.

Parameters

chain (int) – EVM chain id.
phase (str) – Caller-defined scan phase, for example lead_discovery.
cycle_started (datetime.date) – Naive UTC calendar date on which the logical cycle started.
cycle_number (int) – Persistent cycle identifier shared by scanner retries.
stats (eth_defi.provider.rpcdb.RPCRequestStats) – Physical request and error counters for this attempt only.
items_scanned (int) – Non-negative number of logical items submitted during the attempt.

Return type

None

fetch_cycle_calls(chain, cycle_started, cycle_number)

Fetch current-cycle method rows for one chain.

Parameters

chain (int) – EVM chain id.
cycle_started (datetime.date) – UTC cycle date.
cycle_number (int) – Persistent cycle number.

Returns

Rows of (phase, provider_domain, api_call, call_count, items_scanned) aggregated across retry attempts.

Return type

list[tuple[str, str, str, int, int]]

fetch_daily_totals(chain, cycle_started)

Fetch daily provider totals without multiplying retry item counts.

Each cycle contributes the sum of its method rows and the maximum item count reported by any retry. Provider rows display provider-specific calls with the cycle-level item denominator.

Returns

Rows of (phase, provider_domain, call_count, items_scanned).

Parameters

chain (int) –
cycle_started (datetime.date) –

Return type

list[tuple[str, str, int, int]]

fetch_cycle_errors(chain, cycle_started, cycle_number)

Fetch current-cycle error totals for one chain.

Returns

Rows of (phase, provider_domain, error_code, error_message, error_count).

Parameters

chain (int) –
cycle_started (datetime.date) –
cycle_number (int) –

Return type

list[tuple[str, str, str, str, int]]

close()

Checkpoint and close the DuckDB connection explicitly.

Return type: None

format_rpc_usage_report(database, chain, cycle_started, cycle_number)

Format current-cycle and daily JSON-RPC usage for one chain.

The formatter is output-system agnostic: scanners may print the returned text, log it, or embed it in another report. Daily rows keep lead discovery and price scanning separate through their phase column.

Parameters

database (eth_defi.provider.rpcdb.RPCUsageDatabase) – Open tracking database.
chain (int) – EVM chain id to report.
cycle_started (datetime.date) – UTC date of the current logical cycle.
cycle_number (int) – Current persistent cycle number.

Returns

Multi-table plain-text report.

Return type

str