From 6cedbffe9bed8e4537f6898de637cd0d3634f05c Mon Sep 17 00:00:00 2001 From: JuaniRios Date: Thu, 8 Oct 2026 12:50:25 -0300 Subject: [PATCH] [RAI-2998] Publish liq_ prices and exposure and refresh the inventory every 60 s --- SPEC.md | 10 +- docs/observability.md | 2 + scripts/liq-parity/compare.py | 2 + src/dashboard/equity_price.rs | 63 +++++++ src/inventory/broadcasting.rs | 21 ++- src/lib.rs | 83 +++++---- src/metrics/liquidity.rs | 39 ++++- src/metrics/liquidity/inventory.rs | 52 ++++++ src/metrics/liquidity/prices.rs | 269 +++++++++++++++++++++++++++++ src/metrics/liquidity/refresh.rs | 211 ++++++++++++++++++++++ 10 files changed, 711 insertions(+), 41 deletions(-) create mode 100644 src/metrics/liquidity/prices.rs create mode 100644 src/metrics/liquidity/refresh.rs diff --git a/SPEC.md b/SPEC.md index ae90a4afdd..a8c38c6c3e 100644 --- a/SPEC.md +++ b/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, diff --git a/docs/observability.md b/docs/observability.md index 87fb8fd634..5bde2b4d30 100644 --- a/docs/observability.md +++ b/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 diff --git a/scripts/liq-parity/compare.py b/scripts/liq-parity/compare.py index faa053e3dd..d5034db189 100644 --- a/scripts/liq-parity/compare.py +++ b/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. diff --git a/src/dashboard/equity_price.rs b/src/dashboard/equity_price.rs index 52124c3fb3..2650e59d2f 100644 --- a/src/dashboard/equity_price.rs +++ b/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) -> 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 { diff --git a/src/inventory/broadcasting.rs b/src/inventory/broadcasting.rs index c244a3a581..7ee6082177 100644 --- a/src/inventory/broadcasting.rs +++ b/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, diff --git a/src/lib.rs b/src/lib.rs index 20d235250f..c21716f51a 100644 --- a/src/lib.rs +++ b/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(id: uuid::Uuid) -> st0x_finance::Id { struct SessionTaskGuard { conductor: AbortHandle, server_supervisor: SupervisorHandle, - equity_price_supervisor: Option, + 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>, + 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( server_supervisor: SupervisorHandle, - equity_price_supervisor: Option, + auxiliary_supervisor: SupervisorHandle, mut bot_task: JoinHandle>, 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, diff --git a/src/metrics/liquidity.rs b/src/metrics/liquidity.rs index 22ec38d7bd..1c7ed7f7e4 100644 --- a/src/metrics/liquidity.rs +++ b/src/metrics/liquidity.rs @@ -26,6 +26,8 @@ use tracing::{debug, error, warn}; use st0x_float_serde::format_float; pub(crate) mod inventory; +pub(crate) mod prices; +pub(crate) mod refresh; pub(crate) mod settings; /// The process-wide store `/metrics` renders, like the recorder it sits next @@ -77,6 +79,8 @@ pub(crate) enum LiqMetric { UsdcChainAvailable, UsdcChainInflight, UsdcChainRatio, + PositionLastPriceUsd, + EquityExposureUsd, CollectorLastSuccessTsSeconds, } @@ -84,7 +88,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; 42] = [ + pub(crate) const ALL: [Self; 44] = [ Self::BotInfo, Self::BotStartTimestampSeconds, Self::SettingsInfo, @@ -126,6 +130,8 @@ impl LiqMetric { Self::UsdcChainAvailable, Self::UsdcChainInflight, Self::UsdcChainRatio, + Self::PositionLastPriceUsd, + Self::EquityExposureUsd, Self::CollectorLastSuccessTsSeconds, ]; @@ -172,6 +178,8 @@ impl LiqMetric { Self::UsdcChainAvailable => "liq_usdc_chain_available", Self::UsdcChainInflight => "liq_usdc_chain_inflight", Self::UsdcChainRatio => "liq_usdc_chain_ratio", + Self::PositionLastPriceUsd => "liq_position_last_price_usd", + Self::EquityExposureUsd => "liq_equity_exposure_usd", Self::CollectorLastSuccessTsSeconds => "liq_collector_last_success_ts_seconds", } } @@ -258,6 +266,10 @@ impl LiqMetric { Self::UsdcChainRatio => { "Vault share of itself plus gross broker cash; absent until gross is read or both 0" } + Self::PositionLastPriceUsd => { + "Live wrapped-token mid price in USD of each symbol with a position" + } + Self::EquityExposureUsd => "Net position times its live price, in USD", Self::CollectorLastSuccessTsSeconds => { "Unix time each liq_ collector last published its family" } @@ -287,7 +299,9 @@ impl LiqMetric { | Self::EquityTotal | Self::EquityUnwrapped | Self::EquityWrapped - | Self::EquityRatio => &["symbol"], + | Self::EquityRatio + | Self::PositionLastPriceUsd + | Self::EquityExposureUsd => &["symbol"], Self::EquityChainAvailable => &["chain", "symbol"], Self::UsdcCorridorTarget | Self::UsdcCorridorDeviation @@ -365,6 +379,7 @@ impl LiqMetric { | Self::UsdcChainAvailable | Self::UsdcChainInflight | Self::UsdcChainRatio => Some(LiqFamily::Inventory), + Self::PositionLastPriceUsd | Self::EquityExposureUsd => Some(LiqFamily::Prices), Self::CollectorLastSuccessTsSeconds => None, } } @@ -376,6 +391,7 @@ pub(crate) enum LiqFamily { Health, Settings, Inventory, + Prices, } impl LiqFamily { @@ -385,6 +401,7 @@ impl LiqFamily { Self::Health => "health", Self::Settings => "settings", Self::Inventory => "inventory", + Self::Prices => "prices", } } } @@ -482,6 +499,11 @@ impl LiqFamilies { self.generations.fetch_add(1, Ordering::SeqCst) + 1 } + /// The last number [`Self::next_generation`] gave out. + pub(crate) fn current_generation(&self) -> u64 { + self.generations.load(Ordering::SeqCst) + } + /// Like [`Self::replace`], for a family several publishers write: each /// read its source at `generation`, numbered in source order. A replace /// whose generation is older than the stored one is ignored and returns @@ -923,13 +945,20 @@ pub(crate) mod tests { "liq_usdc_chain_available", "liq_usdc_chain_inflight", "liq_usdc_chain_ratio", + "liq_position_last_price_usd", + "liq_equity_exposure_usd", "liq_collector_last_success_ts_seconds", ] ); assert_eq!( - [LiqFamily::Health, LiqFamily::Settings, LiqFamily::Inventory] - .map(LiqFamily::collector), - ["health", "settings", "inventory"] + [ + LiqFamily::Health, + LiqFamily::Settings, + LiqFamily::Inventory, + LiqFamily::Prices, + ] + .map(LiqFamily::collector), + ["health", "settings", "inventory", "prices"] ); } diff --git a/src/metrics/liquidity/inventory.rs b/src/metrics/liquidity/inventory.rs index 4561becf59..49ffe8eb36 100644 --- a/src/metrics/liquidity/inventory.rs +++ b/src/metrics/liquidity/inventory.rs @@ -103,6 +103,17 @@ impl InventoryPublisher { } } + /// Reads the view again without a change, for the periodic republish. + /// Call it while a read lock is held: no write can then move the + /// generation, so the read carries the generation of the last write. + /// `None` before [`Self::start`]: boot has not restored the view yet. + pub(crate) fn read_current(&self, view: &InventoryView) -> Option { + self.started().then(|| InventoryRead { + generation: self.families.current_generation(), + input: InventoryInput::from_view(view), + }) + } + /// Builds the samples and stores them, unless a newer read was stored /// first. Call it after the lock is released. pub(crate) fn publish(&self, read: InventoryRead) { @@ -1041,6 +1052,47 @@ pub(crate) mod tests { ); } + #[test] + fn nothing_is_republished_before_the_start() { + let publisher = publisher(leaked_families()); + + assert!( + publisher + .read_current(&view_with("AAPL", float!(1))) + .is_none() + ); + + publisher.publish(publisher.start(&view_with("AAPL", float!(1)))); + assert!( + publisher + .read_current(&view_with("AAPL", float!(1))) + .is_some() + ); + } + + /// The periodic republish reads at the last write's generation, so a + /// write that lands between its read and its publish wins. + #[test] + fn a_republish_read_before_a_newer_write_is_ignored() { + let families = leaked_families(); + let publisher = publisher(families); + publisher.publish(publisher.start(&view_with("AAPL", float!(1)))); + + let republish = publisher + .read_current(&view_with("AAPL", float!(1))) + .unwrap(); + publisher.publish(publisher.read_changed(&view_with("AAPL", float!(2)))); + publisher.publish(republish); + + assert_eq!( + rendered_store(families).get(&series( + "liq_equity_onchain_available", + &[("symbol", "AAPL")] + )), + Some(&2.0) + ); + } + #[test] fn a_symbol_that_leaves_the_view_leaves_the_series() { let families = leaked_families(); diff --git a/src/metrics/liquidity/prices.rs b/src/metrics/liquidity/prices.rs new file mode 100644 index 0000000000..a80a6f9512 --- /dev/null +++ b/src/metrics/liquidity/prices.rs @@ -0,0 +1,269 @@ +//! The `Prices` family: the live price of each symbol the bot holds a +//! position in, and the position's dollar exposure at that price. +//! +//! Refreshed every 60 seconds by `liq-state-refresh`, like the exporter. A +//! symbol with a price but no position, or a position without a live price, +//! publishes nothing for the missing half. + +use std::collections::HashMap; + +use rain_math_float::Float; +use sqlx::SqlitePool; +use tracing::warn; + +use st0x_event_sorcery::{LoadAllIdsError, SendError, load_all_ids, load_entity}; +use st0x_finance::Symbol; + +use super::{LiqMetric, LiqSample, float_value, push_sample, strip_prefix}; +use crate::position::Position; + +/// One position as the prices builder reads it, owned by the metrics so it +/// does not depend on the dashboard DTO. +#[derive(Debug, Clone)] +pub(crate) struct PositionInput { + pub(crate) symbol: Symbol, + pub(crate) net: Float, +} + +#[derive(Debug, thiserror::Error)] +pub(crate) enum PositionLoadError { + #[error("failed to list positions")] + Ids(#[from] LoadAllIdsError), + #[error("failed to load position {id}")] + Position { + id: Symbol, + #[source] + source: Box>, + }, +} + +/// Every position, or an error. The dashboard's loader turns errors into a +/// shorter list, which would drop exposure series; here a failure keeps the +/// last published family instead. +pub(crate) async fn load_positions( + pool: &SqlitePool, +) -> Result, PositionLoadError> { + let mut positions = Vec::new(); + + for id in load_all_ids::(pool).await? { + match load_entity::(pool, &id).await { + Ok(Some(position)) => positions.push(PositionInput { + symbol: position.symbol, + net: position.net.inner(), + }), + // Listed but not loadable as a live aggregate: the dashboard + // skips it too, so the exporter never saw it. + Ok(None) => warn!(%id, "Position listed without a live aggregate"), + Err(source) => { + return Err(PositionLoadError::Position { + id, + source: Box::new(source), + }); + } + } + } + + Ok(positions) +} + +/// `liq_position_last_price_usd` and `liq_equity_exposure_usd` per position. +/// Positions and prices join on the label symbol, as the exporter joined +/// them, so `tAAPL` prices `AAPL`. +pub(crate) fn price_samples( + positions: &[PositionInput], + live_prices: &[(Symbol, Float)], +) -> Vec { + let prices: HashMap<&str, Float> = live_prices + .iter() + .map(|(symbol, price)| (strip_prefix(symbol.as_str()), *price)) + .collect(); + + let mut samples = Vec::new(); + for position in positions { + let symbol = strip_prefix(position.symbol.as_str()); + let Some(price) = prices.get(symbol).copied() else { + continue; + }; + + let labels = vec![("symbol", symbol.to_string())]; + push_sample( + &mut samples, + LiqMetric::PositionLastPriceUsd, + labels.clone(), + float_value(price), + ); + push_sample( + &mut samples, + LiqMetric::EquityExposureUsd, + labels, + (position.net * price) + .map_err(Into::into) + .and_then(float_value), + ); + } + + samples +} + +#[cfg(test)] +pub(crate) mod tests { + use std::collections::BTreeMap; + + use alloy::primitives::TxHash; + use chrono::{DateTime, Utc}; + + use st0x_config::ExecutionThreshold; + use st0x_event_sorcery::StoreBuilder; + use st0x_evm::Chain; + use st0x_execution::{Direction, FractionalShares}; + use st0x_float_macro::float; + + use super::*; + use crate::metrics::liquidity::LiqFamily; + use crate::metrics::liquidity::inventory::tests::{ + golden_for, render_family, state_fixture, state_reserved_fixture, + }; + use crate::metrics::liquidity::tests::series; + use crate::position::{PositionCommand, TradeId}; + use crate::test_utils::setup_test_db; + + fn position(symbol: &str, net: Float) -> PositionInput { + PositionInput { + symbol: Symbol::new(symbol).unwrap(), + net, + } + } + + fn price(symbol: &str, price: Float) -> (Symbol, Float) { + (Symbol::new(symbol).unwrap(), price) + } + + /// The exporter's live-price rule over the dashboard DTO: an available + /// price that has not expired at `now`. + fn live_prices_from_dto( + prices: &[st0x_dto::EquityPrice], + now: DateTime, + ) -> Vec<(Symbol, Float)> { + prices + .iter() + .filter_map(|price| match price.status { + st0x_dto::EquityPriceStatus::Available { + price_usd, + expires_at, + .. + } if expires_at > now => Some((price.symbol.clone(), price_usd)), + st0x_dto::EquityPriceStatus::Available { .. } + | st0x_dto::EquityPriceStatus::Unavailable => None, + }) + .collect() + } + + #[test] + fn prices_match_the_exporter_goldens() { + let now: DateTime = "2026-10-08T12:00:00Z".parse().unwrap(); + + for (fixture, golden) in [ + (state_fixture(), include_str!("testdata/state.prom")), + ( + state_reserved_fixture(), + include_str!("testdata/state-reserved.prom"), + ), + ] { + let positions: Vec = fixture + .positions + .iter() + .map(|position| PositionInput { + symbol: position.symbol.clone(), + net: position.net, + }) + .collect(); + let live = live_prices_from_dto(&fixture.equity_prices, now); + + assert_eq!( + render_family(LiqFamily::Prices, price_samples(&positions, &live)), + golden_for(LiqFamily::Prices, golden) + ); + } + } + + #[test] + fn only_symbols_with_both_a_position_and_a_live_price_publish() { + let samples = price_samples( + &[position("tAAPL", float!(10)), position("TSLA", float!(-3))], + &[price("AAPL", float!(2.5)), price("tSPYM", float!(7))], + ); + + assert_eq!( + render_family(LiqFamily::Prices, samples), + BTreeMap::from([ + ( + series("liq_position_last_price_usd", &[("symbol", "AAPL")]), + 2.5 + ), + ( + series("liq_equity_exposure_usd", &[("symbol", "AAPL")]), + 25.0 + ), + ]) + ); + } + + #[tokio::test] + async fn load_positions_reads_every_position_aggregate() { + let pool = setup_test_db().await; + let (store, _projection) = StoreBuilder::::new(pool.clone()) + .build(()) + .await + .unwrap(); + for (name, direction) in [("AAPL", Direction::Buy), ("TSLA", Direction::Sell)] { + let symbol = Symbol::new(name).unwrap(); + store + .send( + &symbol, + PositionCommand::AcknowledgeOnChainFill { + symbol: symbol.clone(), + threshold: ExecutionThreshold::whole_share(), + trade_id: TradeId { + chain: Chain::Base, + tx_hash: TxHash::random(), + log_index: 1, + }, + amount: FractionalShares::new(float!(0.5)), + direction, + price_usdc: float!(150), + block_timestamp: Utc::now(), + block_number: None, + }, + ) + .await + .unwrap(); + } + + let mut loaded: Vec<(String, String)> = load_positions(&pool) + .await + .unwrap() + .into_iter() + .map(|position| (position.symbol.to_string(), position.net.format().unwrap())) + .collect(); + loaded.sort(); + + assert_eq!( + loaded, + [ + ("AAPL".to_string(), "0.5".to_string()), + ("TSLA".to_string(), "-0.5".to_string()), + ] + ); + } + + #[tokio::test] + async fn load_positions_fails_instead_of_returning_a_shorter_list() { + let pool = setup_test_db().await; + pool.close().await; + + assert!(matches!( + load_positions(&pool).await, + Err(PositionLoadError::Ids(_)) + )); + } +} diff --git a/src/metrics/liquidity/refresh.rs b/src/metrics/liquidity/refresh.rs new file mode 100644 index 0000000000..8312fd29c1 --- /dev/null +++ b/src/metrics/liquidity/refresh.rs @@ -0,0 +1,211 @@ +//! Periodic `liq_*` refresh tasks. They run under a non-escalating +//! supervisor: a metrics failure never stops trading. + +use std::sync::Arc; +use std::time::{Duration, SystemTime}; + +use chrono::Utc; +use sqlx::SqlitePool; +use task_supervisor::{SupervisedTask, TaskResult}; +use tokio::time::{MissedTickBehavior, timeout}; +use tracing::{info, warn}; + +use super::prices::{load_positions, price_samples}; +use super::{LiqFamilies, LiqFamily}; +use crate::dashboard::equity_price::EquityPriceStore; +use crate::inventory::BroadcastingInventory; + +/// The exporter's cadence and the scrape interval. +const STATE_REFRESH_PERIOD: Duration = Duration::from_secs(60); + +/// Longest a positions load may take before the cycle keeps the last prices. +const POSITION_LOAD_TIMEOUT: Duration = Duration::from_secs(30); + +/// Every 60 seconds: republishes the inventory series, which covers quiet +/// periods, and refreshes the prices and exposure series. +#[derive(Clone)] +pub(crate) struct LiqStateRefresh { + pub(crate) inventory: Arc, + pub(crate) equity_prices: EquityPriceStore, + pub(crate) pool: SqlitePool, + pub(crate) families: &'static LiqFamilies, +} + +impl LiqStateRefresh { + async fn refresh_once(&self) { + self.inventory.republish_liq_metrics().await; + self.refresh_prices().await; + } + + /// A failed or slow positions load keeps the last published prices, so + /// exposure never drops to a shorter list. + async fn refresh_prices(&self) { + let positions = match timeout(POSITION_LOAD_TIMEOUT, load_positions(&self.pool)).await { + Ok(Ok(positions)) => positions, + Ok(Err(error)) => { + warn!( + ?error, + "Kept the last liq_ prices: positions failed to load" + ); + return; + } + Err(_) => { + warn!( + timeout = ?POSITION_LOAD_TIMEOUT, + "Kept the last liq_ prices: positions took too long to load" + ); + return; + } + }; + + let live_prices = self.equity_prices.live_prices(Utc::now()).await; + self.families.replace( + LiqFamily::Prices, + price_samples(&positions, &live_prices), + SystemTime::now(), + ); + } +} + +impl SupervisedTask for LiqStateRefresh { + async fn run(&mut self) -> TaskResult { + info!("liq_ state refresh started"); + + let mut interval = tokio::time::interval(STATE_REFRESH_PERIOD); + interval.set_missed_tick_behavior(MissedTickBehavior::Skip); + + loop { + interval.tick().await; + self.refresh_once().await; + } + } +} + +#[cfg(test)] +mod tests { + use tokio::sync::broadcast; + + use st0x_execution::{FractionalShares, Symbol}; + use st0x_float_macro::float; + + use super::*; + use crate::inventory::InventoryView; + use crate::metrics::liquidity::inventory::tests::{leaked_families, publisher, rendered_store}; + use crate::metrics::liquidity::tests::series; + use crate::test_utils::setup_test_db; + + async fn refresh( + families: &'static LiqFamilies, + equity_prices: EquityPriceStore, + ) -> LiqStateRefresh { + let (sender, _) = broadcast::channel(1); + let view = InventoryView::default().with_equity( + Symbol::new("AAPL").unwrap(), + FractionalShares::new(float!(7)), + FractionalShares::ZERO, + ); + + LiqStateRefresh { + inventory: Arc::new( + BroadcastingInventory::new(view, sender) + .publishing_liq_metrics(publisher(families)), + ), + equity_prices, + pool: setup_test_db().await, + families, + } + } + + fn collector(families: &'static LiqFamilies, name: &str) -> Option { + rendered_store(families) + .get(&series( + "liq_collector_last_success_ts_seconds", + &[("collector", name)], + )) + .copied() + } + + async fn run_first_tick(task: LiqStateRefresh, families: &'static LiqFamilies) { + let mut task = task; + let running = tokio::spawn(async move { task.run().await }); + tokio::time::timeout(Duration::from_secs(5), async { + while collector(families, "prices").is_none() { + tokio::time::sleep(Duration::from_millis(5)).await; + } + }) + .await + .unwrap(); + running.abort(); + } + + fn aapl_onchain(families: &'static LiqFamilies) -> Option { + rendered_store(families) + .get(&series( + "liq_equity_onchain_available", + &[("symbol", "AAPL")], + )) + .copied() + } + + /// The first tick runs at once and puts back the inventory of the last + /// write, here after the stored family was lost. + #[tokio::test] + async fn the_first_tick_republishes_the_inventory_and_publishes_prices() { + let families = leaked_families(); + let task = refresh(families, EquityPriceStore::new([])).await; + task.inventory.start_publishing_liq_metrics().await; + families.replace(LiqFamily::Inventory, Vec::new(), SystemTime::UNIX_EPOCH); + assert_eq!(aapl_onchain(families), None); + + run_first_tick(task, families).await; + + assert_eq!(aapl_onchain(families), Some(7.0)); + } + + /// Before boot restores the view there is nothing to publish: an empty + /// view would read as zero balances. + #[tokio::test] + async fn the_inventory_is_not_published_before_the_start() { + let families = leaked_families(); + let task = refresh(families, EquityPriceStore::new([])).await; + + run_first_tick(task, families).await; + + assert_eq!(collector(families, "inventory"), None); + } + + #[tokio::test] + async fn a_failed_positions_load_keeps_the_last_prices() { + let families = leaked_families(); + let task = refresh( + families, + EquityPriceStore::with_live_mark(Symbol::new("AAPL").unwrap(), float!(2.5)), + ) + .await; + let previous = vec![ + crate::metrics::liquidity::LiqSample::new( + crate::metrics::liquidity::LiqMetric::PositionLastPriceUsd, + vec![("symbol", "AAPL".to_string())], + 2.0, + ) + .unwrap(), + ]; + families.replace( + LiqFamily::Prices, + previous, + SystemTime::UNIX_EPOCH + Duration::from_secs(100), + ); + task.pool.close().await; + + task.refresh_prices().await; + + assert_eq!( + rendered_store(families).get(&series( + "liq_position_last_price_usd", + &[("symbol", "AAPL")] + )), + Some(&2.0) + ); + assert_eq!(collector(families, "prices"), Some(100.0)); + } +}