From 46c5c598bafd20aac647471da07a052f3787b3e2 Mon Sep 17 00:00:00 2001 From: Max Date: Fri, 17 Jul 2026 08:42:59 +0200 Subject: [PATCH 1/8] feat(positions): value open positions against the live mark MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit MarkStore has been write-only since it was added: opra_stream was the only writer and no get/get_all caller existed anywhere in src/, so every quote we decoded was discarded. The module doc claimed "/positions valuation and market-order pre-trade risk read" this store; neither consumer was ever built, and 0011's comment ("unrealized P&L is intentionally absent — it needs a market-data mark") has been waiting for exactly this. Join the mark in at request time: /portfolios/:id/positions now returns mark, market_value, unrealized_pnl and mark_ts. The query gains contract_size from the master (LEFT JOIN, defaulting to 1) so options scale by 100 while spot does not. The mark-dependent fields are Option. A position we can hold but not value — no provider covers it, or the feed is down — renders as null rather than 0, because a zero would read as "flat P&L" instead of "unknown". This is not hypothetical: the three Binance crypto instruments are routable but have no PROVIDER xref row, so they are structurally unmarkable today. The valuation itself is extracted to positions::value() rather than left inline in the handler, because sign errors on shorts are the classic bug here and they belong under test. Both figures are signed through net_qty: a short carries a negative market value and gains when the mark falls below avg_cost. Marks are snapshotted once per request so every row is valued against the same instant rather than racing the feed mid-loop. --- src/handlers.rs | 56 +++++++++++++++++++++++++++++---------- src/positions.rs | 68 ++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 110 insertions(+), 14 deletions(-) diff --git a/src/handlers.rs b/src/handlers.rs index 893c173..b6451a2 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,37 @@ pub async fn get_portfolio_positions( message: format!("failed to load positions: {:?}", err), })?; + // One snapshot for the whole response, so every row is valued against the same + // instant rather than racing the feed mid-loop. + let marks = state.marks().get_all(); + 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).copied()); + 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/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 From 5642f1f8cb67881b8280513511c0f20278f14b28 Mon Sep 17 00:00:00 2001 From: Max Date: Fri, 17 Jul 2026 08:47:49 +0200 Subject: [PATCH 2/8] refactor(streams): one supervisor for reconnect, backoff and health MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit All three stream tasks had grown the same outer loop, verbatim: set the health badge connecting, run a session, set it down, sleep, double the backoff, retry. Only the session body differed. Extract the loop into stream_supervisor and give each stream a Session impl holding what its session needs. This fixes a bug all three shared. backoff_secs was initialized outside the loop and never reset, so it only ever grew for the life of the process: a stream that stayed up for hours and then dropped once would wait the last capped 30s before retrying, as though it were flapping. A session that lasted STABLE_AFTER_SECS is now treated as healthy and its next failure restarts the schedule from 1s. The backoff rule is unit-tested, including that reset. The streams are deliberately not unified further. Alpaca and Binance emit execution reports; OPRA emits quotes. They share mechanics, not purpose, so a single trait over both would be the wrong abstraction — Session is only the reconnect seam. Also drops binance_stream::run's _api_key/_api_secret parameters. They were dead: the adapter already holds the credentials and signs with an Ed25519 key loaded at construction, so main.rs was passing two String::new() to satisfy a signature that existed only to mirror Alpaca's. Streams now name themselves to the supervisor (ALPACA/PAPER, BINANCE/PAPER, DATABENTO/OPRA), so reconnect logs say which stream without reading the module path. Verified: all three connect and reach live/connecting as before; 66 tests pass. --- src/alpaca_stream.rs | 61 +++++++++++++------ src/binance_stream.rs | 55 +++++++++++------ src/main.rs | 3 +- src/opra_stream.rs | 56 ++++++++++------- src/stream_supervisor.rs | 127 +++++++++++++++++++++++++++++++++++++++ 5 files changed, 239 insertions(+), 63 deletions(-) create mode 100644 src/stream_supervisor.rs 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_stream.rs b/src/binance_stream.rs index e02f855..c64ae88 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,14 @@ 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, + }; + supervise("BINANCE/PAPER", health, session).await } async fn connect_and_run( diff --git a/src/main.rs b/src/main.rs index 7d2bb8d..cf3ab1a 100644 --- a/src/main.rs +++ b/src/main.rs @@ -53,6 +53,7 @@ mod binance_stream; mod stream_health; mod marks; mod opra_stream; +mod stream_supervisor; #[derive(OpenApi)] #[openapi( @@ -386,7 +387,7 @@ 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. diff --git a/src/opra_stream.rs b/src/opra_stream.rs index c2070b1..5c13ce0 100644 --- a/src/opra_stream.rs +++ b/src/opra_stream.rs @@ -22,41 +22,51 @@ use databento::{ use sqlx::{PgPool, Row}; use tokio::sync::mpsc; use tokio::time::{sleep, Duration}; -use tracing::{info, warn}; +use tracing::info; use crate::marks::MarkStore; use crate::stream_health::StreamHandle; +use crate::stream_supervisor::{supervise, Session, StreamResult}; const OPRA_DATASET: &str = "OPRA.PILLAR"; -/// `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. +/// Holds what one OPRA session needs. `position_changed_rx` lives here rather than +/// inside a session body: the doorbell outlives reconnects, since it has nothing to +/// do with the Databento session. +struct OpraSession { + pool: PgPool, + marks: MarkStore, + health: StreamHandle, + position_changed_rx: mpsc::Receiver<()>, +} + +#[async_trait::async_trait] +impl Session for OpraSession { + async fn run_once(&mut self) -> StreamResult { + connect_and_run( + &self.pool, + &self.marks, + &self.health, + &mut self.position_changed_rx, + ) + .await + } +} + pub async fn run( pool: PgPool, marks: MarkStore, health: StreamHandle, - mut position_changed_rx: mpsc::Receiver<()>, + 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); - } + let session = OpraSession { + pool, + marks, + health: health.clone(), + position_changed_rx, + }; + supervise("DATABENTO/OPRA", health, session).await } async fn connect_and_run( 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)); + } +} From 89f3fd94e57295b98a2acc8001cb3a75308906ce Mon Sep 17 00:00:00 2001 From: Max Date: Fri, 17 Jul 2026 09:00:06 +0200 Subject: [PATCH 3/8] feat(dataprovider): add the LiveQuoteFeed capability; port OPRA onto it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit crates/dataprovider has advertised a "quote source" capability since it was written — provider.rs names it in the doc comment beside UniverseSource and Enricher — but it never existed, so the OPRA feed grew in the app as one function that connected a socket, queried the database, decoded DBN, and wrote MarkStore. Adding Binance or Bybit meant copying all four concerns. Split them along the seams that actually differ per vendor: LiveQuoteFeed (dataprovider) wire protocol -> Quote. Nothing else. quote_feed::QuoteFeedSession what to subscribe to; watches the doorbell and pushes newly-held instruments mid-session. stream_supervisor reconnect, backoff, health. mark_router the sole writer to MarkStore. A new vendor is now one impl LiveQuoteFeed: no reconnect loop, no health wiring, no SQL, no subscription bookkeeping. Vendor concepts stay contained — Databento's session-scoped numeric ids never reach the trait, because they mean nothing to Binance. Quote carries ts_recv and source_code. ts_recv is deliberately our clock, so a vendor with a skewed or missing timestamp cannot make a stale quote look fresh; source_code is provenance, without which failover cannot arbitrate between sources it cannot name and a mark cannot be audited. Feeds emit rather than write. That makes a feed testable with nothing but a channel, and lets marks/Kafka/cockpit fan out later without reopening any feed. mark_router drops one-sided books: a mid computed from a missing side is silently wrong, and a stale real mark beats a fabricated one. Ranked failover is not built. No instrument currently has more than one provider, so it would be machinery with nothing to arbitrate; mark_router is where it goes when a second feed lands. load_subscribable still reads instrument.symbol and filters to options — it works only because Databento's raw symbol is byte-identical to ours. P3 replaces the body with the instrument_xref PROVIDER join, which is what makes the driver genuinely vendor-neutral. Verified: 66 tests; router starts, feed idles correctly with nothing held, all three streams reach live/connecting as before. --- crates/dataprovider/src/lib.rs | 3 + crates/dataprovider/src/quote.rs | 62 +++++++ src/main.rs | 17 +- src/mark_router.rs | 44 +++++ src/opra_stream.rs | 284 +++++++++++++------------------ src/quote_feed.rs | 133 +++++++++++++++ 6 files changed, 371 insertions(+), 172 deletions(-) create mode 100644 crates/dataprovider/src/quote.rs create mode 100644 src/mark_router.rs create mode 100644 src/quote_feed.rs diff --git a/crates/dataprovider/src/lib.rs b/crates/dataprovider/src/lib.rs index f6ee522..e7388eb 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::{LiveQuoteFeed, 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..ef7ad71 --- /dev/null +++ b/crates/dataprovider/src/quote.rs @@ -0,0 +1,62 @@ +//! 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; + +/// A source of live quotes. +#[async_trait::async_trait] +pub trait LiveQuoteFeed: DataProvider { + /// 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, + ) -> Result<(), ProviderError>; +} diff --git a/src/main.rs b/src/main.rs index cf3ab1a..6b686a4 100644 --- a/src/main.rs +++ b/src/main.rs @@ -54,6 +54,8 @@ mod stream_health; mod marks; mod opra_stream; mod stream_supervisor; +mod quote_feed; +mod mark_router; #[derive(OpenApi)] #[openapi( @@ -390,10 +392,21 @@ async fn serve() { 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())); + if env::var("DATABENTO_API_KEY").map(|k| !k.is_empty()).unwrap_or(false) { 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(), + position_changed_rx, + ); + tokio::spawn(stream_supervisor::supervise("DATABENTO/OPRA", health, session)); } // Register routes diff --git a/src/mark_router.rs b/src/mark_router.rs new file mode 100644 index 0000000..613609d --- /dev/null +++ b/src/mark_router.rs @@ -0,0 +1,44 @@ +//! Routes normalized quotes from every feed into the [`MarkStore`]. +//! +//! This is the only writer. Feeds emit [`Quote`]s onto a channel and never touch +//! the store, so adding a vendor cannot change how marks are decided, and the +//! policy for "which source wins" lives in exactly one place. +//! +//! Today that policy is trivial — one provider covers any given instrument, so +//! every quote is accepted. The ranked primary + failover arbitration (prefer the +//! lowest-ranked healthy source; demote on staleness; log the switch) belongs here +//! when a second feed exists. Building it now would be machinery with nothing to +//! arbitrate: no instrument currently has more than one provider. + +use dataprovider::Quote; +use tokio::sync::mpsc; +use tracing::{debug, info}; + +use crate::marks::MarkStore; + +/// Drain quotes into the store until every feed has dropped its sender. +pub async fn run(mut quotes: mpsc::Receiver, marks: MarkStore) { + info!("mark router: started"); + let mut count: u64 = 0; + + while let Some(q) = quotes.recv().await { + // A one-sided book is not a mark: a mid computed from a missing side would + // be silently wrong, and a stale-but-real mark beats a fabricated one. + if !(q.bid > 0.0 && q.ask > 0.0) { + debug!( + instrument_id = q.instrument_id, + source = q.source_code, + "mark router: skipping one-sided quote" + ); + 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"); +} diff --git a/src/opra_stream.rs b/src/opra_stream.rs index 5c13ce0..1dcff3a 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,188 +17,134 @@ use databento::{ live::Subscription, LiveClient, }; -use sqlx::{PgPool, Row}; +use dataprovider::{DataProvider, LiveQuoteFeed, ProviderError, Quote, SymbolAdds}; use tokio::sync::mpsc; -use tokio::time::{sleep, Duration}; -use tracing::info; - -use crate::marks::MarkStore; -use crate::stream_health::StreamHandle; -use crate::stream_supervisor::{supervise, Session, StreamResult}; +use tracing::{info, warn}; const OPRA_DATASET: &str = "OPRA.PILLAR"; +const SOURCE_CODE: &str = "DATABENTO"; -/// Holds what one OPRA session needs. `position_changed_rx` lives here rather than -/// inside a session body: the doorbell outlives reconnects, since it has nothing to -/// do with the Databento session. -struct OpraSession { - pool: PgPool, - marks: MarkStore, - health: StreamHandle, - position_changed_rx: mpsc::Receiver<()>, -} +pub struct DatabentoOpraFeed; -#[async_trait::async_trait] -impl Session for OpraSession { - async fn run_once(&mut self) -> StreamResult { - connect_and_run( - &self.pool, - &self.marks, - &self.health, - &mut self.position_changed_rx, - ) - .await +impl DataProvider for DatabentoOpraFeed { + fn code(&self) -> &'static str { + SOURCE_CODE } } -pub async fn run( - pool: PgPool, - marks: MarkStore, - health: StreamHandle, - position_changed_rx: mpsc::Receiver<()>, -) { - info!("starting Databento OPRA marks stream"); - let session = OpraSession { - pool, - marks, - health: health.clone(), - position_changed_rx, - }; - supervise("DATABENTO/OPRA", health, session).await -} - -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(()); - } - - // 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_trait::async_trait] +impl LiveQuoteFeed for DatabentoOpraFeed { + async fn run_session( + &self, + symbols: HashMap, + out: &mpsc::Sender, + add_rx: &mut mpsc::Receiver, + ) -> 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()))?; + 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 }; + + // 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/quote_feed.rs b/src/quote_feed.rs new file mode 100644 index 0000000..aa372c1 --- /dev/null +++ b/src/quote_feed.rs @@ -0,0 +1,133 @@ +//! 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_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<()>, +} + +impl QuoteFeedSession { + pub fn new( + feed: F, + pool: PgPool, + out: mpsc::Sender, + position_changed_rx: mpsc::Receiver<()>, + ) -> Self { + Self { feed, pool, out, position_changed_rx } + } +} + +#[async_trait::async_trait] +impl Session for QuoteFeedSession { + async fn run_once(&mut self) -> StreamResult { + let known = load_subscribable(&self.pool, self.feed.code()).await?; + if known.is_empty() { + info!(source = self.feed.code(), "quote feed: nothing held to subscribe, idling"); + sleep(Duration::from_secs(IDLE_RECHECK_SECS)).await; + return Ok(()); + } + 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) => r.map_err(Into::into), + r = watch_held(&self.pool, self.feed.code(), &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, + 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).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. +/// +/// Still the original OPRA-shaped query: held options keyed by the *master* +/// symbol, which works only because Databento's raw symbol happens to be +/// byte-identical to ours. P3 replaces the body with the `instrument_xref` +/// PROVIDER join so the feed's own `external_symbol` drives the subscription and +/// `source_code` selects coverage — that is what makes this genuinely generic, and +/// what lets a feed cover something other than options. +async fn load_subscribable( + pool: &PgPool, + _source_code: &str, +) -> 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()) +} From 19ad7a91aa15f4d4c0ca248a7b0aedb889e1d38c Mon Sep 17 00:00:00 2001 From: Max Date: Fri, 17 Jul 2026 09:12:59 +0200 Subject: [PATCH 4/8] fix(marks): restore feed health reporting; stop dropping zero-bid quotes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Self-review of the P0-P2 refactor caught two regressions it introduced, plus three smaller defects. Health reporting was lost. The pre-refactor opra_stream called health.set_live() after client.start() and health.record_event() per record; the new path called neither, because the feed no longer held a StreamHandle and the supervisor only knows set_connecting/set_down. DATABENTO/OPRA could therefore never leave "connecting" even while streaming, and last_event_at stayed null forever — which is precisely the freshness signal the planned staleness-based failover reads, so the source could never have been selected as active. Fixed with a FeedHealth port in dataprovider (on_connected/on_event) that StreamHandle implements, keeping the crate ignorant of our registry while restoring the exact prior behaviour. The mark router rejected zero-bid quotes. `!(bid > 0.0 && ask > 0.0)` treats a 0.00 x 0.05 book as one-sided, but a zero bid is a real price — that is what a deep-OTM option looks like near expiry, exactly where today's SPY 750 call is headed. Those contracts would report mark=null, indistinguishable from having no data source at all, when in fact the market is quoting them. The missing-side sentinel is UNDEF_PRICE, which feeds already filter. Now rejects only what cannot produce a mid: no offer, negative price, or a crossed book. is_usable() is extracted and tested, since this is the second time a plausible-looking guard here has been wrong. Also: - quote_feed no longer returns Ok(()) from run_once when nothing is held. The supervisor read that as a closed stream, so an idle feed logged a reconnect warning and flapped its badge down->connecting every 60s while behaving exactly as designed. It now waits on the doorbell (with a timer backstop), which also picks up a fill immediately instead of up to a minute later. - binance_stream passed a hardcoded "BINANCE/PAPER" to supervise despite taking an `environment` parameter, so a LIVE stream's reconnects would be logged as paper. - get_portfolio_positions cloned the entire MarkStore per request to read two or three entries; it now looks up per row. Cost was O(all marks), and the clone held the read lock against the router's writes. Verified: 72 tests; no flapping over 20s (was 1 warning/min); positions endpoint unchanged; all three streams reach live/connecting. --- crates/dataprovider/src/lib.rs | 2 +- crates/dataprovider/src/quote.rs | 23 ++++++++++++ src/binance_stream.rs | 3 +- src/handlers.rs | 13 ++++--- src/main.rs | 1 + src/mark_router.rs | 62 +++++++++++++++++++++++++++++--- src/opra_stream.rs | 5 ++- src/quote_feed.rs | 34 +++++++++++++----- src/stream_health.rs | 12 +++++++ 9 files changed, 133 insertions(+), 22 deletions(-) diff --git a/crates/dataprovider/src/lib.rs b/crates/dataprovider/src/lib.rs index e7388eb..813102e 100644 --- a/crates/dataprovider/src/lib.rs +++ b/crates/dataprovider/src/lib.rs @@ -27,7 +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::{LiveQuoteFeed, Quote, SymbolAdds}; +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 index ef7ad71..5f0fa27 100644 --- a/crates/dataprovider/src/quote.rs +++ b/crates/dataprovider/src/quote.rs @@ -44,6 +44,28 @@ impl Quote { /// 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 { @@ -58,5 +80,6 @@ pub trait LiveQuoteFeed: DataProvider { symbols: HashMap, out: &mpsc::Sender, add_rx: &mut mpsc::Receiver, + health: &dyn FeedHealth, ) -> Result<(), ProviderError>; } diff --git a/src/binance_stream.rs b/src/binance_stream.rs index c64ae88..dcf1899 100644 --- a/src/binance_stream.rs +++ b/src/binance_stream.rs @@ -77,7 +77,8 @@ pub async fn run( health: health.clone(), position_changed_tx, }; - supervise("BINANCE/PAPER", health, session).await + 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/handlers.rs b/src/handlers.rs index b6451a2..e46cc4d 100644 --- a/src/handlers.rs +++ b/src/handlers.rs @@ -1007,9 +1007,11 @@ pub async fn get_portfolio_positions( message: format!("failed to load positions: {:?}", err), })?; - // One snapshot for the whole response, so every row is valued against the same - // instant rather than racing the feed mid-loop. - let marks = state.marks().get_all(); + // 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() @@ -1019,10 +1021,7 @@ pub async fn get_portfolio_positions( 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).copied()); + 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)); diff --git a/src/main.rs b/src/main.rs index 6b686a4..f1ce2cd 100644 --- a/src/main.rs +++ b/src/main.rs @@ -405,6 +405,7 @@ async fn serve() { state.pool().clone(), quote_tx.clone(), position_changed_rx, + health.clone(), ); tokio::spawn(stream_supervisor::supervise("DATABENTO/OPRA", health, session)); } diff --git a/src/mark_router.rs b/src/mark_router.rs index 613609d..d929dd0 100644 --- a/src/mark_router.rs +++ b/src/mark_router.rs @@ -16,19 +16,33 @@ use tracing::{debug, info}; use crate::marks::MarkStore; +/// 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 +} + /// Drain quotes into the store until every feed has dropped its sender. pub async fn run(mut quotes: mpsc::Receiver, marks: MarkStore) { info!("mark router: started"); let mut count: u64 = 0; while let Some(q) = quotes.recv().await { - // A one-sided book is not a mark: a mid computed from a missing side would - // be silently wrong, and a stale-but-real mark beats a fabricated one. - if !(q.bid > 0.0 && q.ask > 0.0) { + if !is_usable(q.bid, q.ask) { debug!( instrument_id = q.instrument_id, source = q.source_code, - "mark router: skipping one-sided quote" + bid = q.bid, + ask = q.ask, + "mark router: skipping unusable quote" ); continue; } @@ -42,3 +56,43 @@ pub async fn run(mut quotes: mpsc::Receiver, marks: MarkStore) { info!(quotes = count, "mark router: all feeds closed, stopping"); } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn zero_bid_is_a_real_price() { + // A deep-OTM option near expiry: worth ~0.025, not "no data". + // Rejecting this was a regression that made such contracts unmarkable. + 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() { + // bid == ask is unusual but not contradictory; the mid is well defined. + 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() { + // bid > ask: the sides came from inconsistent states. + 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 1dcff3a..f0012ee 100644 --- a/src/opra_stream.rs +++ b/src/opra_stream.rs @@ -17,7 +17,7 @@ use databento::{ live::Subscription, LiveClient, }; -use dataprovider::{DataProvider, LiveQuoteFeed, ProviderError, Quote, SymbolAdds}; +use dataprovider::{DataProvider, FeedHealth, LiveQuoteFeed, ProviderError, Quote, SymbolAdds}; use tokio::sync::mpsc; use tracing::{info, warn}; @@ -39,6 +39,7 @@ impl LiveQuoteFeed for DatabentoOpraFeed { 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. @@ -57,6 +58,7 @@ impl LiveQuoteFeed for DatabentoOpraFeed { .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, @@ -88,6 +90,7 @@ impl LiveQuoteFeed for DatabentoOpraFeed { 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 diff --git a/src/quote_feed.rs b/src/quote_feed.rs index aa372c1..afd3759 100644 --- a/src/quote_feed.rs +++ b/src/quote_feed.rs @@ -16,6 +16,7 @@ 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 @@ -27,6 +28,7 @@ pub struct QuoteFeedSession { pool: PgPool, out: mpsc::Sender, position_changed_rx: mpsc::Receiver<()>, + health: StreamHandle, } impl QuoteFeedSession { @@ -35,20 +37,36 @@ impl QuoteFeedSession { pool: PgPool, out: mpsc::Sender, position_changed_rx: mpsc::Receiver<()>, + health: StreamHandle, ) -> Self { - Self { feed, pool, out, position_changed_rx } + 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()).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 = load_subscribable(&self.pool, self.feed.code()).await?; - if known.is_empty() { - info!(source = self.feed.code(), "quote feed: nothing held to subscribe, idling"); - sleep(Duration::from_secs(IDLE_RECHECK_SECS)).await; - return Ok(()); - } + 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); @@ -56,7 +74,7 @@ impl Session for QuoteFeedSession { // 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) => r.map_err(Into::into), + 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(), &mut self.position_changed_rx, &add_tx, known) => r, } } 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(); + } +} From 496d4fd9559ed9fdb5dbd7362afb275b11752188 Mon Sep 17 00:00:00 2001 From: Max Date: Fri, 17 Jul 2026 09:15:44 +0200 Subject: [PATCH 5/8] feat(marks): drive subscriptions from instrument_xref, not instrument.symbol MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The quote driver read instrument.symbol and filtered to instrument_class='OPTION' in SQL. That worked only by coincidence: Databento's raw_symbol happens to be byte-identical to our master symbol for OSI options, and for nothing else. Binance calls it SOLUSDT where we call it SOL, so the coincidence ends at the first non-Databento feed — the driver was vendor-neutral in shape only. Subscriptions now come from oms.instrument_xref PROVIDER rows, so the provider's own external_symbol drives the subscription and a feed is free to use whatever strings its venue uses. Verified parity over the full option universe: old and new queries return identical (symbol, id) sets — 27,252 rows, 0 mismatches — so this changes the mechanism without changing behaviour. Coverage needed a second half, which the data made obvious. source_code alone is not coverage: the held SPY *equity* carries a PROVIDER/DATABENTO xref row (external_symbol 'SPY' @ ARCX), so a source_code-only query hands it to the OPRA.PILLAR session — an options-only dataset that cannot quote it. LiveQuoteFeed therefore declares covers() -> &[instrument_class], and DatabentoOpraFeed declares OPTION. Confirmed against the live DB: without the filter the held set is [SPY/SPOT]; with it, correctly empty. covers() is the minimum honest scope, not the final answer. One vendor running several feeds (Databento equities + OPRA) wants a per-feed dataset/schema config; that is what the planned provider_feed_cfg table is for. Until then, a feed declaring its classes beats a class hardcoded in shared SQL. An instrument with no xref row for a source is simply absent from that feed's subscribe set — the "tradable but unmarkable" state, which is a fact about coverage, not an error. Verified: 72 tests; all three streams connect; no errors. --- crates/dataprovider/src/quote.rs | 10 ++++++++ src/opra_stream.rs | 6 +++++ src/quote_feed.rs | 44 ++++++++++++++++++++++---------- 3 files changed, 46 insertions(+), 14 deletions(-) diff --git a/crates/dataprovider/src/quote.rs b/crates/dataprovider/src/quote.rs index 5f0fa27..0e89b96 100644 --- a/crates/dataprovider/src/quote.rs +++ b/crates/dataprovider/src/quote.rs @@ -69,6 +69,16 @@ impl FeedHealth for NoFeedHealth { /// 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 diff --git a/src/opra_stream.rs b/src/opra_stream.rs index f0012ee..247da89 100644 --- a/src/opra_stream.rs +++ b/src/opra_stream.rs @@ -34,6 +34,12 @@ impl DataProvider for DatabentoOpraFeed { #[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"] + } + async fn run_session( &self, symbols: HashMap, diff --git a/src/quote_feed.rs b/src/quote_feed.rs index afd3759..ff29d89 100644 --- a/src/quote_feed.rs +++ b/src/quote_feed.rs @@ -51,7 +51,7 @@ impl QuoteFeedSession { /// 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()).await?; + let held = load_subscribable(&self.pool, self.feed.code(), self.feed.covers()).await?; if !held.is_empty() { return Ok(held); } @@ -75,7 +75,7 @@ impl Session for QuoteFeedSession { // 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(), &mut self.position_changed_rx, &add_tx, known) => r, + r = watch_held(&self.pool, self.feed.code(), self.feed.covers(), &mut self.position_changed_rx, &add_tx, known) => r, } } } @@ -87,6 +87,7 @@ impl Session for QuoteFeedSession { 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, @@ -94,7 +95,7 @@ async fn watch_held( loop { match doorbell.recv().await { Some(()) => { - let latest = load_subscribable(pool, source_code).await?; + let latest = load_subscribable(pool, source_code, covers).await?; let adds: SymbolAdds = latest .into_iter() .filter(|(sym, _)| !known.contains_key(sym)) @@ -122,30 +123,45 @@ async fn watch_held( } } -/// The held instruments this feed can cover. +/// The held instruments this feed can cover, as `external_symbol -> instrument_id`. /// -/// Still the original OPRA-shaped query: held options keyed by the *master* -/// symbol, which works only because Databento's raw symbol happens to be -/// byte-identical to ours. P3 replaces the body with the `instrument_xref` -/// PROVIDER join so the feed's own `external_symbol` drives the subscription and -/// `source_code` selects coverage — that is what makes this genuinely generic, and -/// what lets a feed cover something other than options. +/// 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, + 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 i.symbol, i.id \ + "SELECT DISTINCT x.external_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'", + 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::("symbol"), r.get::("id"))) + .map(|r| (r.get::("external_symbol"), r.get::("id"))) .collect()) } From aa1caad20403c5155008f5b6080a7edd99682443 Mon Sep 17 00:00:00 2001 From: Max Date: Fri, 17 Jul 2026 10:01:03 +0200 Subject: [PATCH 6/8] feat(marks): add the Binance spot feed; crypto is no longer unmarkable MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The three Binance crypto instruments were routable but had no PROVIDER xref row, so nothing in the identity graph claimed to price them: /positions reported a null mark for positions we can freely trade. This closes that gap and, in doing so, is the first real test of the LiveQuoteFeed abstraction — the whole point of the previous three commits was that a second vendor should cost one trait impl. It did. BinanceFeed is protocol only: connect the public bookTicker stream, decode, emit Quote. No reconnect loop, no health wiring, no SQL, no subscription bookkeeping — quote_feed, stream_supervisor and mark_router already owned those, unchanged. Marks come from PRODUCTION Binance even though orders route to testnet. Testnet runs its own synthetic book, so marking against it would value positions at prices nobody can trade; P&L against a testnet fill is meaningless either way, so the choice is between a useful number and a useless one. The stream is public — no key, no signing — and shares nothing with the authenticated WS-API the order stream uses. The doorbell had to fan out. An mpsc has exactly one consumer, so with two feeds the fill path's single signal now goes to a relay task that try_sends to each feed's own doorbell. Keeping one sender on the fill path means execution.rs never learns how many feeds exist. Verified end-to-end against the live market — the first time any quote has flowed through this chain: feed -> Quote -> mark_router -> MarkStore -> /positions SOL: mark 74.605 vs avg_cost 77.82, qty 0.1 -> unrealized -0.3215 BTC: mark 62772.90 vs avg_cost 64100.26 -> unrealized -3.9821 Hand-checked: (74.605 - 77.82) * 0.1 = -0.3215. Marks tick live (74.595 -> 74.615) with mark_ts advancing. 75 tests pass. --- db/scripts/seed_binance_crypto.sql | 21 ++++ src/binance_feed.rs | 187 +++++++++++++++++++++++++++++ src/main.rs | 42 ++++++- 3 files changed, 247 insertions(+), 3 deletions(-) create mode 100644 src/binance_feed.rs diff --git a/db/scripts/seed_binance_crypto.sql b/db/scripts/seed_binance_crypto.sql index e007cea..582efde 100644 --- a/db/scripts/seed_binance_crypto.sql +++ b/db/scripts/seed_binance_crypto.sql @@ -43,6 +43,27 @@ FROM (VALUES JOIN instrument i ON i.symbol = x.base 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.base AND i.venue = 'BINANCE' +ON CONFLICT DO NOTHING; + -- Routing target. Credentials resolved from env (BINANCE_PAPER_API_KEY/SECRET). INSERT INTO oms.broker_connection (code, broker_code, environment, status) VALUES ('binance-paper', 'BINANCE', 'PAPER', 'ACTIVE') diff --git a/src/binance_feed.rs b/src/binance_feed.rs new file mode 100644 index 0000000..4a38c6b --- /dev/null +++ b/src/binance_feed.rs @@ -0,0 +1,187 @@ +//! 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"; + +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 (mut ws, _) = connect_async(WS_URL) + .await + .map_err(|e| ProviderError::Request(e.to_string()))?; + info!(url = WS_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/main.rs b/src/main.rs index f1ce2cd..1b73fb5 100644 --- a/src/main.rs +++ b/src/main.rs @@ -56,6 +56,7 @@ mod opra_stream; mod stream_supervisor; mod quote_feed; mod mark_router; +mod binance_feed; #[derive(OpenApi)] #[openapi( @@ -363,10 +364,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")) { @@ -399,17 +406,46 @@ async fn serve() { tokio::spawn(mark_router::run(quote_rx, state.marks().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"); let session = quote_feed::QuoteFeedSession::new( opra_stream::DatabentoOpraFeed, state.pool().clone(), quote_tx.clone(), - position_changed_rx, + 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)); + } + + // 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 From 8cb36b6c2d5e1d13743da43d15e37f6d0346eb59 Mon Sep 17 00:00:00 2001 From: Max Date: Fri, 17 Jul 2026 13:27:40 +0200 Subject: [PATCH 7/8] feat(marks): add the Bybit feed, ranked failover, and pair-named crypto symbols MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three interlocking changes: Bybit gives crypto a second source, which is the first case the failover design has ever had to arbitrate, and both depend on crypto identity being right first. Symbol collision (0015_CRYPTO_SYMBOL_IS_THE_PAIR) The seed modelled crypto as symbol = base asset (SOL), currency = quote (USDT). That cannot represent SOL/USDT and SOL/BTC at once: both key to (SOL, BINANCE) under instrument_symbol_venue_key, so the model silently permitted exactly one quote currency per base per venue — not a property of the market. The symbol now names the pair (SOLUSDT), which is what every venue calls it and what our own xref already stored as external_symbol; instrument.symbol was the only place saying SOL. Matches Nautilus's raw_symbol convention and CCXT's rule that contracts which differ must have distinct symbols. currency stays USDT so the pair remains decomposable, and recon is untouched (it matches the base asset via external_native_id). Bybit feed orderbook.1 on the v5 public spot stream. Confirmed against the live endpoint before writing the decoder — Bybit and Binance quote SOL within 0.15% of each other, which is what makes one a usable stand-in for the other. Unlike Binance's bookTicker, orderbook.1 is stateful: a snapshot replaces the book, a delta carries only the side that moved, and size "0" deletes a level. The feed tracks the last known top per pair and emits only once both sides are known, so a delta can never publish a half-quote. Needs an application-level {"op":"ping"} every 20s — a protocol ping frame does not reset Bybit's timer. Ranked failover (0018_CREATE_PROVIDER_FEED_POLICY + mark_router) Rank lives in policy keyed by (source_code, instrument_class), not per instrument — that would be ~42k copies of the same integer for a decision that is a property of the source. Binance outranks Bybit for SPOT because we route crypto orders to Binance: its book is the one our fills print against, and Bybit prices the same pair on a different book. The router now owns arbitration: best-ranked live source holds the mark, a better source preempts immediately, a worse one may take over only once the incumbent has been quiet for STALE_AFTER_SECS. Staleness rather than connection state is the trigger because it catches the failure a socket cannot report — up, subscribed, and silently not delivering. Every ownership change logs; a silent failover is one nobody can diagnose later. Deliberately not A/B line arbitration, which is per-message sequence reconciliation between redundant lines of the same feed at microsecond granularity. This is cross-vendor, at seconds, between different books. Verified live, not just in tests: - both feeds up: Bybit connected first and took the mark, Binance preempted it 0.5s later ("promoting better-ranked source") - primary dead (BINANCE_FEED_WS_URL pointed at an unreachable host): BINANCE/SPOT down, BYBIT/SPOT live, SOL still marked at 74.825 with a fresh timestamp - primary recovered: promoted straight back to Binance, mark 74.835 84 tests, including 5 that pin the arbitration state machine. --- .../oms/0018_CREATE_PROVIDER_FEED_POLICY.sql | 37 +++ .../public/0015_CRYPTO_SYMBOL_IS_THE_PAIR.sql | 29 ++ db/scripts/seed_binance_crypto.sql | 21 +- src/binance_feed.rs | 13 +- src/bybit_feed.rs | 240 ++++++++++++++++ src/main.rs | 20 +- src/mark_router.rs | 270 ++++++++++++++++-- 7 files changed, 591 insertions(+), 39 deletions(-) create mode 100644 db/migrations/ods/oms/0018_CREATE_PROVIDER_FEED_POLICY.sql create mode 100644 db/migrations/ods/public/0015_CRYPTO_SYMBOL_IS_THE_PAIR.sql create mode 100644 src/bybit_feed.rs 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 582efde..f5f6fd3 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,7 @@ 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 @@ -61,7 +64,7 @@ FROM (VALUES ('ETH', 'ETHUSDT'), ('SOL', 'SOLUSDT') ) AS x(base, pair) -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; -- Routing target. Credentials resolved from env (BINANCE_PAPER_API_KEY/SECRET). diff --git a/src/binance_feed.rs b/src/binance_feed.rs index 4a38c6b..dd59773 100644 --- a/src/binance_feed.rs +++ b/src/binance_feed.rs @@ -24,6 +24,14 @@ use tracing::{info, warn}; 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 { @@ -48,10 +56,11 @@ impl LiveQuoteFeed for BinanceFeed { // Keys are the pair as Binance reports it in `s` (uppercase, e.g. SOLUSDT). let mut sym_to_id = symbols; - let (mut ws, _) = connect_async(WS_URL) + let url = ws_url(); + let (mut ws, _) = connect_async(&url) .await .map_err(|e| ProviderError::Request(e.to_string()))?; - info!(url = WS_URL, "Binance feed: connected"); + 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?; 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/main.rs b/src/main.rs index 1b73fb5..05b6e4f 100644 --- a/src/main.rs +++ b/src/main.rs @@ -57,6 +57,7 @@ mod stream_supervisor; mod quote_feed; mod mark_router; mod binance_feed; +mod bybit_feed; #[derive(OpenApi)] #[openapi( @@ -403,7 +404,7 @@ async fn serve() { // 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())); + 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); @@ -436,6 +437,23 @@ async fn serve() { 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 { diff --git a/src/mark_router.rs b/src/mark_router.rs index d929dd0..ce73812 100644 --- a/src/mark_router.rs +++ b/src/mark_router.rs @@ -1,40 +1,72 @@ -//! Routes normalized quotes from every feed into the [`MarkStore`]. +//! 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. Feeds emit [`Quote`]s onto a channel and never touch -//! the store, so adding a vendor cannot change how marks are decided, and the -//! policy for "which source wins" lives in exactly one place. +//! 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. //! -//! Today that policy is trivial — one provider covers any given instrument, so -//! every quote is accepted. The ranked primary + failover arbitration (prefer the -//! lowest-ranked healthy source; demote on staleness; log the switch) belongs here -//! when a second feed exists. Building it now would be machinery with nothing to -//! arbitrate: no instrument currently has more than one provider. +//! 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}; +use tracing::{debug, info, warn}; use crate::marks::MarkStore; -/// Can this book produce a meaningful mid? +/// How long a source may go quiet before a worse-ranked source may take over. /// -/// 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 +/// 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) { - info!("mark router: started"); +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!( @@ -47,6 +79,17 @@ pub async fn run(mut quotes: mpsc::Receiver, marks: MarkStore) { 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 { @@ -57,14 +100,189 @@ pub async fn run(mut quotes: mpsc::Receiver, marks: MarkStore) { 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() { - // A deep-OTM option near expiry: worth ~0.025, not "no data". - // Rejecting this was a regression that made such contracts unmarkable. assert!(is_usable(0.0, 0.05)); } @@ -75,7 +293,6 @@ mod tests { #[test] fn locked_book_is_usable() { - // bid == ask is unusual but not contradictory; the mid is well defined. assert!(is_usable(2.30, 2.30)); } @@ -87,7 +304,6 @@ mod tests { #[test] fn crossed_book_is_rejected() { - // bid > ask: the sides came from inconsistent states. assert!(!is_usable(2.50, 2.40)); } From 73264e02dff2934870d7084c6e8d10d079a03569 Mon Sep 17 00:00:00 2001 From: Max Date: Sat, 18 Jul 2026 09:46:12 +0200 Subject: [PATCH 8/8] chore(seed): persist the Bybit PROVIDER xref rows MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The Bybit market-data xref rows were created by a manual INSERT during failover testing and never landed in a seed file, so a fresh install had a BINANCE provider but no BYBIT one — failover could not be reproduced. Add them next to the Binance PROVIDER rows, keyed the same way (same pair string, external_exchange = BYBIT). --- db/scripts/seed_binance_crypto.sql | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/db/scripts/seed_binance_crypto.sql b/db/scripts/seed_binance_crypto.sql index f5f6fd3..6fb7d47 100644 --- a/db/scripts/seed_binance_crypto.sql +++ b/db/scripts/seed_binance_crypto.sql @@ -67,6 +67,19 @@ FROM (VALUES 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). INSERT INTO oms.broker_connection (code, broker_code, environment, status) VALUES ('binance-paper', 'BINANCE', 'PAPER', 'ACTIVE')