feat: mark-to-market P&L with multi-provider quote feeds and ranked failover

crates/dataprovider/src/lib.rs

@@ -4,6 +4,7 @@
 //!   - [`DataProvider`]   — base identity (`code()`).
 //!   - [`UniverseSource`] — discover + price + fetch instrument definitions.
 //!   - [`Enricher`]       — fill the [`Identifiers`] bag from a metadata endpoint.
+//!   - [`LiveQuoteFeed`]  — stream normalized top-of-book [`Quote`]s.
 //!
 //! The seeder (in the `rustoms` app) drives a set of providers and a
 //! `Vec<Box<dyn Enricher>>`. Adding a vendor = new `impl UniverseSource`; adding
@@ -16,6 +17,7 @@ mod enrich;
 mod error;
 mod instrument;
 mod provider;
+mod quote;
 mod universe;
 
 pub mod enrichers;
@@ -25,6 +27,7 @@ pub use enrich::{EnrichReport, Enricher};
 pub use error::ProviderError;
 pub use instrument::{DerivativeDef, Identifiers, InstrumentDef, OptionKind};
 pub use provider::DataProvider;
+pub use quote::{FeedHealth, LiveQuoteFeed, NoFeedHealth, Quote, SymbolAdds};
 pub use universe::{Category, CostEstimate, SType, UniverseSource, UniverseSpec};
 
 pub use enrichers::openfigi::OpenFigiEnricher;

crates/dataprovider/src/quote.rs

@@ -0,0 +1,95 @@
+//! Live quote capability — the "quote source" this crate's [`crate::DataProvider`]
+//! docs have always named but never had.
+//!
+//! A feed's only job is to turn a vendor's wire format into [`Quote`]s and emit
+//! them. It does not decide what to subscribe to (the caller passes a symbol set),
+//! does not touch the database, and does not know who consumes the quotes — that
+//! keeps vendor detail contained and lets a feed be tested with nothing but a
+//! channel. Databento's session-scoped numeric ids, Binance's lowercase pair
+//! strings: neither concept escapes into this interface, because neither means
+//! anything to the other vendor.
+
+use std::collections::HashMap;
+
+use chrono::{DateTime, Utc};
+use tokio::sync::mpsc;
+
+use crate::error::ProviderError;
+use crate::provider::DataProvider;
+
+/// A normalized top-of-book quote, already resolved to our instrument id.
+#[derive(Debug, Clone, Copy)]
+pub struct Quote {
+    pub instrument_id: i64,
+    pub bid: f64,
+    pub ask: f64,
+    pub bid_size: u32,
+    pub ask_size: u32,
+    /// When we received it. This is the staleness clock — deliberately our clock,
+    /// not the venue's, so a vendor with a skewed or absent timestamp cannot make
+    /// a stale quote look fresh.
+    pub ts_recv: DateTime<Utc>,
+    /// Which feed produced this. A mark whose origin is unknown is not auditable,
+    /// and failover cannot arbitrate between sources it cannot name.
+    pub source_code: &'static str,
+}
+
+impl Quote {
+    pub fn mid(&self) -> f64 {
+        (self.bid + self.ask) / 2.0
+    }
+}
+
+/// New symbols to subscribe mid-session, as `external_symbol -> instrument_id`.
+/// Carries the id because the feed must map its own wire identity back to ours.
+pub type SymbolAdds = HashMap<String, i64>;
+
+/// Liveness reporting for a feed session.
+///
+/// A port, so this crate stays ignorant of how the host tracks health. Feeds must
+/// report both: `on_connected` is the only thing that can distinguish "subscribed
+/// and waiting on a quiet book" from "never got up", and `on_event` is the
+/// freshness clock a failover policy reads to decide a source has gone stale.
+pub trait FeedHealth: Send + Sync {
+    /// Subscribed and the venue accepted the session.
+    fn on_connected(&self);
+    /// A frame arrived. Called per record, including ones we discard.
+    fn on_event(&self);
+}
+
+/// A [`FeedHealth`] that reports nowhere — for tests and for hosts that don't
+/// track liveness.
+pub struct NoFeedHealth;
+
+impl FeedHealth for NoFeedHealth {
+    fn on_connected(&self) {}
+    fn on_event(&self) {}
+}
+
+/// A source of live quotes.
+#[async_trait::async_trait]
+pub trait LiveQuoteFeed: DataProvider {
+    /// The `instrument_class` values this feed can quote.
+    ///
+    /// `code()` alone is not coverage: a vendor spans many datasets, and a feed is
+    /// one of them. Databento cross-references both equities and options, but the
+    /// OPRA.PILLAR session can only quote options — handing it an equity symbol
+    /// would at best return nothing and at worst have the gateway reject the whole
+    /// subscription. This is the minimum scope a driver needs to pick the right
+    /// held instruments; a per-feed dataset/schema config is the fuller answer when
+    /// one vendor runs several feeds.
+    fn covers(&self) -> &'static [&'static str];
+    /// Connect, subscribe `symbols`, and emit quotes until the session ends.
+    ///
+    /// `Ok(())` means the venue closed the stream cleanly; the supervisor
+    /// reconnects either way. `add_rx` delivers symbols that became interesting
+    /// after the session started (a new position opened) — each feed subscribes
+    /// them however its protocol allows.
+    async fn run_session(
+        &self,
+        symbols: HashMap<String, i64>,
+        out: &mpsc::Sender<Quote>,
+        add_rx: &mut mpsc::Receiver<SymbolAdds>,
+        health: &dyn FeedHealth,
+    ) -> Result<(), ProviderError>;
+}

