feat: publish the liq_pnl_* windows from the bot every 5 minutes [RAI-3093]

SPEC.md

@@ -1845,6 +1845,34 @@ Every fifth cycle the task publishes `rebalances` over the last 30 days:
 window. A collector that fails or takes longer than 30 seconds keeps its last
 family.
 
+Every 5 minutes a task publishes the `liq_pnl_*` series of six windows, each
+labelled `window`: `1d`, `1w`, `1m`, `ytd`, `1y` and `all`. Each window is one
+`GET /pnl` report built by the same code path as the endpoint, with its own
+ledger catch-up first. Every window ends on the last day with fills
+(`availableRange.lastDate`); the starts are that day, 6 days, 30 days and 364
+days before it, January 1 of its year, and the first day with fills, and no
+start is before the first day with fills. The windows run one at a time inside a
+120-second budget, `1w` first. The task has its own report admission of one, so
+it never takes one of the two live `/pnl` permits.
+
+- Each window is its own family (`collector="pnl_<window>"`). A window whose
+  report fails, times out, or does not fit the budget keeps its last samples,
+  and its collector time stops advancing. The exporter dropped such a window
+  until its next cycle; this is a deliberate difference.
+- Summary, cost, capital, per-symbol and sample series copy `/pnl`'s fields. A
+  decimal that does not parse, and a capital figure that is not computed, leave
+  their series absent. Per-symbol rows whose symbols strip to the same label
+  (`tAAPL` and `wtAAPL`) are all left out, with an error log, rather than one
+  replacing the other; the exporter wrote both lines.
+- `liq_pnl_day_*{day}` come from the report's day buckets. A day sums its
+  symbols after the prefix rule, so `tAAPL` and `wtAAPL` add up. The running
+  totals run over every bucket, and only the last 90 buckets are published, so
+  the last bucket's running total equals the window total. A symbol stays in the
+  running totals after its last day. A day component that does not parse leaves
+  out the day value it adds up and every later running total that includes it,
+  and logs an error. The exporter counted such a component as 0 (a deliberate
+  difference: a 0 would read as a real day).
+
 **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

@@ -113,6 +113,36 @@
     "liq_attestation_last_ms",
 }
 
