feat: publish the liq_ performance families from the bot [RAI-3090]

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

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",

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,
 }
 
 

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):

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

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<startup::StartupTask<dashboard::equity_price::EquityPriceMonitor>>,
     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),

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<PrometheusBuilder, BuildError> {
@@ -258,6 +259,11 @@ pub(crate) fn setup() -> Result<PrometheusHandle, BuildError> {
         "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 \