db/migrations/ods/oms/0018_CREATE_PROVIDER_FEED_POLICY.sql

@@ -0,0 +1,37 @@
+-- Which market-data source wins when several can price the same instrument.
+--
+-- Policy, not per-instrument data: ranking every instrument individually would be
+-- ~42k copies of the same integer, and the answer is almost always a property of
+-- the (source, asset class) pair rather than of one contract. A per-instrument
+-- override belongs on the xref row if a specific name ever needs one; nothing does
+-- today.
+--
+-- rank is ordinal, lower = preferred. Gaps are deliberate so a source can be
+-- slotted between two others without renumbering.
+--
+-- Why Binance outranks Bybit for SPOT: we route crypto orders to Binance, so its
+-- book is the one our fills print against. Bybit prices the same pair within a few
+-- bps (arbitrage), which makes it a good fallback but not the same book — using it
+-- while Binance is healthy would value positions against a venue we do not trade.
+
+CREATE TABLE provider_feed_policy (
+    source_code      TEXT NOT NULL,
+    instrument_class TEXT NOT NULL,
+    rank             INTEGER NOT NULL,
+    enabled          BOOLEAN NOT NULL DEFAULT true,
+    created_at       TIMESTAMPTZ NOT NULL DEFAULT now(),
+    updated_at       TIMESTAMPTZ NOT NULL DEFAULT now(),
+    PRIMARY KEY (source_code, instrument_class),
+    CONSTRAINT provider_feed_policy_rank_positive CHECK (rank > 0)
+);
+
+COMMENT ON TABLE provider_feed_policy IS
+    'Ranked market-data source preference per (source, instrument_class); lower rank wins.';
+COMMENT ON COLUMN provider_feed_policy.rank IS
+    'Ordinal preference, lower = preferred. A source is used only when every better-ranked source is stale or down.';
+
+INSERT INTO provider_feed_policy (source_code, instrument_class, rank) VALUES
+    ('DATABENTO', 'OPTION', 10),   -- only OPRA source; consolidated NBBO
+    ('BINANCE',   'SPOT',   10),   -- the venue we trade
+    ('BYBIT',     'SPOT',   20)    -- fallback: same pair, different book
+ON CONFLICT (source_code, instrument_class) DO NOTHING;

db/migrations/ods/public/0015_CRYPTO_SYMBOL_IS_THE_PAIR.sql

@@ -0,0 +1,29 @@
+-- A crypto instrument is a PAIR, so its symbol must name the pair.
+--
+-- seed_binance_crypto modelled crypto as symbol = base asset (SOL), currency =
+-- quote (USDT), venue = BINANCE. That cannot represent SOL/USDT and SOL/BTC at the
+-- same time: both key to (SOL, BINANCE) under instrument_symbol_venue_key, so
+-- adding any second quote currency for an existing base fails the unique
+-- constraint. The model silently allows exactly one quote currency per base per
+-- venue, which is not a property of the market.
+--
+-- Name the pair, as the venues do. Binance and Bybit both call it SOLUSDT, and our
+-- own xref already stores SOLUSDT as external_symbol on the BROKER and PROVIDER
+-- rows -- instrument.symbol was the only place in the system calling it SOL.
+-- Matches NautilusTrader's raw_symbol convention (SOLUSDT.BINANCE) and CCXT's
+-- rationale for putting base/quote in the symbol: contracts that differ must have
+-- distinct symbols.
+--
+-- currency stays USDT, so the pair remains decomposable and the quote asset is
+-- still first-class for settlement/valuation. The base asset is unchanged on the
+-- BROKER xref's external_native_id ('SOL'), which is what recon matches balances
+-- on -- so this does not disturb reconciliation.
+--
+-- Idempotent: the NOT LIKE guard skips rows already carrying the quote suffix.
+
+UPDATE instrument
+SET symbol = symbol || currency,
+    updated_at = now()
+WHERE asset_class = 'CRYPTO'
+  AND instrument_class = 'SPOT'
+  AND symbol NOT LIKE '%' || currency;

