1use std::{
19 str::FromStr,
20 sync::{
21 Arc,
22 atomic::{AtomicBool, AtomicU32, Ordering},
23 },
24 time::Duration,
25};
26
27use ahash::AHashMap;
28use anyhow::Context;
29use futures_util::{StreamExt, pin_mut};
30use nautilus_common::{
31 clients::DataClient,
32 live::{runner::get_data_event_sender, sender::EventSender},
33 messages::{
34 DataEvent,
35 data::{
36 BarsResponse, BookResponse, CustomDataResponse, DataResponse, FundingRatesResponse,
37 InstrumentResponse, InstrumentsResponse, RequestBars, RequestBookSnapshot,
38 RequestCustomData, RequestFundingRates, RequestInstrument, RequestInstruments,
39 RequestTrades, SubscribeBars, SubscribeBookDeltas, SubscribeCustomData,
40 SubscribeFundingRates, SubscribeIndexPrices, SubscribeInstrument, SubscribeInstruments,
41 SubscribeMarkPrices, SubscribeQuotes, SubscribeTrades, TradesResponse, UnsubscribeBars,
42 UnsubscribeBookDeltas, UnsubscribeCustomData, UnsubscribeFundingRates,
43 UnsubscribeIndexPrices, UnsubscribeMarkPrices, UnsubscribeQuotes, UnsubscribeTrades,
44 subscribe::SubscribeInstrumentStatus, unsubscribe::UnsubscribeInstrumentStatus,
45 },
46 },
47};
48use nautilus_core::{
49 AtomicMap, Params,
50 datetime::datetime_to_unix_nanos,
51 nanos::UnixNanos,
52 time::{AtomicTime, get_atomic_clock_realtime},
53};
54use nautilus_live::{
55 SocketControlFactory,
56 task::{TaskGroup, TaskGroupGuard, TaskSpawner},
57};
58use nautilus_model::{
59 data::{BookOrder, CustomData, Data, DataType, OrderBookDelta, OrderBookDeltas, QuoteTick},
60 enums::{
61 AggregationSource, BookAction, BookType, MarketStatusAction, OrderSide, PriceType,
62 RecordFlag,
63 },
64 identifiers::{ClientId, InstrumentId, Venue},
65 instruments::{Instrument, InstrumentAny},
66 types::{Price, Quantity},
67};
68use parking_lot::{Mutex, RwLock};
69use rust_decimal::Decimal;
70use tokio_util::sync::CancellationToken;
71use ustr::Ustr;
72
73use crate::{
74 common::{
75 bar::{binance_bar_data_type, binance_bars_to_custom_data, parse_binance_bar_type},
76 consts::{BINANCE_BOOK_DEPTHS, BINANCE_VENUE, BINANCE_WS_HEARTBEAT_SECS},
77 enums::{BinanceEnvironment, BinanceProductType},
78 parse::{
79 bar_spec_to_binance_interval, parse_millis, parse_millis_or_init,
80 parse_price_at_precision, parse_quantity_at_precision,
81 parse_required_price_at_precision, parse_required_quantity_at_precision,
82 quote_to_l1_deltas,
83 },
84 status::diff_and_emit_statuses,
85 symbol::{format_binance_stream_symbol, format_binance_symbol},
86 urls::{get_usdm_ws_route_base_url, get_ws_public_base_url},
87 },
88 config::BinanceDataClientConfig,
89 data_types::{
90 BinanceFuturesLiquidation, BinanceFuturesOpenInterest, BinanceFuturesOpenInterestHist,
91 BinanceFuturesOpenInterestHistPoint, register_binance_custom_data,
92 },
93 futures::{
94 http::{
95 client::{BinanceFuturesHttpClient, BinanceFuturesInstrument},
96 models::BinanceOrderBook,
97 query::{BinanceDepthParams, BinanceOpenInterestHistParams, BinanceOpenInterestParams},
98 },
99 websocket::streams::{
100 client::BinanceFuturesWebSocketClient,
101 messages::BinanceFuturesWsStreamsMessage,
102 parse_data::{
103 parse_agg_trade, parse_book_ticker, parse_depth_snapshot, parse_depth_update,
104 parse_kline, parse_mark_price, parse_ticker, parse_trade,
105 },
106 },
107 },
108};
109
110const MAX_SNAPSHOT_RETRIES: u32 = 5;
111const MAX_BUFFERED_DEPTH_UPDATES: usize = 10_000;
112const SNAPSHOT_RETRY_BACKOFF_BASE_MS: u64 = 250;
113const SNAPSHOT_RETRY_BACKOFF_CAP_MS: u64 = 3_000;
114const MARKET_STREAMS_ENDPOINT: &str = "binance-futures-market-streams";
115const PUBLIC_STREAMS_ENDPOINT: &str = "binance-futures-public-streams";
116
117#[derive(Debug)]
119pub struct BinanceFuturesDataClient {
120 clock: &'static AtomicTime,
121 client_id: ClientId,
122 config: BinanceDataClientConfig,
123 product_type: BinanceProductType,
124 http_client: BinanceFuturesHttpClient,
125 ws_client: BinanceFuturesWebSocketClient,
126 ws_public_client: BinanceFuturesWebSocketClient,
127 data_sender: EventSender<DataEvent>,
128 is_connected: AtomicBool,
129 cancellation_token: CancellationToken,
130 session_tasks: TaskGroup,
131 command_tasks: TaskGroup,
132 shutdown_errors: Vec<String>,
133 instruments: Arc<AtomicMap<InstrumentId, InstrumentAny>>,
134 status_cache: Arc<AtomicMap<InstrumentId, MarketStatusAction>>,
135 book_buffers: Arc<AtomicMap<InstrumentId, BookBuffer>>,
136 book_subscriptions: Arc<AtomicMap<InstrumentId, u32>>,
137 book_unsubscribes_pending: Arc<AtomicMap<InstrumentId, Vec<BookDrain>>>,
138 l1_book_subscriptions: Arc<AtomicMap<InstrumentId, u32>>,
139 quote_refs: Arc<AtomicMap<InstrumentId, u32>>,
140 mark_price_refs: Arc<AtomicMap<InstrumentId, u32>>,
142 ticker_refs: Arc<AtomicMap<InstrumentId, u32>>,
143 force_order_refs: Arc<AtomicMap<InstrumentId, u32>>,
144 force_order_all_market_refs: Arc<AtomicU32>,
145 force_order_all_market_stream_active: Arc<AtomicBool>,
146 force_order_ws_lock: Arc<tokio::sync::Mutex<()>>,
147 book_epoch: Arc<RwLock<u64>>,
148 book_command_tail: Arc<Mutex<Option<tokio::sync::oneshot::Receiver<()>>>>,
149 book_drain_generation: u64,
150}
151
152impl BinanceFuturesDataClient {
153 pub fn new(
160 client_id: ClientId,
161 config: BinanceDataClientConfig,
162 product_type: BinanceProductType,
163 ) -> anyhow::Result<Self> {
164 config.validate()?;
165
166 match product_type {
167 BinanceProductType::UsdM | BinanceProductType::CoinM => {}
168 _ => {
169 anyhow::bail!(
170 "BinanceFuturesDataClient requires UsdM or CoinM product type, was {product_type:?}"
171 );
172 }
173 }
174
175 let clock = get_atomic_clock_realtime();
176 let data_sender = get_data_event_sender();
177 let socket_factory = SocketControlFactory::new(client_id, Some(*BINANCE_VENUE));
178 let api_key = config
179 .api_key
180 .as_ref()
181 .map(|value| value.expose_secret().to_owned());
182 let api_secret = config
183 .api_secret
184 .as_ref()
185 .map(|value| value.expose_secret().to_owned());
186 let proxy_url = config
187 .proxy_url
188 .as_ref()
189 .map(|value| value.expose_secret().to_owned());
190
191 let http_client = BinanceFuturesHttpClient::new(
192 product_type,
193 config.environment,
194 clock,
195 api_key.clone(),
196 api_secret.clone(),
197 config.base_url_http.clone(),
198 Some(config.recv_window_ms),
199 None, proxy_url.clone(),
201 false, )?
203 .with_retry_config(config.retry_config());
204
205 let market_url = config.base_url_ws.clone().map(|url| {
206 if product_type == BinanceProductType::UsdM
207 && config.environment == BinanceEnvironment::Live
208 {
209 get_usdm_ws_route_base_url(&url, "market")
210 } else {
211 url
212 }
213 });
214
215 let ws_client = BinanceFuturesWebSocketClient::new(
216 product_type,
217 config.environment,
218 api_key,
219 api_secret,
220 market_url,
221 Some(BINANCE_WS_HEARTBEAT_SECS),
222 config.transport_backend,
223 )?
224 .with_proxy(proxy_url.clone())
225 .with_socket_control(socket_factory.clone(), MARKET_STREAMS_ENDPOINT);
226
227 let public_url = config.base_url_ws.clone().map_or_else(
228 || get_ws_public_base_url(product_type, config.environment).to_string(),
229 |url| {
230 if product_type == BinanceProductType::UsdM
231 && config.environment == BinanceEnvironment::Live
232 {
233 get_usdm_ws_route_base_url(&url, "public")
234 } else {
235 url
236 }
237 },
238 );
239
240 let ws_public_client = BinanceFuturesWebSocketClient::new(
241 product_type,
242 config.environment,
243 None,
244 None,
245 Some(public_url),
246 Some(BINANCE_WS_HEARTBEAT_SECS),
247 config.transport_backend,
248 )?
249 .with_proxy(proxy_url)
250 .with_socket_control(socket_factory, PUBLIC_STREAMS_ENDPOINT);
251
252 let session_tasks = TaskGroup::new();
253 let command_tasks = TaskGroup::new();
254
255 Ok(Self {
256 clock,
257 client_id,
258 config,
259 product_type,
260 http_client,
261 ws_client,
262 ws_public_client,
263 data_sender,
264 is_connected: AtomicBool::new(false),
265 cancellation_token: session_tasks.cancellation_token(),
266 session_tasks,
267 command_tasks,
268 shutdown_errors: Vec::new(),
269 instruments: Arc::new(AtomicMap::new()),
270 status_cache: Arc::new(AtomicMap::new()),
271 book_buffers: Arc::new(AtomicMap::new()),
272 book_subscriptions: Arc::new(AtomicMap::new()),
273 book_unsubscribes_pending: Arc::new(AtomicMap::new()),
274 l1_book_subscriptions: Arc::new(AtomicMap::new()),
275 quote_refs: Arc::new(AtomicMap::new()),
276 mark_price_refs: Arc::new(AtomicMap::new()),
277 ticker_refs: Arc::new(AtomicMap::new()),
278 force_order_refs: Arc::new(AtomicMap::new()),
279 force_order_all_market_refs: Arc::new(AtomicU32::new(0)),
280 force_order_all_market_stream_active: Arc::new(AtomicBool::new(false)),
281 force_order_ws_lock: Arc::new(tokio::sync::Mutex::new(())),
282 book_epoch: Arc::new(RwLock::new(0)),
283 book_command_tail: Arc::new(Mutex::new(None)),
284 book_drain_generation: 0,
285 })
286 }
287
288 fn venue(&self) -> Venue {
289 *BINANCE_VENUE
290 }
291
292 fn send_data(sender: &EventSender<DataEvent>, data: Data) {
293 if let Err(e) = sender.send(DataEvent::Data(data)) {
294 log::error!("Failed to emit data event: {e}");
295 }
296 }
297
298 fn spawn_ws<F>(&self, fut: F, context: &'static str)
299 where
300 F: Future<Output = anyhow::Result<()>> + Send + 'static,
301 {
302 let future = async move {
303 if let Err(e) = fut.await {
304 log::error!("{context}: {e:?}");
305 }
306 };
307
308 if let Err(e) = self.command_tasks.spawn(future) {
309 log::warn!("Skipping Binance Futures {context} after shutdown began: {e}");
310 }
311 }
312
313 fn spawn_command<F>(&self, future: F)
314 where
315 F: Future<Output = ()> + Send + 'static,
316 {
317 if let Err(e) = self.command_tasks.spawn(future) {
318 log::warn!("Skipping Binance Futures data command after shutdown began: {e}");
319 }
320 }
321
322 fn chain_book_command(
325 &self,
326 ) -> (
327 impl Future<Output = ()> + Send + 'static,
328 tokio::sync::oneshot::Sender<()>,
329 ) {
330 let (done_tx, done_rx) = tokio::sync::oneshot::channel();
331 let previous = self.book_command_tail.lock().replace(done_rx);
332 let wait = async move {
333 if let Some(previous) = previous {
334 let _ = previous.await;
335 }
336 };
337 (wait, done_tx)
338 }
339
340 async fn finish_tasks(&self) -> anyhow::Result<()> {
341 let (session_result, command_result) = tokio::join!(
342 self.session_tasks
343 .finish_shutdown(Duration::from_secs(1), Duration::from_secs(2)),
344 self.command_tasks
345 .finish_shutdown(Duration::from_secs(1), Duration::from_secs(2)),
346 );
347 let mut errors = Vec::new();
348 if let Err(e) = session_result {
349 errors.push(format!(
350 "failed to finish Binance Futures data session tasks: {e}"
351 ));
352 }
353
354 if let Err(e) = command_result {
355 errors.push(format!(
356 "failed to finish Binance Futures data command tasks: {e}"
357 ));
358 }
359
360 if !errors.is_empty() {
361 anyhow::bail!(errors.join("; "));
362 }
363 Ok(())
364 }
365
366 async fn prepare_task_groups(&mut self) -> anyhow::Result<()> {
367 if !self.session_tasks.is_open() || !self.command_tasks.is_open() {
368 self.teardown_partial_connect().await?;
369 self.session_tasks
370 .start_generation()
371 .context("failed to start Binance Futures data session task generation")?;
372 self.command_tasks
373 .start_generation()
374 .context("failed to start Binance Futures data command task generation")?;
375 self.cancellation_token = self.session_tasks.cancellation_token();
376 }
377 Ok(())
378 }
379
380 async fn teardown_partial_connect(&mut self) -> anyhow::Result<()> {
381 self.session_tasks.begin_shutdown();
382 self.command_tasks.begin_shutdown();
383 self.ws_client.begin_shutdown();
384 self.ws_public_client.begin_shutdown();
385
386 if let Err(e) = self.ws_client.close().await {
387 self.shutdown_errors
388 .push(format!("market WebSocket close failed: {e}"));
389 }
390
391 if let Err(e) = self.ws_public_client.close().await {
392 self.shutdown_errors
393 .push(format!("public WebSocket close failed: {e}"));
394 }
395
396 if let Err(e) = self.finish_tasks().await {
397 self.shutdown_errors.push(e.to_string());
398 }
399 self.is_connected.store(false, Ordering::Release);
400
401 if !self.shutdown_errors.is_empty() {
402 let errors = std::mem::take(&mut self.shutdown_errors);
403 anyhow::bail!(
404 "Binance Futures data teardown failed: {}",
405 errors.join("; ")
406 );
407 }
408 Ok(())
409 }
410
411 #[expect(clippy::too_many_arguments)]
412 async fn refresh_instrument_catalog(
413 http: &BinanceFuturesHttpClient,
414 provider: &crate::config::BinanceInstrumentProviderConfig,
415 instruments_cache: &Arc<AtomicMap<InstrumentId, InstrumentAny>>,
416 status_cache: &Arc<AtomicMap<InstrumentId, MarketStatusAction>>,
417 ws: &BinanceFuturesWebSocketClient,
418 ws_public: &BinanceFuturesWebSocketClient,
419 sender: &EventSender<DataEvent>,
420 clock: &'static AtomicTime,
421 emit_status_changes: bool,
422 ) -> anyhow::Result<Vec<InstrumentAny>> {
423 let instruments = http
424 .request_instruments_with_config(provider)
425 .await
426 .context("failed to request Binance Futures instruments")?;
427 let venue_statuses = http
428 .request_symbol_statuses()
429 .await
430 .context("failed to request Binance Futures instrument statuses")?;
431
432 let instrument_map = instruments
433 .iter()
434 .map(|instrument| (instrument.id(), instrument.clone()))
435 .collect::<AHashMap<_, _>>();
436 let raw_to_id = instrument_map
437 .values()
438 .map(|instrument| (instrument.raw_symbol().inner(), instrument.id()))
439 .collect::<AHashMap<_, _>>();
440 let status_map = venue_statuses
441 .into_iter()
442 .filter_map(|(symbol, action)| {
443 raw_to_id
444 .get(&symbol)
445 .copied()
446 .map(|instrument_id| (instrument_id, action))
447 })
448 .collect::<AHashMap<_, _>>();
449
450 instruments_cache.store(instrument_map);
451 ws.replace_instruments(&instruments);
452 ws_public.replace_instruments(&instruments);
453
454 if emit_status_changes {
455 let mut cached_statuses = (**status_cache.load()).clone();
456 let ts = clock.get_time_ns();
457 diff_and_emit_statuses(&status_map, &mut cached_statuses, sender, ts, ts);
458 status_cache.store(cached_statuses);
459 } else {
460 status_cache.store(status_map);
461 }
462
463 for instrument in &instruments {
464 if let Err(e) = sender.send(DataEvent::Instrument(instrument.clone())) {
465 log::warn!("Failed to send refreshed Binance Futures instrument: {e}");
466 }
467 }
468
469 Ok(instruments)
470 }
471
472 fn custom_liquidation_instrument_id(
473 data_type: &DataType,
474 ) -> anyhow::Result<Option<InstrumentId>> {
475 let Some(raw_instrument_id) = data_type
476 .metadata()
477 .as_ref()
478 .and_then(|m| m.get("instrument_id"))
479 .and_then(|v| v.as_str())
480 .map(str::trim)
481 .filter(|value| !value.is_empty())
482 else {
483 return Ok(None);
484 };
485
486 let instrument_id = InstrumentId::from_str(raw_instrument_id)
487 .with_context(|| format!("invalid instrument_id metadata `{raw_instrument_id}`"))?;
488
489 Ok(Some(instrument_id))
490 }
491
492 fn required_instrument_id_metadata(data_type: &DataType) -> anyhow::Result<InstrumentId> {
493 let Some(raw_instrument_id) = data_type
494 .metadata()
495 .as_ref()
496 .and_then(|m| m.get("instrument_id"))
497 .and_then(|v| v.as_str())
498 .map(str::trim)
499 .filter(|value| !value.is_empty())
500 else {
501 anyhow::bail!("custom data request requires `instrument_id` metadata");
502 };
503
504 InstrumentId::from_str(raw_instrument_id)
505 .with_context(|| format!("invalid instrument_id metadata `{raw_instrument_id}`"))
506 }
507
508 fn required_period_metadata(data_type: &DataType) -> anyhow::Result<String> {
509 let Some(period) = data_type
510 .metadata()
511 .as_ref()
512 .and_then(|m| m.get("period"))
513 .and_then(|v| v.as_str())
514 .map(str::trim)
515 .filter(|value| !value.is_empty())
516 else {
517 anyhow::bail!("historical open interest request requires `period` metadata");
518 };
519
520 Ok(period.to_string())
521 }
522
523 fn coinm_open_interest_hist_params(
524 http: &BinanceFuturesHttpClient,
525 instrument_id: &InstrumentId,
526 ) -> anyhow::Result<(String, String)> {
527 let symbol = format_binance_symbol(instrument_id);
528 if let Some(pair) = symbol.strip_suffix("_PERP") {
529 return Ok((pair.to_string(), "PERPETUAL".to_string()));
530 }
531
532 let definition = http
533 .instrument_metadata(*instrument_id)
534 .with_context(|| format!("missing COIN-M definition for {instrument_id}"))?;
535 let BinanceFuturesInstrument::CoinM(definition) = definition else {
536 anyhow::bail!("expected a COIN-M definition for {instrument_id}");
537 };
538
539 Ok((definition.pair.to_string(), definition.contract_type))
540 }
541
542 fn parse_open_interest_decimal(field: &str, value: &str) -> anyhow::Result<Decimal> {
543 Decimal::from_str_exact(value)
544 .with_context(|| format!("invalid Binance open interest `{field}` value `{value}`"))
545 }
546
547 fn liquidation_data_type(instrument_id: InstrumentId) -> DataType {
548 let mut metadata = Params::new();
549 metadata.insert(
550 "instrument_id".to_string(),
551 serde_json::Value::String(instrument_id.to_string()),
552 );
553 DataType::new(
554 "BinanceFuturesLiquidation",
555 Some(metadata),
556 Some(instrument_id.to_string()),
557 )
558 }
559
560 fn liquidation_stream(instrument_id: &InstrumentId) -> String {
561 format!("{}@forceOrder", format_binance_stream_symbol(instrument_id))
562 }
563
564 fn spawn_liquidation_stream_reconcile(&self, context: &'static str) {
565 let ws = self.ws_client.clone();
566 let refs = self.force_order_refs.clone();
567 let all_market_refs = self.force_order_all_market_refs.clone();
568 let all_market_stream_active = self.force_order_all_market_stream_active.clone();
569 let ws_lock = self.force_order_ws_lock.clone();
570
571 self.spawn_ws(
572 async move {
573 let _guard = ws_lock.lock().await;
574 let wants_all_market = all_market_refs.load(Ordering::Relaxed) > 0;
575 let all_market_active = all_market_stream_active.load(Ordering::Acquire);
576
577 if wants_all_market {
578 if all_market_active {
579 return Ok(());
580 }
581
582 let specific_streams = refs
583 .load()
584 .keys()
585 .map(Self::liquidation_stream)
586 .collect::<Vec<_>>();
587
588 if !specific_streams.is_empty() {
589 ws.unsubscribe(specific_streams).await.context(
590 "specific forceOrder unsubscribe while enabling all-market",
591 )?;
592 }
593
594 if all_market_refs.load(Ordering::Relaxed) == 0 {
595 let restored_streams = refs
596 .load()
597 .keys()
598 .map(Self::liquidation_stream)
599 .collect::<Vec<_>>();
600
601 if !restored_streams.is_empty() {
602 ws.subscribe(restored_streams).await.context(
603 "specific forceOrder restore after canceled all-market subscription",
604 )?;
605 }
606 all_market_stream_active.store(false, Ordering::Release);
607 return Ok(());
608 }
609
610 all_market_stream_active.store(true, Ordering::Release);
611
612 if let Err(e) = ws
613 .subscribe(vec!["!forceOrder@arr".to_string()])
614 .await
615 .context("all-market forceOrder subscription")
616 {
617 all_market_stream_active.store(false, Ordering::Release);
618 return Err(e);
619 }
620 } else {
621 if !all_market_active {
622 return Ok(());
623 }
624
625 ws.unsubscribe(vec!["!forceOrder@arr".to_string()])
626 .await
627 .context("all-market forceOrder unsubscribe")?;
628
629 let specific_streams = refs
630 .load()
631 .keys()
632 .map(Self::liquidation_stream)
633 .collect::<Vec<_>>();
634
635 if !specific_streams.is_empty() {
636 ws.subscribe(specific_streams).await.context(
637 "specific forceOrder resubscribe after all-market unsubscribe",
638 )?;
639 }
640 all_market_stream_active.store(false, Ordering::Release);
641 }
642
643 Ok(())
644 },
645 context,
646 );
647 }
648
649 #[expect(clippy::too_many_arguments)]
650 fn handle_ws_message(
651 msg: BinanceFuturesWsStreamsMessage,
652 data_sender: &EventSender<DataEvent>,
653 instruments: &Arc<AtomicMap<InstrumentId, InstrumentAny>>,
654 ws_instruments: &Arc<AtomicMap<Ustr, InstrumentAny>>,
655 book_buffers: &Arc<AtomicMap<InstrumentId, BookBuffer>>,
656 book_subscriptions: &Arc<AtomicMap<InstrumentId, u32>>,
657 book_unsubscribes_pending: &Arc<AtomicMap<InstrumentId, Vec<BookDrain>>>,
658 l1_book_subscriptions: &Arc<AtomicMap<InstrumentId, u32>>,
659 force_order_refs: &Arc<AtomicMap<InstrumentId, u32>>,
660 ticker_refs: &Arc<AtomicMap<InstrumentId, u32>>,
661 force_order_all_market_refs: &Arc<AtomicU32>,
662 force_order_all_market_stream_active: &Arc<AtomicBool>,
663 book_epoch: &Arc<RwLock<u64>>,
664 http_client: &BinanceFuturesHttpClient,
665 clock: &'static AtomicTime,
666 command_spawner: &TaskSpawner,
667 ) {
668 let ts_init = clock.get_time_ns();
669 let cache = ws_instruments.load();
670
671 match msg {
672 BinanceFuturesWsStreamsMessage::AggTrade(ref trade_msg) => {
673 if let Some(instrument) = cache.get(&trade_msg.symbol) {
674 match parse_agg_trade(trade_msg, instrument, ts_init) {
675 Ok(trade) => Self::send_data(data_sender, Data::Trade(trade)),
676 Err(e) => log::warn!("Failed to parse aggregate trade: {e}"),
677 }
678 }
679 }
680 BinanceFuturesWsStreamsMessage::Trade(ref trade_msg) => {
681 if let Some(instrument) = cache.get(&trade_msg.symbol) {
682 match parse_trade(trade_msg, instrument, ts_init) {
683 Ok(trade) => Self::send_data(data_sender, Data::Trade(trade)),
684 Err(e) => log::warn!("Failed to parse trade: {e}"),
685 }
686 }
687 }
688 BinanceFuturesWsStreamsMessage::BookTicker(ref ticker_msg) => {
689 if let Some(instrument) = cache.get(&ticker_msg.symbol) {
690 match parse_book_ticker(ticker_msg, instrument, ts_init) {
691 Ok(quote) => Self::send_top_of_book(
692 data_sender,
693 l1_book_subscriptions,
694 quote,
695 ticker_msg.update_id,
696 ),
697 Err(e) => log::warn!("Failed to parse book ticker: {e}"),
698 }
699 }
700 }
701 BinanceFuturesWsStreamsMessage::DepthUpdate(ref depth_msg) => {
702 if let Some(instrument) = cache.get(&depth_msg.symbol) {
703 let Some(depth) = book_subscriptions.load().get(&instrument.id()).copied()
704 else {
705 return;
706 };
707
708 if book_drain_active(book_unsubscribes_pending, instrument.id()) {
713 log::debug!(
714 "Dropping depth frame for {} with unsubscribe confirmation pending",
715 instrument.id()
716 );
717 return;
718 }
719
720 let parsed = if is_partial_book_depth(depth) {
721 parse_depth_snapshot(depth_msg, instrument, ts_init)
722 } else {
723 parse_depth_update(depth_msg, instrument, ts_init)
724 };
725
726 match parsed {
727 Ok(deltas) => {
728 let instrument_id = deltas.instrument_id;
729 let final_update_id = deltas.sequence;
730 let first_update_id = depth_msg.first_update_id;
731 let prev_final_update_id = depth_msg.prev_final_update_id;
732
733 if book_buffers.contains_key(&instrument_id) {
734 let mut was_buffered = false;
735 book_buffers.rcu(|m| {
736 was_buffered = false;
737
738 if let Some(buffer) = m.get_mut(&instrument_id) {
739 buffer.updates.push(BufferedDepthUpdate {
740 deltas: deltas.clone(),
741 first_update_id,
742 final_update_id,
743 prev_final_update_id,
744 });
745 trim_buffered_depth_updates(&mut buffer.updates);
746 was_buffered = true;
747 }
748 });
749
750 if was_buffered {
751 return;
752 }
753 }
754
755 Self::send_data(data_sender, Data::BookDeltas(Box::new(deltas)));
756 }
757 Err(e) => log::warn!("Failed to parse depth update: {e}"),
758 }
759 }
760 }
761 BinanceFuturesWsStreamsMessage::MarkPrice(ref mark_msg) => {
762 if let Some(instrument) = cache.get(&mark_msg.symbol) {
763 match parse_mark_price(mark_msg, instrument, ts_init) {
764 Ok((mark_update, index_update, funding_update, custom_update)) => {
765 Self::send_data(data_sender, Data::MarkPrice(mark_update));
766 Self::send_data(data_sender, Data::IndexPrice(index_update));
767 if let Err(e) = data_sender.send(DataEvent::FundingRate(funding_update))
768 {
769 log::error!("Failed to emit funding rate: {e}");
770 }
771 let data_type = mark_price_data_type(instrument.id());
772 Self::send_data(
773 data_sender,
774 Data::Custom(CustomData::new(Arc::new(custom_update), data_type)),
775 );
776 }
777 Err(e) => log::warn!("Failed to parse mark price: {e}"),
778 }
779 }
780 }
781 BinanceFuturesWsStreamsMessage::Kline(ref kline_msg) => {
782 if let Some(instrument) = cache.get(&kline_msg.symbol) {
783 match parse_kline(kline_msg, instrument, ts_init) {
784 Ok(Some(bar)) => {
785 Self::send_data(data_sender, Data::Bar(bar.bar()));
786 let data_type = binance_bar_data_type(bar.bar_type);
787 Self::send_data(
788 data_sender,
789 Data::Custom(CustomData::new(Arc::new(bar), data_type)),
790 );
791 }
792 Ok(None) => {} Err(e) => log::warn!("Failed to parse kline: {e}"),
794 }
795 }
796 }
797 BinanceFuturesWsStreamsMessage::ForceOrder(ref liq_msg) => {
798 if let Some(instrument) = cache.get(&liq_msg.order.symbol) {
799 let ts_event = parse_millis_or_init(
800 liq_msg.event_time,
801 "Futures liquidation event time",
802 ts_init,
803 );
804 let parse_price = |value: &str, field: &str| -> anyhow::Result<Price> {
805 parse_required_price_at_precision(
806 value,
807 instrument.price_precision(),
808 field,
809 )
810 };
811
812 let parse_quantity = |value: &str, field: &str| -> anyhow::Result<Quantity> {
813 parse_required_quantity_at_precision(
814 value,
815 instrument.size_precision(),
816 field,
817 )
818 };
819
820 match (
821 parse_price(&liq_msg.order.price, "price"),
822 parse_price(&liq_msg.order.average_price, "average_price"),
823 parse_quantity(&liq_msg.order.last_filled_qty, "last_filled_qty"),
824 parse_quantity(&liq_msg.order.accumulated_qty, "accumulated_qty"),
825 ) {
826 (
827 Ok(price),
828 Ok(average_price),
829 Ok(last_filled_qty),
830 Ok(accumulated_qty),
831 ) => {
832 let liquidation = Arc::new(BinanceFuturesLiquidation::new(
833 instrument.id(),
834 OrderSide::from(liq_msg.order.side),
835 price,
836 average_price,
837 last_filled_qty,
838 accumulated_qty,
839 ts_event,
840 ts_init,
841 ));
842
843 let has_all_market_subscription =
844 force_order_all_market_refs.load(Ordering::Relaxed) > 0;
845 let has_all_market_stream =
846 force_order_all_market_stream_active.load(Ordering::Acquire);
847 let has_specific_subscription =
848 force_order_refs.load().contains_key(&instrument.id());
849
850 if has_all_market_subscription || has_all_market_stream {
851 let data_type =
852 DataType::new("BinanceFuturesLiquidation", None, None);
853 Self::send_data(
854 data_sender,
855 Data::Custom(CustomData::new(liquidation, data_type)),
856 );
857 } else if has_specific_subscription {
858 let data_type = Self::liquidation_data_type(instrument.id());
859 Self::send_data(
860 data_sender,
861 Data::Custom(CustomData::new(liquidation, data_type)),
862 );
863 }
864 }
865 (p, ap, lq, aq) => {
866 log::warn!(
867 "Failed to parse Binance liquidation {}: price={:?} avg={:?} \
868 last_qty={:?} accumulated_qty={:?}",
869 liq_msg.order.symbol,
870 p.err(),
871 ap.err(),
872 lq.err(),
873 aq.err(),
874 );
875 }
876 }
877 } else {
878 log::warn!(
879 "Received Binance liquidation for uncached symbol {}",
880 liq_msg.order.symbol
881 );
882 }
883 }
884 BinanceFuturesWsStreamsMessage::Ticker(ref ticker_msg) => {
885 if let Some(instrument) = cache.get(&ticker_msg.symbol) {
886 let instrument_id = instrument.id();
887 if !ticker_refs.load().contains_key(&instrument_id) {
888 return;
889 }
890
891 match parse_ticker(ticker_msg, instrument, ts_init) {
892 Ok(ticker) => {
893 let data_type = ticker_data_type(instrument_id);
894 Self::send_data(
895 data_sender,
896 Data::Custom(CustomData::new(Arc::new(ticker), data_type)),
897 );
898 }
899 Err(e) => log::warn!("Failed to parse ticker: {e}"),
900 }
901 }
902 }
903 BinanceFuturesWsStreamsMessage::AccountUpdate(_)
904 | BinanceFuturesWsStreamsMessage::OrderUpdate(_)
905 | BinanceFuturesWsStreamsMessage::TradeLite(_)
906 | BinanceFuturesWsStreamsMessage::AlgoUpdate(_)
907 | BinanceFuturesWsStreamsMessage::MarginCall(_)
908 | BinanceFuturesWsStreamsMessage::AccountConfigUpdate(_)
909 | BinanceFuturesWsStreamsMessage::ListenKeyExpired => {}
910 BinanceFuturesWsStreamsMessage::Error(e) => {
911 log::warn!(
912 "Binance Futures WebSocket error: code={}, msg={}",
913 e.code,
914 e.msg
915 );
916 }
917 BinanceFuturesWsStreamsMessage::Reconnected(abandoned) => {
918 log::info!("WebSocket reconnected, rebuilding order book snapshots");
919
920 for generation in abandoned {
923 remove_book_drain(book_unsubscribes_pending, generation);
924 }
925
926 let epoch = {
927 let mut guard = book_epoch.write();
928 *guard = guard.wrapping_add(1);
929 *guard
930 };
931
932 let subs: Vec<(InstrumentId, u32)> = {
933 let guard = book_subscriptions.load();
934 guard.iter().map(|(k, v)| (*k, *v)).collect()
935 };
936
937 for (instrument_id, depth) in subs {
938 if is_partial_book_depth(depth) {
939 continue;
940 }
941
942 book_buffers.insert(instrument_id, BookBuffer::new(epoch));
943
944 log::debug!(
945 "OrderBook snapshot rebuild for {instrument_id} @ depth {depth} \
946 starting (reconnect, epoch={epoch})"
947 );
948
949 let http = http_client.clone();
950 let sender = data_sender.clone();
951 let buffers = book_buffers.clone();
952 let insts = instruments.clone();
953
954 if let Err(e) = command_spawner.spawn(async move {
955 Self::fetch_and_emit_snapshot(
956 http,
957 sender,
958 buffers,
959 insts,
960 instrument_id,
961 depth,
962 epoch,
963 clock,
964 )
965 .await;
966 }) {
967 log::warn!(
968 "Skipping Binance Futures snapshot rebuild after shutdown began: {e}"
969 );
970 }
971 }
972 }
973 BinanceFuturesWsStreamsMessage::Unsubscribed {
974 streams,
975 correlation,
976 } => {
977 log::debug!("Unsubscribe confirmed for streams {streams:?}");
978
979 if let Some(correlation) = correlation {
980 remove_book_drain(book_unsubscribes_pending, correlation);
981 }
982 }
983 }
984 }
985
986 fn send_top_of_book(
987 data_sender: &EventSender<DataEvent>,
988 l1_book_subscriptions: &Arc<AtomicMap<InstrumentId, u32>>,
989 quote: QuoteTick,
990 sequence: u64,
991 ) {
992 Self::send_data(data_sender, Data::Quote(quote));
993 if l1_book_subscriptions.contains_key("e.instrument_id) {
994 let deltas = quote_to_l1_deltas(quote, sequence);
995 Self::send_data(data_sender, Data::BookDeltas(Box::new(deltas)));
996 }
997 }
998
999 #[expect(clippy::too_many_arguments)]
1000 async fn fetch_and_emit_snapshot(
1001 http: BinanceFuturesHttpClient,
1002 sender: EventSender<DataEvent>,
1003 buffers: Arc<AtomicMap<InstrumentId, BookBuffer>>,
1004 instruments: Arc<AtomicMap<InstrumentId, InstrumentAny>>,
1005 instrument_id: InstrumentId,
1006 depth: u32,
1007 epoch: u64,
1008 clock: &'static AtomicTime,
1009 ) {
1010 Self::fetch_and_emit_snapshot_inner(
1011 http,
1012 sender,
1013 buffers,
1014 instruments,
1015 instrument_id,
1016 depth,
1017 epoch,
1018 clock,
1019 0,
1020 )
1021 .await;
1022 }
1023
1024 #[expect(clippy::too_many_arguments)]
1025 async fn fetch_and_emit_snapshot_inner(
1026 http: BinanceFuturesHttpClient,
1027 sender: EventSender<DataEvent>,
1028 buffers: Arc<AtomicMap<InstrumentId, BookBuffer>>,
1029 instruments: Arc<AtomicMap<InstrumentId, InstrumentAny>>,
1030 instrument_id: InstrumentId,
1031 depth: u32,
1032 epoch: u64,
1033 clock: &'static AtomicTime,
1034 retry_count: u32,
1035 ) {
1036 if wait_for_buffered_update(&buffers, instrument_id, epoch)
1037 .await
1038 .is_none()
1039 {
1040 return;
1041 }
1042
1043 let symbol = format_binance_stream_symbol(&instrument_id).to_uppercase();
1044 let params = BinanceDepthParams {
1045 symbol,
1046 limit: Some(depth),
1047 };
1048
1049 match http.depth(¶ms).await {
1050 Ok(order_book) => {
1051 let ts_init = clock.get_time_ns();
1052 let last_update_id = order_book.last_update_id as u64;
1053
1054 {
1055 let guard = buffers.load();
1056 match guard.get(&instrument_id) {
1057 None => {
1058 log::debug!(
1059 "OrderBook subscription for {instrument_id} was cancelled, \
1060 discarding snapshot"
1061 );
1062 return;
1063 }
1064 Some(buffer) if buffer.epoch != epoch => {
1065 log::debug!(
1066 "OrderBook snapshot for {instrument_id} is stale \
1067 (epoch {epoch} != {}), discarding",
1068 buffer.epoch
1069 );
1070 return;
1071 }
1072 _ => {}
1073 }
1074 }
1075
1076 let (price_precision, size_precision) = {
1077 let guard = instruments.load();
1078 match guard.get(&instrument_id) {
1079 Some(inst) => (inst.price_precision(), inst.size_precision()),
1080 None => {
1081 log::error!("No instrument in cache for snapshot: {instrument_id}");
1082 buffers.remove(&instrument_id);
1083 return;
1084 }
1085 }
1086 };
1087
1088 let Some(first) = wait_for_first_applicable_update(
1089 &buffers,
1090 instrument_id,
1091 epoch,
1092 last_update_id,
1093 )
1094 .await
1095 else {
1096 return;
1097 };
1098
1099 let target = last_update_id;
1102 let valid_overlap =
1103 first.first_update_id <= target && first.final_update_id >= target;
1104
1105 if !valid_overlap {
1106 if retry_count < MAX_SNAPSHOT_RETRIES {
1107 log::warn!(
1108 "OrderBook overlap validation failed for {instrument_id}: \
1109 lastUpdateId={last_update_id}, first_update_id={}, \
1110 final_update_id={} (need U <= {} <= u), \
1111 retrying snapshot (attempt {}/{})",
1112 first.first_update_id,
1113 first.final_update_id,
1114 target,
1115 retry_count + 1,
1116 MAX_SNAPSHOT_RETRIES
1117 );
1118
1119 tokio::time::sleep(futures_snapshot_retry_backoff(retry_count)).await;
1120
1121 Box::pin(Self::fetch_and_emit_snapshot_inner(
1122 http,
1123 sender,
1124 buffers,
1125 instruments,
1126 instrument_id,
1127 depth,
1128 epoch,
1129 clock,
1130 retry_count + 1,
1131 ))
1132 .await;
1133 return;
1134 }
1135 log::error!(
1136 "OrderBook overlap validation failed for {instrument_id} after \
1137 {MAX_SNAPSHOT_RETRIES} retries; book may be inconsistent"
1138 );
1139 }
1140
1141 let snapshot_deltas = parse_order_book_snapshot(
1142 &order_book,
1143 instrument_id,
1144 price_precision,
1145 size_precision,
1146 ts_init,
1147 );
1148
1149 let buffered = {
1151 let mut taken = Vec::new();
1152 let mut should_return = false;
1153 buffers.rcu(|m| {
1154 taken = Vec::new();
1155 should_return = false;
1156
1157 match m.get_mut(&instrument_id) {
1158 Some(buffer) if buffer.epoch == epoch => {
1159 taken = std::mem::take(&mut buffer.updates);
1160 }
1161 _ => should_return = true,
1162 }
1163 });
1164
1165 if should_return {
1166 return;
1167 }
1168 taken
1169 };
1170
1171 let mut replayed = 0;
1172 let mut last_final_update_id = last_update_id;
1173 let mut is_first = true;
1174 let mut replay_ready = Vec::with_capacity(buffered.len());
1175
1176 for update in buffered {
1177 if update.final_update_id < last_update_id {
1178 continue;
1179 }
1180
1181 if update.final_update_id == last_update_id {
1182 last_final_update_id = update.final_update_id;
1183 is_first = false;
1184 continue;
1185 }
1186
1187 if !is_first && update.prev_final_update_id != last_final_update_id {
1190 if retry_count < MAX_SNAPSHOT_RETRIES {
1191 log::warn!(
1192 "OrderBook continuity break for {instrument_id}: \
1193 expected pu={last_final_update_id}, was pu={}, \
1194 triggering resync (attempt {}/{})",
1195 update.prev_final_update_id,
1196 retry_count + 1,
1197 MAX_SNAPSHOT_RETRIES
1198 );
1199
1200 reset_book_sync_buffer(&buffers, instrument_id, epoch);
1201 tokio::time::sleep(futures_snapshot_retry_backoff(retry_count)).await;
1202
1203 Box::pin(Self::fetch_and_emit_snapshot_inner(
1204 http,
1205 sender,
1206 buffers,
1207 instruments,
1208 instrument_id,
1209 depth,
1210 epoch,
1211 clock,
1212 retry_count + 1,
1213 ))
1214 .await;
1215 return;
1216 }
1217 log::error!(
1218 "OrderBook continuity break for {instrument_id} after \
1219 {MAX_SNAPSHOT_RETRIES} retries: expected pu={last_final_update_id}, \
1220 was pu={}; book may be inconsistent",
1221 update.prev_final_update_id
1222 );
1223 }
1224
1225 last_final_update_id = update.final_update_id;
1226 is_first = false;
1227 replayed += 1;
1228 replay_ready.push(update);
1229 }
1230
1231 if let Err(e) =
1232 sender.send(DataEvent::Data(Data::BookDeltas(Box::new(snapshot_deltas))))
1233 {
1234 log::error!("Failed to send snapshot: {e}");
1235 }
1236
1237 for update in replay_ready {
1238 if let Err(e) =
1239 sender.send(DataEvent::Data(Data::BookDeltas(Box::new(update.deltas))))
1240 {
1241 log::error!("Failed to send replayed deltas: {e}");
1242 }
1243 }
1244
1245 loop {
1247 let more = {
1248 let mut taken = Vec::new();
1249 let mut should_break = false;
1250 buffers.rcu(|m| {
1251 taken = Vec::new();
1252 should_break = false;
1253
1254 match m.get_mut(&instrument_id) {
1255 Some(buffer) if buffer.epoch == epoch => {
1256 if buffer.updates.is_empty() {
1257 m.remove(&instrument_id);
1258 should_break = true;
1259 } else {
1260 taken = std::mem::take(&mut buffer.updates);
1261 }
1262 }
1263 _ => should_break = true,
1264 }
1265 });
1266
1267 if should_break {
1268 break;
1269 }
1270 taken
1271 };
1272
1273 for update in more {
1274 if update.final_update_id <= last_update_id {
1275 continue;
1276 }
1277
1278 if update.prev_final_update_id != last_final_update_id {
1279 if retry_count < MAX_SNAPSHOT_RETRIES {
1280 log::warn!(
1281 "OrderBook continuity break for {instrument_id}: \
1282 expected pu={last_final_update_id}, was pu={}, \
1283 triggering resync (attempt {}/{})",
1284 update.prev_final_update_id,
1285 retry_count + 1,
1286 MAX_SNAPSHOT_RETRIES
1287 );
1288
1289 reset_book_sync_buffer(&buffers, instrument_id, epoch);
1290 tokio::time::sleep(futures_snapshot_retry_backoff(retry_count))
1291 .await;
1292
1293 Box::pin(Self::fetch_and_emit_snapshot_inner(
1294 http,
1295 sender,
1296 buffers,
1297 instruments,
1298 instrument_id,
1299 depth,
1300 epoch,
1301 clock,
1302 retry_count + 1,
1303 ))
1304 .await;
1305 return;
1306 }
1307 log::error!(
1308 "OrderBook continuity break for {instrument_id} after \
1309 {MAX_SNAPSHOT_RETRIES} retries; book may be inconsistent"
1310 );
1311 }
1312
1313 last_final_update_id = update.final_update_id;
1314 replayed += 1;
1315
1316 if let Err(e) =
1317 sender.send(DataEvent::Data(Data::BookDeltas(Box::new(update.deltas))))
1318 {
1319 log::error!("Failed to send replayed deltas: {e}");
1320 }
1321 }
1322 }
1323
1324 log::debug!(
1325 "OrderBook snapshot rebuild for {instrument_id} completed \
1326 (lastUpdateId={last_update_id}, replayed={replayed})"
1327 );
1328 }
1329 Err(e) => {
1330 if retry_count < MAX_SNAPSHOT_RETRIES {
1331 log::warn!(
1332 "Failed to request order book snapshot for {instrument_id}: {e}; \
1333 retrying snapshot (attempt {}/{})",
1334 retry_count + 1,
1335 MAX_SNAPSHOT_RETRIES
1336 );
1337
1338 tokio::time::sleep(futures_snapshot_retry_backoff(retry_count)).await;
1339
1340 Box::pin(Self::fetch_and_emit_snapshot_inner(
1341 http,
1342 sender,
1343 buffers,
1344 instruments,
1345 instrument_id,
1346 depth,
1347 epoch,
1348 clock,
1349 retry_count + 1,
1350 ))
1351 .await;
1352 return;
1353 }
1354
1355 log::error!(
1356 "Failed to request order book snapshot for {instrument_id} after \
1357 {MAX_SNAPSHOT_RETRIES} retries: {e}"
1358 );
1359 buffers.remove(&instrument_id);
1360 }
1361 }
1362 }
1363}
1364
1365fn upsert_instrument(
1366 cache: &Arc<AtomicMap<InstrumentId, InstrumentAny>>,
1367 instrument: InstrumentAny,
1368) {
1369 cache.insert(instrument.id(), instrument);
1370}
1371
1372fn reset_book_sync_buffer(
1373 buffers: &Arc<AtomicMap<InstrumentId, BookBuffer>>,
1374 instrument_id: InstrumentId,
1375 epoch: u64,
1376) {
1377 buffers.rcu(|m| {
1378 if let Some(buffer) = m.get_mut(&instrument_id)
1379 && buffer.epoch == epoch
1380 {
1381 buffer.updates.clear();
1382 }
1383 });
1384}
1385
1386fn trim_buffered_depth_updates(updates: &mut Vec<BufferedDepthUpdate>) {
1387 let excess = updates.len().saturating_sub(MAX_BUFFERED_DEPTH_UPDATES);
1388 if excess > 0 {
1389 updates.drain(..excess);
1390 }
1391}
1392
1393async fn wait_for_buffered_update(
1394 buffers: &Arc<AtomicMap<InstrumentId, BookBuffer>>,
1395 instrument_id: InstrumentId,
1396 epoch: u64,
1397) -> Option<()> {
1398 loop {
1399 let guard = buffers.load();
1400 match guard.get(&instrument_id) {
1401 Some(buffer) if buffer.epoch == epoch && !buffer.updates.is_empty() => return Some(()),
1402 Some(buffer) if buffer.epoch == epoch => {}
1403 _ => return None,
1404 }
1405
1406 drop(guard);
1407 tokio::time::sleep(Duration::from_millis(100)).await;
1408 }
1409}
1410
1411async fn wait_for_first_applicable_update(
1412 buffers: &Arc<AtomicMap<InstrumentId, BookBuffer>>,
1413 instrument_id: InstrumentId,
1414 epoch: u64,
1415 last_update_id: u64,
1416) -> Option<BufferedDepthUpdate> {
1417 loop {
1418 let mut first = None;
1419 let mut waiting = false;
1420 buffers.rcu(|m| {
1421 first = None;
1422 waiting = false;
1423
1424 if let Some(buffer) = m.get_mut(&instrument_id)
1425 && buffer.epoch == epoch
1426 {
1427 buffer
1428 .updates
1429 .retain(|update| update.final_update_id >= last_update_id);
1430 first = buffer
1431 .updates
1432 .iter()
1433 .find(|update| update.final_update_id >= last_update_id)
1434 .cloned();
1435 waiting = first.is_none();
1436 }
1437 });
1438
1439 if first.is_some() {
1440 return first;
1441 }
1442
1443 if !waiting {
1444 return None;
1445 }
1446
1447 tokio::time::sleep(Duration::from_millis(100)).await;
1448 }
1449}
1450
1451fn futures_snapshot_retry_backoff(retry_count: u32) -> Duration {
1452 let multiplier = 1_u64 << retry_count.min(4);
1453 let millis = SNAPSHOT_RETRY_BACKOFF_BASE_MS
1454 .saturating_mul(multiplier)
1455 .min(SNAPSHOT_RETRY_BACKOFF_CAP_MS);
1456 Duration::from_millis(millis)
1457}
1458
1459fn parse_order_book_snapshot(
1460 order_book: &BinanceOrderBook,
1461 instrument_id: InstrumentId,
1462 price_precision: u8,
1463 size_precision: u8,
1464 ts_init: UnixNanos,
1465) -> OrderBookDeltas {
1466 let sequence = order_book.last_update_id as u64;
1467 let ts_event = order_book.transaction_time.map_or(ts_init, |value| {
1468 parse_millis_or_init(
1469 value,
1470 "Futures order book snapshot transaction time",
1471 ts_init,
1472 )
1473 });
1474
1475 let total_levels = order_book.bids.len() + order_book.asks.len();
1476 let mut deltas = Vec::with_capacity(total_levels + 1);
1477
1478 deltas.push(OrderBookDelta::clear(
1479 instrument_id,
1480 sequence,
1481 ts_event,
1482 ts_init,
1483 ));
1484
1485 for (price_str, qty_str) in &order_book.bids {
1486 let Some(price) = parse_price_at_precision(price_str, price_precision) else {
1487 log::warn!(
1488 "Skipping Futures order book bid level for {instrument_id}: invalid or \
1489 non-positive price='{price_str}'"
1490 );
1491 continue;
1492 };
1493 let Some(size) = parse_quantity_at_precision(qty_str, size_precision) else {
1494 log::warn!(
1495 "Skipping Futures order book bid level for {instrument_id}: invalid or \
1496 non-positive quantity='{qty_str}'"
1497 );
1498 continue;
1499 };
1500
1501 let order = BookOrder::new(OrderSide::Buy, price, size, 0);
1502
1503 deltas.push(OrderBookDelta::new(
1504 instrument_id,
1505 BookAction::Add,
1506 order,
1507 0,
1508 sequence,
1509 ts_event,
1510 ts_init,
1511 ));
1512 }
1513
1514 for (price_str, qty_str) in &order_book.asks {
1515 let Some(price) = parse_price_at_precision(price_str, price_precision) else {
1516 log::warn!(
1517 "Skipping Futures order book ask level for {instrument_id}: invalid or \
1518 non-positive price='{price_str}'"
1519 );
1520 continue;
1521 };
1522 let Some(size) = parse_quantity_at_precision(qty_str, size_precision) else {
1523 log::warn!(
1524 "Skipping Futures order book ask level for {instrument_id}: invalid or \
1525 non-positive quantity='{qty_str}'"
1526 );
1527 continue;
1528 };
1529
1530 let order = BookOrder::new(OrderSide::Sell, price, size, 0);
1531
1532 deltas.push(OrderBookDelta::new(
1533 instrument_id,
1534 BookAction::Add,
1535 order,
1536 0,
1537 sequence,
1538 ts_event,
1539 ts_init,
1540 ));
1541 }
1542
1543 if let Some(delta) = deltas.last_mut() {
1544 delta.flags |= RecordFlag::F_LAST as u8;
1545 }
1546
1547 OrderBookDeltas::new(instrument_id, deltas)
1548}
1549
1550#[async_trait::async_trait(?Send)]
1551impl DataClient for BinanceFuturesDataClient {
1552 fn client_id(&self) -> ClientId {
1553 self.client_id
1554 }
1555
1556 fn venue(&self) -> Option<Venue> {
1557 Some(self.venue())
1558 }
1559
1560 fn start(&mut self) -> anyhow::Result<()> {
1561 log::info!(
1562 "Started: client_id={}, product_type={:?}, environment={:?}",
1563 self.client_id,
1564 self.product_type,
1565 self.config.environment,
1566 );
1567 Ok(())
1568 }
1569
1570 fn stop(&mut self) -> anyhow::Result<()> {
1571 log::info!("Stopping {id}", id = self.client_id);
1572 self.session_tasks.begin_shutdown();
1573 self.command_tasks.begin_shutdown();
1574 self.ws_client.begin_shutdown();
1575 self.ws_public_client.begin_shutdown();
1576 self.is_connected.store(false, Ordering::Relaxed);
1577 Ok(())
1578 }
1579
1580 fn reset(&mut self) -> anyhow::Result<()> {
1581 log::debug!("Resetting {id}", id = self.client_id);
1582
1583 self.session_tasks.begin_shutdown();
1584 self.command_tasks.begin_shutdown();
1585 self.ws_client.begin_shutdown();
1586 self.ws_public_client.begin_shutdown();
1587 self.is_connected.store(false, Ordering::Relaxed);
1588
1589 self.mark_price_refs.store(AHashMap::new());
1591 self.ticker_refs.store(AHashMap::new());
1592 self.force_order_refs.store(AHashMap::new());
1593 self.force_order_all_market_refs.store(0, Ordering::Relaxed);
1594 self.force_order_all_market_stream_active
1595 .store(false, Ordering::Release);
1596 self.book_subscriptions.store(AHashMap::new());
1597 self.book_unsubscribes_pending.store(AHashMap::new());
1598 self.l1_book_subscriptions.store(AHashMap::new());
1599 self.quote_refs.store(AHashMap::new());
1600 self.book_buffers.store(AHashMap::new());
1601
1602 Ok(())
1603 }
1604
1605 fn dispose(&mut self) -> anyhow::Result<()> {
1606 log::debug!("Disposing {id}", id = self.client_id);
1607 self.stop()
1608 }
1609
1610 async fn connect(&mut self) -> anyhow::Result<()> {
1611 if self.is_connected() && self.session_tasks.is_open() && self.command_tasks.is_open() {
1612 return Ok(());
1613 }
1614
1615 register_binance_custom_data();
1616
1617 self.prepare_task_groups().await?;
1618 let ws_client = self.ws_client.clone();
1619 let ws_public_client = self.ws_public_client.clone();
1620 let setup_guard =
1621 TaskGroupGuard::new(&[&self.session_tasks, &self.command_tasks], move || {
1622 ws_client.begin_shutdown();
1623 ws_public_client.begin_shutdown();
1624 });
1625
1626 Self::refresh_instrument_catalog(
1627 &self.http_client,
1628 &self.config.instrument_provider,
1629 &self.instruments,
1630 &self.status_cache,
1631 &self.ws_client,
1632 &self.ws_public_client,
1633 &self.data_sender,
1634 self.clock,
1635 false,
1636 )
1637 .await?;
1638
1639 let session_result = async {
1640 log::info!("Connecting to Binance Futures market WebSocket...");
1641 self.ws_client.connect().await.map_err(|e| {
1642 log::error!("Binance Futures market WebSocket connection failed: {e:?}");
1643 anyhow::anyhow!("failed to connect Binance Futures market WebSocket: {e}")
1644 })?;
1645 log::info!("Binance Futures market WebSocket connected");
1646
1647 log::info!("Connecting to Binance Futures public WebSocket...");
1648 self.ws_public_client.connect().await.map_err(|e| {
1649 log::error!("Binance Futures public WebSocket connection failed: {e:?}");
1650 anyhow::anyhow!("failed to connect Binance Futures public WebSocket: {e}")
1651 })?;
1652 log::info!("Binance Futures public WebSocket connected");
1653
1654 let stream = self.ws_client.stream();
1655 let sender = self.data_sender.clone();
1656 let insts = self.instruments.clone();
1657 let ws_insts = self.ws_client.instruments_cache();
1658 let buffers = self.book_buffers.clone();
1659 let book_subs = self.book_subscriptions.clone();
1660 let book_unsubscribes_pending = self.book_unsubscribes_pending.clone();
1661 let l1_book_subs = self.l1_book_subscriptions.clone();
1662 let force_order_refs = self.force_order_refs.clone();
1663 let ticker_refs = self.ticker_refs.clone();
1664 let force_order_all_market_refs = self.force_order_all_market_refs.clone();
1665 let force_order_all_market_stream_active =
1666 self.force_order_all_market_stream_active.clone();
1667 let book_epoch = self.book_epoch.clone();
1668 let http = self.http_client.clone();
1669 let clock = self.clock;
1670 let cancel = self.cancellation_token.clone();
1671 let command_spawner = self
1672 .command_tasks
1673 .spawner()
1674 .context("Binance Futures command task admission is closed")?;
1675
1676 let future = async move {
1677 pin_mut!(stream);
1678
1679 loop {
1680 tokio::select! {
1681 Some(message) = stream.next() => {
1682 Self::handle_ws_message(
1683 message,
1684 &sender,
1685 &insts,
1686 &ws_insts,
1687 &buffers,
1688 &book_subs,
1689 &book_unsubscribes_pending,
1690 &l1_book_subs,
1691 &force_order_refs,
1692 &ticker_refs,
1693 &force_order_all_market_refs,
1694 &force_order_all_market_stream_active,
1695 &book_epoch,
1696 &http,
1697 clock,
1698 &command_spawner,
1699 );
1700 }
1701 () = cancel.cancelled() => {
1702 log::debug!("Market WebSocket stream task cancelled");
1703 break;
1704 }
1705 }
1706 }
1707 };
1708 self.session_tasks
1709 .spawn(future)
1710 .context("failed to register Binance Futures market stream task")?;
1711
1712 let pub_stream = self.ws_public_client.stream();
1713 let pub_sender = self.data_sender.clone();
1714 let pub_insts = self.instruments.clone();
1715 let pub_ws_insts = self.ws_public_client.instruments_cache();
1716 let pub_buffers = self.book_buffers.clone();
1717 let pub_book_subs = self.book_subscriptions.clone();
1718 let pub_book_unsubscribes_pending = self.book_unsubscribes_pending.clone();
1719 let pub_l1_book_subs = self.l1_book_subscriptions.clone();
1720 let pub_force_order_refs = self.force_order_refs.clone();
1721 let pub_ticker_refs = self.ticker_refs.clone();
1722 let pub_force_order_all_market_refs = self.force_order_all_market_refs.clone();
1723 let pub_force_order_all_market_stream_active =
1724 self.force_order_all_market_stream_active.clone();
1725 let pub_book_epoch = self.book_epoch.clone();
1726 let pub_http = self.http_client.clone();
1727 let pub_cancel = self.cancellation_token.clone();
1728 let pub_command_spawner = self
1729 .command_tasks
1730 .spawner()
1731 .context("Binance Futures command task admission is closed")?;
1732
1733 let future = async move {
1734 pin_mut!(pub_stream);
1735
1736 loop {
1737 tokio::select! {
1738 Some(message) = pub_stream.next() => {
1739 Self::handle_ws_message(
1740 message,
1741 &pub_sender,
1742 &pub_insts,
1743 &pub_ws_insts,
1744 &pub_buffers,
1745 &pub_book_subs,
1746 &pub_book_unsubscribes_pending,
1747 &pub_l1_book_subs,
1748 &pub_force_order_refs,
1749 &pub_ticker_refs,
1750 &pub_force_order_all_market_refs,
1751 &pub_force_order_all_market_stream_active,
1752 &pub_book_epoch,
1753 &pub_http,
1754 clock,
1755 &pub_command_spawner,
1756 );
1757 }
1758 () = pub_cancel.cancelled() => {
1759 log::debug!("Public WebSocket stream task cancelled");
1760 break;
1761 }
1762 }
1763 }
1764 };
1765 self.session_tasks
1766 .spawn(future)
1767 .context("failed to register Binance Futures public stream task")?;
1768
1769 let poll_secs = self.config.instrument_status_poll_secs;
1770 if poll_secs > 0 {
1771 let poll_http = self.http_client.clone();
1772 let poll_sender = self.data_sender.clone();
1773 let poll_instruments = self.instruments.clone();
1774 let poll_status_cache = self.status_cache.clone();
1775 let poll_cancel = self.cancellation_token.clone();
1776 let poll_clock = self.clock;
1777
1778 let future = async move {
1779 let mut interval =
1780 tokio::time::interval(tokio::time::Duration::from_secs(poll_secs));
1781 interval.tick().await; loop {
1784 tokio::select! {
1785 _ = interval.tick() => {
1786 match poll_http.request_symbol_statuses().await {
1787 Ok(symbol_statuses) => {
1788 let ts = poll_clock.get_time_ns();
1789 let inst_guard = poll_instruments.load();
1790
1791 let raw_to_id: AHashMap<Ustr, InstrumentId> = inst_guard
1792 .values()
1793 .map(|inst| (inst.raw_symbol().inner(), inst.id()))
1794 .collect();
1795
1796 let mut new_statuses = AHashMap::new();
1797
1798 for (raw_symbol, action) in &symbol_statuses {
1799 if let Some(&id) = raw_to_id.get(raw_symbol) {
1800 new_statuses.insert(id, *action);
1801 }
1802 }
1803 drop(inst_guard);
1804
1805 let mut cache = (**poll_status_cache.load()).clone();
1806 diff_and_emit_statuses(
1807 &new_statuses, &mut cache, &poll_sender, ts, ts,
1808 );
1809 poll_status_cache.store(cache);
1810 }
1811 Err(e) => {
1812 log::warn!("Futures instrument status poll failed: {e}");
1813 }
1814 }
1815 }
1816 () = poll_cancel.cancelled() => {
1817 log::debug!("Futures instrument status polling task cancelled");
1818 break;
1819 }
1820 }
1821 }
1822 };
1823 self.session_tasks
1824 .spawn(future)
1825 .context("failed to register Binance Futures status polling task")?;
1826 log::debug!("Futures instrument status polling started: interval={poll_secs}s");
1827 }
1828
1829 let refresh_secs = self.config.instrument_refresh_interval_secs;
1830 if refresh_secs > 0 {
1831 let http = self.http_client.clone();
1832 let provider = self.config.instrument_provider.clone();
1833 let instruments = self.instruments.clone();
1834 let statuses = self.status_cache.clone();
1835 let ws = self.ws_client.clone();
1836 let ws_public = self.ws_public_client.clone();
1837 let sender = self.data_sender.clone();
1838 let clock = self.clock;
1839 let cancel = self.cancellation_token.clone();
1840
1841 let future = async move {
1842 let mut interval = tokio::time::interval(Duration::from_secs(refresh_secs));
1843 interval.tick().await;
1844
1845 loop {
1846 tokio::select! {
1847 _ = interval.tick() => {
1848 if let Err(e) = Self::refresh_instrument_catalog(
1849 &http,
1850 &provider,
1851 &instruments,
1852 &statuses,
1853 &ws,
1854 &ws_public,
1855 &sender,
1856 clock,
1857 true,
1858 ).await {
1859 log::warn!("Binance Futures instrument refresh failed: {e}");
1860 }
1861 }
1862 () = cancel.cancelled() => {
1863 log::debug!("Binance Futures instrument refresh task cancelled");
1864 break;
1865 }
1866 }
1867 }
1868 };
1869 self.session_tasks
1870 .spawn(future)
1871 .context("failed to register Binance Futures instrument refresh task")?;
1872 log::debug!("Futures instrument refresh started: interval={refresh_secs}s");
1873 }
1874
1875 Ok::<(), anyhow::Error>(())
1876 }
1877 .await;
1878
1879 if let Err(e) = session_result {
1880 if let Err(teardown_error) = self.teardown_partial_connect().await {
1881 return Err(e.context(format!(
1882 "Binance Futures data startup teardown failed: {teardown_error}"
1883 )));
1884 }
1885 return Err(e);
1886 }
1887
1888 setup_guard.disarm();
1889 self.is_connected.store(true, Ordering::Release);
1890 log::info!("Connected: client_id={}", self.client_id);
1891 Ok(())
1892 }
1893
1894 async fn disconnect(&mut self) -> anyhow::Result<()> {
1895 self.teardown_partial_connect().await?;
1896
1897 self.mark_price_refs.store(AHashMap::new());
1899 self.ticker_refs.store(AHashMap::new());
1900 self.force_order_refs.store(AHashMap::new());
1901 self.force_order_all_market_refs.store(0, Ordering::Relaxed);
1902 self.force_order_all_market_stream_active
1903 .store(false, Ordering::Release);
1904 self.book_subscriptions.store(AHashMap::new());
1905 self.book_unsubscribes_pending.store(AHashMap::new());
1906 self.l1_book_subscriptions.store(AHashMap::new());
1907 self.quote_refs.store(AHashMap::new());
1908 self.book_buffers.store(AHashMap::new());
1909
1910 self.is_connected.store(false, Ordering::Release);
1911 log::info!("Disconnected: client_id={}", self.client_id);
1912 Ok(())
1913 }
1914
1915 fn is_connected(&self) -> bool {
1916 self.is_connected.load(Ordering::Relaxed)
1917 }
1918
1919 fn is_disconnected(&self) -> bool {
1920 !self.is_connected()
1921 }
1922
1923 fn subscribe(&mut self, cmd: SubscribeCustomData) -> anyhow::Result<()> {
1924 let data_type = cmd.data_type.type_name();
1925 if data_type == "BinanceFuturesTicker" {
1926 return subscribe_ticker(self, &cmd.data_type);
1927 }
1928
1929 if data_type == "BinanceFuturesMarkPriceUpdate" {
1930 let instrument_id = Self::required_instrument_id_metadata(&cmd.data_type)?;
1931 anyhow::ensure!(
1932 instrument_id.venue == self.venue(),
1933 "Futures mark price requires a BINANCE instrument"
1934 );
1935 let should_subscribe = {
1936 let previous = self
1937 .mark_price_refs
1938 .load()
1939 .get(&instrument_id)
1940 .copied()
1941 .unwrap_or(0);
1942 self.mark_price_refs
1943 .rcu(|refs| *refs.entry(instrument_id).or_insert(0) += 1);
1944 previous == 0
1945 };
1946
1947 if should_subscribe {
1948 let ws = self.ws_client.clone();
1949 let stream = format!(
1950 "{}@markPrice@1s",
1951 format_binance_stream_symbol(&instrument_id)
1952 );
1953 self.spawn_ws(
1954 async move {
1955 ws.subscribe(vec![stream])
1956 .await
1957 .context("mark price custom subscription")
1958 },
1959 "mark price custom subscription",
1960 );
1961 }
1962 return Ok(());
1963 }
1964
1965 if data_type != "BinanceFuturesLiquidation" {
1966 log::warn!("Unsupported custom data subscription: {data_type}");
1967 return Ok(());
1968 }
1969
1970 let instrument_id = Self::custom_liquidation_instrument_id(&cmd.data_type)?;
1971 if let Some(instrument_id) = instrument_id {
1972 if instrument_id.venue != self.venue() {
1973 anyhow::bail!(
1974 "Binance liquidation custom data requires BINANCE venue instrument, received {instrument_id}"
1975 );
1976 }
1977
1978 let should_subscribe = {
1979 let prev = self
1980 .force_order_refs
1981 .load()
1982 .get(&instrument_id)
1983 .copied()
1984 .unwrap_or(0);
1985 self.force_order_refs.rcu(|m| {
1986 let count = m.entry(instrument_id).or_insert(0);
1987 *count += 1;
1988 });
1989 prev == 0
1990 };
1991
1992 let has_all_market_subscription =
1993 self.force_order_all_market_refs.load(Ordering::Relaxed) > 0;
1994 let has_all_market_stream = self
1995 .force_order_all_market_stream_active
1996 .load(Ordering::Acquire);
1997
1998 if should_subscribe && !has_all_market_subscription && !has_all_market_stream {
1999 let ws = self.ws_client.clone();
2000 let stream = Self::liquidation_stream(&instrument_id);
2001 self.spawn_ws(
2002 async move {
2003 ws.subscribe(vec![stream])
2004 .await
2005 .context("forceOrder subscription")
2006 },
2007 "forceOrder subscription",
2008 );
2009 } else if should_subscribe && !has_all_market_subscription {
2010 self.spawn_liquidation_stream_reconcile("forceOrder subscription restore");
2011 }
2012
2013 return Ok(());
2014 }
2015
2016 let should_subscribe = self
2017 .force_order_all_market_refs
2018 .fetch_add(1, Ordering::Relaxed)
2019 == 0;
2020
2021 if should_subscribe {
2022 self.spawn_liquidation_stream_reconcile("all-market forceOrder subscription");
2023 }
2024
2025 Ok(())
2026 }
2027
2028 fn subscribe_instruments(&mut self, _cmd: SubscribeInstruments) -> anyhow::Result<()> {
2029 log::debug!(
2030 "subscribe_instruments: Binance Futures instruments are fetched via HTTP on connect"
2031 );
2032 Ok(())
2033 }
2034
2035 fn subscribe_instrument(&mut self, _cmd: SubscribeInstrument) -> anyhow::Result<()> {
2036 log::debug!(
2037 "subscribe_instrument: Binance Futures instruments are fetched via HTTP on connect"
2038 );
2039 Ok(())
2040 }
2041
2042 fn subscribe_book_deltas(&mut self, cmd: SubscribeBookDeltas) -> anyhow::Result<()> {
2043 if cmd.book_type == BookType::L1_MBP {
2044 anyhow::ensure!(
2045 cmd.depth.is_none_or(|depth| depth.get() == 1),
2046 "Binance Futures L1_MBP supports depth 1 only"
2047 );
2048 anyhow::ensure!(
2049 !self.book_subscriptions.contains_key(&cmd.instrument_id),
2050 "cannot subscribe L1_MBP and L2_MBP for the same Binance Futures instrument"
2051 );
2052 self.l1_book_subscriptions.rcu(|subscriptions| {
2053 *subscriptions.entry(cmd.instrument_id).or_insert(0) += 1;
2054 });
2055 self.subscribe_top_of_book(cmd.instrument_id);
2056 return Ok(());
2057 }
2058
2059 if cmd.book_type != BookType::L2_MBP {
2060 anyhow::bail!("Binance Futures supports L1_MBP and L2_MBP order book subscriptions");
2061 }
2062 anyhow::ensure!(
2063 !self.l1_book_subscriptions.contains_key(&cmd.instrument_id),
2064 "cannot subscribe L1_MBP and L2_MBP for the same Binance Futures instrument"
2065 );
2066
2067 let instrument_id = cmd.instrument_id;
2068 let depth = cmd.depth.map_or(1000, |d| d.get() as u32);
2069
2070 if !BINANCE_BOOK_DEPTHS.contains(&depth) {
2071 anyhow::bail!(
2072 "Invalid depth {depth} for Binance Futures order book. \
2073 Valid values: {BINANCE_BOOK_DEPTHS:?}"
2074 );
2075 }
2076
2077 if let Some(existing) = self.book_subscriptions.load().get(&instrument_id) {
2078 anyhow::ensure!(
2079 *existing == depth,
2080 "Binance Futures book depth cannot change while subscribed"
2081 );
2082 }
2083
2084 let stream = book_stream(&instrument_id, depth);
2087 revive_book_drains(&self.book_unsubscribes_pending, instrument_id, &stream);
2088
2089 self.book_subscriptions.insert(instrument_id, depth);
2090
2091 satisfy_book_drain(&self.book_unsubscribes_pending, instrument_id, &stream);
2094
2095 if is_partial_book_depth(depth) {
2096 let ws = self.ws_public_client.clone();
2097 let (wait, done) = self.chain_book_command();
2098 self.spawn_ws(
2099 async move {
2100 wait.await;
2101 let result = ws
2102 .subscribe(vec![stream])
2103 .await
2104 .context("book deltas subscription");
2105 let _ = done.send(());
2106 result
2107 },
2108 "order book subscription",
2109 );
2110 return Ok(());
2111 }
2112
2113 let epoch = {
2115 let mut guard = self.book_epoch.write();
2116 *guard = guard.wrapping_add(1);
2117 *guard
2118 };
2119
2120 self.book_buffers
2121 .insert(instrument_id, BookBuffer::new(epoch));
2122
2123 log::debug!("OrderBook snapshot rebuild for {instrument_id} @ depth {depth} starting");
2124
2125 let ws = self.ws_public_client.clone();
2127 let (wait, done) = self.chain_book_command();
2128
2129 self.spawn_ws(
2130 async move {
2131 wait.await;
2132 let result = ws
2133 .subscribe(vec![stream])
2134 .await
2135 .context("book deltas subscription");
2136 let _ = done.send(());
2137 result
2138 },
2139 "order book subscription",
2140 );
2141
2142 let http = self.http_client.clone();
2143 let sender = self.data_sender.clone();
2144 let buffers = self.book_buffers.clone();
2145 let instruments = self.instruments.clone();
2146 let clock = self.clock;
2147
2148 self.spawn_command(async move {
2149 Self::fetch_and_emit_snapshot(
2150 http,
2151 sender,
2152 buffers,
2153 instruments,
2154 instrument_id,
2155 depth,
2156 epoch,
2157 clock,
2158 )
2159 .await;
2160 });
2161
2162 Ok(())
2163 }
2164
2165 fn subscribe_quotes(&mut self, cmd: SubscribeQuotes) -> anyhow::Result<()> {
2166 self.subscribe_top_of_book(cmd.instrument_id);
2167 Ok(())
2168 }
2169
2170 fn subscribe_trades(&mut self, cmd: SubscribeTrades) -> anyhow::Result<()> {
2171 let instrument_id = cmd.instrument_id;
2172 let ws = self.ws_client.clone();
2173
2174 let stream = format!("{}@aggTrade", format_binance_stream_symbol(&instrument_id));
2176
2177 self.spawn_ws(
2178 async move {
2179 ws.subscribe(vec![stream])
2180 .await
2181 .context("trades subscription")
2182 },
2183 "trade subscription",
2184 );
2185 Ok(())
2186 }
2187
2188 fn subscribe_bars(&mut self, cmd: SubscribeBars) -> anyhow::Result<()> {
2189 let bar_type = cmd.bar_type;
2190 let ws = self.ws_client.clone();
2191 let interval = bar_spec_to_binance_interval(bar_type.spec())?;
2192 anyhow::ensure!(
2193 interval != crate::common::enums::BinanceKlineInterval::Second1,
2194 "Binance Futures does not support second-level kline intervals"
2195 );
2196
2197 let stream = format!(
2198 "{}@kline_{}",
2199 format_binance_stream_symbol(&bar_type.instrument_id()),
2200 interval.as_str()
2201 );
2202
2203 self.spawn_ws(
2204 async move {
2205 ws.subscribe(vec![stream])
2206 .await
2207 .context("bars subscription")
2208 },
2209 "bar subscription",
2210 );
2211 Ok(())
2212 }
2213
2214 fn subscribe_mark_prices(&mut self, cmd: SubscribeMarkPrices) -> anyhow::Result<()> {
2215 let instrument_id = cmd.instrument_id;
2216
2217 let should_subscribe = {
2218 let prev = self
2219 .mark_price_refs
2220 .load()
2221 .get(&instrument_id)
2222 .copied()
2223 .unwrap_or(0);
2224 self.mark_price_refs.rcu(|m| {
2225 let count = m.entry(instrument_id).or_insert(0);
2226 *count += 1;
2227 });
2228 prev == 0
2229 };
2230
2231 if should_subscribe {
2232 let ws = self.ws_client.clone();
2233 let stream = format!(
2234 "{}@markPrice@1s",
2235 format_binance_stream_symbol(&instrument_id)
2236 );
2237
2238 self.spawn_ws(
2239 async move {
2240 ws.subscribe(vec![stream])
2241 .await
2242 .context("mark prices subscription")
2243 },
2244 "mark prices subscription",
2245 );
2246 }
2247 Ok(())
2248 }
2249
2250 fn subscribe_index_prices(&mut self, cmd: SubscribeIndexPrices) -> anyhow::Result<()> {
2251 let instrument_id = cmd.instrument_id;
2252
2253 let should_subscribe = {
2254 let prev = self
2255 .mark_price_refs
2256 .load()
2257 .get(&instrument_id)
2258 .copied()
2259 .unwrap_or(0);
2260 self.mark_price_refs.rcu(|m| {
2261 let count = m.entry(instrument_id).or_insert(0);
2262 *count += 1;
2263 });
2264 prev == 0
2265 };
2266
2267 if should_subscribe {
2268 let ws = self.ws_client.clone();
2269 let stream = format!(
2270 "{}@markPrice@1s",
2271 format_binance_stream_symbol(&instrument_id)
2272 );
2273
2274 self.spawn_ws(
2275 async move {
2276 ws.subscribe(vec![stream])
2277 .await
2278 .context("index prices subscription")
2279 },
2280 "index prices subscription",
2281 );
2282 }
2283 Ok(())
2284 }
2285
2286 fn subscribe_funding_rates(&mut self, cmd: SubscribeFundingRates) -> anyhow::Result<()> {
2287 let instrument_id = cmd.instrument_id;
2288
2289 let should_subscribe = {
2290 let prev = self
2291 .mark_price_refs
2292 .load()
2293 .get(&instrument_id)
2294 .copied()
2295 .unwrap_or(0);
2296 self.mark_price_refs.rcu(|m| {
2297 let count = m.entry(instrument_id).or_insert(0);
2298 *count += 1;
2299 });
2300 prev == 0
2301 };
2302
2303 if should_subscribe {
2304 let ws = self.ws_client.clone();
2305 let stream = format!(
2306 "{}@markPrice@1s",
2307 format_binance_stream_symbol(&instrument_id)
2308 );
2309
2310 self.spawn_ws(
2311 async move {
2312 ws.subscribe(vec![stream])
2313 .await
2314 .context("funding rates subscription")
2315 },
2316 "funding rates subscription",
2317 );
2318 }
2319 Ok(())
2320 }
2321
2322 fn subscribe_instrument_status(
2323 &mut self,
2324 cmd: SubscribeInstrumentStatus,
2325 ) -> anyhow::Result<()> {
2326 log::debug!(
2327 "subscribe_instrument_status: {id} (status changes detected via periodic exchange info polling)",
2328 id = cmd.instrument_id,
2329 );
2330 Ok(())
2331 }
2332
2333 fn unsubscribe_book_deltas(&mut self, cmd: &UnsubscribeBookDeltas) -> anyhow::Result<()> {
2334 let instrument_id = cmd.instrument_id;
2335
2336 if let Some(count) = self
2337 .l1_book_subscriptions
2338 .load()
2339 .get(&instrument_id)
2340 .copied()
2341 {
2342 if count == 1 {
2343 self.l1_book_subscriptions.remove(&instrument_id);
2344 } else {
2345 self.l1_book_subscriptions.rcu(|subscriptions| {
2346 if let Some(existing) = subscriptions.get_mut(&instrument_id) {
2347 *existing -= 1;
2348 }
2349 });
2350 }
2351 self.unsubscribe_top_of_book(instrument_id);
2352 return Ok(());
2353 }
2354 let ws = self.ws_public_client.clone();
2355
2356 let Some(depth) = self.book_subscriptions.load().get(&instrument_id).copied() else {
2357 return Ok(());
2358 };
2359 self.book_subscriptions.remove(&instrument_id);
2360
2361 self.book_buffers.remove(&instrument_id);
2363
2364 let stream = book_stream(&instrument_id, depth);
2365 let generation = self.book_drain_generation;
2366 self.book_drain_generation += 1;
2367
2368 arm_book_drain(
2372 &self.book_unsubscribes_pending,
2373 instrument_id,
2374 generation,
2375 &stream,
2376 );
2377
2378 let book_unsubscribes_pending = self.book_unsubscribes_pending.clone();
2379 let (wait, done) = self.chain_book_command();
2380 self.spawn_ws(
2381 async move {
2382 wait.await;
2383 let sent = ws
2384 .unsubscribe_correlated(vec![stream.clone()], generation)
2385 .await
2386 .context("book deltas unsubscribe");
2387
2388 let result = match sent {
2389 Ok(sent) if sent.contains(&stream) => Ok(()),
2390 Ok(_) => {
2391 remove_book_drain(&book_unsubscribes_pending, generation);
2392 Ok(())
2393 }
2394 Err(e) => {
2395 remove_book_drain(&book_unsubscribes_pending, generation);
2396 Err(e)
2397 }
2398 };
2399 let _ = done.send(());
2400 result
2401 },
2402 "order book unsubscribe",
2403 );
2404 Ok(())
2405 }
2406
2407 fn unsubscribe_quotes(&mut self, cmd: &UnsubscribeQuotes) -> anyhow::Result<()> {
2408 self.unsubscribe_top_of_book(cmd.instrument_id);
2409 Ok(())
2410 }
2411
2412 fn unsubscribe_trades(&mut self, cmd: &UnsubscribeTrades) -> anyhow::Result<()> {
2413 let instrument_id = cmd.instrument_id;
2414 let ws = self.ws_client.clone();
2415
2416 let stream = format!("{}@aggTrade", format_binance_stream_symbol(&instrument_id));
2417
2418 self.spawn_ws(
2419 async move {
2420 ws.unsubscribe(vec![stream])
2421 .await
2422 .context("trades unsubscribe")
2423 .map(|_| ())
2424 },
2425 "trade unsubscribe",
2426 );
2427 Ok(())
2428 }
2429
2430 fn unsubscribe(&mut self, cmd: &UnsubscribeCustomData) -> anyhow::Result<()> {
2431 let data_type = cmd.data_type.type_name();
2432 if data_type == "BinanceFuturesTicker" {
2433 return unsubscribe_ticker(self, &cmd.data_type);
2434 }
2435
2436 if data_type == "BinanceFuturesMarkPriceUpdate" {
2437 let instrument_id = Self::required_instrument_id_metadata(&cmd.data_type)?;
2438 let should_unsubscribe = match self.mark_price_refs.load().get(&instrument_id).copied()
2439 {
2440 Some(1) => {
2441 self.mark_price_refs.remove(&instrument_id);
2442 true
2443 }
2444 Some(count) if count > 1 => {
2445 self.mark_price_refs.rcu(|refs| {
2446 if let Some(existing) = refs.get_mut(&instrument_id) {
2447 *existing -= 1;
2448 }
2449 });
2450 false
2451 }
2452 _ => false,
2453 };
2454
2455 if should_unsubscribe {
2456 let ws = self.ws_client.clone();
2457 let stream = format!(
2458 "{}@markPrice@1s",
2459 format_binance_stream_symbol(&instrument_id)
2460 );
2461 self.spawn_ws(
2462 async move {
2463 ws.unsubscribe(vec![stream])
2464 .await
2465 .context("mark price custom unsubscribe")
2466 .map(|_| ())
2467 },
2468 "mark price custom unsubscribe",
2469 );
2470 }
2471 return Ok(());
2472 }
2473
2474 if data_type != "BinanceFuturesLiquidation" {
2475 log::warn!("Unsupported custom data unsubscription: {data_type}");
2476 return Ok(());
2477 }
2478
2479 let instrument_id = Self::custom_liquidation_instrument_id(&cmd.data_type)?;
2480 if let Some(instrument_id) = instrument_id {
2481 if instrument_id.venue != self.venue() {
2482 anyhow::bail!(
2483 "Binance liquidation custom data requires BINANCE venue instrument, received {instrument_id}"
2484 );
2485 }
2486
2487 let should_unsubscribe = {
2488 let prev = self.force_order_refs.load().get(&instrument_id).copied();
2489 match prev {
2490 Some(1) => {
2491 self.force_order_refs.remove(&instrument_id);
2492 true
2493 }
2494 Some(count) if count > 1 => {
2495 self.force_order_refs.rcu(|m| {
2496 if let Some(existing) = m.get_mut(&instrument_id) {
2497 *existing -= 1;
2498 }
2499 });
2500 false
2501 }
2502 _ => false,
2503 }
2504 };
2505
2506 let has_all_market_subscription =
2507 self.force_order_all_market_refs.load(Ordering::Relaxed) > 0;
2508 let has_all_market_stream = self
2509 .force_order_all_market_stream_active
2510 .load(Ordering::Acquire);
2511
2512 if should_unsubscribe && !has_all_market_subscription {
2513 let ws = self.ws_client.clone();
2514 let stream = Self::liquidation_stream(&instrument_id);
2515 let ws_lock = self.force_order_ws_lock.clone();
2516 self.spawn_ws(
2517 async move {
2518 let _guard = if has_all_market_stream {
2519 Some(ws_lock.lock().await)
2520 } else {
2521 None
2522 };
2523 ws.unsubscribe(vec![stream])
2524 .await
2525 .context("forceOrder unsubscribe")
2526 .map(|_| ())
2527 },
2528 "forceOrder unsubscribe",
2529 );
2530 }
2531
2532 return Ok(());
2533 }
2534
2535 let should_unsubscribe = self
2536 .force_order_all_market_refs
2537 .try_update(Ordering::Relaxed, Ordering::Relaxed, |current| {
2538 if current == 0 {
2539 None
2540 } else {
2541 Some(current - 1)
2542 }
2543 })
2544 .is_ok_and(|prev| prev == 1);
2545
2546 if should_unsubscribe {
2547 self.spawn_liquidation_stream_reconcile("all-market forceOrder unsubscribe");
2548 }
2549
2550 Ok(())
2551 }
2552
2553 fn unsubscribe_bars(&mut self, cmd: &UnsubscribeBars) -> anyhow::Result<()> {
2554 let bar_type = cmd.bar_type;
2555 let ws = self.ws_client.clone();
2556 let interval = bar_spec_to_binance_interval(bar_type.spec())?;
2557
2558 let stream = format!(
2559 "{}@kline_{}",
2560 format_binance_stream_symbol(&bar_type.instrument_id()),
2561 interval.as_str()
2562 );
2563
2564 self.spawn_ws(
2565 async move {
2566 ws.unsubscribe(vec![stream])
2567 .await
2568 .context("bars unsubscribe")
2569 .map(|_| ())
2570 },
2571 "bar unsubscribe",
2572 );
2573 Ok(())
2574 }
2575
2576 fn unsubscribe_mark_prices(&mut self, cmd: &UnsubscribeMarkPrices) -> anyhow::Result<()> {
2577 let instrument_id = cmd.instrument_id;
2578
2579 let should_unsubscribe = {
2580 let prev = self.mark_price_refs.load().get(&instrument_id).copied();
2581 match prev {
2582 Some(count) if count <= 1 => {
2583 self.mark_price_refs.remove(&instrument_id);
2584 true
2585 }
2586 Some(_) => {
2587 self.mark_price_refs.rcu(|m| {
2588 if let Some(count) = m.get_mut(&instrument_id) {
2589 *count = count.saturating_sub(1);
2590 }
2591 });
2592 false
2593 }
2594 None => false,
2595 }
2596 };
2597
2598 if should_unsubscribe {
2599 let ws = self.ws_client.clone();
2600 let symbol_lower = format_binance_stream_symbol(&instrument_id);
2601 let streams = vec![
2602 format!("{symbol_lower}@markPrice"),
2603 format!("{symbol_lower}@markPrice@1s"),
2604 format!("{symbol_lower}@markPrice@3s"),
2605 ];
2606
2607 self.spawn_ws(
2608 async move {
2609 ws.unsubscribe(streams)
2610 .await
2611 .context("mark prices unsubscribe")
2612 .map(|_| ())
2613 },
2614 "mark prices unsubscribe",
2615 );
2616 }
2617 Ok(())
2618 }
2619
2620 fn unsubscribe_index_prices(&mut self, cmd: &UnsubscribeIndexPrices) -> anyhow::Result<()> {
2621 let instrument_id = cmd.instrument_id;
2622
2623 let should_unsubscribe = {
2624 let prev = self.mark_price_refs.load().get(&instrument_id).copied();
2625 match prev {
2626 Some(count) if count <= 1 => {
2627 self.mark_price_refs.remove(&instrument_id);
2628 true
2629 }
2630 Some(_) => {
2631 self.mark_price_refs.rcu(|m| {
2632 if let Some(count) = m.get_mut(&instrument_id) {
2633 *count = count.saturating_sub(1);
2634 }
2635 });
2636 false
2637 }
2638 None => false,
2639 }
2640 };
2641
2642 if should_unsubscribe {
2643 let ws = self.ws_client.clone();
2644 let symbol_lower = format_binance_stream_symbol(&instrument_id);
2645 let streams = vec![
2646 format!("{symbol_lower}@markPrice"),
2647 format!("{symbol_lower}@markPrice@1s"),
2648 format!("{symbol_lower}@markPrice@3s"),
2649 ];
2650
2651 self.spawn_ws(
2652 async move {
2653 ws.unsubscribe(streams)
2654 .await
2655 .context("index prices unsubscribe")
2656 .map(|_| ())
2657 },
2658 "index prices unsubscribe",
2659 );
2660 }
2661 Ok(())
2662 }
2663
2664 fn unsubscribe_funding_rates(&mut self, cmd: &UnsubscribeFundingRates) -> anyhow::Result<()> {
2665 let instrument_id = cmd.instrument_id;
2666
2667 let should_unsubscribe = {
2668 let prev = self.mark_price_refs.load().get(&instrument_id).copied();
2669 match prev {
2670 Some(count) if count <= 1 => {
2671 self.mark_price_refs.remove(&instrument_id);
2672 true
2673 }
2674 Some(_) => {
2675 self.mark_price_refs.rcu(|m| {
2676 if let Some(count) = m.get_mut(&instrument_id) {
2677 *count = count.saturating_sub(1);
2678 }
2679 });
2680 false
2681 }
2682 None => false,
2683 }
2684 };
2685
2686 if should_unsubscribe {
2687 let ws = self.ws_client.clone();
2688 let symbol_lower = format_binance_stream_symbol(&instrument_id);
2689 let streams = vec![
2690 format!("{symbol_lower}@markPrice"),
2691 format!("{symbol_lower}@markPrice@1s"),
2692 format!("{symbol_lower}@markPrice@3s"),
2693 ];
2694
2695 self.spawn_ws(
2696 async move {
2697 ws.unsubscribe(streams)
2698 .await
2699 .context("funding rates unsubscribe")
2700 .map(|_| ())
2701 },
2702 "funding rates unsubscribe",
2703 );
2704 }
2705 Ok(())
2706 }
2707
2708 fn unsubscribe_instrument_status(
2709 &mut self,
2710 cmd: &UnsubscribeInstrumentStatus,
2711 ) -> anyhow::Result<()> {
2712 log::debug!(
2713 "unsubscribe_instrument_status: {id}",
2714 id = cmd.instrument_id,
2715 );
2716 Ok(())
2717 }
2718
2719 fn request_instruments(&self, request: RequestInstruments) -> anyhow::Result<()> {
2720 let http = self.http_client.clone();
2721 let sender = self.data_sender.clone();
2722 let instruments_cache = self.instruments.clone();
2723 let request_id = request.request_id;
2724 let client_id = request.client_id.unwrap_or(self.client_id);
2725 let venue = self.venue();
2726 let start = request.start;
2727 let end = request.end;
2728 let params = request.params;
2729 let clock = self.clock;
2730 let provider = self.config.instrument_provider.clone();
2731 let start_nanos = datetime_to_unix_nanos(start);
2732 let end_nanos = datetime_to_unix_nanos(end);
2733
2734 self.spawn_command(async move {
2735 match http.request_instruments_with_config(&provider).await {
2736 Ok(instruments) => {
2737 for instrument in &instruments {
2738 upsert_instrument(&instruments_cache, instrument.clone());
2739 }
2740
2741 let response = DataResponse::Instruments(InstrumentsResponse::new(
2742 request_id,
2743 client_id,
2744 venue,
2745 instruments,
2746 start_nanos,
2747 end_nanos,
2748 clock.get_time_ns(),
2749 params,
2750 ));
2751
2752 if let Err(e) = sender.send(DataEvent::Response(response)) {
2753 log::error!("Failed to send instruments response: {e}");
2754 }
2755 }
2756 Err(e) => log::error!("Instruments request failed: {e:?}"),
2757 }
2758 });
2759
2760 Ok(())
2761 }
2762
2763 fn request_instrument(&self, request: RequestInstrument) -> anyhow::Result<()> {
2764 let http = self.http_client.clone();
2765 let sender = self.data_sender.clone();
2766 let instruments = self.instruments.clone();
2767 let instrument_id = request.instrument_id;
2768 let request_id = request.request_id;
2769 let client_id = request.client_id.unwrap_or(self.client_id);
2770 let start = request.start;
2771 let end = request.end;
2772 let params = request.params;
2773 let clock = self.clock;
2774 let provider = self.config.instrument_provider.clone();
2775 let start_nanos = datetime_to_unix_nanos(start);
2776 let end_nanos = datetime_to_unix_nanos(end);
2777
2778 self.spawn_command(async move {
2779 match http.request_instruments_with_config(&provider).await {
2780 Ok(all_instruments) => {
2781 for instrument in &all_instruments {
2782 upsert_instrument(&instruments, instrument.clone());
2783 }
2784
2785 let instrument = all_instruments
2786 .into_iter()
2787 .find(|i| i.id() == instrument_id);
2788
2789 if let Some(instrument) = instrument {
2790 let response = DataResponse::Instrument(Box::new(InstrumentResponse::new(
2791 request_id,
2792 client_id,
2793 instrument.id(),
2794 instrument,
2795 start_nanos,
2796 end_nanos,
2797 clock.get_time_ns(),
2798 params,
2799 )));
2800
2801 if let Err(e) = sender.send(DataEvent::Response(response)) {
2802 log::error!("Failed to send instrument response: {e}");
2803 }
2804 } else {
2805 log::error!("Instrument not found: {instrument_id}");
2806 }
2807 }
2808 Err(e) => log::error!("Instrument request failed: {e:?}"),
2809 }
2810 });
2811
2812 Ok(())
2813 }
2814
2815 fn request_data(&self, request: RequestCustomData) -> anyhow::Result<()> {
2820 let data_type = request.data_type.clone();
2821 let data_type_name = data_type.type_name().to_string();
2822
2823 if data_type_name == "BinanceBar" {
2824 let bar_type = parse_binance_bar_type(&data_type)?;
2825 anyhow::ensure!(
2826 bar_type.aggregation_source() == AggregationSource::External,
2827 "historical BinanceBar requests require EXTERNAL aggregation"
2828 );
2829 anyhow::ensure!(
2830 bar_type.spec().price_type == PriceType::Last,
2831 "historical BinanceBar requests require LAST price type"
2832 );
2833 anyhow::ensure!(
2834 bar_type.spec().is_time_aggregated(),
2835 "historical BinanceBar requests require time aggregation"
2836 );
2837 let http = self.http_client.clone();
2838 let sender = self.data_sender.clone();
2839 let request_id = request.request_id;
2840 let client_id = request.client_id;
2841 let start = request.start;
2842 let end = request.end;
2843 let limit = request.limit.map(|value| value.get() as u32);
2844 let params = request.params;
2845 let clock = self.clock;
2846 let venue = self.venue();
2847 let start_nanos = datetime_to_unix_nanos(start);
2848 let end_nanos = datetime_to_unix_nanos(end);
2849 self.spawn_command(async move {
2850 match http.request_binance_bars(bar_type, start, end, limit).await {
2851 Ok(bars) => {
2852 let response = DataResponse::Data(CustomDataResponse::new(
2853 request_id,
2854 client_id,
2855 Some(venue),
2856 data_type,
2857 binance_bars_to_custom_data(bar_type, bars),
2858 start_nanos,
2859 end_nanos,
2860 clock.get_time_ns(),
2861 params,
2862 ));
2863
2864 if let Err(e) = sender.send(DataEvent::Response(response)) {
2865 log::error!("Failed to send BinanceBar response: {e}");
2866 }
2867 }
2868 Err(e) => log::error!("BinanceBar request failed for {bar_type}: {e:?}"),
2869 }
2870 });
2871 return Ok(());
2872 }
2873
2874 if data_type_name != "BinanceFuturesOpenInterest"
2875 && data_type_name != "BinanceFuturesOpenInterestHist"
2876 {
2877 log::warn!("Unsupported custom data request: {data_type_name}");
2878 return Ok(());
2879 }
2880
2881 let instrument_id = Self::required_instrument_id_metadata(&data_type)?;
2882
2883 if instrument_id.venue != self.venue() {
2884 anyhow::bail!(
2885 "Binance Futures custom data requires BINANCE venue instrument, received {instrument_id}"
2886 );
2887 }
2888
2889 let period = if data_type_name == "BinanceFuturesOpenInterestHist" {
2890 Some(Self::required_period_metadata(&data_type)?)
2891 } else {
2892 None
2893 };
2894
2895 let http = self.http_client.clone();
2896 let sender = self.data_sender.clone();
2897 let request_id = request.request_id;
2898 let client_id = request.client_id;
2899 let params = request.params;
2900 let clock = self.clock;
2901 let venue = self.venue();
2902 let limit = request.limit.map(|n| n.get() as u32);
2903 let start_nanos = datetime_to_unix_nanos(request.start);
2904 let end_nanos = datetime_to_unix_nanos(request.end);
2905 let start_ms = request.start.map(|dt| dt.as_millisecond());
2906 let end_ms = request.end.map(|dt| dt.as_millisecond());
2907
2908 self.spawn_command(async move {
2909 let response = if data_type_name == "BinanceFuturesOpenInterest" {
2910 let response_data_type = data_type.clone();
2911 let query = BinanceOpenInterestParams {
2912 symbol: format_binance_symbol(&instrument_id),
2913 };
2914
2915 match http
2916 .open_interest(&query)
2917 .await
2918 .context("failed to request current open interest from Binance Futures")
2919 {
2920 Ok(open_interest) => {
2921 let ts_init = clock.get_time_ns();
2922 let open_interest_value = match Self::parse_open_interest_decimal(
2923 "open_interest",
2924 &open_interest.open_interest,
2925 ) {
2926 Ok(value) => value,
2927 Err(e) => {
2928 log::error!(
2929 "Current open interest request failed for {instrument_id}: {e:?}"
2930 );
2931 return;
2932 }
2933 };
2934 let ts_event = match parse_millis(
2935 open_interest.time,
2936 "Futures open interest time",
2937 ) {
2938 Ok(value) => value,
2939 Err(e) => {
2940 log::error!(
2941 "Current open interest request failed for {instrument_id}: {e:?}"
2942 );
2943 return;
2944 }
2945 };
2946 let payload = Arc::new(BinanceFuturesOpenInterest::new(
2947 instrument_id,
2948 open_interest_value,
2949 ts_event,
2950 ts_init,
2951 ));
2952 let custom = CustomData::new(payload, response_data_type.clone());
2953
2954 Some(DataResponse::Data(CustomDataResponse::new(
2955 request_id,
2956 client_id,
2957 Some(venue),
2958 response_data_type,
2959 custom,
2960 start_nanos,
2961 end_nanos,
2962 ts_init,
2963 params,
2964 )))
2965 }
2966 Err(e) => {
2967 log::error!("Current open interest request failed for {instrument_id}: {e:?}");
2968 None
2969 }
2970 }
2971 } else {
2972 let response_data_type = data_type.clone();
2973 let period = period.expect("period required for historical open interest");
2974 let query = match http.product_type() {
2975 BinanceProductType::UsdM => BinanceOpenInterestHistParams {
2976 symbol: Some(format_binance_symbol(&instrument_id)),
2977 pair: None,
2978 contract_type: None,
2979 period: period.clone(),
2980 start_time: start_ms,
2981 end_time: end_ms,
2982 limit,
2983 },
2984 BinanceProductType::CoinM => {
2985 let (pair, contract_type) =
2986 match Self::coinm_open_interest_hist_params(&http, &instrument_id) {
2987 Ok(values) => values,
2988 Err(e) => {
2989 log::error!(
2990 "Historical open interest request failed for {instrument_id}: {e:?}"
2991 );
2992 return;
2993 }
2994 };
2995 BinanceOpenInterestHistParams {
2996 symbol: None,
2997 pair: Some(pair),
2998 contract_type: Some(contract_type),
2999 period: period.clone(),
3000 start_time: start_ms,
3001 end_time: end_ms,
3002 limit,
3003 }
3004 }
3005 product_type => {
3006 log::error!(
3007 "Historical open interest request failed for {instrument_id}: unsupported product type {product_type:?}"
3008 );
3009 return;
3010 }
3011 };
3012
3013 match http
3014 .open_interest_hist(&query)
3015 .await
3016 .context("failed to request historical open interest from Binance Futures")
3017 {
3018 Ok(history) => {
3019 let ts_init = clock.get_time_ns();
3020 let points: Vec<BinanceFuturesOpenInterestHistPoint> = match history
3021 .into_iter()
3022 .map(|point| -> anyhow::Result<_> {
3023 Ok(BinanceFuturesOpenInterestHistPoint::new(
3024 Self::parse_open_interest_decimal(
3025 "sum_open_interest",
3026 &point.sum_open_interest,
3027 )?,
3028 Self::parse_open_interest_decimal(
3029 "sum_open_interest_value",
3030 &point.sum_open_interest_value,
3031 )?,
3032 parse_millis(
3033 point.timestamp,
3034 "Futures historical open interest timestamp",
3035 )?,
3036 ))
3037 })
3038 .collect()
3039 {
3040 Ok(points) => points,
3041 Err(e) => {
3042 log::error!(
3043 "Historical open interest request failed for {instrument_id}: {e:?}"
3044 );
3045 return;
3046 }
3047 };
3048 let ts_event = points.last().map_or(ts_init, |point| point.ts_event);
3049 let payload = Arc::new(BinanceFuturesOpenInterestHist::new(
3050 instrument_id,
3051 period,
3052 points,
3053 ts_event,
3054 ts_init,
3055 ));
3056 let custom = CustomData::new(payload, response_data_type.clone());
3057
3058 Some(DataResponse::Data(CustomDataResponse::new(
3059 request_id,
3060 client_id,
3061 Some(venue),
3062 response_data_type,
3063 custom,
3064 start_nanos,
3065 end_nanos,
3066 ts_init,
3067 params,
3068 )))
3069 }
3070 Err(e) => {
3071 log::error!(
3072 "Historical open interest request failed for {instrument_id}: {e:?}"
3073 );
3074 None
3075 }
3076 }
3077 };
3078
3079 if let Some(response) = response
3080 && let Err(e) = sender.send(DataEvent::Response(response))
3081 {
3082 log::error!("Failed to send custom data response: {e}");
3083 }
3084 });
3085
3086 Ok(())
3087 }
3088
3089 fn request_trades(&self, request: RequestTrades) -> anyhow::Result<()> {
3090 let http = self.http_client.clone();
3091 let sender = self.data_sender.clone();
3092 let instrument_id = request.instrument_id;
3093 let limit = request.limit.map(|n| n.get() as u32);
3094 let request_id = request.request_id;
3095 let client_id = request.client_id.unwrap_or(self.client_id);
3096 let params = request.params;
3097 let clock = self.clock;
3098 let start_nanos = datetime_to_unix_nanos(request.start);
3099 let end_nanos = datetime_to_unix_nanos(request.end);
3100 let start = request.start;
3101 let end = request.end;
3102 anyhow::ensure!(
3103 limit.is_none_or(|value| value <= 1000),
3104 "Binance Futures trade limit must not exceed 1000"
3105 );
3106
3107 self.spawn_command(async move {
3108 let result = if start.is_some() || end.is_some() {
3109 http.request_agg_trades(instrument_id, start, end, limit)
3110 .await
3111 } else {
3112 http.request_trades(instrument_id, limit).await
3113 };
3114
3115 match result.context("failed to request trades from Binance Futures") {
3116 Ok(trades) => {
3117 let response = DataResponse::Trades(TradesResponse::new(
3118 request_id,
3119 client_id,
3120 instrument_id,
3121 trades,
3122 start_nanos,
3123 end_nanos,
3124 clock.get_time_ns(),
3125 params,
3126 ));
3127
3128 if let Err(e) = sender.send(DataEvent::Response(response)) {
3129 log::error!("Failed to send trades response: {e}");
3130 }
3131 }
3132 Err(e) => log::error!("Trade request failed: {e:?}"),
3133 }
3134 });
3135
3136 Ok(())
3137 }
3138
3139 fn request_funding_rates(&self, request: RequestFundingRates) -> anyhow::Result<()> {
3140 let http = self.http_client.clone();
3141 let sender = self.data_sender.clone();
3142 let instrument_id = request.instrument_id;
3143 let start = request.start;
3144 let end = request.end;
3145 let limit = request.limit.map(|n| n.get() as u32);
3146 let request_id = request.request_id;
3147 let client_id = request.client_id.unwrap_or(self.client_id);
3148 let params = request.params;
3149 let clock = self.clock;
3150 let start_nanos = datetime_to_unix_nanos(start);
3151 let end_nanos = datetime_to_unix_nanos(end);
3152
3153 self.spawn_command(async move {
3154 match http
3155 .request_funding_rates(instrument_id, start, end, limit)
3156 .await
3157 .context("failed to request funding rates from Binance Futures")
3158 {
3159 Ok(funding_rates) => {
3160 let response = DataResponse::FundingRates(FundingRatesResponse::new(
3161 request_id,
3162 client_id,
3163 instrument_id,
3164 funding_rates,
3165 start_nanos,
3166 end_nanos,
3167 clock.get_time_ns(),
3168 params,
3169 ));
3170
3171 if let Err(e) = sender.send(DataEvent::Response(response)) {
3172 log::error!("Failed to send funding rates response: {e}");
3173 }
3174 }
3175 Err(e) => log::error!("Funding rates request failed for {instrument_id}: {e:?}"),
3176 }
3177 });
3178
3179 Ok(())
3180 }
3181
3182 fn request_bars(&self, request: RequestBars) -> anyhow::Result<()> {
3183 let http = self.http_client.clone();
3184 let sender = self.data_sender.clone();
3185 let bar_type = request.bar_type;
3186 let start = request.start;
3187 let end = request.end;
3188 let limit = request.limit.map(|n| n.get() as u32);
3189 let request_id = request.request_id;
3190 let client_id = request.client_id.unwrap_or(self.client_id);
3191 let params = request.params;
3192 let clock = self.clock;
3193 let start_nanos = datetime_to_unix_nanos(start);
3194 let end_nanos = datetime_to_unix_nanos(end);
3195 anyhow::ensure!(
3196 bar_type.aggregation_source() == AggregationSource::External,
3197 "Binance historical bars require EXTERNAL aggregation"
3198 );
3199 anyhow::ensure!(
3200 bar_type.spec().price_type == PriceType::Last,
3201 "Binance historical bars require LAST price type"
3202 );
3203 anyhow::ensure!(
3204 bar_type.spec().is_time_aggregated(),
3205 "Binance historical bars require time aggregation"
3206 );
3207
3208 self.spawn_command(async move {
3209 let result = http.request_bars(bar_type, start, end, limit).await;
3210
3211 match result.context("failed to request bars from Binance Futures") {
3212 Ok(bars) => {
3213 let response = DataResponse::Bars(BarsResponse::new(
3214 request_id,
3215 client_id,
3216 bar_type,
3217 bars,
3218 start_nanos,
3219 end_nanos,
3220 clock.get_time_ns(),
3221 params,
3222 ));
3223
3224 if let Err(e) = sender.send(DataEvent::Response(response)) {
3225 log::error!("Failed to send bars response: {e}");
3226 }
3227 }
3228 Err(e) => log::error!("Bar request failed: {e:?}"),
3229 }
3230 });
3231
3232 Ok(())
3233 }
3234
3235 fn request_book_snapshot(&self, request: RequestBookSnapshot) -> anyhow::Result<()> {
3236 let depth = request.depth.map_or(1000, |value| value.get() as u32);
3237 anyhow::ensure!(
3238 BINANCE_BOOK_DEPTHS.contains(&depth),
3239 "invalid Binance Futures order-book depth {depth}; valid values are {BINANCE_BOOK_DEPTHS:?}"
3240 );
3241 let http = self.http_client.clone();
3242 let sender = self.data_sender.clone();
3243 let instrument_id = request.instrument_id;
3244 let request_id = request.request_id;
3245 let client_id = request.client_id.unwrap_or(self.client_id);
3246 let params = request.params;
3247 let clock = self.clock;
3248
3249 self.spawn_command(async move {
3250 match http.request_book_snapshot(instrument_id, Some(depth)).await {
3251 Ok(book) => {
3252 let response = DataResponse::Book(BookResponse::new(
3253 request_id,
3254 client_id,
3255 instrument_id,
3256 book,
3257 None,
3258 None,
3259 clock.get_time_ns(),
3260 params,
3261 ));
3262
3263 if let Err(e) = sender.send(DataEvent::Response(response)) {
3264 log::error!("Failed to send book snapshot response: {e}");
3265 }
3266 }
3267 Err(e) => log::error!("Book snapshot request failed for {instrument_id}: {e:?}"),
3268 }
3269 });
3270 Ok(())
3271 }
3272}
3273
3274impl BinanceFuturesDataClient {
3275 fn subscribe_top_of_book(&self, instrument_id: InstrumentId) {
3276 let should_subscribe = {
3277 let previous = self
3278 .quote_refs
3279 .load()
3280 .get(&instrument_id)
3281 .copied()
3282 .unwrap_or(0);
3283 self.quote_refs
3284 .rcu(|refs| *refs.entry(instrument_id).or_insert(0) += 1);
3285 previous == 0
3286 };
3287
3288 if should_subscribe {
3289 let ws = self.ws_public_client.clone();
3290 let stream = format!(
3291 "{}@bookTicker",
3292 format_binance_stream_symbol(&instrument_id)
3293 );
3294 self.spawn_ws(
3295 async move {
3296 ws.subscribe(vec![stream])
3297 .await
3298 .context("top-of-book subscription")
3299 },
3300 "top-of-book subscription",
3301 );
3302 }
3303 }
3304
3305 fn unsubscribe_top_of_book(&self, instrument_id: InstrumentId) {
3306 let should_unsubscribe = match self.quote_refs.load().get(&instrument_id).copied() {
3307 Some(1) => {
3308 self.quote_refs.remove(&instrument_id);
3309 true
3310 }
3311 Some(count) if count > 1 => {
3312 self.quote_refs.rcu(|refs| {
3313 if let Some(existing) = refs.get_mut(&instrument_id) {
3314 *existing -= 1;
3315 }
3316 });
3317 false
3318 }
3319 _ => false,
3320 };
3321
3322 if should_unsubscribe {
3323 let ws = self.ws_public_client.clone();
3324 let stream = format!(
3325 "{}@bookTicker",
3326 format_binance_stream_symbol(&instrument_id)
3327 );
3328 self.spawn_ws(
3329 async move {
3330 ws.unsubscribe(vec![stream])
3331 .await
3332 .context("top-of-book unsubscribe")
3333 .map(|_| ())
3334 },
3335 "top-of-book unsubscribe",
3336 );
3337 }
3338 }
3339}
3340
3341#[derive(Debug, Clone)]
3342struct BufferedDepthUpdate {
3343 deltas: OrderBookDeltas,
3344 first_update_id: u64,
3345 final_update_id: u64,
3346 prev_final_update_id: u64,
3347}
3348
3349#[derive(Debug, Clone)]
3350struct BookBuffer {
3351 updates: Vec<BufferedDepthUpdate>,
3352 epoch: u64,
3353}
3354
3355impl BookBuffer {
3356 fn new(epoch: u64) -> Self {
3357 Self {
3358 updates: Vec::new(),
3359 epoch,
3360 }
3361 }
3362}
3363
3364fn is_partial_book_depth(depth: u32) -> bool {
3365 matches!(depth, 5 | 10 | 20)
3366}
3367
3368fn book_stream(instrument_id: &InstrumentId, depth: u32) -> String {
3369 let symbol = format_binance_stream_symbol(instrument_id);
3370 if is_partial_book_depth(depth) {
3371 format!("{symbol}@depth{depth}@100ms")
3372 } else {
3373 format!("{symbol}@depth@0ms")
3374 }
3375}
3376
3377#[derive(Clone, Debug)]
3378struct BookDrain {
3379 generation: u64,
3380 stream: String,
3381 satisfied: bool,
3384}
3385
3386fn arm_book_drain(
3387 map: &AtomicMap<InstrumentId, Vec<BookDrain>>,
3388 instrument_id: InstrumentId,
3389 generation: u64,
3390 stream: &str,
3391) {
3392 map.rcu(|m| {
3393 m.entry(instrument_id).or_default().push(BookDrain {
3394 generation,
3395 stream: stream.to_owned(),
3396 satisfied: false,
3397 });
3398 });
3399}
3400
3401fn satisfy_book_drain(
3402 map: &AtomicMap<InstrumentId, Vec<BookDrain>>,
3403 instrument_id: InstrumentId,
3404 stream: &str,
3405) {
3406 map.rcu(|m| {
3407 if let Some(drains) = m.get_mut(&instrument_id) {
3408 for drain in drains.iter_mut().filter(|drain| drain.stream == stream) {
3409 drain.satisfied = true;
3410 }
3411 }
3412 });
3413}
3414
3415fn remove_book_drain(map: &AtomicMap<InstrumentId, Vec<BookDrain>>, generation: u64) {
3418 map.rcu(|m| {
3419 for drains in m.values_mut() {
3420 drains.retain(|drain| drain.generation != generation);
3421 }
3422 m.retain(|_, drains| !drains.is_empty());
3423 });
3424}
3425
3426fn book_drain_active(
3427 map: &AtomicMap<InstrumentId, Vec<BookDrain>>,
3428 instrument_id: InstrumentId,
3429) -> bool {
3430 map.load()
3431 .get(&instrument_id)
3432 .is_some_and(|drains| drains.iter().any(|drain| !drain.satisfied))
3433}
3434
3435fn revive_book_drains(
3439 map: &AtomicMap<InstrumentId, Vec<BookDrain>>,
3440 instrument_id: InstrumentId,
3441 stream: &str,
3442) {
3443 map.rcu(|m| {
3444 if let Some(drains) = m.get_mut(&instrument_id) {
3445 for drain in drains.iter_mut().filter(|drain| drain.stream != stream) {
3446 drain.satisfied = false;
3447 }
3448 }
3449 });
3450}
3451
3452fn subscribe_ticker(client: &BinanceFuturesDataClient, data_type: &DataType) -> anyhow::Result<()> {
3453 let instrument_id = BinanceFuturesDataClient::required_instrument_id_metadata(data_type)?;
3454 if instrument_id.venue != client.venue() {
3455 anyhow::bail!(
3456 "Binance Futures ticker custom data requires BINANCE venue instrument, received {instrument_id}"
3457 );
3458 }
3459
3460 let should_subscribe = {
3461 let prev = client
3462 .ticker_refs
3463 .load()
3464 .get(&instrument_id)
3465 .copied()
3466 .unwrap_or(0);
3467 client.ticker_refs.rcu(|m| {
3468 let count = m.entry(instrument_id).or_insert(0);
3469 *count += 1;
3470 });
3471 prev == 0
3472 };
3473
3474 if should_subscribe {
3475 let ws = client.ws_client.clone();
3476 let stream = ticker_stream(&instrument_id);
3477 client.spawn_ws(
3478 async move {
3479 ws.subscribe(vec![stream])
3480 .await
3481 .context("ticker subscription")
3482 },
3483 "ticker subscription",
3484 );
3485 }
3486
3487 Ok(())
3488}
3489
3490fn unsubscribe_ticker(
3491 client: &BinanceFuturesDataClient,
3492 data_type: &DataType,
3493) -> anyhow::Result<()> {
3494 let instrument_id = BinanceFuturesDataClient::required_instrument_id_metadata(data_type)?;
3495 if instrument_id.venue != client.venue() {
3496 anyhow::bail!(
3497 "Binance Futures ticker custom data requires BINANCE venue instrument, received {instrument_id}"
3498 );
3499 }
3500
3501 let should_unsubscribe = {
3502 let prev = client.ticker_refs.load().get(&instrument_id).copied();
3503 match prev {
3504 Some(count) if count <= 1 => {
3505 client.ticker_refs.remove(&instrument_id);
3506 true
3507 }
3508 Some(_) => {
3509 client.ticker_refs.rcu(|m| {
3510 if let Some(count) = m.get_mut(&instrument_id) {
3511 *count = count.saturating_sub(1);
3512 }
3513 });
3514 false
3515 }
3516 None => false,
3517 }
3518 };
3519
3520 if should_unsubscribe {
3521 let ws = client.ws_client.clone();
3522 let stream = ticker_stream(&instrument_id);
3523 client.spawn_ws(
3524 async move {
3525 ws.unsubscribe(vec![stream])
3526 .await
3527 .context("ticker unsubscribe")
3528 .map(|_| ())
3529 },
3530 "ticker unsubscribe",
3531 );
3532 }
3533
3534 Ok(())
3535}
3536
3537fn ticker_data_type(instrument_id: InstrumentId) -> DataType {
3538 let mut metadata = Params::new();
3539 metadata.insert(
3540 "instrument_id".to_string(),
3541 serde_json::Value::String(instrument_id.to_string()),
3542 );
3543 DataType::new(
3544 "BinanceFuturesTicker",
3545 Some(metadata),
3546 Some(instrument_id.to_string()),
3547 )
3548}
3549
3550fn mark_price_data_type(instrument_id: InstrumentId) -> DataType {
3551 let mut metadata = Params::new();
3552 metadata.insert(
3553 "instrument_id".to_string(),
3554 serde_json::Value::String(instrument_id.to_string()),
3555 );
3556 DataType::new(
3557 "BinanceFuturesMarkPriceUpdate",
3558 Some(metadata),
3559 Some(instrument_id.to_string()),
3560 )
3561}
3562
3563fn ticker_stream(instrument_id: &InstrumentId) -> String {
3564 format!("{}@ticker", format_binance_stream_symbol(instrument_id))
3565}
3566
3567#[cfg(test)]
3568mod tests {
3569 use rstest::rstest;
3570 use rust_decimal_macros::dec;
3571
3572 use super::*;
3573
3574 #[rstest]
3575 #[case(0, 250)]
3576 #[case(1, 500)]
3577 #[case(2, 1_000)]
3578 #[case(3, 2_000)]
3579 #[case(4, 3_000)]
3580 #[case(5, 3_000)]
3581 fn test_snapshot_retry_backoff_exponentially_increases_then_caps(
3582 #[case] retry_count: u32,
3583 #[case] expected_ms: u64,
3584 ) {
3585 assert_eq!(
3586 futures_snapshot_retry_backoff(retry_count),
3587 Duration::from_millis(expected_ms)
3588 );
3589 }
3590
3591 #[rstest]
3592 fn test_parse_order_book_snapshot_skips_invalid_levels() {
3593 let instrument_id = InstrumentId::from("BTCUSDT-PERP.BINANCE");
3594 let order_book = BinanceOrderBook {
3595 last_update_id: 10,
3596 bids: vec![
3597 ("not-a-price".to_string(), "1.0".to_string()),
3598 ("100.00".to_string(), "0.5".to_string()),
3599 ],
3600 asks: vec![
3601 ("101.00".to_string(), "not-a-quantity".to_string()),
3602 ("102.00".to_string(), "0.7".to_string()),
3603 ],
3604 event_time: None,
3605 transaction_time: None,
3606 };
3607
3608 let deltas =
3609 parse_order_book_snapshot(&order_book, instrument_id, 2, 3, UnixNanos::from(1));
3610
3611 assert_eq!(deltas.deltas.len(), 3);
3612 assert_eq!(deltas.deltas[1].order.side, OrderSide::Buy.into());
3613 assert_eq!(deltas.deltas[1].order.price.as_decimal(), dec!(100.00));
3614 assert_eq!(deltas.deltas[1].order.size.as_decimal(), dec!(0.500));
3615 assert_eq!(deltas.deltas[2].order.side, OrderSide::Sell.into());
3616 assert_eq!(deltas.deltas[2].order.price.as_decimal(), dec!(102.00));
3617 assert_eq!(deltas.deltas[2].order.size.as_decimal(), dec!(0.700));
3618 assert_eq!(deltas.deltas[2].flags, RecordFlag::F_LAST as u8);
3619 assert_eq!(deltas.ts_event, UnixNanos::from(1));
3620 assert_eq!(deltas.ts_init, UnixNanos::from(1));
3621 }
3622
3623 #[rstest]
3624 #[case::negative(-1)]
3625 #[case::overflow(i64::MAX)]
3626 fn test_parse_order_book_snapshot_falls_back_for_invalid_timestamp(
3627 #[case] transaction_time: i64,
3628 ) {
3629 let order_book = BinanceOrderBook {
3630 last_update_id: 10,
3631 bids: vec![],
3632 asks: vec![],
3633 event_time: None,
3634 transaction_time: Some(transaction_time),
3635 };
3636
3637 let ts_init = UnixNanos::from(1);
3638 let deltas = parse_order_book_snapshot(
3639 &order_book,
3640 InstrumentId::from("BTCUSDT-PERP.BINANCE"),
3641 2,
3642 3,
3643 ts_init,
3644 );
3645
3646 assert_eq!(deltas.ts_event, ts_init);
3647 assert_eq!(deltas.ts_init, ts_init);
3648 }
3649
3650 #[rstest]
3651 fn test_parse_order_book_snapshot_all_invalid_levels_marks_clear_last() {
3652 let instrument_id = InstrumentId::from("BTCUSDT-PERP.BINANCE");
3653 let order_book = BinanceOrderBook {
3654 last_update_id: 10,
3655 bids: vec![("not-a-price".to_string(), "1.0".to_string())],
3656 asks: vec![("101.00".to_string(), "not-a-quantity".to_string())],
3657 event_time: None,
3658 transaction_time: None,
3659 };
3660
3661 let deltas =
3662 parse_order_book_snapshot(&order_book, instrument_id, 2, 3, UnixNanos::from(1));
3663
3664 assert_eq!(deltas.deltas.len(), 1);
3665 assert_eq!(deltas.deltas[0].action, BookAction::Clear);
3666 assert_eq!(
3667 deltas.deltas[0].flags,
3668 RecordFlag::F_SNAPSHOT as u8 | RecordFlag::F_LAST as u8
3669 );
3670 }
3671
3672 #[rstest]
3673 fn test_book_drain_gate_follows_unsatisfied_entries() {
3674 let map = AtomicMap::new();
3675 let instrument_id = InstrumentId::from("BTCUSDT-PERP.BINANCE");
3676
3677 assert!(!book_drain_active(&map, instrument_id));
3678 arm_book_drain(&map, instrument_id, 1, "btcusdt@depth20@100ms");
3679 assert!(book_drain_active(&map, instrument_id));
3680
3681 satisfy_book_drain(&map, instrument_id, "btcusdt@depth20@100ms");
3682 assert!(!book_drain_active(&map, instrument_id));
3683
3684 arm_book_drain(&map, instrument_id, 2, "btcusdt@depth10@100ms");
3685 assert!(book_drain_active(&map, instrument_id));
3686 }
3687
3688 #[rstest]
3689 fn test_remove_book_drain_resolves_only_matching_generation() {
3690 let map = AtomicMap::new();
3691 let instrument_id = InstrumentId::from("BTCUSDT-PERP.BINANCE");
3692
3693 arm_book_drain(&map, instrument_id, 1, "btcusdt@depth20@100ms");
3696 satisfy_book_drain(&map, instrument_id, "btcusdt@depth20@100ms");
3697 arm_book_drain(&map, instrument_id, 2, "btcusdt@depth20@100ms");
3698
3699 remove_book_drain(&map, 1);
3700 assert!(book_drain_active(&map, instrument_id));
3701
3702 remove_book_drain(&map, 2);
3703 assert!(!book_drain_active(&map, instrument_id));
3704 assert!(!map.load().contains_key(&instrument_id));
3705 }
3706
3707 #[rstest]
3708 fn test_semantics_change_revives_satisfied_drains() {
3709 let map = AtomicMap::new();
3710 let instrument_id = InstrumentId::from("BTCUSDT-PERP.BINANCE");
3711
3712 arm_book_drain(&map, instrument_id, 1, "btcusdt@depth20@100ms");
3713 satisfy_book_drain(&map, instrument_id, "btcusdt@depth20@100ms");
3714 assert!(!book_drain_active(&map, instrument_id));
3715
3716 revive_book_drains(&map, instrument_id, "btcusdt@depth5@100ms");
3719 assert!(book_drain_active(&map, instrument_id));
3720
3721 satisfy_book_drain(&map, instrument_id, "btcusdt@depth20@100ms");
3723 revive_book_drains(&map, instrument_id, "btcusdt@depth20@100ms");
3724 assert!(!book_drain_active(&map, instrument_id));
3725 }
3726
3727 #[rstest]
3728 fn test_remove_book_drain_leaves_other_generations_pending() {
3729 let map = AtomicMap::new();
3730 let instrument_id = InstrumentId::from("BTCUSDT-PERP.BINANCE");
3731
3732 arm_book_drain(&map, instrument_id, 1, "btcusdt@depth20@100ms");
3733 arm_book_drain(&map, instrument_id, 2, "btcusdt@depth10@100ms");
3734
3735 remove_book_drain(&map, 2);
3737 assert!(book_drain_active(&map, instrument_id));
3738
3739 remove_book_drain(&map, 1);
3740 assert!(!book_drain_active(&map, instrument_id));
3741 }
3742}