+# Every PnL series: one family per window, `window` label on each.
+PNL_NAMES = {
+    "liq_pnl_summary_usd",
+    "liq_pnl_summary_shares",
+    "liq_pnl_summary_count",
+    "liq_pnl_cost_usd",
+    "liq_pnl_revenue_usd",
+    "liq_pnl_cost_entries",
+    "liq_pnl_cost_missing_observations",
+    "liq_pnl_cost_coverage",
+    "liq_pnl_capital_avg_deployed_usd",
+    "liq_pnl_capital_annualized_return_pct",
+    "liq_pnl_capital_coverage_days",
+    "liq_pnl_capital_sample_days",
+    "liq_pnl_symbol_usd",
+    "liq_pnl_symbol_shares",
+    "liq_pnl_symbol_lots",
+    "liq_pnl_symbol_volume_shares",
+    "liq_pnl_sample_total_fills",
+    "liq_pnl_sample_symbols",
+    "liq_pnl_sample_first_ts_seconds",
+    "liq_pnl_sample_last_ts_seconds",
+    "liq_pnl_warnings",
+    "liq_pnl_day_usd",
+    "liq_pnl_day_cum_usd",
+    "liq_pnl_day_stream_usd",
+    "liq_pnl_day_cum_stream_usd",
+}
+PORTED |= PNL_NAMES
+
 # Names only the bot publishes.
 BOT_ONLY = {
     "liq_collector_last_success_ts_seconds",
@@ -177,8 +207,12 @@
 
 # Documented differences. ("absolute", tolerance): values may differ by up to
 # the tolerance. "values": keys are compared, values are not. "ignore": the
-# name is not compared at all. The item that ports a name adds its entry, with
-# the evidence for it.
+# name is not compared at all. "kept_window": a series of a `window` the
+# exporter has no series of at all is not reported as extra in the bot.
+# "kept_window_bot_absent": as "kept_window", and a series the exporter has
+# and the bot lacks is not reported either, while the bot publishes that name
+# for the same window; values are still compared where both have the series. The item that ports a name adds its entry, with the
+# evidence for it.
 KNOWN_DIFFS = {
     # The exporter derives the start from integer uptime at poll time.
     "liq_bot_start_timestamp_seconds": ("absolute", 2.0),
@@ -219,6 +253,27 @@
     "liq_reliability_log_count_24h": "values",
     "liq_log_target_count_24h": "ignore",
 }
+# A PnL window whose report failed or did not fit the cycle budget keeps its
+# last value in the bot, with a stalled liq_collector_last_success_ts_seconds
+# {collector="pnl_<window>"}. The exporter drops that window from its pnl
+# family until its next cycle (exporter.py collect_pnl: a failed or skipped
+# window adds no samples before set_family). A deliberate difference.
+KNOWN_DIFFS.update({name: "kept_window" for name in PNL_NAMES})
+# A day component of /pnl that does not parse: the exporter counts it as 0
+# (t0.devops 8dca7be exporter.py pnl_day_samples, `dec(row.get(field)) or
+# Decimal(0)` and `dec(row.get("totalPnlUsd")) or Decimal(0)`), so it still
+# publishes that day's value and every running total after it. The bot leaves
+# those series out and logs an error (src/metrics/liquidity/pnl.rs
+# day_component; every_window_matches_the_exporter_golden pins the two series
+# the fixture's bad component feeds). So a day series can be missing from the
+# bot only. The values both sides publish are still compared.
+PNL_DAY_NAMES = {
+    "liq_pnl_day_usd",
+    "liq_pnl_day_cum_usd",
+    "liq_pnl_day_stream_usd",
+    "liq_pnl_day_cum_stream_usd",
+}
+KNOWN_DIFFS.update({name: "kept_window_bot_absent" for name in PNL_DAY_NAMES})
 
 # (bot type, exporter type) that --known-diffs accepts for every name. The
 # bot publishes every liq_* name as a gauge; the exporter is untyped.
@@ -364,11 +419,32 @@ def compared(key):
     exporter = {key: value for key, value in exporter.items()
                 if key[0] not in not_ported}
 
+    def window(key):
+        return dict(key[1]).get("window")
+
+    windowed = {"kept_window", "kept_window_bot_absent"}
+    exporter_windows = {window(key) for key in exporter
+                        if known.get(key[0]) in windowed}
+
+    def kept_window(key):
+        return known.get(key[0]) in windowed and window(key) not in exporter_windows
+
+    # A bot_absent name excuses a missing series only while the bot still
+    # publishes that name for the window: a bad day component leaves out
+    # single series, so a whole name missing from a window is a finding.
+    bot_absent_windows = {(key[0], window(key)) for key in bot
+                          if known.get(key[0]) == "kept_window_bot_absent"}
+
+    def excused_absence(key):
+        return (known.get(key[0]) == "kept_window_bot_absent"
+                and (key[0], window(key)) in bot_absent_windows)
+
     findings = type_findings(bot_text, exporter_text, dropped, known_diffs)
     for key in exporter.keys() - bot.keys():
-        findings[key] = f"missing from bot: {format_series(key)}"
+        if not excused_absence(key):
+            findings[key] = f"missing from bot: {format_series(key)}"
     for key in bot.keys() - exporter.keys():
-        if key[0] not in BOT_ONLY:
+        if key[0] not in BOT_ONLY and not kept_window(key):
             findings[key] = f"extra in bot: {format_series(key)}"
     for key in bot.keys() & exporter.keys():
         if values_differ(bot[key], exporter[key], known.get(key[0])):

scripts/liq-parity/golden.py

@@ -78,6 +78,27 @@ def run_rebalance(exporter, fixture):
     exporter.collect_rebalance_timings("from", "to")
 
 
+def run_pnl(exporter, fixture):
+    """Every /pnl call, the range probe and each window, returns the
+    fixture, so each window publishes the same report under its own
+    window label."""
+    def bot_get(path, params=None, timeout=15):
+        if path != "/pnl":
+            raise RuntimeError(f"unexpected bot read {path}")
+        return fixture
+
+    exporter.bot_get = bot_get
+    exporter.collect_pnl()
+
+
+def run_pnl_window_all(exporter, fixture):
+    """One report as the `all` window, through the exporter's two PnL
+    builders, as collect_pnl combines them."""
+    exporter.REGISTRY.set_family(
+        "pnl",
+        exporter.pnl_samples(fixture, "all") + exporter.pnl_day_samples(fixture, "all"))
+
+
 # case name -> how the fixture reaches the exporter
 CASES = {
     "settings": run_state,
@@ -88,6 +109,9 @@ def run_rebalance(exporter, fixture):
     "reliability": run_reliability,
     "infra": run_infra,
     "rebalance": run_rebalance,
+    # PnL windows and day buckets
+    "pnl": run_pnl,
+    "pnl-days": run_pnl_window_all,
 }
 
 

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_pending_orders_total, liq_pnl_summary_usd\n"
+            "liq_future_total, liq_future_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_equity_ratio_deviation, liq_pending_orders_total, "
-            "liq_pnl_summary_usd, liq_up\n"
+            "liq_asset_flags, liq_equity_ratio_deviation, liq_future_total, "
+            "liq_future_usd, liq_up\n"
             "10 finding(s)\n"
         ))
 
