nautilus_binance/futures/

data.rs

// -------------------------------------------------------------------------------------------------
//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
//  https://nautechsystems.io
//
//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
//  You may not use this file except in compliance with the License.
//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
//
//  Unless required by applicable law or agreed to in writing, software
//  distributed under the License is distributed on an "AS IS" BASIS,
//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//  See the License for the specific language governing permissions and
//  limitations under the License.
// -------------------------------------------------------------------------------------------------

//! Live market data client implementation for the Binance Futures adapter.

use std::{
    num::NonZeroU32,
    str::FromStr,
    sync::{
        Arc,
        atomic::{AtomicBool, AtomicU32, Ordering},
    },
    time::Duration,
};

use ahash::AHashMap;
use anyhow::Context;
use futures_util::{StreamExt, pin_mut};
use nautilus_common::{
    clients::{DataClient, execute_bar_subscriptions, execute_instrument_subscriptions},
    live::{runner::get_data_event_sender, sender::EventSender},
    messages::{
        DataEvent,
        data::{
            BarsResponse, BookResponse, CustomDataResponse, DataResponse, FundingRatesResponse,
            InstrumentResponse, InstrumentsResponse, RequestBars, RequestBookSnapshot,
            RequestCustomData, RequestFundingRates, RequestInstrument, RequestInstruments,
            RequestTrades, SubscribeBars, SubscribeBookDeltas, SubscribeCustomData,
            SubscribeFundingRates, SubscribeIndexPrices, SubscribeInstrument, SubscribeInstruments,
            SubscribeMarkPrices, SubscribeQuotes, SubscribeTrades, TradesResponse, UnsubscribeBars,
            UnsubscribeBookDeltas, UnsubscribeCustomData, UnsubscribeFundingRates,
            UnsubscribeIndexPrices, UnsubscribeMarkPrices, UnsubscribeQuotes, UnsubscribeTrades,
            subscribe::SubscribeInstrumentStatus, unsubscribe::UnsubscribeInstrumentStatus,
        },
    },
};
use nautilus_core::{
    AtomicMap, Params,
    datetime::datetime_to_unix_nanos,
    nanos::UnixNanos,
    time::{AtomicTime, get_atomic_clock_realtime},
};
use nautilus_live::{
    SocketControlFactory,
    task::{TaskGroup, TaskGroupGuard, TaskSpawner},
};
use nautilus_model::{
    data::{BookOrder, CustomData, Data, DataType, OrderBookDelta, OrderBookDeltas, QuoteTick},
    enums::{
        AggregationSource, BookAction, BookType, MarketStatusAction, OrderSide, PriceType,
        RecordFlag,
    },
    identifiers::{ClientId, InstrumentId, Venue},
    instruments::{Instrument, InstrumentAny},
    types::{Price, Quantity},
};
use parking_lot::Mutex;
use rust_decimal::Decimal;
use tokio_util::sync::CancellationToken;
use ustr::Ustr;

use crate::{
    book::{
        BinanceBookError,
        pacing::SnapshotPacer,
        recovery::spawn_recovery,
        sync::{BookSyncTracker, DepthSequencing, DepthSnapshot, DepthUpdate},
    },
    common::{
        bar::{binance_bar_data_type, binance_bars_to_custom_data, parse_binance_bar_type},
        consts::{BINANCE_BOOK_DEPTHS, BINANCE_VENUE, BINANCE_WS_HEARTBEAT_SECS},
        enums::{BinanceEnvironment, BinanceProductType},
        parse::{
            bar_spec_to_binance_interval, parse_millis, parse_millis_or_init,
            parse_price_at_precision, parse_quantity_at_precision,
            parse_required_price_at_precision, parse_required_quantity_at_precision,
            quote_to_l1_deltas,
        },
        status::diff_and_emit_statuses,
        symbol::{format_binance_stream_symbol, format_binance_symbol},
        urls::{get_usdm_ws_route_base_url, get_ws_public_base_url},
        websocket::chain_command,
    },
    config::BinanceDataClientConfig,
    data_types::{
        BinanceFuturesLiquidation, BinanceFuturesOpenInterest, BinanceFuturesOpenInterestHist,
        BinanceFuturesOpenInterestHistPoint, register_binance_custom_data,
    },
    futures::{
        http::{
            client::{BinanceFuturesHttpClient, BinanceFuturesInstrument},
            error::BinanceFuturesHttpError,
            models::BinanceOrderBook,
            query::{BinanceDepthParams, BinanceOpenInterestHistParams, BinanceOpenInterestParams},
        },
        websocket::streams::{
            client::BinanceFuturesWebSocketClient,
            messages::BinanceFuturesWsStreamsMessage,
            parse_data::{
                parse_agg_trade, parse_book_ticker, parse_depth_snapshot, parse_depth_update,
                parse_kline, parse_mark_price, parse_ticker, parse_trade,
            },
        },
    },
};

const MARKET_STREAMS_ENDPOINT: &str = "binance-futures-market-streams";
const PUBLIC_STREAMS_ENDPOINT: &str = "binance-futures-public-streams";

// Half the 2,400 per-minute request weight that USD-M and COIN-M share
const SNAPSHOT_WEIGHT_PER_MINUTE: NonZeroU32 = NonZeroU32::new(1_200).expect("non-zero");

/// Binance Futures data client for USD-M and COIN-M markets.
#[derive(Debug)]
pub struct BinanceFuturesDataClient {
    clock: &'static AtomicTime,
    client_id: ClientId,
    config: BinanceDataClientConfig,
    product_type: BinanceProductType,
    http_client: BinanceFuturesHttpClient,
    ws_client: BinanceFuturesWebSocketClient,
    ws_public_client: BinanceFuturesWebSocketClient,
    data_sender: EventSender<DataEvent>,
    is_connected: AtomicBool,
    cancellation_token: CancellationToken,
    session_tasks: TaskGroup,
    command_tasks: TaskGroup,
    shutdown_errors: Vec<String>,
    instruments: Arc<AtomicMap<InstrumentId, InstrumentAny>>,
    status_cache: Arc<AtomicMap<InstrumentId, MarketStatusAction>>,
    instrument_update_lock: Arc<tokio::sync::Mutex<()>>,
    book_sync: BookSyncTracker,
    book_subscriptions: Arc<AtomicMap<InstrumentId, u32>>,
    book_unsubscribes_pending: Arc<AtomicMap<InstrumentId, Vec<BookDrain>>>,
    l1_book_subscriptions: Arc<AtomicMap<InstrumentId, u32>>,
    quote_refs: Arc<AtomicMap<InstrumentId, u32>>,
    // Mark, index, and funding subscriptions share one ref-counted `@markPrice@1s` stream
    mark_price_refs: Arc<AtomicMap<InstrumentId, u32>>,
    ticker_refs: Arc<AtomicMap<InstrumentId, u32>>,
    force_order_refs: Arc<AtomicMap<InstrumentId, u32>>,
    force_order_all_market_refs: Arc<AtomicU32>,
    force_order_all_market_stream_active: Arc<AtomicBool>,
    force_order_ws_lock: Arc<tokio::sync::Mutex<()>>,
    book_command_tail: Arc<Mutex<Option<tokio::sync::oneshot::Receiver<()>>>>,
    book_drain_generation: u64,
}

impl BinanceFuturesDataClient {
    /// Creates a new [`BinanceFuturesDataClient`] instance.
    ///
    /// # Errors
    ///
    /// Returns an error if the client fails to initialize or if the product type
    /// is not a futures type (UsdM or CoinM).
    pub fn new(
        client_id: ClientId,
        config: BinanceDataClientConfig,
        product_type: BinanceProductType,
    ) -> anyhow::Result<Self> {
        config.validate()?;

        match product_type {
            BinanceProductType::UsdM | BinanceProductType::CoinM => {}
            _ => {
                anyhow::bail!(
                    "BinanceFuturesDataClient requires UsdM or CoinM product type, was {product_type:?}"
                );
            }
        }

        let clock = get_atomic_clock_realtime();
        let data_sender = get_data_event_sender();
        let socket_factory = SocketControlFactory::new(client_id, Some(*BINANCE_VENUE));
        let api_key = config
            .api_key
            .as_ref()
            .map(|value| value.expose_secret().to_owned());
        let api_secret = config
            .api_secret
            .as_ref()
            .map(|value| value.expose_secret().to_owned());
        let proxy_url = config
            .proxy_url
            .as_ref()
            .map(|value| value.expose_secret().to_owned());

        let http_client = BinanceFuturesHttpClient::new(
            product_type,
            config.environment,
            clock,
            api_key.clone(),
            api_secret.clone(),
            config.base_url_http.clone(),
            Some(config.recv_window_ms),
            None, // timeout_secs
            proxy_url.clone(),
            false, // treat_expired_as_canceled
        )?
        .with_retry_config(config.retry_config());

        let market_url = config.base_url_ws.clone().map(|url| {
            if product_type == BinanceProductType::UsdM
                && config.environment == BinanceEnvironment::Live
            {
                get_usdm_ws_route_base_url(&url, "market")
            } else {
                url
            }
        });

        let ws_client = BinanceFuturesWebSocketClient::new(
            product_type,
            config.environment,
            api_key,
            api_secret,
            market_url,
            Some(BINANCE_WS_HEARTBEAT_SECS),
            config.transport_backend,
        )?
        .with_proxy(proxy_url.clone())
        .with_socket_control(socket_factory.clone(), MARKET_STREAMS_ENDPOINT);

        let public_url = config.base_url_ws.clone().map_or_else(
            || get_ws_public_base_url(product_type, config.environment).to_string(),
            |url| {
                if product_type == BinanceProductType::UsdM
                    && config.environment == BinanceEnvironment::Live
                {
                    get_usdm_ws_route_base_url(&url, "public")
                } else {
                    url
                }
            },
        );

        let ws_public_client = BinanceFuturesWebSocketClient::new(
            product_type,
            config.environment,
            None,
            None,
            Some(public_url),
            Some(BINANCE_WS_HEARTBEAT_SECS),
            config.transport_backend,
        )?
        .with_proxy(proxy_url)
        .with_socket_control(socket_factory, PUBLIC_STREAMS_ENDPOINT);

        let snapshot_pacer = Arc::new(SnapshotPacer::new(SNAPSHOT_WEIGHT_PER_MINUTE));

        let book_sync = BookSyncTracker::new(
            DepthSequencing::Futures,
            data_sender.clone(),
            snapshot_pacer,
        );

        let session_tasks = TaskGroup::new();
        let command_tasks = TaskGroup::new();

        Ok(Self {
            clock,
            client_id,
            config,
            product_type,
            http_client,
            ws_client,
            ws_public_client,
            data_sender,
            is_connected: AtomicBool::new(false),
            cancellation_token: session_tasks.cancellation_token(),
            session_tasks,
            command_tasks,
            shutdown_errors: Vec::new(),
            instruments: Arc::new(AtomicMap::new()),
            status_cache: Arc::new(AtomicMap::new()),
            instrument_update_lock: Arc::new(tokio::sync::Mutex::new(())),
            book_sync,
            book_subscriptions: Arc::new(AtomicMap::new()),
            book_unsubscribes_pending: Arc::new(AtomicMap::new()),
            l1_book_subscriptions: Arc::new(AtomicMap::new()),
            quote_refs: Arc::new(AtomicMap::new()),
            mark_price_refs: Arc::new(AtomicMap::new()),
            ticker_refs: Arc::new(AtomicMap::new()),
            force_order_refs: Arc::new(AtomicMap::new()),
            force_order_all_market_refs: Arc::new(AtomicU32::new(0)),
            force_order_all_market_stream_active: Arc::new(AtomicBool::new(false)),
            force_order_ws_lock: Arc::new(tokio::sync::Mutex::new(())),
            book_command_tail: Arc::new(Mutex::new(None)),
            book_drain_generation: 0,
        })
    }

    fn venue(&self) -> Venue {
        *BINANCE_VENUE
    }

    fn send_data(sender: &EventSender<DataEvent>, data: Data) {
        if let Err(e) = sender.send(DataEvent::Data(data)) {
            log::error!("Failed to emit data event: {e}");
        }
    }