db/scripts/seed_binance_crypto.sql

@@ -2,10 +2,13 @@
 -- quote currency, a few spot pairs as instruments, their BROKER xref rows (order
 -- routing + recon), and the broker_connection. Idempotent.
 --
--- Crypto instrument model: symbol = base asset (BTC), currency = quote (USDT),
--- venue = BINANCE. The xref carries the Binance pair as external_symbol (BTCUSDT,
--- used to route orders) and the base asset as external_native_id (BTC, what
--- account balances report and reconciliation matches on).
+-- Crypto instrument model: symbol = the PAIR (BTCUSDT), currency = quote (USDT),
+-- venue = BINANCE. The symbol names the pair because that is what the instrument
+-- is -- symbol = base alone cannot represent BTC/USDT and BTC/BTC-quote variants
+-- at once under UNIQUE (symbol, venue), and every venue calls it BTCUSDT anyway.
+-- The xref carries that pair as external_symbol (order routing + market data) and
+-- the base asset as external_native_id (BTC, what account balances report and
+-- reconciliation matches on).
 --
 -- Account creation (principal/portfolio/account routing to binance-paper) is left
 -- to the admin/cockpit — this seeds only the shared reference data.
@@ -24,9 +27,9 @@ INSERT INTO instrument
     (symbol, venue, name, asset_class, instrument_class, currency, status,
      price_precision, size_precision, price_increment, size_increment)
 VALUES
-    ('BTC', 'BINANCE', 'Bitcoin',  'CRYPTO', 'SPOT', 'USDT', 'ACTIVE', 2, 5, 0.01,  0.00001),
-    ('ETH', 'BINANCE', 'Ethereum', 'CRYPTO', 'SPOT', 'USDT', 'ACTIVE', 2, 4, 0.01,  0.0001),
-    ('SOL', 'BINANCE', 'Solana',   'CRYPTO', 'SPOT', 'USDT', 'ACTIVE', 2, 3, 0.01,  0.001)
+    ('BTCUSDT', 'BINANCE', 'Bitcoin',  'CRYPTO', 'SPOT', 'USDT', 'ACTIVE', 2, 5, 0.01,  0.00001),
+    ('ETHUSDT', 'BINANCE', 'Ethereum', 'CRYPTO', 'SPOT', 'USDT', 'ACTIVE', 2, 4, 0.01,  0.0001),
+    ('SOLUSDT', 'BINANCE', 'Solana',   'CRYPTO', 'SPOT', 'USDT', 'ACTIVE', 2, 3, 0.01,  0.001)
 ON CONFLICT (symbol, venue) DO NOTHING;
 
 -- BROKER xref: external_symbol = Binance pair (order routing);
@@ -40,7 +43,41 @@ FROM (VALUES
     ('ETH', 'ETHUSDT', 0.0001),
     ('SOL', 'SOLUSDT', 0.001)
 ) AS x(base, pair, min_qty)
