diff --git a/crates/dataprovider/src/lib.rs b/crates/dataprovider/src/lib.rs index f6ee522..813102e 100644 --- a/crates/dataprovider/src/lib.rs +++ b/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>`. 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; diff --git a/crates/dataprovider/src/quote.rs b/crates/dataprovider/src/quote.rs new file mode 100644 index 0000000..0e89b96 --- /dev/null +++ b/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, + /// 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; + +/// 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, + out: &mpsc::Sender, + add_rx: &mut mpsc::Receiver, + health: &dyn FeedHealth, + ) -> Result<(), ProviderError>; +} diff --git a/db/migrations/ods/oms/0018_CREATE_PROVIDER_FEED_POLICY.sql b/db/migrations/ods/oms/0018_CREATE_PROVIDER_FEED_POLICY.sql new file mode 100644 index 0000000..0c69b9e --- /dev/null +++ b/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; diff --git a/db/migrations/ods/public/0015_CRYPTO_SYMBOL_IS_THE_PAIR.sql b/db/migrations/ods/public/0015_CRYPTO_SYMBOL_IS_THE_PAIR.sql new file mode 100644 index 0000000..3c44bb7 --- /dev/null +++ b/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; diff --git a/db/scripts/seed_binance_crypto.sql b/db/scripts/seed_binance_crypto.sql index e007cea..6fb7d47 100644 --- a/db/scripts/seed_binance_crypto.sql +++ b/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). diff --git a/src/alpaca_stream.rs b/src/alpaca_stream.rs index dcac437..b1adc23 100644 --- a/src/alpaca_stream.rs +++ b/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, + adapter: Arc, + health: StreamHandle, + position_changed_tx: Option>, +} + +#[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( diff --git a/src/binance_feed.rs b/src/binance_feed.rs new file mode 100644 index 0000000..dd59773 --- /dev/null +++ b/src/binance_feed.rs @@ -0,0 +1,196 @@ +//! Binance spot top-of-book, as a [`LiveQuoteFeed`]. +//! +//! Subscribes to `@bookTicker` — best bid/ask pushed on every change. This +//! is a public stream: no key, no signing, nothing in common with the authenticated +//! WS-API the order stream uses (`binance_stream.rs`). +//! +//! **Production, deliberately, even though orders route to testnet.** Testnet runs +//! its own synthetic book, so marking against it would value positions at prices no +//! one can trade. Real prices make the mark meaningful; P&L against a testnet fill +//! is meaningless either way, so the choice is between a useful number and a +//! useless one. + +use std::collections::HashMap; + +use chrono::Utc; +use dataprovider::{DataProvider, FeedHealth, LiveQuoteFeed, ProviderError, Quote, SymbolAdds}; +use futures_util::{SinkExt, StreamExt}; +use tokio::sync::mpsc; +use tokio_tungstenite::{connect_async, tungstenite::Message}; +use tracing::{info, warn}; + +/// Public market-data endpoint. Raw (`/ws`) rather than combined (`/stream`), so +/// symbols can be added mid-session with a SUBSCRIBE frame. +const WS_URL: &str = "wss://stream.binance.com:9443/ws"; +const SOURCE_CODE: &str = "BINANCE"; + +/// Override the endpoint — for pointing at a mock, or at testnet's book if you ever +/// want marks consistent with testnet fills rather than with the real market. +/// Setting it to an unreachable host is also how the failover path gets exercised +/// without unplugging anything. +fn ws_url() -> String { + std::env::var("BINANCE_FEED_WS_URL").unwrap_or_else(|_| WS_URL.to_string()) +} + +pub struct BinanceFeed; + +impl DataProvider for BinanceFeed { + fn code(&self) -> &'static str { + SOURCE_CODE + } +} + +#[async_trait::async_trait] +impl LiveQuoteFeed for BinanceFeed { + fn covers(&self) -> &'static [&'static str] { + &["SPOT"] + } + + async fn run_session( + &self, + symbols: HashMap, + out: &mpsc::Sender, + add_rx: &mut mpsc::Receiver, + health: &dyn FeedHealth, + ) -> Result<(), ProviderError> { + // Keys are the pair as Binance reports it in `s` (uppercase, e.g. SOLUSDT). + let mut sym_to_id = symbols; + + let url = ws_url(); + let (mut ws, _) = connect_async(&url) + .await + .map_err(|e| ProviderError::Request(e.to_string()))?; + info!(url = %url, "Binance feed: connected"); + + let mut req_id: u64 = 1; + subscribe(&mut ws, sym_to_id.keys().cloned().collect(), &mut req_id).await?; + health.on_connected(); + info!(count = sym_to_id.len(), "Binance feed: live"); + + loop { + let text = tokio::select! { + adds = add_rx.recv() => { + match adds { + Some(adds) => { + let pairs: Vec = adds.keys().cloned().collect(); + info!(count = pairs.len(), "Binance feed: subscribing new symbols"); + sym_to_id.extend(adds); + subscribe(&mut ws, pairs, &mut req_id).await?; + } + // Driver gone; nothing more will be added, but keep streaming. + None => std::future::pending::<()>().await, + } + continue; + } + msg = ws.next() => { + let Some(msg) = msg else { break }; + health.on_event(); + match msg.map_err(|e| ProviderError::Request(e.to_string()))? { + Message::Text(t) => t, + // Binance pings every few minutes and closes the socket if we + // do not pong. tungstenite does not answer for us. + Message::Ping(p) => { + ws.send(Message::Pong(p)) + .await + .map_err(|e| ProviderError::Request(e.to_string()))?; + continue; + } + Message::Close(_) => break, + _ => continue, + } + } + }; + + let v: serde_json::Value = match serde_json::from_str(&text) { + Ok(v) => v, + Err(e) => { + warn!(error = %e, raw = %text, "Binance feed: parse failed"); + continue; + } + }; + + // SUBSCRIBE acks look like {"result":null,"id":1} — no `s` field. + let Some(pair) = v["s"].as_str() else { continue }; + let Some(&instrument_id) = sym_to_id.get(pair) else { continue }; + + // bookTicker sends prices as strings; a malformed one is not a reason to + // drop the session, but it must not become a 0.0 mark either. + let (Some(bid), Some(ask)) = (num(&v["b"]), num(&v["a"])) else { + warn!(pair, raw = %text, "Binance feed: unparseable price, skipping"); + continue; + }; + + let quote = Quote { + instrument_id, + bid, + ask, + bid_size: num(&v["B"]).unwrap_or(0.0) as u32, + ask_size: num(&v["A"]).unwrap_or(0.0) as u32, + ts_recv: Utc::now(), + source_code: SOURCE_CODE, + }; + if out.send(quote).await.is_err() { + warn!("Binance feed: mark router gone, ending session"); + return Ok(()); + } + } + + Ok(()) + } +} + +/// Binance wants lowercase stream names (`solusdt@bookTicker`) but reports the pair +/// uppercase in `s`. Our xref stores the uppercase form, so only the request side +/// is lowered. +async fn subscribe(ws: &mut S, pairs: Vec, req_id: &mut u64) -> Result<(), ProviderError> +where + S: SinkExt + Unpin, + >::Error: std::fmt::Display, +{ + if pairs.is_empty() { + return Ok(()); + } + let params: Vec = pairs + .iter() + .map(|p| format!("{}@bookTicker", p.to_lowercase())) + .collect(); + let msg = serde_json::json!({ "method": "SUBSCRIBE", "params": params, "id": req_id }); + *req_id += 1; + ws.send(Message::Text(msg.to_string())) + .await + .map_err(|e| ProviderError::Request(e.to_string())) +} + +/// Binance encodes every number as a JSON string. +fn num(v: &serde_json::Value) -> Option { + v.as_str()?.parse().ok() +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn parses_binance_string_numbers() { + let v: serde_json::Value = serde_json::from_str(r#"{"b":"25.35190000"}"#).unwrap(); + assert_eq!(num(&v["b"]), Some(25.3519)); + } + + #[test] + fn missing_or_malformed_price_is_none() { + let v: serde_json::Value = serde_json::from_str(r#"{"b":"","a":null}"#).unwrap(); + assert_eq!(num(&v["b"]), None); + assert_eq!(num(&v["a"]), None); + assert_eq!(num(&v["nope"]), None); + } + + /// A real bookTicker frame must yield both sides. + #[test] + fn decodes_a_book_ticker_frame() { + let raw = r#"{"u":400900217,"s":"SOLUSDT","b":"25.35190000","B":"31.21000000","a":"25.36520000","A":"40.66000000"}"#; + let v: serde_json::Value = serde_json::from_str(raw).unwrap(); + assert_eq!(v["s"].as_str(), Some("SOLUSDT")); + assert_eq!(num(&v["b"]), Some(25.3519)); + assert_eq!(num(&v["a"]), Some(25.3652)); + } +} diff --git a/src/binance_stream.rs b/src/binance_stream.rs index e02f855..dcf1899 100644 --- a/src/binance_stream.rs +++ b/src/binance_stream.rs @@ -14,7 +14,7 @@ use chrono::Utc; use futures_util::{SinkExt, StreamExt}; use sqlx::{PgPool, Row}; use tokio::sync::mpsc; -use tokio::time::{sleep, Duration}; +use tokio::time::Duration; use tokio_tungstenite::{connect_async, tungstenite::Message}; use tracing::{error, info, warn}; use uuid::Uuid; @@ -24,6 +24,7 @@ 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}; const VENUE: &str = "BINANCE"; @@ -32,10 +33,33 @@ const VENUE: &str = "BINANCE"; /// invisible; the ping surfaces a dead socket and the sweep recovers the fill. const HEARTBEAT_SECS: u64 = 30; +struct BinanceSession { + pool: PgPool, + kafka: Option, + adapter: Arc, + health: StreamHandle, + position_changed_tx: Option>, +} + +#[async_trait::async_trait] +impl Session for BinanceSession { + async fn run_once(&mut self) -> StreamResult { + connect_and_run( + &self.adapter, + &self.pool, + &self.kafka, + &self.health, + self.position_changed_tx.as_ref(), + ) + .await + } +} + +/// Credentials are not taken here: the adapter already holds them (it signs with +/// an Ed25519 key loaded at construction), which is why the old `_api_key` / +/// `_api_secret` parameters were unused. pub async fn run( environment: &'static str, - _api_key: String, - _api_secret: String, pool: PgPool, kafka: Option, adapter: Arc, @@ -46,23 +70,15 @@ pub async fn run( // Catch up on anything missed while disconnected, then stream live. reconcile_routed_orders(&pool, &kafka, &adapter, position_changed_tx.as_ref()).await; - let mut backoff_secs: u64 = 1; - loop { - health.set_connecting(); - info!(env = environment, "Binance WS-API connecting"); - match connect_and_run(&adapter, &pool, &kafka, &health, position_changed_tx.as_ref()).await { - Ok(()) => { - warn!(env = environment, "Binance stream closed, reconnecting in {backoff_secs}s"); - health.set_down("stream closed"); - } - Err(e) => { - warn!(env = environment, error = %e, "Binance 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 session = BinanceSession { + pool, + kafka, + adapter, + health: health.clone(), + position_changed_tx, + }; + let name: &'static str = if environment == "LIVE" { "BINANCE/LIVE" } else { "BINANCE/PAPER" }; + supervise(name, health, session).await } async fn connect_and_run( diff --git a/src/bybit_feed.rs b/src/bybit_feed.rs new file mode 100644 index 0000000..4015648 --- /dev/null +++ b/src/bybit_feed.rs @@ -0,0 +1,240 @@ +//! Bybit spot top-of-book, as a [`LiveQuoteFeed`]. +//! +//! Subscribes to `orderbook.1.` on the v5 public spot stream — no key, no +//! signing. Like the Binance feed this is production data used to mark positions +//! held on a paper venue: a real price is the only useful one. +//! +//! Bybit exists here as a *second* source for the same crypto instruments, so a +//! Binance outage does not leave positions unmarked. It is ranked below Binance +//! (see `provider_feed_policy`) because we trade on Binance — its book is the one +//! our fills actually print against. Bybit's price for the same pair is a close +//! proxy (arbitrage holds them within a few bps), not the same book. + +use std::collections::HashMap; + +use chrono::Utc; +use dataprovider::{DataProvider, FeedHealth, LiveQuoteFeed, ProviderError, Quote, SymbolAdds}; +use futures_util::{SinkExt, StreamExt}; +use tokio::sync::mpsc; +use tokio::time::{interval, Duration}; +use tokio_tungstenite::{connect_async, tungstenite::Message}; +use tracing::{info, warn}; + +const WS_URL: &str = "wss://stream.bybit.com/v5/public/spot"; +const SOURCE_CODE: &str = "BYBIT"; + +/// Bybit closes a connection that has been silent for 30s; its docs prescribe an +/// application-level `{"op":"ping"}` every 20s. This is *not* a WebSocket ping +/// frame — the protocol-level one does not reset their timer. +const PING_SECS: u64 = 20; + +/// Best bid/ask carried across messages. +/// +/// `orderbook.1` is a stateful topic: a `snapshot` replaces the book, a `delta` +/// carries only the side that changed. Emitting on a delta without remembering the +/// other side would publish a half-quote, so the last known level is kept per pair +/// and a quote is only emitted once both sides are known. +#[derive(Default, Clone, Copy)] +struct Top { + bid: Option<(f64, f64)>, + ask: Option<(f64, f64)>, +} + +pub struct BybitFeed; + +impl DataProvider for BybitFeed { + fn code(&self) -> &'static str { + SOURCE_CODE + } +} + +#[async_trait::async_trait] +impl LiveQuoteFeed for BybitFeed { + fn covers(&self) -> &'static [&'static str] { + &["SPOT"] + } + + async fn run_session( + &self, + symbols: HashMap, + out: &mpsc::Sender, + add_rx: &mut mpsc::Receiver, + health: &dyn FeedHealth, + ) -> Result<(), ProviderError> { + let mut sym_to_id = symbols; + let mut books: HashMap = HashMap::new(); + + let (mut ws, _) = connect_async(WS_URL) + .await + .map_err(|e| ProviderError::Request(e.to_string()))?; + info!(url = WS_URL, "Bybit feed: connected"); + + subscribe(&mut ws, sym_to_id.keys().cloned().collect()).await?; + health.on_connected(); + info!(count = sym_to_id.len(), "Bybit feed: live"); + + let mut ping = interval(Duration::from_secs(PING_SECS)); + ping.tick().await; // consume the immediate first tick + + loop { + let text = tokio::select! { + _ = ping.tick() => { + ws.send(Message::Text(r#"{"op":"ping"}"#.to_string())) + .await + .map_err(|e| ProviderError::Request(e.to_string()))?; + continue; + } + adds = add_rx.recv() => { + match adds { + Some(adds) => { + let pairs: Vec = adds.keys().cloned().collect(); + info!(count = pairs.len(), "Bybit feed: subscribing new symbols"); + sym_to_id.extend(adds); + subscribe(&mut ws, pairs).await?; + } + None => std::future::pending::<()>().await, + } + continue; + } + msg = ws.next() => { + let Some(msg) = msg else { break }; + health.on_event(); + match msg.map_err(|e| ProviderError::Request(e.to_string()))? { + Message::Text(t) => t, + Message::Ping(p) => { + ws.send(Message::Pong(p)) + .await + .map_err(|e| ProviderError::Request(e.to_string()))?; + continue; + } + Message::Close(_) => break, + _ => continue, + } + } + }; + + let v: serde_json::Value = match serde_json::from_str(&text) { + Ok(v) => v, + Err(e) => { + warn!(error = %e, raw = %text, "Bybit feed: parse failed"); + continue; + } + }; + + // Subscribe acks and pongs carry `op`, not `topic`. + if v.get("topic").is_none() { + if v["success"] == serde_json::json!(false) { + warn!(raw = %text, "Bybit feed: request rejected"); + } + continue; + } + + let data = &v["data"]; + let Some(pair) = data["s"].as_str() else { continue }; + let Some(&instrument_id) = sym_to_id.get(pair) else { continue }; + + let is_snapshot = v["type"].as_str() == Some("snapshot"); + let book = books.entry(pair.to_string()).or_default(); + if is_snapshot { + // A snapshot is the whole book: a side absent from it is genuinely + // gone, so do not carry the previous level forward. + *book = Top::default(); + } + if let Some(bid) = level(&data["b"]) { + book.bid = bid; + } + if let Some(ask) = level(&data["a"]) { + book.ask = ask; + } + + let (Some((bid, bid_sz)), Some((ask, ask_sz))) = (book.bid, book.ask) else { + continue; // only one side known so far + }; + + let quote = Quote { + instrument_id, + bid, + ask, + bid_size: bid_sz as u32, + ask_size: ask_sz as u32, + ts_recv: Utc::now(), + source_code: SOURCE_CODE, + }; + if out.send(quote).await.is_err() { + warn!("Bybit feed: mark router gone, ending session"); + return Ok(()); + } + } + + Ok(()) + } +} + +async fn subscribe(ws: &mut S, pairs: Vec) -> Result<(), ProviderError> +where + S: SinkExt + Unpin, + >::Error: std::fmt::Display, +{ + if pairs.is_empty() { + return Ok(()); + } + let args: Vec = pairs.iter().map(|p| format!("orderbook.1.{p}")).collect(); + let msg = serde_json::json!({ "op": "subscribe", "args": args }); + ws.send(Message::Text(msg.to_string())) + .await + .map_err(|e| ProviderError::Request(e.to_string())) +} + +/// Read one side of an `orderbook.1` update. +/// +/// Returns `None` when the side is absent (a delta that didn't touch it — keep the +/// previous level) and `Some(None)` when it was explicitly removed (size "0"), +/// which must clear the level rather than leave a stale price standing. +#[allow(clippy::option_option)] +fn level(side: &serde_json::Value) -> Option> { + let arr = side.as_array()?; + let first = arr.first()?; + let price: f64 = first.get(0)?.as_str()?.parse().ok()?; + let size: f64 = first.get(1)?.as_str()?.parse().ok()?; + Some(if size == 0.0 { None } else { Some((price, size)) }) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn json(s: &str) -> serde_json::Value { + serde_json::from_str(s).unwrap() + } + + #[test] + fn reads_a_level() { + let v = json(r#"{"b":[["74.72","138.8075"]]}"#); + assert_eq!(level(&v["b"]), Some(Some((74.72, 138.8075)))); + } + + #[test] + fn absent_side_leaves_previous_level_alone() { + // A delta that didn't touch this side: [] means "no change", not "no bid". + let v = json(r#"{"b":[]}"#); + assert_eq!(level(&v["b"]), None); + } + + #[test] + fn zero_size_clears_the_level() { + // Bybit deletes a level by sending size "0" — carrying the old price + // forward here would leave a stale bid standing after it was pulled. + let v = json(r#"{"b":[["74.72","0"]]}"#); + assert_eq!(level(&v["b"]), Some(None)); + } + + #[test] + fn decodes_a_real_snapshot_frame() { + let raw = r#"{"topic":"orderbook.1.SOLUSDT","ts":1784287273269,"type":"snapshot","data":{"s":"SOLUSDT","b":[["74.72","138.8075"]],"a":[["74.73","16.3301"]],"u":8666785,"seq":155684298279}}"#; + let v = json(raw); + assert_eq!(v["data"]["s"].as_str(), Some("SOLUSDT")); + assert_eq!(v["type"].as_str(), Some("snapshot")); + assert_eq!(level(&v["data"]["b"]), Some(Some((74.72, 138.8075)))); + assert_eq!(level(&v["data"]["a"]), Some(Some((74.73, 16.3301)))); + } +} diff --git a/src/handlers.rs b/src/handlers.rs index 893c173..e46cc4d 100644 --- a/src/handlers.rs +++ b/src/handlers.rs @@ -23,7 +23,7 @@ use crate::domain::orders::commands::{OrderCommand, RouteOrder, SubmitOrder}; use crate::domain::orders::errors::{CommandRejection, RejectionCode}; use crate::domain::orders::events::OrderDomainEvent; use crate::auth::AuthContext; -use crate::positions::Position; +use crate::positions::{value as value_position, Position}; use crate::domain::orders::state::{OrderAggregateState, OrderSide, OrderStatus, OrderType, TimeInForce}; use crate::event_store::{OrderEventStore, NewOrderEvent}; use crate::kafka::publish_events; @@ -985,13 +985,19 @@ pub async fn get_portfolio_positions( }); } + // contract_size scales the mark into money (x100 for options, 1 for spot). + // LEFT JOIN: a position whose master row is missing still returns, valued at 1x. + // position.instrument_id is TEXT holding the numeric instrument.id. let rows = sqlx::query( - "SELECT portfolio_id, instrument_id, \ - net_qty::double precision AS net_qty, \ - avg_cost::double precision AS avg_cost, \ - realized_pnl::double precision AS realized_pnl, \ - updated_at \ - FROM position WHERE portfolio_id = $1 ORDER BY instrument_id", + "SELECT p.portfolio_id, p.instrument_id, \ + p.net_qty::double precision AS net_qty, \ + p.avg_cost::double precision AS avg_cost, \ + p.realized_pnl::double precision AS realized_pnl, \ + p.updated_at, \ + COALESCE(i.contract_size, 1)::double precision AS contract_size \ + FROM position p \ + LEFT JOIN instrument i ON i.id::text = p.instrument_id \ + WHERE p.portfolio_id = $1 ORDER BY p.instrument_id", ) .bind(portfolio_id) .fetch_all(state.pool()) @@ -1001,15 +1007,36 @@ pub async fn get_portfolio_positions( message: format!("failed to load positions: {:?}", err), })?; + // Look up per row rather than snapshotting the whole store: a portfolio holds a + // handful of instruments, while the store carries every marked instrument in the + // universe, so a clone would cost O(all marks) per request and hold the read + // lock against the router's writes for the duration. + let marks = state.marks(); + let positions = rows .iter() - .map(|r| Position { - portfolio_id: r.get("portfolio_id"), - instrument_id: r.get("instrument_id"), - net_qty: r.get("net_qty"), - avg_cost: r.get("avg_cost"), - realized_pnl: r.get("realized_pnl"), - updated_at: r.get("updated_at"), + .map(|r| { + let instrument_id: String = r.get("instrument_id"); + let net_qty: f64 = r.get("net_qty"); + let avg_cost: f64 = r.get("avg_cost"); + let contract_size: f64 = r.get("contract_size"); + + let mark = instrument_id.parse::().ok().and_then(|id| marks.get(id)); + let mid = mark.map(|m| m.mid()); + let valuation = mid.map(|mid| value_position(net_qty, avg_cost, contract_size, mid)); + + Position { + portfolio_id: r.get("portfolio_id"), + instrument_id, + net_qty, + avg_cost, + realized_pnl: r.get("realized_pnl"), + updated_at: r.get("updated_at"), + mark: mid, + market_value: valuation.map(|v| v.market_value), + unrealized_pnl: valuation.map(|v| v.unrealized_pnl), + mark_ts: mark.map(|m| m.ts), + } }) .collect(); Ok(Json(positions)) diff --git a/src/main.rs b/src/main.rs index 7d2bb8d..05b6e4f 100644 --- a/src/main.rs +++ b/src/main.rs @@ -53,6 +53,11 @@ mod binance_stream; mod stream_health; mod marks; mod opra_stream; +mod stream_supervisor; +mod quote_feed; +mod mark_router; +mod binance_feed; +mod bybit_feed; #[derive(OpenApi)] #[openapi( @@ -360,10 +365,16 @@ async fn serve() { Err(e) => error!(error = ?e, "failed to check position projection"), } - // Doorbell from the fill path to the marks feed: a fill moved a position, so + // Doorbell from the fill path to the marks feeds: a fill moved a position, so // re-read the held set. bounded(1) — a queued signal already means "reload", // so extras are redundant and try_send never blocks the fill path. - let (position_changed_tx, position_changed_rx) = tokio::sync::mpsc::channel::<()>(1); + // + // The fill path rings once; an mpsc has exactly one consumer, so a fan-out task + // (spawned below, once every feed has registered) relays to each feed's own + // doorbell. Keeping the fill path on a single sender means execution.rs need + // not know how many feeds exist. + let (position_changed_tx, mut position_changed_rx) = tokio::sync::mpsc::channel::<()>(1); + let mut marks_doorbells: Vec> = Vec::new(); // Spawn Alpaca trade-update stream tasks (one per configured environment) if let (Ok(key), Ok(secret)) = (env::var("ALPACA_PAPER_API_KEY"), env::var("ALPACA_PAPER_API_SECRET")) { @@ -386,15 +397,73 @@ async fn serve() { // Spawn the Binance user-data stream when configured. if let Some(adapter) = binance_paper { let health = state.stream_health().handle("BINANCE", "PAPER"); - tokio::spawn(binance_stream::run("PAPER", String::new(), String::new(), state.pool().clone(), state.kafka().cloned(), adapter, health, Some(position_changed_tx.clone()))); + tokio::spawn(binance_stream::run("PAPER", state.pool().clone(), state.kafka().cloned(), adapter, health, Some(position_changed_tx.clone()))); } - // Spawn the Databento OPRA marks stream when configured. + // Market data: feeds emit quotes onto one channel; the router is the sole + // writer to MarkStore. Adding a vendor means spawning another feed here — + // nothing downstream changes. + let (quote_tx, quote_rx) = tokio::sync::mpsc::channel::(1024); + tokio::spawn(mark_router::run(quote_rx, state.marks().clone(), state.pool().clone())); + if env::var("DATABENTO_API_KEY").map(|k| !k.is_empty()).unwrap_or(false) { + let (opra_pos_tx, opra_pos_rx) = tokio::sync::mpsc::channel::<()>(1); + marks_doorbells.push(opra_pos_tx); let health = state.stream_health().handle("DATABENTO", "OPRA"); - tokio::spawn(opra_stream::run(state.pool().clone(), state.marks().clone(), health, position_changed_rx)); + let session = quote_feed::QuoteFeedSession::new( + opra_stream::DatabentoOpraFeed, + state.pool().clone(), + quote_tx.clone(), + opra_pos_rx, + health.clone(), + ); + tokio::spawn(stream_supervisor::supervise("DATABENTO/OPRA", health, session)); } + // Binance public market data — no credentials, so it is always on. Each feed + // needs its own doorbell receiver (an mpsc has exactly one consumer), so the + // sender is cloned per feed rather than shared. + { + let (binance_pos_tx, binance_pos_rx) = tokio::sync::mpsc::channel::<()>(1); + marks_doorbells.push(binance_pos_tx); + let health = state.stream_health().handle("BINANCE", "SPOT"); + let session = quote_feed::QuoteFeedSession::new( + binance_feed::BinanceFeed, + state.pool().clone(), + quote_tx.clone(), + binance_pos_rx, + health.clone(), + ); + tokio::spawn(stream_supervisor::supervise("BINANCE/SPOT", health, session)); + } + + // Bybit public market data — a second source for the same crypto pairs, so a + // Binance outage does not leave positions unmarked. Ranked below Binance in + // provider_feed_policy; the router decides which one owns the mark. + { + let (bybit_pos_tx, bybit_pos_rx) = tokio::sync::mpsc::channel::<()>(1); + marks_doorbells.push(bybit_pos_tx); + let health = state.stream_health().handle("BYBIT", "SPOT"); + let session = quote_feed::QuoteFeedSession::new( + bybit_feed::BybitFeed, + state.pool().clone(), + quote_tx.clone(), + bybit_pos_rx, + health.clone(), + ); + tokio::spawn(stream_supervisor::supervise("BYBIT/SPOT", health, session)); + } + + // Relay the fill path's single doorbell to every feed. try_send: a full + // per-feed channel already means "reload pending", and this must never block. + tokio::spawn(async move { + while position_changed_rx.recv().await.is_some() { + for doorbell in &marks_doorbells { + let _ = doorbell.try_send(()); + } + } + }); + // Register routes // 1) Register order routes diff --git a/src/mark_router.rs b/src/mark_router.rs new file mode 100644 index 0000000..ce73812 --- /dev/null +++ b/src/mark_router.rs @@ -0,0 +1,314 @@ +//! Routes normalized quotes from every feed into the [`MarkStore`], choosing which +//! source wins when more than one can price an instrument. +//! +//! This is the only writer, so "which source is authoritative" is decided in one +//! place rather than by whichever feed ticked last. The policy is **ranked primary +//! with staleness-driven failover**: the best-ranked source that is still producing +//! quotes owns the mark; when it goes quiet, the next one takes over and the switch +//! is logged. +//! +//! Staleness — rather than a health/disconnect signal — is the trigger because it +//! catches the failure that a connection state cannot: a socket that is up and +//! subscribed but silently no longer delivering. A feed that stops ticking is +//! useless whether or not its TCP connection admits it. +//! +//! This is not A/B line arbitration. That is per-message sequence-number +//! reconciliation between two redundant lines of the *same* feed, at microsecond +//! granularity. This is cross-vendor, at seconds, between genuinely different books +//! — Binance and Bybit quote the same pair but are not the same market. + +use std::collections::HashMap; + +use chrono::{DateTime, Utc}; +use dataprovider::Quote; +use sqlx::{PgPool, Row}; +use tokio::sync::mpsc; +use tracing::{debug, info, warn}; + +use crate::marks::MarkStore; + +/// How long a source may go quiet before a worse-ranked source may take over. +/// +/// Must exceed the natural gap between quotes on the quietest instrument a feed +/// covers, or the mark will flap between sources on an idle book. Safe today: the +/// only multi-source class is crypto SPOT, which ticks continuously; options have a +/// single source, so nothing can preempt them however long they sit quiet. +const STALE_AFTER_SECS: i64 = 10; + +/// Rank applied to a source with no policy row. Higher than any configured rank, so +/// an unconfigured source is usable but never preferred over a configured one. +const UNRANKED: i32 = 1000; + +/// Who currently owns an instrument's mark. +struct Active { + source: &'static str, + rank: i32, + at: DateTime, +} + +/// Drain quotes into the store until every feed has dropped its sender. +pub async fn run(mut quotes: mpsc::Receiver, marks: MarkStore, pool: PgPool) { + let policy = match load_policy(&pool).await { + Ok(p) => { + info!(rules = p.len(), "mark router: loaded source policy"); + p + } + Err(e) => { + // Not fatal: every source falls back to UNRANKED, so marks still flow + // first-come. Degrading to "some mark" beats no marks at all. + warn!(error = %e, "mark router: policy load failed, all sources unranked"); + HashMap::new() + } + }; + + let mut classes: HashMap = HashMap::new(); + let mut active: HashMap = HashMap::new(); + let mut count: u64 = 0; + + info!("mark router: started"); + + while let Some(q) = quotes.recv().await { + if !is_usable(q.bid, q.ask) { + debug!( + instrument_id = q.instrument_id, + source = q.source_code, + bid = q.bid, + ask = q.ask, + "mark router: skipping unusable quote" + ); + continue; + } + + // instrument_class is immutable, so cache it forever after one lookup. + let class = match class_of(&pool, &mut classes, q.instrument_id).await { + Some(c) => c, + None => continue, + }; + let rank = *policy.get(&(q.source_code.to_string(), class)).unwrap_or(&UNRANKED); + + if !accept(&mut active, &q, rank) { + continue; + } + + marks.set(q.instrument_id, q.bid, q.ask); + count += 1; + if count % 10_000 == 0 { + debug!(quotes = count, "mark router: throughput"); + } + } + + info!(quotes = count, "mark router: all feeds closed, stopping"); +} + +/// Decide whether this quote may set the mark, updating ownership as a side effect. +/// +/// Logs every ownership change at info: a silent failover is one you cannot +/// diagnose afterwards, and "why is the mark 0.15% off" is exactly the question +/// this answers. +fn accept(active: &mut HashMap, q: &Quote, rank: i32) -> bool { + match active.get(&q.instrument_id) { + None => { + info!( + instrument_id = q.instrument_id, + source = q.source_code, + rank, + "mark router: source active" + ); + active.insert( + q.instrument_id, + Active { source: q.source_code, rank, at: q.ts_recv }, + ); + true + } + Some(cur) if cur.source == q.source_code => { + // Incumbent: refresh its freshness clock. + active.insert( + q.instrument_id, + Active { source: q.source_code, rank, at: q.ts_recv }, + ); + true + } + Some(cur) if rank < cur.rank => { + // A better source appeared (or recovered) — take over immediately. + info!( + instrument_id = q.instrument_id, + from = cur.source, + to = q.source_code, + rank, + "mark router: promoting better-ranked source" + ); + active.insert( + q.instrument_id, + Active { source: q.source_code, rank, at: q.ts_recv }, + ); + true + } + Some(cur) => { + // Worse or equal rank: only usable if the incumbent has gone quiet. + let quiet_for = (q.ts_recv - cur.at).num_seconds(); + if quiet_for >= STALE_AFTER_SECS { + warn!( + instrument_id = q.instrument_id, + from = cur.source, + to = q.source_code, + quiet_for_secs = quiet_for, + "mark router: primary stale, failing over" + ); + active.insert( + q.instrument_id, + Active { source: q.source_code, rank, at: q.ts_recv }, + ); + true + } else { + false + } + } + } +} + +/// Can this book produce a meaningful mid? +/// +/// A zero *bid* is a real price, not a missing side: a deep-OTM option quoted +/// 0.00 x 0.05 is genuinely worth about 0.025, and rejecting it would report "no +/// data" for a contract the market is actively quoting — indistinguishable from +/// having no feed at all. The missing-side sentinel is the vendor's undefined +/// price, which feeds filter out before emitting. +/// +/// Genuinely unusable: no offer at all, a negative price, or a crossed book +/// (bid > ask), which means the two sides were read from inconsistent states. +fn is_usable(bid: f64, ask: f64) -> bool { + ask > 0.0 && bid >= 0.0 && bid <= ask +} + +async fn class_of( + pool: &PgPool, + cache: &mut HashMap, + instrument_id: i64, +) -> Option { + if let Some(c) = cache.get(&instrument_id) { + return Some(c.clone()); + } + let row = sqlx::query("SELECT instrument_class FROM instrument WHERE id = $1") + .bind(instrument_id) + .fetch_optional(pool) + .await + .ok()??; + let class: String = row.get("instrument_class"); + cache.insert(instrument_id, class.clone()); + Some(class) +} + +async fn load_policy(pool: &PgPool) -> Result, sqlx::Error> { + let rows = sqlx::query( + "SELECT source_code, instrument_class, rank \ + FROM oms.provider_feed_policy WHERE enabled", + ) + .fetch_all(pool) + .await?; + + Ok(rows + .iter() + .map(|r| { + ( + (r.get::("source_code"), r.get::("instrument_class")), + r.get::("rank"), + ) + }) + .collect()) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn quote(source: &'static str, at: DateTime) -> Quote { + Quote { + instrument_id: 1, + bid: 10.0, + ask: 10.1, + bid_size: 1, + ask_size: 1, + ts_recv: at, + source_code: source, + } + } + + fn t(secs: i64) -> DateTime { + DateTime::from_timestamp(1_700_000_000 + secs, 0).unwrap() + } + + #[test] + fn first_source_wins_the_instrument() { + let mut a = HashMap::new(); + assert!(accept(&mut a, "e("BINANCE", t(0)), 10)); + } + + #[test] + fn worse_ranked_source_is_ignored_while_primary_is_fresh() { + let mut a = HashMap::new(); + assert!(accept(&mut a, "e("BINANCE", t(0)), 10)); + // Bybit ticks 1s later; Binance is healthy, so Bybit must not touch the mark. + assert!(!accept(&mut a, "e("BYBIT", t(1)), 20)); + } + + #[test] + fn worse_ranked_source_takes_over_once_primary_goes_quiet() { + let mut a = HashMap::new(); + assert!(accept(&mut a, "e("BINANCE", t(0)), 10)); + assert!(!accept(&mut a, "e("BYBIT", t(5)), 20)); + // Binance silent for STALE_AFTER_SECS: Bybit is now the best live source. + assert!(accept(&mut a, "e("BYBIT", t(STALE_AFTER_SECS)), 20)); + } + + #[test] + fn primary_reclaims_immediately_when_it_returns() { + let mut a = HashMap::new(); + assert!(accept(&mut a, "e("BINANCE", t(0)), 10)); + assert!(accept(&mut a, "e("BYBIT", t(20)), 20)); // failed over + // No waiting: a better-ranked source preempts as soon as it ticks again. + assert!(accept(&mut a, "e("BINANCE", t(21)), 10)); + assert!(!accept(&mut a, "e("BYBIT", t(22)), 20)); + } + + #[test] + fn incumbent_keeps_refreshing_its_own_clock() { + let mut a = HashMap::new(); + assert!(accept(&mut a, "e("BINANCE", t(0)), 10)); + for s in 1..30 { + assert!(accept(&mut a, "e("BINANCE", t(s)), 10)); + } + // Never went stale, so the fallback still cannot preempt. + assert!(!accept(&mut a, "e("BYBIT", t(30)), 20)); + } + + #[test] + fn zero_bid_is_a_real_price() { + assert!(is_usable(0.0, 0.05)); + } + + #[test] + fn normal_two_sided_book_is_usable() { + assert!(is_usable(2.28, 2.29)); + } + + #[test] + fn locked_book_is_usable() { + assert!(is_usable(2.30, 2.30)); + } + + #[test] + fn no_offer_is_not_a_mark() { + assert!(!is_usable(0.0, 0.0)); + assert!(!is_usable(1.0, 0.0)); + } + + #[test] + fn crossed_book_is_rejected() { + assert!(!is_usable(2.50, 2.40)); + } + + #[test] + fn negative_price_is_rejected() { + assert!(!is_usable(-1.0, 2.0)); + } +} diff --git a/src/opra_stream.rs b/src/opra_stream.rs index c2070b1..247da89 100644 --- a/src/opra_stream.rs +++ b/src/opra_stream.rs @@ -1,16 +1,14 @@ -//! Databento OPRA live top-of-book → marks store. +//! Databento OPRA live top-of-book, as a [`LiveQuoteFeed`]. //! -//! Subscribes to `cmbp-1` (consolidated top-of-book across OPRA's venues — the -//! NBBO, not a single publisher's book) for the option symbols we currently hold -//! (`position.net_qty <> 0`) and writes each quote's mid inputs into the shared -//! [`crate::marks::MarkStore`], which the `/positions` valuation and -//! market-order pre-trade risk read. Mirrors the reconnect/backoff + -//! `stream_health` badge shape of `binance_stream.rs`. +//! Subscribes to `cmbp-1` — the consolidated top-of-book across OPRA's ~16 venues, +//! i.e. the NBBO rather than one publisher's book. (`bbo-1s` is not offered on this +//! dataset; `tcbbo` decodes to the same record but only ticks on trades, which +//! would leave an illiquid contract unmarked for hours.) //! -//! OPRA is enormous, so we subscribe only to the held set. Databento maps each -//! `raw_symbol` (the space-padded OSI, stored verbatim in `instrument.symbol`) -//! to a numeric `instrument_id` via a `SymbolMappingMsg`; quote records are then -//! keyed by that numeric id, so we keep a `dbn_id → our_id` table. +//! Everything not specific to Databento's wire format lives elsewhere: +//! [`crate::quote_feed`] decides what to subscribe to and watches for new +//! positions, [`crate::stream_supervisor`] reconnects, and +//! [`crate::mark_router`] decides what becomes a mark. use std::collections::HashMap; @@ -19,178 +17,143 @@ use databento::{ live::Subscription, LiveClient, }; -use sqlx::{PgPool, Row}; +use dataprovider::{DataProvider, FeedHealth, LiveQuoteFeed, ProviderError, Quote, SymbolAdds}; use tokio::sync::mpsc; -use tokio::time::{sleep, Duration}; use tracing::{info, warn}; -use crate::marks::MarkStore; -use crate::stream_health::StreamHandle; - const OPRA_DATASET: &str = "OPRA.PILLAR"; +const SOURCE_CODE: &str = "DATABENTO"; -/// `position_changed_rx` is owned here and borrowed into each session: the -/// doorbell outlives reconnects, since it has nothing to do with the Databento -/// session. -pub async fn run( - pool: PgPool, - marks: MarkStore, - health: StreamHandle, - mut position_changed_rx: mpsc::Receiver<()>, -) { - info!("starting Databento OPRA marks stream"); - - let mut backoff_secs: u64 = 1; - loop { - // Set the health status to `connecting`, then reset it to live on `success`. - health.set_connecting(); - match connect_and_run(&pool, &marks, &health, &mut position_changed_rx).await { - Ok(()) => { - warn!("OPRA stream closed, reconnecting in {backoff_secs}s"); - health.set_down("stream closed"); - } - Err(e) => { - warn!(error = %e, "OPRA 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); +pub struct DatabentoOpraFeed; + +impl DataProvider for DatabentoOpraFeed { + fn code(&self) -> &'static str { + SOURCE_CODE } } -async fn connect_and_run( - pool: &PgPool, - marks: &MarkStore, - health: &StreamHandle, - position_changed_rx: &mut mpsc::Receiver<()>, -) -> Result<(), Box> { - // Get {osi-symbol: instrument_id} map for currently-held options. - let mut sym_to_id = load_held_option_symbols(pool).await?; - - // If we have no open positions, sleep until a reconnect re-checks the held set. - if sym_to_id.is_empty() { - info!("OPRA stream: no open option positions, nothing to subscribe"); - sleep(Duration::from_secs(60)).await; - return Ok(()); +#[async_trait::async_trait] +impl LiveQuoteFeed for DatabentoOpraFeed { + /// OPRA.PILLAR is an options dataset. Databento cross-references our equities + /// too, but this session cannot quote them. + fn covers(&self) -> &'static [&'static str] { + &["OPTION"] } - // Collect symbol to subscribe to. - let symbols: Vec = sym_to_id.keys().cloned().collect(); - info!(count = symbols.len(), "OPRA stream: subscribing held options"); - - // Connect + subscribe to OPRA live - // TODO: Add other live data feeds dynamically. - let mut client = LiveClient::builder() - .key_from_env()? // DATABENTO_API_KEY - .dataset(OPRA_DATASET) - .build() - .await?; - - client - .subscribe( - Subscription::builder() - .symbols(symbols) - .schema(Schema::Cmbp1) - .stype_in(SType::RawSymbol) - .build(), - ) - .await?; - - client.start().await?; - health.set_live(); - info!("OPRA stream: live"); - - // Databento's numeric instrument_id is session-scoped, so this map is rebuilt - // from each session's SymbolMappingMsgs and dies with the session. It has to - // live here, not in a helper: if it sat inside a future `select!` can drop, - // losing that branch would take the session's id map with it. - let mut dbn_to_id: HashMap = HashMap::new(); - // Symbols held but not yet subscribed, queued by the doorbell branch below. - let mut pending: Vec = Vec::new(); - - loop { - // Subscribe outside the select: `subscribe` is NOT cancel-safe — a dropped - // partial send makes the gateway reject it and close the connection. - if !pending.is_empty() { - info!(count = pending.len(), "OPRA stream: subscribing newly-held options"); - client - .subscribe( - Subscription::builder() - .symbols(std::mem::take(&mut pending)) - .schema(Schema::Cmbp1) - .stype_in(SType::RawSymbol) - .build(), - ) - .await?; - } + async fn run_session( + &self, + symbols: HashMap, + out: &mpsc::Sender, + add_rx: &mut mpsc::Receiver, + health: &dyn FeedHealth, + ) -> Result<(), ProviderError> { + // Databento's raw_symbol is the space-padded OSI, stored verbatim in + // instrument.symbol, so the keys here need no transform. + let mut sym_to_id = symbols; + + let mut client = LiveClient::builder() + .key_from_env() + .map_err(|e| ProviderError::Config(e.to_string()))? + .dataset(OPRA_DATASET) + .build() + .await + .map_err(|e| ProviderError::Request(e.to_string()))?; + + subscribe(&mut client, sym_to_id.keys().cloned().collect()).await?; + client + .start() + .await + .map_err(|e| ProviderError::Request(e.to_string()))?; + health.on_connected(); + info!("OPRA feed: live"); + + // Databento's numeric instrument_id is scoped to (dataset, day) and rotates, + // so this map is rebuilt from each session's SymbolMappingMsgs and dies with + // the session. It must live here, outside any future `select!` can drop — + // losing it mid-session would silently mis-key every subsequent quote. + let mut dbn_to_id: HashMap = HashMap::new(); + let mut pending: Vec = Vec::new(); + + loop { + // Subscribe outside the select: `subscribe` is not cancel-safe — the + // crate warns a dropped partial send makes the gateway reject it and + // close the connection. + if !pending.is_empty() { + info!(count = pending.len(), "OPRA feed: subscribing new symbols"); + subscribe(&mut client, std::mem::take(&mut pending)).await?; + } - tokio::select! { - // Doorbell rang: a fill moved a position. Re-read the held set and diff - // it — we query the DB rather than trust a payload, which is why the - // signal carries none. - Some(()) = position_changed_rx.recv() => { - for (sym, id) in load_held_option_symbols(pool).await? { - if !sym_to_id.contains_key(&sym) { - pending.push(sym.clone()); - sym_to_id.insert(sym, id); + tokio::select! { + adds = add_rx.recv() => match adds { + Some(adds) => { + pending.extend(adds.keys().cloned()); + sym_to_id.extend(adds); } - } - } - // Cancel-safe, so losing this branch just drops it mid-await. - rec = client.next_record() => { - let Some(rec) = rec? else { break }; - health.record_event(); - - // Symbol mapping: numeric instrument_id ↔ raw OSI. Handled in the - // loop, not once at startup, so symbols added mid-session map too. - if let Some(sm) = rec.get::() { - if let Ok(osi) = sm.stype_out_symbol() { - if let Some(&our_id) = sym_to_id.get(osi) { - dbn_to_id.insert(sm.hd.instrument_id, our_id); + // Driver gone; nothing more will be added, but keep streaming. + None => std::future::pending::<()>().await, + }, + // `next_record` is cancel-safe, so losing this branch merely drops it. + rec = client.next_record() => { + let rec = rec.map_err(|e| ProviderError::Request(e.to_string()))?; + let Some(rec) = rec else { break }; + health.on_event(); + + // Symbol mapping: the gateway's numeric id for this session. + // Handled in the loop, not once at startup, so symbols added + // mid-session get mapped too. + if let Some(sm) = rec.get::() { + if let Ok(osi) = sm.stype_out_symbol() { + if let Some(&our_id) = sym_to_id.get(osi) { + dbn_to_id.insert(sm.hd.instrument_id, our_id); + } } + continue; } - continue; - } - // Consolidated top-of-book: update the mark for the mapped instrument. - if let Some(quote) = rec.get::() { - let level = "e.levels[0]; - if let (Some(bid), Some(ask)) = (px(level.bid_px), px(level.ask_px)) { - if let Some(&our_id) = dbn_to_id.get("e.hd.instrument_id) { - marks.set(our_id, bid, ask); + if let Some(q) = rec.get::() { + let level = &q.levels[0]; + let (Some(bid), Some(ask)) = (px(level.bid_px), px(level.ask_px)) else { + continue; + }; + let Some(&instrument_id) = dbn_to_id.get(&q.hd.instrument_id) else { + continue; // quote for something we never mapped + }; + let quote = Quote { + instrument_id, + bid, + ask, + bid_size: level.bid_sz, + ask_size: level.ask_sz, + ts_recv: chrono::Utc::now(), + source_code: SOURCE_CODE, + }; + if out.send(quote).await.is_err() { + warn!("OPRA feed: mark router gone, ending session"); + return Ok(()); } } } } } - } - Ok(()) + Ok(()) + } } +async fn subscribe(client: &mut LiveClient, symbols: Vec) -> Result<(), ProviderError> { + client + .subscribe( + Subscription::builder() + .symbols(symbols) + .schema(Schema::Cmbp1) + .stype_in(SType::RawSymbol) + .build(), + ) + .await + .map_err(|e| ProviderError::Request(e.to_string())) +} /// Fixed-point Databento price → f64, or `None` when undefined. fn px(v: i64) -> Option { (v != UNDEF_PRICE).then_some(v as f64 / 1e9) } - -/// Load the space-padded OSI → our `instrument.id` map for currently-held options. -async fn load_held_option_symbols( - pool: &PgPool, -) -> Result, sqlx::Error> { - let rows = sqlx::query( - // position.instrument_id is text holding the numeric instrument.id. - "SELECT i.symbol, i.id \ - FROM position p \ - JOIN instrument i ON i.id::text = p.instrument_id \ - WHERE p.net_qty <> 0 AND i.instrument_class = 'OPTION'", - ) - .fetch_all(pool) - .await?; - - Ok(rows - .iter() - .map(|r| (r.get::("symbol"), r.get::("id"))) - .collect()) -} diff --git a/src/positions.rs b/src/positions.rs index 8211a68..0922528 100644 --- a/src/positions.rs +++ b/src/positions.rs @@ -82,7 +82,31 @@ impl PositionMath { } } +/// Mark-to-market valuation of an open position. +/// +/// Both figures are signed through `net_qty`, so a short (`net_qty < 0`) yields a +/// negative market value and profits when the mark falls below `avg_cost`. +/// `contract_size` scales into money — 100 for an option, 1 for spot. +pub fn value(net_qty: f64, avg_cost: f64, contract_size: f64, mid: f64) -> Valuation { + Valuation { + market_value: net_qty * mid * contract_size, + unrealized_pnl: (mid - avg_cost) * net_qty * contract_size, + } +} + +#[derive(Debug, Clone, Copy)] +pub struct Valuation { + pub market_value: f64, + pub unrealized_pnl: f64, +} + /// A position row as returned by the API. +/// +/// `realized_pnl` comes from the projection; the mark-dependent fields are joined +/// in from the in-memory [`crate::marks::MarkStore`] at request time. They are +/// `None` when no live mark exists for the instrument — a position we can hold but +/// not value (no market-data source, or the feed is down) is a first-class state, +/// not an error, so it renders as a gap rather than a misleading zero. #[derive(Debug, Serialize, utoipa::ToSchema)] pub struct Position { pub portfolio_id: Uuid, @@ -91,6 +115,14 @@ pub struct Position { pub avg_cost: f64, pub realized_pnl: f64, pub updated_at: DateTime, + /// Mid of the current top-of-book quote. + pub mark: Option, + /// `net_qty * mark * contract_size` — signed, so shorts are negative. + pub market_value: Option, + /// `(mark - avg_cost) * net_qty * contract_size` — signed-correct for shorts. + pub unrealized_pnl: Option, + /// When the mark was received; lets a caller judge staleness. + pub mark_ts: Option>, } /// Read the current position (FOR UPDATE), or FLAT if none. @@ -247,6 +279,42 @@ mod tests { approx(p.realized_pnl, 0.0); } + #[test] + fn value_long_option_scales_by_contract_size() { + // 1 contract, paid 2.00, now marked 3.28 -> +1.28 x 100. + let v = value(1.0, 2.00, 100.0, 3.28); + approx(v.market_value, 328.0); + approx(v.unrealized_pnl, 128.0); + } + + #[test] + fn value_short_is_signed_correct() { + // Short 1 contract at 3.00; mark falls to 2.00 -> the short PROFITS. + let v = value(-1.0, 3.00, 100.0, 2.00); + approx(v.market_value, -200.0); // liability + approx(v.unrealized_pnl, 100.0); // gain + } + + #[test] + fn value_short_loses_when_mark_rises() { + let v = value(-1.0, 3.00, 100.0, 4.00); + approx(v.unrealized_pnl, -100.0); + } + + #[test] + fn value_spot_uses_unit_contract_size() { + // 0.1 SOL bought at 100, marked 150. + let v = value(0.1, 100.0, 1.0, 150.0); + approx(v.market_value, 15.0); + approx(v.unrealized_pnl, 5.0); + } + + #[test] + fn value_at_cost_is_flat() { + let v = value(10.0, 42.0, 1.0, 42.0); + approx(v.unrealized_pnl, 0.0); + } + #[test] fn add_to_long_blends_cost() { let p = PositionMath::FLAT diff --git a/src/quote_feed.rs b/src/quote_feed.rs new file mode 100644 index 0000000..ff29d89 --- /dev/null +++ b/src/quote_feed.rs @@ -0,0 +1,167 @@ +//! Generic driver for any [`LiveQuoteFeed`]. +//! +//! Owns everything that is the same regardless of vendor: work out what to +//! subscribe to, hand it to the feed, watch the position doorbell, and push newly +//! held instruments in mid-session. The feed itself only speaks its wire protocol. +//! +//! Pairing this with [`crate::stream_supervisor`] means a new vendor is one +//! `impl LiveQuoteFeed` — no reconnect loop, no health wiring, no DB access, no +//! subscription bookkeeping. + +use std::collections::HashMap; + +use dataprovider::{LiveQuoteFeed, Quote, SymbolAdds}; +use sqlx::{PgPool, Row}; +use tokio::sync::mpsc; +use tokio::time::{sleep, Duration}; +use tracing::info; + +use crate::stream_health::StreamHandle; +use crate::stream_supervisor::{Session, StreamResult}; + +/// How long to idle before re-checking when nothing is held. A reconnect re-runs +/// the query, so this is just the poll interval for "did we buy anything yet". +const IDLE_RECHECK_SECS: u64 = 60; + +pub struct QuoteFeedSession { + feed: F, + pool: PgPool, + out: mpsc::Sender, + position_changed_rx: mpsc::Receiver<()>, + health: StreamHandle, +} + +impl QuoteFeedSession { + pub fn new( + feed: F, + pool: PgPool, + out: mpsc::Sender, + position_changed_rx: mpsc::Receiver<()>, + health: StreamHandle, + ) -> Self { + Self { feed, pool, out, position_changed_rx, health } + } + + /// Block until there is something to subscribe to. + /// + /// Deliberately does *not* return to the supervisor when nothing is held: an + /// idle feed is not a closed stream, and reporting it as one would flap the + /// health badge and log a reconnect warning every minute for a feed behaving + /// exactly as designed. Waits on the doorbell so a fill is picked up at once, + /// with a timer as the backstop for positions this process didn't see. + async fn wait_for_subscribable(&mut self) -> Result, sqlx::Error> { + loop { + let held = load_subscribable(&self.pool, self.feed.code(), self.feed.covers()).await?; + if !held.is_empty() { + return Ok(held); + } + tokio::select! { + _ = self.position_changed_rx.recv() => {} + _ = sleep(Duration::from_secs(IDLE_RECHECK_SECS)) => {} + } + } + } +} + +#[async_trait::async_trait] +impl Session for QuoteFeedSession { + async fn run_once(&mut self) -> StreamResult { + let known = self.wait_for_subscribable().await?; + info!(source = self.feed.code(), count = known.len(), "quote feed: subscribing held set"); + + let (add_tx, mut add_rx) = mpsc::channel::(8); + + // The watcher never ends the session — only the feed decides that. Racing + // them lets a fill be picked up without waiting for a reconnect. + tokio::select! { + r = self.feed.run_session(known.clone(), &self.out, &mut add_rx, &self.health) => r.map_err(Into::into), + r = watch_held(&self.pool, self.feed.code(), self.feed.covers(), &mut self.position_changed_rx, &add_tx, known) => r, + } + } +} + +/// On each doorbell ring, re-read the held set and push anything new to the feed. +/// +/// Re-reads rather than trusting a payload: the DB is the truth, and a signal that +/// carries no data cannot carry stale data. +async fn watch_held( + pool: &PgPool, + source_code: &'static str, + covers: &'static [&'static str], + doorbell: &mut mpsc::Receiver<()>, + add_tx: &mpsc::Sender, + mut known: HashMap, +) -> StreamResult { + loop { + match doorbell.recv().await { + Some(()) => { + let latest = load_subscribable(pool, source_code, covers).await?; + let adds: SymbolAdds = latest + .into_iter() + .filter(|(sym, _)| !known.contains_key(sym)) + .collect(); + if adds.is_empty() { + continue; + } + info!( + source = source_code, + count = adds.len(), + "quote feed: subscribing newly-held instruments" + ); + known.extend(adds.iter().map(|(s, i)| (s.clone(), *i))); + if add_tx.send(adds).await.is_err() { + // The feed dropped its receiver: the session is ending anyway. + // Park rather than return, so the feed's own result is what the + // supervisor sees. + std::future::pending::<()>().await; + } + } + // Doorbell closed (only if every fill stream is gone). Never end the + // session over it — a feed with no doorbell still streams fine. + None => std::future::pending::<()>().await, + } + } +} + +/// The held instruments this feed can cover, as `external_symbol -> instrument_id`. +/// +/// Driven by `instrument_xref`, which is the point: the *provider's own* symbol +/// drives the subscription, so a feed is no longer required to use strings that +/// happen to match ours. The previous query read `instrument.symbol` directly and +/// worked only by the coincidence that Databento's raw symbol is byte-identical to +/// our master symbol — a coincidence that holds for OSI options and for nothing +/// else. Binance calls it `SOLUSDT` where we call it `SOL`. +/// +/// Coverage is `(source_code, instrument_class)`. Both halves are load-bearing: +/// source_code alone would hand this feed the held SPY *equity*, which is +/// cross-referenced to DATABENTO but not quotable on OPRA.PILLAR. +/// +/// An instrument with no xref row for this source is silently absent — that is the +/// "tradable but unmarkable" state, not an error. +async fn load_subscribable( + pool: &PgPool, + source_code: &str, + covers: &[&str], +) -> Result, sqlx::Error> { + let classes: Vec = covers.iter().map(|c| c.to_string()).collect(); + let rows = sqlx::query( + // position.instrument_id is text holding the numeric instrument.id. + "SELECT DISTINCT x.external_symbol, i.id \ + FROM position p \ + JOIN instrument i ON i.id::text = p.instrument_id \ + JOIN oms.instrument_xref x ON x.instrument_id = i.id \ + AND x.source_type = 'PROVIDER' AND x.source_code = $1 \ + WHERE p.net_qty <> 0 \ + AND i.instrument_class = ANY($2) \ + AND x.external_symbol IS NOT NULL", + ) + .bind(source_code) + .bind(&classes) + .fetch_all(pool) + .await?; + + Ok(rows + .iter() + .map(|r| (r.get::("external_symbol"), r.get::("id"))) + .collect()) +} diff --git a/src/stream_health.rs b/src/stream_health.rs index 575bf8a..9f9b663 100644 --- a/src/stream_health.rs +++ b/src/stream_health.rs @@ -132,3 +132,15 @@ impl StreamHandle { self.mutate(|h| h.last_event_at = Some(now)); } } + +/// Lets a `dataprovider` feed report liveness without that crate knowing about +/// this registry. +impl dataprovider::FeedHealth for StreamHandle { + fn on_connected(&self) { + self.set_live(); + } + + fn on_event(&self) { + self.record_event(); + } +} diff --git a/src/stream_supervisor.rs b/src/stream_supervisor.rs new file mode 100644 index 0000000..0436468 --- /dev/null +++ b/src/stream_supervisor.rs @@ -0,0 +1,127 @@ +//! Shared supervision for long-lived reconnecting streams. +//! +//! Every stream task — Alpaca and Binance order streams, the Databento OPRA marks +//! feed — wants the same outer loop: mark the health badge `connecting`, run one +//! session until it closes or errors, mark it `down`, wait, reconnect. Only the +//! session body differs. This module owns the loop; [`Session`] is the seam. +//! +//! The streams are deliberately *not* unified beyond this. Order streams emit +//! execution reports; the marks feed emits quotes. They share mechanics, not +//! purpose — a single trait over both would be the wrong abstraction. + +use std::time::Instant; + +use tokio::time::{sleep, Duration}; +use tracing::{info, warn}; + +use crate::stream_health::StreamHandle; + +pub type StreamResult = Result<(), Box>; + +/// One connected session of a stream. +/// +/// `run_once` returns when the connection ends: `Ok(())` for a clean close, +/// `Err` for a failure. Either way the supervisor reconnects — the distinction +/// only affects what gets logged and shown on the health badge. +#[async_trait::async_trait] +pub trait Session: Send { + async fn run_once(&mut self) -> StreamResult; +} + +const INITIAL_BACKOFF_SECS: u64 = 1; +const MAX_BACKOFF_SECS: u64 = 30; + +/// A session that stayed up at least this long is treated as healthy, so its next +/// failure restarts the backoff from the bottom. Without this, backoff only ever +/// grows for the life of the process: a stream that ran for hours and then dropped +/// once would wait the last capped delay before retrying, as if it were flapping. +const STABLE_AFTER_SECS: u64 = 60; + +/// Run `session` forever, reconnecting with exponential backoff and keeping the +/// health badge current. +pub async fn supervise(name: &'static str, health: StreamHandle, mut session: S) -> ! { + let mut backoff_secs = INITIAL_BACKOFF_SECS; + + loop { + health.set_connecting(); + info!(stream = name, "connecting"); + + let started = Instant::now(); + match session.run_once().await { + Ok(()) => { + health.set_down("stream closed"); + warn!(stream = name, "stream closed, reconnecting in {backoff_secs}s"); + } + Err(e) => { + health.set_down(e.to_string()); + warn!(stream = name, error = %e, "stream error, reconnecting in {backoff_secs}s"); + } + } + + if started.elapsed() >= Duration::from_secs(STABLE_AFTER_SECS) { + backoff_secs = INITIAL_BACKOFF_SECS; + } + + sleep(Duration::from_secs(backoff_secs)).await; + backoff_secs = (backoff_secs * 2).min(MAX_BACKOFF_SECS); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + /// The backoff schedule, extracted so the growth/reset rule is testable without + /// running a supervisor. Mirrors the arithmetic in `supervise`. + fn next_backoff(current: u64, session_lasted: Duration) -> u64 { + let base = if session_lasted >= Duration::from_secs(STABLE_AFTER_SECS) { + INITIAL_BACKOFF_SECS + } else { + current + }; + (base * 2).min(MAX_BACKOFF_SECS) + } + + #[test] + fn backoff_grows_while_flapping() { + let brief = Duration::from_secs(0); + let mut b = INITIAL_BACKOFF_SECS; + b = next_backoff(b, brief); + assert_eq!(b, 2); + b = next_backoff(b, brief); + assert_eq!(b, 4); + b = next_backoff(b, brief); + assert_eq!(b, 8); + } + + #[test] + fn backoff_caps() { + let brief = Duration::from_secs(0); + let mut b = INITIAL_BACKOFF_SECS; + for _ in 0..12 { + b = next_backoff(b, brief); + } + assert_eq!(b, MAX_BACKOFF_SECS); + } + + /// The bug this module exists to fix: a long-lived session used to inherit the + /// last capped delay, so a stream up for hours retried as if it were flapping. + #[test] + fn stable_session_resets_backoff() { + let brief = Duration::from_secs(0); + let mut b = INITIAL_BACKOFF_SECS; + for _ in 0..10 { + b = next_backoff(b, brief); + } + assert_eq!(b, MAX_BACKOFF_SECS, "should be capped after a flapping burst"); + + let stable = Duration::from_secs(STABLE_AFTER_SECS); + assert_eq!(next_backoff(b, stable), 2, "a healthy session restarts the schedule"); + } + + #[test] + fn session_just_under_threshold_does_not_reset() { + let almost = Duration::from_secs(STABLE_AFTER_SECS - 1); + assert_eq!(next_backoff(16, almost), 32.min(MAX_BACKOFF_SECS)); + } +}