    fn spawn_ws<F>(&self, fut: F, context: &'static str)
    where
        F: Future<Output = anyhow::Result<()>> + Send + 'static,
    {
        let future = async move {
            if let Err(e) = fut.await {
                log::error!("{context}: {e:?}");
            }
        };

        if let Err(e) = self.command_tasks.spawn(future) {
            log::warn!("Skipping Binance Futures {context} after shutdown began: {e}");
        }
    }

    fn spawn_command<F>(&self, future: F)
    where
        F: Future<Output = ()> + Send + 'static,
    {
        if let Err(e) = self.command_tasks.spawn(future) {
            log::warn!("Skipping Binance Futures data command after shutdown began: {e}");
        }
    }

    async fn finish_tasks(&self) -> anyhow::Result<()> {
        let (session_result, command_result) = tokio::join!(
            self.session_tasks
                .finish_shutdown(Duration::from_secs(1), Duration::from_secs(2)),
            self.command_tasks
                .finish_shutdown(Duration::from_secs(1), Duration::from_secs(2)),
        );
        let mut errors = Vec::new();
        if let Err(e) = session_result {
            errors.push(format!(
                "failed to finish Binance Futures data session tasks: {e}"
            ));
        }

        if let Err(e) = command_result {
            errors.push(format!(
                "failed to finish Binance Futures data command tasks: {e}"
            ));
        }

        if !errors.is_empty() {
            anyhow::bail!(errors.join("; "));
        }
        Ok(())
    }

    async fn prepare_task_groups(&mut self) -> anyhow::Result<()> {
        if !self.session_tasks.is_open() || !self.command_tasks.is_open() {
            self.teardown_partial_connect().await?;
            self.session_tasks
                .start_generation()
                .context("failed to start Binance Futures data session task generation")?;
            self.command_tasks
                .start_generation()
                .context("failed to start Binance Futures data command task generation")?;
            self.cancellation_token = self.session_tasks.cancellation_token();
        }
        Ok(())
    }

    async fn teardown_partial_connect(&mut self) -> anyhow::Result<()> {
        self.session_tasks.begin_shutdown();
        self.command_tasks.begin_shutdown();
        self.ws_client.begin_shutdown();
        self.ws_public_client.begin_shutdown();

        if let Err(e) = self.ws_client.close().await {
            self.shutdown_errors
                .push(format!("market WebSocket close failed: {e}"));
        }

        if let Err(e) = self.ws_public_client.close().await {
            self.shutdown_errors
                .push(format!("public WebSocket close failed: {e}"));
        }

        if let Err(e) = self.finish_tasks().await {
            self.shutdown_errors.push(e.to_string());
        }
        self.is_connected.store(false, Ordering::Release);

        if !self.shutdown_errors.is_empty() {
            let errors = std::mem::take(&mut self.shutdown_errors);
            anyhow::bail!(
                "Binance Futures data teardown failed: {}",
                errors.join("; ")
            );
        }
        Ok(())
    }

    #[expect(clippy::too_many_arguments)]
    async fn refresh_instrument_catalog(
        http: &BinanceFuturesHttpClient,
        provider: &crate::config::BinanceInstrumentProviderConfig,
        instruments_cache: &Arc<AtomicMap<InstrumentId, InstrumentAny>>,
        status_cache: &Arc<AtomicMap<InstrumentId, MarketStatusAction>>,
        instrument_update_lock: &tokio::sync::Mutex<()>,
        ws: &BinanceFuturesWebSocketClient,
        ws_public: &BinanceFuturesWebSocketClient,
        sender: &EventSender<DataEvent>,
        clock: &'static AtomicTime,
        emit_status_changes: bool,
    ) -> anyhow::Result<Vec<InstrumentAny>> {
        let _guard = instrument_update_lock.lock().await;
        let instruments = http
            .request_instruments_with_config(provider)
            .await
            .context("failed to request Binance Futures instruments")?;
        let venue_statuses = http
            .request_symbol_statuses()
            .await
            .context("failed to request Binance Futures instrument statuses")?;

        let instrument_map = instruments
            .iter()
            .map(|instrument| (instrument.id(), instrument.clone()))
            .collect::<AHashMap<_, _>>();
        let raw_to_id = instrument_map
            .values()
            .map(|instrument| (instrument.raw_symbol().inner(), instrument.id()))
            .collect::<AHashMap<_, _>>();
        let status_map = venue_statuses
            .into_iter()
            .filter_map(|(symbol, action)| {
                raw_to_id
                    .get(&symbol)
                    .copied()
                    .map(|instrument_id| (instrument_id, action))
            })
            .collect::<AHashMap<_, _>>();

        instruments_cache.store(instrument_map);
        ws.replace_instruments(&instruments);
        ws_public.replace_instruments(&instruments);

        if emit_status_changes {
            let mut cached_statuses = (**status_cache.load()).clone();
            let ts = clock.get_time_ns();
            diff_and_emit_statuses(&status_map, &mut cached_statuses, sender, ts, ts);
            status_cache.store(cached_statuses);
        } else {
            status_cache.store(status_map);
        }

        for instrument in &instruments {
            if let Err(e) = sender.send(DataEvent::Data(instrument.clone().into())) {
                log::warn!("Failed to send refreshed Binance Futures instrument: {e}");
            }
        }

        Ok(instruments)
    }

    fn custom_liquidation_instrument_id(
        data_type: &DataType,
    ) -> anyhow::Result<Option<InstrumentId>> {
        let Some(raw_instrument_id) = data_type
            .metadata()
            .as_ref()
            .and_then(|m| m.get("instrument_id"))
            .and_then(|v| v.as_str())
            .map(str::trim)
            .filter(|value| !value.is_empty())
        else {
            return Ok(None);
        };

        let instrument_id = InstrumentId::from_str(raw_instrument_id)
            .with_context(|| format!("invalid instrument_id metadata `{raw_instrument_id}`"))?;

        Ok(Some(instrument_id))
    }

    fn required_instrument_id_metadata(data_type: &DataType) -> anyhow::Result<InstrumentId> {
        let Some(raw_instrument_id) = data_type
            .metadata()
            .as_ref()
            .and_then(|m| m.get("instrument_id"))
            .and_then(|v| v.as_str())
            .map(str::trim)
            .filter(|value| !value.is_empty())
        else {
            anyhow::bail!("custom data request requires `instrument_id` metadata");
        };

        InstrumentId::from_str(raw_instrument_id)
            .with_context(|| format!("invalid instrument_id metadata `{raw_instrument_id}`"))
    }

    fn required_period_metadata(data_type: &DataType) -> anyhow::Result<String> {
        let Some(period) = data_type
            .metadata()
            .as_ref()
            .and_then(|m| m.get("period"))
            .and_then(|v| v.as_str())
            .map(str::trim)
            .filter(|value| !value.is_empty())
        else {
            anyhow::bail!("historical open interest request requires `period` metadata");
        };

        Ok(period.to_string())
    }

    fn coinm_open_interest_hist_params(
        http: &BinanceFuturesHttpClient,
        instrument_id: &InstrumentId,
    ) -> anyhow::Result<(String, String)> {
        let symbol = format_binance_symbol(instrument_id);
        if let Some(pair) = symbol.strip_suffix("_PERP") {
            return Ok((pair.to_string(), "PERPETUAL".to_string()));
        }

        let definition = http
            .instrument_metadata(*instrument_id)
            .with_context(|| format!("missing COIN-M definition for {instrument_id}"))?;
        let BinanceFuturesInstrument::CoinM(definition) = definition else {
            anyhow::bail!("expected a COIN-M definition for {instrument_id}");
        };

        Ok((definition.pair.to_string(), definition.contract_type))
    }

    fn parse_open_interest_decimal(field: &str, value: &str) -> anyhow::Result<Decimal> {
        Decimal::from_str_exact(value)
            .with_context(|| format!("invalid Binance open interest `{field}` value `{value}`"))
    }

    fn liquidation_data_type(instrument_id: InstrumentId) -> DataType {
        let metadata = Params::new(
            [(
                "instrument_id".to_string(),
                serde_json::Value::String(instrument_id.to_string()),
            )]
            .into(),
        );
        DataType::new(
            "BinanceFuturesLiquidation",
            Some(metadata),
            Some(instrument_id.to_string()),
        )
    }

    fn liquidation_stream(instrument_id: &InstrumentId) -> String {
        format!("{}@forceOrder", format_binance_stream_symbol(instrument_id))
    }

    fn spawn_liquidation_stream_reconcile(&self, context: &'static str) {
        let ws = self.ws_client.clone();
        let refs = self.force_order_refs.clone();
        let all_market_refs = self.force_order_all_market_refs.clone();
        let all_market_stream_active = self.force_order_all_market_stream_active.clone();
        let ws_lock = self.force_order_ws_lock.clone();

        self.spawn_ws(
            async move {
                let _guard = ws_lock.lock().await;
                let wants_all_market = all_market_refs.load(Ordering::Relaxed) > 0;
                let all_market_active = all_market_stream_active.load(Ordering::Acquire);

                if wants_all_market {
                    if all_market_active {
                        return Ok(());
                    }

                    let specific_streams = refs
                        .load()
                        .keys()
                        .map(Self::liquidation_stream)
                        .collect::<Vec<_>>();

                    if !specific_streams.is_empty() {
                        ws.unsubscribe(specific_streams).await.context(
                            "specific forceOrder unsubscribe while enabling all-market",
                        )?;
                    }

                    if all_market_refs.load(Ordering::Relaxed) == 0 {
                        let restored_streams = refs
                            .load()
                            .keys()
                            .map(Self::liquidation_stream)
                            .collect::<Vec<_>>();

                        if !restored_streams.is_empty() {
                            ws.subscribe(restored_streams).await.context(
                                "specific forceOrder restore after canceled all-market subscription",
                            )?;
                        }
                        all_market_stream_active.store(false, Ordering::Release);
                        return Ok(());
                    }

                    all_market_stream_active.store(true, Ordering::Release);

                    if let Err(e) = ws
                        .subscribe(vec!["!forceOrder@arr".to_string()])
                        .await
                        .context("all-market forceOrder subscription")
                    {
                        all_market_stream_active.store(false, Ordering::Release);
                        return Err(e);
                    }
                } else {
                    if !all_market_active {
                        return Ok(());
                    }

                    ws.unsubscribe(vec!["!forceOrder@arr".to_string()])
                        .await
                        .context("all-market forceOrder unsubscribe")?;

                    let specific_streams = refs
                        .load()
                        .keys()
                        .map(Self::liquidation_stream)
                        .collect::<Vec<_>>();

                    if !specific_streams.is_empty() {
                        ws.subscribe(specific_streams).await.context(
                            "specific forceOrder resubscribe after all-market unsubscribe",
                        )?;
                    }
                    all_market_stream_active.store(false, Ordering::Release);
                }

                Ok(())
            },
            context,
        );
    }

    #[expect(clippy::too_many_arguments)]
    fn handle_ws_message(
        msg: BinanceFuturesWsStreamsMessage,
        data_sender: &EventSender<DataEvent>,
        instruments: &Arc<AtomicMap<InstrumentId, InstrumentAny>>,
        ws_instruments: &Arc<AtomicMap<Ustr, InstrumentAny>>,
        book_sync: &BookSyncTracker,
        book_subscriptions: &Arc<AtomicMap<InstrumentId, u32>>,
        book_unsubscribes_pending: &Arc<AtomicMap<InstrumentId, Vec<BookDrain>>>,
        l1_book_subscriptions: &Arc<AtomicMap<InstrumentId, u32>>,
        force_order_refs: &Arc<AtomicMap<InstrumentId, u32>>,
        ticker_refs: &Arc<AtomicMap<InstrumentId, u32>>,
        force_order_all_market_refs: &Arc<AtomicU32>,
        force_order_all_market_stream_active: &Arc<AtomicBool>,
        http_client: &BinanceFuturesHttpClient,
        snapshot_timeout: Duration,
        clock: &'static AtomicTime,
        command_spawner: &TaskSpawner,
    ) {
        let ts_init = clock.get_time_ns();
        let cache = ws_instruments.load();

        match msg {
            BinanceFuturesWsStreamsMessage::AggTrade(ref trade_msg) => {
                if let Some(instrument) = cache.get(&trade_msg.symbol) {
                    match parse_agg_trade(trade_msg, instrument, ts_init) {
                        Ok(trade) => Self::send_data(data_sender, Data::Trade(trade)),
                        Err(e) => log::warn!("Failed to parse aggregate trade: {e}"),
                    }
                }
            }
            BinanceFuturesWsStreamsMessage::Trade(ref trade_msg) => {
                if let Some(instrument) = cache.get(&trade_msg.symbol) {
                    match parse_trade(trade_msg, instrument, ts_init) {
                        Ok(trade) => Self::send_data(data_sender, Data::Trade(trade)),
                        Err(e) => log::warn!("Failed to parse trade: {e}"),
                    }
                }
            }
            BinanceFuturesWsStreamsMessage::BookTicker(ref ticker_msg) => {
                if let Some(instrument) = cache.get(&ticker_msg.symbol) {
                    match parse_book_ticker(ticker_msg, instrument, ts_init) {
                        Ok(quote) => Self::send_top_of_book(
                            data_sender,
                            l1_book_subscriptions,
                            quote,
                            ticker_msg.update_id,
                        ),
                        Err(e) => log::warn!("Failed to parse book ticker: {e}"),
                    }
                }
            }
            BinanceFuturesWsStreamsMessage::DepthUpdate(ref depth_msg) => {
                if let Some(instrument) = cache.get(&depth_msg.symbol) {
                    let instrument_id = instrument.id();

                    let Some(depth) = book_subscriptions.load().get(&instrument_id).copied() else {
                        return;
                    };

                    // Depth frames carry no stream identity on the wire; drop them until
                    // the venue confirms the prior stream's unsubscribe, so an in-flight
                    // frame from the old stream cannot be parsed with the new
                    // subscription's semantics
                    if book_drain_active(book_unsubscribes_pending, instrument_id) {
                        log::debug!(
                            "Dropping depth frame for {instrument_id} with unsubscribe confirmation pending"
                        );
                        return;
                    }

                    if is_partial_book_depth(depth) {
                        match parse_depth_snapshot(depth_msg, instrument, ts_init) {
                            Ok(deltas) => {
                                Self::send_data(data_sender, Data::BookDeltas(Box::new(deltas)));
                            }
                            Err(e) => log::warn!("Failed to parse depth update: {e}"),
                        }

                        return;
                    }

                    // A diff without levels still advances the sequence, and a dropped
                    // diff surfaces as a sequence gap on the next one
                    let deltas = if depth_msg.bids.is_empty() && depth_msg.asks.is_empty() {
                        None
                    } else {
                        match parse_depth_update(depth_msg, instrument, ts_init) {
                            Ok(deltas) => Some(deltas),
                            Err(e) => {
                                log::warn!("Failed to parse depth update: {e}");
                                return;
                            }
                        }
                    };

                    let update = DepthUpdate {
                        first_update_id: depth_msg.first_update_id,
                        final_update_id: depth_msg.final_update_id,
                        prev_final_update_id: Some(depth_msg.prev_final_update_id),
                        deltas,
                    };

                    if let Some(recovery) = book_sync.handle_update(instrument_id, update) {
                        let http = http_client.clone();
                        let instruments = instruments.clone();

                        spawn_recovery(
                            instrument_id,
                            recovery,
                            book_sync.clone(),
                            move || {
                                Self::fetch_depth_snapshot(
                                    http.clone(),
                                    instruments.clone(),
                                    instrument_id,
                                    depth,
                                    clock,
                                )
                            },
                            depth_request_weight(depth),
                            snapshot_timeout,
                            command_spawner,
                        );
                    }
                }
            }
            BinanceFuturesWsStreamsMessage::MarkPrice(ref mark_msg) => {
                if let Some(instrument) = cache.get(&mark_msg.symbol) {
                    match parse_mark_price(mark_msg, instrument, ts_init) {
                        Ok((mark_update, index_update, funding_update, custom_update)) => {
                            Self::send_data(data_sender, Data::MarkPrice(mark_update));
                            Self::send_data(data_sender, Data::IndexPrice(index_update));

                            if let Err(e) =
                                data_sender.send(DataEvent::Data(Data::FundingRate(funding_update)))
                            {
                                log::error!("Failed to emit funding rate: {e}");
                            }
                            let data_type = mark_price_data_type(instrument.id());
                            Self::send_data(
                                data_sender,
                                Data::Custom(CustomData::new(Arc::new(custom_update), data_type)),
                            );
                        }
                        Err(e) => log::warn!("Failed to parse mark price: {e}"),
                    }
                }
            }
            BinanceFuturesWsStreamsMessage::Kline(ref kline_msg) => {
                if let Some(instrument) = cache.get(&kline_msg.symbol) {
                    match parse_kline(kline_msg, instrument, ts_init) {
                        Ok(Some(bar)) => {
                            Self::send_data(data_sender, Data::Bar(bar.bar()));
                            let data_type = binance_bar_data_type(bar.bar_type);
                            Self::send_data(
                                data_sender,
                                Data::Custom(CustomData::new(Arc::new(bar), data_type)),
                            );
                        }
                        Ok(None) => {} // Kline not closed yet
                        Err(e) => log::warn!("Failed to parse kline: {e}"),
                    }
                }
            }
            BinanceFuturesWsStreamsMessage::ForceOrder(ref liq_msg) => {
                if let Some(instrument) = cache.get(&liq_msg.order.symbol) {
                    let ts_event = parse_millis_or_init(
                        liq_msg.event_time,
                        "Futures liquidation event time",
                        ts_init,
                    );
                    let parse_price = |value: &str, field: &str| -> anyhow::Result<Price> {
                        parse_required_price_at_precision(
                            value,
                            instrument.price_precision(),
                            field,
                        )
                    };

                    let parse_quantity = |value: &str, field: &str| -> anyhow::Result<Quantity> {
                        parse_required_quantity_at_precision(
                            value,
                            instrument.size_precision(),
                            field,
                        )
                    };

                    match (
                        parse_price(&liq_msg.order.price, "price"),
                        parse_price(&liq_msg.order.average_price, "average_price"),
                        parse_quantity(&liq_msg.order.last_filled_qty, "last_filled_qty"),
                        parse_quantity(&liq_msg.order.accumulated_qty, "accumulated_qty"),
                    ) {
                        (
                            Ok(price),
                            Ok(average_price),
                            Ok(last_filled_qty),
                            Ok(accumulated_qty),
                        ) => {
                            let liquidation = Arc::new(BinanceFuturesLiquidation::new(
                                instrument.id(),
                                OrderSide::from(liq_msg.order.side),
                                price,
                                average_price,
                                last_filled_qty,
                                accumulated_qty,
                                ts_event,
                                ts_init,
                            ));

                            let has_all_market_subscription =
                                force_order_all_market_refs.load(Ordering::Relaxed) > 0;
                            let has_all_market_stream =
                                force_order_all_market_stream_active.load(Ordering::Acquire);
                            let has_specific_subscription =
                                force_order_refs.load().contains_key(&instrument.id());

                            if has_all_market_subscription || has_all_market_stream {
                                let data_type =
                                    DataType::new("BinanceFuturesLiquidation", None, None);
                                Self::send_data(
                                    data_sender,
                                    Data::Custom(CustomData::new(liquidation, data_type)),
                                );
                            } else if has_specific_subscription {
                                let data_type = Self::liquidation_data_type(instrument.id());
                                Self::send_data(
                                    data_sender,
                                    Data::Custom(CustomData::new(liquidation, data_type)),
                                );
                            }
                        }
                        (p, ap, lq, aq) => {
                            log::warn!(
                                "Failed to parse Binance liquidation {}: price={:?} avg={:?} \
                                last_qty={:?} accumulated_qty={:?}",
                                liq_msg.order.symbol,
                                p.err(),
                                ap.err(),
                                lq.err(),
                                aq.err(),
                            );
                        }
                    }
                } else {
                    log::debug!(
                        "Received Binance liquidation for uncached symbol {}",
                        liq_msg.order.symbol
                    );
                }
            }
            BinanceFuturesWsStreamsMessage::Ticker(ref ticker_msg) => {
                if let Some(instrument) = cache.get(&ticker_msg.symbol) {
                    let instrument_id = instrument.id();
                    if !ticker_refs.load().contains_key(&instrument_id) {
                        return;
                    }

                    match parse_ticker(ticker_msg, instrument, ts_init) {
                        Ok(ticker) => {
                            let data_type = ticker_data_type(instrument_id);
                            Self::send_data(
                                data_sender,
                                Data::Custom(CustomData::new(Arc::new(ticker), data_type)),
                            );
                        }
                        Err(e) => log::warn!("Failed to parse ticker: {e}"),
                    }
                }
            }
            BinanceFuturesWsStreamsMessage::AccountUpdate(_)
            | BinanceFuturesWsStreamsMessage::OrderUpdate(_)
            | BinanceFuturesWsStreamsMessage::TradeLite(_)
            | BinanceFuturesWsStreamsMessage::AlgoUpdate(_)
            | BinanceFuturesWsStreamsMessage::MarginCall(_)
            | BinanceFuturesWsStreamsMessage::AccountConfigUpdate(_)
            | BinanceFuturesWsStreamsMessage::ListenKeyExpired => {}
            BinanceFuturesWsStreamsMessage::Error(e) => {
                log::warn!(
                    "Binance Futures WebSocket error: code={}, msg={}",
                    e.code,
                    e.msg
                );
            }
            BinanceFuturesWsStreamsMessage::Reconnected(abandoned) => {
                log::info!("WebSocket reconnected, rebuilding order book snapshots");

                // Only unsubscribe lifecycles the reconnecting connection abandoned are
                // resolved; their confirmations can no longer arrive
                for generation in abandoned {
                    remove_book_drain(book_unsubscribes_pending, generation);
                }

                book_sync.reset_on_reconnect();
            }
            BinanceFuturesWsStreamsMessage::Unsubscribed {
                streams,
                correlation,
            } => {
                log::debug!("Unsubscribe confirmed for streams {streams:?}");

                if let Some(correlation) = correlation {
                    remove_book_drain(book_unsubscribes_pending, correlation);
                }
            }
        }
    }

    fn send_top_of_book(
        data_sender: &EventSender<DataEvent>,
        l1_book_subscriptions: &Arc<AtomicMap<InstrumentId, u32>>,
        quote: QuoteTick,
        sequence: u64,
    ) {
        Self::send_data(data_sender, Data::Quote(quote));
        if l1_book_subscriptions.contains_key(&quote.instrument_id) {
            let deltas = quote_to_l1_deltas(quote, sequence);
            Self::send_data(data_sender, Data::BookDeltas(Box::new(deltas)));
        }
    }

    async fn fetch_depth_snapshot(
        http: BinanceFuturesHttpClient,
        instruments: Arc<AtomicMap<InstrumentId, InstrumentAny>>,
        instrument_id: InstrumentId,
        depth: u32,
        clock: &'static AtomicTime,
    ) -> Result<DepthSnapshot, BinanceBookError> {
        let symbol = format_binance_stream_symbol(&instrument_id).to_uppercase();
        let params = BinanceDepthParams {
            symbol,
            limit: Some(depth),
        };

        let order_book = http
            .depth(&params)
            .await
            .map_err(|e| depth_snapshot_error(instrument_id, &e))?;

        let ts_init = clock.get_time_ns();

        let (price_precision, size_precision) = instruments
            .load()
            .get(&instrument_id)
            .map(|instrument| (instrument.price_precision(), instrument.size_precision()))
            .ok_or_else(|| {
                BinanceBookError::Permanent(format!("no instrument cached for {instrument_id}"))
            })?;

        let deltas = parse_order_book_snapshot(
            &order_book,
            instrument_id,
            price_precision,
            size_precision,
            ts_init,
        );

        Ok(DepthSnapshot {
            last_update_id: order_book.last_update_id as u64,
            deltas,
            has_event_time: true,
        })
    }
}

fn upsert_instruments(
    cache: &Arc<AtomicMap<InstrumentId, InstrumentAny>>,
    instruments: &[InstrumentAny],
) {
    if instruments.is_empty() {
        return;
    }

    cache.rcu(|cache| {
        for instrument in instruments {
            cache.insert(instrument.id(), instrument.clone());
        }
    });
}

fn parse_order_book_snapshot(
    order_book: &BinanceOrderBook,
    instrument_id: InstrumentId,
    price_precision: u8,
    size_precision: u8,
    ts_init: UnixNanos,
) -> OrderBookDeltas {
    let sequence = order_book.last_update_id as u64;
    let ts_event = order_book.transaction_time.map_or(ts_init, |value| {
        parse_millis_or_init(
            value,
            "Futures order book snapshot transaction time",
            ts_init,
        )
    });

    let total_levels = order_book.bids.len() + order_book.asks.len();
    let mut deltas = Vec::with_capacity(total_levels + 1);

    deltas.push(OrderBookDelta::clear(
        instrument_id,
        sequence,
        ts_event,
        ts_init,
    ));

    for (price_str, qty_str) in &order_book.bids {
        let Some(price) = parse_price_at_precision(price_str, price_precision) else {
            log::warn!(
                "Skipping Futures order book bid level for {instrument_id}: invalid or \
                non-positive price='{price_str}'"
            );
            continue;
        };
        let Some(size) = parse_quantity_at_precision(qty_str, size_precision) else {
            log::warn!(
                "Skipping Futures order book bid level for {instrument_id}: invalid or \
                non-positive quantity='{qty_str}'"
            );
            continue;
        };

        let order = BookOrder::new(OrderSide::Buy, price, size, 0);

        deltas.push(OrderBookDelta::new(
            instrument_id,
            BookAction::Add,
            order,
            RecordFlag::F_SNAPSHOT as u8,
            sequence,
            ts_event,
            ts_init,
        ));
    }

    for (price_str, qty_str) in &order_book.asks {
        let Some(price) = parse_price_at_precision(price_str, price_precision) else {
            log::warn!(
                "Skipping Futures order book ask level for {instrument_id}: invalid or \
                non-positive price='{price_str}'"
            );
            continue;
        };
        let Some(size) = parse_quantity_at_precision(qty_str, size_precision) else {
            log::warn!(
                "Skipping Futures order book ask level for {instrument_id}: invalid or \
                non-positive quantity='{qty_str}'"
            );
            continue;
        };

        let order = BookOrder::new(OrderSide::Sell, price, size, 0);

        deltas.push(OrderBookDelta::new(
            instrument_id,
            BookAction::Add,
            order,
            RecordFlag::F_SNAPSHOT as u8,
            sequence,
            ts_event,
            ts_init,
        ));
    }

    if let Some(delta) = deltas.last_mut() {
        delta.flags |= RecordFlag::F_LAST as u8;
    }

    OrderBookDeltas::new(instrument_id, deltas)
}

// Futures `/depth` request weight by level limit, shared by USD-M and COIN-M
fn depth_request_weight(limit: u32) -> u32 {
    match limit {
        ..=50 => 2,
        51..=100 => 5,
        101..=500 => 10,
        _ => 20,
    }
}

// The HTTP client has already spent its own retries, so an exhausted budget stays retryable
fn depth_snapshot_error(
    instrument_id: InstrumentId,
    e: &BinanceFuturesHttpError,
) -> BinanceBookError {
    let message = format!("depth snapshot request for {instrument_id} failed: {e}");

    if e.is_retryable() || matches!(e, BinanceFuturesHttpError::RetryBudgetExceeded(_)) {
        BinanceBookError::Retryable(message)
    } else {
        BinanceBookError::Permanent(message)
    }
}

#[async_trait::async_trait(?Send)]
impl DataClient for BinanceFuturesDataClient {
    fn client_id(&self) -> ClientId {
        self.client_id
    }

