1use std::{
19 num::NonZeroU32,
20 str::FromStr,
21 sync::{
22 Arc,
23 atomic::{AtomicBool, Ordering},
24 },
25 time::Duration,
26};
27
28use ahash::AHashMap;
29use anyhow::Context;
30use futures_util::{StreamExt, pin_mut};
31use nautilus_common::{
32 clients::{DataClient, execute_bar_subscriptions, execute_instrument_subscriptions},
33 live::{runner::get_data_event_sender, sender::EventSender},
34 messages::{
35 DataEvent,
36 data::{
37 BarsResponse, BookResponse, CustomDataResponse, DataResponse, InstrumentResponse,
38 InstrumentsResponse, RequestBars, RequestBookSnapshot, RequestCustomData,
39 RequestInstrument, RequestInstruments, RequestTrades, SubscribeBars,
40 SubscribeBookDeltas, SubscribeCustomData, SubscribeInstrument, SubscribeInstruments,
41 SubscribeQuotes, SubscribeTrades, TradesResponse, UnsubscribeBars,
42 UnsubscribeBookDeltas, UnsubscribeCustomData, UnsubscribeQuotes, UnsubscribeTrades,
43 subscribe::SubscribeInstrumentStatus, unsubscribe::UnsubscribeInstrumentStatus,
44 },
45 },
46};
47use nautilus_core::{
48 AtomicMap, Params,
49 datetime::datetime_to_unix_nanos,
50 nanos::UnixNanos,
51 time::{AtomicTime, get_atomic_clock_realtime},
52};
53use nautilus_live::{
54 SocketControlFactory,
55 task::{TaskGroup, TaskGroupGuard, TaskSpawner},
56};
57use nautilus_model::{
58 data::{BookOrder, CustomData, Data, DataType, OrderBookDelta, OrderBookDeltas, QuoteTick},
59 enums::{
60 AggregationSource, BookAction, BookType, MarketStatusAction, OrderSide, PriceType,
61 RecordFlag,
62 },
63 identifiers::{ClientId, InstrumentId, Venue},
64 instruments::{Instrument, InstrumentAny},
65 types::{Price, Quantity},
66};
67use parking_lot::Mutex;
68use tokio_util::sync::CancellationToken;
69use ustr::Ustr;
70
71use crate::{
72 book::{
73 BinanceBookError,
74 pacing::SnapshotPacer,
75 recovery::spawn_recovery,
76 sync::{BookSyncTracker, DepthSequencing, DepthSnapshot, DepthUpdate},
77 },
78 common::{
79 bar::{binance_bar_data_type, binance_bars_to_custom_data, parse_binance_bar_type},
80 consts::{BINANCE_VENUE, BINANCE_WS_HEARTBEAT_SECS},
81 credential::resolve_credentials,
82 enums::{BinanceEnvironment, BinanceProductType},
83 parse::{bar_spec_to_binance_interval, quote_to_l1_deltas},
84 status::diff_and_emit_statuses,
85 urls::{get_http_base_url_with_us, get_ws_base_url_with_us},
86 websocket::chain_command,
87 },
88 config::{BinanceDataClientConfig, BinanceSpotMarketDataMode},
89 data_types::register_binance_custom_data,
90 spot::{
91 http::{BinanceDepth, BinanceSpotHttpError, DepthParams, client::BinanceSpotHttpClient},
92 websocket::{
93 public_json::{
94 BinanceSpotPublicJsonWebSocketClient,
95 messages::BinanceSpotPublicWsMessage,
96 parse::{
97 parse_book_ticker as parse_json_book_ticker,
98 parse_depth_diff as parse_json_depth_diff,
99 parse_depth_snapshot as parse_json_depth_snapshot,
100 parse_kline as parse_json_kline, parse_ticker as parse_json_ticker,
101 parse_trade as parse_json_trade,
102 },
103 },
104 streams::{
105 client::BinanceSpotWebSocketClient,
106 messages::BinanceSpotWsMessage,
107 parse::{
108 parse_bbo_event, parse_depth_diff, parse_depth_snapshot, parse_trades_event,
109 },
110 },
111 },
112 },
113};
114
115const BOOK_DEPTHS_JSON: [usize; 3] = [5, 10, 20];
116
117const SNAPSHOT_DEPTH: u32 = 5000;
119
120const SNAPSHOT_WEIGHT_PER_MINUTE: NonZeroU32 = NonZeroU32::new(3_000).expect("non-zero");
122
123#[derive(Debug)]
125pub struct BinanceSpotDataClient {
126 clock: &'static AtomicTime,
127 client_id: ClientId,
128 config: BinanceDataClientConfig,
129 http_client: BinanceSpotHttpClient,
130 ws_client: SpotWsClient,
131 spot_market_data_mode: BinanceSpotMarketDataMode,
132 is_connected: AtomicBool,
133 cancellation_token: CancellationToken,
134 session_tasks: TaskGroup,
135 command_tasks: TaskGroup,
136 shutdown_errors: Vec<String>,
137 data_sender: EventSender<DataEvent>,
138 instruments: Arc<AtomicMap<InstrumentId, InstrumentAny>>,
139 status_cache: Arc<AtomicMap<InstrumentId, MarketStatusAction>>,
140 instrument_update_lock: Arc<tokio::sync::Mutex<()>>,
141 book_sync: BookSyncTracker,
142 book_subscriptions: Arc<AtomicMap<InstrumentId, String>>,
143 l1_book_subscriptions: Arc<AtomicMap<InstrumentId, u32>>,
144 quote_refs: Arc<AtomicMap<InstrumentId, u32>>,
145 ticker_refs: Arc<AtomicMap<InstrumentId, u32>>,
146 book_command_tail: Mutex<Option<tokio::sync::oneshot::Receiver<()>>>,
147}
148
149impl BinanceSpotDataClient {
150 pub fn new(client_id: ClientId, config: BinanceDataClientConfig) -> anyhow::Result<Self> {
156 config.validate()?;
157 let clock = get_atomic_clock_realtime();
158 let spot_market_data_mode = config.spot_market_data_mode;
159 let base_url_http = config.base_url_http.clone().or_else(|| {
160 config.us.then(|| {
161 get_http_base_url_with_us(config.product_type, config.environment, true).to_string()
162 })
163 });
164 let api_key = config
165 .api_key
166 .as_ref()
167 .map(|value| value.expose_secret().to_owned());
168 let api_secret = config
169 .api_secret
170 .as_ref()
171 .map(|value| value.expose_secret().to_owned());
172 let proxy_url = config
173 .proxy_url
174 .as_ref()
175 .map(|value| value.expose_secret().to_owned());
176
177 let http_client = BinanceSpotHttpClient::new_with_json_responses(
178 config.environment,
179 clock,
180 api_key.clone(),
181 api_secret.clone(),
182 base_url_http,
183 Some(config.recv_window_ms),
184 None, proxy_url.clone(),
186 config.us,
187 )?
188 .with_retry_config(config.retry_config());
189
190 let creds = if spot_market_data_mode == BinanceSpotMarketDataMode::Sbe {
191 resolve_credentials(api_key, api_secret, config.environment, config.product_type)
192 .inspect_err(|e| {
193 log::warn!(
194 "Failed to resolve Binance API credentials ({e}). \
195 Spot SBE WebSocket streams require an Ed25519 API key. \
196 Set the appropriate env vars for your environment, \
197 or provide api_key/api_secret in the data client config"
198 );
199 })
200 .ok()
201 } else {
202 None
203 };
204
205 let socket_factory = SocketControlFactory::new(client_id, Some(*BINANCE_VENUE));
206 let ws_client = match spot_market_data_mode {
207 BinanceSpotMarketDataMode::Sbe => SpotWsClient::Sbe(
209 BinanceSpotWebSocketClient::new(
210 config.base_url_ws.clone(),
211 creds.as_ref().map(|(k, _)| k.clone()),
212 creds.as_ref().map(|(_, s)| s.clone()),
213 Some(BINANCE_WS_HEARTBEAT_SECS),
214 config.transport_backend,
215 )?
216 .with_proxy(proxy_url)
217 .with_socket_control(socket_factory, "binance-spot-sbe-data-streams"),
218 ),
219 BinanceSpotMarketDataMode::Json => SpotWsClient::JsonPublic(
220 BinanceSpotPublicJsonWebSocketClient::new(
221 Some(resolve_spot_json_ws_url(
222 config.base_url_ws.clone(),
223 config.environment,
224 config.us,
225 )),
226 Some(BINANCE_WS_HEARTBEAT_SECS),
227 config.transport_backend,
228 )
229 .with_proxy(proxy_url)
230 .with_socket_control(socket_factory, "binance-spot-json-data-streams"),
231 ),
232 };
233 let data_sender = get_data_event_sender();
234 let snapshot_pacer = Arc::new(SnapshotPacer::new(SNAPSHOT_WEIGHT_PER_MINUTE));
235 let book_sync =
236 BookSyncTracker::new(DepthSequencing::Spot, data_sender.clone(), snapshot_pacer);
237
238 log::debug!("Configured Spot market data mode: {spot_market_data_mode:?}");
239
240 let session_tasks = TaskGroup::new();
241 let command_tasks = TaskGroup::new();
242
243 Ok(Self {
244 clock,
245 client_id,
246 config,
247 http_client,
248 ws_client,
249 spot_market_data_mode,
250 is_connected: AtomicBool::new(false),
251 cancellation_token: session_tasks.cancellation_token(),
252 session_tasks,
253 command_tasks,
254 shutdown_errors: Vec::new(),
255 data_sender,
256 instruments: Arc::new(AtomicMap::new()),
257 status_cache: Arc::new(AtomicMap::new()),
258 instrument_update_lock: Arc::new(tokio::sync::Mutex::new(())),
259 book_sync,
260 book_subscriptions: Arc::new(AtomicMap::new()),
261 l1_book_subscriptions: Arc::new(AtomicMap::new()),
262 quote_refs: Arc::new(AtomicMap::new()),
263 ticker_refs: Arc::new(AtomicMap::new()),
264 book_command_tail: Mutex::new(None),
265 })
266 }
267
268 fn venue(&self) -> Venue {
269 *BINANCE_VENUE
270 }
271
272 fn send_data(sender: &EventSender<DataEvent>, data: Data) {
273 if let Err(e) = sender.send(DataEvent::Data(data)) {
274 log::error!("Failed to emit data event: {e}");
275 }
276 }
277
278 fn spawn_ws<F>(&self, fut: F, context: &'static str)
279 where
280 F: std::future::Future<Output = anyhow::Result<()>> + Send + 'static,
281 {
282 let future = async move {
283 if let Err(e) = fut.await {
284 log::error!("{context}: {e:?}");
285 }
286 };
287
288 if let Err(e) = self.command_tasks.spawn(future) {
289 log::warn!("Skipping Binance Spot {context} after shutdown began: {e}");
290 }
291 }
292
293 fn spawn_command<F>(&self, future: F)
294 where
295 F: std::future::Future<Output = ()> + Send + 'static,
296 {
297 if let Err(e) = self.command_tasks.spawn(future) {
298 log::warn!("Skipping Binance Spot data command after shutdown began: {e}");
299 }
300 }
301
302 async fn finish_tasks(&self) -> anyhow::Result<()> {
303 let (session_result, command_result) = tokio::join!(
304 self.session_tasks
305 .finish_shutdown(Duration::from_secs(1), Duration::from_secs(2)),
306 self.command_tasks
307 .finish_shutdown(Duration::from_secs(1), Duration::from_secs(2)),
308 );
309 let mut errors = Vec::new();
310 if let Err(e) = session_result {
311 errors.push(format!(
312 "failed to finish Binance Spot data session tasks: {e}"
313 ));
314 }
315
316 if let Err(e) = command_result {
317 errors.push(format!(
318 "failed to finish Binance Spot data command tasks: {e}"
319 ));
320 }
321
322 if !errors.is_empty() {
323 anyhow::bail!(errors.join("; "));
324 }
325 Ok(())
326 }
327
328 async fn prepare_task_groups(&mut self) -> anyhow::Result<()> {
329 if !self.session_tasks.is_open() || !self.command_tasks.is_open() {
330 self.teardown_partial_connect().await?;
331 self.session_tasks
332 .start_generation()
333 .context("failed to start Binance Spot data session task generation")?;
334 self.command_tasks
335 .start_generation()
336 .context("failed to start Binance Spot data command task generation")?;
337 self.cancellation_token = self.session_tasks.cancellation_token();
338 }
339 Ok(())
340 }
341
342 async fn teardown_partial_connect(&mut self) -> anyhow::Result<()> {
343 self.session_tasks.begin_shutdown();
344 self.command_tasks.begin_shutdown();
345 self.ws_client.begin_shutdown();
346 if let Err(e) = self.ws_client.close().await {
347 self.shutdown_errors
348 .push(format!("WebSocket close failed: {e}"));
349 }
350
351 if let Err(e) = self.finish_tasks().await {
352 self.shutdown_errors.push(e.to_string());
353 }
354 self.is_connected.store(false, Ordering::Release);
355
356 if !self.shutdown_errors.is_empty() {
357 let errors = std::mem::take(&mut self.shutdown_errors);
358 anyhow::bail!("Binance Spot data teardown failed: {}", errors.join("; "));
359 }
360 Ok(())
361 }
362
363 #[expect(clippy::too_many_arguments)]
364 async fn refresh_instrument_catalog(
365 http: &BinanceSpotHttpClient,
366 provider: &crate::config::BinanceInstrumentProviderConfig,
367 us: bool,
368 instruments_cache: &Arc<AtomicMap<InstrumentId, InstrumentAny>>,
369 status_cache: &Arc<AtomicMap<InstrumentId, MarketStatusAction>>,
370 instrument_update_lock: &tokio::sync::Mutex<()>,
371 ws: &SpotWsClient,
372 sender: &EventSender<DataEvent>,
373 clock: &'static AtomicTime,
374 emit_status_changes: bool,
375 ) -> anyhow::Result<Vec<InstrumentAny>> {
376 let _guard = instrument_update_lock.lock().await;
377 let instruments = http
378 .request_instruments_with_config(provider, us)
379 .await
380 .context("failed to request Binance Spot instruments")?;
381 let venue_statuses = http
382 .request_symbol_statuses(us)
383 .await
384 .context("failed to request Binance Spot instrument statuses")?;
385
386 let instrument_map = instruments
387 .iter()
388 .map(|instrument| (instrument.id(), instrument.clone()))
389 .collect::<AHashMap<_, _>>();
390 let status_map = venue_statuses
391 .into_iter()
392 .filter(|(instrument_id, _)| instrument_map.contains_key(instrument_id))
393 .collect::<AHashMap<_, _>>();
394
395 instruments_cache.store(instrument_map);
396 ws.replace_instruments(&instruments);
397
398 if emit_status_changes {
399 let mut cached_statuses = (**status_cache.load()).clone();
400 let ts = clock.get_time_ns();
401 diff_and_emit_statuses(&status_map, &mut cached_statuses, sender, ts, ts);
402 status_cache.store(cached_statuses);
403 } else {
404 status_cache.store(status_map);
405 }
406
407 for instrument in &instruments {
408 if let Err(e) = sender.send(DataEvent::Data(instrument.clone().into())) {
409 log::warn!("Failed to send refreshed Binance Spot instrument: {e}");
410 }
411 }
412
413 Ok(instruments)
414 }
415
416 #[expect(clippy::too_many_arguments)]
417 fn handle_ws_message(
418 msg: BinanceSpotWsMessage,
419 data_sender: &EventSender<DataEvent>,
420 instruments: &Arc<AtomicMap<InstrumentId, InstrumentAny>>,
421 ws_instruments: &Arc<AtomicMap<Ustr, InstrumentAny>>,
422 book_sync: &BookSyncTracker,
423 l1_book_subscriptions: &Arc<AtomicMap<InstrumentId, u32>>,
424 http_client: &BinanceSpotHttpClient,
425 snapshot_timeout: Duration,
426 clock: &'static AtomicTime,
427 command_spawner: &TaskSpawner,
428 ) {
429 let ts_init = clock.get_time_ns();
430
431 match msg {
432 BinanceSpotWsMessage::Trades(ref event) => {
433 let symbol = event.symbol;
434 let cache = ws_instruments.load();
435 if let Some(instrument) = cache.get(&symbol) {
436 match parse_trades_event(event, instrument, ts_init) {
437 Ok(trades) => {
438 for data in trades {
439 Self::send_data(data_sender, data);
440 }
441 }
442 Err(e) => log::warn!("Failed to parse Spot SBE trades: {e}"),
443 }
444 }
445 }
446 BinanceSpotWsMessage::BestBidAsk(ref event) => {
447 let symbol = event.symbol;
448 let cache = ws_instruments.load();
449 if let Some(instrument) = cache.get(&symbol) {
450 match parse_bbo_event(event, instrument, ts_init) {
451 Ok(quote) => Self::send_top_of_book(
452 data_sender,
453 l1_book_subscriptions,
454 quote,
455 event.book_update_id as u64,
456 ),
457 Err(e) => log::warn!("Failed to parse Spot SBE best bid/ask: {e}"),
458 }
459 }
460 }
461 BinanceSpotWsMessage::DepthSnapshot(ref event) => {
462 let symbol = event.symbol;
463 let cache = ws_instruments.load();
464 if let Some(instrument) = cache.get(&symbol) {
465 match parse_depth_snapshot(event, instrument, ts_init) {
466 Ok(deltas) => {
467 Self::send_data(data_sender, Data::BookDeltas(Box::new(deltas)));
468 }
469 Err(e) => log::warn!("Failed to parse Spot SBE depth snapshot: {e}"),
470 }
471 }
472 }
473 BinanceSpotWsMessage::DepthDiff(ref event) => {
474 let symbol = event.symbol;
475 let cache = ws_instruments.load();
476 if let Some(instrument) = cache.get(&symbol) {
477 match parse_depth_diff(event, instrument, ts_init) {
478 Ok(deltas) => {
479 let update = DepthUpdate {
480 first_update_id: event.first_book_update_id as u64,
481 final_update_id: event.last_book_update_id as u64,
482 prev_final_update_id: None,
483 deltas,
484 };
485
486 Self::route_depth_diff(
487 instrument.id(),
488 update,
489 book_sync,
490 instruments,
491 http_client,
492 snapshot_timeout,
493 clock,
494 command_spawner,
495 );
496 }
497 Err(e) => log::warn!("Failed to parse Spot SBE depth update: {e}"),
498 }
499 }
500 }
501 BinanceSpotWsMessage::ServerShutdown(ref msg) => {
502 log::warn!(
503 "Binance server shutdown notice (event_time={}); disconnect expected within ~10 minutes",
504 msg.event_time,
505 );
506 }
507 BinanceSpotWsMessage::RawBinary(data) => {
508 log::debug!("Unhandled binary message: {} bytes", data.len());
509 }
510 BinanceSpotWsMessage::RawJson(value) => {
511 log::debug!("Unhandled JSON message: {value:?}");
512 }
513 BinanceSpotWsMessage::Error(e) => {
514 log::warn!("Binance WebSocket error: code={}, msg={}", e.code, e.msg);
515 }
516 BinanceSpotWsMessage::Reconnected => {
517 log::info!("WebSocket reconnected, rebuilding order book snapshots");
518 book_sync.reset_on_reconnect();
519 }
520 }
521 }
522
523 #[expect(clippy::too_many_arguments)]
524 fn handle_public_json_ws_message(
525 msg: BinanceSpotPublicWsMessage,
526 data_sender: &EventSender<DataEvent>,
527 instruments: &Arc<AtomicMap<InstrumentId, InstrumentAny>>,
528 ws_instruments: &Arc<AtomicMap<Ustr, InstrumentAny>>,
529 book_sync: &BookSyncTracker,
530 l1_book_subscriptions: &Arc<AtomicMap<InstrumentId, u32>>,
531 http_client: &BinanceSpotHttpClient,
532 snapshot_timeout: Duration,
533 clock: &'static AtomicTime,
534 command_spawner: &TaskSpawner,
535 ) {
536 let ts_init = clock.get_time_ns();
537
538 match msg {
539 BinanceSpotPublicWsMessage::Trade(ref event) => {
540 let symbol = event.symbol;
541 let cache = ws_instruments.load();
542 if let Some(instrument) = cache.get(&symbol) {
543 match parse_json_trade(event, instrument, ts_init) {
544 Ok(trade) => Self::send_data(data_sender, Data::Trade(trade)),
545 Err(e) => log::warn!("Failed to parse Spot JSON trade: {e}"),
546 }
547 }
548 }
549 BinanceSpotPublicWsMessage::BookTicker(ref event) => {
550 let symbol = event.symbol;
551 let cache = ws_instruments.load();
552 if let Some(instrument) = cache.get(&symbol) {
553 match parse_json_book_ticker(event, instrument, ts_init) {
554 Ok(quote) => Self::send_top_of_book(
555 data_sender,
556 l1_book_subscriptions,
557 quote,
558 event.book_update_id,
559 ),
560 Err(e) => log::warn!("Failed to parse Spot JSON book ticker: {e}"),
561 }
562 }
563 }
564 BinanceSpotPublicWsMessage::DepthSnapshot(ref event) => {
565 let symbol = event.symbol;
566 let cache = ws_instruments.load();
567 if let Some(instrument) = cache.get(&symbol) {
568 let deltas = parse_json_depth_snapshot(event, instrument, ts_init);
569 Self::send_data(data_sender, Data::BookDeltas(Box::new(deltas)));
570 }
571 }
572 BinanceSpotPublicWsMessage::DepthDiff(ref event) => {
573 let symbol = event.symbol;
574 let cache = ws_instruments.load();
575 if let Some(instrument) = cache.get(&symbol) {
576 match parse_json_depth_diff(event, instrument, ts_init) {
578 Ok(deltas) => {
579 let update = DepthUpdate {
580 first_update_id: event.first_update_id,
581 final_update_id: event.final_update_id,
582 prev_final_update_id: None,
583 deltas,
584 };
585
586 Self::route_depth_diff(
587 instrument.id(),
588 update,
589 book_sync,
590 instruments,
591 http_client,
592 snapshot_timeout,
593 clock,
594 command_spawner,
595 );
596 }
597 Err(e) => log::warn!("Failed to parse Spot JSON depth update: {e}"),
598 }
599 }
600 }
601 BinanceSpotPublicWsMessage::Kline(ref event) => {
602 let symbol = event.symbol;
603 let cache = ws_instruments.load();
604 if let Some(instrument) = cache.get(&symbol) {
605 match parse_json_kline(event, instrument, ts_init) {
606 Ok(Some(bar)) => {
607 Self::send_data(data_sender, Data::Bar(bar.bar()));
608 let data_type = binance_bar_data_type(bar.bar_type);
609 Self::send_data(
610 data_sender,
611 Data::Custom(CustomData::new(Arc::new(bar), data_type)),
612 );
613 }
614 Ok(None) => {} Err(e) => log::warn!("Failed to parse Spot JSON kline: {e}"),
616 }
617 }
618 }
619 BinanceSpotPublicWsMessage::Ticker(ref event) => {
620 let symbol = event.symbol;
621 let cache = ws_instruments.load();
622 if let Some(instrument) = cache.get(&symbol) {
623 match parse_json_ticker(event, instrument, ts_init) {
624 Ok(ticker) => {
625 let data_type = spot_ticker_data_type(instrument.id());
626 Self::send_data(
627 data_sender,
628 Data::Custom(CustomData::new(Arc::new(ticker), data_type)),
629 );
630 }
631 Err(e) => log::warn!("Failed to parse Spot JSON ticker: {e}"),
632 }
633 }
634 }
635 BinanceSpotPublicWsMessage::ServerShutdown(ref msg) => {
636 log::warn!(
637 "Binance Spot JSON server shutdown notice (event_time={}); disconnect expected within ~10 minutes",
638 msg.event_time,
639 );
640 }
641 BinanceSpotPublicWsMessage::RawJson(value) => {
642 log::debug!("Unhandled Spot JSON message: {value:?}");
643 }
644 BinanceSpotPublicWsMessage::Error(e) => {
645 log::warn!("Spot JSON WebSocket error: code={}, msg={}", e.code, e.msg);
646 }
647 BinanceSpotPublicWsMessage::Reconnected => {
648 log::info!("Spot JSON WebSocket reconnected, rebuilding order book snapshots");
649 book_sync.reset_on_reconnect();
650 }
651 }
652 }
653
654 fn send_top_of_book(
655 data_sender: &EventSender<DataEvent>,
656 l1_book_subscriptions: &Arc<AtomicMap<InstrumentId, u32>>,
657 quote: QuoteTick,
658 sequence: u64,
659 ) {
660 Self::send_data(data_sender, Data::Quote(quote));
661 if l1_book_subscriptions.contains_key("e.instrument_id) {
662 let deltas = quote_to_l1_deltas(quote, sequence);
663 Self::send_data(data_sender, Data::BookDeltas(Box::new(deltas)));
664 }
665 }
666
667 #[expect(
668 clippy::too_many_arguments,
669 reason = "book recovery requires the full subscription and command ownership context"
670 )]
671 fn route_depth_diff(
672 instrument_id: InstrumentId,
673 update: DepthUpdate,
674 book_sync: &BookSyncTracker,
675 instruments: &Arc<AtomicMap<InstrumentId, InstrumentAny>>,
676 http_client: &BinanceSpotHttpClient,
677 snapshot_timeout: Duration,
678 clock: &'static AtomicTime,
679 command_spawner: &TaskSpawner,
680 ) {
681 let Some(recovery) = book_sync.handle_update(instrument_id, update) else {
682 return;
683 };
684
685 let http = http_client.clone();
686 let instruments = instruments.clone();
687
688 spawn_recovery(
689 instrument_id,
690 recovery,
691 book_sync.clone(),
692 move || {
693 Self::fetch_depth_snapshot(http.clone(), instruments.clone(), instrument_id, clock)
694 },
695 depth_request_weight(SNAPSHOT_DEPTH),
696 snapshot_timeout,
697 command_spawner,
698 );
699 }
700
701 fn quote_stream_suffix(&self) -> &'static str {
702 match self.spot_market_data_mode {
703 BinanceSpotMarketDataMode::Sbe => "bestBidAsk",
704 BinanceSpotMarketDataMode::Json => "bookTicker",
705 }
706 }
707
708 fn required_instrument_id_metadata(data_type: &DataType) -> anyhow::Result<InstrumentId> {
709 let raw = data_type
710 .metadata()
711 .as_ref()
712 .and_then(|metadata| metadata.get("instrument_id"))
713 .and_then(|value| value.as_str())
714 .map(str::trim)
715 .filter(|value| !value.is_empty())
716 .context("custom data subscription requires `instrument_id` metadata")?;
717 InstrumentId::from_str(raw)
718 .with_context(|| format!("invalid instrument_id metadata `{raw}`"))
719 }
720
721 async fn fetch_depth_snapshot(
722 http: BinanceSpotHttpClient,
723 instruments: Arc<AtomicMap<InstrumentId, InstrumentAny>>,
724 instrument_id: InstrumentId,
725 clock: &'static AtomicTime,
726 ) -> Result<DepthSnapshot, BinanceBookError> {
727 let params = DepthParams {
728 symbol: instrument_id.symbol.as_str().to_uppercase(),
729 limit: Some(SNAPSHOT_DEPTH),
730 };
731
732 let depth = http
733 .inner()
734 .depth(¶ms)
735 .await
736 .map_err(|e| depth_snapshot_error(instrument_id, &e))?;
737
738 let ts_init = clock.get_time_ns();
739
740 let (price_precision, size_precision) = instruments
741 .load()
742 .get(&instrument_id)
743 .map(|instrument| (instrument.price_precision(), instrument.size_precision()))
744 .ok_or_else(|| {
745 BinanceBookError::Permanent(format!("no instrument cached for {instrument_id}"))
746 })?;
747
748 let deltas = parse_spot_depth_snapshot(
749 &depth,
750 instrument_id,
751 price_precision,
752 size_precision,
753 ts_init,
754 )
755 .map_err(|e| {
756 BinanceBookError::Retryable(format!(
757 "failed to parse depth snapshot for {instrument_id}: {e}"
758 ))
759 })?;
760
761 Ok(DepthSnapshot {
762 last_update_id: depth.last_update_id as u64,
763 deltas,
764 has_event_time: false,
765 })
766 }
767}
768
769fn spot_ticker_data_type(instrument_id: InstrumentId) -> DataType {
770 let metadata = Params::new(
771 [(
772 "instrument_id".to_string(),
773 serde_json::Value::String(instrument_id.to_string()),
774 )]
775 .into(),
776 );
777 DataType::new(
778 "BinanceSpotTicker",
779 Some(metadata),
780 Some(instrument_id.to_string()),
781 )
782}
783
784fn upsert_instruments(
785 cache: &Arc<AtomicMap<InstrumentId, InstrumentAny>>,
786 instruments: &[InstrumentAny],
787) {
788 if instruments.is_empty() {
789 return;
790 }
791
792 cache.rcu(|cache| {
793 for instrument in instruments {
794 cache.insert(instrument.id(), instrument.clone());
795 }
796 });
797}
798
799fn parse_spot_depth_snapshot(
800 depth: &BinanceDepth,
801 instrument_id: InstrumentId,
802 price_precision: u8,
803 size_precision: u8,
804 ts_init: UnixNanos,
805) -> anyhow::Result<OrderBookDeltas> {
806 let sequence = depth.last_update_id as u64;
807
808 let total_levels = depth.bids.len() + depth.asks.len();
809 let mut deltas = Vec::with_capacity(total_levels + 1);
810
811 deltas.push(OrderBookDelta::clear(
813 instrument_id,
814 sequence,
815 ts_init,
816 ts_init,
817 ));
818
819 for (side, levels) in [
820 (OrderSide::Buy, &depth.bids),
821 (OrderSide::Sell, &depth.asks),
822 ] {
823 for level in levels {
824 let price = Price::from_mantissa_exponent_checked(
825 level.price_mantissa,
826 depth.price_exponent,
827 price_precision,
828 )?;
829 let size = Quantity::from_mantissa_exponent_checked(
830 level.qty_mantissa as u64,
831 depth.qty_exponent,
832 size_precision,
833 )?;
834 let order = BookOrder::new(side, price, size, 0);
835
836 deltas.push(OrderBookDelta::new(
837 instrument_id,
838 BookAction::Add,
839 order,
840 RecordFlag::F_SNAPSHOT as u8,
841 sequence,
842 ts_init,
843 ts_init,
844 ));
845 }
846 }
847
848 if let Some(delta) = deltas.last_mut() {
849 delta.flags |= RecordFlag::F_LAST as u8;
850 }
851
852 Ok(OrderBookDeltas::new(instrument_id, deltas))
853}
854
855fn depth_request_weight(limit: u32) -> u32 {
857 match limit {
858 ..=100 => 5,
859 101..=500 => 25,
860 501..=1000 => 50,
861 _ => 250,
862 }
863}
864
865fn depth_snapshot_error(instrument_id: InstrumentId, e: &BinanceSpotHttpError) -> BinanceBookError {
867 let message = format!("depth snapshot request for {instrument_id} failed: {e}");
868
869 if e.is_retryable() || matches!(e, BinanceSpotHttpError::RetryBudgetExceeded(_)) {
870 BinanceBookError::Retryable(message)
871 } else {
872 BinanceBookError::Permanent(message)
873 }
874}
875
876#[async_trait::async_trait(?Send)]
877impl DataClient for BinanceSpotDataClient {
878 fn client_id(&self) -> ClientId {
879 self.client_id
880 }
881
882 fn venue(&self) -> Option<Venue> {
883 Some(self.venue())
884 }
885
886 fn start(&mut self) -> anyhow::Result<()> {
887 log::info!(
888 "Started: client_id={}, product_type={:?}, environment={:?}",
889 self.client_id,
890 self.config.product_type,
891 self.config.environment,
892 );
893 Ok(())
894 }
895
896 fn stop(&mut self) -> anyhow::Result<()> {
897 log::info!("Stopping {id}", id = self.client_id);
898 self.session_tasks.begin_shutdown();
899 self.command_tasks.begin_shutdown();
900 self.ws_client.begin_shutdown();
901 self.is_connected.store(false, Ordering::Relaxed);
902 Ok(())
903 }
904
905 fn reset(&mut self) -> anyhow::Result<()> {
906 log::debug!("Resetting {id}", id = self.client_id);
907
908 self.session_tasks.begin_shutdown();
909 self.command_tasks.begin_shutdown();
910 self.ws_client.begin_shutdown();
911 self.is_connected.store(false, Ordering::Relaxed);
912
913 self.book_subscriptions.store(AHashMap::new());
914 self.l1_book_subscriptions.store(AHashMap::new());
915 self.quote_refs.store(AHashMap::new());
916 self.ticker_refs.store(AHashMap::new());
917 self.book_sync.clear();
918
919 Ok(())
920 }
921
922 fn dispose(&mut self) -> anyhow::Result<()> {
923 log::debug!("Disposing {id}", id = self.client_id);
924 self.stop()
925 }
926
927 async fn connect(&mut self) -> anyhow::Result<()> {
928 if self.is_connected() && self.session_tasks.is_open() && self.command_tasks.is_open() {
929 return Ok(());
930 }
931
932 register_binance_custom_data();
933
934 if self.spot_market_data_mode == BinanceSpotMarketDataMode::Sbe
935 && !self.ws_client.has_credentials()
936 {
937 anyhow::bail!(
938 "Binance Spot market data mode SBE requires Ed25519 API credentials. \
939 Set the appropriate env vars for your environment, \
940 or provide api_key/api_secret in the data client config"
941 );
942 }
943
944 self.prepare_task_groups().await?;
945 let ws_client = self.ws_client.clone();
946 let setup_guard =
947 TaskGroupGuard::new(&[&self.session_tasks, &self.command_tasks], move || {
948 ws_client.begin_shutdown();
949 });
950
951 Self::refresh_instrument_catalog(
952 &self.http_client,
953 &self.config.instrument_provider,
954 self.config.us,
955 &self.instruments,
956 &self.status_cache,
957 &self.instrument_update_lock,
958 &self.ws_client,
959 &self.data_sender,
960 self.clock,
961 false,
962 )
963 .await?;
964
965 let session_result = async {
966 match &mut self.ws_client {
967 SpotWsClient::Sbe(ws_client) => {
968 log::info!("Connecting to Binance Spot SBE WebSocket...");
969 ws_client.connect().await.map_err(|e| {
970 anyhow::anyhow!("failed to connect Binance Spot SBE WebSocket: {e}")
971 })?;
972 log::info!("Binance Spot SBE WebSocket connected");
973
974 let stream = ws_client.stream();
975 let sender = self.data_sender.clone();
976 let insts = self.instruments.clone();
977 let ws_insts = ws_client.instruments_cache();
978 let book_sync = self.book_sync.clone();
979 let l1_book_subs = self.l1_book_subscriptions.clone();
980 let http = self.http_client.clone();
981 let snapshot_timeout =
982 Duration::from_secs(self.config.book_snapshot_timeout_secs);
983 let clock = self.clock;
984 let cancel = self.cancellation_token.clone();
985 let command_spawner = self
986 .command_tasks
987 .spawner()
988 .context("Binance Spot command task admission is closed")?;
989
990 let future = async move {
991 pin_mut!(stream);
992
993 loop {
994 tokio::select! {
995 Some(message) = stream.next() => {
996 Self::handle_ws_message(
997 message,
998 &sender,
999 &insts,
1000 &ws_insts,
1001 &book_sync,
1002 &l1_book_subs,
1003 &http,
1004 snapshot_timeout,
1005 clock,
1006 &command_spawner,
1007 );
1008 }
1009 () = cancel.cancelled() => {
1010 log::debug!("Spot SBE WebSocket stream task cancelled");
1011 break;
1012 }
1013 }
1014 }
1015 };
1016 self.session_tasks
1017 .spawn(future)
1018 .context("failed to register Binance Spot SBE stream task")?;
1019 }
1020 SpotWsClient::JsonPublic(ws_client) => {
1021 log::info!("Connecting to Binance Spot public JSON WebSocket...");
1022 ws_client.connect().await.map_err(|e| {
1023 anyhow::anyhow!("failed to connect Binance Spot public JSON WebSocket: {e}")
1024 })?;
1025 log::info!("Binance Spot public JSON WebSocket connected");
1026
1027 let stream = ws_client.stream();
1028 let sender = self.data_sender.clone();
1029 let insts = self.instruments.clone();
1030 let ws_insts = ws_client.instruments_cache();
1031 let book_sync = self.book_sync.clone();
1032 let l1_book_subs = self.l1_book_subscriptions.clone();
1033 let http = self.http_client.clone();
1034 let snapshot_timeout =
1035 Duration::from_secs(self.config.book_snapshot_timeout_secs);
1036 let clock = self.clock;
1037 let cancel = self.cancellation_token.clone();
1038 let command_spawner = self
1039 .command_tasks
1040 .spawner()
1041 .context("Binance Spot command task admission is closed")?;
1042
1043 let future = async move {
1044 pin_mut!(stream);
1045
1046 loop {
1047 tokio::select! {
1048 Some(message) = stream.next() => {
1049 Self::handle_public_json_ws_message(
1050 message,
1051 &sender,
1052 &insts,
1053 &ws_insts,
1054 &book_sync,
1055 &l1_book_subs,
1056 &http,
1057 snapshot_timeout,
1058 clock,
1059 &command_spawner,
1060 );
1061 }
1062 () = cancel.cancelled() => {
1063 log::debug!("Spot JSON WebSocket stream task cancelled");
1064 break;
1065 }
1066 }
1067 }
1068 };
1069 self.session_tasks
1070 .spawn(future)
1071 .context("failed to register Binance Spot JSON stream task")?;
1072 }
1073 }
1074
1075 let poll_secs = self.config.instrument_status_poll_secs;
1076 if poll_secs > 0 {
1077 let http = self.http_client.clone();
1078 let poll_sender = self.data_sender.clone();
1079 let poll_instruments = self.instruments.clone();
1080 let poll_status_cache = self.status_cache.clone();
1081 let poll_update_lock = self.instrument_update_lock.clone();
1082 let poll_cancel = self.cancellation_token.clone();
1083 let clock = self.clock;
1084 let us = self.config.us;
1085
1086 let future = async move {
1087 let mut interval =
1088 tokio::time::interval(tokio::time::Duration::from_secs(poll_secs));
1089 interval.tick().await; loop {
1092 tokio::select! {
1093 _ = interval.tick() => {
1094 let _guard = poll_update_lock.lock().await;
1095
1096 match http.request_symbol_statuses(us).await {
1097 Ok(statuses) => {
1098 let ts = clock.get_time_ns();
1099 let inst_guard = poll_instruments.load();
1100 let new_statuses = statuses
1101 .into_iter()
1102 .filter(|(instrument_id, _)| {
1103 inst_guard.contains_key(instrument_id)
1104 })
1105 .collect();
1106 drop(inst_guard);
1107
1108 let mut cache =
1109 (**poll_status_cache.load()).clone();
1110 diff_and_emit_statuses(
1111 &new_statuses, &mut cache, &poll_sender, ts, ts,
1112 );
1113 poll_status_cache.store(cache);
1114 }
1115 Err(e) => {
1116 log::warn!("Instrument status poll failed: {e}");
1117 }
1118 }
1119 }
1120 () = poll_cancel.cancelled() => {
1121 log::debug!("Instrument status polling task cancelled");
1122 break;
1123 }
1124 }
1125 }
1126 };
1127 self.session_tasks
1128 .spawn(future)
1129 .context("failed to register Binance Spot status polling task")?;
1130 log::debug!("Instrument status polling started: interval={poll_secs}s");
1131 }
1132
1133 let refresh_secs = self.config.instrument_refresh_interval_secs;
1134 if refresh_secs > 0 {
1135 let http = self.http_client.clone();
1136 let provider = self.config.instrument_provider.clone();
1137 let us = self.config.us;
1138 let instruments = self.instruments.clone();
1139 let statuses = self.status_cache.clone();
1140 let update_lock = self.instrument_update_lock.clone();
1141 let ws = self.ws_client.clone();
1142 let sender = self.data_sender.clone();
1143 let clock = self.clock;
1144 let cancel = self.cancellation_token.clone();
1145
1146 let future = async move {
1147 let mut interval = tokio::time::interval(Duration::from_secs(refresh_secs));
1148 interval.tick().await;
1149
1150 loop {
1151 tokio::select! {
1152 _ = interval.tick() => {
1153 if let Err(e) = Self::refresh_instrument_catalog(
1154 &http,
1155 &provider,
1156 us,
1157 &instruments,
1158 &statuses,
1159 &update_lock,
1160 &ws,
1161 &sender,
1162 clock,
1163 true,
1164 ).await {
1165 log::warn!("Binance Spot instrument refresh failed: {e}");
1166 }
1167 }
1168 () = cancel.cancelled() => {
1169 log::debug!("Binance Spot instrument refresh task cancelled");
1170 break;
1171 }
1172 }
1173 }
1174 };
1175 self.session_tasks
1176 .spawn(future)
1177 .context("failed to register Binance Spot instrument refresh task")?;
1178 log::debug!("Instrument refresh started: interval={refresh_secs}s");
1179 }
1180
1181 Ok::<(), anyhow::Error>(())
1182 }
1183 .await;
1184
1185 if let Err(e) = session_result {
1186 if let Err(teardown_error) = self.teardown_partial_connect().await {
1187 return Err(e.context(format!(
1188 "Binance Spot data startup teardown failed: {teardown_error}"
1189 )));
1190 }
1191 return Err(e);
1192 }
1193
1194 setup_guard.disarm();
1195 self.is_connected.store(true, Ordering::Release);
1196 log::info!("Connected: client_id={}", self.client_id);
1197 Ok(())
1198 }
1199
1200 async fn disconnect(&mut self) -> anyhow::Result<()> {
1201 self.teardown_partial_connect().await?;
1202
1203 self.book_subscriptions.store(AHashMap::new());
1204 self.l1_book_subscriptions.store(AHashMap::new());
1205 self.quote_refs.store(AHashMap::new());
1206 self.ticker_refs.store(AHashMap::new());
1207 self.book_sync.clear();
1208
1209 self.is_connected.store(false, Ordering::Release);
1210 log::info!("Disconnected: client_id={}", self.client_id);
1211 Ok(())
1212 }
1213
1214 fn is_connected(&self) -> bool {
1215 self.is_connected.load(Ordering::Relaxed)
1216 }
1217
1218 fn is_disconnected(&self) -> bool {
1219 !self.is_connected()
1220 }
1221
1222 fn subscribe(&mut self, cmd: SubscribeCustomData) -> anyhow::Result<()> {
1223 if cmd.data_type.type_name() != "BinanceSpotTicker" {
1224 log::warn!(
1225 "Unsupported custom data subscription: {}",
1226 cmd.data_type.type_name()
1227 );
1228 return Ok(());
1229 }
1230 anyhow::ensure!(
1231 self.spot_market_data_mode == BinanceSpotMarketDataMode::Json,
1232 "Binance Spot 24-hour ticker custom data requires JSON market-data mode"
1233 );
1234 let instrument_id = Self::required_instrument_id_metadata(&cmd.data_type)?;
1235 anyhow::ensure!(
1236 instrument_id.venue == self.venue(),
1237 "Spot ticker requires a BINANCE instrument"
1238 );
1239 let should_subscribe = {
1240 let previous = self
1241 .ticker_refs
1242 .load()
1243 .get(&instrument_id)
1244 .copied()
1245 .unwrap_or(0);
1246 self.ticker_refs
1247 .rcu(|refs| *refs.entry(instrument_id).or_insert(0) += 1);
1248 previous == 0
1249 };
1250
1251 if should_subscribe {
1252 let ws = self.ws_client.clone();
1253 let stream = format!("{}@ticker", instrument_id.symbol.as_str().to_lowercase());
1254 self.spawn_ws(
1255 async move {
1256 ws.subscribe(vec![stream])
1257 .await
1258 .context("ticker subscription")
1259 },
1260 "ticker subscription",
1261 );
1262 }
1263 Ok(())
1264 }
1265
1266 fn subscribe_instruments(&mut self, _cmd: SubscribeInstruments) -> anyhow::Result<()> {
1267 log::debug!("subscribe_instruments: Binance instruments are fetched via HTTP on connect");
1268 Ok(())
1269 }
1270
1271 fn subscribe_instrument(&mut self, _cmd: SubscribeInstrument) -> anyhow::Result<()> {
1272 log::debug!("subscribe_instrument: Binance instruments are fetched via HTTP on connect");
1273 Ok(())
1274 }
1275
1276 fn subscribe_book_deltas(&mut self, cmd: SubscribeBookDeltas) -> anyhow::Result<()> {
1277 execute_instrument_subscriptions(&cmd.instrument_ids, |instrument_id| {
1278 let cmd = cmd.with_instrument_ids(instrument_id);
1279 if cmd.book_type == BookType::L1_MBP {
1280 anyhow::ensure!(
1281 cmd.depth.is_none_or(|depth| depth.get() == 1),
1282 "Binance Spot L1_MBP supports depth 1 only"
1283 );
1284 anyhow::ensure!(
1285 !self.book_subscriptions.contains_key(&instrument_id),
1286 "cannot subscribe L1_MBP and L2_MBP for the same Binance Spot instrument"
1287 );
1288 self.l1_book_subscriptions.rcu(|subscriptions| {
1289 *subscriptions.entry(instrument_id).or_insert(0) += 1;
1290 });
1291
1292 self.subscribe_top_of_book(instrument_id);
1293 return Ok(());
1294 }
1295
1296 if cmd.book_type != BookType::L2_MBP {
1297 anyhow::bail!("Binance Spot supports L1_MBP and L2_MBP order book subscriptions");
1298 }
1299
1300 anyhow::ensure!(
1301 !self.l1_book_subscriptions.contains_key(&instrument_id),
1302 "cannot subscribe L1_MBP and L2_MBP for the same Binance Spot instrument"
1303 );
1304
1305 let ws = self.ws_client.clone();
1306 let symbol_lower = instrument_id.symbol.as_str().to_lowercase();
1307 let stream = self.depth_stream(
1308 &symbol_lower,
1309 cmd.depth.map(|d| d.get()),
1310 cmd.params.as_ref(),
1311 )?;
1312
1313 if let Some(existing) = self.book_subscriptions.load().get(&instrument_id) {
1314 anyhow::ensure!(
1315 existing == &stream,
1316 "unsubscribe before changing Binance Spot book depth or update_speed"
1317 );
1318 return Ok(());
1319 }
1320
1321 if self.spot_market_data_mode == BinanceSpotMarketDataMode::Json
1322 && let Some(depth) = cmd.depth
1323 {
1324 let depth = depth.get();
1327
1328 if !BOOK_DEPTHS_JSON.contains(&depth) {
1329 anyhow::bail!(
1330 "Invalid depth {depth} for Binance Spot JSON order book. \
1331 Valid values: {BOOK_DEPTHS_JSON:?}"
1332 );
1333 }
1334
1335 self.book_subscriptions
1336 .insert(instrument_id, stream.clone());
1337 self.spawn_ws(
1338 chain_command(&self.book_command_tail, async move {
1339 ws.subscribe(vec![stream])
1340 .await
1341 .context("book deltas subscription")
1342 }),
1343 "order book subscription",
1344 );
1345
1346 return Ok(());
1347 }
1348
1349 match cmd.depth.map(|d| d.get()) {
1350 Some(depth) => {
1352 anyhow::ensure!(
1353 depth == 20,
1354 "Binance Spot SBE partial books support depth 20 only; use JSON market data for other depths"
1355 );
1356 self.book_subscriptions
1357 .insert(instrument_id, stream.clone());
1358 self.spawn_ws(
1359 chain_command(&self.book_command_tail, async move {
1360 ws.subscribe(vec![stream])
1361 .await
1362 .context("book deltas subscription")
1363 }),
1364 "order book subscription",
1365 );
1366 }
1367 None => {
1369 self.book_subscriptions
1370 .insert(instrument_id, stream.clone());
1371
1372 self.book_sync.subscribe(instrument_id);
1374
1375 self.spawn_ws(
1376 chain_command(&self.book_command_tail, async move {
1377 ws.subscribe(vec![stream])
1378 .await
1379 .context("book deltas subscription")
1380 }),
1381 "order book subscription",
1382 );
1383 }
1384 }
1385
1386 Ok(())
1387 })
1388 }
1389
1390 fn subscribe_quotes(&mut self, cmd: SubscribeQuotes) -> anyhow::Result<()> {
1391 execute_instrument_subscriptions(&cmd.instrument_ids, |instrument_id| {
1392 self.subscribe_top_of_book(instrument_id);
1393 Ok(())
1394 })
1395 }
1396
1397 fn subscribe_trades(&mut self, cmd: SubscribeTrades) -> anyhow::Result<()> {
1398 execute_instrument_subscriptions(&cmd.instrument_ids, |instrument_id| {
1399 let ws = self.ws_client.clone();
1400
1401 let stream = format!("{}@trade", instrument_id.symbol.as_str().to_lowercase());
1402
1403 self.spawn_ws(
1404 async move {
1405 ws.subscribe(vec![stream])
1406 .await
1407 .context("trades subscription")
1408 },
1409 "trade subscription",
1410 );
1411
1412 Ok(())
1413 })
1414 }
1415
1416 fn subscribe_bars(&mut self, cmd: SubscribeBars) -> anyhow::Result<()> {
1417 execute_bar_subscriptions(&cmd.bar_types, |bar_type| {
1418 anyhow::ensure!(
1419 self.spot_market_data_mode == BinanceSpotMarketDataMode::Json,
1420 "Binance Spot kline subscriptions require JSON market-data mode"
1421 );
1422 let ws = self.ws_client.clone();
1423 let interval = bar_spec_to_binance_interval(bar_type.spec())?;
1424
1425 let stream = format!(
1426 "{}@kline_{}",
1427 bar_type.instrument_id().symbol.as_str().to_lowercase(),
1428 interval.as_str()
1429 );
1430
1431 self.spawn_ws(
1432 async move {
1433 ws.subscribe(vec![stream])
1434 .await
1435 .context("bars subscription")
1436 },
1437 "bar subscription",
1438 );
1439
1440 Ok(())
1441 })
1442 }
1443
1444 fn subscribe_instrument_status(
1445 &mut self,
1446 cmd: SubscribeInstrumentStatus,
1447 ) -> anyhow::Result<()> {
1448 execute_instrument_subscriptions(&cmd.instrument_ids, |instrument_id| {
1449 log::debug!(
1450 "subscribe_instrument_status: {instrument_id} (status changes detected via periodic exchange info polling)",
1451 );
1452 Ok(())
1453 })
1454 }
1455
1456 fn unsubscribe_book_deltas(&mut self, cmd: &UnsubscribeBookDeltas) -> anyhow::Result<()> {
1457 execute_instrument_subscriptions(&cmd.instrument_ids, |instrument_id| {
1458 if let Some(count) = self
1459 .l1_book_subscriptions
1460 .load()
1461 .get(&instrument_id)
1462 .copied()
1463 {
1464 if count == 1 {
1465 self.l1_book_subscriptions.remove(&instrument_id);
1466 } else {
1467 self.l1_book_subscriptions.rcu(|subscriptions| {
1468 if let Some(existing) = subscriptions.get_mut(&instrument_id) {
1469 *existing -= 1;
1470 }
1471 });
1472 }
1473
1474 self.unsubscribe_top_of_book(instrument_id);
1475 return Ok(());
1476 }
1477
1478 let ws = self.ws_client.clone();
1479
1480 let stream = self.book_subscriptions.load().get(&instrument_id).cloned();
1482 self.book_subscriptions.remove(&instrument_id);
1483 self.book_sync.remove(instrument_id);
1484
1485 let Some(stream) = stream else {
1486 return Ok(());
1487 };
1488
1489 self.spawn_ws(
1490 chain_command(&self.book_command_tail, async move {
1491 ws.unsubscribe(vec![stream])
1492 .await
1493 .context("book deltas unsubscribe")
1494 }),
1495 "order book unsubscribe",
1496 );
1497
1498 Ok(())
1499 })
1500 }
1501
1502 fn unsubscribe_quotes(&mut self, cmd: &UnsubscribeQuotes) -> anyhow::Result<()> {
1503 execute_instrument_subscriptions(&cmd.instrument_ids, |instrument_id| {
1504 self.unsubscribe_top_of_book(instrument_id);
1505 Ok(())
1506 })
1507 }
1508
1509 fn unsubscribe(&mut self, cmd: &UnsubscribeCustomData) -> anyhow::Result<()> {
1510 if cmd.data_type.type_name() != "BinanceSpotTicker" {
1511 log::warn!(
1512 "Unsupported custom data unsubscription: {}",
1513 cmd.data_type.type_name()
1514 );
1515 return Ok(());
1516 }
1517 let instrument_id = Self::required_instrument_id_metadata(&cmd.data_type)?;
1518 let should_unsubscribe = match self.ticker_refs.load().get(&instrument_id).copied() {
1519 Some(1) => {
1520 self.ticker_refs.remove(&instrument_id);
1521 true
1522 }
1523 Some(count) if count > 1 => {
1524 self.ticker_refs.rcu(|refs| {
1525 if let Some(existing) = refs.get_mut(&instrument_id) {
1526 *existing -= 1;
1527 }
1528 });
1529 false
1530 }
1531 _ => false,
1532 };
1533
1534 if should_unsubscribe {
1535 let ws = self.ws_client.clone();
1536 let stream = format!("{}@ticker", instrument_id.symbol.as_str().to_lowercase());
1537 self.spawn_ws(
1538 async move {
1539 ws.unsubscribe(vec![stream])
1540 .await
1541 .context("ticker unsubscribe")
1542 },
1543 "ticker unsubscribe",
1544 );
1545 }
1546 Ok(())
1547 }
1548
1549 fn unsubscribe_trades(&mut self, cmd: &UnsubscribeTrades) -> anyhow::Result<()> {
1550 execute_instrument_subscriptions(&cmd.instrument_ids, |instrument_id| {
1551 let ws = self.ws_client.clone();
1552
1553 let stream = format!("{}@trade", instrument_id.symbol.as_str().to_lowercase());
1554
1555 self.spawn_ws(
1556 async move {
1557 ws.unsubscribe(vec![stream])
1558 .await
1559 .context("trades unsubscribe")
1560 },
1561 "trade unsubscribe",
1562 );
1563
1564 Ok(())
1565 })
1566 }
1567
1568 fn unsubscribe_bars(&mut self, cmd: &UnsubscribeBars) -> anyhow::Result<()> {
1569 execute_bar_subscriptions(&cmd.bar_types, |bar_type| {
1570 let ws = self.ws_client.clone();
1571 let interval = bar_spec_to_binance_interval(bar_type.spec())?;
1572
1573 let stream = format!(
1574 "{}@kline_{}",
1575 bar_type.instrument_id().symbol.as_str().to_lowercase(),
1576 interval.as_str()
1577 );
1578
1579 self.spawn_ws(
1580 async move {
1581 ws.unsubscribe(vec![stream])
1582 .await
1583 .context("bars unsubscribe")
1584 },
1585 "bar unsubscribe",
1586 );
1587
1588 Ok(())
1589 })
1590 }
1591
1592 fn unsubscribe_instrument_status(
1593 &mut self,
1594 cmd: &UnsubscribeInstrumentStatus,
1595 ) -> anyhow::Result<()> {
1596 execute_instrument_subscriptions(&cmd.instrument_ids, |instrument_id| {
1597 log::debug!("unsubscribe_instrument_status: {instrument_id}",);
1598 Ok(())
1599 })
1600 }
1601
1602 fn request_instruments(&self, request: RequestInstruments) -> anyhow::Result<()> {
1603 let http = self.http_client.clone();
1604 let sender = self.data_sender.clone();
1605 let instruments_cache = self.instruments.clone();
1606 let request_id = request.request_id;
1607 let client_id = request.client_id.unwrap_or(self.client_id);
1608 let venue = self.venue();
1609 let start = request.start;
1610 let end = request.end;
1611 let params = request.params;
1612 let clock = self.clock;
1613 let provider = self.config.instrument_provider.clone();
1614 let us = self.config.us;
1615 let start_nanos = datetime_to_unix_nanos(start);
1616 let end_nanos = datetime_to_unix_nanos(end);
1617
1618 self.spawn_command(async move {
1619 match http.request_instruments_with_config(&provider, us).await {
1620 Ok(instruments) => {
1621 upsert_instruments(&instruments_cache, &instruments);
1622
1623 let response = DataResponse::Instruments(InstrumentsResponse::new(
1624 request_id,
1625 client_id,
1626 venue,
1627 instruments,
1628 start_nanos,
1629 end_nanos,
1630 clock.get_time_ns(),
1631 params,
1632 ));
1633
1634 if let Err(e) = sender.send(DataEvent::Response(response)) {
1635 log::error!("Failed to send instruments response: {e}");
1636 }
1637 }
1638 Err(e) => log::error!("Instruments request failed: {e:?}"),
1639 }
1640 });
1641
1642 Ok(())
1643 }
1644
1645 fn request_instrument(&self, request: RequestInstrument) -> anyhow::Result<()> {
1646 let instrument_id = request.single_instrument_id()?;
1647
1648 let http = self.http_client.clone();
1649 let sender = self.data_sender.clone();
1650 let instruments = self.instruments.clone();
1651 let request_id = request.request_id;
1652 let client_id = request.client_id.unwrap_or(self.client_id);
1653 let start = request.start;
1654 let end = request.end;
1655 let params = request.params;
1656 let clock = self.clock;
1657 let provider = self.config.instrument_provider.clone();
1658 let us = self.config.us;
1659 let start_nanos = datetime_to_unix_nanos(start);
1660 let end_nanos = datetime_to_unix_nanos(end);
1661
1662 self.spawn_command(async move {
1663 match http.request_instruments_with_config(&provider, us).await {
1664 Ok(all_instruments) => {
1665 upsert_instruments(&instruments, &all_instruments);
1666
1667 let instrument = all_instruments
1668 .into_iter()
1669 .find(|i| i.id() == instrument_id);
1670
1671 if let Some(instrument) = instrument {
1672 let response = DataResponse::Instrument(Box::new(InstrumentResponse::new(
1673 request_id,
1674 client_id,
1675 instrument.id(),
1676 instrument,
1677 start_nanos,
1678 end_nanos,
1679 clock.get_time_ns(),
1680 params,
1681 )));
1682
1683 if let Err(e) = sender.send(DataEvent::Response(response)) {
1684 log::error!("Failed to send instrument response: {e}");
1685 }
1686 } else {
1687 log::error!("Instrument not found: {instrument_id}");
1688 }
1689 }
1690 Err(e) => log::error!("Instrument request failed: {e:?}"),
1691 }
1692 });
1693
1694 Ok(())
1695 }
1696
1697 fn request_data(&self, request: RequestCustomData) -> anyhow::Result<()> {
1698 if request.data_type.type_name() != "BinanceBar" {
1699 log::warn!(
1700 "Unsupported custom data request: {}",
1701 request.data_type.type_name()
1702 );
1703 return Ok(());
1704 }
1705 let bar_type = parse_binance_bar_type(&request.data_type)?;
1706 anyhow::ensure!(
1707 bar_type.aggregation_source() == AggregationSource::External,
1708 "historical BinanceBar requests require EXTERNAL aggregation"
1709 );
1710 anyhow::ensure!(
1711 bar_type.spec().price_type == PriceType::Last,
1712 "historical BinanceBar requests require LAST price type"
1713 );
1714 anyhow::ensure!(
1715 bar_type.spec().is_time_aggregated(),
1716 "historical BinanceBar requests require time aggregation"
1717 );
1718 let http = self.http_client.clone();
1719 let sender = self.data_sender.clone();
1720 let request_id = request.request_id;
1721 let client_id = request.client_id;
1722 let data_type = request.data_type;
1723 let start = request.start;
1724 let end = request.end;
1725 let limit = request.limit.map(|value| value.get() as u32);
1726 let params = request.params;
1727 let clock = self.clock;
1728 let venue = self.venue();
1729 let start_nanos = datetime_to_unix_nanos(start);
1730 let end_nanos = datetime_to_unix_nanos(end);
1731
1732 self.spawn_command(async move {
1733 match http.request_binance_bars(bar_type, start, end, limit).await {
1734 Ok(bars) => {
1735 let response = DataResponse::Data(CustomDataResponse::new(
1736 request_id,
1737 client_id,
1738 Some(venue),
1739 data_type,
1740 binance_bars_to_custom_data(bar_type, bars),
1741 start_nanos,
1742 end_nanos,
1743 clock.get_time_ns(),
1744 params,
1745 ));
1746
1747 if let Err(e) = sender.send(DataEvent::Response(response)) {
1748 log::error!("Failed to send BinanceBar response: {e}");
1749 }
1750 }
1751 Err(e) => log::error!("BinanceBar request failed for {bar_type}: {e:?}"),
1752 }
1753 });
1754 Ok(())
1755 }
1756
1757 fn request_trades(&self, request: RequestTrades) -> anyhow::Result<()> {
1758 let instrument_id = request.single_instrument_id()?;
1759
1760 let http = self.http_client.clone();
1761 let sender = self.data_sender.clone();
1762 let limit = request.limit.map(|n| n.get() as u32);
1763 let request_id = request.request_id;
1764 let client_id = request.client_id.unwrap_or(self.client_id);
1765 let params = request.params;
1766 let clock = self.clock;
1767 let start_nanos = datetime_to_unix_nanos(request.start);
1768 let end_nanos = datetime_to_unix_nanos(request.end);
1769 let start = request.start;
1770 let end = request.end;
1771 anyhow::ensure!(
1772 limit.is_none_or(|value| value <= 1000),
1773 "Binance Spot trade limit must not exceed 1000"
1774 );
1775
1776 self.spawn_command(async move {
1777 let result = if start.is_some() || end.is_some() {
1778 http.request_agg_trades(instrument_id, start, end, limit)
1779 .await
1780 } else {
1781 http.request_trades(instrument_id, limit).await
1782 };
1783
1784 match result.context("failed to request trades from Binance") {
1785 Ok(trades) => {
1786 let response = DataResponse::Trades(TradesResponse::new(
1787 request_id,
1788 client_id,
1789 instrument_id,
1790 trades,
1791 start_nanos,
1792 end_nanos,
1793 clock.get_time_ns(),
1794 params,
1795 ));
1796
1797 if let Err(e) = sender.send(DataEvent::Response(response)) {
1798 log::error!("Failed to send trades response: {e}");
1799 }
1800 }
1801 Err(e) => log::error!("Trade request failed: {e:?}"),
1802 }
1803 });
1804
1805 Ok(())
1806 }
1807
1808 fn request_bars(&self, request: RequestBars) -> anyhow::Result<()> {
1809 let bar_type = request.single_bar_type()?;
1810
1811 let http = self.http_client.clone();
1812 let sender = self.data_sender.clone();
1813 let start = request.start;
1814 let end = request.end;
1815 let limit = request.limit.map(|n| n.get() as u32);
1816 let request_id = request.request_id;
1817 let client_id = request.client_id.unwrap_or(self.client_id);
1818 let params = request.params;
1819 let clock = self.clock;
1820 let start_nanos = datetime_to_unix_nanos(start);
1821 let end_nanos = datetime_to_unix_nanos(end);
1822 anyhow::ensure!(
1823 bar_type.aggregation_source() == AggregationSource::External,
1824 "Binance historical bars require EXTERNAL aggregation"
1825 );
1826 anyhow::ensure!(
1827 bar_type.spec().price_type == PriceType::Last,
1828 "Binance historical bars require LAST price type"
1829 );
1830 anyhow::ensure!(
1831 bar_type.spec().is_time_aggregated(),
1832 "Binance historical bars require time aggregation"
1833 );
1834
1835 self.spawn_command(async move {
1836 let result = http.request_bars(bar_type, start, end, limit).await;
1837
1838 match result.context("failed to request bars from Binance") {
1839 Ok(bars) => {
1840 let response = DataResponse::Bars(BarsResponse::new(
1841 request_id,
1842 client_id,
1843 bar_type,
1844 bars,
1845 start_nanos,
1846 end_nanos,
1847 clock.get_time_ns(),
1848 params,
1849 ));
1850
1851 if let Err(e) = sender.send(DataEvent::Response(response)) {
1852 log::error!("Failed to send bars response: {e}");
1853 }
1854 }
1855 Err(e) => log::error!("Bar request failed: {e:?}"),
1856 }
1857 });
1858
1859 Ok(())
1860 }
1861
1862 fn request_book_snapshot(&self, request: RequestBookSnapshot) -> anyhow::Result<()> {
1863 let instrument_id = request.single_instrument_id()?;
1864
1865 let depth = request.depth.map(|value| value.get() as u32);
1866 anyhow::ensure!(
1867 depth.is_none_or(|value| (1..=5000).contains(&value)),
1868 "Binance Spot order-book depth must be between 1 and 5000"
1869 );
1870 let http = self.http_client.clone();
1871 let book_sync = self.book_sync.clone();
1872 let sender = self.data_sender.clone();
1873 let request_id = request.request_id;
1874 let client_id = request.client_id.unwrap_or(self.client_id);
1875 let params = request.params;
1876 let clock = self.clock;
1877
1878 self.spawn_command(async move {
1879 let weight = depth_request_weight(depth.unwrap_or(100));
1881 book_sync.pacer().acquire(weight).await;
1882
1883 match http.request_book_snapshot(instrument_id, depth).await {
1884 Ok(book) => {
1885 let response = DataResponse::Book(BookResponse::new(
1886 request_id,
1887 client_id,
1888 instrument_id,
1889 book,
1890 None,
1891 None,
1892 clock.get_time_ns(),
1893 params,
1894 ));
1895
1896 if let Err(e) = sender.send(DataEvent::Response(response)) {
1897 log::error!("Failed to send book snapshot response: {e}");
1898 }
1899 }
1900 Err(e) => log::error!("Book snapshot request failed for {instrument_id}: {e:?}"),
1901 }
1902 });
1903 Ok(())
1904 }
1905}
1906
1907impl BinanceSpotDataClient {
1908 fn depth_stream(
1909 &self,
1910 symbol: &str,
1911 depth: Option<usize>,
1912 params: Option<&Params>,
1913 ) -> anyhow::Result<String> {
1914 let mut stream = match depth {
1915 Some(depth) => format!("{symbol}@depth{depth}"),
1916 None => format!("{symbol}@depth"),
1917 };
1918
1919 if self.spot_market_data_mode == BinanceSpotMarketDataMode::Json {
1920 let speed = params.and_then(|p| p.get("update_speed"));
1921 if let Some(speed) = speed {
1922 anyhow::ensure!(
1923 matches!(speed.as_u64(), Some(100 | 1000)),
1924 "Binance Spot JSON update_speed must be 100 or 1000, was {speed}"
1925 );
1926 }
1927
1928 if speed.is_none_or(|speed| speed.as_u64() == Some(100)) {
1929 stream.push_str("@100ms");
1930 }
1931 }
1932
1933 Ok(stream)
1934 }
1935
1936 fn subscribe_top_of_book(&self, instrument_id: InstrumentId) {
1937 let should_subscribe = {
1938 let previous = self
1939 .quote_refs
1940 .load()
1941 .get(&instrument_id)
1942 .copied()
1943 .unwrap_or(0);
1944 self.quote_refs.rcu(|refs| {
1945 *refs.entry(instrument_id).or_insert(0) += 1;
1946 });
1947 previous == 0
1948 };
1949
1950 if should_subscribe {
1951 let ws = self.ws_client.clone();
1952 let suffix = self.quote_stream_suffix();
1953 let stream = format!("{}@{suffix}", instrument_id.symbol.as_str().to_lowercase());
1954 self.spawn_ws(
1955 async move {
1956 ws.subscribe(vec![stream])
1957 .await
1958 .context("top-of-book subscription")
1959 },
1960 "top-of-book subscription",
1961 );
1962 }
1963 }
1964
1965 fn unsubscribe_top_of_book(&self, instrument_id: InstrumentId) {
1966 let should_unsubscribe = match self.quote_refs.load().get(&instrument_id).copied() {
1967 Some(1) => {
1968 self.quote_refs.remove(&instrument_id);
1969 true
1970 }
1971 Some(count) if count > 1 => {
1972 self.quote_refs.rcu(|refs| {
1973 if let Some(existing) = refs.get_mut(&instrument_id) {
1974 *existing -= 1;
1975 }
1976 });
1977 false
1978 }
1979 _ => false,
1980 };
1981
1982 if should_unsubscribe {
1983 let ws = self.ws_client.clone();
1984 let suffix = self.quote_stream_suffix();
1985 let stream = format!("{}@{suffix}", instrument_id.symbol.as_str().to_lowercase());
1986 self.spawn_ws(
1987 async move {
1988 ws.unsubscribe(vec![stream])
1989 .await
1990 .context("top-of-book unsubscribe")
1991 },
1992 "top-of-book unsubscribe",
1993 );
1994 }
1995 }
1996}
1997
1998#[derive(Debug, Clone)]
1999enum SpotWsClient {
2000 Sbe(BinanceSpotWebSocketClient),
2001 JsonPublic(BinanceSpotPublicJsonWebSocketClient),
2002}
2003
2004impl SpotWsClient {
2005 fn has_credentials(&self) -> bool {
2006 match self {
2007 Self::Sbe(client) => client.has_credentials(),
2008 Self::JsonPublic(_) => true, }
2010 }
2011
2012 fn replace_instruments(&self, instruments: &[InstrumentAny]) {
2013 match self {
2014 Self::Sbe(client) => client.replace_instruments(instruments),
2015 Self::JsonPublic(client) => client.replace_instruments(instruments),
2016 }
2017 }
2018
2019 async fn subscribe(&self, streams: Vec<String>) -> anyhow::Result<()> {
2020 match self {
2021 Self::Sbe(client) => client.subscribe(streams).await.map_err(Into::into),
2022 Self::JsonPublic(client) => client.subscribe(streams).await,
2023 }
2024 }
2025
2026 async fn unsubscribe(&self, streams: Vec<String>) -> anyhow::Result<()> {
2027 match self {
2028 Self::Sbe(client) => client.unsubscribe(streams).await.map_err(Into::into),
2029 Self::JsonPublic(client) => client.unsubscribe(streams).await,
2030 }
2031 }
2032
2033 async fn close(&mut self) -> anyhow::Result<()> {
2034 match self {
2035 Self::Sbe(client) => client.close().await.map_err(Into::into),
2036 Self::JsonPublic(client) => client.close().await,
2037 }
2038 }
2039
2040 fn begin_shutdown(&self) {
2041 match self {
2042 Self::Sbe(client) => client.begin_shutdown(),
2043 Self::JsonPublic(client) => client.begin_shutdown(),
2044 }
2045 }
2046}
2047
2048fn resolve_spot_json_ws_url(
2049 base_url_ws: Option<String>,
2050 environment: BinanceEnvironment,
2051 us: bool,
2052) -> String {
2053 let default_url =
2054 get_ws_base_url_with_us(BinanceProductType::Spot, environment, us).to_string();
2055
2056 match base_url_ws {
2057 Some(url) if looks_like_spot_sbe_ws_url(&url) => {
2058 log::warn!(
2059 "Spot JSON market-data mode received an SBE WebSocket URL override (`{url}`); \
2060 using Spot JSON WebSocket default for {environment:?}: {default_url}",
2061 );
2062 default_url
2063 }
2064 Some(url) => url,
2065 None => default_url,
2066 }
2067}
2068
2069fn looks_like_spot_sbe_ws_url(base_url: &str) -> bool {
2070 let without_scheme = base_url
2071 .split_once("://")
2072 .map_or(base_url, |(_, rest)| rest);
2073 let host = without_scheme
2074 .split(['/', ':'])
2075 .next()
2076 .unwrap_or(without_scheme);
2077 host.starts_with("stream-sbe") || host.starts_with("demo-stream-sbe")
2078}
2079
2080#[cfg(test)]
2081mod tests {
2082 use std::{num::NonZeroU32, sync::Arc, time::Duration};
2083
2084 use nautilus_common::{
2085 live::{runner::replace_data_event_sender, sender::EventSender},
2086 messages::DataEvent,
2087 testing::wait_until_async,
2088 };
2089 use nautilus_core::{AtomicMap, nanos::UnixNanos, time::AtomicTime};
2090 use nautilus_live::task::TaskGroup;
2091 use nautilus_model::{
2092 data::Data,
2093 enums::{BookAction, RecordFlag},
2094 identifiers::InstrumentId,
2095 instruments::{Instrument, InstrumentAny, stubs::currency_pair_btcusdt},
2096 };
2097 use parking_lot::Mutex;
2098 use rstest::rstest;
2099 use rust_decimal_macros::dec;
2100 use ustr::Ustr;
2101
2102 use super::{
2103 BinanceBookError, BinanceDepth, BinanceEnvironment, BinanceSpotDataClient,
2104 BinanceSpotHttpError, BinanceSpotMarketDataMode, depth_request_weight,
2105 depth_snapshot_error, parse_spot_depth_snapshot, resolve_spot_json_ws_url,
2106 };
2107 use crate::{
2108 book::{
2109 pacing::SnapshotPacer,
2110 sync::{BookSyncTracker, DepthSequencing},
2111 },
2112 common::{
2113 consts::{BINANCE_CLIENT_ID, BINANCE_SPOT_WS_URL},
2114 websocket::chain_command,
2115 },
2116 config::BinanceDataClientConfig,
2117 spot::{
2118 http::{BinancePriceLevel, BinanceSpotHttpClient},
2119 sbe::stream::{
2120 BestBidAskStreamEvent, DepthDiffStreamEvent, DepthSnapshotStreamEvent, PriceLevel,
2121 Trade, TradesStreamEvent,
2122 },
2123 websocket::streams::messages::BinanceSpotWsMessage,
2124 },
2125 };
2126
2127 #[rstest]
2128 #[case::trades("trades")]
2129 #[case::bbo("bbo")]
2130 #[case::snapshot("snapshot")]
2131 #[case::diff("diff")]
2132 fn handle_ws_message_rejects_invalid_sbe_then_processes_valid_bbo(#[case] kind: &str) {
2133 let ts_init = UnixNanos::from(1_800_000_000_000_000_000u64);
2134 let clock = Box::leak(Box::new(AtomicTime::new(false, ts_init)));
2135 let instrument = InstrumentAny::CurrencyPair(currency_pair_btcusdt());
2136 let instruments = Arc::new(AtomicMap::new());
2137 instruments.insert(instrument.id(), instrument.clone());
2138 let ws_instruments = Arc::new(AtomicMap::new());
2139 ws_instruments.insert(Ustr::from("BTCUSDT"), instrument);
2140 let l1_book_subscriptions = Arc::new(AtomicMap::<InstrumentId, u32>::new());
2141 let http_client = BinanceSpotHttpClient::new(
2142 BinanceEnvironment::Testnet,
2143 clock,
2144 None,
2145 None,
2146 None,
2147 None,
2148 None,
2149 None,
2150 )
2151 .unwrap();
2152 let (sender, mut receiver) = tokio::sync::mpsc::unbounded_channel();
2153 let sender = EventSender::from(sender);
2154 let pacer = SnapshotPacer::new(NonZeroU32::new(60_000).unwrap());
2155 let book_sync =
2156 BookSyncTracker::new(DepthSequencing::Spot, sender.clone(), Arc::new(pacer));
2157 let event_time_us = 1_700_000_000_000_000;
2158 let message = BinanceSpotWsMessage::BestBidAsk(BestBidAskStreamEvent {
2159 event_time_us,
2160 book_update_id: 123,
2161 price_exponent: -2,
2162 qty_exponent: -4,
2163 bid_price_mantissa: 12_345,
2164 bid_qty_mantissa: 25_000,
2165 ask_price_mantissa: 12_350,
2166 ask_qty_mantissa: 30_000,
2167 symbol: Ustr::from("BTCUSDT"),
2168 });
2169 let command_tasks = TaskGroup::new();
2170 let command_spawner = command_tasks.spawner().unwrap();
2171
2172 let invalid_level = PriceLevel {
2173 price_mantissa: i64::MAX,
2174 qty_mantissa: 1,
2175 };
2176
2177 let invalid = match kind {
2178 "trades" => BinanceSpotWsMessage::Trades(TradesStreamEvent {
2179 event_time_us,
2180 transact_time_us: event_time_us,
2181 price_exponent: 0,
2182 qty_exponent: 0,
2183 trades: vec![
2184 Trade {
2185 id: 1,
2186 price_mantissa: 1,
2187 qty_mantissa: 1,
2188 is_buyer_maker: false,
2189 },
2190 Trade {
2191 id: 2,
2192 price_mantissa: i64::MAX,
2193 qty_mantissa: 1,
2194 is_buyer_maker: true,
2195 },
2196 ],
2197 symbol: Ustr::from("BTCUSDT"),
2198 }),
2199 "bbo" => BinanceSpotWsMessage::BestBidAsk(BestBidAskStreamEvent {
2200 event_time_us,
2201 book_update_id: 122,
2202 price_exponent: 0,
2203 qty_exponent: 0,
2204 bid_price_mantissa: i64::MAX,
2205 bid_qty_mantissa: 1,
2206 ask_price_mantissa: 1,
2207 ask_qty_mantissa: 1,
2208 symbol: Ustr::from("BTCUSDT"),
2209 }),
2210 "snapshot" => BinanceSpotWsMessage::DepthSnapshot(DepthSnapshotStreamEvent {
2211 event_time_us,
2212 book_update_id: 122,
2213 price_exponent: 0,
2214 qty_exponent: 0,
2215 bids: vec![invalid_level],
2216 asks: vec![],
2217 symbol: Ustr::from("BTCUSDT"),
2218 }),
2219 "diff" => BinanceSpotWsMessage::DepthDiff(DepthDiffStreamEvent {
2220 event_time_us,
2221 first_book_update_id: 121,
2222 last_book_update_id: 122,
2223 price_exponent: 0,
2224 qty_exponent: 0,
2225 bids: vec![invalid_level],
2226 asks: vec![],
2227 symbol: Ustr::from("BTCUSDT"),
2228 }),
2229 _ => unreachable!(),
2230 };
2231
2232 BinanceSpotDataClient::handle_ws_message(
2233 invalid,
2234 &sender,
2235 &instruments,
2236 &ws_instruments,
2237 &book_sync,
2238 &l1_book_subscriptions,
2239 &http_client,
2240 Duration::from_secs(10),
2241 clock,
2242 &command_spawner,
2243 );
2244 assert!(matches!(
2245 receiver.try_recv(),
2246 Err(tokio::sync::mpsc::error::TryRecvError::Empty)
2247 ));
2248
2249 BinanceSpotDataClient::handle_ws_message(
2250 message,
2251 &sender,
2252 &instruments,
2253 &ws_instruments,
2254 &book_sync,
2255 &l1_book_subscriptions,
2256 &http_client,
2257 Duration::from_secs(10),
2258 clock,
2259 &command_spawner,
2260 );
2261
2262 let DataEvent::Data(Data::Quote(quote)) = receiver.try_recv().unwrap() else {
2263 panic!("expected quote data event");
2264 };
2265
2266 assert_eq!(quote.instrument_id, InstrumentId::from("BTCUSDT.BINANCE"));
2267 assert_eq!(quote.bid_price.as_decimal(), dec!(123.45));
2268 assert_eq!(quote.ask_price.as_decimal(), dec!(123.50));
2269 assert_eq!(quote.bid_size.as_decimal(), dec!(2.5));
2270 assert_eq!(quote.ask_size.as_decimal(), dec!(3.0));
2271 assert_eq!(quote.ts_event, UnixNanos::from_micros(event_time_us as u64));
2272 assert_eq!(quote.ts_init, ts_init);
2273 }
2274
2275 #[rstest]
2276 #[tokio::test]
2277 async fn chained_book_commands_run_in_issue_order() {
2278 let (sender, _receiver) = tokio::sync::mpsc::unbounded_channel();
2279 replace_data_event_sender(sender);
2280
2281 let config = BinanceDataClientConfig {
2282 spot_market_data_mode: BinanceSpotMarketDataMode::Json,
2283 ..Default::default()
2284 };
2285
2286 let client = BinanceSpotDataClient::new(*BINANCE_CLIENT_ID, config).unwrap();
2287 let order = Arc::new(Mutex::new(Vec::new()));
2288 let unsubscribe = Arc::clone(&order);
2289 let subscribe = Arc::clone(&order);
2290
2291 client.spawn_ws(
2292 chain_command(&client.book_command_tail, async move {
2293 tokio::time::sleep(Duration::from_millis(50)).await;
2294 unsubscribe.lock().push("unsubscribe");
2295 Ok(())
2296 }),
2297 "order book unsubscribe",
2298 );
2299
2300 client.spawn_ws(
2301 chain_command(&client.book_command_tail, async move {
2302 subscribe.lock().push("subscribe");
2303 Ok(())
2304 }),
2305 "order book subscription",
2306 );
2307
2308 wait_until_async(
2309 || {
2310 let finished = order.lock().len() == 2;
2311 async move { finished }
2312 },
2313 Duration::from_secs(5),
2314 )
2315 .await;
2316
2317 assert_eq!(*order.lock(), vec!["unsubscribe", "subscribe"]);
2318 }
2319
2320 #[rstest]
2321 fn parse_spot_depth_snapshot_sets_sequence_and_last_flag() {
2322 let instrument_id = InstrumentId::from("BTCUSDT.BINANCE");
2323 let depth = depth_snapshot(
2324 vec![price_level(10_000, 1_000)],
2325 vec![price_level(10_100, 2_000)],
2326 );
2327
2328 let deltas =
2329 parse_spot_depth_snapshot(&depth, instrument_id, 2, 3, UnixNanos::from(2)).unwrap();
2330
2331 assert_eq!(deltas.deltas.len(), 3);
2332 assert_eq!(deltas.deltas[0].sequence, 123);
2333 assert_eq!(deltas.deltas[1].sequence, 123);
2334 assert_eq!(deltas.deltas[2].sequence, 123);
2335 assert_eq!(deltas.ts_event, UnixNanos::from(2));
2336 assert_eq!(deltas.ts_init, UnixNanos::from(2));
2337 assert_eq!(deltas.deltas[1].order.price.as_decimal(), dec!(100.00));
2338 assert_eq!(deltas.deltas[1].order.size.as_decimal(), dec!(1.000));
2339 assert_eq!(deltas.deltas[1].flags, RecordFlag::F_SNAPSHOT as u8);
2340 assert_eq!(
2341 deltas.deltas[2].flags,
2342 RecordFlag::F_SNAPSHOT as u8 | RecordFlag::F_LAST as u8
2343 );
2344 }
2345
2346 #[rstest]
2347 fn parse_spot_depth_snapshot_sets_last_flag_for_bid_only_snapshot() {
2348 let instrument_id = InstrumentId::from("BTCUSDT.BINANCE");
2349 let depth = depth_snapshot(vec![price_level(10_000, 1_000)], vec![]);
2350
2351 let deltas =
2352 parse_spot_depth_snapshot(&depth, instrument_id, 2, 3, UnixNanos::from(2)).unwrap();
2353
2354 assert_eq!(deltas.deltas.len(), 2);
2355 assert_eq!(
2356 deltas.deltas[1].flags,
2357 RecordFlag::F_SNAPSHOT as u8 | RecordFlag::F_LAST as u8
2358 );
2359 }
2360
2361 #[rstest]
2362 fn parse_spot_depth_snapshot_returns_clear_only_for_empty_book() {
2363 let instrument_id = InstrumentId::from("BTCUSDT.BINANCE");
2364 let depth = depth_snapshot(vec![], vec![]);
2365
2366 let deltas =
2367 parse_spot_depth_snapshot(&depth, instrument_id, 2, 3, UnixNanos::from(2)).unwrap();
2368
2369 assert_eq!(deltas.deltas.len(), 1);
2370 assert_eq!(deltas.deltas[0].action, BookAction::Clear);
2371 assert_eq!(deltas.deltas[0].sequence, 123);
2372 assert_eq!(
2373 deltas.deltas[0].flags,
2374 RecordFlag::F_SNAPSHOT as u8 | RecordFlag::F_LAST as u8
2375 );
2376 }
2377
2378 #[rstest]
2379 fn parse_spot_depth_snapshot_rejects_out_of_range_price() {
2380 let instrument_id = InstrumentId::from("BTCUSDT.BINANCE");
2381 let depth = BinanceDepth {
2382 last_update_id: 123,
2383 price_exponent: 100,
2384 qty_exponent: -3,
2385 bids: vec![price_level(i64::MAX, 1_000)],
2386 asks: vec![],
2387 };
2388
2389 let result = parse_spot_depth_snapshot(&depth, instrument_id, 2, 3, UnixNanos::from(2));
2390
2391 assert!(result.is_err());
2392 }
2393
2394 #[rstest]
2395 fn parse_spot_depth_snapshot_rejects_out_of_range_quantity() {
2396 let instrument_id = InstrumentId::from("BTCUSDT.BINANCE");
2397 let depth = BinanceDepth {
2398 last_update_id: 123,
2399 price_exponent: -2,
2400 qty_exponent: 100,
2401 bids: vec![price_level(10_000, i64::MAX)],
2402 asks: vec![],
2403 };
2404
2405 let result = parse_spot_depth_snapshot(&depth, instrument_id, 2, 3, UnixNanos::from(2));
2406
2407 assert!(result.is_err());
2408 }
2409
2410 fn depth_snapshot(bids: Vec<BinancePriceLevel>, asks: Vec<BinancePriceLevel>) -> BinanceDepth {
2411 BinanceDepth {
2412 last_update_id: 123,
2413 price_exponent: -2,
2414 qty_exponent: -3,
2415 bids,
2416 asks,
2417 }
2418 }
2419
2420 fn price_level(price_mantissa: i64, qty_mantissa: i64) -> BinancePriceLevel {
2421 BinancePriceLevel {
2422 price_mantissa,
2423 qty_mantissa,
2424 }
2425 }
2426
2427 #[rstest]
2428 fn test_spot_market_data_mode_default_is_sbe() {
2429 assert_eq!(
2430 BinanceSpotMarketDataMode::default(),
2431 BinanceSpotMarketDataMode::Sbe
2432 );
2433 }
2434
2435 #[rstest]
2436 fn test_resolve_spot_json_ws_url_uses_environment_default_without_override() {
2437 assert_eq!(
2438 resolve_spot_json_ws_url(None, BinanceEnvironment::Live, false),
2439 BINANCE_SPOT_WS_URL.to_string()
2440 );
2441 }
2442
2443 #[rstest]
2444 fn test_resolve_spot_json_ws_url_rewrites_sbe_override_to_spot_default() {
2445 assert_eq!(
2446 resolve_spot_json_ws_url(
2447 Some("wss://stream-sbe.binance.com/ws".to_string()),
2448 BinanceEnvironment::Live,
2449 false,
2450 ),
2451 BINANCE_SPOT_WS_URL.to_string()
2452 );
2453 }
2454
2455 #[rstest]
2456 fn test_resolve_spot_json_ws_url_preserves_non_sbe_override() {
2457 let custom = "wss://example.com/ws".to_string();
2458 assert_eq!(
2459 resolve_spot_json_ws_url(Some(custom.clone()), BinanceEnvironment::Live, false),
2460 custom
2461 );
2462 }
2463
2464 #[rstest]
2465 fn test_resolve_spot_json_ws_url_uses_binance_us_default() {
2466 assert_eq!(
2467 resolve_spot_json_ws_url(None, BinanceEnvironment::Live, true),
2468 "wss://stream.binance.us:9443/ws"
2469 );
2470 }
2471
2472 #[rstest]
2473 #[case::smallest(1, 5)]
2474 #[case::tier_one_max(100, 5)]
2475 #[case::tier_two_min(101, 25)]
2476 #[case::tier_two_max(500, 25)]
2477 #[case::tier_three_min(501, 50)]
2478 #[case::tier_three_max(1000, 50)]
2479 #[case::tier_four_min(1001, 250)]
2480 #[case::largest(5000, 250)]
2481 fn depth_request_weight_follows_venue_tiers(#[case] limit: u32, #[case] expected: u32) {
2482 assert_eq!(depth_request_weight(limit), expected);
2483 }
2484
2485 #[rstest]
2486 #[case::network(BinanceSpotHttpError::NetworkError("connection reset".to_string()), true)]
2487 #[case::server_error(
2488 BinanceSpotHttpError::UnexpectedStatus {
2489 status: 503,
2490 body: String::new(),
2491 retry_after: None,
2492 },
2493 true
2494 )]
2495 #[case::rate_limited(
2496 BinanceSpotHttpError::BinanceError {
2497 code: -1003,
2498 message: "Too many requests".to_string(),
2499 status: 429,
2500 retry_after: None,
2501 },
2502 true
2503 )]
2504 #[case::retry_budget_exceeded(
2505 BinanceSpotHttpError::RetryBudgetExceeded("elapsed budget exhausted".to_string()),
2506 true
2507 )]
2508 #[case::invalid_symbol(
2509 BinanceSpotHttpError::BinanceError {
2510 code: -1121,
2511 message: "Invalid symbol.".to_string(),
2512 status: 400,
2513 retry_after: None,
2514 },
2515 false
2516 )]
2517 fn depth_snapshot_error_retries_transient_failures(
2518 #[case] error: BinanceSpotHttpError,
2519 #[case] retryable: bool,
2520 ) {
2521 let instrument_id = InstrumentId::from("BTCUSDT.BINANCE");
2522 let message = format!("depth snapshot request for {instrument_id} failed: {error}");
2523
2524 let classified = depth_snapshot_error(instrument_id, &error);
2525
2526 let expected = if retryable {
2527 BinanceBookError::Retryable(message)
2528 } else {
2529 BinanceBookError::Permanent(message)
2530 };
2531
2532 assert_eq!(classified, expected);
2533 }
2534}