@@ -194,6 +194,59 @@ def test_a_mismatch_whose_values_move_between_pairs_is_still_reported(self):
             "value differs: liq_bot_start_timestamp_seconds bot=100.0 exporter=103.3",
         ])
 
+    def test_a_pnl_window_only_the_bot_kept_is_not_a_finding(self):
+        bot = ("liq_bot_info 1\n"
+               'liq_pnl_warnings{window="1w"} 2\n'
+               'liq_pnl_warnings{window="1d"} 2\n'
+               'liq_pnl_day_usd{day="2026-03-02",symbol="AAPL",window="1w"} 1\n')
+        exporter = ("liq_bot_info 1\n"
+                    'liq_pnl_warnings{window="1w"} 2\n')
+
+        findings, _ = compare.compare([(bot, exporter)], True, True)
+        self.assertEqual(findings, [
+            'extra in bot: liq_pnl_day_usd{day="2026-03-02",symbol="AAPL",window="1w"}',
+        ])
+
+        findings, _ = compare.compare([(bot, exporter)], True, False)
+        self.assertEqual(findings, [
+            'extra in bot: liq_pnl_day_usd{day="2026-03-02",symbol="AAPL",window="1w"}',
+            'extra in bot: liq_pnl_warnings{window="1d"}',
+        ])
+
+    def test_a_pnl_day_series_only_the_exporter_has_is_not_a_finding(self):
+        bot = ("liq_bot_info 1\n"
+               'liq_pnl_warnings{window="1w"} 2\n'
+               'liq_pnl_day_usd{day="2026-03-02",symbol="AAPL",window="1w"} 1\n')
+        exporter = ("liq_bot_info 1\n"
+                    'liq_pnl_warnings{window="1w"} 2\n'
+                    'liq_pnl_day_usd{day="2026-03-02",symbol="AAPL",window="1w"} 1.5\n'
+                    'liq_pnl_day_usd{day="2026-03-03",symbol="AAPL",window="1w"} 0\n'
+                    'liq_pnl_warnings{window="1d"} 2\n')
+
+        findings, _ = compare.compare([(bot, exporter)], True, True)
+        self.assertEqual(findings, [
+            'missing from bot: liq_pnl_warnings{window="1d"}',
+            'value differs: liq_pnl_day_usd{day="2026-03-02",symbol="AAPL",window="1w"} '
+            "bot=1.0 exporter=1.5",
+        ])
+
+        findings, _ = compare.compare([(bot, exporter)], True, False)
+        self.assertIn(
+            'missing from bot: liq_pnl_day_usd{day="2026-03-03",symbol="AAPL",window="1w"}',
+            findings)
+
+    def test_a_pnl_day_name_the_bot_drops_for_a_whole_window_is_a_finding(self):
+        bot = ("liq_bot_info 1\n"
+               'liq_pnl_warnings{window="1w"} 2\n')
+        exporter = ("liq_bot_info 1\n"
+                    'liq_pnl_warnings{window="1w"} 2\n'
+                    'liq_pnl_day_usd{day="2026-03-02",symbol="AAPL",window="1w"} 1\n')
+
+        findings, _ = compare.compare([(bot, exporter)], True, True)
+        self.assertEqual(findings, [
+            'missing from bot: liq_pnl_day_usd{day="2026-03-02",symbol="AAPL",window="1w"}',
+        ])
+
     def test_a_snapshot_without_liq_series_fails_instead_of_passing(self):
         empty = os.path.join(HERE, "testdata", "empty.prom")
         err = io.StringIO()