    fn venue(&self) -> Option<Venue> {
        Some(self.venue())
    }

    fn start(&mut self) -> anyhow::Result<()> {
        log::info!(
            "Started: client_id={}, product_type={:?}, environment={:?}",
            self.client_id,
            self.product_type,
            self.config.environment,
        );
        Ok(())
    }

    fn stop(&mut self) -> anyhow::Result<()> {
        log::info!("Stopping {id}", id = self.client_id);
        self.session_tasks.begin_shutdown();
        self.command_tasks.begin_shutdown();
        self.ws_client.begin_shutdown();
        self.ws_public_client.begin_shutdown();
        self.is_connected.store(false, Ordering::Relaxed);
        Ok(())
    }

    fn reset(&mut self) -> anyhow::Result<()> {
        log::debug!("Resetting {id}", id = self.client_id);

        self.session_tasks.begin_shutdown();
        self.command_tasks.begin_shutdown();
        self.ws_client.begin_shutdown();
        self.ws_public_client.begin_shutdown();
        self.is_connected.store(false, Ordering::Relaxed);

        // Clear subscription state so resubscribes issue fresh WS subscribes
        self.mark_price_refs.store(AHashMap::new());
        self.ticker_refs.store(AHashMap::new());
        self.force_order_refs.store(AHashMap::new());
        self.force_order_all_market_refs.store(0, Ordering::Relaxed);
        self.force_order_all_market_stream_active
            .store(false, Ordering::Release);
        self.book_subscriptions.store(AHashMap::new());
        self.book_unsubscribes_pending.store(AHashMap::new());
        self.l1_book_subscriptions.store(AHashMap::new());
        self.quote_refs.store(AHashMap::new());
        self.book_sync.clear();

        Ok(())
    }

