1use std::{
19 sync::{
20 Arc, RwLock,
21 atomic::{AtomicBool, Ordering},
22 },
23 time::Duration,
24};
25
26use ahash::AHashMap;
27use anyhow::Context;
28use futures_util::{StreamExt, pin_mut};
29use nautilus_common::{
30 clients::DataClient,
31 live::{runner::get_data_event_sender, runtime::get_runtime},
32 messages::{
33 DataEvent,
34 data::{
35 BarsResponse, DataResponse, InstrumentResponse, InstrumentsResponse, RequestBars,
36 RequestInstrument, RequestInstruments, RequestTrades, SubscribeBars,
37 SubscribeBookDeltas, SubscribeInstrument, SubscribeInstruments, SubscribeQuotes,
38 SubscribeTrades, TradesResponse, UnsubscribeBars, UnsubscribeBookDeltas,
39 UnsubscribeQuotes, UnsubscribeTrades, subscribe::SubscribeInstrumentStatus,
40 unsubscribe::UnsubscribeInstrumentStatus,
41 },
42 },
43};
44use nautilus_core::{
45 AtomicMap, MUTEX_POISONED,
46 datetime::datetime_to_unix_nanos,
47 nanos::UnixNanos,
48 time::{AtomicTime, get_atomic_clock_realtime},
49};
50use nautilus_model::{
51 data::{BookOrder, Data, OrderBookDelta, OrderBookDeltas, OrderBookDeltas_API},
52 enums::{BookAction, BookType, MarketStatusAction, OrderSide, RecordFlag},
53 identifiers::{ClientId, InstrumentId, Symbol, Venue},
54 instruments::{Instrument, InstrumentAny},
55 types::{Price, Quantity},
56};
57use tokio::task::JoinHandle;
58use tokio_util::sync::CancellationToken;
59use ustr::Ustr;
60
61use crate::{
62 common::{
63 consts::BINANCE_VENUE,
64 credential::resolve_credentials,
65 enums::{BinanceEnvironment, BinanceProductType},
66 parse::bar_spec_to_binance_interval,
67 status::diff_and_emit_statuses,
68 urls::get_ws_base_url,
69 },
70 config::{BinanceDataClientConfig, BinanceSpotMarketDataMode},
71 spot::{
72 http::{BinanceDepth, DepthParams, client::BinanceSpotHttpClient},
73 sbe::generated::symbol_status::SymbolStatus,
74 websocket::{
75 public_json::{
76 BinanceSpotPublicJsonWebSocketClient,
77 messages::BinanceSpotPublicWsMessage,
78 parse::{
79 parse_book_ticker as parse_json_book_ticker,
80 parse_depth_diff as parse_json_depth_diff,
81 parse_depth_snapshot as parse_json_depth_snapshot,
82 parse_kline as parse_json_kline, parse_trade as parse_json_trade,
83 },
84 },
85 streams::{
86 client::BinanceSpotWebSocketClient,
87 messages::BinanceSpotWsMessage,
88 parse::{
89 parse_bbo_event, parse_depth_diff, parse_depth_snapshot, parse_trades_event,
90 },
91 },
92 },
93 },
94};
95
96const MAX_SNAPSHOT_RETRIES: u32 = 5;
97const MAX_BUFFERED_DEPTH_UPDATES: usize = 10_000;
98const SNAPSHOT_RETRY_BACKOFF_BASE_MS: u64 = 250;
99const SNAPSHOT_RETRY_BACKOFF_CAP_MS: u64 = 3_000;
100
101#[derive(Debug, Clone)]
102struct BufferedDepthUpdate {
103 deltas: OrderBookDeltas,
104 first_update_id: u64,
105 final_update_id: u64,
106}
107
108#[derive(Debug, Clone, Copy, PartialEq, Eq)]
109enum BookSyncStatus {
110 Buffering,
111 Failed,
112}
113
114#[derive(Debug, Clone)]
115struct BookBuffer {
116 updates: Vec<BufferedDepthUpdate>,
117 epoch: u64,
118 status: BookSyncStatus,
119}
120
121impl BookBuffer {
122 fn new(epoch: u64) -> Self {
123 Self {
124 updates: Vec::new(),
125 epoch,
126 status: BookSyncStatus::Buffering,
127 }
128 }
129}
130
131#[derive(Debug, Clone)]
132enum SpotWsClient {
133 Sbe(BinanceSpotWebSocketClient),
134 JsonPublic(BinanceSpotPublicJsonWebSocketClient),
135}
136
137impl SpotWsClient {
138 fn has_credentials(&self) -> bool {
139 match self {
140 Self::Sbe(client) => client.has_credentials(),
141 Self::JsonPublic(_) => true, }
143 }
144
145 fn cache_instruments(&self, instruments: &[InstrumentAny]) {
146 match self {
147 Self::Sbe(client) => client.cache_instruments(instruments),
148 Self::JsonPublic(client) => client.cache_instruments(instruments),
149 }
150 }
151
152 async fn subscribe(&self, streams: Vec<String>) -> anyhow::Result<()> {
153 match self {
154 Self::Sbe(client) => client.subscribe(streams).await.map_err(Into::into),
155 Self::JsonPublic(client) => client.subscribe(streams).await,
156 }
157 }
158
159 async fn unsubscribe(&self, streams: Vec<String>) -> anyhow::Result<()> {
160 match self {
161 Self::Sbe(client) => client.unsubscribe(streams).await.map_err(Into::into),
162 Self::JsonPublic(client) => client.unsubscribe(streams).await,
163 }
164 }
165
166 async fn close(&mut self) -> anyhow::Result<()> {
167 match self {
168 Self::Sbe(client) => client.close().await.map_err(Into::into),
169 Self::JsonPublic(client) => client.close().await,
170 }
171 }
172}
173
174fn looks_like_spot_sbe_ws_url(base_url: &str) -> bool {
175 let without_scheme = base_url
176 .split_once("://")
177 .map_or(base_url, |(_, rest)| rest);
178 let host = without_scheme
179 .split(['/', ':'])
180 .next()
181 .unwrap_or(without_scheme);
182 host.starts_with("stream-sbe") || host.starts_with("demo-stream-sbe")
183}
184
185fn resolve_spot_json_ws_url(
186 base_url_ws: Option<String>,
187 environment: BinanceEnvironment,
188) -> String {
189 let default_url = get_ws_base_url(BinanceProductType::Spot, environment).to_string();
190
191 match base_url_ws {
192 Some(url) if looks_like_spot_sbe_ws_url(&url) => {
193 log::warn!(
194 "Spot JSON market-data mode received an SBE WebSocket URL override (`{url}`); \
195 using Spot JSON WebSocket default for {environment:?}: {default_url}",
196 );
197 default_url
198 }
199 Some(url) => url,
200 None => default_url,
201 }
202}
203
204#[derive(Debug)]
206pub struct BinanceSpotDataClient {
207 clock: &'static AtomicTime,
208 client_id: ClientId,
209 config: BinanceDataClientConfig,
210 http_client: BinanceSpotHttpClient,
211 ws_client: SpotWsClient,
212 spot_market_data_mode: BinanceSpotMarketDataMode,
213 is_connected: AtomicBool,
214 cancellation_token: CancellationToken,
215 tasks: Vec<JoinHandle<()>>,
216 data_sender: tokio::sync::mpsc::UnboundedSender<DataEvent>,
217 instruments: Arc<AtomicMap<InstrumentId, InstrumentAny>>,
218 status_cache: Arc<AtomicMap<InstrumentId, MarketStatusAction>>,
219 book_buffers: Arc<AtomicMap<InstrumentId, BookBuffer>>,
220 book_subscriptions: Arc<AtomicMap<InstrumentId, u32>>,
221 book_epoch: Arc<RwLock<u64>>,
222}
223
224impl BinanceSpotDataClient {
225 pub fn new(client_id: ClientId, config: BinanceDataClientConfig) -> anyhow::Result<Self> {
231 let clock = get_atomic_clock_realtime();
232 let spot_market_data_mode = config.spot_market_data_mode;
233
234 let http_client = BinanceSpotHttpClient::new(
235 config.environment,
236 clock,
237 config.api_key.clone(),
238 config.api_secret.clone(),
239 config.base_url_http.clone(),
240 None, None, None, )?;
244
245 let creds = if spot_market_data_mode == BinanceSpotMarketDataMode::Sbe {
246 resolve_credentials(
247 config.api_key.clone(),
248 config.api_secret.clone(),
249 config.environment,
250 config.product_type,
251 )
252 .inspect_err(|e| {
253 log::warn!(
254 "Failed to resolve Binance API credentials ({e}). \
255 Spot SBE WebSocket streams require an Ed25519 API key. \
256 Set the appropriate env vars for your environment, \
257 or provide api_key/api_secret in the data client config"
258 );
259 })
260 .ok()
261 } else {
262 None
263 };
264
265 let ws_client = match spot_market_data_mode {
266 BinanceSpotMarketDataMode::Sbe => SpotWsClient::Sbe(BinanceSpotWebSocketClient::new(
268 config.base_url_ws.clone(),
269 creds.as_ref().map(|(k, _)| k.clone()),
270 creds.as_ref().map(|(_, s)| s.clone()),
271 Some(20), config.transport_backend,
273 )?),
274 BinanceSpotMarketDataMode::Json => {
275 SpotWsClient::JsonPublic(BinanceSpotPublicJsonWebSocketClient::new(
276 Some(resolve_spot_json_ws_url(
277 config.base_url_ws.clone(),
278 config.environment,
279 )),
280 Some(20), config.transport_backend,
282 ))
283 }
284 };
285 let data_sender = get_data_event_sender();
286
287 log::debug!("Configured Spot market data mode: {spot_market_data_mode:?}");
288
289 Ok(Self {
290 clock,
291 client_id,
292 config,
293 http_client,
294 ws_client,
295 spot_market_data_mode,
296 is_connected: AtomicBool::new(false),
297 cancellation_token: CancellationToken::new(),
298 tasks: Vec::new(),
299 data_sender,
300 instruments: Arc::new(AtomicMap::new()),
301 status_cache: Arc::new(AtomicMap::new()),
302 book_buffers: Arc::new(AtomicMap::new()),
303 book_subscriptions: Arc::new(AtomicMap::new()),
304 book_epoch: Arc::new(RwLock::new(0)),
305 })
306 }
307
308 fn venue(&self) -> Venue {
309 *BINANCE_VENUE
310 }
311
312 fn send_data(sender: &tokio::sync::mpsc::UnboundedSender<DataEvent>, data: Data) {
313 if let Err(e) = sender.send(DataEvent::Data(data)) {
314 log::error!("Failed to emit data event: {e}");
315 }
316 }
317
318 fn spawn_ws<F>(&self, fut: F, context: &'static str)
319 where
320 F: std::future::Future<Output = anyhow::Result<()>> + Send + 'static,
321 {
322 get_runtime().spawn(async move {
323 if let Err(e) = fut.await {
324 log::error!("{context}: {e:?}");
325 }
326 });
327 }
328
329 #[expect(clippy::too_many_arguments)]
330 fn handle_ws_message(
331 msg: BinanceSpotWsMessage,
332 data_sender: &tokio::sync::mpsc::UnboundedSender<DataEvent>,
333 instruments: &Arc<AtomicMap<InstrumentId, InstrumentAny>>,
334 ws_instruments: &Arc<AtomicMap<Ustr, InstrumentAny>>,
335 book_buffers: &Arc<AtomicMap<InstrumentId, BookBuffer>>,
336 book_subscriptions: &Arc<AtomicMap<InstrumentId, u32>>,
337 book_epoch: &Arc<RwLock<u64>>,
338 http_client: &BinanceSpotHttpClient,
339 clock: &'static AtomicTime,
340 ) {
341 match msg {
342 BinanceSpotWsMessage::Trades(ref event) => {
343 let symbol = Ustr::from(&event.symbol);
344 let cache = ws_instruments.load();
345 if let Some(instrument) = cache.get(&symbol) {
346 let trades = parse_trades_event(event, instrument);
347 for data in trades {
348 Self::send_data(data_sender, data);
349 }
350 }
351 }
352 BinanceSpotWsMessage::BestBidAsk(ref event) => {
353 let symbol = Ustr::from(&event.symbol);
354 let cache = ws_instruments.load();
355 if let Some(instrument) = cache.get(&symbol) {
356 let quote = parse_bbo_event(event, instrument);
357 Self::send_data(data_sender, Data::from(quote));
358 }
359 }
360 BinanceSpotWsMessage::DepthSnapshot(ref event) => {
361 let symbol = Ustr::from(&event.symbol);
362 let cache = ws_instruments.load();
363 if let Some(instrument) = cache.get(&symbol)
364 && let Some(deltas) = parse_depth_snapshot(event, instrument)
365 {
366 Self::send_data(data_sender, Data::Deltas(OrderBookDeltas_API::new(deltas)));
367 }
368 }
369 BinanceSpotWsMessage::DepthDiff(ref event) => {
370 let symbol = Ustr::from(&event.symbol);
371 let cache = ws_instruments.load();
372 if let Some(instrument) = cache.get(&symbol)
373 && let Some(deltas) = parse_depth_diff(event, instrument)
374 {
375 let first_update_id = event.first_book_update_id as u64;
376 let final_update_id = event.last_book_update_id as u64;
377
378 Self::route_depth_diff(
379 data_sender,
380 book_buffers,
381 deltas,
382 first_update_id,
383 final_update_id,
384 );
385 }
386 }
387 BinanceSpotWsMessage::ServerShutdown(ref msg) => {
388 log::warn!(
389 "Binance server shutdown notice (event_time={}); disconnect expected within ~10 minutes",
390 msg.event_time,
391 );
392 }
393 BinanceSpotWsMessage::RawBinary(data) => {
394 log::debug!("Unhandled binary message: {} bytes", data.len());
395 }
396 BinanceSpotWsMessage::RawJson(value) => {
397 log::debug!("Unhandled JSON message: {value:?}");
398 }
399 BinanceSpotWsMessage::Error(e) => {
400 log::warn!("Binance WebSocket error: code={}, msg={}", e.code, e.msg);
401 }
402 BinanceSpotWsMessage::Reconnected => {
403 log::info!("WebSocket reconnected, rebuilding order book snapshots");
404
405 Self::rebuild_full_depth_books(
406 data_sender,
407 instruments,
408 book_buffers,
409 book_subscriptions,
410 book_epoch,
411 http_client,
412 clock,
413 );
414 }
415 }
416 }
417
418 #[expect(clippy::too_many_arguments)]
419 fn handle_public_json_ws_message(
420 msg: BinanceSpotPublicWsMessage,
421 data_sender: &tokio::sync::mpsc::UnboundedSender<DataEvent>,
422 instruments: &Arc<AtomicMap<InstrumentId, InstrumentAny>>,
423 ws_instruments: &Arc<AtomicMap<Ustr, InstrumentAny>>,
424 book_buffers: &Arc<AtomicMap<InstrumentId, BookBuffer>>,
425 book_subscriptions: &Arc<AtomicMap<InstrumentId, u32>>,
426 book_epoch: &Arc<RwLock<u64>>,
427 http_client: &BinanceSpotHttpClient,
428 clock: &'static AtomicTime,
429 ) {
430 let ts_init = clock.get_time_ns();
431
432 match msg {
433 BinanceSpotPublicWsMessage::Trade(ref event) => {
434 let symbol = event.symbol;
435 let cache = ws_instruments.load();
436 if let Some(instrument) = cache.get(&symbol) {
437 match parse_json_trade(event, instrument, ts_init) {
438 Ok(trade) => Self::send_data(data_sender, Data::Trade(trade)),
439 Err(e) => log::warn!("Failed to parse Spot JSON trade: {e}"),
440 }
441 }
442 }
443 BinanceSpotPublicWsMessage::BookTicker(ref event) => {
444 let symbol = event.symbol;
445 let cache = ws_instruments.load();
446 if let Some(instrument) = cache.get(&symbol) {
447 match parse_json_book_ticker(event, instrument, ts_init) {
448 Ok(quote) => Self::send_data(data_sender, Data::Quote(quote)),
449 Err(e) => log::warn!("Failed to parse Spot JSON book ticker: {e}"),
450 }
451 }
452 }
453 BinanceSpotPublicWsMessage::DepthSnapshot(ref event) => {
454 let symbol = event.symbol;
455 let cache = ws_instruments.load();
456 if let Some(instrument) = cache.get(&symbol)
457 && let Some(deltas) = parse_json_depth_snapshot(event, instrument, ts_init)
458 {
459 Self::send_data(data_sender, Data::Deltas(OrderBookDeltas_API::new(deltas)));
460 }
461 }
462 BinanceSpotPublicWsMessage::DepthDiff(ref event) => {
463 let symbol = event.symbol;
464 let cache = ws_instruments.load();
465 if let Some(instrument) = cache.get(&symbol) {
466 match parse_json_depth_diff(event, instrument, ts_init) {
467 Ok(Some(deltas)) => Self::route_depth_diff(
468 data_sender,
469 book_buffers,
470 deltas,
471 event.first_update_id,
472 event.final_update_id,
473 ),
474 Ok(None) => {}
475 Err(e) => log::warn!("Failed to parse Spot JSON depth update: {e}"),
476 }
477 }
478 }
479 BinanceSpotPublicWsMessage::Kline(ref event) => {
480 let symbol = event.symbol;
481 let cache = ws_instruments.load();
482 if let Some(instrument) = cache.get(&symbol) {
483 match parse_json_kline(event, instrument, ts_init) {
484 Ok(Some(bar)) => Self::send_data(data_sender, Data::Bar(bar)),
485 Ok(None) => {} Err(e) => log::warn!("Failed to parse Spot JSON kline: {e}"),
487 }
488 }
489 }
490 BinanceSpotPublicWsMessage::ServerShutdown(ref msg) => {
491 log::warn!(
492 "Binance Spot JSON server shutdown notice (event_time={}); disconnect expected within ~10 minutes",
493 msg.event_time,
494 );
495 }
496 BinanceSpotPublicWsMessage::RawJson(value) => {
497 log::debug!("Unhandled Spot JSON message: {value:?}");
498 }
499 BinanceSpotPublicWsMessage::Error(e) => {
500 log::warn!("Spot JSON WebSocket error: code={}, msg={}", e.code, e.msg);
501 }
502 BinanceSpotPublicWsMessage::Reconnected => {
503 log::info!("Spot JSON WebSocket reconnected, rebuilding order book snapshots");
504
505 Self::rebuild_full_depth_books(
506 data_sender,
507 instruments,
508 book_buffers,
509 book_subscriptions,
510 book_epoch,
511 http_client,
512 clock,
513 );
514 }
515 }
516 }
517
518 fn route_depth_diff(
519 data_sender: &tokio::sync::mpsc::UnboundedSender<DataEvent>,
520 book_buffers: &Arc<AtomicMap<InstrumentId, BookBuffer>>,
521 deltas: OrderBookDeltas,
522 first_update_id: u64,
523 final_update_id: u64,
524 ) {
525 let instrument_id = deltas.instrument_id;
526
527 if book_buffers.contains_key(&instrument_id) {
528 let mut handled_by_sync = false;
529 book_buffers.rcu(|m| {
530 handled_by_sync = false;
531
532 if let Some(buffer) = m.get_mut(&instrument_id) {
533 handled_by_sync = true;
534
535 if buffer.status == BookSyncStatus::Buffering {
536 buffer.updates.push(BufferedDepthUpdate {
537 deltas: deltas.clone(),
538 first_update_id,
539 final_update_id,
540 });
541 trim_buffered_depth_updates(&mut buffer.updates);
542 }
543 }
544 });
545
546 if handled_by_sync {
547 return;
548 }
549 }
550
551 Self::send_data(data_sender, Data::Deltas(OrderBookDeltas_API::new(deltas)));
552 }
553
554 fn rebuild_full_depth_books(
555 data_sender: &tokio::sync::mpsc::UnboundedSender<DataEvent>,
556 instruments: &Arc<AtomicMap<InstrumentId, InstrumentAny>>,
557 book_buffers: &Arc<AtomicMap<InstrumentId, BookBuffer>>,
558 book_subscriptions: &Arc<AtomicMap<InstrumentId, u32>>,
559 book_epoch: &Arc<RwLock<u64>>,
560 http_client: &BinanceSpotHttpClient,
561 clock: &'static AtomicTime,
562 ) {
563 let epoch = {
564 let mut guard = book_epoch.write().expect(MUTEX_POISONED);
565 *guard = guard.wrapping_add(1);
566 *guard
567 };
568
569 let subs: Vec<(InstrumentId, u32)> = {
570 let guard = book_subscriptions.load();
571 guard.iter().map(|(k, v)| (*k, *v)).collect()
572 };
573
574 for (instrument_id, depth) in subs {
575 if depth != 0 {
576 continue;
577 }
578
579 book_buffers.insert(instrument_id, BookBuffer::new(epoch));
580
581 log::debug!(
582 "OrderBook snapshot rebuild for {instrument_id} starting \
583 (reconnect, epoch={epoch})"
584 );
585
586 let http = http_client.clone();
587 let sender = data_sender.clone();
588 let buffers = book_buffers.clone();
589 let insts = instruments.clone();
590
591 get_runtime().spawn(async move {
592 Self::fetch_and_emit_snapshot(
593 http,
594 sender,
595 buffers,
596 insts,
597 instrument_id,
598 epoch,
599 clock,
600 )
601 .await;
602 });
603 }
604 }
605
606 fn quote_stream_suffix(&self) -> &'static str {
607 match self.spot_market_data_mode {
608 BinanceSpotMarketDataMode::Sbe => "bestBidAsk",
609 BinanceSpotMarketDataMode::Json => "bookTicker",
610 }
611 }
612
613 async fn fetch_and_emit_snapshot(
614 http: BinanceSpotHttpClient,
615 sender: tokio::sync::mpsc::UnboundedSender<DataEvent>,
616 buffers: Arc<AtomicMap<InstrumentId, BookBuffer>>,
617 instruments: Arc<AtomicMap<InstrumentId, InstrumentAny>>,
618 instrument_id: InstrumentId,
619 epoch: u64,
620 clock: &'static AtomicTime,
621 ) {
622 Self::fetch_and_emit_snapshot_inner(
623 http,
624 sender,
625 buffers,
626 instruments,
627 instrument_id,
628 epoch,
629 clock,
630 0,
631 )
632 .await;
633 }
634
635 #[expect(clippy::too_many_arguments)]
636 async fn fetch_and_emit_snapshot_inner(
637 http: BinanceSpotHttpClient,
638 sender: tokio::sync::mpsc::UnboundedSender<DataEvent>,
639 buffers: Arc<AtomicMap<InstrumentId, BookBuffer>>,
640 instruments: Arc<AtomicMap<InstrumentId, InstrumentAny>>,
641 instrument_id: InstrumentId,
642 epoch: u64,
643 clock: &'static AtomicTime,
644 retry_count: u32,
645 ) {
646 const SNAPSHOT_DEPTH: u32 = 5000;
647
648 if Self::wait_for_buffered_update(&buffers, instrument_id, epoch)
649 .await
650 .is_none()
651 {
652 return;
653 }
654
655 let params = DepthParams {
656 symbol: instrument_id.symbol.as_str().to_uppercase(),
657 limit: Some(SNAPSHOT_DEPTH),
658 };
659
660 match http.inner().depth(¶ms).await {
661 Ok(depth_snapshot) => {
662 let ts_init = clock.get_time_ns();
663 let last_update_id = depth_snapshot.last_update_id as u64;
664
665 {
666 let guard = buffers.load();
667 match guard.get(&instrument_id) {
668 None => {
669 log::debug!(
670 "OrderBook subscription for {instrument_id} was cancelled, \
671 discarding snapshot"
672 );
673 return;
674 }
675 Some(buffer) if buffer.epoch != epoch => {
676 log::debug!(
677 "OrderBook snapshot for {instrument_id} is stale \
678 (epoch {epoch} != {}), discarding",
679 buffer.epoch
680 );
681 return;
682 }
683 Some(buffer) if buffer.status == BookSyncStatus::Failed => {
684 log::debug!(
685 "OrderBook snapshot for {instrument_id} belongs to a failed \
686 sync, discarding"
687 );
688 return;
689 }
690 _ => {}
691 }
692 }
693
694 let (price_precision, size_precision) = {
695 let guard = instruments.load();
696 match guard.get(&instrument_id) {
697 Some(inst) => (inst.price_precision(), inst.size_precision()),
698 None => {
699 log::error!("No instrument in cache for snapshot: {instrument_id}");
700 Self::mark_book_sync_failed(&buffers, instrument_id, epoch);
701 return;
702 }
703 }
704 };
705
706 let Some(first) = Self::wait_for_first_applicable_update(
707 &buffers,
708 instrument_id,
709 epoch,
710 last_update_id,
711 )
712 .await
713 else {
714 return;
715 };
716
717 let target = last_update_id + 1;
718 if !spot_overlap_valid(first.first_update_id, first.final_update_id, last_update_id)
719 {
720 if retry_count < MAX_SNAPSHOT_RETRIES {
721 log::warn!(
722 "OrderBook overlap validation failed for {instrument_id}: \
723 lastUpdateId={last_update_id}, first_update_id={}, \
724 final_update_id={} (need U <= {} <= u), \
725 retrying snapshot (attempt {}/{})",
726 first.first_update_id,
727 first.final_update_id,
728 target,
729 retry_count + 1,
730 MAX_SNAPSHOT_RETRIES
731 );
732
733 tokio::time::sleep(spot_snapshot_retry_backoff(retry_count)).await;
734
735 Box::pin(Self::fetch_and_emit_snapshot_inner(
736 http,
737 sender,
738 buffers,
739 instruments,
740 instrument_id,
741 epoch,
742 clock,
743 retry_count + 1,
744 ))
745 .await;
746 return;
747 }
748
749 log::error!(
750 "OrderBook overlap validation failed for {instrument_id} after \
751 {MAX_SNAPSHOT_RETRIES} retries; no deltas will be emitted until \
752 resubscribe or reconnect"
753 );
754 Self::mark_book_sync_failed(&buffers, instrument_id, epoch);
755 return;
756 }
757
758 let Some(buffered) =
759 Self::take_buffered_depth_updates(&buffers, instrument_id, epoch)
760 else {
761 return;
762 };
763
764 let mut replayed = 0;
765 let mut last_final_update_id = last_update_id;
766 let mut is_first = true;
767 let mut replay_ready = Vec::with_capacity(buffered.len());
768
769 for update in buffered {
770 if update.final_update_id <= last_update_id {
771 continue;
772 }
773
774 if !spot_continuity_ok(is_first, update.first_update_id, last_final_update_id) {
775 if retry_count < MAX_SNAPSHOT_RETRIES {
776 log::warn!(
777 "OrderBook continuity break for {instrument_id}: \
778 expected U={}, was U={}, triggering resync (attempt {}/{})",
779 last_final_update_id + 1,
780 update.first_update_id,
781 retry_count + 1,
782 MAX_SNAPSHOT_RETRIES
783 );
784
785 Self::reset_book_sync_buffer(&buffers, instrument_id, epoch);
786 tokio::time::sleep(spot_snapshot_retry_backoff(retry_count)).await;
787
788 Box::pin(Self::fetch_and_emit_snapshot_inner(
789 http,
790 sender,
791 buffers,
792 instruments,
793 instrument_id,
794 epoch,
795 clock,
796 retry_count + 1,
797 ))
798 .await;
799 return;
800 }
801
802 log::error!(
803 "OrderBook continuity break for {instrument_id} after \
804 {MAX_SNAPSHOT_RETRIES} retries; no deltas will be emitted until \
805 resubscribe or reconnect"
806 );
807 Self::mark_book_sync_failed(&buffers, instrument_id, epoch);
808 return;
809 }
810
811 last_final_update_id = update.final_update_id;
812 is_first = false;
813 replayed += 1;
814 replay_ready.push(update);
815 }
816
817 let snapshot_ts_event = replay_ready
818 .first()
819 .map_or(ts_init, |update| update.deltas.ts_event);
820
821 let snapshot_deltas = match parse_spot_depth_snapshot(
822 &depth_snapshot,
823 instrument_id,
824 price_precision,
825 size_precision,
826 snapshot_ts_event,
827 ts_init,
828 ) {
829 Ok(Some(deltas)) => deltas,
830 Ok(None) => {
831 if retry_count < MAX_SNAPSHOT_RETRIES {
832 log::warn!(
833 "OrderBook snapshot for {instrument_id} contained no levels; \
834 retrying snapshot (attempt {}/{})",
835 retry_count + 1,
836 MAX_SNAPSHOT_RETRIES
837 );
838
839 tokio::time::sleep(spot_snapshot_retry_backoff(retry_count)).await;
840
841 Box::pin(Self::fetch_and_emit_snapshot_inner(
842 http,
843 sender,
844 buffers,
845 instruments,
846 instrument_id,
847 epoch,
848 clock,
849 retry_count + 1,
850 ))
851 .await;
852 return;
853 }
854
855 log::error!(
856 "OrderBook snapshot for {instrument_id} contained no levels after \
857 {MAX_SNAPSHOT_RETRIES} retries; no deltas will be emitted until \
858 resubscribe or reconnect"
859 );
860 Self::mark_book_sync_failed(&buffers, instrument_id, epoch);
861 return;
862 }
863 Err(e) => {
864 if retry_count < MAX_SNAPSHOT_RETRIES {
865 log::warn!(
866 "Failed to parse order book snapshot for {instrument_id}: {e}; \
867 retrying snapshot (attempt {}/{})",
868 retry_count + 1,
869 MAX_SNAPSHOT_RETRIES
870 );
871
872 tokio::time::sleep(spot_snapshot_retry_backoff(retry_count)).await;
873
874 Box::pin(Self::fetch_and_emit_snapshot_inner(
875 http,
876 sender,
877 buffers,
878 instruments,
879 instrument_id,
880 epoch,
881 clock,
882 retry_count + 1,
883 ))
884 .await;
885 return;
886 }
887
888 log::error!(
889 "Failed to parse order book snapshot for {instrument_id} after \
890 {MAX_SNAPSHOT_RETRIES} retries: {e}; no deltas will be emitted \
891 until resubscribe or reconnect"
892 );
893 Self::mark_book_sync_failed(&buffers, instrument_id, epoch);
894 return;
895 }
896 };
897
898 if let Err(e) = sender.send(DataEvent::Data(Data::Deltas(
899 OrderBookDeltas_API::new(snapshot_deltas),
900 ))) {
901 log::error!("Failed to send snapshot: {e}");
902 }
903
904 for update in replay_ready {
905 if let Err(e) = sender.send(DataEvent::Data(Data::Deltas(
906 OrderBookDeltas_API::new(update.deltas),
907 ))) {
908 log::error!("Failed to send replayed deltas: {e}");
909 }
910 }
911
912 while let Some(more) =
913 Self::drain_buffered_depth_updates(&buffers, instrument_id, epoch)
914 {
915 for update in more {
916 if update.final_update_id <= last_update_id {
917 continue;
918 }
919
920 if !spot_continuity_ok(
921 is_first,
922 update.first_update_id,
923 last_final_update_id,
924 ) {
925 if retry_count < MAX_SNAPSHOT_RETRIES {
926 log::warn!(
927 "OrderBook continuity break for {instrument_id}: \
928 expected U={}, was U={}, triggering resync (attempt {}/{})",
929 last_final_update_id + 1,
930 update.first_update_id,
931 retry_count + 1,
932 MAX_SNAPSHOT_RETRIES
933 );
934
935 Self::reset_book_sync_buffer(&buffers, instrument_id, epoch);
936 tokio::time::sleep(spot_snapshot_retry_backoff(retry_count)).await;
937
938 Box::pin(Self::fetch_and_emit_snapshot_inner(
939 http,
940 sender,
941 buffers,
942 instruments,
943 instrument_id,
944 epoch,
945 clock,
946 retry_count + 1,
947 ))
948 .await;
949 return;
950 }
951 log::error!(
952 "OrderBook continuity break for {instrument_id} after \
953 {MAX_SNAPSHOT_RETRIES} retries; no deltas will be emitted \
954 until resubscribe or reconnect"
955 );
956 Self::mark_book_sync_failed(&buffers, instrument_id, epoch);
957 return;
958 }
959
960 last_final_update_id = update.final_update_id;
961 is_first = false;
962 replayed += 1;
963
964 if let Err(e) = sender.send(DataEvent::Data(Data::Deltas(
965 OrderBookDeltas_API::new(update.deltas),
966 ))) {
967 log::error!("Failed to send replayed deltas: {e}");
968 }
969 }
970 }
971
972 log::debug!(
973 "OrderBook snapshot rebuild for {instrument_id} completed \
974 (lastUpdateId={last_update_id}, replayed={replayed})"
975 );
976 }
977 Err(e) => {
978 if retry_count < MAX_SNAPSHOT_RETRIES {
979 log::warn!(
980 "Failed to request order book snapshot for {instrument_id}: {e}; \
981 retrying snapshot (attempt {}/{})",
982 retry_count + 1,
983 MAX_SNAPSHOT_RETRIES
984 );
985
986 tokio::time::sleep(spot_snapshot_retry_backoff(retry_count)).await;
987
988 Box::pin(Self::fetch_and_emit_snapshot_inner(
989 http,
990 sender,
991 buffers,
992 instruments,
993 instrument_id,
994 epoch,
995 clock,
996 retry_count + 1,
997 ))
998 .await;
999 return;
1000 }
1001
1002 log::error!(
1003 "Failed to request order book snapshot for {instrument_id} after \
1004 {MAX_SNAPSHOT_RETRIES} retries: {e}; no deltas will be emitted until \
1005 resubscribe or reconnect"
1006 );
1007 Self::mark_book_sync_failed(&buffers, instrument_id, epoch);
1008 }
1009 }
1010 }
1011
1012 async fn wait_for_buffered_update(
1013 buffers: &Arc<AtomicMap<InstrumentId, BookBuffer>>,
1014 instrument_id: InstrumentId,
1015 epoch: u64,
1016 ) -> Option<()> {
1017 loop {
1018 let guard = buffers.load();
1019 match guard.get(&instrument_id) {
1020 Some(buffer)
1021 if buffer.epoch == epoch
1022 && buffer.status == BookSyncStatus::Buffering
1023 && !buffer.updates.is_empty() =>
1024 {
1025 return Some(());
1026 }
1027 Some(buffer)
1028 if buffer.epoch == epoch && buffer.status == BookSyncStatus::Buffering => {}
1029 _ => return None,
1030 }
1031
1032 drop(guard);
1033 tokio::time::sleep(Duration::from_millis(100)).await;
1034 }
1035 }
1036
1037 async fn wait_for_first_applicable_update(
1038 buffers: &Arc<AtomicMap<InstrumentId, BookBuffer>>,
1039 instrument_id: InstrumentId,
1040 epoch: u64,
1041 last_update_id: u64,
1042 ) -> Option<BufferedDepthUpdate> {
1043 loop {
1044 let mut first = None;
1045 let mut waiting = false;
1046 buffers.rcu(|m| {
1047 first = None;
1048 waiting = false;
1049
1050 if let Some(buffer) = m.get_mut(&instrument_id)
1051 && buffer.epoch == epoch
1052 && buffer.status == BookSyncStatus::Buffering
1053 {
1054 buffer
1055 .updates
1056 .retain(|update| update.final_update_id > last_update_id);
1057 first = first_applicable_spot_update(&buffer.updates, last_update_id).cloned();
1058 waiting = first.is_none();
1059 }
1060 });
1061
1062 if first.is_some() {
1063 return first;
1064 }
1065
1066 if !waiting {
1067 return None;
1068 }
1069
1070 tokio::time::sleep(Duration::from_millis(100)).await;
1071 }
1072 }
1073
1074 fn take_buffered_depth_updates(
1075 buffers: &Arc<AtomicMap<InstrumentId, BookBuffer>>,
1076 instrument_id: InstrumentId,
1077 epoch: u64,
1078 ) -> Option<Vec<BufferedDepthUpdate>> {
1079 let mut taken = None;
1080 buffers.rcu(|m| {
1081 taken = None;
1082
1083 if let Some(buffer) = m.get_mut(&instrument_id)
1084 && buffer.epoch == epoch
1085 && buffer.status == BookSyncStatus::Buffering
1086 {
1087 taken = Some(std::mem::take(&mut buffer.updates));
1088 }
1089 });
1090 taken
1091 }
1092
1093 fn drain_buffered_depth_updates(
1094 buffers: &Arc<AtomicMap<InstrumentId, BookBuffer>>,
1095 instrument_id: InstrumentId,
1096 epoch: u64,
1097 ) -> Option<Vec<BufferedDepthUpdate>> {
1098 let mut taken = None;
1099 buffers.rcu(|m| {
1100 taken = None;
1101
1102 if let Some(buffer) = m.get_mut(&instrument_id)
1103 && buffer.epoch == epoch
1104 && buffer.status == BookSyncStatus::Buffering
1105 {
1106 if buffer.updates.is_empty() {
1107 m.remove(&instrument_id);
1108 } else {
1109 taken = Some(std::mem::take(&mut buffer.updates));
1110 }
1111 }
1112 });
1113 taken
1114 }
1115
1116 fn reset_book_sync_buffer(
1117 buffers: &Arc<AtomicMap<InstrumentId, BookBuffer>>,
1118 instrument_id: InstrumentId,
1119 epoch: u64,
1120 ) {
1121 buffers.rcu(|m| {
1122 if let Some(buffer) = m.get_mut(&instrument_id)
1123 && buffer.epoch == epoch
1124 {
1125 buffer.updates.clear();
1126 buffer.status = BookSyncStatus::Buffering;
1127 }
1128 });
1129 }
1130
1131 fn mark_book_sync_failed(
1132 buffers: &Arc<AtomicMap<InstrumentId, BookBuffer>>,
1133 instrument_id: InstrumentId,
1134 epoch: u64,
1135 ) {
1136 buffers.rcu(|m| {
1137 if let Some(buffer) = m.get_mut(&instrument_id)
1138 && buffer.epoch == epoch
1139 {
1140 buffer.updates.clear();
1141 buffer.status = BookSyncStatus::Failed;
1142 }
1143 });
1144 }
1145}
1146
1147fn upsert_instrument(
1148 cache: &Arc<AtomicMap<InstrumentId, InstrumentAny>>,
1149 instrument: InstrumentAny,
1150) {
1151 cache.insert(instrument.id(), instrument);
1152}
1153
1154fn spot_overlap_valid(first_update_id: u64, final_update_id: u64, last_update_id: u64) -> bool {
1156 let target = last_update_id + 1;
1157 first_update_id <= target && final_update_id >= target
1158}
1159
1160fn spot_continuity_ok(is_first: bool, first_update_id: u64, prev_final_update_id: u64) -> bool {
1162 is_first || first_update_id == prev_final_update_id + 1
1163}
1164
1165fn spot_snapshot_retry_backoff(retry_count: u32) -> Duration {
1166 let multiplier = 1_u64 << retry_count.min(4);
1167 let millis = SNAPSHOT_RETRY_BACKOFF_BASE_MS
1168 .saturating_mul(multiplier)
1169 .min(SNAPSHOT_RETRY_BACKOFF_CAP_MS);
1170 Duration::from_millis(millis)
1171}
1172
1173fn first_applicable_spot_update(
1174 updates: &[BufferedDepthUpdate],
1175 last_update_id: u64,
1176) -> Option<&BufferedDepthUpdate> {
1177 updates
1178 .iter()
1179 .find(|update| update.final_update_id > last_update_id)
1180}
1181
1182fn trim_buffered_depth_updates(updates: &mut Vec<BufferedDepthUpdate>) {
1183 let excess = updates.len().saturating_sub(MAX_BUFFERED_DEPTH_UPDATES);
1184 if excess > 0 {
1185 updates.drain(..excess);
1186 }
1187}
1188
1189fn parse_spot_depth_snapshot(
1190 depth: &BinanceDepth,
1191 instrument_id: InstrumentId,
1192 price_precision: u8,
1193 size_precision: u8,
1194 ts_event: UnixNanos,
1195 ts_init: UnixNanos,
1196) -> anyhow::Result<Option<OrderBookDeltas>> {
1197 let sequence = depth.last_update_id as u64;
1198
1199 let total_levels = depth.bids.len() + depth.asks.len();
1200 let mut deltas = Vec::with_capacity(total_levels + 1);
1201
1202 deltas.push(OrderBookDelta::clear(
1204 instrument_id,
1205 sequence,
1206 ts_event,
1207 ts_init,
1208 ));
1209
1210 for (i, level) in depth.bids.iter().enumerate() {
1211 let price = Price::from_mantissa_exponent_checked(
1212 level.price_mantissa,
1213 depth.price_exponent,
1214 price_precision,
1215 )?;
1216 let size = Quantity::from_mantissa_exponent_checked(
1217 level.qty_mantissa as u64,
1218 depth.qty_exponent,
1219 size_precision,
1220 )?;
1221 let flags = if i == depth.bids.len() - 1 && depth.asks.is_empty() {
1222 RecordFlag::F_LAST as u8
1223 } else {
1224 0
1225 };
1226
1227 let order = BookOrder::new(OrderSide::Buy, price, size, 0);
1228
1229 deltas.push(OrderBookDelta::new(
1230 instrument_id,
1231 BookAction::Add,
1232 order,
1233 flags,
1234 sequence,
1235 ts_event,
1236 ts_init,
1237 ));
1238 }
1239
1240 for (i, level) in depth.asks.iter().enumerate() {
1241 let price = Price::from_mantissa_exponent_checked(
1242 level.price_mantissa,
1243 depth.price_exponent,
1244 price_precision,
1245 )?;
1246 let size = Quantity::from_mantissa_exponent_checked(
1247 level.qty_mantissa as u64,
1248 depth.qty_exponent,
1249 size_precision,
1250 )?;
1251 let flags = if i == depth.asks.len() - 1 {
1252 RecordFlag::F_LAST as u8
1253 } else {
1254 0
1255 };
1256
1257 let order = BookOrder::new(OrderSide::Sell, price, size, 0);
1258
1259 deltas.push(OrderBookDelta::new(
1260 instrument_id,
1261 BookAction::Add,
1262 order,
1263 flags,
1264 sequence,
1265 ts_event,
1266 ts_init,
1267 ));
1268 }
1269
1270 if deltas.len() <= 1 {
1271 return Ok(None);
1272 }
1273
1274 Ok(Some(OrderBookDeltas::new(instrument_id, deltas)))
1275}
1276
1277#[async_trait::async_trait(?Send)]
1278impl DataClient for BinanceSpotDataClient {
1279 fn client_id(&self) -> ClientId {
1280 self.client_id
1281 }
1282
1283 fn venue(&self) -> Option<Venue> {
1284 Some(self.venue())
1285 }
1286
1287 fn start(&mut self) -> anyhow::Result<()> {
1288 log::info!(
1289 "Started: client_id={}, product_type={:?}, environment={:?}",
1290 self.client_id,
1291 self.config.product_type,
1292 self.config.environment,
1293 );
1294 Ok(())
1295 }
1296
1297 fn stop(&mut self) -> anyhow::Result<()> {
1298 log::info!("Stopping {id}", id = self.client_id);
1299 self.cancellation_token.cancel();
1300 self.is_connected.store(false, Ordering::Relaxed);
1301 Ok(())
1302 }
1303
1304 fn reset(&mut self) -> anyhow::Result<()> {
1305 log::debug!("Resetting {id}", id = self.client_id);
1306
1307 self.cancellation_token.cancel();
1308
1309 for task in self.tasks.drain(..) {
1310 task.abort();
1311 }
1312
1313 let mut ws = self.ws_client.clone();
1314 get_runtime().spawn(async move {
1315 let _ = ws.close().await;
1316 });
1317
1318 self.book_subscriptions.store(AHashMap::new());
1319 self.book_buffers.store(AHashMap::new());
1320
1321 self.is_connected.store(false, Ordering::Relaxed);
1322 self.cancellation_token = CancellationToken::new();
1323 Ok(())
1324 }
1325
1326 fn dispose(&mut self) -> anyhow::Result<()> {
1327 log::debug!("Disposing {id}", id = self.client_id);
1328 self.stop()
1329 }
1330
1331 async fn connect(&mut self) -> anyhow::Result<()> {
1332 if self.is_connected() {
1333 return Ok(());
1334 }
1335
1336 if self.spot_market_data_mode == BinanceSpotMarketDataMode::Sbe
1337 && !self.ws_client.has_credentials()
1338 {
1339 anyhow::bail!(
1340 "Binance Spot market data mode SBE requires Ed25519 API credentials. \
1341 Set the appropriate env vars for your environment, \
1342 or provide api_key/api_secret in the data client config"
1343 );
1344 }
1345
1346 self.cancellation_token = CancellationToken::new();
1348
1349 let exchange_info = self
1351 .http_client
1352 .exchange_info()
1353 .await
1354 .map_err(|e| anyhow::anyhow!("failed to request Binance exchange info: {e}"))?;
1355
1356 let instruments = self
1357 .http_client
1358 .request_instruments()
1359 .await
1360 .context("failed to request Binance instruments")?;
1361
1362 self.http_client.cache_instruments(instruments.clone());
1363
1364 {
1365 let mut inst_map = AHashMap::new();
1366 let mut status_map = AHashMap::new();
1367
1368 for instrument in &instruments {
1369 inst_map.insert(instrument.id(), instrument.clone());
1370 }
1371
1372 for symbol_info in &exchange_info.symbols {
1374 let instrument_id =
1375 InstrumentId::new(Symbol::from(symbol_info.symbol.as_str()), *BINANCE_VENUE);
1376
1377 if inst_map.contains_key(&instrument_id) {
1378 let action = MarketStatusAction::from(SymbolStatus::from(symbol_info.status));
1379 status_map.insert(instrument_id, action);
1380 }
1381 }
1382
1383 self.instruments.store(inst_map);
1384 self.status_cache.store(status_map);
1385 }
1386
1387 for instrument in instruments.clone() {
1388 if let Err(e) = self.data_sender.send(DataEvent::Instrument(instrument)) {
1389 log::warn!("Failed to send instrument: {e}");
1390 }
1391 }
1392
1393 self.ws_client.cache_instruments(&instruments);
1394
1395 match &mut self.ws_client {
1396 SpotWsClient::Sbe(ws_client) => {
1397 log::info!("Connecting to Binance Spot SBE WebSocket...");
1398 ws_client.connect().await.map_err(|e| {
1399 log::error!("Binance Spot SBE WebSocket connection failed: {e:?}");
1400 anyhow::anyhow!("failed to connect Binance Spot SBE WebSocket: {e}")
1401 })?;
1402 log::info!("Binance Spot SBE WebSocket connected");
1403
1404 let stream = ws_client.stream();
1405 let sender = self.data_sender.clone();
1406 let insts = self.instruments.clone();
1407 let ws_insts = ws_client.instruments_cache();
1408 let buffers = self.book_buffers.clone();
1409 let book_subs = self.book_subscriptions.clone();
1410 let book_epoch = self.book_epoch.clone();
1411 let http = self.http_client.clone();
1412 let clock = self.clock;
1413 let cancel = self.cancellation_token.clone();
1414
1415 let handle = get_runtime().spawn(async move {
1416 pin_mut!(stream);
1417
1418 loop {
1419 tokio::select! {
1420 Some(message) = stream.next() => {
1421 Self::handle_ws_message(
1422 message,
1423 &sender,
1424 &insts,
1425 &ws_insts,
1426 &buffers,
1427 &book_subs,
1428 &book_epoch,
1429 &http,
1430 clock,
1431 );
1432 }
1433 () = cancel.cancelled() => {
1434 log::debug!("Spot SBE WebSocket stream task cancelled");
1435 break;
1436 }
1437 }
1438 }
1439 });
1440 self.tasks.push(handle);
1441 }
1442 SpotWsClient::JsonPublic(ws_client) => {
1443 log::info!("Connecting to Binance Spot public JSON WebSocket...");
1444 ws_client.connect().await.map_err(|e| {
1445 log::error!("Binance Spot public JSON WebSocket connection failed: {e:?}");
1446 anyhow::anyhow!("failed to connect Binance Spot public JSON WebSocket: {e}")
1447 })?;
1448 log::info!("Binance Spot public JSON WebSocket connected");
1449
1450 let stream = ws_client.stream();
1451 let sender = self.data_sender.clone();
1452 let insts = self.instruments.clone();
1453 let ws_insts = ws_client.instruments_cache();
1454 let buffers = self.book_buffers.clone();
1455 let book_subs = self.book_subscriptions.clone();
1456 let book_epoch = self.book_epoch.clone();
1457 let http = self.http_client.clone();
1458 let clock = self.clock;
1459 let cancel = self.cancellation_token.clone();
1460
1461 let handle = get_runtime().spawn(async move {
1462 pin_mut!(stream);
1463
1464 loop {
1465 tokio::select! {
1466 Some(message) = stream.next() => {
1467 Self::handle_public_json_ws_message(
1468 message,
1469 &sender,
1470 &insts,
1471 &ws_insts,
1472 &buffers,
1473 &book_subs,
1474 &book_epoch,
1475 &http,
1476 clock,
1477 );
1478 }
1479 () = cancel.cancelled() => {
1480 log::debug!("Spot JSON WebSocket stream task cancelled");
1481 break;
1482 }
1483 }
1484 }
1485 });
1486 self.tasks.push(handle);
1487 }
1488 }
1489
1490 let poll_secs = self.config.instrument_status_poll_secs;
1492 if poll_secs > 0 {
1493 let http = self.http_client.clone();
1494 let poll_sender = self.data_sender.clone();
1495 let poll_instruments = self.instruments.clone();
1496 let poll_status_cache = self.status_cache.clone();
1497 let poll_cancel = self.cancellation_token.clone();
1498 let clock = self.clock;
1499
1500 let poll_handle = get_runtime().spawn(async move {
1501 let mut interval =
1502 tokio::time::interval(tokio::time::Duration::from_secs(poll_secs));
1503 interval.tick().await; loop {
1506 tokio::select! {
1507 _ = interval.tick() => {
1508 match http.exchange_info().await {
1509 Ok(info) => {
1510 let ts = clock.get_time_ns();
1511 let inst_guard = poll_instruments.load();
1512
1513 let mut new_statuses = AHashMap::new();
1514 for symbol_info in &info.symbols {
1515 let instrument_id = InstrumentId::new(
1516 Symbol::from(
1517 symbol_info.symbol.as_str(),
1518 ),
1519 *BINANCE_VENUE,
1520 );
1521
1522 if inst_guard.contains_key(&instrument_id) {
1523 let action = MarketStatusAction::from(
1524 SymbolStatus::from(symbol_info.status),
1525 );
1526 new_statuses.insert(instrument_id, action);
1527 }
1528 }
1529 drop(inst_guard);
1530
1531 let mut cache =
1532 (**poll_status_cache.load()).clone();
1533 diff_and_emit_statuses(
1534 &new_statuses, &mut cache, &poll_sender, ts, ts,
1535 );
1536 poll_status_cache.store(cache);
1537 }
1538 Err(e) => {
1539 log::warn!("Instrument status poll failed: {e}");
1540 }
1541 }
1542 }
1543 () = poll_cancel.cancelled() => {
1544 log::debug!("Instrument status polling task cancelled");
1545 break;
1546 }
1547 }
1548 }
1549 });
1550 self.tasks.push(poll_handle);
1551 log::debug!("Instrument status polling started: interval={poll_secs}s");
1552 }
1553
1554 self.is_connected.store(true, Ordering::Release);
1555 log::info!("Connected: client_id={}", self.client_id);
1556 Ok(())
1557 }
1558
1559 async fn disconnect(&mut self) -> anyhow::Result<()> {
1560 if self.is_disconnected() {
1561 return Ok(());
1562 }
1563
1564 self.cancellation_token.cancel();
1565
1566 let _ = self.ws_client.close().await;
1567
1568 let handles: Vec<_> = self.tasks.drain(..).collect();
1569 for handle in handles {
1570 if let Err(e) = handle.await {
1571 log::error!("Error joining WebSocket task: {e}");
1572 }
1573 }
1574
1575 self.book_subscriptions.store(AHashMap::new());
1576 self.book_buffers.store(AHashMap::new());
1577
1578 self.is_connected.store(false, Ordering::Release);
1579 log::info!("Disconnected: client_id={}", self.client_id);
1580 Ok(())
1581 }
1582
1583 fn is_connected(&self) -> bool {
1584 self.is_connected.load(Ordering::Relaxed)
1585 }
1586
1587 fn is_disconnected(&self) -> bool {
1588 !self.is_connected()
1589 }
1590
1591 fn subscribe_instruments(&mut self, _cmd: SubscribeInstruments) -> anyhow::Result<()> {
1592 log::debug!("subscribe_instruments: Binance instruments are fetched via HTTP on connect");
1593 Ok(())
1594 }
1595
1596 fn subscribe_instrument(&mut self, _cmd: SubscribeInstrument) -> anyhow::Result<()> {
1597 log::debug!("subscribe_instrument: Binance instruments are fetched via HTTP on connect");
1598 Ok(())
1599 }
1600
1601 fn subscribe_book_deltas(&mut self, cmd: SubscribeBookDeltas) -> anyhow::Result<()> {
1602 if cmd.book_type != BookType::L2_MBP {
1603 anyhow::bail!("Binance SBE only supports L2_MBP order book deltas");
1604 }
1605
1606 let instrument_id = cmd.instrument_id;
1607 let ws = self.ws_client.clone();
1608 let symbol_lower = instrument_id.symbol.as_str().to_lowercase();
1609
1610 if self.spot_market_data_mode == BinanceSpotMarketDataMode::Json && cmd.depth.is_some() {
1611 let depth_level = match cmd.depth.map(|d| d.get()) {
1614 Some(1..=5) => 5,
1615 Some(6..=10) => 10,
1616 _ => 20,
1617 };
1618 self.book_subscriptions.insert(instrument_id, depth_level);
1619
1620 let stream = format!("{symbol_lower}@depth{depth_level}");
1621 self.spawn_ws(
1622 async move {
1623 ws.subscribe(vec![stream])
1624 .await
1625 .context("book deltas subscription")
1626 },
1627 "order book subscription",
1628 );
1629 return Ok(());
1630 }
1631
1632 match cmd.depth.map(|d| d.get()) {
1633 Some(depth) => {
1635 let depth_level = match depth {
1636 1..=5 => 5,
1637 6..=10 => 10,
1638 _ => 20,
1639 };
1640 self.book_subscriptions.insert(instrument_id, depth_level);
1641
1642 let stream = format!("{symbol_lower}@depth{depth_level}");
1643 self.spawn_ws(
1644 async move {
1645 ws.subscribe(vec![stream])
1646 .await
1647 .context("book deltas subscription")
1648 },
1649 "order book subscription",
1650 );
1651 }
1652 None => {
1654 self.book_subscriptions.insert(instrument_id, 0);
1655
1656 let epoch = {
1658 let mut guard = self.book_epoch.write().expect(MUTEX_POISONED);
1659 *guard = guard.wrapping_add(1);
1660 *guard
1661 };
1662
1663 self.book_buffers
1665 .insert(instrument_id, BookBuffer::new(epoch));
1666
1667 log::debug!("OrderBook full snapshot rebuild for {instrument_id} starting");
1668
1669 let stream = format!("{symbol_lower}@depth");
1670 self.spawn_ws(
1671 async move {
1672 ws.subscribe(vec![stream])
1673 .await
1674 .context("book deltas subscription")
1675 },
1676 "order book subscription",
1677 );
1678
1679 let http = self.http_client.clone();
1680 let sender = self.data_sender.clone();
1681 let buffers = self.book_buffers.clone();
1682 let instruments = self.instruments.clone();
1683 let clock = self.clock;
1684
1685 get_runtime().spawn(async move {
1686 Self::fetch_and_emit_snapshot(
1687 http,
1688 sender,
1689 buffers,
1690 instruments,
1691 instrument_id,
1692 epoch,
1693 clock,
1694 )
1695 .await;
1696 });
1697 }
1698 }
1699 Ok(())
1700 }
1701
1702 fn subscribe_quotes(&mut self, cmd: SubscribeQuotes) -> anyhow::Result<()> {
1703 let instrument_id = cmd.instrument_id;
1704 let ws = self.ws_client.clone();
1705 let suffix = self.quote_stream_suffix();
1706
1707 let stream = format!("{}@{suffix}", instrument_id.symbol.as_str().to_lowercase());
1708
1709 self.spawn_ws(
1710 async move {
1711 ws.subscribe(vec![stream])
1712 .await
1713 .context("quotes subscription")
1714 },
1715 "quote subscription",
1716 );
1717 Ok(())
1718 }
1719
1720 fn subscribe_trades(&mut self, cmd: SubscribeTrades) -> anyhow::Result<()> {
1721 let instrument_id = cmd.instrument_id;
1722 let ws = self.ws_client.clone();
1723
1724 let stream = format!("{}@trade", instrument_id.symbol.as_str().to_lowercase());
1725
1726 self.spawn_ws(
1727 async move {
1728 ws.subscribe(vec![stream])
1729 .await
1730 .context("trades subscription")
1731 },
1732 "trade subscription",
1733 );
1734 Ok(())
1735 }
1736
1737 fn subscribe_bars(&mut self, cmd: SubscribeBars) -> anyhow::Result<()> {
1738 let bar_type = cmd.bar_type;
1739 let ws = self.ws_client.clone();
1740 let interval = bar_spec_to_binance_interval(bar_type.spec())?;
1741
1742 let stream = format!(
1743 "{}@kline_{}",
1744 bar_type.instrument_id().symbol.as_str().to_lowercase(),
1745 interval.as_str()
1746 );
1747
1748 self.spawn_ws(
1749 async move {
1750 ws.subscribe(vec![stream])
1751 .await
1752 .context("bars subscription")
1753 },
1754 "bar subscription",
1755 );
1756 Ok(())
1757 }
1758
1759 fn subscribe_instrument_status(
1760 &mut self,
1761 cmd: SubscribeInstrumentStatus,
1762 ) -> anyhow::Result<()> {
1763 log::debug!(
1764 "subscribe_instrument_status: {id} (status changes detected via periodic exchange info polling)",
1765 id = cmd.instrument_id,
1766 );
1767 Ok(())
1768 }
1769
1770 fn unsubscribe_book_deltas(&mut self, cmd: &UnsubscribeBookDeltas) -> anyhow::Result<()> {
1771 let instrument_id = cmd.instrument_id;
1772 let ws = self.ws_client.clone();
1773
1774 self.book_subscriptions.remove(&instrument_id);
1776 self.book_buffers.remove(&instrument_id);
1777
1778 let symbol_lower = instrument_id.symbol.as_str().to_lowercase();
1779 let streams = vec![
1780 format!("{symbol_lower}@depth"),
1781 format!("{symbol_lower}@depth5"),
1782 format!("{symbol_lower}@depth10"),
1783 format!("{symbol_lower}@depth20"),
1784 ];
1785
1786 self.spawn_ws(
1787 async move {
1788 ws.unsubscribe(streams)
1789 .await
1790 .context("book deltas unsubscribe")
1791 },
1792 "order book unsubscribe",
1793 );
1794 Ok(())
1795 }
1796
1797 fn unsubscribe_quotes(&mut self, cmd: &UnsubscribeQuotes) -> anyhow::Result<()> {
1798 let instrument_id = cmd.instrument_id;
1799 let ws = self.ws_client.clone();
1800 let suffix = self.quote_stream_suffix();
1801
1802 let stream = format!("{}@{suffix}", instrument_id.symbol.as_str().to_lowercase());
1803
1804 self.spawn_ws(
1805 async move {
1806 ws.unsubscribe(vec![stream])
1807 .await
1808 .context("quotes unsubscribe")
1809 },
1810 "quote unsubscribe",
1811 );
1812 Ok(())
1813 }
1814
1815 fn unsubscribe_trades(&mut self, cmd: &UnsubscribeTrades) -> anyhow::Result<()> {
1816 let instrument_id = cmd.instrument_id;
1817 let ws = self.ws_client.clone();
1818
1819 let stream = format!("{}@trade", instrument_id.symbol.as_str().to_lowercase());
1820
1821 self.spawn_ws(
1822 async move {
1823 ws.unsubscribe(vec![stream])
1824 .await
1825 .context("trades unsubscribe")
1826 },
1827 "trade unsubscribe",
1828 );
1829 Ok(())
1830 }
1831
1832 fn unsubscribe_bars(&mut self, cmd: &UnsubscribeBars) -> anyhow::Result<()> {
1833 let bar_type = cmd.bar_type;
1834 let ws = self.ws_client.clone();
1835 let interval = bar_spec_to_binance_interval(bar_type.spec())?;
1836
1837 let stream = format!(
1838 "{}@kline_{}",
1839 bar_type.instrument_id().symbol.as_str().to_lowercase(),
1840 interval.as_str()
1841 );
1842
1843 self.spawn_ws(
1844 async move {
1845 ws.unsubscribe(vec![stream])
1846 .await
1847 .context("bars unsubscribe")
1848 },
1849 "bar unsubscribe",
1850 );
1851 Ok(())
1852 }
1853
1854 fn unsubscribe_instrument_status(
1855 &mut self,
1856 cmd: &UnsubscribeInstrumentStatus,
1857 ) -> anyhow::Result<()> {
1858 log::debug!(
1859 "unsubscribe_instrument_status: {id}",
1860 id = cmd.instrument_id,
1861 );
1862 Ok(())
1863 }
1864
1865 fn request_instruments(&self, request: RequestInstruments) -> anyhow::Result<()> {
1866 let http = self.http_client.clone();
1867 let sender = self.data_sender.clone();
1868 let instruments_cache = self.instruments.clone();
1869 let request_id = request.request_id;
1870 let client_id = request.client_id.unwrap_or(self.client_id);
1871 let venue = self.venue();
1872 let start = request.start;
1873 let end = request.end;
1874 let params = request.params;
1875 let clock = self.clock;
1876 let start_nanos = datetime_to_unix_nanos(start);
1877 let end_nanos = datetime_to_unix_nanos(end);
1878
1879 get_runtime().spawn(async move {
1880 match http.request_instruments().await {
1881 Ok(instruments) => {
1882 for instrument in &instruments {
1883 upsert_instrument(&instruments_cache, instrument.clone());
1884 }
1885
1886 let response = DataResponse::Instruments(InstrumentsResponse::new(
1887 request_id,
1888 client_id,
1889 venue,
1890 instruments,
1891 start_nanos,
1892 end_nanos,
1893 clock.get_time_ns(),
1894 params,
1895 ));
1896
1897 if let Err(e) = sender.send(DataEvent::Response(response)) {
1898 log::error!("Failed to send instruments response: {e}");
1899 }
1900 }
1901 Err(e) => log::error!("Instruments request failed: {e:?}"),
1902 }
1903 });
1904
1905 Ok(())
1906 }
1907
1908 fn request_instrument(&self, request: RequestInstrument) -> anyhow::Result<()> {
1909 let http = self.http_client.clone();
1910 let sender = self.data_sender.clone();
1911 let instruments = self.instruments.clone();
1912 let instrument_id = request.instrument_id;
1913 let request_id = request.request_id;
1914 let client_id = request.client_id.unwrap_or(self.client_id);
1915 let start = request.start;
1916 let end = request.end;
1917 let params = request.params;
1918 let clock = self.clock;
1919 let start_nanos = datetime_to_unix_nanos(start);
1920 let end_nanos = datetime_to_unix_nanos(end);
1921
1922 get_runtime().spawn(async move {
1923 match http.request_instruments().await {
1924 Ok(all_instruments) => {
1925 for instrument in &all_instruments {
1926 upsert_instrument(&instruments, instrument.clone());
1927 }
1928
1929 let instrument = all_instruments
1930 .into_iter()
1931 .find(|i| i.id() == instrument_id);
1932
1933 if let Some(instrument) = instrument {
1934 let response = DataResponse::Instrument(Box::new(InstrumentResponse::new(
1935 request_id,
1936 client_id,
1937 instrument.id(),
1938 instrument,
1939 start_nanos,
1940 end_nanos,
1941 clock.get_time_ns(),
1942 params,
1943 )));
1944
1945 if let Err(e) = sender.send(DataEvent::Response(response)) {
1946 log::error!("Failed to send instrument response: {e}");
1947 }
1948 } else {
1949 log::error!("Instrument not found: {instrument_id}");
1950 }
1951 }
1952 Err(e) => log::error!("Instrument request failed: {e:?}"),
1953 }
1954 });
1955
1956 Ok(())
1957 }
1958
1959 fn request_trades(&self, request: RequestTrades) -> anyhow::Result<()> {
1960 let http = self.http_client.clone();
1961 let sender = self.data_sender.clone();
1962 let instrument_id = request.instrument_id;
1963 let limit = request.limit.map(|n| n.get() as u32);
1964 let request_id = request.request_id;
1965 let client_id = request.client_id.unwrap_or(self.client_id);
1966 let params = request.params;
1967 let clock = self.clock;
1968 let start_nanos = datetime_to_unix_nanos(request.start);
1969 let end_nanos = datetime_to_unix_nanos(request.end);
1970
1971 get_runtime().spawn(async move {
1972 match http
1973 .request_trades(instrument_id, limit)
1974 .await
1975 .context("failed to request trades from Binance")
1976 {
1977 Ok(trades) => {
1978 let response = DataResponse::Trades(TradesResponse::new(
1979 request_id,
1980 client_id,
1981 instrument_id,
1982 trades,
1983 start_nanos,
1984 end_nanos,
1985 clock.get_time_ns(),
1986 params,
1987 ));
1988
1989 if let Err(e) = sender.send(DataEvent::Response(response)) {
1990 log::error!("Failed to send trades response: {e}");
1991 }
1992 }
1993 Err(e) => log::error!("Trade request failed: {e:?}"),
1994 }
1995 });
1996
1997 Ok(())
1998 }
1999
2000 fn request_bars(&self, request: RequestBars) -> anyhow::Result<()> {
2001 let http = self.http_client.clone();
2002 let sender = self.data_sender.clone();
2003 let bar_type = request.bar_type;
2004 let start = request.start;
2005 let end = request.end;
2006 let limit = request.limit.map(|n| n.get() as u32);
2007 let request_id = request.request_id;
2008 let client_id = request.client_id.unwrap_or(self.client_id);
2009 let params = request.params;
2010 let clock = self.clock;
2011 let start_nanos = datetime_to_unix_nanos(start);
2012 let end_nanos = datetime_to_unix_nanos(end);
2013
2014 get_runtime().spawn(async move {
2015 match http
2016 .request_bars(bar_type, start, end, limit)
2017 .await
2018 .context("failed to request bars from Binance")
2019 {
2020 Ok(bars) => {
2021 let response = DataResponse::Bars(BarsResponse::new(
2022 request_id,
2023 client_id,
2024 bar_type,
2025 bars,
2026 start_nanos,
2027 end_nanos,
2028 clock.get_time_ns(),
2029 params,
2030 ));
2031
2032 if let Err(e) = sender.send(DataEvent::Response(response)) {
2033 log::error!("Failed to send bars response: {e}");
2034 }
2035 }
2036 Err(e) => log::error!("Bar request failed: {e:?}"),
2037 }
2038 });
2039
2040 Ok(())
2041 }
2042}
2043
2044#[cfg(test)]
2045mod tests {
2046 use std::time::Duration;
2047
2048 use nautilus_core::nanos::UnixNanos;
2049 use nautilus_model::{
2050 data::{BookOrder, OrderBookDelta, OrderBookDeltas},
2051 enums::{BookAction, OrderSide, RecordFlag},
2052 identifiers::InstrumentId,
2053 types::{Price, Quantity},
2054 };
2055 use rstest::rstest;
2056 use rust_decimal_macros::dec;
2057
2058 use super::{
2059 BinanceDepth, BinanceEnvironment, BinanceSpotMarketDataMode, BufferedDepthUpdate,
2060 first_applicable_spot_update, parse_spot_depth_snapshot, resolve_spot_json_ws_url,
2061 spot_continuity_ok, spot_overlap_valid, spot_snapshot_retry_backoff,
2062 };
2063 use crate::{common::consts::BINANCE_SPOT_WS_URL, spot::http::BinancePriceLevel};
2064
2065 #[rstest]
2066 fn overlap_accepts_first_diff_straddling_snapshot() {
2067 assert!(spot_overlap_valid(90, 110, 100));
2068 assert!(spot_overlap_valid(101, 101, 100));
2069 assert!(spot_overlap_valid(101, 200, 100));
2070 }
2071
2072 #[rstest]
2073 fn overlap_rejects_gap_and_stale() {
2074 assert!(!spot_overlap_valid(103, 110, 100));
2075 assert!(!spot_overlap_valid(90, 100, 100));
2076 }
2077
2078 #[rstest]
2079 fn continuity_skips_first_then_requires_contiguous_u() {
2080 assert!(spot_continuity_ok(true, 999, 100));
2081 assert!(spot_continuity_ok(false, 101, 100));
2082 assert!(!spot_continuity_ok(false, 102, 100));
2083 assert!(!spot_continuity_ok(false, 100, 100));
2084 }
2085
2086 #[rstest]
2087 #[case(0, 250)]
2088 #[case(1, 500)]
2089 #[case(2, 1_000)]
2090 #[case(3, 2_000)]
2091 #[case(4, 3_000)]
2092 #[case(5, 3_000)]
2093 fn snapshot_retry_backoff_exponentially_increases_then_caps(
2094 #[case] retry_count: u32,
2095 #[case] expected_ms: u64,
2096 ) {
2097 assert_eq!(
2098 spot_snapshot_retry_backoff(retry_count),
2099 Duration::from_millis(expected_ms)
2100 );
2101 }
2102
2103 #[rstest]
2104 fn first_applicable_update_skips_stale_diffs() {
2105 let updates = vec![
2106 buffered_update(90, 100),
2107 buffered_update(101, 101),
2108 buffered_update(102, 103),
2109 ];
2110
2111 let update = first_applicable_spot_update(&updates, 100).unwrap();
2112
2113 assert_eq!(update.first_update_id, 101);
2114 assert_eq!(update.final_update_id, 101);
2115 assert!(first_applicable_spot_update(&updates, 103).is_none());
2116 }
2117
2118 #[rstest]
2119 fn parse_spot_depth_snapshot_sets_sequence_and_last_flag() {
2120 let instrument_id = InstrumentId::from("BTCUSDT.BINANCE");
2121 let depth = depth_snapshot(
2122 vec![price_level(10_000, 1_000)],
2123 vec![price_level(10_100, 2_000)],
2124 );
2125
2126 let deltas = parse_spot_depth_snapshot(
2127 &depth,
2128 instrument_id,
2129 2,
2130 3,
2131 UnixNanos::from(1),
2132 UnixNanos::from(2),
2133 )
2134 .unwrap()
2135 .unwrap();
2136
2137 assert_eq!(deltas.deltas.len(), 3);
2138 assert_eq!(deltas.deltas[0].sequence, 123);
2139 assert_eq!(deltas.deltas[1].sequence, 123);
2140 assert_eq!(deltas.deltas[2].sequence, 123);
2141 assert_eq!(deltas.ts_event, UnixNanos::from(1));
2142 assert_eq!(deltas.ts_init, UnixNanos::from(2));
2143 assert_eq!(deltas.deltas[1].order.price.as_decimal(), dec!(100.00));
2144 assert_eq!(deltas.deltas[1].order.size.as_decimal(), dec!(1.000));
2145 assert_eq!(deltas.deltas[1].flags, 0);
2146 assert_eq!(deltas.deltas[2].flags, RecordFlag::F_LAST as u8);
2147 }
2148
2149 #[rstest]
2150 fn parse_spot_depth_snapshot_sets_last_flag_for_bid_only_snapshot() {
2151 let instrument_id = InstrumentId::from("BTCUSDT.BINANCE");
2152 let depth = depth_snapshot(vec![price_level(10_000, 1_000)], vec![]);
2153
2154 let deltas = parse_spot_depth_snapshot(
2155 &depth,
2156 instrument_id,
2157 2,
2158 3,
2159 UnixNanos::from(1),
2160 UnixNanos::from(2),
2161 )
2162 .unwrap()
2163 .unwrap();
2164
2165 assert_eq!(deltas.deltas.len(), 2);
2166 assert_eq!(deltas.deltas[1].flags, RecordFlag::F_LAST as u8);
2167 }
2168
2169 #[rstest]
2170 fn parse_spot_depth_snapshot_returns_none_for_empty_book() {
2171 let instrument_id = InstrumentId::from("BTCUSDT.BINANCE");
2172 let depth = depth_snapshot(vec![], vec![]);
2173
2174 let deltas = parse_spot_depth_snapshot(
2175 &depth,
2176 instrument_id,
2177 2,
2178 3,
2179 UnixNanos::from(1),
2180 UnixNanos::from(2),
2181 )
2182 .unwrap();
2183
2184 assert!(deltas.is_none());
2185 }
2186
2187 #[rstest]
2188 fn parse_spot_depth_snapshot_rejects_out_of_range_price() {
2189 let instrument_id = InstrumentId::from("BTCUSDT.BINANCE");
2190 let depth = BinanceDepth {
2191 last_update_id: 123,
2192 price_exponent: 100,
2193 qty_exponent: -3,
2194 bids: vec![price_level(i64::MAX, 1_000)],
2195 asks: vec![],
2196 };
2197
2198 let result = parse_spot_depth_snapshot(
2199 &depth,
2200 instrument_id,
2201 2,
2202 3,
2203 UnixNanos::from(1),
2204 UnixNanos::from(2),
2205 );
2206
2207 assert!(result.is_err());
2208 }
2209
2210 #[rstest]
2211 fn parse_spot_depth_snapshot_rejects_out_of_range_quantity() {
2212 let instrument_id = InstrumentId::from("BTCUSDT.BINANCE");
2213 let depth = BinanceDepth {
2214 last_update_id: 123,
2215 price_exponent: -2,
2216 qty_exponent: 100,
2217 bids: vec![price_level(10_000, i64::MAX)],
2218 asks: vec![],
2219 };
2220
2221 let result = parse_spot_depth_snapshot(
2222 &depth,
2223 instrument_id,
2224 2,
2225 3,
2226 UnixNanos::from(1),
2227 UnixNanos::from(2),
2228 );
2229
2230 assert!(result.is_err());
2231 }
2232
2233 fn buffered_update(first_update_id: u64, final_update_id: u64) -> BufferedDepthUpdate {
2234 let instrument_id = InstrumentId::from("BTCUSDT.BINANCE");
2235 let ts = UnixNanos::default();
2236 let order = BookOrder::new(
2237 OrderSide::Buy,
2238 Price::from_raw(1, 0),
2239 Quantity::from_raw(1, 0),
2240 0,
2241 );
2242 let delta = OrderBookDelta::new(
2243 instrument_id,
2244 BookAction::Update,
2245 order,
2246 0,
2247 final_update_id,
2248 ts,
2249 ts,
2250 );
2251 let deltas = OrderBookDeltas::new(instrument_id, vec![delta]);
2252
2253 BufferedDepthUpdate {
2254 deltas,
2255 first_update_id,
2256 final_update_id,
2257 }
2258 }
2259
2260 fn depth_snapshot(bids: Vec<BinancePriceLevel>, asks: Vec<BinancePriceLevel>) -> BinanceDepth {
2261 BinanceDepth {
2262 last_update_id: 123,
2263 price_exponent: -2,
2264 qty_exponent: -3,
2265 bids,
2266 asks,
2267 }
2268 }
2269
2270 fn price_level(price_mantissa: i64, qty_mantissa: i64) -> BinancePriceLevel {
2271 BinancePriceLevel {
2272 price_mantissa,
2273 qty_mantissa,
2274 }
2275 }
2276
2277 #[rstest]
2278 fn test_spot_market_data_mode_default_is_sbe() {
2279 assert_eq!(
2280 BinanceSpotMarketDataMode::default(),
2281 BinanceSpotMarketDataMode::Sbe
2282 );
2283 }
2284
2285 #[rstest]
2286 fn test_resolve_spot_json_ws_url_uses_environment_default_without_override() {
2287 assert_eq!(
2288 resolve_spot_json_ws_url(None, BinanceEnvironment::Live),
2289 BINANCE_SPOT_WS_URL.to_string()
2290 );
2291 }
2292
2293 #[rstest]
2294 fn test_resolve_spot_json_ws_url_rewrites_sbe_override_to_spot_default() {
2295 assert_eq!(
2296 resolve_spot_json_ws_url(
2297 Some("wss://stream-sbe.binance.com/ws".to_string()),
2298 BinanceEnvironment::Live,
2299 ),
2300 BINANCE_SPOT_WS_URL.to_string()
2301 );
2302 }
2303
2304 #[rstest]
2305 fn test_resolve_spot_json_ws_url_preserves_non_sbe_override() {
2306 let custom = "wss://example.com/ws".to_string();
2307 assert_eq!(
2308 resolve_spot_json_ws_url(Some(custom.clone()), BinanceEnvironment::Live),
2309 custom
2310 );
2311 }
2312}