Skip to main content

nautilus_binance/futures/
data.rs

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