    fn dispose(&mut self) -> anyhow::Result<()> {
        log::debug!("Disposing {id}", id = self.client_id);
        self.stop()
    }

    async fn connect(&mut self) -> anyhow::Result<()> {
        if self.is_connected() && self.session_tasks.is_open() && self.command_tasks.is_open() {
            return Ok(());
        }

        register_binance_custom_data();

        self.prepare_task_groups().await?;
        let ws_client = self.ws_client.clone();
        let ws_public_client = self.ws_public_client.clone();
        let setup_guard =
            TaskGroupGuard::new(&[&self.session_tasks, &self.command_tasks], move || {
                ws_client.begin_shutdown();
                ws_public_client.begin_shutdown();
            });

        Self::refresh_instrument_catalog(
            &self.http_client,
            &self.config.instrument_provider,
            &self.instruments,
            &self.status_cache,
            &self.instrument_update_lock,
            &self.ws_client,
            &self.ws_public_client,
            &self.data_sender,
            self.clock,
            false,
        )
        .await?;

        let session_result = async {
            log::info!("Connecting to Binance Futures market WebSocket...");
            self.ws_client.connect().await.map_err(|e| {
                anyhow::anyhow!("failed to connect Binance Futures market WebSocket: {e}")
            })?;
            log::info!("Binance Futures market WebSocket connected");

            log::info!("Connecting to Binance Futures public WebSocket...");
            self.ws_public_client.connect().await.map_err(|e| {
                anyhow::anyhow!("failed to connect Binance Futures public WebSocket: {e}")
            })?;
            log::info!("Binance Futures public WebSocket connected");

            let stream = self.ws_client.stream();
            let sender = self.data_sender.clone();
            let insts = self.instruments.clone();
            let ws_insts = self.ws_client.instruments_cache();
            let book_sync = self.book_sync.clone();
            let book_subs = self.book_subscriptions.clone();
            let book_unsubscribes_pending = self.book_unsubscribes_pending.clone();
            let l1_book_subs = self.l1_book_subscriptions.clone();
            let force_order_refs = self.force_order_refs.clone();
            let ticker_refs = self.ticker_refs.clone();
            let force_order_all_market_refs = self.force_order_all_market_refs.clone();
            let force_order_all_market_stream_active =
                self.force_order_all_market_stream_active.clone();
            let http = self.http_client.clone();
            let snapshot_timeout = Duration::from_secs(self.config.book_snapshot_timeout_secs);
            let clock = self.clock;
            let cancel = self.cancellation_token.clone();
            let command_spawner = self
                .command_tasks
                .spawner()
                .context("Binance Futures command task admission is closed")?;

            let future = async move {
                pin_mut!(stream);

                loop {
                    tokio::select! {
                        Some(message) = stream.next() => {
                            Self::handle_ws_message(
                                message,
                                &sender,
                                &insts,
                                &ws_insts,
                                &book_sync,
                                &book_subs,
                                &book_unsubscribes_pending,
                                &l1_book_subs,
                                &force_order_refs,
                                &ticker_refs,
                                &force_order_all_market_refs,
                                &force_order_all_market_stream_active,
                                &http,
                                snapshot_timeout,
                                clock,
                                &command_spawner,
                            );
                        }
                        () = cancel.cancelled() => {
                            log::debug!("Market WebSocket stream task cancelled");
                            break;
                        }
                    }
                }
            };
            self.session_tasks
                .spawn(future)
                .context("failed to register Binance Futures market stream task")?;

            let pub_stream = self.ws_public_client.stream();
            let pub_sender = self.data_sender.clone();
            let pub_insts = self.instruments.clone();
            let pub_ws_insts = self.ws_public_client.instruments_cache();
            let pub_book_sync = self.book_sync.clone();
            let pub_book_subs = self.book_subscriptions.clone();
            let pub_book_unsubscribes_pending = self.book_unsubscribes_pending.clone();
            let pub_l1_book_subs = self.l1_book_subscriptions.clone();
            let pub_force_order_refs = self.force_order_refs.clone();
            let pub_ticker_refs = self.ticker_refs.clone();
            let pub_force_order_all_market_refs = self.force_order_all_market_refs.clone();
            let pub_force_order_all_market_stream_active =
                self.force_order_all_market_stream_active.clone();
            let pub_http = self.http_client.clone();
            let pub_cancel = self.cancellation_token.clone();
            let pub_command_spawner = self
                .command_tasks
                .spawner()
                .context("Binance Futures command task admission is closed")?;

            let future = async move {
                pin_mut!(pub_stream);

                loop {
                    tokio::select! {
                        Some(message) = pub_stream.next() => {
                            Self::handle_ws_message(
                                message,
                                &pub_sender,
                                &pub_insts,
                                &pub_ws_insts,
                                &pub_book_sync,
                                &pub_book_subs,
                                &pub_book_unsubscribes_pending,
                                &pub_l1_book_subs,
                                &pub_force_order_refs,
                                &pub_ticker_refs,
                                &pub_force_order_all_market_refs,
                                &pub_force_order_all_market_stream_active,
                                &pub_http,
                                snapshot_timeout,
                                clock,
                                &pub_command_spawner,
                            );
                        }
                        () = pub_cancel.cancelled() => {
                            log::debug!("Public WebSocket stream task cancelled");
                            break;
                        }
                    }
                }
            };
            self.session_tasks
                .spawn(future)
                .context("failed to register Binance Futures public stream task")?;

            let poll_secs = self.config.instrument_status_poll_secs;
            if poll_secs > 0 {
                let poll_http = self.http_client.clone();
                let poll_sender = self.data_sender.clone();
                let poll_instruments = self.instruments.clone();
                let poll_status_cache = self.status_cache.clone();
                let poll_update_lock = self.instrument_update_lock.clone();
                let poll_cancel = self.cancellation_token.clone();
                let poll_clock = self.clock;

                let future = async move {
                    let mut interval =
                        tokio::time::interval(tokio::time::Duration::from_secs(poll_secs));
                    interval.tick().await; // Skip first immediate tick

                    loop {
                        tokio::select! {
                            _ = interval.tick() => {
                                let _guard = poll_update_lock.lock().await;

                                match poll_http.request_symbol_statuses().await {
                                    Ok(symbol_statuses) => {
                                        let ts = poll_clock.get_time_ns();
                                        let inst_guard = poll_instruments.load();

                                        let raw_to_id: AHashMap<Ustr, InstrumentId> = inst_guard
                                            .values()
                                            .map(|inst| (inst.raw_symbol().inner(), inst.id()))
                                            .collect();

                                        let mut new_statuses = AHashMap::new();

                                        for (raw_symbol, action) in &symbol_statuses {
                                            if let Some(&id) = raw_to_id.get(raw_symbol) {
                                                new_statuses.insert(id, *action);
                                            }
                                        }
                                        drop(inst_guard);

                                        let mut cache = (**poll_status_cache.load()).clone();
                                        diff_and_emit_statuses(
                                            &new_statuses, &mut cache, &poll_sender, ts, ts,
                                        );
                                        poll_status_cache.store(cache);
                                    }
                                    Err(e) => {
                                        log::warn!("Futures instrument status poll failed: {e}");
                                    }
                                }
                            }
                            () = poll_cancel.cancelled() => {
                                log::debug!("Futures instrument status polling task cancelled");
                                break;
                            }
                        }
                    }
                };
                self.session_tasks
                    .spawn(future)
                    .context("failed to register Binance Futures status polling task")?;
                log::debug!("Futures instrument status polling started: interval={poll_secs}s");
            }

            let refresh_secs = self.config.instrument_refresh_interval_secs;
            if refresh_secs > 0 {
                let http = self.http_client.clone();
                let provider = self.config.instrument_provider.clone();
                let instruments = self.instruments.clone();
                let statuses = self.status_cache.clone();
                let update_lock = self.instrument_update_lock.clone();
                let ws = self.ws_client.clone();
                let ws_public = self.ws_public_client.clone();
                let sender = self.data_sender.clone();
                let clock = self.clock;
                let cancel = self.cancellation_token.clone();

                let future = async move {
                    let mut interval = tokio::time::interval(Duration::from_secs(refresh_secs));
                    interval.tick().await;

                    loop {
                        tokio::select! {
                            _ = interval.tick() => {
                                if let Err(e) = Self::refresh_instrument_catalog(
                                    &http,
                                    &provider,
                                    &instruments,
                                    &statuses,
                                    &update_lock,
                                    &ws,
                                    &ws_public,
                                    &sender,
                                    clock,
                                    true,
                                ).await {
                                    log::warn!("Binance Futures instrument refresh failed: {e}");
                                }
                            }
                            () = cancel.cancelled() => {
                                log::debug!("Binance Futures instrument refresh task cancelled");
                                break;
                            }
                        }
                    }
                };
                self.session_tasks
                    .spawn(future)
                    .context("failed to register Binance Futures instrument refresh task")?;
                log::debug!("Futures instrument refresh started: interval={refresh_secs}s");
            }

            Ok::<(), anyhow::Error>(())
        }
        .await;

        if let Err(e) = session_result {
            if let Err(teardown_error) = self.teardown_partial_connect().await {
                return Err(e.context(format!(
                    "Binance Futures data startup teardown failed: {teardown_error}"
                )));
            }
            return Err(e);
        }

        setup_guard.disarm();
        self.is_connected.store(true, Ordering::Release);
        log::info!("Connected: client_id={}", self.client_id);
        Ok(())
    }

    async fn disconnect(&mut self) -> anyhow::Result<()> {
        self.teardown_partial_connect().await?;

        // Clear subscription state so resubscribes issue fresh WS subscribes
        self.mark_price_refs.store(AHashMap::new());
        self.ticker_refs.store(AHashMap::new());
        self.force_order_refs.store(AHashMap::new());
        self.force_order_all_market_refs.store(0, Ordering::Relaxed);
        self.force_order_all_market_stream_active
            .store(false, Ordering::Release);
        self.book_subscriptions.store(AHashMap::new());
        self.book_unsubscribes_pending.store(AHashMap::new());
        self.l1_book_subscriptions.store(AHashMap::new());
        self.quote_refs.store(AHashMap::new());
        self.book_sync.clear();

        self.is_connected.store(false, Ordering::Release);
        log::info!("Disconnected: client_id={}", self.client_id);
        Ok(())
    }

    fn is_connected(&self) -> bool {
        self.is_connected.load(Ordering::Relaxed)
    }

    fn is_disconnected(&self) -> bool {
        !self.is_connected()
    }

    fn subscribe(&mut self, cmd: SubscribeCustomData) -> anyhow::Result<()> {
        let data_type = cmd.data_type.type_name();
        if data_type == "BinanceFuturesTicker" {
            return subscribe_ticker(self, &cmd.data_type);
        }

        if data_type == "BinanceFuturesMarkPriceUpdate" {
            let instrument_id = Self::required_instrument_id_metadata(&cmd.data_type)?;
            anyhow::ensure!(
                instrument_id.venue == self.venue(),
                "Futures mark price requires a BINANCE instrument"
            );
            let should_subscribe = {
                let previous = self
                    .mark_price_refs
                    .load()
                    .get(&instrument_id)
                    .copied()
                    .unwrap_or(0);
                self.mark_price_refs
                    .rcu(|refs| *refs.entry(instrument_id).or_insert(0) += 1);
                previous == 0
            };

            if should_subscribe {
                let ws = self.ws_client.clone();
                let stream = format!(
                    "{}@markPrice@1s",
                    format_binance_stream_symbol(&instrument_id)
                );
                self.spawn_ws(
                    async move {
                        ws.subscribe(vec![stream])
                            .await
                            .context("mark price custom subscription")
                    },
                    "mark price custom subscription",
                );
            }
            return Ok(());
        }

        if data_type != "BinanceFuturesLiquidation" {
            log::warn!("Unsupported custom data subscription: {data_type}");
            return Ok(());
        }

        let instrument_id = Self::custom_liquidation_instrument_id(&cmd.data_type)?;
        if let Some(instrument_id) = instrument_id {
            if instrument_id.venue != self.venue() {
                anyhow::bail!(
                    "Binance liquidation custom data requires BINANCE venue instrument, received {instrument_id}"
                );
            }

            let should_subscribe = {
                let prev = self
                    .force_order_refs
                    .load()
                    .get(&instrument_id)
                    .copied()
                    .unwrap_or(0);
                self.force_order_refs.rcu(|m| {
                    let count = m.entry(instrument_id).or_insert(0);
                    *count += 1;
                });
                prev == 0
            };

            let has_all_market_subscription =
                self.force_order_all_market_refs.load(Ordering::Relaxed) > 0;
            let has_all_market_stream = self
                .force_order_all_market_stream_active
                .load(Ordering::Acquire);

            if should_subscribe && !has_all_market_subscription && !has_all_market_stream {
                let ws = self.ws_client.clone();
                let stream = Self::liquidation_stream(&instrument_id);
                self.spawn_ws(
                    async move {
                        ws.subscribe(vec![stream])
                            .await
                            .context("forceOrder subscription")
                    },
                    "forceOrder subscription",
                );
            } else if should_subscribe && !has_all_market_subscription {
                self.spawn_liquidation_stream_reconcile("forceOrder subscription restore");
            }

            return Ok(());
        }

        let should_subscribe = self
            .force_order_all_market_refs
            .fetch_add(1, Ordering::Relaxed)
            == 0;

        if should_subscribe {
            self.spawn_liquidation_stream_reconcile("all-market forceOrder subscription");
        }

        Ok(())
    }