-JOIN instrument i ON i.symbol = x.base AND i.venue = 'BINANCE'
+JOIN instrument i ON i.symbol = x.pair AND i.venue = 'BINANCE'
+ON CONFLICT DO NOTHING;
+
+-- PROVIDER xref: the same pair, but as a market-data identity rather than a
+-- routing one. Without these rows the crypto is tradable but unmarkable — the
+-- quote feed has no way to know Binance can price it, so /positions reports a
+-- null mark for a position we can freely trade.
+--
+-- external_symbol is the pair as Binance's market-data streams report it (the `s`
+-- field of bookTicker, uppercase); external_exchange is the venue, matching
+-- instrument.venue. No native_id: the feed resolves by symbol, and Binance's
+-- market data has no separate id worth storing.
+INSERT INTO oms.instrument_xref
+    (instrument_id, source_type, source_code, external_symbol, external_exchange,
+     method, confidence)
+SELECT i.id, 'PROVIDER', 'BINANCE', x.pair, 'BINANCE', 'manual', 'resolved'
+FROM (VALUES
+    ('BTC', 'BTCUSDT'),
+    ('ETH', 'ETHUSDT'),
+    ('SOL', 'SOLUSDT')
+) AS x(base, pair)
+JOIN instrument i ON i.symbol = x.pair AND i.venue = 'BINANCE'
+ON CONFLICT DO NOTHING;
+
+-- A SECOND market-data source for the same pairs. Bybit quotes the identical
+-- symbol; ranked below Binance in provider_feed_policy, it is the failover feed
+-- when Binance goes quiet. external_symbol is the same pair string (Bybit's v5
+-- orderbook.1 reports it identically); external_exchange = BYBIT so the two
+-- PROVIDER rows are distinct identities for one instrument.
+INSERT INTO oms.instrument_xref
+    (instrument_id, source_type, source_code, external_symbol, external_exchange,
+     method, confidence)
+SELECT i.id, 'PROVIDER', 'BYBIT', i.symbol, 'BYBIT', 'manual', 'resolved'
+FROM instrument i
+WHERE i.asset_class = 'CRYPTO' AND i.instrument_class = 'SPOT'
 ON CONFLICT DO NOTHING;
 
 -- Routing target. Credentials resolved from env (BINANCE_PAPER_API_KEY/SECRET).

src/alpaca_stream.rs

@@ -1,6 +1,6 @@
 use futures_util::{SinkExt, StreamExt};
 use sqlx::{PgPool, Row};
-use tokio::time::{sleep, Duration};
+use tokio::time::Duration;
 use tokio_tungstenite::{connect_async, tungstenite::Message};
 use tracing::{error, info, warn};
 use uuid::Uuid;
@@ -13,10 +13,39 @@ use crate::domain::orders::commands::ExecutionReport;
 use crate::execution::process_execution_report;
 use crate::kafka::KafkaClient;
 use crate::stream_health::StreamHandle;
+use crate::stream_supervisor::{supervise, Session, StreamResult};
 
 /// Ping + missed-fill sweep cadence; see the Binance stream for the rationale.
 const HEARTBEAT_SECS: u64 = 30;
 
+struct AlpacaSession {
+    ws_url: &'static str,
+    api_key: String,
+    api_secret: String,
+    pool: PgPool,
+    kafka: Option<KafkaClient>,
+    adapter: Arc<AlpacaAdapter>,
+    health: StreamHandle,
+    position_changed_tx: Option<mpsc::Sender<()>>,
+}
+
+#[async_trait::async_trait]
+impl Session for AlpacaSession {
+    async fn run_once(&mut self) -> StreamResult {
+        connect_and_run(
+            self.ws_url,
+            &self.api_key,
+            &self.api_secret,
+            &self.pool,
+            &self.kafka,
+            &self.adapter,
+            &self.health,
+            self.position_changed_tx.as_ref(),
+        )
+        .await
+    }
+}
+
 pub async fn run(
     environment: &'static str,
     api_key: String,
@@ -36,24 +65,18 @@ pub async fn run(
         "wss://paper-api.alpaca.markets/stream"
     };
 
-    let mut backoff_secs: u64 = 1;
-
-    loop {
-        health.set_connecting();
-        info!(env = environment, attempt = backoff_secs, "Alpaca stream connecting");
-        match connect_and_run(ws_url, &api_key, &api_secret, &pool, &kafka, &adapter, &health, position_changed_tx.as_ref()).await {
-            Ok(()) => {
-                warn!(env = environment, "Alpaca stream closed cleanly, reconnecting in {backoff_secs}s");
-                health.set_down("stream closed");
-            }
-            Err(e) => {
-                warn!(env = environment, error = %e, "Alpaca stream error, reconnecting in {backoff_secs}s");
-                health.set_down(e.to_string());
-            }
-        }
-        sleep(Duration::from_secs(backoff_secs)).await;
-        backoff_secs = (backoff_secs * 2).min(30);
-    }
+    let name: &'static str = if environment == "LIVE" { "ALPACA/LIVE" } else { "ALPACA/PAPER" };
+    let session = AlpacaSession {
+        ws_url,
+        api_key,
+        api_secret,
+        pool,
+        kafka,
+        adapter,
+        health: health.clone(),
+        position_changed_tx,
+    };
+    supervise(name, health, session).await
 }
 
 async fn reconcile_routed_orders(