feat: publish liq_ prices and exposure and refresh the inventory every 60 s [RAI-3088]

SPEC.md

@@ -1772,7 +1772,15 @@ write order, and the store ignores a copy older than the one it holds, so a slow
 publisher never puts older balances back. Publishing starts once boot has
 restored the inventory from its snapshots: before that the view is empty, and
 its series would read as zero balances. If boot fails to read a snapshot, the
-inventory series stay unpublished until the next start and an error is logged.
+inventory series stay unpublished until the next start and an error is logged. A
+task republishes the inventory every 60 seconds, which covers quiet periods.
+
+Every 60 seconds the same task publishes `liq_position_last_price_usd{symbol}`
+and `liq_equity_exposure_usd{symbol}` (net position times that price) for each
+symbol with a position. The price is the live, unexpired wrapped-token price
+from the pricing feed; positions and prices join on the `symbol` label. A symbol
+with a price but no position, or a position without a live price, has no series.
+If the positions fail to load, the last published prices stay.
 
 - `liq_equity_*{symbol}` and the unlabelled `liq_usdc_*` series keep the
   exporter's meaning: the onchain values are the primary chain's vault only,

docs/observability.md

@@ -168,6 +168,8 @@ second writer. A test scans `src/` for this.
 | `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                                                           |
 
 `symbol` labels drop a leading `wt` or `t` only before an uppercase letter

scripts/liq-parity/compare.py

@@ -90,6 +90,8 @@
     "liq_usdc_total",
     "liq_usdc_ratio",
     "liq_usdc_rebalanceable",
+    "liq_position_last_price_usd",
+    "liq_equity_exposure_usd",
 }
 
 # Names only the bot publishes.

src/dashboard/equity_price.rs

@@ -190,6 +190,25 @@ impl EquityPriceStore {
         snapshot
     }
 
+    /// Every symbol's wrapped-token price that has not expired at `now`, in
+    /// symbol order.
+    pub(crate) async fn live_prices(&self, now: DateTime<Utc>) -> Vec<(Symbol, Float)> {
+        let mut live: Vec<(Symbol, Float)> = self
+            .prices
+            .read()
+            .await
+            .iter()
+            .filter_map(|(symbol, price)| {
+                price
+                    .as_ref()
+                    .filter(|price| price.expires_at > now)
+                    .map(|price| (symbol.clone(), price.price_usd))
+            })
+            .collect();
+        live.sort_by(|(left, _), (right, _)| left.cmp(right));
+        live
+    }
+
     /// A store holding one live mark for `symbol`, observed now, with a
     /// wrapper ratio of 1.
     #[cfg(test)]
@@ -909,6 +928,50 @@ mod tests {
 
     use super::*;
 
+    #[tokio::test]
+    async fn live_prices_lists_unexpired_prices_in_symbol_order() {
+        let now = Utc::now();
+        let available = |price_usd, expires_at| {
+            Some(AvailablePrice {
+                price_usd,
+                underlying_price_usd: None,
+                observed_at: now,
+                expires_at,
+            })
+        };
+        let store = EquityPriceStore {
+            prices: Arc::new(RwLock::new(HashMap::from([
+                (
+                    Symbol::new("TSLA").unwrap(),
+                    available(float!(9), now + TimeDelta::seconds(30)),
+                ),
+                (
+                    Symbol::new("AAPL").unwrap(),
+                    available(float!(2.5), now + TimeDelta::seconds(30)),
+                ),
+                (Symbol::new("RKLB").unwrap(), available(float!(4), now)),
+                (Symbol::new("SPYM").unwrap(), None),
+            ]))),
+            mark_listener: Arc::default(),
+            last_marks: Arc::default(),
+        };
+
+        let live: Vec<(String, String)> = store
+            .live_prices(now)
+            .await
+            .into_iter()
+            .map(|(symbol, price)| (symbol.to_string(), price.format().unwrap()))
+            .collect();
+
+        assert_eq!(
+            live,
+            [
+                ("AAPL".to_string(), "2.5".to_string()),
+                ("TSLA".to_string(), "9".to_string()),
+            ]
+        );
+    }
+
     struct AssertDashboardAuthorization;
 
     impl Callback for AssertDashboardAuthorization {

src/inventory/broadcasting.rs

@@ -9,7 +9,7 @@ use std::ops::{Deref, DerefMut};
 
 use chrono::Utc;
 use tokio::sync::{RwLock, RwLockReadGuard, RwLockWriteGuard, broadcast};
-use tracing::warn;
+use tracing::{debug, warn};
 
 use st0x_dto::{InventorySnapshot, Statement};
 
@@ -59,6 +59,25 @@ impl BroadcastingInventory {
         self.view.read().await
     }
 
+    /// Publishes the `liq_*` inventory series again from the current view.
+    /// Covers quiet periods and any write path that skipped the publish; a
+    /// write that lands meanwhile carries a newer generation and wins.
+    pub(crate) async fn republish_liq_metrics(&self) {
+        let Some(publisher) = &self.liq_metrics else {
+            warn!(target: "inventory", "No liq_ inventory publisher attached; skipped the republish");
+            return;
+        };
+
+        let Some(read) = publisher.read_current(&*self.view.read().await) else {
+            debug!(
+                target: "inventory",
+                "Skipped the liq_ inventory republish: boot has not restored the inventory yet"
+            );
+            return;
+        };
+        publisher.publish(read);
+    }
+
     pub(crate) async fn write(&self) -> BroadcastingWriteGuard<'_> {
         BroadcastingWriteGuard {
             guard: self.view.write().await,

src/lib.rs

@@ -401,12 +401,15 @@ async fn run_bot_session_inner(
         startup_barrier.token(),
         startup_barrier.token(),
     );
-    let equity_price_supervisor = equity_price_task.map(|task| {
-        non_escalating_supervisor_builder()
-            .with_task("dashboard-equity-prices", task)
-            .build()
-            .run()
-    });
+    let auxiliary_supervisor = spawn_auxiliary_supervisor(
+        equity_price_task,
+        metrics::liquidity::refresh::LiqStateRefresh {
+            inventory: inventory.clone(),
+            equity_prices: equity_prices.clone(),
+            pool: pools.cqrs.clone(),
+            families: &metrics::liquidity::LIQ_FAMILIES,
+        },
+    );
     let mut bot_task = tokio::spawn(Box::pin(run_conductor_session(
         ctx.clone(),
         pools,
@@ -439,7 +442,7 @@ async fn run_bot_session_inner(
     let mut session_guard = SessionTaskGuard {
         conductor: bot_task.abort_handle(),
         server_supervisor: server_supervisor.clone(),
-        equity_price_supervisor: equity_price_supervisor.clone(),
+        auxiliary_supervisor: auxiliary_supervisor.clone(),
         armed: true,
     };
 
@@ -488,7 +491,7 @@ async fn run_bot_session_inner(
             StartupOutcome::Started => {
                 await_shutdown(
                     server_supervisor,
-                    equity_price_supervisor,
+                    auxiliary_supervisor,
                     bot_task,
                     &detached_tasks,
                     shutdown_token,
@@ -498,7 +501,7 @@ async fn run_bot_session_inner(
                 .await
             }
             StartupOutcome::ShutdownSignal => {
-                shutdown_equity_price_supervisor(equity_price_supervisor.as_ref());
+                shutdown_auxiliary_supervisor(&auxiliary_supervisor);
                 drain_for_shutdown_signal(
                     &server_supervisor,
                     bot_task,
@@ -609,7 +612,7 @@ fn transfer_id<Tag>(id: uuid::Uuid) -> st0x_finance::Id<Tag> {
 struct SessionTaskGuard {
     conductor: AbortHandle,
     server_supervisor: SupervisorHandle,
-    equity_price_supervisor: Option<SupervisorHandle>,
+    auxiliary_supervisor: SupervisorHandle,
     armed: bool,
 }
 
@@ -627,7 +630,7 @@ impl Drop for SessionTaskGuard {
 
         self.conductor.abort();
         shutdown_supervisor(&self.server_supervisor);
-        shutdown_equity_price_supervisor(self.equity_price_supervisor.as_ref());
+        shutdown_auxiliary_supervisor(&self.auxiliary_supervisor);
     }
 }
 
@@ -745,13 +748,30 @@ fn spawn_server_supervisor(
         .run()
 }
 
+/// The supervisor for tasks that must never stop trading: the dashboard
+/// price feed (when pricing is configured) and the `liq_*` refresh.
+fn spawn_auxiliary_supervisor(
+    equity_price_task: Option<startup::StartupTask<dashboard::equity_price::EquityPriceMonitor>>,
+    liq_state_refresh: metrics::liquidity::refresh::LiqStateRefresh,
+) -> SupervisorHandle {
+    let builder =
+        non_escalating_supervisor_builder().with_task("liq-state-refresh", liq_state_refresh);
+
+    match equity_price_task {
+        Some(task) => builder.with_task("dashboard-equity-prices", task),
+        None => builder,
+    }
+    .build()
+    .run()
+}
+
 fn non_escalating_supervisor_builder() -> SupervisorBuilder {
     SupervisorBuilder::default().with_unlimited_restarts()
 }
 
 async fn await_shutdown<S>(
     server_supervisor: SupervisorHandle,
-    equity_price_supervisor: Option<SupervisorHandle>,
+    auxiliary_supervisor: SupervisorHandle,
     mut bot_task: JoinHandle<anyhow::Result<()>>,
     detached_tasks: &TaskTracker,
     shutdown_token: CancellationToken,
@@ -768,7 +788,7 @@ where
         result = server_supervisor.wait() => ShutdownTrigger::ServerExit(result),
         result = &mut bot_task => ShutdownTrigger::BotExit(result),
     };
-    shutdown_equity_price_supervisor(equity_price_supervisor.as_ref());
+    shutdown_auxiliary_supervisor(&auxiliary_supervisor);
 
     match trigger {
         ShutdownTrigger::Signal => {
@@ -914,10 +934,8 @@ fn shutdown_supervisor(handle: &SupervisorHandle) {
     shutdown_named_supervisor("server", handle);
 }
 
-fn shutdown_equity_price_supervisor(handle: Option<&SupervisorHandle>) {
-    if let Some(handle) = handle {
-        shutdown_named_supervisor("dashboard-equity-prices", handle);
-    }
+fn shutdown_auxiliary_supervisor(handle: &SupervisorHandle) {
+    shutdown_named_supervisor("auxiliary", handle);
 }
 
 fn shutdown_named_supervisor(name: &'static str, handle: &SupervisorHandle) {
@@ -1297,7 +1315,7 @@ mod tests {
         let mut guard = SessionTaskGuard {
             conductor: conductor_task.abort_handle(),
             server_supervisor: supervisor.clone(),
-            equity_price_supervisor: None,
+            auxiliary_supervisor: pending_test_supervisor(),
             armed: true,
         };
 
@@ -1332,7 +1350,7 @@ mod tests {
         let guard = SessionTaskGuard {
             conductor: conductor_task.abort_handle(),
             server_supervisor: supervisor.clone(),
-            equity_price_supervisor: None,
+            auxiliary_supervisor: pending_test_supervisor(),
             armed: true,
         };
 
@@ -1366,7 +1384,7 @@ mod tests {
             let mut guard = SessionTaskGuard {
                 conductor: conductor_abort,
                 server_supervisor: supervisor,
-                equity_price_supervisor: None,
+                auxiliary_supervisor: pending_test_supervisor(),
                 armed: true,
             };
             let coordinated_shutdown = async move {
@@ -1425,8 +1443,8 @@ mod tests {
 
         let supervisor = SupervisorBuilder::default().build().run();
         let supervisor_handle_for_assert = supervisor.clone();
-        let equity_price_supervisor = pending_test_supervisor();
-        let equity_price_handle_for_assert = equity_price_supervisor.clone();
+        let auxiliary_supervisor = pending_test_supervisor();
+        let auxiliary_handle_for_assert = auxiliary_supervisor.clone();
 
         let (signal_tx, signal_rx) = tokio::sync::oneshot::channel::<()>();
         let signal_fut = async move {
@@ -1440,7 +1458,7 @@ mod tests {
 
         await_shutdown(
             supervisor,
-            Some(equity_price_supervisor),
+            auxiliary_supervisor,
             bot_task,
             &TaskTracker::new(),
             shutdown_token,
@@ -1461,14 +1479,11 @@ mod tests {
             post_wait.is_ok(),
             "supervisor.wait() should return after signal-triggered shutdown"
         );
-        let equity_price_post_wait = tokio::time::timeout(
-            Duration::from_secs(1),
-            equity_price_handle_for_assert.wait(),
-        )
-        .await;
+        let auxiliary_post_wait =
+            tokio::time::timeout(Duration::from_secs(1), auxiliary_handle_for_assert.wait()).await;
         assert!(
-            equity_price_post_wait.is_ok(),
-            "equity price supervisor should stop during signal-triggered shutdown"
+            auxiliary_post_wait.is_ok(),
+            "auxiliary supervisor should stop during signal-triggered shutdown"
         );
     }
 
@@ -1483,7 +1498,7 @@ mod tests {
 
         await_shutdown(
             supervisor,
-            None,
+            pending_test_supervisor(),
             bot_task,
             &TaskTracker::new(),
             shutdown_token,
@@ -1525,7 +1540,7 @@ mod tests {
 
         await_shutdown(
             supervisor,
-            None,
+            pending_test_supervisor(),
             bot_task,
             &TaskTracker::new(),
             shutdown_token,
@@ -1603,7 +1618,7 @@ mod tests {
                 async move {
                     await_shutdown(
                         supervisor,
-                        None,
+                        pending_test_supervisor(),
                         bot_task,
                         &detached_tasks,
                         shutdown_token,
@@ -1660,7 +1675,7 @@ mod tests {
 
         await_shutdown(
             supervisor,
-            None,
+            pending_test_supervisor(),
             bot_task,
             &TaskTracker::new(),
             shutdown_token,