    fn subscribe_instruments(&mut self, _cmd: SubscribeInstruments) -> anyhow::Result<()> {
        log::debug!(
            "subscribe_instruments: Binance Futures instruments are fetched via HTTP on connect"
        );
        Ok(())
    }

    fn subscribe_instrument(&mut self, _cmd: SubscribeInstrument) -> anyhow::Result<()> {
        log::debug!(
            "subscribe_instrument: Binance Futures instruments are fetched via HTTP on connect"
        );
        Ok(())
    }

    fn subscribe_book_deltas(&mut self, cmd: SubscribeBookDeltas) -> anyhow::Result<()> {
        execute_instrument_subscriptions(&cmd.instrument_ids, |instrument_id| {
            let cmd = cmd.with_instrument_ids(instrument_id);
            if cmd.book_type == BookType::L1_MBP {
                anyhow::ensure!(
                    cmd.depth.is_none_or(|depth| depth.get() == 1),
                    "Binance Futures L1_MBP supports depth 1 only"
                );
                anyhow::ensure!(
                    !self.book_subscriptions.contains_key(&instrument_id),
                    "cannot subscribe L1_MBP and L2_MBP for the same Binance Futures instrument"
                );
                self.l1_book_subscriptions.rcu(|subscriptions| {
                    *subscriptions.entry(instrument_id).or_insert(0) += 1;
                });

                self.subscribe_top_of_book(instrument_id);
                return Ok(());
            }

            if cmd.book_type != BookType::L2_MBP {
                anyhow::bail!(
                    "Binance Futures supports L1_MBP and L2_MBP order book subscriptions"
                );
            }

            anyhow::ensure!(
                !self.l1_book_subscriptions.contains_key(&instrument_id),
                "cannot subscribe L1_MBP and L2_MBP for the same Binance Futures instrument"
            );

            let depth = cmd.depth.map_or(1000, |d| d.get() as u32);

            if !BINANCE_BOOK_DEPTHS.contains(&depth) {
                anyhow::bail!(
                    "Invalid depth {depth} for Binance Futures order book. \
                    Valid values: {BINANCE_BOOK_DEPTHS:?}"
                );
            }

            if let Some(existing) = self.book_subscriptions.load().get(&instrument_id) {
                anyhow::ensure!(
                    *existing == depth,
                    "Binance Futures book depth cannot change while subscribed"
                );
            }

            // Establish gate protection before publishing the new semantics, so a
            // concurrent consumer cannot parse an old-stream frame with them
            let stream = book_stream(&instrument_id, depth);
            revive_book_drains(&self.book_unsubscribes_pending, instrument_id, &stream);

            self.book_subscriptions.insert(instrument_id, depth);

            // A resubscription to the drained stream carries identical frame semantics, so
            // the gate no longer needs to hold for it; its confirmation may never arrive
            satisfy_book_drain(&self.book_unsubscribes_pending, instrument_id, &stream);

            if is_partial_book_depth(depth) {
                let ws = self.ws_public_client.clone();
                self.spawn_ws(
                    chain_command(&self.book_command_tail, async move {
                        ws.subscribe(vec![stream])
                            .await
                            .context("book deltas subscription")
                    }),
                    "order book subscription",
                );

                return Ok(());
            }

            // Resync from the next diff, cancelling any recovery from a prior subscription
            self.book_sync.subscribe(instrument_id);

            // Subscribe to the unthrottled diff depth stream for Futures.
            let ws = self.ws_public_client.clone();
            self.spawn_ws(
                chain_command(&self.book_command_tail, async move {
                    ws.subscribe(vec![stream])
                        .await
                        .context("book deltas subscription")
                }),
                "order book subscription",
            );

            Ok(())
        })
    }

    fn subscribe_quotes(&mut self, cmd: SubscribeQuotes) -> anyhow::Result<()> {
        execute_instrument_subscriptions(&cmd.instrument_ids, |instrument_id| {
            self.subscribe_top_of_book(instrument_id);
            Ok(())
        })
    }

    fn subscribe_trades(&mut self, cmd: SubscribeTrades) -> anyhow::Result<()> {
        execute_instrument_subscriptions(&cmd.instrument_ids, |instrument_id| {
            let ws = self.ws_client.clone();

            // Binance Futures uses aggTrade for aggregate trades
            let stream = format!("{}@aggTrade", format_binance_stream_symbol(&instrument_id));

            self.spawn_ws(
                async move {
                    ws.subscribe(vec![stream])
                        .await
                        .context("trades subscription")
                },
                "trade subscription",
            );

            Ok(())
        })
    }

    fn subscribe_bars(&mut self, cmd: SubscribeBars) -> anyhow::Result<()> {
        execute_bar_subscriptions(&cmd.bar_types, |bar_type| {
            let ws = self.ws_client.clone();
            let interval = bar_spec_to_binance_interval(bar_type.spec())?;
            anyhow::ensure!(
                interval != crate::common::enums::BinanceKlineInterval::Second1,
                "Binance Futures does not support second-level kline intervals"
            );

            let stream = format!(
                "{}@kline_{}",
                format_binance_stream_symbol(&bar_type.instrument_id()),
                interval.as_str()
            );

            self.spawn_ws(
                async move {
                    ws.subscribe(vec![stream])
                        .await
                        .context("bars subscription")
                },
                "bar subscription",
            );

            Ok(())
        })
    }

    fn subscribe_mark_prices(&mut self, cmd: SubscribeMarkPrices) -> anyhow::Result<()> {
        execute_instrument_subscriptions(&cmd.instrument_ids, |instrument_id| {
            let should_subscribe = {
                let prev = self
                    .mark_price_refs
                    .load()
                    .get(&instrument_id)
                    .copied()
                    .unwrap_or(0);
                self.mark_price_refs.rcu(|m| {
                    let count = m.entry(instrument_id).or_insert(0);
                    *count += 1;
                });

                prev == 0
            };

            if should_subscribe {
                let ws = self.ws_client.clone();
                let stream = format!(
                    "{}@markPrice@1s",
                    format_binance_stream_symbol(&instrument_id)
                );

                self.spawn_ws(
                    async move {
                        ws.subscribe(vec![stream])
                            .await
                            .context("mark prices subscription")
                    },
                    "mark prices subscription",
                );
            }

            Ok(())
        })
    }

    fn subscribe_index_prices(&mut self, cmd: SubscribeIndexPrices) -> anyhow::Result<()> {
        execute_instrument_subscriptions(&cmd.instrument_ids, |instrument_id| {
            let should_subscribe = {
                let prev = self
                    .mark_price_refs
                    .load()
                    .get(&instrument_id)
                    .copied()
                    .unwrap_or(0);
                self.mark_price_refs.rcu(|m| {
                    let count = m.entry(instrument_id).or_insert(0);
                    *count += 1;
                });

                prev == 0
            };

            if should_subscribe {
                let ws = self.ws_client.clone();
                let stream = format!(
                    "{}@markPrice@1s",
                    format_binance_stream_symbol(&instrument_id)
                );

                self.spawn_ws(
                    async move {
                        ws.subscribe(vec![stream])
                            .await
                            .context("index prices subscription")
                    },
                    "index prices subscription",
                );
            }

            Ok(())
        })
    }

    fn subscribe_funding_rates(&mut self, cmd: SubscribeFundingRates) -> anyhow::Result<()> {
        execute_instrument_subscriptions(&cmd.instrument_ids, |instrument_id| {
            let should_subscribe = {
                let prev = self
                    .mark_price_refs
                    .load()
                    .get(&instrument_id)
                    .copied()
                    .unwrap_or(0);
                self.mark_price_refs.rcu(|m| {
                    let count = m.entry(instrument_id).or_insert(0);
                    *count += 1;
                });

                prev == 0
            };

            if should_subscribe {
                let ws = self.ws_client.clone();
                let stream = format!(
                    "{}@markPrice@1s",
                    format_binance_stream_symbol(&instrument_id)
                );

                self.spawn_ws(
                    async move {
                        ws.subscribe(vec![stream])
                            .await
                            .context("funding rates subscription")
                    },
                    "funding rates subscription",
                );
            }

            Ok(())
        })
    }

    fn subscribe_instrument_status(
        &mut self,
        cmd: SubscribeInstrumentStatus,
    ) -> anyhow::Result<()> {
        execute_instrument_subscriptions(&cmd.instrument_ids, |instrument_id| {
            log::debug!(
                "subscribe_instrument_status: {instrument_id} (status changes detected via periodic exchange info polling)",
            );
            Ok(())
        })
    }

    fn unsubscribe_book_deltas(&mut self, cmd: &UnsubscribeBookDeltas) -> anyhow::Result<()> {
        execute_instrument_subscriptions(&cmd.instrument_ids, |instrument_id| {
            if let Some(count) = self
                .l1_book_subscriptions
                .load()
                .get(&instrument_id)
                .copied()
            {
                if count == 1 {
                    self.l1_book_subscriptions.remove(&instrument_id);
                } else {
                    self.l1_book_subscriptions.rcu(|subscriptions| {
                        if let Some(existing) = subscriptions.get_mut(&instrument_id) {
                            *existing -= 1;
                        }
                    });
                }

                self.unsubscribe_top_of_book(instrument_id);
                return Ok(());
            }

            let ws = self.ws_public_client.clone();

            let Some(depth) = self.book_subscriptions.load().get(&instrument_id).copied() else {
                return Ok(());
            };

            self.book_subscriptions.remove(&instrument_id);

            // Stop book sync so an in-flight snapshot cannot emit after unsubscribe
            self.book_sync.remove(instrument_id);

            let stream = book_stream(&instrument_id, depth);
            let generation = self.book_drain_generation;
            self.book_drain_generation += 1;

            // Gate depth frames on the venue's unsubscribe confirmation. Arm eagerly so a
            // not-yet-registered subscribe cannot slip frames past the gate; the spawned
            // task disarms when the pool reports no unsubscribe was sent
            arm_book_drain(
                &self.book_unsubscribes_pending,
                instrument_id,
                generation,
                &stream,
            );

            let book_unsubscribes_pending = self.book_unsubscribes_pending.clone();
            self.spawn_ws(
                chain_command(&self.book_command_tail, async move {
                    let sent = ws
                        .unsubscribe_correlated(vec![stream.clone()], generation)
                        .await
                        .context("book deltas unsubscribe");

                    match sent {
                        Ok(sent) if sent.contains(&stream) => Ok(()),
                        Ok(_) => {
                            remove_book_drain(&book_unsubscribes_pending, generation);
                            Ok(())
                        }
                        Err(e) => {
                            remove_book_drain(&book_unsubscribes_pending, generation);
                            Err(e)
                        }
                    }
                }),
                "order book unsubscribe",
            );

            Ok(())
        })
    }

    fn unsubscribe_quotes(&mut self, cmd: &UnsubscribeQuotes) -> anyhow::Result<()> {
        execute_instrument_subscriptions(&cmd.instrument_ids, |instrument_id| {
            self.unsubscribe_top_of_book(instrument_id);
            Ok(())
        })
    }

    fn unsubscribe_trades(&mut self, cmd: &UnsubscribeTrades) -> anyhow::Result<()> {
        execute_instrument_subscriptions(&cmd.instrument_ids, |instrument_id| {
            let ws = self.ws_client.clone();

            let stream = format!("{}@aggTrade", format_binance_stream_symbol(&instrument_id));

            self.spawn_ws(
                async move {
                    ws.unsubscribe(vec![stream])
                        .await
                        .context("trades unsubscribe")
                        .map(|_| ())
                },
                "trade unsubscribe",
            );

            Ok(())
        })
    }

