diff --git a/SPEC.md b/SPEC.md index a8c38c6c3e..0463888519 100644 --- a/SPEC.md +++ b/SPEC.md @@ -1804,6 +1804,32 @@ If the positions fail to load, the last published prices stay. in. It is not the corridor's trigger ratio, which uses the vault total, in flight included. +Every 60 seconds a second task publishes the performance families from the same +loaders as the `/performance/*` endpoints, over the last 24 hours: + +- `latencies`: `liq_hedge_latency_ms{stage,quantile}` (nearest-rank `p50`, + `p90`, `p95`, `p99` and `max`; a stage with no samples has no series), + `liq_hedge_latency_ms_samples{stage}`, and per open exposure + `liq_open_exposure_fill_count{symbol}` and + `liq_open_exposure_oldest_ts_seconds{symbol}`. +- `reliability`: `liq_failure_event_count_24h{event_type}` and + `liq_job_queue{job_type,state}`. The job queue counts are the queue now, not a + window. +- `infra`: `liq_block_lag_blocks{chain}`, + `liq_block_lag_sampled_ts_seconds{chain}`, + `liq_poll_{cycles,errors,skipped_ticks}_24h{chain}`, + `liq_poll_duration_ms{chain,quantile}`, and + `liq_dependency_{calls_24h,errors_24h}{dependency,operation}` with + `liq_dependency_latency_ms{dependency,operation,quantile}`. The block-lag and + poll series carry one `chain` per hedged chain; the exporter's unlabelled + versions came from a report shape the bot no longer serves. + +Every fifth cycle the task publishes `rebalances` over the last 30 days: +`liq_rebalance_stage_ms{kind,stage,quantile}` (`kind` is `usdc` or `equity`) and +`liq_attestation_last_ms{kind="usdc"}`, absent with no attestation in the +window. A collector that fails or takes longer than 30 seconds keeps its last +family. + **Exception to financial-integrity rules.** Some ported series reproduce the exporter's published sentinels: a ratio with a zero denominator is published as 0, and a missing balance component counts as 0. These follow the exporter's diff --git a/docs/observability.md b/docs/observability.md index 3910f68dcc..f9249b9c1e 100644 --- a/docs/observability.md +++ b/docs/observability.md @@ -149,55 +149,73 @@ sample time, so only `outcome="paused"` shows that stall. | `order_fill_poll_duration_seconds` | histogram | `chain` | each order-fill poll cycle | | `order_fill_block_lag_blocks` | gauge | `chain` | each poll that knows the cutoff block and the checkpoint | | `order_fill_block_lag_sampled_timestamp_seconds` | gauge | `chain` | same, set to the poll's sample time | +| `metrics_refresh_duration_seconds` | histogram | `collector` | each performance collector run, failed or not | ### Catalog -| Name | Labels | Family | Published | -| --------------------------------------- | -------------------------------------------------------------------------------------------------------------------------- | ----------- | -------------------------------------------------------------------------------------------------- | -| `liq_bot_info` | `git_commit` | `health` | `1`; first 12 characters of the build commit (`dev` in local builds) | -| `liq_bot_start_timestamp_seconds` | | `health` | Unix time the process started | -| `liq_settings_info` | `broker`, `log_level`, `orderbook`, `server_port`, `trading_mode`, `turnkey_organization`, `wallet_address`, `wallet_kind` | `settings` | `1`; a missing wallet or organization gives `""`, port 0 gives `""`, `trading_mode` is always `""` | -| `liq_settings_equity_target` | | `settings` | primary chain default target share; absent with only per-symbol targets | -| `liq_settings_equity_deviation` | | `settings` | always | -| `liq_settings_usdc_target` | | `settings` | the one active corridor's target (the primary chain's with several); absent without one | -| `liq_settings_usdc_deviation` | | `settings` | as above | -| `liq_settings_cash_reserved` | | `settings` | absent when not configured | -| `liq_settings_execution_threshold_usd` | | `settings` | dollar threshold; absent for a share-count threshold | -| `liq_settings_order_polling_seconds` | | `settings` | always | -| `liq_settings_inventory_poll_seconds` | | `settings` | always | -| `liq_settings_deployment_block` | | `settings` | always | -| `liq_asset_counter_trading` | `symbol` | `settings` | `1` or `0`, every symbol the primary chain lists | -| `liq_asset_extended_hours` | `symbol` | `settings` | `1` or `0`; absent while counter trading is disabled | -| `liq_asset_rebalancing` | `symbol` | `settings` | `1` when the symbol starts new rebalancing operations | -| `liq_usdc_corridor_target` | `chain` | `settings` | every configured USDC corridor, whatever the USDC mode | -| `liq_usdc_corridor_deviation` | `chain` | `settings` | as above | -| `liq_equity_onchain_available` | `symbol` | `inventory` | primary chain vault | -| `liq_equity_offchain_available` | `symbol` | `inventory` | broker | -| `liq_equity_inflight_total` | `symbol` | `inventory` | primary chain vault plus broker in flight | -| `liq_equity_total` | `symbol` | `inventory` | onchain plus offchain plus in flight; excludes wallet tokens and other chains | -| `liq_equity_unwrapped` | `symbol` | `inventory` | Base wallet unwrapped tokens | -| `liq_equity_wrapped` | `symbol` | `inventory` | Base wallet wrapped tokens | -| `liq_equity_ratio` | `symbol` | `inventory` | onchain over onchain plus offchain; `0` when both are 0 | -| `liq_equity_chain_available` | `chain`, `symbol` | `inventory` | each chain vault a snapshot read, in its wrapped shares: never sum across chains or with broker | -| `liq_usdc_onchain_available` | | `inventory` | primary chain vault | -| `liq_usdc_onchain_inflight` | | `inventory` | primary chain vault in flight | -| `liq_usdc_offchain_available` | | `inventory` | broker cash after the reserve | -| `liq_usdc_offchain_gross` | | `inventory` | broker cash before the reserve; absent until read | -| `liq_usdc_offchain_inflight` | | `inventory` | broker cash in flight | -| `liq_usdc_alpaca_usdc` | | `inventory` | USDC token in the broker account; absent until read | -| `liq_usdc_alpaca_total` | | `inventory` | gross broker cash, or available broker cash until the gross is read | -| `liq_usdc_inflight_total` | | `inventory` | vault plus broker in flight | -| `liq_usdc_inflight_ethereum_wallet` | | `inventory` | absent until read | -| `liq_usdc_inflight_base_wallet` | | `inventory` | absent until read | -| `liq_usdc_total` | | `inventory` | vault plus broker total plus in flight | -| `liq_usdc_ratio` | | `inventory` | vault over vault plus broker total; `0` when both are 0 | -| `liq_usdc_rebalanceable` | | `inventory` | withdrawable minus the reserve, never below 0; absent until withdrawable is read | -| `liq_usdc_chain_available` | `chain` | `inventory` | each hedged chain vault a snapshot has read | -| `liq_usdc_chain_inflight` | `chain` | `inventory` | as above | -| `liq_usdc_chain_ratio` | `chain` | `inventory` | vault available over itself plus gross broker cash; absent until the gross is read or both are 0 | -| `liq_position_last_price_usd` | `symbol` | `prices` | live price of each symbol with a position; refreshed every 60 s | -| `liq_equity_exposure_usd` | `symbol` | `prices` | net position times that price | -| `liq_collector_last_success_ts_seconds` | `collector` | store | Unix time each family was last published | +| Name | Labels | Family | Published | +| --------------------------------------- | -------------------------------------------------------------------------------------------------------------------------- | ------------- | -------------------------------------------------------------------------------------------------- | +| `liq_bot_info` | `git_commit` | `health` | `1`; first 12 characters of the build commit (`dev` in local builds) | +| `liq_bot_start_timestamp_seconds` | | `health` | Unix time the process started | +| `liq_settings_info` | `broker`, `log_level`, `orderbook`, `server_port`, `trading_mode`, `turnkey_organization`, `wallet_address`, `wallet_kind` | `settings` | `1`; a missing wallet or organization gives `""`, port 0 gives `""`, `trading_mode` is always `""` | +| `liq_settings_equity_target` | | `settings` | primary chain default target share; absent with only per-symbol targets | +| `liq_settings_equity_deviation` | | `settings` | always | +| `liq_settings_usdc_target` | | `settings` | the one active corridor's target (the primary chain's with several); absent without one | +| `liq_settings_usdc_deviation` | | `settings` | as above | +| `liq_settings_cash_reserved` | | `settings` | absent when not configured | +| `liq_settings_execution_threshold_usd` | | `settings` | dollar threshold; absent for a share-count threshold | +| `liq_settings_order_polling_seconds` | | `settings` | always | +| `liq_settings_inventory_poll_seconds` | | `settings` | always | +| `liq_settings_deployment_block` | | `settings` | always | +| `liq_asset_counter_trading` | `symbol` | `settings` | `1` or `0`, every symbol the primary chain lists | +| `liq_asset_extended_hours` | `symbol` | `settings` | `1` or `0`; absent while counter trading is disabled | +| `liq_asset_rebalancing` | `symbol` | `settings` | `1` when the symbol starts new rebalancing operations | +| `liq_usdc_corridor_target` | `chain` | `settings` | every configured USDC corridor, whatever the USDC mode | +| `liq_usdc_corridor_deviation` | `chain` | `settings` | as above | +| `liq_equity_onchain_available` | `symbol` | `inventory` | primary chain vault | +| `liq_equity_offchain_available` | `symbol` | `inventory` | broker | +| `liq_equity_inflight_total` | `symbol` | `inventory` | primary chain vault plus broker in flight | +| `liq_equity_total` | `symbol` | `inventory` | onchain plus offchain plus in flight; excludes wallet tokens and other chains | +| `liq_equity_unwrapped` | `symbol` | `inventory` | Base wallet unwrapped tokens | +| `liq_equity_wrapped` | `symbol` | `inventory` | Base wallet wrapped tokens | +| `liq_equity_ratio` | `symbol` | `inventory` | onchain over onchain plus offchain; `0` when both are 0 | +| `liq_equity_chain_available` | `chain`, `symbol` | `inventory` | each chain vault a snapshot read, in its wrapped shares: never sum across chains or with broker | +| `liq_usdc_onchain_available` | | `inventory` | primary chain vault | +| `liq_usdc_onchain_inflight` | | `inventory` | primary chain vault in flight | +| `liq_usdc_offchain_available` | | `inventory` | broker cash after the reserve | +| `liq_usdc_offchain_gross` | | `inventory` | broker cash before the reserve; absent until read | +| `liq_usdc_offchain_inflight` | | `inventory` | broker cash in flight | +| `liq_usdc_alpaca_usdc` | | `inventory` | USDC token in the broker account; absent until read | +| `liq_usdc_alpaca_total` | | `inventory` | gross broker cash, or available broker cash until the gross is read | +| `liq_usdc_inflight_total` | | `inventory` | vault plus broker in flight | +| `liq_usdc_inflight_ethereum_wallet` | | `inventory` | absent until read | +| `liq_usdc_inflight_base_wallet` | | `inventory` | absent until read | +| `liq_usdc_total` | | `inventory` | vault plus broker total plus in flight | +| `liq_usdc_ratio` | | `inventory` | vault over vault plus broker total; `0` when both are 0 | +| `liq_usdc_rebalanceable` | | `inventory` | withdrawable minus the reserve, never below 0; absent until withdrawable is read | +| `liq_usdc_chain_available` | `chain` | `inventory` | each hedged chain vault a snapshot has read | +| `liq_usdc_chain_inflight` | `chain` | `inventory` | as above | +| `liq_usdc_chain_ratio` | `chain` | `inventory` | vault available over itself plus gross broker cash; absent until the gross is read or both are 0 | +| `liq_position_last_price_usd` | `symbol` | `prices` | live price of each symbol with a position; refreshed every 60 s | +| `liq_equity_exposure_usd` | `symbol` | `prices` | net position times that price | +| `liq_hedge_latency_ms` | `quantile`, `stage` | `latencies` | 24 h nearest-rank `p50`, `p90`, `p95`, `p99`, `max`; absent for a stage without samples | +| `liq_hedge_latency_ms_samples` | `stage` | `latencies` | samples behind each stage | +| `liq_open_exposure_fill_count` | `symbol` | `latencies` | fills after the symbol's latest hedge placement | +| `liq_open_exposure_oldest_ts_seconds` | `symbol` | `latencies` | block time of the oldest unhedged fill | +| `liq_failure_event_count_24h` | `event_type` | `reliability` | lifecycle failure events in 24 h | +| `liq_job_queue` | `job_type`, `state` | `reliability` | queue counts now, not windowed | +| `liq_block_lag_blocks` | `chain` | `infra` | latest sampled lag of each hedged chain; absent until known | +| `liq_block_lag_sampled_ts_seconds` | `chain` | `infra` | time of that sample | +| `liq_poll_cycles_24h` | `chain` | `infra` | order-fill poll cycles in 24 h | +| `liq_poll_errors_24h` | `chain` | `infra` | failed cycles in 24 h | +| `liq_poll_skipped_ticks_24h` | `chain` | `infra` | dropped poll ticks in 24 h | +| `liq_poll_duration_ms` | `chain`, `quantile` | `infra` | 24 h poll duration; absent with no cycles | +| `liq_dependency_calls_24h` | `dependency`, `operation` | `infra` | external calls in 24 h | +| `liq_dependency_errors_24h` | `dependency`, `operation` | `infra` | failed calls in 24 h | +| `liq_dependency_latency_ms` | `dependency`, `operation`, `quantile` | `infra` | 24 h call latency | +| `liq_rebalance_stage_ms` | `kind`, `quantile`, `stage` | `rebalances` | 30 d completed stage duration; refreshed every 5 min | +| `liq_attestation_last_ms` | `kind` | `rebalances` | latest CCTP attestation in 30 d; absent with none | +| `liq_collector_last_success_ts_seconds` | `collector` | store | Unix time each family was last published | `symbol` labels drop a leading `wt` or `t` only before an uppercase letter (`tAAPL` and `wtAAPL` give `AAPL`; `tsla` stays). diff --git a/scripts/liq-parity/compare.py b/scripts/liq-parity/compare.py index d5034db189..de6b49b78f 100644 --- a/scripts/liq-parity/compare.py +++ b/scripts/liq-parity/compare.py @@ -92,6 +92,23 @@ "liq_usdc_rebalanceable", "liq_position_last_price_usd", "liq_equity_exposure_usd", + "liq_hedge_latency_ms", + "liq_hedge_latency_ms_samples", + "liq_open_exposure_fill_count", + "liq_open_exposure_oldest_ts_seconds", + "liq_failure_event_count_24h", + "liq_job_queue", + "liq_block_lag_blocks", + "liq_block_lag_sampled_ts_seconds", + "liq_poll_cycles_24h", + "liq_poll_errors_24h", + "liq_poll_skipped_ticks_24h", + "liq_poll_duration_ms", + "liq_dependency_calls_24h", + "liq_dependency_errors_24h", + "liq_dependency_latency_ms", + "liq_rebalance_stage_ms", + "liq_attestation_last_ms", } # Names only the bot publishes. @@ -163,6 +180,22 @@ KNOWN_DIFFS = { # The exporter derives the start from integer uptime at poll time. "liq_bot_start_timestamp_seconds": ("absolute", 2.0), + # The exporter's infra collector has failed on every cycle since the bot + # began serving /performance/infra per chain: t0.devops 226c029 + # exporter.py collect_infra calls poll.get() on the per-chain `poll` + # list, guarded() swallows the AttributeError, and the infra family is + # never set. So the exporter has no infra series to compare with, and + # the bot's carry a `chain` label the exporter never had. The golden + # tests pin the dependency series against the exporter instead. + "liq_block_lag_blocks": "ignore", + "liq_block_lag_sampled_ts_seconds": "ignore", + "liq_poll_cycles_24h": "ignore", + "liq_poll_errors_24h": "ignore", + "liq_poll_skipped_ticks_24h": "ignore", + "liq_poll_duration_ms": "ignore", + "liq_dependency_calls_24h": "ignore", + "liq_dependency_errors_24h": "ignore", + "liq_dependency_latency_ms": "ignore", # The endpoint the exporter reads truncates at 50,000 entries and counts # an exact window; the bot counts every entry in one-minute buckets. "liq_reliability_log_count_24h": "values", diff --git a/scripts/liq-parity/golden.py b/scripts/liq-parity/golden.py index 8db19b91d8..e965b99c22 100644 --- a/scripts/liq-parity/golden.py +++ b/scripts/liq-parity/golden.py @@ -47,12 +47,47 @@ def run_full_state(exporter, fixture): exporter.collect_state() +def serve(fixture_by_path): + """A bot_get stub that answers each path from the fixture.""" + def bot_get(path, params=None, timeout=15): + return fixture_by_path[path] + return bot_get + + +def run_latencies(exporter, fixture): + exporter.bot_get = serve({"/performance/latencies": fixture}) + exporter.collect_latencies("from", "to") + + +def run_reliability(exporter, fixture): + exporter.bot_get = serve({"/performance/reliability": fixture}) + exporter.collect_reliability("from", "to") + + +def run_infra(exporter, fixture): + exporter.bot_get = serve({"/performance/infra": fixture}) + exporter.collect_infra("from", "to") + + +def run_rebalance(exporter, fixture): + """The fixture holds both reports: {"usdc": ..., "equity": ...}.""" + exporter.bot_get = serve({ + "/performance/rebalances": fixture["usdc"], + "/performance/equity-rebalances": fixture["equity"], + }) + exporter.collect_rebalance_timings("from", "to") + + # case name -> how the fixture reaches the exporter CASES = { "settings": run_state, "settings-minimal": run_state, "state": run_full_state, "state-reserved": run_full_state, + "latencies": run_latencies, + "reliability": run_reliability, + "infra": run_infra, + "rebalance": run_rebalance, } diff --git a/scripts/liq-parity/test_compare.py b/scripts/liq-parity/test_compare.py index d507441b69..6e35b00ed2 100644 --- a/scripts/liq-parity/test_compare.py +++ b/scripts/liq-parity/test_compare.py @@ -41,7 +41,7 @@ def test_reports_missing_extra_and_value_findings_after_drop_list_and_known_diff 'missing from bot: liq_asset_rebalancing{symbol="TSLA"}\n' "value differs: liq_settings_usdc_target bot=0.45 exporter=0.4\n" "not yet ported (exporter-only names, ignored): " - "liq_block_lag_blocks, liq_hedge_latency_ms_samples\n" + "liq_pending_orders_total, liq_pnl_summary_usd\n" "3 finding(s)\n" )) @@ -63,8 +63,8 @@ def test_without_known_diffs_the_start_time_jitter_and_the_gauge_types_are_findi "bot=1700000000.0 exporter=1700000001.9\n" "value differs: liq_settings_usdc_target bot=0.45 exporter=0.4\n" "not yet ported (exporter-only names, ignored): " - "liq_asset_flags, liq_block_lag_blocks, liq_equity_ratio_deviation, " - "liq_hedge_latency_ms_samples, liq_up\n" + "liq_asset_flags, liq_equity_ratio_deviation, liq_pending_orders_total, " + "liq_pnl_summary_usd, liq_up\n" "10 finding(s)\n" )) @@ -246,14 +246,14 @@ def test_a_legitimately_absent_series_does_not_make_a_snapshot_unusable(self): []) def test_a_bot_name_published_in_one_pair_is_compared_in_every_pair(self): - exporter = "liq_bot_info{git_commit=\"a\"} 1\nliq_job_queue{state=\"x\"} 2\n" + exporter = "liq_bot_info{git_commit=\"a\"} 1\nliq_pending_orders_total 2\n" with_name = exporter without_name = "liq_bot_info{git_commit=\"a\"} 1\n" findings, not_ported = compare.compare( [(with_name, exporter), (without_name, exporter)], True, True) - self.assertEqual(findings, ["unlisted bot name: liq_job_queue"]) + self.assertEqual(findings, ["unlisted bot name: liq_pending_orders_total"]) self.assertEqual(not_ported, []) def test_a_bot_name_outside_the_lists_is_a_finding(self): diff --git a/scripts/liq-parity/testdata/exporter.prom b/scripts/liq-parity/testdata/exporter.prom index b594a2e7ac..e3eb817c73 100644 --- a/scripts/liq-parity/testdata/exporter.prom +++ b/scripts/liq-parity/testdata/exporter.prom @@ -8,5 +8,5 @@ liq_asset_rebalancing{symbol="TSLA"} 0.0 liq_asset_flags{counter_trading="enabled",extended_hours="true",rebalancing="true",symbol="AAPL"} 1.0 liq_equity_total{symbol="AAPL"} 5.0 liq_equity_ratio_deviation{symbol="AAPL"} 0.25 -liq_hedge_latency_ms_samples{stage="detection"} 5.0 -liq_block_lag_blocks{chain="base"} 3.0 +liq_pnl_summary_usd{stream="total",window="1d"} 5.0 +liq_pending_orders_total 3.0 diff --git a/src/lib.rs b/src/lib.rs index 0702c603ae..b71c373278 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -410,6 +410,11 @@ async fn run_bot_session_inner( pool: pools.cqrs.clone(), families: &metrics::liquidity::LIQ_FAMILIES, }, + metrics::liquidity::refresh::LiqPerformanceRefresh { + pool: pools.cqrs.clone(), + chains: ctx.chains.clone(), + families: &metrics::liquidity::LIQ_FAMILIES, + }, ); let mut bot_task = tokio::spawn(Box::pin(run_conductor_session( ctx.clone(), @@ -751,11 +756,12 @@ fn spawn_server_supervisor( /// The supervisor for tasks that must never stop trading: the dashboard /// price feed (when pricing is configured), the metrics recorder upkeep and -/// the `liq_*` refresh. +/// the `liq_*` refreshes. fn spawn_auxiliary_supervisor( equity_price_task: Option>, metrics_handle: PrometheusHandle, liq_state_refresh: metrics::liquidity::refresh::LiqStateRefresh, + liq_performance_refresh: metrics::liquidity::refresh::LiqPerformanceRefresh, ) -> SupervisorHandle { let builder = non_escalating_supervisor_builder() .with_task( @@ -764,7 +770,8 @@ fn spawn_auxiliary_supervisor( handle: metrics_handle, }, ) - .with_task("liq-state-refresh", liq_state_refresh); + .with_task("liq-state-refresh", liq_state_refresh) + .with_task("liq-performance-refresh", liq_performance_refresh); match equity_price_task { Some(task) => builder.with_task("dashboard-equity-prices", task), diff --git a/src/metrics.rs b/src/metrics.rs index 7e1626f7b6..047f76e458 100644 --- a/src/metrics.rs +++ b/src/metrics.rs @@ -25,9 +25,10 @@ const DURATION_BUCKETS: [f64; 13] = [ ]; /// Duration histograms rendered with [`DURATION_BUCKETS`]. -const BUCKETED_HISTOGRAMS: [&str; 2] = [ +const BUCKETED_HISTOGRAMS: [&str; 3] = [ "dependency_call_duration_seconds", "order_fill_poll_duration_seconds", + "metrics_refresh_duration_seconds", ]; fn builder() -> Result { @@ -258,6 +259,11 @@ pub(crate) fn setup() -> Result { "Time of the poll that last set order_fill_block_lag_blocks, by chain; it stops \ advancing while the lag is unknown" ); + metrics::describe_histogram!( + "metrics_refresh_duration_seconds", + metrics::Unit::Seconds, + "Time one liq_ collector took to load and build its family, by collector" + ); metrics::describe_counter!( "bot_gas_redrive_total", "Bot-gas receipt-cost enqueue failures redriven instead of failing the triggering \ diff --git a/src/metrics/liquidity.rs b/src/metrics/liquidity.rs index 1c7ed7f7e4..deb24b5a0f 100644 --- a/src/metrics/liquidity.rs +++ b/src/metrics/liquidity.rs @@ -19,13 +19,16 @@ use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{Arc, LazyLock, Mutex, PoisonError}; use std::time::{SystemTime, UNIX_EPOCH}; +use chrono::{DateTime, Utc}; use metrics_exporter_prometheus::formatting::{write_help_line, write_type_line}; use rain_math_float::{Float, FloatError}; +use serde::Serialize; use tracing::{debug, error, warn}; use st0x_float_serde::format_float; pub(crate) mod inventory; +pub(crate) mod performance; pub(crate) mod prices; pub(crate) mod refresh; pub(crate) mod settings; @@ -81,6 +84,23 @@ pub(crate) enum LiqMetric { UsdcChainRatio, PositionLastPriceUsd, EquityExposureUsd, + HedgeLatencyMs, + HedgeLatencyMsSamples, + OpenExposureFillCount, + OpenExposureOldestTsSeconds, + FailureEventCount24h, + JobQueue, + BlockLagBlocks, + BlockLagSampledTsSeconds, + PollCycles24h, + PollErrors24h, + PollSkippedTicks24h, + PollDurationMs, + DependencyCalls24h, + DependencyErrors24h, + DependencyLatencyMs, + RebalanceStageMs, + AttestationLastMs, CollectorLastSuccessTsSeconds, } @@ -88,7 +108,7 @@ impl LiqMetric { /// Every variant, for the catalog tests. A new variant needs an entry /// here and its name in the catalog test. #[cfg(test)] - pub(crate) const ALL: [Self; 44] = [ + pub(crate) const ALL: [Self; 61] = [ Self::BotInfo, Self::BotStartTimestampSeconds, Self::SettingsInfo, @@ -132,6 +152,23 @@ impl LiqMetric { Self::UsdcChainRatio, Self::PositionLastPriceUsd, Self::EquityExposureUsd, + Self::HedgeLatencyMs, + Self::HedgeLatencyMsSamples, + Self::OpenExposureFillCount, + Self::OpenExposureOldestTsSeconds, + Self::FailureEventCount24h, + Self::JobQueue, + Self::BlockLagBlocks, + Self::BlockLagSampledTsSeconds, + Self::PollCycles24h, + Self::PollErrors24h, + Self::PollSkippedTicks24h, + Self::PollDurationMs, + Self::DependencyCalls24h, + Self::DependencyErrors24h, + Self::DependencyLatencyMs, + Self::RebalanceStageMs, + Self::AttestationLastMs, Self::CollectorLastSuccessTsSeconds, ]; @@ -180,6 +217,23 @@ impl LiqMetric { Self::UsdcChainRatio => "liq_usdc_chain_ratio", Self::PositionLastPriceUsd => "liq_position_last_price_usd", Self::EquityExposureUsd => "liq_equity_exposure_usd", + Self::HedgeLatencyMs => "liq_hedge_latency_ms", + Self::HedgeLatencyMsSamples => "liq_hedge_latency_ms_samples", + Self::OpenExposureFillCount => "liq_open_exposure_fill_count", + Self::OpenExposureOldestTsSeconds => "liq_open_exposure_oldest_ts_seconds", + Self::FailureEventCount24h => "liq_failure_event_count_24h", + Self::JobQueue => "liq_job_queue", + Self::BlockLagBlocks => "liq_block_lag_blocks", + Self::BlockLagSampledTsSeconds => "liq_block_lag_sampled_ts_seconds", + Self::PollCycles24h => "liq_poll_cycles_24h", + Self::PollErrors24h => "liq_poll_errors_24h", + Self::PollSkippedTicks24h => "liq_poll_skipped_ticks_24h", + Self::PollDurationMs => "liq_poll_duration_ms", + Self::DependencyCalls24h => "liq_dependency_calls_24h", + Self::DependencyErrors24h => "liq_dependency_errors_24h", + Self::DependencyLatencyMs => "liq_dependency_latency_ms", + Self::RebalanceStageMs => "liq_rebalance_stage_ms", + Self::AttestationLastMs => "liq_attestation_last_ms", Self::CollectorLastSuccessTsSeconds => "liq_collector_last_success_ts_seconds", } } @@ -270,6 +324,51 @@ impl LiqMetric { "Live wrapped-token mid price in USD of each symbol with a position" } Self::EquityExposureUsd => "Net position times its live price, in USD", + Self::HedgeLatencyMs => { + "Hedge pipeline stage latency over the last 24 hours, by stage and \ + nearest-rank quantile (p50, p90, p95, p99, max)" + } + Self::HedgeLatencyMsSamples => "Samples behind each liq_hedge_latency_ms stage", + Self::OpenExposureFillCount => { + "Fills observed after the symbol's latest hedge placement" + } + Self::OpenExposureOldestTsSeconds => { + "Block time of the oldest fill not yet covered by a hedge" + } + Self::FailureEventCount24h => { + "Money-at-risk lifecycle failure events over the last 24 hours, by event type" + } + Self::JobQueue => "Jobs in each apalis queue now, by job type and state; not windowed", + Self::BlockLagBlocks => "Latest sampled order-fill block lag of each hedged chain", + Self::BlockLagSampledTsSeconds => { + "Unix time of each hedged chain's latest block-lag sample" + } + Self::PollCycles24h => { + "Order-fill poll cycles of each hedged chain over the last 24 hours" + } + Self::PollErrors24h => { + "Failed order-fill poll cycles of each hedged chain over the last 24 hours" + } + Self::PollSkippedTicks24h => { + "Order-fill poll ticks each hedged chain dropped over the last 24 hours" + } + Self::PollDurationMs => { + "Order-fill poll cycle duration over the last 24 hours, by chain and quantile" + } + Self::DependencyCalls24h => { + "External dependency calls over the last 24 hours, by dependency and operation" + } + Self::DependencyErrors24h => "Failed external dependency calls over the last 24 hours", + Self::DependencyLatencyMs => { + "External dependency call latency over the last 24 hours, by quantile" + } + Self::RebalanceStageMs => { + "Completed rebalance stage duration over the last 30 days, by kind (usdc, \ + equity), stage and quantile" + } + Self::AttestationLastMs => { + "Duration of the latest CCTP attestation in the last 30 days" + } Self::CollectorLastSuccessTsSeconds => { "Unix time each liq_ collector last published its family" } @@ -301,13 +400,29 @@ impl LiqMetric { | Self::EquityWrapped | Self::EquityRatio | Self::PositionLastPriceUsd - | Self::EquityExposureUsd => &["symbol"], + | Self::EquityExposureUsd + | Self::OpenExposureFillCount + | Self::OpenExposureOldestTsSeconds => &["symbol"], Self::EquityChainAvailable => &["chain", "symbol"], Self::UsdcCorridorTarget | Self::UsdcCorridorDeviation | Self::UsdcChainAvailable | Self::UsdcChainInflight - | Self::UsdcChainRatio => &["chain"], + | Self::UsdcChainRatio + | Self::BlockLagBlocks + | Self::BlockLagSampledTsSeconds + | Self::PollCycles24h + | Self::PollErrors24h + | Self::PollSkippedTicks24h => &["chain"], + Self::HedgeLatencyMs => &["quantile", "stage"], + Self::HedgeLatencyMsSamples => &["stage"], + Self::FailureEventCount24h => &["event_type"], + Self::JobQueue => &["job_type", "state"], + Self::PollDurationMs => &["chain", "quantile"], + Self::DependencyCalls24h | Self::DependencyErrors24h => &["dependency", "operation"], + Self::DependencyLatencyMs => &["dependency", "operation", "quantile"], + Self::RebalanceStageMs => &["kind", "quantile", "stage"], + Self::AttestationLastMs => &["kind"], Self::CollectorLastSuccessTsSeconds => &["collector"], Self::BotStartTimestampSeconds | Self::SettingsEquityTarget @@ -380,6 +495,21 @@ impl LiqMetric { | Self::UsdcChainInflight | Self::UsdcChainRatio => Some(LiqFamily::Inventory), Self::PositionLastPriceUsd | Self::EquityExposureUsd => Some(LiqFamily::Prices), + Self::HedgeLatencyMs + | Self::HedgeLatencyMsSamples + | Self::OpenExposureFillCount + | Self::OpenExposureOldestTsSeconds => Some(LiqFamily::Latencies), + Self::FailureEventCount24h | Self::JobQueue => Some(LiqFamily::Reliability), + Self::BlockLagBlocks + | Self::BlockLagSampledTsSeconds + | Self::PollCycles24h + | Self::PollErrors24h + | Self::PollSkippedTicks24h + | Self::PollDurationMs + | Self::DependencyCalls24h + | Self::DependencyErrors24h + | Self::DependencyLatencyMs => Some(LiqFamily::Infra), + Self::RebalanceStageMs | Self::AttestationLastMs => Some(LiqFamily::Rebalances), Self::CollectorLastSuccessTsSeconds => None, } } @@ -392,6 +522,10 @@ pub(crate) enum LiqFamily { Settings, Inventory, Prices, + Latencies, + Reliability, + Infra, + Rebalances, } impl LiqFamily { @@ -402,6 +536,10 @@ impl LiqFamily { Self::Settings => "settings", Self::Inventory => "inventory", Self::Prices => "prices", + Self::Latencies => "latencies", + Self::Reliability => "reliability", + Self::Infra => "infra", + Self::Rebalances => "rebalances", } } } @@ -653,8 +791,47 @@ pub(crate) fn integer_value(value: u64) -> Result { Ok(value.to_string().parse()?) } +/// Converts a signed integer to `f64`, refusing values `f64` cannot hold +/// exactly. +pub(crate) fn signed_integer_value(value: i64) -> Result { + let magnitude = integer_value(value.unsigned_abs())?; + Ok(if value < 0 { -magnitude } else { magnitude }) +} + +/// Converts a count to `f64`, refusing values `f64` cannot hold exactly. +pub(crate) fn count_value(count: usize) -> Result { + integer_value(u64::try_from(count)?) +} + +/// Unix seconds with microsecond precision, as the exporter computed them +/// from the RFC 3339 text: whole microseconds divided by one million. +pub(crate) fn timestamp_value(at: DateTime) -> Result { + const MICROS_PER_SECOND: f64 = 1_000_000.0; + + Ok(signed_integer_value(at.timestamp_micros())? / MICROS_PER_SECOND) +} + +/// The serde wire name of a unit enum variant, which is the label value the +/// exporter read from the bot's JSON. +pub(crate) fn wire_name(value: &T) -> Result { + match serde_json::to_value(value)? { + serde_json::Value::String(name) => Ok(name), + _ => Err(WireNameError::NotAString), + } +} + +#[derive(Debug, thiserror::Error)] +pub(crate) enum WireNameError { + #[error("failed to serialize the value")] + Serialize(#[from] serde_json::Error), + #[error("the value does not serialize to a string")] + NotAString, +} + #[derive(Debug, thiserror::Error)] pub(crate) enum LiqValueError { + #[error("count does not fit in u64")] + Count(#[from] std::num::TryFromIntError), #[error("failed to format the Float value")] Format(#[from] FloatError), #[error("formatted Float value does not parse as f64")] @@ -947,6 +1124,23 @@ pub(crate) mod tests { "liq_usdc_chain_ratio", "liq_position_last_price_usd", "liq_equity_exposure_usd", + "liq_hedge_latency_ms", + "liq_hedge_latency_ms_samples", + "liq_open_exposure_fill_count", + "liq_open_exposure_oldest_ts_seconds", + "liq_failure_event_count_24h", + "liq_job_queue", + "liq_block_lag_blocks", + "liq_block_lag_sampled_ts_seconds", + "liq_poll_cycles_24h", + "liq_poll_errors_24h", + "liq_poll_skipped_ticks_24h", + "liq_poll_duration_ms", + "liq_dependency_calls_24h", + "liq_dependency_errors_24h", + "liq_dependency_latency_ms", + "liq_rebalance_stage_ms", + "liq_attestation_last_ms", "liq_collector_last_success_ts_seconds", ] ); @@ -956,9 +1150,22 @@ pub(crate) mod tests { LiqFamily::Settings, LiqFamily::Inventory, LiqFamily::Prices, + LiqFamily::Latencies, + LiqFamily::Reliability, + LiqFamily::Infra, + LiqFamily::Rebalances, ] .map(LiqFamily::collector), - ["health", "settings", "inventory", "prices"] + [ + "health", + "settings", + "inventory", + "prices", + "latencies", + "reliability", + "infra", + "rebalances", + ] ); } @@ -1015,6 +1222,31 @@ pub(crate) mod tests { )); } + #[test] + fn signed_count_and_timestamp_values_are_exact() { + assert_eq!(signed_integer_value(-12).ok(), Some(-12.0)); + assert_eq!(signed_integer_value(i64::MIN).ok(), None); + assert_eq!(count_value(42).ok(), Some(42.0)); + + let at = DateTime::parse_from_rfc3339("2026-03-01T12:30:15.250Z") + .unwrap() + .with_timezone(&Utc); + assert_eq!(timestamp_value(at).ok(), Some(1_772_368_215.25)); + } + + #[test] + fn wire_name_reads_the_serde_name() { + assert_eq!( + wire_name(&st0x_dto::ChainName::HyperEvm).unwrap(), + "hyperevm" + ); + assert_eq!( + wire_name(&st0x_dto::FailureEventType::OffchainOrderFailed).unwrap(), + "OffchainOrderEvent::Failed" + ); + assert!(matches!(wire_name(&3), Err(WireNameError::NotAString))); + } + #[test] fn sample_rejects_non_finite_values() { for value in [f64::NAN, f64::INFINITY, f64::NEG_INFINITY] { diff --git a/src/metrics/liquidity/performance.rs b/src/metrics/liquidity/performance.rs new file mode 100644 index 0000000000..edf23e71c0 --- /dev/null +++ b/src/metrics/liquidity/performance.rs @@ -0,0 +1,624 @@ +//! The performance families: hedge latencies, reliability, infrastructure +//! and rebalance timings. +//! +//! Each builder takes the report the matching `/performance/*` endpoint +//! serves, so the series hold the numbers the exporter read from those +//! endpoints. `liq-performance-refresh` builds them every 60 seconds over the +//! last 24 hours, and the rebalance family every fifth cycle over the last 30 +//! days, like the exporter. +//! +//! The infrastructure series carry a `chain` label per hedged chain. The +//! exporter published them without labels, from a report shape the bot no +//! longer serves. + +use tracing::warn; + +use st0x_dto::{ + EquityTimings, FailureEventCount, HedgeLatencies, InfraReport, JobQueueHealth, LatencyStats, + RebalanceTimings, +}; + +use super::{ + LiqMetric, LiqSample, count_value, push_sample, signed_integer_value, strip_prefix, + timestamp_value, wire_name, +}; + +/// `liq_hedge_latency_ms`, its sample counts, and the open exposures. +pub(crate) fn latency_samples(report: &HedgeLatencies) -> Vec { + let stages = &report.summary.stages; + let mut samples = Vec::new(); + + for (stage, stats) in [ + ("detection", &stages.detection), + ("decision", &stages.decision), + ("submission", &stages.submission), + ("execution", &stages.execution), + ("exposure_window", &stages.exposure_window), + ] { + let Some(stats) = stats else { + continue; + }; + let labels = vec![("stage", stage.to_string())]; + + push_quantiles(&mut samples, LiqMetric::HedgeLatencyMs, &labels, stats); + push_sample( + &mut samples, + LiqMetric::HedgeLatencyMsSamples, + labels, + count_value(stats.sample_count), + ); + } + + for exposure in &report.open_exposures { + let labels = vec![("symbol", strip_prefix(exposure.symbol.as_str()).to_string())]; + + push_sample( + &mut samples, + LiqMetric::OpenExposureFillCount, + labels.clone(), + count_value(exposure.fill_count), + ); + push_sample( + &mut samples, + LiqMetric::OpenExposureOldestTsSeconds, + labels, + timestamp_value(exposure.oldest_fill_block_timestamp), + ); + } + + samples +} + +/// Failure events in the window and the instantaneous job queue counts. +pub(crate) fn reliability_samples( + failure_events: &[FailureEventCount], + job_queues: &[JobQueueHealth], +) -> Vec { + let mut samples = Vec::new(); + + for event in failure_events { + push_sample( + &mut samples, + LiqMetric::FailureEventCount24h, + vec![("event_type", event.event_type.as_str().to_string())], + count_value(event.count), + ); + } + + for queue in job_queues { + for (state, count) in [ + ("pending", queue.pending), + ("running", queue.running), + ("done", queue.done), + ("failed", queue.failed), + ("awaiting_retry", queue.awaiting_retry), + ("killed", queue.killed), + ("retried", queue.retried), + ] { + push_sample( + &mut samples, + LiqMetric::JobQueue, + vec![ + ("job_type", queue.job_type.clone()), + ("state", state.to_string()), + ], + count_value(count), + ); + } + } + + samples +} + +/// Block lag and poll health per hedged chain, and dependency call stats. +pub(crate) fn infra_samples(report: &InfraReport) -> Vec { + let mut samples = Vec::new(); + + for lag in &report.monitor.block_lag { + let Some(chain) = label_value(&lag.chain) else { + continue; + }; + let labels = vec![("chain", chain)]; + + if let Some(blocks) = lag.current_lag_blocks { + push_sample( + &mut samples, + LiqMetric::BlockLagBlocks, + labels.clone(), + signed_integer_value(blocks), + ); + } + if let Some(sampled_at) = lag.current_lag_sampled_at { + push_sample( + &mut samples, + LiqMetric::BlockLagSampledTsSeconds, + labels, + timestamp_value(sampled_at), + ); + } + } + + for poll in &report.monitor.poll { + let Some(chain) = label_value(&poll.chain) else { + continue; + }; + let labels = vec![("chain", chain)]; + + for (metric, count) in [ + (LiqMetric::PollCycles24h, poll.cycles), + (LiqMetric::PollErrors24h, poll.errors), + (LiqMetric::PollSkippedTicks24h, poll.skipped_ticks), + ] { + push_sample(&mut samples, metric, labels.clone(), count_value(count)); + } + if let Some(duration) = &poll.duration { + push_quantiles(&mut samples, LiqMetric::PollDurationMs, &labels, duration); + } + } + + for dependency in &report.dependencies { + let Some(name) = label_value(&dependency.dependency) else { + continue; + }; + let labels = vec![ + ("dependency", name), + ("operation", dependency.operation.clone()), + ]; + + push_sample( + &mut samples, + LiqMetric::DependencyCalls24h, + labels.clone(), + count_value(dependency.calls), + ); + push_sample( + &mut samples, + LiqMetric::DependencyErrors24h, + labels.clone(), + count_value(dependency.errors), + ); + if let Some(latency) = &dependency.latency { + push_quantiles( + &mut samples, + LiqMetric::DependencyLatencyMs, + &labels, + latency, + ); + } + } + + samples +} + +/// Stage percentiles of USDC and equity rebalances, and the latest CCTP +/// attestation. Equity rebalances have no attestation. +pub(crate) fn rebalance_samples(usdc: &RebalanceTimings, equity: &EquityTimings) -> Vec { + let mut samples = Vec::new(); + + for entry in &usdc.stage_summary { + push_stage(&mut samples, "usdc", &entry.stage, &entry.stats); + } + if let Some(last) = usdc.attestation_trend.last() { + push_sample( + &mut samples, + LiqMetric::AttestationLastMs, + vec![("kind", "usdc".to_string())], + signed_integer_value(last.duration_ms), + ); + } + + for entry in &equity.stage_summary { + push_stage(&mut samples, "equity", &entry.stage, &entry.stats); + } + + samples +} + +fn push_stage( + samples: &mut Vec, + kind: &str, + stage: &Stage, + stats: &LatencyStats, +) { + let Some(stage) = label_value(stage) else { + return; + }; + + push_quantiles( + samples, + LiqMetric::RebalanceStageMs, + &[("kind", kind.to_string()), ("stage", stage)], + stats, + ); +} + +/// One sample per exporter quantile label, from the nearest-rank stats. +fn push_quantiles( + samples: &mut Vec, + metric: LiqMetric, + labels: &[(&'static str, String)], + stats: &LatencyStats, +) { + for (quantile, millis) in [ + ("p50", stats.p50_ms), + ("p90", stats.p90_ms), + ("p95", stats.p95_ms), + ("p99", stats.p99_ms), + ("max", stats.max_ms), + ] { + let mut quantile_labels = labels.to_vec(); + quantile_labels.push(("quantile", quantile.to_string())); + push_sample( + samples, + metric, + quantile_labels, + signed_integer_value(millis), + ); + } +} + +/// The wire name of an enum label value, or `None` after logging. +fn label_value(value: &T) -> Option { + wire_name(value) + .inspect_err(|error| warn!(?value, %error, "Skipped liq_ samples with no label value")) + .ok() +} + +#[cfg(test)] +pub(crate) mod tests { + use std::collections::BTreeMap; + + use chrono::{DateTime, Utc}; + use serde_json::Value; + + use st0x_dto::{ + AttestationSample, ChainBlockLag, ChainName, ChainPollHealth, CountedLogLevel, + DependencyName, DependencyStats, EquityStageName, EquityStageStats, FailureEventType, + LatencySummary, LogTargetCount, LogVolumeBucket, MonitorTelemetry, OpenExposureReport, + RebalanceStageName, RebalanceStageStats, ReliabilityReport, StageLatencies, + }; + use st0x_finance::Symbol; + + use super::*; + use crate::metrics::liquidity::LiqFamily; + use crate::metrics::liquidity::inventory::tests::{golden_for, render_family}; + use crate::metrics::liquidity::tests::{SeriesKey, series}; + + pub(crate) fn at(rfc3339: &str) -> DateTime { + DateTime::parse_from_rfc3339(rfc3339) + .unwrap() + .with_timezone(&Utc) + } + + fn stats(p50_ms: i64, max_ms: i64, sample_count: usize) -> LatencyStats { + LatencyStats { + p50_ms, + p90_ms: p50_ms + 10, + p95_ms: p50_ms + 20, + p99_ms: p50_ms + 30, + max_ms, + sample_count, + } + } + + fn assert_fixture(dto: &T, fixture_json: &str) { + let fixture: Value = serde_json::from_str(fixture_json).unwrap(); + assert_eq!(serde_json::to_value(dto).unwrap(), fixture); + } + + pub(crate) fn latencies_fixture() -> HedgeLatencies { + HedgeLatencies { + summary: LatencySummary { + fill_count: 7, + stages: StageLatencies { + detection: Some(stats(1_200, 4_000, 7)), + decision: Some(stats(-12, 40, 6)), + submission: None, + execution: Some(stats(850, 9_100, 5)), + exposure_window: Some(stats(3_400, 61_000, 5)), + }, + }, + buckets: Vec::new(), + cycles: Vec::new(), + total_cycles: 3, + open_exposures: vec![ + OpenExposureReport { + symbol: Symbol::new("tAAPL").unwrap(), + fill_count: 2, + oldest_fill_block_timestamp: at("2026-03-01T12:00:00Z"), + }, + OpenExposureReport { + symbol: Symbol::new("RKLB").unwrap(), + fill_count: 1, + oldest_fill_block_timestamp: at("2026-03-01T12:30:15.250Z"), + }, + ], + } + } + + pub(crate) fn reliability_fixture() -> ReliabilityReport { + ReliabilityReport { + log_buckets: vec![LogVolumeBucket { + start: at("2026-03-01T00:00:00Z"), + errors: 3, + warnings: 5, + }], + log_targets: vec![ + LogTargetCount { + target: "inventory".to_string(), + level: CountedLogLevel::Warn, + count: 5, + sparkline: vec![5], + }, + LogTargetCount { + target: "hedge".to_string(), + level: CountedLogLevel::Error, + count: 3, + sparkline: vec![3], + }, + ], + failure_events: vec![ + FailureEventCount { + event_type: FailureEventType::OffchainOrderFailed, + count: 2, + last_at: at("2026-03-01T09:00:00Z"), + }, + FailureEventCount { + event_type: FailureEventType::AttestationTimedOut, + count: 1, + last_at: at("2026-03-01T10:00:00Z"), + }, + ], + job_queues: vec![JobQueueHealth { + job_type: "PlaceHedge".to_string(), + pending: 1, + running: 0, + done: 40, + failed: 2, + awaiting_retry: 1, + killed: 0, + retried: 3, + oldest_pending_run_at: Some(at("2026-03-01T11:00:00Z")), + }], + log_entries_truncated: false, + } + } + + /// The exporter fails on a non-empty per-chain `poll` list, so the + /// golden case leaves it empty and pins only the dependency series; the + /// per-chain series have their own test. + fn infra_fixture() -> InfraReport { + InfraReport { + monitor: MonitorTelemetry { + enabled_chains: vec![ChainName::Base, ChainName::Robinhood], + block_lag: vec![ + ChainBlockLag { + chain: ChainName::Base, + current_lag_blocks: Some(3), + current_lag_sampled_at: Some(at("2026-03-01T12:00:05Z")), + points: Vec::new(), + }, + ChainBlockLag { + chain: ChainName::Robinhood, + current_lag_blocks: None, + current_lag_sampled_at: None, + points: Vec::new(), + }, + ], + poll: Vec::new(), + }, + dependencies: vec![ + DependencyStats { + dependency: DependencyName::Broker, + operation: "get_order_status".to_string(), + calls: 30, + errors: 0, + latency: None, + buckets: Vec::new(), + }, + DependencyStats { + dependency: DependencyName::Rpc, + operation: "eth_getLogs".to_string(), + calls: 120, + errors: 2, + latency: Some(stats(95, 2_300, 120)), + buckets: Vec::new(), + }, + ], + } + } + + fn rebalance_fixture() -> (RebalanceTimings, EquityTimings) { + ( + RebalanceTimings { + operations: Vec::new(), + total_operations: 2, + skipped_operations: 0, + stage_summary: vec![ + RebalanceStageStats { + stage: RebalanceStageName::Conversion, + stats: stats(2_000, 2_500, 2), + }, + RebalanceStageStats { + stage: RebalanceStageName::Attestation, + stats: stats(52_000, 60_000, 2), + }, + ], + attestation_trend: vec![ + AttestationSample { + burned_at: at("2026-02-20T08:00:00Z"), + duration_ms: 60_000, + }, + AttestationSample { + burned_at: at("2026-02-27T08:00:00Z"), + duration_ms: 45_000, + }, + ], + }, + EquityTimings { + operations: Vec::new(), + total_operations: 1, + skipped_operations: 1, + stage_summary: vec![EquityStageStats { + stage: EquityStageName::MintAcceptance, + stats: stats(30_000, 30_050, 1), + }], + }, + ) + } + + #[test] + fn latencies_match_the_exporter_golden() { + let report = latencies_fixture(); + assert_fixture(&report, include_str!("testdata/latencies.json")); + + assert_eq!( + render_family(LiqFamily::Latencies, latency_samples(&report)), + golden_for( + LiqFamily::Latencies, + include_str!("testdata/latencies.prom") + ) + ); + } + + #[test] + fn reliability_matches_the_exporter_golden() { + let report = reliability_fixture(); + assert_fixture(&report, include_str!("testdata/reliability.json")); + + assert_eq!( + render_family( + LiqFamily::Reliability, + reliability_samples(&report.failure_events, &report.job_queues), + ), + golden_for( + LiqFamily::Reliability, + include_str!("testdata/reliability.prom") + ) + ); + } + + #[test] + fn infra_dependencies_match_the_exporter_golden() { + let report = infra_fixture(); + assert_fixture(&report, include_str!("testdata/infra.json")); + + let dependency_names = [ + "liq_dependency_calls_24h", + "liq_dependency_errors_24h", + "liq_dependency_latency_ms", + ]; + let dependencies: BTreeMap = + render_family(LiqFamily::Infra, infra_samples(&report)) + .into_iter() + .filter(|((name, _), _)| dependency_names.contains(&name.as_str())) + .collect(); + + assert_eq!( + dependencies, + golden_for(LiqFamily::Infra, include_str!("testdata/infra.prom")) + ); + } + + #[test] + fn rebalances_match_the_exporter_golden() { + let (usdc, equity) = rebalance_fixture(); + assert_fixture( + &serde_json::json!({ "usdc": usdc, "equity": equity }), + include_str!("testdata/rebalance.json"), + ); + + assert_eq!( + render_family(LiqFamily::Rebalances, rebalance_samples(&usdc, &equity)), + golden_for( + LiqFamily::Rebalances, + include_str!("testdata/rebalance.prom") + ) + ); + } + + /// The exporter has no working baseline for these (it fails on the + /// per-chain report), so the expectation is written by hand: one series + /// per hedged chain, and nothing for a value not sampled yet. + #[test] + fn infra_publishes_block_lag_and_poll_health_per_chain() { + let mut report = infra_fixture(); + report.dependencies.clear(); + report.monitor.poll = vec![ + ChainPollHealth { + chain: ChainName::Base, + cycles: 17_280, + errors: 4, + skipped_ticks: 1, + duration: Some(stats(180, 2_400, 17_280)), + }, + ChainPollHealth { + chain: ChainName::Robinhood, + cycles: 0, + errors: 0, + skipped_ticks: 0, + duration: None, + }, + ]; + + let base = [("chain", "base")]; + let robinhood = [("chain", "robinhood")]; + let duration = |quantile| { + series( + "liq_poll_duration_ms", + &[("chain", "base"), ("quantile", quantile)], + ) + }; + assert_eq!( + render_family(LiqFamily::Infra, infra_samples(&report)), + BTreeMap::from([ + (series("liq_block_lag_blocks", &base), 3.0), + ( + series("liq_block_lag_sampled_ts_seconds", &base), + 1_772_366_405.0 + ), + (series("liq_poll_cycles_24h", &base), 17_280.0), + (series("liq_poll_errors_24h", &base), 4.0), + (series("liq_poll_skipped_ticks_24h", &base), 1.0), + (duration("p50"), 180.0), + (duration("p90"), 190.0), + (duration("p95"), 200.0), + (duration("p99"), 210.0), + (duration("max"), 2_400.0), + (series("liq_poll_cycles_24h", &robinhood), 0.0), + (series("liq_poll_errors_24h", &robinhood), 0.0), + (series("liq_poll_skipped_ticks_24h", &robinhood), 0.0), + ]) + ); + } + + #[test] + fn an_empty_attestation_trend_publishes_no_attestation() { + let (mut usdc, equity) = rebalance_fixture(); + usdc.attestation_trend.clear(); + + let rendered = render_family(LiqFamily::Rebalances, rebalance_samples(&usdc, &equity)); + + assert!( + rendered + .keys() + .all(|(name, _)| name != "liq_attestation_last_ms"), + "{rendered:?}" + ); + } + + #[test] + fn a_stage_without_samples_publishes_nothing() { + let mut report = latencies_fixture(); + report.summary.stages = StageLatencies { + detection: None, + decision: None, + submission: None, + execution: None, + exposure_window: None, + }; + report.open_exposures.clear(); + + assert_eq!(latency_samples(&report), Vec::new()); + } +} diff --git a/src/metrics/liquidity/refresh.rs b/src/metrics/liquidity/refresh.rs index 8312fd29c1..8b928fd5bf 100644 --- a/src/metrics/liquidity/refresh.rs +++ b/src/metrics/liquidity/refresh.rs @@ -7,13 +7,24 @@ use std::time::{Duration, SystemTime}; use chrono::Utc; use sqlx::SqlitePool; use task_supervisor::{SupervisedTask, TaskResult}; -use tokio::time::{MissedTickBehavior, timeout}; +use tokio::time::{Instant, MissedTickBehavior, timeout}; use tracing::{info, warn}; +use st0x_config::ChainRegistry; +use st0x_dto::InfraReport; + +use super::performance::{infra_samples, latency_samples, rebalance_samples, reliability_samples}; use super::prices::{load_positions, price_samples}; -use super::{LiqFamilies, LiqFamily}; +use super::{LiqFamilies, LiqFamily, LiqSample}; use crate::dashboard::equity_price::EquityPriceStore; use crate::inventory::BroadcastingInventory; +use crate::performance::equity_timing::load_equity_timings; +use crate::performance::infra::{load_dependency_stats, load_monitor_telemetry}; +use crate::performance::rebalance::load_rebalance_timings; +use crate::performance::reliability::{load_failure_events, load_job_queue_health}; +use crate::performance::{ + PerformanceError, ReportRange, hedge_latency_report, load_hedge_performance, +}; /// The exporter's cadence and the scrape interval. const STATE_REFRESH_PERIOD: Duration = Duration::from_secs(60); @@ -67,6 +78,141 @@ impl LiqStateRefresh { } } +/// The exporter's cadence for the performance collectors. +const PERFORMANCE_REFRESH_PERIOD: Duration = Duration::from_secs(60); + +/// The rebalance family refreshes on every fifth performance cycle, like the +/// exporter's slow cycle. +const REBALANCE_EVERY_CYCLES: u64 = 5; + +/// The window of the 24-hour performance series. +const PERFORMANCE_WINDOW: chrono::Duration = chrono::Duration::hours(24); + +/// The window of the rebalance timings. +const REBALANCE_WINDOW: chrono::Duration = chrono::Duration::days(30); + +/// Longest one collector may take before the cycle keeps its last family. +const COLLECTOR_TIMEOUT: Duration = Duration::from_secs(30); + +/// A collector slower than this is logged: the loaders replay history and +/// the exporter runs the same loaders while both publish. +const SLOW_COLLECTOR: Duration = Duration::from_secs(10); + +/// Every 60 seconds: the latencies, reliability and infra families over the +/// last 24 hours, and every fifth cycle the rebalances over the last 30 +/// days. Each family comes from the loaders its `/performance/*` endpoint +/// uses. A collector that fails or times out keeps its last family, and its +/// `liq_collector_last_success_ts_seconds` stops advancing. +#[derive(Clone)] +pub(crate) struct LiqPerformanceRefresh { + pub(crate) pool: SqlitePool, + pub(crate) chains: ChainRegistry, + pub(crate) families: &'static LiqFamilies, +} + +impl LiqPerformanceRefresh { + async fn refresh_once(&self, cycle: u64) { + let now = Utc::now(); + let day = ReportRange { + from: now - PERFORMANCE_WINDOW, + to: now, + }; + + self.collect(LiqFamily::Latencies, async { + let performances = load_hedge_performance(&self.pool, &day).await?; + Ok(latency_samples(&hedge_latency_report(&performances, &day))) + }) + .await; + + self.collect(LiqFamily::Reliability, async { + let (failure_events, job_queues) = tokio::try_join!( + load_failure_events(&self.pool, &day), + load_job_queue_health(&self.pool), + )?; + Ok(reliability_samples(&failure_events, &job_queues)) + }) + .await; + + self.collect(LiqFamily::Infra, async { + let (monitor, dependencies) = tokio::try_join!( + load_monitor_telemetry(&self.pool, &day, &self.chains), + load_dependency_stats(&self.pool, &day), + )?; + Ok(infra_samples(&InfraReport { + monitor, + dependencies, + })) + }) + .await; + + if cycle.is_multiple_of(REBALANCE_EVERY_CYCLES) { + let month = ReportRange { + from: now - REBALANCE_WINDOW, + to: now, + }; + self.collect(LiqFamily::Rebalances, async { + let (usdc, equity) = tokio::try_join!( + load_rebalance_timings(&self.pool, &month), + load_equity_timings(&self.pool, &month), + )?; + Ok(rebalance_samples(&usdc, &equity)) + }) + .await; + } + } + + /// Publishes `family` from `build`, or keeps the last one after logging + /// why. The duration goes to `metrics_refresh_duration_seconds`. + async fn collect( + &self, + family: LiqFamily, + build: impl Future, PerformanceError>>, + ) { + let collector = family.collector(); + let started = Instant::now(); + let result = timeout(COLLECTOR_TIMEOUT, build).await; + let elapsed = started.elapsed(); + + metrics::histogram!("metrics_refresh_duration_seconds", "collector" => collector) + .record(elapsed); + if elapsed > SLOW_COLLECTOR { + warn!(collector, ?elapsed, "A liq_ collector was slow"); + } + + match result { + Ok(Ok(samples)) => self.families.replace(family, samples, SystemTime::now()), + Ok(Err(error)) => { + warn!( + collector, + ?error, + "Kept the last liq_ family: its loader failed" + ); + } + Err(_) => warn!( + collector, + timeout = ?COLLECTOR_TIMEOUT, + "Kept the last liq_ family: its loader took too long" + ), + } + } +} + +impl SupervisedTask for LiqPerformanceRefresh { + async fn run(&mut self) -> TaskResult { + info!("liq_ performance refresh started"); + + let mut interval = tokio::time::interval(PERFORMANCE_REFRESH_PERIOD); + interval.set_missed_tick_behavior(MissedTickBehavior::Skip); + + let mut cycle: u64 = 0; + loop { + interval.tick().await; + self.refresh_once(cycle).await; + cycle = cycle.wrapping_add(1); + } + } +} + impl SupervisedTask for LiqStateRefresh { async fn run(&mut self) -> TaskResult { info!("liq_ state refresh started"); @@ -83,17 +229,144 @@ impl SupervisedTask for LiqStateRefresh { #[cfg(test)] mod tests { + use alloy::primitives::Address; use tokio::sync::broadcast; + use st0x_config::create_test_ctx_with_order_owner; + use st0x_execution::{FractionalShares, Symbol}; use st0x_float_macro::float; use super::*; use crate::inventory::InventoryView; + use crate::metrics::liquidity::LiqMetric; use crate::metrics::liquidity::inventory::tests::{leaked_families, publisher, rendered_store}; - use crate::metrics::liquidity::tests::series; + use crate::metrics::liquidity::tests::{parse_exposition, series}; use crate::test_utils::setup_test_db; + fn performance(families: &'static LiqFamilies, pool: SqlitePool) -> LiqPerformanceRefresh { + LiqPerformanceRefresh { + pool, + chains: create_test_ctx_with_order_owner(Address::ZERO).chains, + families, + } + } + + fn performance_collectors(families: &'static LiqFamilies) -> Vec> { + ["latencies", "reliability", "infra", "rebalances"] + .into_iter() + .map(|name| collector(families, name)) + .collect() + } + + /// On an empty database every family publishes: the counts are zero + /// where the loaders report zero, and absent where they report nothing. + #[tokio::test] + async fn the_first_performance_cycle_publishes_every_family() { + let families = leaked_families(); + let task = performance(families, setup_test_db().await); + let recorder = crate::metrics::local_recorder(); + let handle = recorder.handle(); + let _local = metrics::set_default_local_recorder(&recorder); + + task.refresh_once(0).await; + + let durations = parse_exposition(&handle.render()); + for collector in ["latencies", "reliability", "infra", "rebalances"] { + assert_eq!( + durations.get(&series( + "metrics_refresh_duration_seconds_count", + &[("collector", collector)], + )), + Some(&1.0), + "{collector}" + ); + } + assert!( + performance_collectors(families).iter().all(Option::is_some), + "{:?}", + performance_collectors(families) + ); + let rendered = rendered_store(families); + let base = [("chain", "base")]; + assert_eq!( + rendered.get(&series("liq_poll_cycles_24h", &base)), + Some(&0.0) + ); + assert_eq!(rendered.get(&series("liq_block_lag_blocks", &base)), None); + } + + #[tokio::test] + async fn rebalances_refresh_only_every_fifth_cycle() { + let families = leaked_families(); + let task = performance(families, setup_test_db().await); + + task.refresh_once(1).await; + assert_eq!(collector(families, "rebalances"), None); + assert!(collector(families, "latencies").is_some()); + + task.refresh_once(5).await; + assert!(collector(families, "rebalances").is_some()); + } + + #[tokio::test] + async fn a_failing_loader_keeps_the_last_family_and_its_timestamp() { + let families = leaked_families(); + let task = performance(families, setup_test_db().await); + let previous = vec![ + LiqSample::new( + LiqMetric::JobQueue, + vec![ + ("job_type", "PlaceHedge".to_string()), + ("state", "pending".to_string()), + ], + 4.0, + ) + .unwrap(), + ]; + families.replace( + LiqFamily::Reliability, + previous, + SystemTime::UNIX_EPOCH + Duration::from_secs(100), + ); + task.pool.close().await; + + task.refresh_once(0).await; + + assert_eq!( + rendered_store(families).get(&series( + "liq_job_queue", + &[("job_type", "PlaceHedge"), ("state", "pending")] + )), + Some(&4.0) + ); + assert_eq!( + performance_collectors(families), + [None, Some(100.0), None, None] + ); + } + + #[tokio::test(start_paused = true)] + async fn a_collector_that_times_out_keeps_the_last_family() { + let families = leaked_families(); + // The pending collector never reads the pool; a lazy one does no I/O + // while time is paused. + let task = performance( + families, + SqlitePool::connect_lazy("sqlite::memory:").unwrap(), + ); + families.replace( + LiqFamily::Latencies, + Vec::new(), + SystemTime::UNIX_EPOCH + Duration::from_secs(100), + ); + + task.collect(LiqFamily::Latencies, std::future::pending()) + .await; + + assert_eq!(collector(families, "latencies"), Some(100.0)); + } + async fn refresh( families: &'static LiqFamilies, equity_prices: EquityPriceStore, diff --git a/src/metrics/liquidity/testdata/infra.json b/src/metrics/liquidity/testdata/infra.json new file mode 100644 index 0000000000..ad3f0f0044 --- /dev/null +++ b/src/metrics/liquidity/testdata/infra.json @@ -0,0 +1,48 @@ +{ + "monitor": { + "enabledChains": [ + "base", + "robinhood" + ], + "blockLag": [ + { + "chain": "base", + "currentLagBlocks": 3, + "currentLagSampledAt": "2026-03-01T12:00:05Z", + "points": [] + }, + { + "chain": "robinhood", + "currentLagBlocks": null, + "currentLagSampledAt": null, + "points": [] + } + ], + "poll": [] + }, + "dependencies": [ + { + "dependency": "broker", + "operation": "get_order_status", + "calls": 30, + "errors": 0, + "latency": null, + "buckets": [] + }, + { + "dependency": "rpc", + "operation": "eth_getLogs", + "calls": 120, + "errors": 2, + "latency": { + "p50Ms": 95, + "p90Ms": 105, + "p95Ms": 115, + "p99Ms": 125, + "maxMs": 2300, + "sampleCount": 120 + }, + "buckets": [] + } + ] +} diff --git a/src/metrics/liquidity/testdata/infra.prom b/src/metrics/liquidity/testdata/infra.prom new file mode 100644 index 0000000000..d4e967870f --- /dev/null +++ b/src/metrics/liquidity/testdata/infra.prom @@ -0,0 +1,12 @@ +# Expected exporter output for infra.json, written by +# scripts/liq-parity/golden.py from t0.devops revision 226c029. +liq_dependency_calls_24h{dependency="broker",operation="get_order_status"} 30.0 +liq_dependency_errors_24h{dependency="broker",operation="get_order_status"} 0.0 +liq_dependency_calls_24h{dependency="rpc",operation="eth_getLogs"} 120.0 +liq_dependency_errors_24h{dependency="rpc",operation="eth_getLogs"} 2.0 +liq_dependency_latency_ms{dependency="rpc",operation="eth_getLogs",quantile="p50"} 95.0 +liq_dependency_latency_ms{dependency="rpc",operation="eth_getLogs",quantile="p90"} 105.0 +liq_dependency_latency_ms{dependency="rpc",operation="eth_getLogs",quantile="p95"} 115.0 +liq_dependency_latency_ms{dependency="rpc",operation="eth_getLogs",quantile="p99"} 125.0 +liq_dependency_latency_ms{dependency="rpc",operation="eth_getLogs",quantile="max"} 2300.0 +liq_dependency_latency_ms_samples{dependency="rpc",operation="eth_getLogs"} 120.0 diff --git a/src/metrics/liquidity/testdata/latencies.json b/src/metrics/liquidity/testdata/latencies.json new file mode 100644 index 0000000000..b739577d88 --- /dev/null +++ b/src/metrics/liquidity/testdata/latencies.json @@ -0,0 +1,55 @@ +{ + "summary": { + "fillCount": 7, + "stages": { + "detection": { + "p50Ms": 1200, + "p90Ms": 1210, + "p95Ms": 1220, + "p99Ms": 1230, + "maxMs": 4000, + "sampleCount": 7 + }, + "decision": { + "p50Ms": -12, + "p90Ms": -2, + "p95Ms": 8, + "p99Ms": 18, + "maxMs": 40, + "sampleCount": 6 + }, + "submission": null, + "execution": { + "p50Ms": 850, + "p90Ms": 860, + "p95Ms": 870, + "p99Ms": 880, + "maxMs": 9100, + "sampleCount": 5 + }, + "exposureWindow": { + "p50Ms": 3400, + "p90Ms": 3410, + "p95Ms": 3420, + "p99Ms": 3430, + "maxMs": 61000, + "sampleCount": 5 + } + } + }, + "buckets": [], + "cycles": [], + "totalCycles": 3, + "openExposures": [ + { + "symbol": "tAAPL", + "fillCount": 2, + "oldestFillBlockTimestamp": "2026-03-01T12:00:00Z" + }, + { + "symbol": "RKLB", + "fillCount": 1, + "oldestFillBlockTimestamp": "2026-03-01T12:30:15.250Z" + } + ] +} diff --git a/src/metrics/liquidity/testdata/latencies.prom b/src/metrics/liquidity/testdata/latencies.prom new file mode 100644 index 0000000000..d5693cb3f7 --- /dev/null +++ b/src/metrics/liquidity/testdata/latencies.prom @@ -0,0 +1,32 @@ +# Expected exporter output for latencies.json, written by +# scripts/liq-parity/golden.py from t0.devops revision 226c029. +liq_hedge_latency_ms{quantile="p50",stage="detection"} 1200.0 +liq_hedge_latency_ms{quantile="p90",stage="detection"} 1210.0 +liq_hedge_latency_ms{quantile="p95",stage="detection"} 1220.0 +liq_hedge_latency_ms{quantile="p99",stage="detection"} 1230.0 +liq_hedge_latency_ms{quantile="max",stage="detection"} 4000.0 +liq_hedge_latency_ms_samples{stage="detection"} 7.0 +liq_hedge_latency_ms{quantile="p50",stage="decision"} -12.0 +liq_hedge_latency_ms{quantile="p90",stage="decision"} -2.0 +liq_hedge_latency_ms{quantile="p95",stage="decision"} 8.0 +liq_hedge_latency_ms{quantile="p99",stage="decision"} 18.0 +liq_hedge_latency_ms{quantile="max",stage="decision"} 40.0 +liq_hedge_latency_ms_samples{stage="decision"} 6.0 +liq_hedge_latency_ms{quantile="p50",stage="execution"} 850.0 +liq_hedge_latency_ms{quantile="p90",stage="execution"} 860.0 +liq_hedge_latency_ms{quantile="p95",stage="execution"} 870.0 +liq_hedge_latency_ms{quantile="p99",stage="execution"} 880.0 +liq_hedge_latency_ms{quantile="max",stage="execution"} 9100.0 +liq_hedge_latency_ms_samples{stage="execution"} 5.0 +liq_hedge_latency_ms{quantile="p50",stage="exposure_window"} 3400.0 +liq_hedge_latency_ms{quantile="p90",stage="exposure_window"} 3410.0 +liq_hedge_latency_ms{quantile="p95",stage="exposure_window"} 3420.0 +liq_hedge_latency_ms{quantile="p99",stage="exposure_window"} 3430.0 +liq_hedge_latency_ms{quantile="max",stage="exposure_window"} 61000.0 +liq_hedge_latency_ms_samples{stage="exposure_window"} 5.0 +liq_hedge_fill_count_24h 7.0 +liq_hedge_cycles_24h 3.0 +liq_open_exposure_fill_count{symbol="AAPL"} 2.0 +liq_open_exposure_oldest_ts_seconds{symbol="AAPL"} 1772366400.0 +liq_open_exposure_fill_count{symbol="RKLB"} 1.0 +liq_open_exposure_oldest_ts_seconds{symbol="RKLB"} 1772368215.25 diff --git a/src/metrics/liquidity/testdata/rebalance.json b/src/metrics/liquidity/testdata/rebalance.json new file mode 100644 index 0000000000..8dc99326fd --- /dev/null +++ b/src/metrics/liquidity/testdata/rebalance.json @@ -0,0 +1,59 @@ +{ + "usdc": { + "operations": [], + "totalOperations": 2, + "skippedOperations": 0, + "stageSummary": [ + { + "stage": "conversion", + "stats": { + "p50Ms": 2000, + "p90Ms": 2010, + "p95Ms": 2020, + "p99Ms": 2030, + "maxMs": 2500, + "sampleCount": 2 + } + }, + { + "stage": "attestation", + "stats": { + "p50Ms": 52000, + "p90Ms": 52010, + "p95Ms": 52020, + "p99Ms": 52030, + "maxMs": 60000, + "sampleCount": 2 + } + } + ], + "attestationTrend": [ + { + "burnedAt": "2026-02-20T08:00:00Z", + "durationMs": 60000 + }, + { + "burnedAt": "2026-02-27T08:00:00Z", + "durationMs": 45000 + } + ] + }, + "equity": { + "operations": [], + "totalOperations": 1, + "skippedOperations": 1, + "stageSummary": [ + { + "stage": "mint_acceptance", + "stats": { + "p50Ms": 30000, + "p90Ms": 30010, + "p95Ms": 30020, + "p99Ms": 30030, + "maxMs": 30050, + "sampleCount": 1 + } + } + ] + } +} diff --git a/src/metrics/liquidity/testdata/rebalance.prom b/src/metrics/liquidity/testdata/rebalance.prom new file mode 100644 index 0000000000..49516f4420 --- /dev/null +++ b/src/metrics/liquidity/testdata/rebalance.prom @@ -0,0 +1,25 @@ +# Expected exporter output for rebalance.json, written by +# scripts/liq-parity/golden.py from t0.devops revision 226c029. +liq_rebalance_operations{kind="usdc"} 2.0 +liq_rebalance_skipped{kind="usdc"} 0.0 +liq_rebalance_stage_ms{kind="usdc",quantile="p50",stage="conversion"} 2000.0 +liq_rebalance_stage_ms{kind="usdc",quantile="p90",stage="conversion"} 2010.0 +liq_rebalance_stage_ms{kind="usdc",quantile="p95",stage="conversion"} 2020.0 +liq_rebalance_stage_ms{kind="usdc",quantile="p99",stage="conversion"} 2030.0 +liq_rebalance_stage_ms{kind="usdc",quantile="max",stage="conversion"} 2500.0 +liq_rebalance_stage_ms_samples{kind="usdc",stage="conversion"} 2.0 +liq_rebalance_stage_ms{kind="usdc",quantile="p50",stage="attestation"} 52000.0 +liq_rebalance_stage_ms{kind="usdc",quantile="p90",stage="attestation"} 52010.0 +liq_rebalance_stage_ms{kind="usdc",quantile="p95",stage="attestation"} 52020.0 +liq_rebalance_stage_ms{kind="usdc",quantile="p99",stage="attestation"} 52030.0 +liq_rebalance_stage_ms{kind="usdc",quantile="max",stage="attestation"} 60000.0 +liq_rebalance_stage_ms_samples{kind="usdc",stage="attestation"} 2.0 +liq_attestation_last_ms{kind="usdc"} 45000.0 +liq_rebalance_operations{kind="equity"} 1.0 +liq_rebalance_skipped{kind="equity"} 1.0 +liq_rebalance_stage_ms{kind="equity",quantile="p50",stage="mint_acceptance"} 30000.0 +liq_rebalance_stage_ms{kind="equity",quantile="p90",stage="mint_acceptance"} 30010.0 +liq_rebalance_stage_ms{kind="equity",quantile="p95",stage="mint_acceptance"} 30020.0 +liq_rebalance_stage_ms{kind="equity",quantile="p99",stage="mint_acceptance"} 30030.0 +liq_rebalance_stage_ms{kind="equity",quantile="max",stage="mint_acceptance"} 30050.0 +liq_rebalance_stage_ms_samples{kind="equity",stage="mint_acceptance"} 1.0 diff --git a/src/metrics/liquidity/testdata/reliability.json b/src/metrics/liquidity/testdata/reliability.json new file mode 100644 index 0000000000..ceb3a1740f --- /dev/null +++ b/src/metrics/liquidity/testdata/reliability.json @@ -0,0 +1,53 @@ +{ + "logBuckets": [ + { + "start": "2026-03-01T00:00:00Z", + "errors": 3, + "warnings": 5 + } + ], + "logTargets": [ + { + "target": "inventory", + "level": "WARN", + "count": 5, + "sparkline": [ + 5 + ] + }, + { + "target": "hedge", + "level": "ERROR", + "count": 3, + "sparkline": [ + 3 + ] + } + ], + "failureEvents": [ + { + "eventType": "OffchainOrderEvent::Failed", + "count": 2, + "lastAt": "2026-03-01T09:00:00Z" + }, + { + "eventType": "UsdcRebalanceEvent::AttestationTimedOut", + "count": 1, + "lastAt": "2026-03-01T10:00:00Z" + } + ], + "jobQueues": [ + { + "jobType": "PlaceHedge", + "pending": 1, + "running": 0, + "done": 40, + "failed": 2, + "awaitingRetry": 1, + "killed": 0, + "retried": 3, + "oldestPendingRunAt": "2026-03-01T11:00:00Z" + } + ], + "logEntriesTruncated": false +} diff --git a/src/metrics/liquidity/testdata/reliability.prom b/src/metrics/liquidity/testdata/reliability.prom new file mode 100644 index 0000000000..657ad94d00 --- /dev/null +++ b/src/metrics/liquidity/testdata/reliability.prom @@ -0,0 +1,18 @@ +# Expected exporter output for reliability.json, written by +# scripts/liq-parity/golden.py from t0.devops revision 226c029. +liq_reliability_log_count_24h{level="error"} 3.0 +liq_reliability_log_count_24h{level="warning"} 5.0 +liq_log_target_count_24h{level="WARN",target="inventory"} 5.0 +liq_log_target_count_24h{level="ERROR",target="hedge"} 3.0 +liq_failure_event_count_24h{event_type="OffchainOrderEvent::Failed"} 2.0 +liq_failure_event_last_ts_seconds{event_type="OffchainOrderEvent::Failed"} 1772355600.0 +liq_failure_event_count_24h{event_type="UsdcRebalanceEvent::AttestationTimedOut"} 1.0 +liq_failure_event_last_ts_seconds{event_type="UsdcRebalanceEvent::AttestationTimedOut"} 1772359200.0 +liq_job_queue{job_type="PlaceHedge",state="pending"} 1.0 +liq_job_queue{job_type="PlaceHedge",state="running"} 0.0 +liq_job_queue{job_type="PlaceHedge",state="done"} 40.0 +liq_job_queue{job_type="PlaceHedge",state="failed"} 2.0 +liq_job_queue{job_type="PlaceHedge",state="awaiting_retry"} 1.0 +liq_job_queue{job_type="PlaceHedge",state="killed"} 0.0 +liq_job_queue{job_type="PlaceHedge",state="retried"} 3.0 +liq_job_queue_oldest_pending_ts_seconds{job_type="PlaceHedge"} 1772362800.0