@@ -246,14 +299,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_pending_orders_total 2\n"
+        exporter = "liq_bot_info{git_commit=\"a\"} 1\nliq_future_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_pending_orders_total"])
+        self.assertEqual(findings, ["unlisted bot name: liq_future_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_pnl_summary_usd{stream="total",window="1d"} 5.0
-liq_pending_orders_total 3.0
+liq_future_usd{stream="total",window="1d"} 5.0
+liq_future_total 3.0

src/api.rs

@@ -4789,6 +4789,7 @@ mod tests {
     use crate::inventory::{
         self, BroadcastingInventory, PortfolioAsset, PortfolioBalanceRow, PortfolioLocation,
     };
+    use crate::metrics::liquidity::pnl_refresh::{LivePnlReports, PnlReports};
     use crate::offchain::order::{
         OffchainOrder, OffchainOrderEvent, OffchainOrderId, OrderPlacementResult,
     };
@@ -6133,6 +6134,30 @@ mod tests {
         assert_eq!(second.summary.matched_lot_count, 2);
     }
 
+    #[tokio::test]
+    async fn pnl_metrics_reports_run_while_live_requests_hold_every_permit() {
+        let (state, _broker_mock) = pnl_test_state().await;
+        let _live_permits = (0..crate::dashboard::pnl::MAX_CONCURRENT_PNL_REPORTS)
+            .map(|_| acquire_pnl_report_permit(&state.pnl_report_admission).unwrap())
+            .collect::<Vec<_>>();
+        let reports = LivePnlReports::new(&state.ctx, state.pool.clone(), state.pnl_ledger.clone());
+
+        let report = reports
+            .report(PnlQuery {
+                limit: Some(1),
+                ..PnlQuery::default()
+            })
+            .await
+            .unwrap();
+        assert_eq!(report.summary.matched_lot_count, 0);
+
+        let response = build_app(state)
+            .oneshot(Request::builder().uri("/pnl").body(Body::empty()).unwrap())
+            .await
+            .unwrap();
+        assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE);
+    }
+
     /// End-to-end `/pnl` coverage for capital/return-on-capital figures: three
     /// portfolio_snapshot days are captured through the real
     /// aggregate (`Store::send`, never raw SQL), each `captured_at` built

src/dashboard/pnl/mod.rs

@@ -22,7 +22,11 @@ mod tests;
 
 pub(crate) use ledger::{LedgerHead, PnlLedger, PnlLedgerReactor};
 pub(crate) use query::{PnlError, PnlQuery};
-pub(crate) use response::PnlResponse;
+#[cfg(test)]
+pub(crate) use response::{
+    PnlAvailableRange, PnlCapitalSummary, PnlCostCoverage, PnlCostSummary, PnlSampleStats,
+};
+pub(crate) use response::{PnlResponse, PnlSummary, PnlSymbolSummary, PnlWindow, PnlWindowSymbol};
 #[cfg(test)]
 pub(crate) use source::{
     MAX_CONCURRENT_PNL_REPORTS, acquire_pnl_report_permit, build_pnl_report,

src/lib.rs

@@ -411,11 +411,9 @@ async fn run_bot_session_inner(
             pool: pools.cqrs.clone(),
             families: &metrics::liquidity::LIQ_FAMILIES,
         },
-        metrics::liquidity::refresh::LiqPerformanceRefresh::new(
-            &ctx,
-            pools.cqrs.clone(),
-            &metrics::liquidity::LIQ_FAMILIES,
-        ),
+        &ctx,
+        &pools.cqrs,
+        &pnl_ledger,
     );
     let mut bot_task = tokio::spawn(Box::pin(run_conductor_session(
         ctx.clone(),
@@ -762,8 +760,18 @@ 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,
+    ctx: &Ctx,
+    pool: &SqlitePool,
+    pnl_ledger: &Arc<dashboard::pnl::PnlLedger>,
 ) -> SupervisorHandle {
+    let liq_performance_refresh = metrics::liquidity::refresh::LiqPerformanceRefresh::new(
+        ctx,
+        pool.clone(),
+        &metrics::liquidity::LIQ_FAMILIES,
+    );
+    let liq_pnl_refresh =
+        metrics::liquidity::pnl_refresh::LiqPnlRefresh::live(ctx, pool.clone(), pnl_ledger.clone());
+
     let builder = non_escalating_supervisor_builder()
         .with_task(
             "metrics-recorder-upkeep",
@@ -772,7 +780,8 @@ fn spawn_auxiliary_supervisor(
             },
         )
         .with_task("liq-state-refresh", liq_state_refresh)
-        .with_task("liq-performance-refresh", liq_performance_refresh);
+        .with_task("liq-performance-refresh", liq_performance_refresh)
+        .with_task("liq-pnl-refresh", liq_pnl_refresh);
 
     match equity_price_task {
         Some(task) => builder.with_task("dashboard-equity-prices", task),