    fn unsubscribe(&mut self, cmd: &UnsubscribeCustomData) -> anyhow::Result<()> {
        let data_type = cmd.data_type.type_name();
        if data_type == "BinanceFuturesTicker" {
            return unsubscribe_ticker(self, &cmd.data_type);
        }

        if data_type == "BinanceFuturesMarkPriceUpdate" {
            let instrument_id = Self::required_instrument_id_metadata(&cmd.data_type)?;
            let should_unsubscribe = match self.mark_price_refs.load().get(&instrument_id).copied()
            {
                Some(1) => {
                    self.mark_price_refs.remove(&instrument_id);
                    true
                }
                Some(count) if count > 1 => {
                    self.mark_price_refs.rcu(|refs| {
                        if let Some(existing) = refs.get_mut(&instrument_id) {
                            *existing -= 1;
                        }
                    });
                    false
                }
                _ => false,
            };

            if should_unsubscribe {
                let ws = self.ws_client.clone();
                let stream = format!(
                    "{}@markPrice@1s",
                    format_binance_stream_symbol(&instrument_id)
                );
                self.spawn_ws(
                    async move {
                        ws.unsubscribe(vec![stream])
                            .await
                            .context("mark price custom unsubscribe")
                            .map(|_| ())
                    },
                    "mark price custom unsubscribe",
                );
            }
            return Ok(());
        }

        if data_type != "BinanceFuturesLiquidation" {
            log::warn!("Unsupported custom data unsubscription: {data_type}");
            return Ok(());
        }

        let instrument_id = Self::custom_liquidation_instrument_id(&cmd.data_type)?;
        if let Some(instrument_id) = instrument_id {
            if instrument_id.venue != self.venue() {
                anyhow::bail!(
                    "Binance liquidation custom data requires BINANCE venue instrument, received {instrument_id}"
                );
            }

            let should_unsubscribe = {
                let prev = self.force_order_refs.load().get(&instrument_id).copied();
                match prev {
                    Some(1) => {
                        self.force_order_refs.remove(&instrument_id);
                        true
                    }
                    Some(count) if count > 1 => {
                        self.force_order_refs.rcu(|m| {
                            if let Some(existing) = m.get_mut(&instrument_id) {
                                *existing -= 1;
                            }
                        });
                        false
                    }
                    _ => false,
                }
            };

            let has_all_market_subscription =
                self.force_order_all_market_refs.load(Ordering::Relaxed) > 0;
            let has_all_market_stream = self
                .force_order_all_market_stream_active
                .load(Ordering::Acquire);

            if should_unsubscribe && !has_all_market_subscription {
                let ws = self.ws_client.clone();
                let stream = Self::liquidation_stream(&instrument_id);
                let ws_lock = self.force_order_ws_lock.clone();
                self.spawn_ws(
                    async move {
                        let _guard = if has_all_market_stream {
                            Some(ws_lock.lock().await)
                        } else {
                            None
                        };
                        ws.unsubscribe(vec![stream])
                            .await
                            .context("forceOrder unsubscribe")
                            .map(|_| ())
                    },
                    "forceOrder unsubscribe",
                );
            }

            return Ok(());
        }

        let should_unsubscribe = self
            .force_order_all_market_refs
            .try_update(Ordering::Relaxed, Ordering::Relaxed, |current| {
                if current == 0 {
                    None
                } else {
                    Some(current - 1)
                }
            })
            .is_ok_and(|prev| prev == 1);

        if should_unsubscribe {
            self.spawn_liquidation_stream_reconcile("all-market forceOrder unsubscribe");
        }

        Ok(())
    }

    fn unsubscribe_bars(&mut self, cmd: &UnsubscribeBars) -> anyhow::Result<()> {
        execute_bar_subscriptions(&cmd.bar_types, |bar_type| {
            let ws = self.ws_client.clone();
            let interval = bar_spec_to_binance_interval(bar_type.spec())?;

            let stream = format!(
                "{}@kline_{}",
                format_binance_stream_symbol(&bar_type.instrument_id()),
                interval.as_str()
            );

            self.spawn_ws(
                async move {
                    ws.unsubscribe(vec![stream])
                        .await
                        .context("bars unsubscribe")
                        .map(|_| ())
                },
                "bar unsubscribe",
            );

            Ok(())
        })
    }

    fn unsubscribe_mark_prices(&mut self, cmd: &UnsubscribeMarkPrices) -> anyhow::Result<()> {
        execute_instrument_subscriptions(&cmd.instrument_ids, |instrument_id| {
            let should_unsubscribe = {
                let prev = self.mark_price_refs.load().get(&instrument_id).copied();
                match prev {
                    Some(count) if count <= 1 => {
                        self.mark_price_refs.remove(&instrument_id);
                        true
                    }
                    Some(_) => {
                        self.mark_price_refs.rcu(|m| {
                            if let Some(count) = m.get_mut(&instrument_id) {
                                *count = count.saturating_sub(1);
                            }
                        });

                        false
                    }
                    None => false,
                }
            };

            if should_unsubscribe {
                let ws = self.ws_client.clone();
                let symbol_lower = format_binance_stream_symbol(&instrument_id);
                let streams = vec![
                    format!("{symbol_lower}@markPrice"),
                    format!("{symbol_lower}@markPrice@1s"),
                    format!("{symbol_lower}@markPrice@3s"),
                ];

                self.spawn_ws(
                    async move {
                        ws.unsubscribe(streams)
                            .await
                            .context("mark prices unsubscribe")
                            .map(|_| ())
                    },
                    "mark prices unsubscribe",
                );
            }

            Ok(())
        })
    }

    fn unsubscribe_index_prices(&mut self, cmd: &UnsubscribeIndexPrices) -> anyhow::Result<()> {
        execute_instrument_subscriptions(&cmd.instrument_ids, |instrument_id| {
            let should_unsubscribe = {
                let prev = self.mark_price_refs.load().get(&instrument_id).copied();
                match prev {
                    Some(count) if count <= 1 => {
                        self.mark_price_refs.remove(&instrument_id);
                        true
                    }
                    Some(_) => {
                        self.mark_price_refs.rcu(|m| {
                            if let Some(count) = m.get_mut(&instrument_id) {
                                *count = count.saturating_sub(1);
                            }
                        });

                        false
                    }
                    None => false,
                }
            };

            if should_unsubscribe {
                let ws = self.ws_client.clone();
                let symbol_lower = format_binance_stream_symbol(&instrument_id);
                let streams = vec![
                    format!("{symbol_lower}@markPrice"),
                    format!("{symbol_lower}@markPrice@1s"),
                    format!("{symbol_lower}@markPrice@3s"),
                ];

                self.spawn_ws(
                    async move {
                        ws.unsubscribe(streams)
                            .await
                            .context("index prices unsubscribe")
                            .map(|_| ())
                    },
                    "index prices unsubscribe",
                );
            }

            Ok(())
        })
    }

    fn unsubscribe_funding_rates(&mut self, cmd: &UnsubscribeFundingRates) -> anyhow::Result<()> {
        execute_instrument_subscriptions(&cmd.instrument_ids, |instrument_id| {
            let should_unsubscribe = {
                let prev = self.mark_price_refs.load().get(&instrument_id).copied();
                match prev {
                    Some(count) if count <= 1 => {
                        self.mark_price_refs.remove(&instrument_id);
                        true
                    }
                    Some(_) => {
                        self.mark_price_refs.rcu(|m| {
                            if let Some(count) = m.get_mut(&instrument_id) {
                                *count = count.saturating_sub(1);
                            }
                        });

                        false
                    }
                    None => false,
                }
            };

            if should_unsubscribe {
                let ws = self.ws_client.clone();
                let symbol_lower = format_binance_stream_symbol(&instrument_id);
                let streams = vec![
                    format!("{symbol_lower}@markPrice"),
                    format!("{symbol_lower}@markPrice@1s"),
                    format!("{symbol_lower}@markPrice@3s"),
                ];

                self.spawn_ws(
                    async move {
                        ws.unsubscribe(streams)
                            .await
                            .context("funding rates unsubscribe")
                            .map(|_| ())
                    },
                    "funding rates unsubscribe",
                );
            }

            Ok(())
        })
    }

    fn unsubscribe_instrument_status(
        &mut self,
        cmd: &UnsubscribeInstrumentStatus,
    ) -> anyhow::Result<()> {
        execute_instrument_subscriptions(&cmd.instrument_ids, |instrument_id| {
            log::debug!("unsubscribe_instrument_status: {instrument_id}",);
            Ok(())
        })
    }

    fn request_instruments(&self, request: RequestInstruments) -> anyhow::Result<()> {
        let http = self.http_client.clone();
        let sender = self.data_sender.clone();
        let instruments_cache = self.instruments.clone();
        let request_id = request.request_id;
        let client_id = request.client_id.unwrap_or(self.client_id);
        let venue = self.venue();
        let start = request.start;
        let end = request.end;
        let params = request.params;
        let clock = self.clock;
        let provider = self.config.instrument_provider.clone();
        let start_nanos = datetime_to_unix_nanos(start);
        let end_nanos = datetime_to_unix_nanos(end);

        self.spawn_command(async move {
            match http.request_instruments_with_config(&provider).await {
                Ok(instruments) => {
                    upsert_instruments(&instruments_cache, &instruments);

                    let response = DataResponse::Instruments(InstrumentsResponse::new(
                        request_id,
                        client_id,
                        venue,
                        instruments,
                        start_nanos,
                        end_nanos,
                        clock.get_time_ns(),
                        params,
                    ));

                    if let Err(e) = sender.send(DataEvent::Response(response)) {
                        log::error!("Failed to send instruments response: {e}");
                    }
                }
                Err(e) => log::error!("Instruments request failed: {e:?}"),
            }
        });

        Ok(())
    }

    fn request_instrument(&self, request: RequestInstrument) -> anyhow::Result<()> {
        let http = self.http_client.clone();
        let sender = self.data_sender.clone();
        let instruments = self.instruments.clone();
        let instrument_id = request.single_instrument_id()?;
        let request_id = request.request_id;
        let client_id = request.client_id.unwrap_or(self.client_id);
        let start = request.start;
        let end = request.end;
        let params = request.params;
        let clock = self.clock;
        let provider = self.config.instrument_provider.clone();
        let start_nanos = datetime_to_unix_nanos(start);
        let end_nanos = datetime_to_unix_nanos(end);

        self.spawn_command(async move {
            match http.request_instruments_with_config(&provider).await {
                Ok(all_instruments) => {
                    upsert_instruments(&instruments, &all_instruments);

                    let instrument = all_instruments
                        .into_iter()
                        .find(|i| i.id() == instrument_id);

                    if let Some(instrument) = instrument {
                        let response = DataResponse::Instrument(Box::new(InstrumentResponse::new(
                            request_id,
                            client_id,
                            instrument.id(),
                            instrument,
                            start_nanos,
                            end_nanos,
                            clock.get_time_ns(),
                            params,
                        )));

                        if let Err(e) = sender.send(DataEvent::Response(response)) {
                            log::error!("Failed to send instrument response: {e}");
                        }
                    } else {
                        log::error!("Instrument not found: {instrument_id}");
                    }
                }
                Err(e) => log::error!("Instrument request failed: {e:?}"),
            }
        });

        Ok(())
    }

    /// Requests Binance futures custom data.
    ///
    /// Spawned fetch failures are logged and no response is emitted, matching
    /// the existing request-path behavior for other Binance adapter requests.
    fn request_data(&self, request: RequestCustomData) -> anyhow::Result<()> {
        let data_type = request.data_type.clone();
        let data_type_name = data_type.type_name().to_string();

        if data_type_name == "BinanceBar" {
            let bar_type = parse_binance_bar_type(&data_type)?;
            anyhow::ensure!(
                bar_type.aggregation_source() == AggregationSource::External,
                "historical BinanceBar requests require EXTERNAL aggregation"
            );
            anyhow::ensure!(
                bar_type.spec().price_type == PriceType::Last,
                "historical BinanceBar requests require LAST price type"
            );
            anyhow::ensure!(
                bar_type.spec().is_time_aggregated(),
                "historical BinanceBar requests require time aggregation"
            );
            let http = self.http_client.clone();
            let sender = self.data_sender.clone();
            let request_id = request.request_id;
            let client_id = request.client_id;
            let start = request.start;
            let end = request.end;
            let limit = request.limit.map(|value| value.get() as u32);
            let params = request.params;
            let clock = self.clock;
            let venue = self.venue();
            let start_nanos = datetime_to_unix_nanos(start);
            let end_nanos = datetime_to_unix_nanos(end);
            self.spawn_command(async move {
                match http.request_binance_bars(bar_type, start, end, limit).await {
                    Ok(bars) => {
                        let response = DataResponse::Data(CustomDataResponse::new(
                            request_id,
                            client_id,
                            Some(venue),
                            data_type,
                            binance_bars_to_custom_data(bar_type, bars),
                            start_nanos,
                            end_nanos,
                            clock.get_time_ns(),
                            params,
                        ));

                        if let Err(e) = sender.send(DataEvent::Response(response)) {
                            log::error!("Failed to send BinanceBar response: {e}");
                        }
                    }
                    Err(e) => log::error!("BinanceBar request failed for {bar_type}: {e:?}"),
                }
            });
            return Ok(());
        }

        if data_type_name != "BinanceFuturesOpenInterest"
            && data_type_name != "BinanceFuturesOpenInterestHist"
        {
            log::warn!("Unsupported custom data request: {data_type_name}");
            return Ok(());
        }

        let instrument_id = Self::required_instrument_id_metadata(&data_type)?;

        if instrument_id.venue != self.venue() {
            anyhow::bail!(
                "Binance Futures custom data requires BINANCE venue instrument, received {instrument_id}"
            );
        }

        let period = if data_type_name == "BinanceFuturesOpenInterestHist" {
            Some(Self::required_period_metadata(&data_type)?)
        } else {
            None
        };

        let http = self.http_client.clone();
        let sender = self.data_sender.clone();
        let request_id = request.request_id;
        let client_id = request.client_id;
        let params = request.params;
        let clock = self.clock;
        let venue = self.venue();
        let limit = request.limit.map(|n| n.get() as u32);
        let start_nanos = datetime_to_unix_nanos(request.start);
        let end_nanos = datetime_to_unix_nanos(request.end);
        let start_ms = request.start.map(|dt| dt.as_millisecond());
        let end_ms = request.end.map(|dt| dt.as_millisecond());

        self.spawn_command(async move {
            let response = if data_type_name == "BinanceFuturesOpenInterest" {
                let response_data_type = data_type.clone();
                let query = BinanceOpenInterestParams {
                    symbol: format_binance_symbol(&instrument_id),
                };

                match http
                    .open_interest(&query)
                    .await
                    .context("failed to request current open interest from Binance Futures")
                {
                    Ok(open_interest) => {
                        let ts_init = clock.get_time_ns();
                        let open_interest_value = match Self::parse_open_interest_decimal(
                            "open_interest",
                            &open_interest.open_interest,
                        ) {
                            Ok(value) => value,
                            Err(e) => {
                                log::error!(
                                    "Current open interest request failed for {instrument_id}: {e:?}"
                                );
                                return;
                            }
                        };
                        let ts_event = match parse_millis(
                            open_interest.time,
                            "Futures open interest time",
                        ) {
                                Ok(value) => value,
                                Err(e) => {
                                    log::error!(
                                        "Current open interest request failed for {instrument_id}: {e:?}"
                                    );
                                    return;
                                }
                            };
                        let payload = Arc::new(BinanceFuturesOpenInterest::new(
                            instrument_id,
                            open_interest_value,
                            ts_event,
                            ts_init,
                        ));
                        let custom = CustomData::new(payload, response_data_type.clone());

                        Some(DataResponse::Data(CustomDataResponse::new(
                            request_id,
                            client_id,
                            Some(venue),
                            response_data_type,
                            custom,
                            start_nanos,
                            end_nanos,
                            ts_init,
                            params,
                        )))
                    }
                    Err(e) => {
                        log::error!("Current open interest request failed for {instrument_id}: {e:?}");
                        None
                    }
                }
            } else {
                let response_data_type = data_type.clone();
                let period = period.expect("period required for historical open interest");
                let query = match http.product_type() {
                    BinanceProductType::UsdM => BinanceOpenInterestHistParams {
                        symbol: Some(format_binance_symbol(&instrument_id)),
                        pair: None,
                        contract_type: None,
                        period: period.clone(),
                        start_time: start_ms,
                        end_time: end_ms,
                        limit,
                    },
                    BinanceProductType::CoinM => {
                        let (pair, contract_type) =
                            match Self::coinm_open_interest_hist_params(&http, &instrument_id) {
                                Ok(values) => values,
                                Err(e) => {
                                    log::error!(
                                        "Historical open interest request failed for {instrument_id}: {e:?}"
                                    );
                                    return;
                                }
                            };
                        BinanceOpenInterestHistParams {
                            symbol: None,
                            pair: Some(pair),
                            contract_type: Some(contract_type),
                            period: period.clone(),
                            start_time: start_ms,
                            end_time: end_ms,
                            limit,
                        }
                    }
                    product_type => {
                        log::error!(
                            "Historical open interest request failed for {instrument_id}: unsupported product type {product_type:?}"
                        );
                        return;
                    }
                };

                match http
                    .open_interest_hist(&query)
                    .await
                    .context("failed to request historical open interest from Binance Futures")
                {
                    Ok(history) => {
                        let ts_init = clock.get_time_ns();
                        let points: Vec<BinanceFuturesOpenInterestHistPoint> = match history
                            .into_iter()
                            .map(|point| -> anyhow::Result<_> {
                                Ok(BinanceFuturesOpenInterestHistPoint::new(
                                    Self::parse_open_interest_decimal(
                                        "sum_open_interest",
                                        &point.sum_open_interest,
                                    )?,
                                    Self::parse_open_interest_decimal(
                                        "sum_open_interest_value",
                                        &point.sum_open_interest_value,
                                    )?,
                                    parse_millis(
                                        point.timestamp,
                                        "Futures historical open interest timestamp",
                                    )?,
                                ))
                            })
                            .collect()
                        {
                            Ok(points) => points,
                            Err(e) => {
                                log::error!(
                                    "Historical open interest request failed for {instrument_id}: {e:?}"
                                );
                                return;
                            }
                        };
                        let ts_event = points.last().map_or(ts_init, |point| point.ts_event);
                        let payload = Arc::new(BinanceFuturesOpenInterestHist::new(
                            instrument_id,
                            period,
                            points,
                            ts_event,
                            ts_init,
                        ));
                        let custom = CustomData::new(payload, response_data_type.clone());

                        Some(DataResponse::Data(CustomDataResponse::new(
                            request_id,
                            client_id,
                            Some(venue),
                            response_data_type,
                            custom,
                            start_nanos,
                            end_nanos,
                            ts_init,
                            params,
                        )))
                    }
                    Err(e) => {
                        log::error!(
                            "Historical open interest request failed for {instrument_id}: {e:?}"
                        );
                        None
                    }
                }
            };

            if let Some(response) = response
                && let Err(e) = sender.send(DataEvent::Response(response))
            {
                log::error!("Failed to send custom data response: {e}");
            }
        });

        Ok(())
    }

    fn request_trades(&self, request: RequestTrades) -> anyhow::Result<()> {
        let http = self.http_client.clone();
        let sender = self.data_sender.clone();
        let instrument_id = request.single_instrument_id()?;
        let limit = request.limit.map(|n| n.get() as u32);
        let request_id = request.request_id;
        let client_id = request.client_id.unwrap_or(self.client_id);
        let params = request.params;
        let clock = self.clock;
        let start_nanos = datetime_to_unix_nanos(request.start);
        let end_nanos = datetime_to_unix_nanos(request.end);
        let start = request.start;
        let end = request.end;
        anyhow::ensure!(
            limit.is_none_or(|value| value <= 1000),
            "Binance Futures trade limit must not exceed 1000"
        );

        self.spawn_command(async move {
            let result = if start.is_some() || end.is_some() {
                http.request_agg_trades(instrument_id, start, end, limit)
                    .await
            } else {
                http.request_trades(instrument_id, limit).await
            };

            match result.context("failed to request trades from Binance Futures") {
                Ok(trades) => {
                    let response = DataResponse::Trades(TradesResponse::new(
                        request_id,
                        client_id,
                        instrument_id,
                        trades,
                        start_nanos,
                        end_nanos,
                        clock.get_time_ns(),
                        params,
                    ));

                    if let Err(e) = sender.send(DataEvent::Response(response)) {
                        log::error!("Failed to send trades response: {e}");
                    }
                }
                Err(e) => log::error!("Trade request failed: {e:?}"),
            }
        });

        Ok(())
    }

    fn request_funding_rates(&self, request: RequestFundingRates) -> anyhow::Result<()> {
        let http = self.http_client.clone();
        let sender = self.data_sender.clone();
        let instrument_id = request.single_instrument_id()?;
        let start = request.start;
        let end = request.end;
        let limit = request.limit.map(|n| n.get() as u32);
        let request_id = request.request_id;
        let client_id = request.client_id.unwrap_or(self.client_id);
        let params = request.params;
        let clock = self.clock;
        let start_nanos = datetime_to_unix_nanos(start);
        let end_nanos = datetime_to_unix_nanos(end);

        self.spawn_command(async move {
            match http
                .request_funding_rates(instrument_id, start, end, limit)
                .await
                .context("failed to request funding rates from Binance Futures")
            {
                Ok(funding_rates) => {
                    let response = DataResponse::FundingRates(FundingRatesResponse::new(
                        request_id,
                        client_id,
                        instrument_id,
                        funding_rates,
                        start_nanos,
                        end_nanos,
                        clock.get_time_ns(),
                        params,
                    ));

                    if let Err(e) = sender.send(DataEvent::Response(response)) {
                        log::error!("Failed to send funding rates response: {e}");
                    }
                }
                Err(e) => log::error!("Funding rates request failed for {instrument_id}: {e:?}"),
            }
        });

        Ok(())
    }

    fn request_bars(&self, request: RequestBars) -> anyhow::Result<()> {
        let http = self.http_client.clone();
        let sender = self.data_sender.clone();
        let bar_type = request.single_bar_type()?;
        let start = request.start;
        let end = request.end;
        let limit = request.limit.map(|n| n.get() as u32);
        let request_id = request.request_id;
        let client_id = request.client_id.unwrap_or(self.client_id);
        let params = request.params;
        let clock = self.clock;
        let start_nanos = datetime_to_unix_nanos(start);
        let end_nanos = datetime_to_unix_nanos(end);
        anyhow::ensure!(
            bar_type.aggregation_source() == AggregationSource::External,
            "Binance historical bars require EXTERNAL aggregation"
        );
        anyhow::ensure!(
            bar_type.spec().price_type == PriceType::Last,
            "Binance historical bars require LAST price type"
        );
        anyhow::ensure!(
            bar_type.spec().is_time_aggregated(),
            "Binance historical bars require time aggregation"
        );

        self.spawn_command(async move {
            let result = http.request_bars(bar_type, start, end, limit).await;

            match result.context("failed to request bars from Binance Futures") {
                Ok(bars) => {
                    let response = DataResponse::Bars(BarsResponse::new(
                        request_id,
                        client_id,
                        bar_type,
                        bars,
                        start_nanos,
                        end_nanos,
                        clock.get_time_ns(),
                        params,
                    ));

                    if let Err(e) = sender.send(DataEvent::Response(response)) {
                        log::error!("Failed to send bars response: {e}");
                    }
                }
                Err(e) => log::error!("Bar request failed: {e:?}"),
            }
        });

        Ok(())
    }

    fn request_book_snapshot(&self, request: RequestBookSnapshot) -> anyhow::Result<()> {
        let depth = request.depth.map_or(1000, |value| value.get() as u32);
        anyhow::ensure!(
            BINANCE_BOOK_DEPTHS.contains(&depth),
            "invalid Binance Futures order-book depth {depth}; valid values are {BINANCE_BOOK_DEPTHS:?}"
        );
        let http = self.http_client.clone();
        let book_sync = self.book_sync.clone();
        let sender = self.data_sender.clone();
        let instrument_id = request.single_instrument_id()?;
        let request_id = request.request_id;
        let client_id = request.client_id.unwrap_or(self.client_id);
        let params = request.params;
        let clock = self.clock;

        self.spawn_command(async move {
            book_sync.pacer().acquire(depth_request_weight(depth)).await;

            match http.request_book_snapshot(instrument_id, Some(depth)).await {
                Ok(book) => {
                    let response = DataResponse::Book(BookResponse::new(
                        request_id,
                        client_id,
                        instrument_id,
                        book,
                        None,
                        None,
                        clock.get_time_ns(),
                        params,
                    ));

                    if let Err(e) = sender.send(DataEvent::Response(response)) {
                        log::error!("Failed to send book snapshot response: {e}");
                    }
                }
                Err(e) => log::error!("Book snapshot request failed for {instrument_id}: {e:?}"),
            }
        });
        Ok(())
    }
}

impl BinanceFuturesDataClient {
    fn subscribe_top_of_book(&self, instrument_id: InstrumentId) {
        let should_subscribe = {
            let previous = self
                .quote_refs
                .load()
                .get(&instrument_id)
                .copied()
                .unwrap_or(0);
            self.quote_refs
                .rcu(|refs| *refs.entry(instrument_id).or_insert(0) += 1);
            previous == 0
        };

        if should_subscribe {
            let ws = self.ws_public_client.clone();
            let stream = format!(
                "{}@bookTicker",
                format_binance_stream_symbol(&instrument_id)
            );
            self.spawn_ws(
                async move {
                    ws.subscribe(vec![stream])
                        .await
                        .context("top-of-book subscription")
                },
                "top-of-book subscription",
            );
        }
    }

    fn unsubscribe_top_of_book(&self, instrument_id: InstrumentId) {
        let should_unsubscribe = match self.quote_refs.load().get(&instrument_id).copied() {
            Some(1) => {
                self.quote_refs.remove(&instrument_id);
                true
            }
            Some(count) if count > 1 => {
                self.quote_refs.rcu(|refs| {
                    if let Some(existing) = refs.get_mut(&instrument_id) {
                        *existing -= 1;
                    }
                });
                false
            }
            _ => false,
        };

        if should_unsubscribe {
            let ws = self.ws_public_client.clone();
            let stream = format!(
                "{}@bookTicker",
                format_binance_stream_symbol(&instrument_id)
            );
            self.spawn_ws(
                async move {
                    ws.unsubscribe(vec![stream])
                        .await
                        .context("top-of-book unsubscribe")
                        .map(|_| ())
                },
                "top-of-book unsubscribe",
            );
        }
    }
}

fn is_partial_book_depth(depth: u32) -> bool {
    matches!(depth, 5 | 10 | 20)
}

fn book_stream(instrument_id: &InstrumentId, depth: u32) -> String {
    let symbol = format_binance_stream_symbol(instrument_id);
    if is_partial_book_depth(depth) {
        format!("{symbol}@depth{depth}@100ms")
    } else {
        format!("{symbol}@depth@0ms")
    }
}

#[derive(Clone, Debug)]
struct BookDrain {
    generation: u64,
    stream: String,
    // A same-stream resubscribe satisfies the drain: its frames carry identical
    // semantics, so the gate no longer needs to hold for it
    satisfied: bool,
}

fn arm_book_drain(
    map: &AtomicMap<InstrumentId, Vec<BookDrain>>,
    instrument_id: InstrumentId,
    generation: u64,
    stream: &str,
) {
    map.rcu(|m| {
        m.entry(instrument_id).or_default().push(BookDrain {
            generation,
            stream: stream.to_owned(),
            satisfied: false,
        });
    });
}

fn satisfy_book_drain(
    map: &AtomicMap<InstrumentId, Vec<BookDrain>>,
    instrument_id: InstrumentId,
    stream: &str,
) {
    map.rcu(|m| {
        if let Some(drains) = m.get_mut(&instrument_id) {
            for drain in drains.iter_mut().filter(|drain| drain.stream == stream) {
                drain.satisfied = true;
            }
        }
    });
}

// Drains are correlated by generation: a confirmation or reconnect abandonment only
// resolves the exact unsubscribe lifecycle it belongs to
fn remove_book_drain(map: &AtomicMap<InstrumentId, Vec<BookDrain>>, generation: u64) {
    map.rcu(|m| {
        for drains in m.values_mut() {
            drains.retain(|drain| drain.generation != generation);
        }
        m.retain(|_, drains| !drains.is_empty());
    });
}

fn book_drain_active(
    map: &AtomicMap<InstrumentId, Vec<BookDrain>>,
    instrument_id: InstrumentId,
) -> bool {
    map.load()
        .get(&instrument_id)
        .is_some_and(|drains| drains.iter().any(|drain| !drain.satisfied))
}

// A satisfied drain only releases frames whose semantics match; when the subscription
// changes to a different stream, an unresolved old copy may still be live elsewhere,
// so its protection must hold again
fn revive_book_drains(
    map: &AtomicMap<InstrumentId, Vec<BookDrain>>,
    instrument_id: InstrumentId,
    stream: &str,
) {
    map.rcu(|m| {
        if let Some(drains) = m.get_mut(&instrument_id) {
            for drain in drains.iter_mut().filter(|drain| drain.stream != stream) {
                drain.satisfied = false;
            }
        }
    });
}

fn subscribe_ticker(client: &BinanceFuturesDataClient, data_type: &DataType) -> anyhow::Result<()> {
    let instrument_id = BinanceFuturesDataClient::required_instrument_id_metadata(data_type)?;
    if instrument_id.venue != client.venue() {
        anyhow::bail!(
            "Binance Futures ticker custom data requires BINANCE venue instrument, received {instrument_id}"
        );
    }

    let should_subscribe = {
        let prev = client
            .ticker_refs
            .load()
            .get(&instrument_id)
            .copied()
            .unwrap_or(0);
        client.ticker_refs.rcu(|m| {
            let count = m.entry(instrument_id).or_insert(0);
            *count += 1;
        });
        prev == 0
    };

    if should_subscribe {
        let ws = client.ws_client.clone();
        let stream = ticker_stream(&instrument_id);
        client.spawn_ws(
            async move {
                ws.subscribe(vec![stream])
                    .await
                    .context("ticker subscription")
            },
            "ticker subscription",
        );
    }

    Ok(())
}

fn unsubscribe_ticker(
    client: &BinanceFuturesDataClient,
    data_type: &DataType,
) -> anyhow::Result<()> {
    let instrument_id = BinanceFuturesDataClient::required_instrument_id_metadata(data_type)?;
    if instrument_id.venue != client.venue() {
        anyhow::bail!(
            "Binance Futures ticker custom data requires BINANCE venue instrument, received {instrument_id}"
        );
    }

    let should_unsubscribe = {
        let prev = client.ticker_refs.load().get(&instrument_id).copied();
        match prev {
            Some(count) if count <= 1 => {
                client.ticker_refs.remove(&instrument_id);
                true
            }
            Some(_) => {
                client.ticker_refs.rcu(|m| {
                    if let Some(count) = m.get_mut(&instrument_id) {
                        *count = count.saturating_sub(1);
                    }
                });
                false
            }
            None => false,
        }
    };

    if should_unsubscribe {
        let ws = client.ws_client.clone();
        let stream = ticker_stream(&instrument_id);
        client.spawn_ws(
            async move {
                ws.unsubscribe(vec![stream])
                    .await
                    .context("ticker unsubscribe")
                    .map(|_| ())
            },
            "ticker unsubscribe",
        );
    }

    Ok(())
}

fn ticker_data_type(instrument_id: InstrumentId) -> DataType {
    let metadata = Params::new(
        [(
            "instrument_id".to_string(),
            serde_json::Value::String(instrument_id.to_string()),
        )]
        .into(),
    );
    DataType::new(
        "BinanceFuturesTicker",
        Some(metadata),
        Some(instrument_id.to_string()),
    )
}

fn mark_price_data_type(instrument_id: InstrumentId) -> DataType {
    let metadata = Params::new(
        [(
            "instrument_id".to_string(),
            serde_json::Value::String(instrument_id.to_string()),
        )]
        .into(),
    );
    DataType::new(
        "BinanceFuturesMarkPriceUpdate",
        Some(metadata),
        Some(instrument_id.to_string()),
    )
}

fn ticker_stream(instrument_id: &InstrumentId) -> String {
    format!("{}@ticker", format_binance_stream_symbol(instrument_id))
}

#[cfg(test)]
mod tests {
    use rstest::rstest;
    use rust_decimal_macros::dec;

    use super::*;

    #[rstest]
    fn test_parse_order_book_snapshot_skips_invalid_levels() {
        let instrument_id = InstrumentId::from("BTCUSDT-PERP.BINANCE");
        let order_book = BinanceOrderBook {
            last_update_id: 10,
            bids: vec![
                ("not-a-price".to_string(), "1.0".to_string()),
                ("100.00".to_string(), "0.5".to_string()),
            ],
            asks: vec![
                ("101.00".to_string(), "not-a-quantity".to_string()),
                ("102.00".to_string(), "0.7".to_string()),
            ],
            event_time: None,
            transaction_time: None,
        };

        let deltas =
            parse_order_book_snapshot(&order_book, instrument_id, 2, 3, UnixNanos::from(1));

        assert_eq!(deltas.deltas.len(), 3);
        assert_eq!(deltas.deltas[1].order.side, OrderSide::Buy.into());
        assert_eq!(deltas.deltas[1].order.price.as_decimal(), dec!(100.00));
        assert_eq!(deltas.deltas[1].order.size.as_decimal(), dec!(0.500));
        assert_eq!(deltas.deltas[2].order.side, OrderSide::Sell.into());
        assert_eq!(deltas.deltas[2].order.price.as_decimal(), dec!(102.00));
        assert_eq!(deltas.deltas[2].order.size.as_decimal(), dec!(0.700));
        assert_eq!(deltas.deltas[1].flags, RecordFlag::F_SNAPSHOT as u8);
        assert_eq!(
            deltas.deltas[2].flags,
            RecordFlag::F_SNAPSHOT as u8 | RecordFlag::F_LAST as u8
        );
        assert_eq!(deltas.ts_event, UnixNanos::from(1));
        assert_eq!(deltas.ts_init, UnixNanos::from(1));
    }

    #[rstest]
    #[case::negative(-1)]
    #[case::overflow(i64::MAX)]
    fn test_parse_order_book_snapshot_falls_back_for_invalid_timestamp(
        #[case] transaction_time: i64,
    ) {
        let order_book = BinanceOrderBook {
            last_update_id: 10,
            bids: vec![],
            asks: vec![],
            event_time: None,
            transaction_time: Some(transaction_time),
        };

        let ts_init = UnixNanos::from(1);
        let deltas = parse_order_book_snapshot(
            &order_book,
            InstrumentId::from("BTCUSDT-PERP.BINANCE"),
            2,
            3,
            ts_init,
        );

        assert_eq!(deltas.ts_event, ts_init);
        assert_eq!(deltas.ts_init, ts_init);
    }

    #[rstest]
    fn test_parse_order_book_snapshot_all_invalid_levels_marks_clear_last() {
        let instrument_id = InstrumentId::from("BTCUSDT-PERP.BINANCE");
        let order_book = BinanceOrderBook {
            last_update_id: 10,
            bids: vec![("not-a-price".to_string(), "1.0".to_string())],
            asks: vec![("101.00".to_string(), "not-a-quantity".to_string())],
            event_time: None,
            transaction_time: None,
        };

        let deltas =
            parse_order_book_snapshot(&order_book, instrument_id, 2, 3, UnixNanos::from(1));

        assert_eq!(deltas.deltas.len(), 1);
        assert_eq!(deltas.deltas[0].action, BookAction::Clear);
        assert_eq!(
            deltas.deltas[0].flags,
            RecordFlag::F_SNAPSHOT as u8 | RecordFlag::F_LAST as u8
        );
    }

    #[rstest]
    fn test_book_drain_gate_follows_unsatisfied_entries() {
        let map = AtomicMap::new();
        let instrument_id = InstrumentId::from("BTCUSDT-PERP.BINANCE");

        assert!(!book_drain_active(&map, instrument_id));
        arm_book_drain(&map, instrument_id, 1, "btcusdt@depth20@100ms");
        assert!(book_drain_active(&map, instrument_id));

        satisfy_book_drain(&map, instrument_id, "btcusdt@depth20@100ms");
        assert!(!book_drain_active(&map, instrument_id));

        arm_book_drain(&map, instrument_id, 2, "btcusdt@depth10@100ms");
        assert!(book_drain_active(&map, instrument_id));
    }

    #[rstest]
    fn test_remove_book_drain_resolves_only_matching_generation() {
        let map = AtomicMap::new();
        let instrument_id = InstrumentId::from("BTCUSDT-PERP.BINANCE");

        // A satisfied tombstone never produces a confirmation, so a late one for
        // the older lifecycle must not strand the newer drain
        arm_book_drain(&map, instrument_id, 1, "btcusdt@depth20@100ms");
        satisfy_book_drain(&map, instrument_id, "btcusdt@depth20@100ms");
        arm_book_drain(&map, instrument_id, 2, "btcusdt@depth20@100ms");

        remove_book_drain(&map, 1);
        assert!(book_drain_active(&map, instrument_id));

        remove_book_drain(&map, 2);
        assert!(!book_drain_active(&map, instrument_id));
        assert!(!map.load().contains_key(&instrument_id));
    }

    #[rstest]
    fn test_semantics_change_revives_satisfied_drains() {
        let map = AtomicMap::new();
        let instrument_id = InstrumentId::from("BTCUSDT-PERP.BINANCE");

        arm_book_drain(&map, instrument_id, 1, "btcusdt@depth20@100ms");
        satisfy_book_drain(&map, instrument_id, "btcusdt@depth20@100ms");
        assert!(!book_drain_active(&map, instrument_id));

        // A depth change revives the drain: its stream may still be live and its
        // frames no longer match the subscription's semantics
        revive_book_drains(&map, instrument_id, "btcusdt@depth5@100ms");
        assert!(book_drain_active(&map, instrument_id));

        // A same-stream resubscribe does not revive it
        satisfy_book_drain(&map, instrument_id, "btcusdt@depth20@100ms");
        revive_book_drains(&map, instrument_id, "btcusdt@depth20@100ms");
        assert!(!book_drain_active(&map, instrument_id));
    }

    #[rstest]
    fn test_remove_book_drain_leaves_other_generations_pending() {
        let map = AtomicMap::new();
        let instrument_id = InstrumentId::from("BTCUSDT-PERP.BINANCE");

        arm_book_drain(&map, instrument_id, 1, "btcusdt@depth20@100ms");
        arm_book_drain(&map, instrument_id, 2, "btcusdt@depth10@100ms");

        // A confirmation for the second drain leaves the failed first one pending
        remove_book_drain(&map, 2);
        assert!(book_drain_active(&map, instrument_id));

        remove_book_drain(&map, 1);
        assert!(!book_drain_active(&map, instrument_id));
    }

    #[rstest]
    #[case::smallest(5, 2)]
    #[case::tier_one_max(50, 2)]
    #[case::tier_two_min(51, 5)]
    #[case::tier_two_max(100, 5)]
    #[case::tier_three_min(101, 10)]
    #[case::tier_three_max(500, 10)]
    #[case::tier_four_min(501, 20)]
    #[case::largest(1000, 20)]
    fn test_depth_request_weight_follows_venue_tiers(#[case] limit: u32, #[case] expected: u32) {
        assert_eq!(depth_request_weight(limit), expected);
    }

    #[rstest]
    #[case::network(BinanceFuturesHttpError::NetworkError("connection reset".to_string()), true)]
    #[case::server_error(
        BinanceFuturesHttpError::UnexpectedStatus {
            status: 503,
            body: String::new(),
            retry_after: None,
        },
        true
    )]
    #[case::rate_limited(
        BinanceFuturesHttpError::BinanceError {
            code: -1003,
            message: "Too many requests".to_string(),
            status: 429,
            retry_after: None,
        },
        true
    )]
    #[case::retry_budget_exceeded(
        BinanceFuturesHttpError::RetryBudgetExceeded("elapsed budget exhausted".to_string()),
        true
    )]
    #[case::invalid_symbol(
        BinanceFuturesHttpError::BinanceError {
            code: -1121,
            message: "Invalid symbol.".to_string(),
            status: 400,
            retry_after: None,
        },
        false
    )]
    fn test_depth_snapshot_error_retries_transient_failures(
        #[case] error: BinanceFuturesHttpError,
        #[case] retryable: bool,
    ) {
        let instrument_id = InstrumentId::from("BTCUSDT-PERP.BINANCE");
        let message = format!("depth snapshot request for {instrument_id} failed: {error}");

        let classified = depth_snapshot_error(instrument_id, &error);

        let expected = if retryable {
            BinanceBookError::Retryable(message)
        } else {
            BinanceBookError::Permanent(message)
        };

        assert_eq!(classified, expected);
    }
}