nautilus_binance/spot/

execution.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 execution client implementation for the Binance Spot adapter.

use std::{
    future::Future,
    sync::{
        Arc,
        atomic::{AtomicBool, Ordering},
    },
    time::Duration,
};

use ahash::{AHashMap, AHashSet};
use anyhow::Context;
use async_trait::async_trait;
use jiff::Timestamp;
use nautilus_common::{
    cache::fifo::FifoCache,
    clients::ExecutionClient,
    live::runner::get_exec_event_sender,
    messages::execution::{
        BatchCancelOrders, CancelAllOrders, CancelOrder, GenerateFillReports,
        GenerateFillReportsBuilder, GenerateOrderStatusReport, GenerateOrderStatusReports,
        GenerateOrderStatusReportsBuilder, GeneratePositionStatusReports,
        GeneratePositionStatusReportsBuilder, ModifyOrder, QueryAccount, QueryOrder, SubmitOrder,
        SubmitOrderList,
    },
};
use nautilus_core::{
    Params, UUID4, UnixNanos,
    datetime::{NANOSECONDS_IN_MILLISECOND, checked_mins_to_nanos},
    time::{AtomicTime, get_atomic_clock_realtime},
};
use nautilus_live::{
    ExecutionClientCore, ExecutionEventEmitter, SocketControlFactory,
    task::{TaskGroup, TaskGroupGuard, TaskSpawner},
};
use nautilus_model::{
    accounts::AccountAny,
    enums::{ContingencyType, LiquiditySide, OmsType, OrderStatus, OrderType, TimeInForce},
    events::{
        AccountState, OrderAccepted, OrderCancelRejected, OrderCanceled, OrderEventAny,
        OrderExpired, OrderFilled, OrderModifyRejected, OrderRejected, OrderUpdated,
    },
    identifiers::{
        AccountId, ClientId, ClientOrderId, InstrumentId, StrategyId, TradeId, TraderId, Venue,
        VenueOrderId,
    },
    instruments::Instrument,
    orders::{Order, OrderAny},
    reports::{ExecutionMassStatus, FillReport, OrderStatusReport, PositionStatusReport},
    types::{AccountBalance, Currency, MarginBalance, Money, Price, Quantity},
};
use parking_lot::Mutex;
use rust_decimal::Decimal;
use ustr::Ustr;

use super::websocket::trading::{
    client::BinanceSpotWsTradingClient,
    messages::BinanceSpotWsTradingMessage,
    parse::{
        parse_spot_account_position, parse_spot_exec_report_to_fill,
        parse_spot_exec_report_to_order_status,
    },
    user_data::{BinanceSpotExecutionReport, BinanceSpotExecutionType},
};
use crate::{
    common::{
        consts::{
            BINANCE_GTX_ORDER_REJECT_CODE, BINANCE_NAUTILUS_SPOT_BROKER_ID,
            BINANCE_NEW_ORDER_REJECTED_CODE, BINANCE_SPOT_POST_ONLY_REJECT_MSG,
            BINANCE_SPOT_SBE_WS_API_DEMO_URL, BINANCE_SPOT_SBE_WS_API_TESTNET_URL,
            BINANCE_SPOT_SBE_WS_API_URL, BINANCE_STATUS_UNKNOWN_CODE,
            BINANCE_UNEXPECTED_RESPONSE_CODE, BINANCE_VENUE, BINANCE_WS_HEARTBEAT_SECS,
        },
        credential::resolve_credentials,
        dispatch::{
            OrderIdentity, PendingOperation, PendingRequest, WsDispatchState,
            ensure_accepted_emitted,
        },
        encoder::{decode_client_order_id, encode_broker_id},
        enums::{BinanceEnvironment, BinanceSide, BinanceTimeInForce},
        parse::{
            parse_millis_or_init, parse_required_decimal, parse_required_price_at_precision,
            parse_required_quantity_at_precision,
        },
        urls::{get_http_base_url_with_us, get_spot_user_stream_url},
    },
    config::BinanceExecutionClientConfig,
    spot::{
        enums::{
            BinanceCancelReplaceMode, BinanceOrderResponseType, BinanceSpotOrderType,
            order_type_to_binance_spot, time_in_force_to_binance_spot,
        },
        http::{
            client::BinanceSpotHttpClient,
            error::BinanceSpotHttpError,
            models::BatchCancelResult,
            query::{
                BatchCancelItem, CancelOrderParams, CancelReplaceOrderParams,
                NewOcoOrderListParams, NewOrderParams,
            },
        },
    },
};

const ACCOUNT_TRADES_MAX_INTERVAL_MS: i64 = 24 * 60 * 60 * 1_000;

const ACCOUNT_TRADES_PAGE_LIMIT: u32 = 1_000;

const WS_RECONNECT_SETUP_RETRY_DELAY: Duration = Duration::from_secs(1);

/// Live execution client for Binance Spot trading.
///
/// Implements the [`ExecutionClient`] trait for order management on Binance Spot
/// and Spot Margin markets. Uses WebSocket or HTTP for order commands. The WebSocket
/// User Data Stream is required for successful execution lifecycle events.
#[derive(Debug)]
pub struct BinanceSpotExecutionClient {
    core: ExecutionClientCore,
    clock: &'static AtomicTime,
    config: BinanceExecutionClientConfig,
    emitter: ExecutionEventEmitter,
    dispatch_state: Arc<WsDispatchState>,
    http_client: BinanceSpotHttpClient,
    socket_factory: SocketControlFactory,
    ws_trading_client: Option<BinanceSpotWsTradingClient>,
    ws_trading_dispatch_active: Arc<AtomicBool>,
    ws_user_data_client: Arc<Mutex<Option<BinanceSpotWsTradingClient>>>,
    ws_user_data_dispatch_active: Arc<AtomicBool>,
    listen_key: Option<String>,
    us_credentials: Option<(String, String)>,
    ws_authenticated: Arc<tokio::sync::Notify>,
    ws_user_data_subscribed: Arc<tokio::sync::Notify>,
    session_tasks: TaskGroup,
    pending_tasks: TaskGroup,
    shutdown_errors: Vec<String>,
}

impl BinanceSpotExecutionClient {
    /// Creates a new [`BinanceSpotExecutionClient`].
    ///
    /// # Errors
    ///
    /// Returns an error if the HTTP client fails to initialize or credentials are missing.
    pub fn new(
        core: ExecutionClientCore,
        config: BinanceExecutionClientConfig,
    ) -> anyhow::Result<Self> {
        config.validate()?;
        let (api_key, api_secret) = resolve_credentials(
            config.api_key.clone(),
            config.api_secret.clone(),
            config.environment,
            config.product_type,
        )?;

        let clock = get_atomic_clock_realtime();
        let socket_factory = SocketControlFactory::new(core.client_id, Some(*BINANCE_VENUE));
        let base_url_http = config.base_url_http.clone().or_else(|| {
            config.us.then(|| {
                get_http_base_url_with_us(config.product_type, config.environment, true).to_string()
            })
        });

        let http_client = BinanceSpotHttpClient::new_with_json_responses(
            config.environment,
            clock,
            Some(api_key.clone()),
            Some(api_secret.clone()),
            base_url_http,
            Some(config.recv_window_ms),
            None, // timeout_secs
            config.proxy_url.clone(),
            config.us,
        )
        .context("failed to construct Binance Spot HTTP client")?;
        let emitter = ExecutionEventEmitter::new(
            clock,
            core.trader_id,
            core.account_id,
            core.account_type,
            core.base_currency,
        );

        let ws_trading_client = if config.us {
            None
        } else {
            let url = Some(Self::resolve_ws_trading_url(
                config.base_url_ws_trading.clone(),
                config.environment,
            ));
            Some(
                BinanceSpotWsTradingClient::new(
                    url,
                    api_key.clone(),
                    api_secret.clone(),
                    Some(BINANCE_WS_HEARTBEAT_SECS),
                    config.transport_backend,
                )
                .with_proxy(config.proxy_url.clone())
                .with_recv_window(Some(config.recv_window_ms))
                .with_socket_control(socket_factory.control("binance-spot-trading")),
            )
        };
        let us_credentials = config.us.then_some((api_key, api_secret));

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

        Ok(Self {
            core,
            clock,
            config,
            emitter,
            dispatch_state: Arc::new(WsDispatchState::default()),
            http_client,
            socket_factory,
            ws_trading_client,
            ws_trading_dispatch_active: Arc::new(AtomicBool::new(false)),
            ws_user_data_client: Arc::new(Mutex::new(None)),
            ws_user_data_dispatch_active: Arc::new(AtomicBool::new(false)),
            listen_key: None,
            us_credentials,
            ws_authenticated: Arc::new(tokio::sync::Notify::new()),
            ws_user_data_subscribed: Arc::new(tokio::sync::Notify::new()),
            session_tasks,
            pending_tasks,
            shutdown_errors: Vec::new(),
        })
    }

    fn resolve_ws_trading_url(base_url: Option<String>, environment: BinanceEnvironment) -> String {
        base_url.unwrap_or_else(|| {
            match environment {
                BinanceEnvironment::Live => BINANCE_SPOT_SBE_WS_API_URL,
                BinanceEnvironment::Testnet => BINANCE_SPOT_SBE_WS_API_TESTNET_URL,
                BinanceEnvironment::Demo => BINANCE_SPOT_SBE_WS_API_DEMO_URL,
            }
            .to_string()
        })
    }

    async fn refresh_account_state(&self) -> anyhow::Result<AccountState> {
        self.http_client
            .request_account_state(self.core.account_id)
            .await
    }

    fn update_account_state(&self) {
        let http_client = self.http_client.clone();
        let account_id = self.core.account_id;
        let emitter = self.emitter.clone();
        let clock = self.clock;

        self.spawn_task("query_account", async move {
            let account_state = http_client.request_account_state(account_id).await?;
            let ts_now = clock.get_time_ns();
            emitter.emit_account_state(
                account_state.balances.clone(),
                account_state.margins.clone(),
                account_state.is_reported,
                ts_now,
                account_state.info,
            );
            Ok(())
        });
    }

    fn ws_user_data_active(&self) -> bool {
        let dispatch_running = if self.config.us {
            self.ws_user_data_dispatch_active.load(Ordering::Acquire)
        } else {
            self.ws_trading_dispatch_active.load(Ordering::Acquire)
        };
        let user_data_active = if self.config.us {
            self.ws_user_data_client
                .lock()
                .as_ref()
                .is_some_and(BinanceSpotWsTradingClient::is_user_data_active)
        } else {
            self.ws_trading_client
                .as_ref()
                .is_some_and(BinanceSpotWsTradingClient::is_user_data_active)
        };

        user_data_active && dispatch_running
    }

    fn ensure_ws_user_data_active(&self) -> anyhow::Result<()> {
        anyhow::ensure!(
            self.ws_user_data_active(),
            "Binance Spot user data stream is not active",
        );
        Ok(())
    }

    fn ws_order_transport_active(&self) -> bool {
        self.config.use_ws_trading && self.ws_trading_client.is_some() && self.ws_user_data_active()
    }

    fn submit_order_internal(&self, cmd: &SubmitOrder) -> anyhow::Result<()> {
        let order = self.core.cache().try_order_owned(&cmd.client_order_id)?;

        let event_emitter = self.emitter.clone();
        let trader_id = self.core.trader_id;
        let account_id = self.core.account_id;
        let client_order_id = order.client_order_id();
        let strategy_id = order.strategy_id();
        let instrument_id = order.instrument_id();
        let order_side = order.order_side();
        let order_type = order.order_type();
        let quantity = order.quantity();
        let time_in_force = order.time_in_force();
        let price = order.price();
        let trigger_price = order.trigger_price();
        let is_post_only = order.is_post_only();
        let is_quote_quantity = order.is_quote_quantity();
        let display_qty = order.display_qty();
        let use_gtd = self.config.use_gtd;
        let clock = self.clock;
        let ts_init = self.clock.get_time_ns();

        // Register identity for tracked/external dispatch routing
        self.dispatch_state.order_identities.insert(
            client_order_id,
            OrderIdentity {
                instrument_id,
                strategy_id,
                order_side,
                order_type,
                price,
                quantity,
                venue_position_id: None,
            },
        );

        if self.ws_order_transport_active() {
            let ws_client = self.ws_trading_client.as_ref().unwrap().clone();
            let dispatch_state = self.dispatch_state.clone();
            let params = build_new_order_params(
                &order,
                client_order_id,
                is_post_only,
                is_quote_quantity,
                use_gtd,
            )?;

            // Pre-register before sending to avoid response racing the insert
            let request_id = ws_client.next_request_id();
            dispatch_state.pending_requests.insert(
                request_id.clone(),
                PendingRequest {
                    client_order_id,
                    venue_order_id: None,
                    operation: PendingOperation::Place,
                },
            );

            self.spawn_task("submit_order_ws", async move {
                if let Err(e) = ws_client
                    .place_order_with_id(request_id.clone(), params)
                    .await
                {
                    dispatch_state.pending_requests.remove(&request_id);
                    log::warn!(
                        "WS submit request failed for {client_order_id}, awaiting reconciliation: {e}"
                    );
                    anyhow::bail!("WS submit order failed: {e}");
                }
                Ok(())
            });
        } else {
            let http_client = self.http_client.clone();
            let dispatch_state = self.dispatch_state.clone();
            log::debug!("WS trading not active, falling back to HTTP for submit_order");

            self.spawn_task("submit_order_http", async move {
                let result = http_client
                    .submit_order(
                        account_id,
                        instrument_id,
                        client_order_id,
                        order_side,
                        order_type,
                        quantity,
                        time_in_force,
                        price,
                        trigger_price,
                        is_post_only,
                        is_quote_quantity,
                        display_qty,
                        use_gtd,
                    )
                    .await;

                match result {
                    Ok(report) => handle_spot_order_submit_success(
                        client_order_id,
                        report.venue_order_id,
                    ),
                    Err(e) => {
                        if is_ambiguous_submit_error(&e) {
                            log::warn!(
                                "Ambiguous submit failure for {client_order_id}, awaiting reconciliation: {e}"
                            );
                        } else if is_structured_venue_rejection(&e)
                            || is_local_command_failure(&e)
                        {
                            let due_post_only = e
                                .downcast_ref::<BinanceSpotHttpError>()
                                .is_some_and(is_spot_post_only_rejection);
                            dispatch_state.cleanup_terminal(client_order_id);
                            let rejected = OrderRejected::new(
                                trader_id,
                                strategy_id,
                                instrument_id,
                                client_order_id,
                                account_id,
                                format!("submit-order-error: {e}").into(),
                                UUID4::new(),
                                ts_init,
                                clock.get_time_ns(),
                                false,
                                due_post_only,
                            );
                            event_emitter.send_order_event(OrderEventAny::Rejected(rejected));
                        } else {
                            log::warn!(
                                "Ambiguous submit failure for {client_order_id}, awaiting reconciliation: {e}"
                            );
                        }
                        return Err(e);
                    }
                }
                Ok(())
            });
        }

        Ok(())
    }

    fn cancel_order_internal(&self, cmd: &CancelOrder) {
        let event_emitter = self.emitter.clone();
        let trader_id = self.core.trader_id;
        let account_id = self.core.account_id;
        let clock = self.clock;
        let command = cmd.clone();

        if self.ws_order_transport_active() {
            let ws_client = self.ws_trading_client.as_ref().unwrap().clone();
            let dispatch_state = self.dispatch_state.clone();
            let params = build_cancel_order_params(&command);

            // Pre-register before sending to avoid response racing the insert
            let request_id = ws_client.next_request_id();
            dispatch_state.pending_requests.insert(
                request_id.clone(),
                PendingRequest {
                    client_order_id: command.client_order_id,
                    venue_order_id: command.venue_order_id,
                    operation: PendingOperation::Cancel,
                },
            );

            self.spawn_task("cancel_order_ws", async move {
                if let Err(e) = ws_client
                    .cancel_order_with_id(request_id.clone(), params)
                    .await
                {
                    dispatch_state.pending_requests.remove(&request_id);
                    log::warn!(
                        "WS cancel request failed for {}, awaiting reconciliation: {e}",
                        command.client_order_id
                    );
                    anyhow::bail!("WS cancel order failed: {e}");
                }
                Ok(())
            });
        } else {
            let http_client = self.http_client.clone();
            let dispatch_state = self.dispatch_state.clone();
            log::debug!("WS trading not active, falling back to HTTP for cancel_order");

            self.spawn_task("cancel_order_http", async move {
                let result = http_client
                    .cancel_order(
                        command.instrument_id,
                        command.venue_order_id,
                        Some(command.client_order_id),
                    )
                    .await;

                match result {
                    Ok(venue_order_id) => {
                        dispatch_state.cleanup_terminal(command.client_order_id);
                        let ts_now = clock.get_time_ns();
                        let canceled_event = OrderCanceled::new(
                            trader_id,
                            command.strategy_id,
                            command.instrument_id,
                            command.client_order_id,
                            UUID4::new(),
                            ts_now,
                            ts_now,
                            false,
                            Some(venue_order_id),
                            Some(account_id),
                        );
                        event_emitter.send_order_event(OrderEventAny::Canceled(canceled_event));
                    }
                    Err(e) => {
                        if is_structured_venue_rejection(&e) {
                            let ts_now = clock.get_time_ns();
                            let rejected_event = OrderCancelRejected::new(
                                trader_id,
                                command.strategy_id,
                                command.instrument_id,
                                command.client_order_id,
                                format!("cancel-order-error: {e}").into(),
                                UUID4::new(),
                                ts_now,
                                ts_now,
                                false,
                                command.venue_order_id,
                                Some(account_id),
                            );
                            event_emitter
                                .send_order_event(OrderEventAny::CancelRejected(rejected_event));
                        } else if is_local_command_failure(&e) {
                            log::warn!(
                                "Cancel command failed local validation for {}: {e}",
                                command.client_order_id
                            );
                        } else {
                            log::warn!(
                                "Ambiguous cancel failure for {}, awaiting reconciliation: {e}",
                                command.client_order_id
                            );
                        }
                        return Err(e);
                    }
                }
                Ok(())
            });
        }
    }

    fn spawn_task<F>(&self, description: &'static str, fut: F)
    where
        F: Future<Output = anyhow::Result<()>> + Send + 'static,
    {
        crate::common::execution::spawn_task(&self.pending_tasks, description, fut);
    }

    fn abort_pending_tasks(&self) {
        crate::common::execution::abort_pending_tasks(&self.pending_tasks);
    }

    fn abort_session_tasks(&self) {
        self.session_tasks.begin_shutdown();
    }

    async fn await_pending_tasks(&self) -> anyhow::Result<()> {
        crate::common::execution::await_pending_tasks(&self.pending_tasks).await
    }

    async fn await_session_tasks(&self) -> anyhow::Result<()> {
        self.session_tasks.begin_shutdown();
        self.session_tasks
            .finish_shutdown(Duration::from_secs(1), Duration::from_secs(2))
            .await
            .map_err(|e| anyhow::anyhow!("Failed to terminate Binance Spot session tasks: {e}"))?;
        Ok(())
    }

    async fn ws_setup_failure(
        &mut self,
        mut ws_trading: BinanceSpotWsTradingClient,
        reason: String,
    ) -> anyhow::Error {
        ws_trading.mark_user_data_inactive();
        log::error!("{reason}; Binance Spot private user data is required for execution");

        self.abort_session_tasks();

        if let Err(e) = ws_trading.disconnect().await {
            log::warn!("Failed to stop Binance Spot trading WebSocket after setup failure: {e}");
        }
        self.disconnect_us_user_data().await;

        if let Err(e) = self.await_session_tasks().await {
            log::warn!("Failed to drain Binance Spot session tasks after setup failure: {e}");
        }
        self.ws_trading_client = Some(ws_trading);
        anyhow::anyhow!(reason)
    }

    async fn connect_us_user_data(&mut self) -> anyhow::Result<()> {
        let (api_key, api_secret) = self
            .us_credentials
            .clone()
            .context("Binance US user data credentials are unavailable")?;
        let listen_key = self
            .http_client
            .inner()
            .create_listen_key()
            .await
            .context("failed to create Binance US listen key")?
            .listen_key;
        self.listen_key = Some(listen_key.clone());
        let url = get_spot_user_stream_url(self.config.base_url_ws.as_deref(), &listen_key);
        let mut ws_user_data = BinanceSpotWsTradingClient::new(
            Some(url),
            api_key,
            api_secret,
            Some(BINANCE_WS_HEARTBEAT_SECS),
            self.config.transport_backend,
        )
        .with_proxy(self.config.proxy_url.clone())
        .with_socket_control(self.socket_factory.control("binance-spot-user-streams"));
        *self.ws_user_data_client.lock() = Some(ws_user_data.clone());
        ws_user_data
            .connect()
            .await
            .context("failed to connect Binance US user data stream")?;

        let ws_clone = ws_user_data.clone();
        let emitter = self.emitter.clone();
        let account_id = self.core.account_id;
        let clock = self.clock;
        let http_client = self.http_client.clone();
        let dispatch_state = self.dispatch_state.clone();
        let treat_expired_as_canceled = self.config.treat_expired_as_canceled;
        let ws_authenticated = self.ws_authenticated.clone();
        let ws_user_data_subscribed = self.ws_user_data_subscribed.clone();
        let (setup_error_tx, _setup_error_rx) = tokio::sync::mpsc::unbounded_channel();
        let seen_trade_ids = Arc::new(Mutex::new(FifoCache::new()));
        let dispatch_active = Arc::clone(&self.ws_user_data_dispatch_active);
        let task_spawner = self
            .session_tasks
            .spawner()
            .context("Binance Spot session task admission is closed")?;

        if let Err(e) = self.session_tasks.spawn(async move {
            dispatch_active.store(true, Ordering::Release);
            let _active = DispatchActiveGuard(dispatch_active);

            while let Some(message) = ws_clone.recv().await {
                if matches!(&message, BinanceSpotWsTradingMessage::Reconnected) {
                    ws_clone.mark_user_data_active();
                }
                dispatch_ws_trading_message(
                    message,
                    &emitter,
                    &http_client,
                    account_id,
                    treat_expired_as_canceled,
                    clock,
                    &dispatch_state,
                    &ws_authenticated,
                    &ws_user_data_subscribed,
                    &setup_error_tx,
                    &seen_trade_ids,
                    &task_spawner,
                );
            }
            log::warn!("Binance US user data dispatch loop ended");
        }) {
            return Err(e.into());
        }

        let keepalive_http = self.http_client.clone();
        let keepalive_key = listen_key.clone();

        if let Err(e) = self.session_tasks.spawn(async move {
            let mut interval = tokio::time::interval(Duration::from_secs(30 * 60));
            interval.tick().await;

            loop {
                interval.tick().await;

                if let Err(e) = keepalive_http
                    .inner()
                    .extend_listen_key(&keepalive_key)
                    .await
                {
                    log::warn!("Binance US listen key keepalive failed: {e}");
                }
            }
        }) {
            return Err(e.into());
        }

        ws_user_data.mark_user_data_active();
        *self.ws_user_data_client.lock() = Some(ws_user_data);
        Ok(())
    }

    async fn disconnect_us_user_data(&mut self) {
        let mut client_drained = true;
        let client = self.ws_user_data_client.lock().clone();

        if let Some(mut client) = client {
            client.mark_user_data_inactive();
            if let Err(e) = client.disconnect().await {
                client_drained = false;
                self.shutdown_errors.push(format!(
                    "failed to stop Binance US user data WebSocket: {e}"
                ));
            }
        }

        if client_drained {
            *self.ws_user_data_client.lock() = None;
        }

        if let Some(listen_key) = self.listen_key.clone() {
            match self.http_client.inner().close_listen_key(&listen_key).await {
                Ok(()) if self.listen_key.as_deref() == Some(listen_key.as_str()) => {
                    self.listen_key = None;
                }
                Ok(()) => {}
                Err(e) => self
                    .shutdown_errors
                    .push(format!("failed to close Binance US listen key: {e}")),
            }
        }
    }

    async fn teardown_partial_connect(&mut self) -> anyhow::Result<()> {
        if let Some(client) = self.ws_trading_client.as_ref() {
            client.mark_user_data_inactive();
            client.begin_shutdown();
        }

        if let Some(client) = self.ws_user_data_client.lock().as_ref() {
            client.mark_user_data_inactive();
            client.begin_shutdown();
        }

        self.abort_session_tasks();
        self.abort_pending_tasks();

        if let Some(ref mut ws_trading) = self.ws_trading_client
            && let Err(e) = ws_trading.disconnect().await
        {
            self.shutdown_errors
                .push(format!("trading WebSocket shutdown failed: {e}"));
        }
        self.disconnect_us_user_data().await;

        let (session_result, pending_result) =
            tokio::join!(self.await_session_tasks(), self.await_pending_tasks());
        self.core.set_disconnected();

        if let Err(e) = session_result {
            self.shutdown_errors.push(e.to_string());
        }

        if let Err(e) = pending_result {
            self.shutdown_errors.push(e.to_string());
        }

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

#[async_trait(?Send)]
impl ExecutionClient for BinanceSpotExecutionClient {
    fn is_connected(&self) -> bool {
        self.core.is_connected()
    }

    fn client_id(&self) -> ClientId {
        self.core.client_id
    }

    fn account_id(&self) -> AccountId {
        self.core.account_id
    }

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

    fn oms_type(&self) -> OmsType {
        self.core.oms_type
    }

    fn get_account(&self) -> Option<AccountAny> {
        self.core.cache().account_owned(&self.core.account_id)
    }

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

        if !self.pending_tasks.is_open() || !self.session_tasks.is_open() {
            self.teardown_partial_connect().await?;
        }

        if !self.pending_tasks.is_open() {
            self.await_pending_tasks().await?;
            self.pending_tasks.start_generation().map_err(|e| {
                anyhow::anyhow!("Failed to start Binance Spot task generation: {e}")
            })?;
        }

        if !self.session_tasks.is_open() {
            self.await_session_tasks().await?;
            self.session_tasks.start_generation().map_err(|e| {
                anyhow::anyhow!("Failed to start Binance Spot session generation: {e}")
            })?;
        }
        let ws_trading_client = self.ws_trading_client.clone();
        let ws_user_data_client = Arc::clone(&self.ws_user_data_client);
        let setup_guard =
            TaskGroupGuard::new(&[&self.session_tasks, &self.pending_tasks], move || {
                if let Some(client) = ws_trading_client {
                    client.begin_shutdown();
                }

                if let Some(client) = ws_user_data_client.lock().as_ref() {
                    client.begin_shutdown();
                }
            });

        let ws_setup_timeout = Duration::from_millis(self.config.ws_trading_setup_timeout_ms);

        // Load instruments if not already done
        if !self.core.instruments_initialized() {
            let instruments = self
                .http_client
                .request_instruments_with_config(&self.config.instrument_provider, self.config.us)
                .await
                .context("failed to request Binance Spot instruments")?;

            if instruments.is_empty() {
                log::warn!("No instruments returned for Binance Spot");
            } else {
                log::debug!("Loaded {} Spot instruments", instruments.len());
                self.http_client.cache_instruments(instruments);
            }

            self.core.set_instruments_initialized();
        }

        // Request initial account state
        let account_state = self
            .refresh_account_state()
            .await
            .context("failed to request Binance account state")?;

        if !account_state.balances.is_empty() {
            log::debug!(
                "Received account state with {} balance(s)",
                account_state.balances.len()
            );
        }

        self.emitter.send_account_state(account_state);

        // Wait for account to be registered in cache before completing connect
        crate::common::execution::await_account_registered(&self.core, self.core.account_id, 30.0)
            .await?;

        let session_result = async {
        if self.config.us {
            self.connect_us_user_data().await?;
        }

        if let Some(mut ws_trading) = self.ws_trading_client.clone() {
            match ws_trading.connect().await {
                Ok(()) => {
                    log::debug!("Connected to Binance Spot WS trading API");

                    let ws_trading_clone = ws_trading.clone();
                    let emitter = self.emitter.clone();
                    let account_id = self.core.account_id;
                    let clock = self.clock;
                    let http_client = self.http_client.clone();
                    let dispatch_state = self.dispatch_state.clone();
                    let treat_expired_as_canceled = self.config.treat_expired_as_canceled;
                    let ws_authenticated = self.ws_authenticated.clone();
                    let ws_user_data_subscribed = self.ws_user_data_subscribed.clone();
                    let (ws_setup_error_tx, mut ws_setup_error_rx) =
                        tokio::sync::mpsc::unbounded_channel();
                    let seen_trade_ids = std::sync::Arc::new(Mutex::new(FifoCache::new()));
                    let dispatch_active = Arc::clone(&self.ws_trading_dispatch_active);
                    let task_spawner = self
                        .session_tasks
                        .spawner()
                        .context("Binance Spot session task admission is closed")?;

                    self.session_tasks.spawn(async move {
                        dispatch_active.store(true, Ordering::Release);
                        let _active = DispatchActiveGuard(dispatch_active);
                        let mut resubscribing = false;

                        loop {
                            match ws_trading_clone.recv().await {
                                Some(msg) => {
                                    match &msg {
                                        BinanceSpotWsTradingMessage::Reconnected => {
                                            ws_trading_clone.mark_user_data_inactive();
                                            resubscribing = true;
                                            if let Err(e) = ws_trading_clone.session_logon().await {
                                                resubscribing = false;
                                                log::error!(
                                                    "Failed to re-authenticate Binance Spot user data stream: {e}"
                                                );
                                            }
                                        }
                                        BinanceSpotWsTradingMessage::Authenticated if resubscribing => {
                                            if let Err(e) =
                                                ws_trading_clone.subscribe_user_data().await
                                            {
                                                resubscribing = false;
                                                log::error!(
                                                    "Failed to resubscribe Binance Spot user data stream: {e}"
                                                );
                                            }
                                            continue;
                                        }
                                        BinanceSpotWsTradingMessage::UserDataSubscribed { .. } => {
                                            let was_resubscribing = resubscribing;
                                            resubscribing = false;
                                            ws_trading_clone.mark_user_data_active();

                                            if was_resubscribing {
                                                continue;
                                            }
                                        }
                                        BinanceSpotWsTradingMessage::AuthenticationRejected(reason)
                                            if resubscribing =>
                                        {
                                            log::warn!(
                                                "Binance Spot reconnect authentication failed; retrying: {reason}"
                                            );
                                            tokio::time::sleep(WS_RECONNECT_SETUP_RETRY_DELAY).await;

                                            if let Err(e) = ws_trading_clone.session_logon().await {
                                                resubscribing = false;
                                                log::error!(
                                                    "Failed to retry Binance Spot reconnect authentication: {e}"
                                                );
                                            }
                                            continue;
                                        }
                                        BinanceSpotWsTradingMessage::UserDataSubscriptionRejected(reason)
                                            if resubscribing =>
                                        {
                                            log::warn!(
                                                "Binance Spot reconnect user data subscription failed; retrying: {reason}"
                                            );
                                            tokio::time::sleep(WS_RECONNECT_SETUP_RETRY_DELAY).await;

                                            if let Err(e) =
                                                ws_trading_clone.subscribe_user_data().await
                                            {
                                                resubscribing = false;
                                                log::error!(
                                                    "Failed to retry Binance Spot user data subscription: {e}"
                                                );
                                            }
                                            continue;
                                        }
                                        _ => {}
                                    }

                                    dispatch_ws_trading_message(
                                        msg,
                                        &emitter,
                                        &http_client,
                                        account_id,
                                        treat_expired_as_canceled,
                                        clock,
                                        &dispatch_state,
                                        &ws_authenticated,
                                        &ws_user_data_subscribed,
                                        &ws_setup_error_tx,
                                        &seen_trade_ids,
                                        &task_spawner,
                                    );
                                }
                                None => {
                                    log::warn!("WS trading dispatch loop ended");
                                    break;
                                }
                            }
                        }
                    })?;

                    if let Err(e) = ws_trading.session_logon().await {
                        let reason = format!("WS session logon failed: {e}");
                        return Err(self.ws_setup_failure(ws_trading, reason).await);
                    } else {
                        let auth_result = wait_for_ws_setup_response(
                            ws_setup_timeout,
                            self.ws_authenticated.notified(),
                            &mut ws_setup_error_rx,
                            "WS session authentication timed out",
                        )
                        .await;

                        if let Err(e) = auth_result {
                            return Err(self.ws_setup_failure(ws_trading, e.to_string()).await);
                        } else if let Err(e) = ws_trading.subscribe_user_data().await {
                            let reason = format!("WS user data subscribe failed: {e}");
                            return Err(self.ws_setup_failure(ws_trading, reason).await);
                        } else {
                            let subscribe_result = wait_for_ws_setup_response(
                                ws_setup_timeout,
                                self.ws_user_data_subscribed.notified(),
                                &mut ws_setup_error_rx,
                                "WS user data subscription timed out",
                            )
                            .await;

                            if let Err(e) = subscribe_result {
                                return Err(self.ws_setup_failure(ws_trading, e.to_string()).await);
                            } else {
                                self.ws_trading_client = Some(ws_trading);
                            }
                        }
                    }
                }
                Err(e) => {
                    let reason = format!("Failed to connect WS trading API: {e}");
                    return Err(self.ws_setup_failure(ws_trading, reason).await);
                }
            }
        }

        let refresh_secs = self.config.instrument_refresh_interval_secs;
        if refresh_secs > 0 {
            let http_client = self.http_client.clone();
            let provider = self.config.instrument_provider.clone();
            let us = self.config.us;

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

                loop {
                    interval.tick().await;

                    match http_client
                        .request_instruments_with_config(&provider, us)
                        .await
                    {
                        Ok(instruments) => log::debug!(
                            "Refreshed Binance Spot execution instruments: count={}",
                            instruments.len()
                        ),
                        Err(e) => {
                            log::warn!("Binance Spot execution instrument refresh failed: {e}");
                        }
                    }
                }
            })?;
        }

        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 Spot startup teardown failed: {teardown_error}"
                )));
            }
            return Err(e);
        }

        setup_guard.disarm();
        self.core.set_connected();
        log::info!("Connected: client_id={}", self.core.client_id);
        Ok(())
    }

    async fn disconnect(&mut self) -> anyhow::Result<()> {
        self.teardown_partial_connect().await?;
        log::info!("Disconnected: client_id={}", self.core.client_id);
        Ok(())
    }

    fn query_account(&self, _cmd: QueryAccount) -> anyhow::Result<()> {
        self.update_account_state();
        Ok(())
    }

    fn query_order(&self, cmd: QueryOrder) -> anyhow::Result<()> {
        log::debug!("query_order: client_order_id={}", cmd.client_order_id);

        let http_client = self.http_client.clone();
        let command = cmd;
        let event_emitter = self.emitter.clone();
        let account_id = self.core.account_id;
        let treat_expired_as_canceled = self.config.treat_expired_as_canceled;

        self.spawn_task("query_order", async move {
            let result = http_client
                .request_order_status_report(
                    account_id,
                    command.instrument_id,
                    command.venue_order_id,
                    Some(command.client_order_id),
                )
                .await;

            match result {
                Ok(Some(mut report)) => {
                    normalize_spot_order_status_report(&mut report, treat_expired_as_canceled);
                    event_emitter.send_order_status_report(report);
                }
                Ok(None) => log::debug!(
                    "No order status report returned: client_order_id={}",
                    command.client_order_id
                ),
                Err(e) => log::warn!("Failed to query order status: {e}"),
            }

            Ok(())
        });

        Ok(())
    }

    fn generate_account_state(
        &self,
        balances: Vec<AccountBalance>,
        margins: Vec<MarginBalance>,
        reported: bool,
        ts_event: UnixNanos,
        info: Option<Params>,
    ) -> anyhow::Result<()> {
        self.emitter
            .emit_account_state(balances, margins, reported, ts_event, info);
        Ok(())
    }

    fn start(&mut self) -> anyhow::Result<()> {
        if self.core.is_started() {
            return Ok(());
        }

        self.emitter.set_sender(get_exec_event_sender());
        self.core.set_started();

        // Spawn instrument bootstrap task
        let http_client = self.http_client.clone();
        let provider = self.config.instrument_provider.clone();
        let us = self.config.us;

        self.session_tasks.spawn(async move {
            match http_client
                .request_instruments_with_config(&provider, us)
                .await
            {
                Ok(instruments) => {
                    if instruments.is_empty() {
                        log::warn!("No instruments returned for Binance Spot");
                    } else {
                        http_client.cache_instruments(instruments);
                        log::debug!("Instruments initialized");
                    }
                }
                Err(e) => {
                    log::error!("Failed to request Binance Spot instruments: {e}");
                }
            }
        })?;

        log::info!(
            "Started: client_id={}, account_id={}, account_type={:?}, environment={:?}, product_type={:?}",
            self.core.client_id,
            self.core.account_id,
            self.core.account_type,
            self.config.environment,
            self.config.product_type,
        );
        Ok(())
    }

    fn stop(&mut self) -> anyhow::Result<()> {
        if self.core.is_stopped() {
            return Ok(());
        }

        if let Some(client) = self.ws_trading_client.as_ref() {
            client.mark_user_data_inactive();
            client.begin_shutdown();
        }

        if let Some(client) = self.ws_user_data_client.lock().as_ref() {
            client.mark_user_data_inactive();
            client.begin_shutdown();
        }

        self.core.set_stopped();
        self.core.set_disconnected();
        self.abort_session_tasks();
        self.abort_pending_tasks();
        log::info!("Stopped: client_id={}", self.core.client_id);
        Ok(())
    }

    async fn generate_order_status_report(
        &self,
        cmd: &GenerateOrderStatusReport,
    ) -> anyhow::Result<Option<OrderStatusReport>> {
        let Some(instrument_id) = cmd.instrument_id else {
            log::warn!("generate_order_status_report requires instrument_id: {cmd:?}");
            return Ok(None);
        };

        // Convert ClientOrderId to VenueOrderId if provided (API naming quirk)
        let venue_order_id = cmd
            .venue_order_id
            .as_ref()
            .map(|id| VenueOrderId::new(id.inner()));

        let report = self
            .http_client
            .request_order_status_report(
                self.core.account_id,
                instrument_id,
                venue_order_id,
                cmd.client_order_id,
            )
            .await?;

        Ok(report.map(|mut report| {
            normalize_spot_order_status_report(&mut report, self.config.treat_expired_as_canceled);
            report
        }))
    }

    async fn generate_order_status_reports(
        &self,
        cmd: &GenerateOrderStatusReports,
    ) -> anyhow::Result<Vec<OrderStatusReport>> {
        let start_dt = cmd.start.map(|nanos| nanos.to_datetime_utc());
        let end_dt = cmd.end.map(|nanos| nanos.to_datetime_utc());

        let mut reports = self
            .http_client
            .request_order_status_reports(
                self.core.account_id,
                cmd.instrument_id,
                start_dt,
                end_dt,
                cmd.open_only,
                None, // limit
            )
            .await?;

        normalize_spot_order_status_reports(&mut reports, self.config.treat_expired_as_canceled);

        Ok(reports)
    }

    async fn generate_fill_reports(
        &self,
        cmd: GenerateFillReports,
    ) -> anyhow::Result<Vec<FillReport>> {
        let Some(instrument_id) = cmd.instrument_id else {
            log::warn!("generate_fill_reports requires instrument_id for Binance Spot");
            return Ok(Vec::new());
        };

        // Convert ClientOrderId to VenueOrderId if provided (API naming quirk)
        let venue_order_id = cmd
            .venue_order_id
            .as_ref()
            .map(|id| VenueOrderId::new(id.inner()));
        let requested_start_time = cmd
            .start
            .map(|start| start.as_i64() / NANOSECONDS_IN_MILLISECOND as i64);
        let requested_end_time = cmd
            .end
            .map(|end| end.as_i64() / NANOSECONDS_IN_MILLISECOND as i64);
        if let (Some(start), Some(end)) = (requested_start_time, requested_end_time) {
            anyhow::ensure!(
                start <= end,
                "fill report start time must not exceed end time"
            );
        }

        let mut reports = Vec::new();
        let mut seen_trade_ids = AHashSet::new();

        if venue_order_id.is_some() {
            let mut from_id = 0;

            loop {
                let page = self
                    .http_client
                    .request_fill_reports_with_cursor(
                        self.core.account_id,
                        instrument_id,
                        venue_order_id,
                        None,
                        None,
                        Some(from_id),
                        Some(ACCOUNT_TRADES_PAGE_LIMIT),
                    )
                    .await?;

                if page.is_empty() {
                    break;
                }

                let page_len = page.len();
                let max_trade_id = max_trade_id(&page)?;
                let passed_end = requested_end_time.is_some_and(|end_time| {
                    page.iter().any(|report| report_time_ms(report) > end_time)
                });

                reports.extend(page.into_iter().filter(|report| {
                    requested_start_time
                        .is_none_or(|start_time| report_time_ms(report) >= start_time)
                        && requested_end_time
                            .is_none_or(|end_time| report_time_ms(report) <= end_time)
                        && seen_trade_ids.insert(report.trade_id)
                }));

                if page_len < ACCOUNT_TRADES_PAGE_LIMIT as usize || passed_end {
                    break;
                }

                let next_from_id = max_trade_id
                    .checked_add(1)
                    .context("Binance Spot trade ID overflow during pagination")?;
                anyhow::ensure!(
                    next_from_id > from_id,
                    "Binance Spot account-trades pagination made no progress"
                );
                from_id = next_from_id;
            }
        } else if let Some(query_start_time) = requested_start_time {
            let query_end_time = requested_end_time.unwrap_or_else(|| {
                self.clock.get_time_ns().as_i64() / NANOSECONDS_IN_MILLISECOND as i64
            });
            anyhow::ensure!(
                query_start_time <= query_end_time,
                "fill report start time must not exceed end time"
            );
            let mut window_start = query_start_time;

            loop {
                let window_end = window_start
                    .saturating_add(ACCOUNT_TRADES_MAX_INTERVAL_MS)
                    .min(query_end_time);
                let mut from_id = None;

                loop {
                    let start = if from_id.is_none() {
                        Some(
                            Timestamp::from_millisecond(window_start)
                                .context("invalid Binance Spot account-trades start time")?,
                        )
                    } else {
                        None
                    };
                    let end = if from_id.is_none() {
                        Some(
                            Timestamp::from_millisecond(window_end)
                                .context("invalid Binance Spot account-trades end time")?,
                        )
                    } else {
                        None
                    };
                    let page = self
                        .http_client
                        .request_fill_reports_with_cursor(
                            self.core.account_id,
                            instrument_id,
                            None,
                            start,
                            end,
                            from_id,
                            Some(ACCOUNT_TRADES_PAGE_LIMIT),
                        )
                        .await?;

                    if page.is_empty() {
                        break;
                    }

                    let page_len = page.len();
                    let max_trade_id = max_trade_id(&page)?;
                    let passed_window_end = page
                        .iter()
                        .any(|report| report_time_ms(report) > window_end);

                    reports.extend(page.into_iter().filter(|report| {
                        let report_time = report_time_ms(report);
                        report_time >= window_start
                            && report_time <= window_end
                            && seen_trade_ids.insert(report.trade_id)
                    }));

                    if page_len < ACCOUNT_TRADES_PAGE_LIMIT as usize || passed_window_end {
                        break;
                    }

                    let next_from_id = max_trade_id
                        .checked_add(1)
                        .context("Binance Spot trade ID overflow during pagination")?;
                    anyhow::ensure!(
                        from_id.is_none_or(|cursor| next_from_id > cursor),
                        "Binance Spot account-trades pagination made no progress"
                    );
                    from_id = Some(next_from_id);
                }

                if window_end >= query_end_time {
                    break;
                }
                window_start = window_end.saturating_add(1);
            }
        } else {
            let mut from_id = 0;

            loop {
                let page = self
                    .http_client
                    .request_fill_reports_with_cursor(
                        self.core.account_id,
                        instrument_id,
                        None,
                        None,
                        None,
                        Some(from_id),
                        Some(ACCOUNT_TRADES_PAGE_LIMIT),
                    )
                    .await?;

                if page.is_empty() {
                    break;
                }

                let page_len = page.len();
                let max_trade_id = max_trade_id(&page)?;
                let passed_end = requested_end_time.is_some_and(|end_time| {
                    page.iter().any(|report| report_time_ms(report) > end_time)
                });

                reports.extend(page.into_iter().filter(|report| {
                    requested_end_time.is_none_or(|end_time| report_time_ms(report) <= end_time)
                        && seen_trade_ids.insert(report.trade_id)
                }));

                if page_len < ACCOUNT_TRADES_PAGE_LIMIT as usize || passed_end {
                    break;
                }

                let next_from_id = max_trade_id
                    .checked_add(1)
                    .context("Binance Spot trade ID overflow during pagination")?;
                anyhow::ensure!(
                    next_from_id > from_id,
                    "Binance Spot account-trades pagination made no progress"
                );
                from_id = next_from_id;
            }
        }

        let mut reports_with_trade_ids = reports
            .into_iter()
            .map(|report| parse_trade_id(&report).map(|trade_id| (report, trade_id)))
            .collect::<anyhow::Result<Vec<_>>>()?;
        reports_with_trade_ids
            .sort_unstable_by_key(|(report, trade_id)| (report.ts_event, *trade_id));
        Ok(reports_with_trade_ids
            .into_iter()
            .map(|(report, _)| report)
            .collect())
    }

    async fn generate_position_status_reports(
        &self,
        _cmd: &GeneratePositionStatusReports,
    ) -> anyhow::Result<Vec<PositionStatusReport>> {
        // Spot trading doesn't have positions in the traditional sense
        // Returns empty for spot, could be extended for margin positions
        Ok(Vec::new())
    }

    async fn generate_mass_status(
        &self,
        lookback_mins: Option<u64>,
    ) -> anyhow::Result<Option<ExecutionMassStatus>> {
        log::info!("Generating ExecutionMassStatus (lookback_mins={lookback_mins:?})");

        let ts_now = self.clock.get_time_ns();

        let start = if let Some(mins) = lookback_mins {
            let lookback_ns = checked_mins_to_nanos(mins)
                .context("lookback minutes exceed the nanosecond range")?;
            Some(UnixNanos::from(ts_now.as_u64().saturating_sub(lookback_ns)))
        } else {
            None
        };

        // Binance requires instrument_id for historical orders (open_only=false).
        // Use open_only=true for mass status to get all open orders across instruments.
        let order_cmd = GenerateOrderStatusReportsBuilder::default()
            .ts_init(ts_now)
            .open_only(true)
            .start(start)
            .build()
            .map_err(|e| anyhow::anyhow!("{e}"))?;

        let position_cmd = GeneratePositionStatusReportsBuilder::default()
            .ts_init(ts_now)
            .start(start)
            .build()
            .map_err(|e| anyhow::anyhow!("{e}"))?;

        let (order_reports, position_reports) = tokio::try_join!(
            self.generate_order_status_reports(&order_cmd),
            self.generate_position_status_reports(&position_cmd),
        )?;

        let mut instrument_ids: Vec<_> = order_reports
            .iter()
            .map(|report| report.instrument_id)
            .collect();
        {
            let cache = self.core.cache();
            instrument_ids.extend(
                cache
                    .orders_open(
                        Some(&BINANCE_VENUE),
                        None,
                        None,
                        Some(&self.core.account_id),
                        None,
                    )
                    .into_iter()
                    .chain(cache.orders_inflight(
                        Some(&BINANCE_VENUE),
                        None,
                        None,
                        Some(&self.core.account_id),
                        None,
                    ))
                    .map(|order| order.instrument_id())
                    .filter(|instrument_id| {
                        self.http_client
                            .get_instrument(&instrument_id.symbol.inner())
                            .is_some_and(|instrument| instrument.id() == *instrument_id)
                    }),
            );
        }
        instrument_ids.sort_unstable();
        instrument_ids.dedup();

        let mut fill_reports = Vec::new();

        for instrument_id in instrument_ids {
            let fill_cmd = GenerateFillReportsBuilder::default()
                .ts_init(ts_now)
                .instrument_id(Some(instrument_id))
                .start(start)
                .build()
                .map_err(|e| anyhow::anyhow!("{e}"))?;
            fill_reports.extend(self.generate_fill_reports(fill_cmd).await?);
        }

        log::info!("Received {} OrderStatusReports", order_reports.len());
        log::info!("Received {} FillReports", fill_reports.len());
        log::info!("Received {} PositionReports", position_reports.len());

        let mut mass_status = ExecutionMassStatus::new(
            self.core.client_id,
            self.core.account_id,
            *BINANCE_VENUE,
            ts_now,
            None,
        );

        mass_status.add_order_reports(order_reports);
        mass_status.add_fill_reports(fill_reports);
        mass_status.add_position_reports(position_reports);

        Ok(Some(mass_status))
    }

    fn submit_order(&self, cmd: SubmitOrder) -> anyhow::Result<()> {
        let order = self.core.cache().try_order_owned(&cmd.client_order_id)?;

        if order.is_closed() {
            let client_order_id = order.client_order_id();
            log::warn!("Cannot submit closed order {client_order_id}");
            return Ok(());
        }

        if order.time_in_force() == TimeInForce::Gtd && self.config.use_gtd {
            time_in_force_to_binance_spot(order.time_in_force(), self.config.use_gtd)?;
        }

        self.ensure_ws_user_data_active()?;

        log::debug!("OrderSubmitted client_order_id={}", order.client_order_id());
        self.emitter.emit_order_submitted(&order);

        self.submit_order_internal(&cmd)
    }

    fn submit_order_list(&self, cmd: SubmitOrderList) -> anyhow::Result<()> {
        if cmd.order_list.client_order_ids.is_empty() {
            log::debug!("submit_order_list called with empty order list");
            return Ok(());
        }

        self.ensure_ws_user_data_active()?;

        let orders = self.core.get_orders_for_list(&cmd.order_list)?;

        if let Some(order) = orders.iter().find(|order| order.is_closed()) {
            let reason = format!("Cannot submit closed order {}", order.client_order_id());
            for order in &orders {
                self.emitter.emit_order_denied(order, &reason);
            }
            return Ok(());
        }

        let params = match build_spot_order_list_params(
            cmd.order_list.id.as_ref(),
            &orders,
            self.config.use_gtd,
        ) {
            Ok(request) => request,
            Err(reason) => {
                for order in &orders {
                    self.emitter.emit_order_denied(order, &reason);
                }
                return Ok(());
            }
        };

        for order in &orders {
            self.dispatch_state.order_identities.insert(
                order.client_order_id(),
                OrderIdentity {
                    instrument_id: order.instrument_id(),
                    strategy_id: order.strategy_id(),
                    order_side: order.order_side(),
                    order_type: order.order_type(),
                    price: order.price(),
                    quantity: order.quantity(),
                    venue_position_id: None,
                },
            );
            self.emitter.emit_order_submitted(order);
        }

        let event_emitter = self.emitter.clone();
        let trader_id = self.core.trader_id;
        let account_id = self.core.account_id;
        let clock = self.clock;
        let http_client = self.http_client.clone();
        let dispatch_state = self.dispatch_state.clone();

        self.spawn_task("submit_order_list_http", async move {
            if let Err(e) = submit_spot_order_list(&http_client, &params).await {
                handle_spot_order_list_submit_error(
                    &event_emitter,
                    &dispatch_state,
                    trader_id,
                    account_id,
                    clock,
                    &orders,
                    e,
                )?;
            }

            Ok(())
        });

        Ok(())
    }

    fn modify_order(&self, cmd: ModifyOrder) -> anyhow::Result<()> {
        // Binance Spot uses cancel-replace for order modification, which requires
        // the full order specification (side, type, time_in_force). Since ModifyOrder
        // doesn't include these fields, we need to look up the original order from cache.
        let order = self
            .core
            .cache()
            .order(&cmd.client_order_id)
            .map(|o| o.clone());

        let Some(order) = order else {
            log::warn!(
                "Cannot modify order {}: not found in cache",
                cmd.client_order_id
            );
            let ts_init = self.clock.get_time_ns();
            let rejected_event = OrderModifyRejected::new(
                self.core.trader_id,
                cmd.strategy_id,
                cmd.instrument_id,
                cmd.client_order_id,
                "Order not found in cache for modify".into(),
                UUID4::new(),
                ts_init, // no venue timestamp, rejected locally
                ts_init,
                false,
                cmd.venue_order_id,
                Some(self.core.account_id),
            );

            self.emitter
                .send_order_event(OrderEventAny::ModifyRejected(rejected_event));
            return Ok(());
        };

        let event_emitter = self.emitter.clone();
        let trader_id = self.core.trader_id;
        let account_id = self.core.account_id;
        let clock = self.clock;

        let order_side = order.order_side();
        let order_type = order.order_type();
        let time_in_force = order.time_in_force();
        let quantity = cmd.quantity.unwrap_or_else(|| order.quantity());
        let use_gtd = self.config.use_gtd;

        if self.ws_order_transport_active() {
            let command = cmd;
            let ws_client = self.ws_trading_client.as_ref().unwrap().clone();
            let dispatch_state = self.dispatch_state.clone();
            let params = build_cancel_replace_params(&command, &order, quantity, use_gtd)?;

            // Pre-register before sending to avoid response racing the insert
            let request_id = ws_client.next_request_id();
            dispatch_state.pending_requests.insert(
                request_id.clone(),
                PendingRequest {
                    client_order_id: command.client_order_id,
                    venue_order_id: command.venue_order_id,
                    operation: PendingOperation::Modify,
                },
            );

            self.spawn_task("modify_order_ws", async move {
                if let Err(e) = ws_client
                    .cancel_replace_order_with_id(request_id.clone(), params)
                    .await
                {
                    dispatch_state.pending_requests.remove(&request_id);
                    log::warn!(
                        "WS modify request failed for {}, awaiting reconciliation: {e}",
                        command.client_order_id
                    );
                    anyhow::bail!("WS modify order failed: {e}");
                }
                Ok(())
            });
        } else {
            let command = cmd;
            let http_client = self.http_client.clone();
            log::debug!("WS trading not active, falling back to HTTP for modify_order");

            self.spawn_task("modify_order_http", async move {
                let result = match command.venue_order_id {
                    Some(venue_order_id) => {
                        http_client
                            .modify_order(
                                account_id,
                                command.instrument_id,
                                venue_order_id,
                                command.client_order_id,
                                order_side,
                                order_type,
                                quantity,
                                time_in_force,
                                command.price,
                                use_gtd,
                            )
                            .await
                    }
                    None => Err(anyhow::anyhow!(BinanceSpotHttpError::ValidationError(
                        "venue_order_id required for modify".to_string()
                    ))),
                };

                match result {
                    Ok(report) => {
                        let ts_now = clock.get_time_ns();
                        let updated_event = OrderUpdated::new(
                            trader_id,
                            command.strategy_id,
                            command.instrument_id,
                            command.client_order_id,
                            report.quantity,
                            UUID4::new(),
                            ts_now,
                            ts_now,
                            false,
                            Some(report.venue_order_id),
                            Some(account_id),
                            report.price,
                            None,  // trigger_price
                            None,  // protection_price
                            false, // is_quote_quantity
                        );
                        event_emitter.send_order_event(OrderEventAny::Updated(updated_event));
                    }
                    Err(e) => {
                        if is_structured_venue_rejection(&e) || is_local_command_failure(&e) {
                            let ts_now = clock.get_time_ns();
                            let rejected_event = OrderModifyRejected::new(
                                trader_id,
                                command.strategy_id,
                                command.instrument_id,
                                command.client_order_id,
                                format!("modify-order-error: {e}").into(),
                                UUID4::new(),
                                ts_now,
                                ts_now,
                                false,
                                command.venue_order_id,
                                Some(account_id),
                            );
                            event_emitter
                                .send_order_event(OrderEventAny::ModifyRejected(rejected_event));
                        } else {
                            log::warn!(
                                "Ambiguous modify failure for {}, awaiting reconciliation: {e}",
                                command.client_order_id
                            );
                        }
                        return Err(e);
                    }
                }
                Ok(())
            });
        }

        Ok(())
    }

    fn cancel_order(&self, cmd: CancelOrder) -> anyhow::Result<()> {
        self.cancel_order_internal(&cmd);
        Ok(())
    }

    fn cancel_all_orders(&self, cmd: CancelAllOrders) -> anyhow::Result<()> {
        let event_emitter = self.emitter.clone();
        let trader_id = self.core.trader_id;
        let account_id = self.core.account_id;
        let clock = self.clock;

        if self.ws_order_transport_active() {
            let ws_client = self.ws_trading_client.as_ref().unwrap().clone();
            let symbol = cmd.instrument_id.symbol.to_string();

            self.spawn_task("cancel_all_orders_ws", async move {
                if let Err(e) = ws_client.cancel_all_orders(symbol).await {
                    log::error!("WS cancel_all_orders failed: {e}");
                }
                // Individual cancel confirmations dispatched via WS trading message loop
                Ok(())
            });

            return Ok(());
        }

        log::debug!("WS trading not active, falling back to HTTP for cancel_all_orders");
        let http_client = self.http_client.clone();

        // Build strategy lookup from cache before spawning (cache is not Send)
        let strategy_lookup: AHashMap<ClientOrderId, StrategyId> = {
            let cache = self.core.cache();
            cache
                .orders_open(None, Some(&cmd.instrument_id), None, None, None)
                .into_iter()
                .map(|order| (order.client_order_id(), order.strategy_id()))
                .collect()
        };

        let command = cmd;
        self.spawn_task("cancel_all_orders_http", async move {
            let canceled_orders = http_client.cancel_all_orders(command.instrument_id).await?;

            for (venue_order_id, client_order_id) in canceled_orders {
                let strategy_id = strategy_lookup
                    .get(&client_order_id)
                    .copied()
                    .unwrap_or(command.strategy_id);

                let canceled_event = OrderCanceled::new(
                    trader_id,
                    strategy_id,
                    command.instrument_id,
                    client_order_id,
                    UUID4::new(),
                    command.ts_init,
                    clock.get_time_ns(),
                    false,
                    Some(venue_order_id),
                    Some(account_id),
                );

                event_emitter.send_order_event(OrderEventAny::Canceled(canceled_event));
            }

            Ok(())
        });

        Ok(())
    }

    fn batch_cancel_orders(&self, cmd: BatchCancelOrders) -> anyhow::Result<()> {
        const BATCH_SIZE: usize = 5;

        if cmd.cancels.is_empty() {
            return Ok(());
        }

        let http_client = self.http_client.clone();
        let command = cmd;

        let event_emitter = self.emitter.clone();
        let trader_id = self.core.trader_id;
        let account_id = self.core.account_id;
        let clock = self.clock;

        self.spawn_task("batch_cancel_orders", async move {
            for chunk in command.cancels.chunks(BATCH_SIZE) {
                let batch_items: Vec<BatchCancelItem> = chunk
                    .iter()
                    .map(|cancel| {
                        if let Some(venue_order_id) = cancel.venue_order_id {
                            let order_id = venue_order_id.inner().parse::<i64>().unwrap_or(0);
                            if order_id != 0 {
                                BatchCancelItem::by_order_id(
                                    command.instrument_id.symbol.to_string(),
                                    order_id,
                                )
                            } else {
                                BatchCancelItem::by_client_order_id(
                                    command.instrument_id.symbol.to_string(),
                                    encode_broker_id(
                                        &cancel.client_order_id,
                                        BINANCE_NAUTILUS_SPOT_BROKER_ID,
                                    ),
                                )
                            }
                        } else {
                            BatchCancelItem::by_client_order_id(
                                command.instrument_id.symbol.to_string(),
                                encode_broker_id(
                                    &cancel.client_order_id,
                                    BINANCE_NAUTILUS_SPOT_BROKER_ID,
                                ),
                            )
                        }
                    })
                    .collect();

                match http_client.batch_cancel_orders(&batch_items).await {
                    Ok(results) => {
                        for (i, result) in results.iter().enumerate() {
                            let cancel = &chunk[i];

                            match result {
                                BatchCancelResult::Success(success) => {
                                    let venue_order_id =
                                        VenueOrderId::new(success.order_id.to_string());
                                    let canceled_event = OrderCanceled::new(
                                        trader_id,
                                        cancel.strategy_id,
                                        cancel.instrument_id,
                                        cancel.client_order_id,
                                        UUID4::new(),
                                        cancel.ts_init,
                                        clock.get_time_ns(),
                                        false,
                                        Some(venue_order_id),
                                        Some(account_id),
                                    );

                                    event_emitter
                                        .send_order_event(OrderEventAny::Canceled(canceled_event));
                                }
                                BatchCancelResult::Error(error) => {
                                    let rejected_event = OrderCancelRejected::new(
                                        trader_id,
                                        cancel.strategy_id,
                                        cancel.instrument_id,
                                        cancel.client_order_id,
                                        format!(
                                            "batch-cancel-error: code={}, msg={}",
                                            error.code, error.msg
                                        )
                                        .into(),
                                        UUID4::new(),
                                        clock.get_time_ns(),
                                        cancel.ts_init,
                                        false,
                                        cancel.venue_order_id,
                                        Some(account_id),
                                    );

                                    event_emitter.send_order_event(OrderEventAny::CancelRejected(
                                        rejected_event,
                                    ));
                                }
                            }
                        }
                    }
                    Err(e) => {
                        if is_local_http_command_failure(&e) {
                            log::warn!(
                                "Batch cancel command failed local validation for {} orders: {e}",
                                chunk.len()
                            );
                        } else {
                            log::warn!(
                                "Ambiguous batch cancel failure for {} orders, awaiting reconciliation: {e}",
                                chunk.len()
                            );
                        }
                    }
                }
            }

            Ok(())
        });

        Ok(())
    }
}

struct DispatchActiveGuard(Arc<AtomicBool>);

impl Drop for DispatchActiveGuard {
    fn drop(&mut self) {
        self.0.store(false, Ordering::Release);
    }
}

fn max_trade_id(reports: &[FillReport]) -> anyhow::Result<i64> {
    let mut max_trade_id = None;

    for report in reports {
        let trade_id = parse_trade_id(report)?;
        max_trade_id = Some(max_trade_id.map_or(trade_id, |current: i64| current.max(trade_id)));
    }

    max_trade_id.context("Binance Spot account-trades page was empty")
}

fn parse_trade_id(report: &FillReport) -> anyhow::Result<i64> {
    report
        .trade_id
        .to_string()
        .parse::<i64>()
        .with_context(|| format!("invalid Binance Spot trade ID {}", report.trade_id))
}

fn report_time_ms(report: &FillReport) -> i64 {
    report.ts_event.as_i64() / NANOSECONDS_IN_MILLISECOND as i64
}

fn normalize_spot_order_status_report(
    report: &mut OrderStatusReport,
    treat_expired_as_canceled: bool,
) {
    if treat_expired_as_canceled && report.order_status == OrderStatus::Expired {
        report.order_status = OrderStatus::Canceled;
    }
}

fn normalize_spot_order_status_reports(
    reports: &mut [OrderStatusReport],
    treat_expired_as_canceled: bool,
) {
    for report in reports {
        normalize_spot_order_status_report(report, treat_expired_as_canceled);
    }
}

async fn wait_for_ws_setup_response(
    timeout: Duration,
    success: impl Future<Output = ()>,
    setup_errors: &mut tokio::sync::mpsc::UnboundedReceiver<String>,
    timeout_message: &'static str,
) -> anyhow::Result<()> {
    tokio::pin!(success);

    let result = tokio::time::timeout(timeout, async {
        tokio::select! {
            () = &mut success => Ok(()),
            err = setup_errors.recv() => {
                anyhow::bail!(
                    "{}",
                    err.unwrap_or_else(|| "WS setup error channel closed".to_string()),
                )
            }
        }
    })
    .await;

    result.map_err(|_| anyhow::anyhow!(timeout_message))?
}

#[expect(clippy::too_many_arguments)]
fn dispatch_ws_trading_message(
    msg: BinanceSpotWsTradingMessage,
    emitter: &ExecutionEventEmitter,
    http_client: &BinanceSpotHttpClient,
    account_id: AccountId,
    treat_expired_as_canceled: bool,
    clock: &'static AtomicTime,
    dispatch_state: &WsDispatchState,
    ws_authenticated: &tokio::sync::Notify,
    ws_user_data_subscribed: &tokio::sync::Notify,
    ws_setup_error_tx: &tokio::sync::mpsc::UnboundedSender<String>,
    seen_trade_ids: &std::sync::Arc<Mutex<FifoCache<(Ustr, i64), 10_000>>>,
    task_spawner: &TaskSpawner,
) {
    match msg {
        BinanceSpotWsTradingMessage::OrderAccepted {
            request_id,
            response,
        } => {
            dispatch_state.pending_requests.remove(&request_id);
            log::debug!(
                "WS order accepted: request_id={request_id}, order_id={}",
                response.order_id
            );
            // OrderAccepted event is synthesized from UDS executionReport (New)
        }
        BinanceSpotWsTradingMessage::OrderRejected {
            request_id,
            code,
            msg,
        } => {
            log::debug!("WS order rejected: request_id={request_id}, code={code}, msg={msg}");
            if let Some((_, pending)) = dispatch_state.pending_requests.remove(&request_id) {
                let code_i64 = i64::from(code);
                if matches!(
                    code_i64,
                    BINANCE_UNEXPECTED_RESPONSE_CODE | BINANCE_STATUS_UNKNOWN_CODE
                ) {
                    log::warn!(
                        "Ambiguous WS submit failure for {}, awaiting reconciliation: code={code}, msg={msg}",
                        pending.client_order_id,
                    );
                    return;
                }

                // Clone to drop the DashMap read guard before cleanup_terminal
                let identity = dispatch_state
                    .order_identities
                    .get(&pending.client_order_id)
                    .map(|r| r.clone());

                if let Some(identity) = identity {
                    let due_post_only = code_i64 == BINANCE_GTX_ORDER_REJECT_CODE
                        || (code_i64 == BINANCE_NEW_ORDER_REJECTED_CODE
                            && msg == BINANCE_SPOT_POST_ONLY_REJECT_MSG);
                    let ts_now = clock.get_time_ns();
                    let rejected = OrderRejected::new(
                        emitter.trader_id(),
                        identity.strategy_id,
                        identity.instrument_id,
                        pending.client_order_id,
                        account_id,
                        Ustr::from(&format!("code={code}: {msg}")),
                        UUID4::new(),
                        ts_now,
                        ts_now,
                        false,
                        due_post_only,
                    );
                    dispatch_state.cleanup_terminal(pending.client_order_id);
                    emitter.send_order_event(OrderEventAny::Rejected(rejected));
                } else {
                    log::warn!(
                        "No order identity for {}, cannot emit OrderRejected",
                        pending.client_order_id
                    );
                }
            } else {
                log::warn!("No pending request for {request_id}, cannot emit OrderRejected");
            }
        }
        BinanceSpotWsTradingMessage::OrderCanceled {
            request_id,
            response,
        } => {
            dispatch_state.pending_requests.remove(&request_id);
            log::debug!(
                "WS order canceled: request_id={request_id}, order_id={}",
                response.order_id
            );
            // OrderCanceled event is synthesized from UDS executionReport (Canceled)
        }
        BinanceSpotWsTradingMessage::CancelRejected {
            request_id,
            code,
            msg,
        } => {
            log::warn!("WS cancel rejected: request_id={request_id}, code={code}, msg={msg}");
            if let Some((_, pending)) = dispatch_state.pending_requests.remove(&request_id)
                && let Some(identity) = dispatch_state
                    .order_identities
                    .get(&pending.client_order_id)
            {
                let ts_now = clock.get_time_ns();
                let rejected = OrderCancelRejected::new(
                    emitter.trader_id(),
                    identity.strategy_id,
                    identity.instrument_id,
                    pending.client_order_id,
                    Ustr::from(&format!("code={code}: {msg}")),
                    UUID4::new(),
                    ts_now,
                    ts_now,
                    false,
                    pending.venue_order_id,
                    Some(account_id),
                );
                emitter.send_order_event(OrderEventAny::CancelRejected(rejected));
            }
        }
        BinanceSpotWsTradingMessage::CancelReplaceAccepted {
            request_id,
            cancel_response,
            new_order_response,
        } => {
            dispatch_state.pending_requests.remove(&request_id);
            log::debug!(
                "WS cancel-replace accepted: request_id={request_id}, \
                 canceled_id={}, new_id={}",
                cancel_response.order_id,
                new_order_response.order_id,
            );
            // OrderUpdated event is synthesized from UDS executionReport (Replaced)
        }
        BinanceSpotWsTradingMessage::CancelReplaceRejected {
            request_id,
            code,
            msg,
        } => {
            log::warn!(
                "WS cancel-replace rejected: request_id={request_id}, code={code}, msg={msg}"
            );

            if let Some((_, pending)) = dispatch_state.pending_requests.remove(&request_id)
                && let Some(identity) = dispatch_state
                    .order_identities
                    .get(&pending.client_order_id)
            {
                let ts_now = clock.get_time_ns();
                let rejected = OrderModifyRejected::new(
                    emitter.trader_id(),
                    identity.strategy_id,
                    identity.instrument_id,
                    pending.client_order_id,
                    Ustr::from(&format!("code={code}: {msg}")),
                    UUID4::new(),
                    ts_now,
                    ts_now,
                    false,
                    pending.venue_order_id,
                    Some(account_id),
                );
                emitter.send_order_event(OrderEventAny::ModifyRejected(rejected));
            }
        }
        BinanceSpotWsTradingMessage::RequestFailed { request_id, msg } => {
            dispatch_state.pending_requests.remove(&request_id);
            log::error!(
                "WS trading request failed without structured venue response: request_id={request_id}, {msg}"
            );
        }
        BinanceSpotWsTradingMessage::AllOrdersCanceled {
            request_id,
            responses,
        } => {
            dispatch_state.pending_requests.remove(&request_id);
            log::debug!(
                "WS all orders canceled: request_id={request_id}, count={}",
                responses.len()
            );
            // Individual OrderCanceled events arrive via UDS executionReport
        }
        BinanceSpotWsTradingMessage::UserDataSubscribed { subscription_id } => {
            log::debug!("User data stream subscribed: id={subscription_id}");
            ws_user_data_subscribed.notify_one();
        }
        BinanceSpotWsTradingMessage::ExecutionReport(report) => {
            let ts_init = clock.get_time_ns();
            dispatch_execution_report(
                &report,
                emitter,
                http_client,
                account_id,
                treat_expired_as_canceled,
                dispatch_state,
                seen_trade_ids,
                ts_init,
            );
        }
        BinanceSpotWsTradingMessage::AccountPosition(position) => {
            let ts_init = clock.get_time_ns();
            let state = parse_spot_account_position(&position, account_id, ts_init);
            emitter.send_account_state(state);
        }
        BinanceSpotWsTradingMessage::BalanceUpdate(update) => {
            log::debug!(
                "Balance update: asset={}, delta={}",
                update.asset,
                update.delta,
            );
            let http_client = http_client.clone();
            let emitter = emitter.clone();

            if let Err(e) = task_spawner.spawn(async move {
                match http_client.request_account_state(account_id).await {
                    Ok(state) => emitter.send_account_state(state),
                    Err(e) => {
                        log::error!("Failed to refresh account state after balance update: {e}");
                    }
                }
            }) {
                log::warn!("Skipping Binance Spot balance refresh after shutdown began: {e}");
            }
        }
        BinanceSpotWsTradingMessage::Connected => {
            log::debug!("WS trading API connected");
        }
        BinanceSpotWsTradingMessage::Authenticated => {
            log::debug!("WS trading API authenticated");
            ws_authenticated.notify_one();
        }
        BinanceSpotWsTradingMessage::AuthenticationRejected(reason) => {
            log::error!("WS trading API authentication failed: {reason}");
            let _ = ws_setup_error_tx.send(reason);
        }
        BinanceSpotWsTradingMessage::Reconnected => {
            log::info!("WS trading API reconnected");
        }
        BinanceSpotWsTradingMessage::ServerShutdown { event_time } => {
            log::warn!(
                "WS trading API server shutdown notice (event_time={event_time}); reconnect expected within ~10 minutes"
            );
        }
        BinanceSpotWsTradingMessage::Error(err) => {
            log::error!("WS trading API error: {err}");
            let _ = ws_setup_error_tx.send(err);
        }
        BinanceSpotWsTradingMessage::UserDataSubscriptionRejected(reason) => {
            log::error!("WS trading API user data subscription failed: {reason}");
            let _ = ws_setup_error_tx.send(reason);
        }
    }
}

fn build_new_order_params(
    order: &impl Order,
    client_order_id: ClientOrderId,
    is_post_only: bool,
    is_quote_quantity: bool,
    use_gtd: bool,
) -> anyhow::Result<NewOrderParams> {
    let binance_side = BinanceSide::try_from(order.order_side())?;
    let binance_order_type = order_type_to_binance_spot(order.order_type(), is_post_only)?;

    let requires_trigger = matches!(
        order.order_type(),
        OrderType::StopMarket
            | OrderType::StopLimit
            | OrderType::MarketIfTouched
            | OrderType::LimitIfTouched
    );

    if requires_trigger && order.trigger_price().is_none() {
        anyhow::bail!("Conditional orders require a trigger price");
    }

    let supports_tif = matches!(
        binance_order_type,
        BinanceSpotOrderType::Limit
            | BinanceSpotOrderType::StopLossLimit
            | BinanceSpotOrderType::TakeProfitLimit
    );
    let binance_tif = time_in_force_to_binance_spot(order.time_in_force(), use_gtd)?;
    let binance_tif = if supports_tif {
        Some(binance_tif)
    } else {
        None
    };

    let qty_str = order.quantity().to_string();
    let (base_qty, quote_qty) = if is_quote_quantity {
        (None, Some(qty_str))
    } else {
        (Some(qty_str), None)
    };

    let client_id_str = encode_broker_id(&client_order_id, BINANCE_NAUTILUS_SPOT_BROKER_ID);

    Ok(NewOrderParams {
        symbol: order.instrument_id().symbol.to_string(),
        side: binance_side,
        order_type: binance_order_type,
        time_in_force: binance_tif,
        quantity: base_qty,
        quote_order_qty: quote_qty,
        price: order.price().map(|p| p.to_string()),
        new_client_order_id: Some(client_id_str),
        stop_price: order.trigger_price().map(|p| p.to_string()),
        trailing_delta: None,
        iceberg_qty: order.display_qty().map(|q| q.to_string()),
        new_order_resp_type: Some(BinanceOrderResponseType::Full),
        self_trade_prevention_mode: None,
        strategy_id: None,
        strategy_type: None,
    })
}

fn build_spot_order_list_params(
    order_list_id: &str,
    orders: &[OrderAny],
    use_gtd: bool,
) -> Result<NewOcoOrderListParams, String> {
    let has_grouped_order = orders.iter().any(is_grouped_order);

    if has_grouped_order {
        return build_spot_oco_order_list_params(order_list_id, orders, use_gtd);
    }

    Err("Binance Spot order-list submission currently supports only OCO lists".to_string())
}

fn build_spot_oco_order_list_params(
    order_list_id: &str,
    orders: &[OrderAny],
    use_gtd: bool,
) -> Result<NewOcoOrderListParams, String> {
    if orders.len() != 2 {
        return Err(format!(
            "Binance Spot OCO order-list submission requires exactly 2 orders, was {}",
            orders.len()
        ));
    }

    if orders
        .iter()
        .any(|order| order.contingency_type() != Some(ContingencyType::Oco))
    {
        return Err(
            "Binance Spot grouped order-list submission currently supports only OCO lists"
                .to_string(),
        );
    }

    let first = &orders[0];
    let second = &orders[1];
    if first.instrument_id() != second.instrument_id() {
        return Err("Binance Spot OCO order-list legs must use the same instrument".to_string());
    }

    if first.order_side() != second.order_side() {
        return Err("Binance Spot OCO order-list legs must use the same side".to_string());
    }

    if first.quantity() != second.quantity() {
        return Err("Binance Spot OCO order-list legs must use the same quantity".to_string());
    }

    if first.is_quote_quantity() || second.is_quote_quantity() {
        return Err("Binance Spot OCO order-list legs do not support quote quantity".to_string());
    }

    let mut above = None;
    let mut below = None;

    for order in orders {
        let params = build_new_order_params(
            order,
            order.client_order_id(),
            order.is_post_only(),
            false,
            use_gtd,
        )
        .map_err(|e| e.to_string())?;

        match spot_oco_leg_position(params.side, params.order_type)? {
            SpotOcoLegPosition::Above => {
                if above.replace(params).is_some() {
                    return Err(
                        "Binance Spot OCO order-list resolved more than one above leg".to_string(),
                    );
                }
            }
            SpotOcoLegPosition::Below => {
                if below.replace(params).is_some() {
                    return Err(
                        "Binance Spot OCO order-list resolved more than one below leg".to_string(),
                    );
                }
            }
        }
    }

    let above = above.ok_or_else(|| "Binance Spot OCO order-list missing above leg".to_string())?;
    let below = below.ok_or_else(|| "Binance Spot OCO order-list missing below leg".to_string())?;
    let quantity = above
        .quantity
        .clone()
        .ok_or_else(|| "Binance Spot OCO order-list requires base quantity".to_string())?;

    Ok(NewOcoOrderListParams {
        symbol: first.instrument_id().symbol.to_string(),
        list_client_order_id: Some(order_list_id.to_string()),
        side: above.side,
        quantity,
        above_type: above.order_type,
        above_client_order_id: above.new_client_order_id,
        above_iceberg_qty: above.iceberg_qty,
        above_price: above.price,
        above_stop_price: above.stop_price,
        above_time_in_force: above.time_in_force,
        below_type: below.order_type,
        below_client_order_id: below.new_client_order_id,
        below_iceberg_qty: below.iceberg_qty,
        below_price: below.price,
        below_stop_price: below.stop_price,
        below_time_in_force: below.time_in_force,
        new_order_resp_type: Some(BinanceOrderResponseType::Full),
        self_trade_prevention_mode: None,
    })
}

enum SpotOcoLegPosition {
    Above,
    Below,
}

fn spot_oco_leg_position(
    side: BinanceSide,
    order_type: BinanceSpotOrderType,
) -> Result<SpotOcoLegPosition, String> {
    match (side, order_type) {
        (
            BinanceSide::Sell,
            BinanceSpotOrderType::LimitMaker
            | BinanceSpotOrderType::TakeProfit
            | BinanceSpotOrderType::TakeProfitLimit,
        )
        | (
            BinanceSide::Buy,
            BinanceSpotOrderType::StopLoss | BinanceSpotOrderType::StopLossLimit,
        ) => Ok(SpotOcoLegPosition::Above),
        (
            BinanceSide::Sell,
            BinanceSpotOrderType::StopLoss | BinanceSpotOrderType::StopLossLimit,
        )
        | (
            BinanceSide::Buy,
            BinanceSpotOrderType::LimitMaker
            | BinanceSpotOrderType::TakeProfit
            | BinanceSpotOrderType::TakeProfitLimit,
        ) => Ok(SpotOcoLegPosition::Below),
        (_, unsupported) => Err(format!(
            "Unsupported Binance Spot OCO leg order type: {unsupported:?}"
        )),
    }
}

fn is_grouped_order(order: &OrderAny) -> bool {
    order.contingency_type().is_some()
        || order
            .linked_order_ids()
            .is_some_and(|linked_order_ids| !linked_order_ids.is_empty())
}

fn handle_spot_order_submit_success(client_order_id: ClientOrderId, venue_order_id: VenueOrderId) {
    log::debug!(
        "Order submit succeeded: client_order_id={client_order_id}, venue_order_id={venue_order_id}",
    );
}

async fn submit_spot_order_list(
    http_client: &BinanceSpotHttpClient,
    params: &NewOcoOrderListParams,
) -> Result<(), BinanceSpotHttpError> {
    let response = http_client.submit_oco_order_list(params).await?;
    log::debug!(
        "Order list submit succeeded: order_list_id={}, order_count={}",
        response.order_list_id,
        response.orders.len(),
    );
    Ok(())
}

fn handle_spot_order_list_submit_error(
    event_emitter: &ExecutionEventEmitter,
    dispatch_state: &WsDispatchState,
    trader_id: TraderId,
    account_id: AccountId,
    clock: &'static AtomicTime,
    orders: &[OrderAny],
    error: BinanceSpotHttpError,
) -> anyhow::Result<()> {
    let ambiguous = matches!(
        error,
        BinanceSpotHttpError::BinanceError {
            code: BINANCE_UNEXPECTED_RESPONSE_CODE | BINANCE_STATUS_UNKNOWN_CODE,
            ..
        }
    );

    if ambiguous {
        log::error!("Ambiguous order-list submit failure, awaiting reconciliation: {error}");
        return Err(error.into());
    }

    let reject_orders = matches!(
        error,
        BinanceSpotHttpError::BinanceError { .. }
            | BinanceSpotHttpError::MissingCredentials
            | BinanceSpotHttpError::ValidationError(_)
    );

    if reject_orders {
        let ts_now = clock.get_time_ns();
        let reason = format!("submit-order-list-error: {error}");
        for order in orders {
            let client_order_id = order.client_order_id();
            dispatch_state.cleanup_terminal(client_order_id);
            let rejected = OrderRejected::new(
                trader_id,
                order.strategy_id(),
                order.instrument_id(),
                client_order_id,
                account_id,
                reason.clone().into(),
                UUID4::new(),
                ts_now,
                ts_now,
                false,
                false,
            );
            event_emitter.send_order_event(OrderEventAny::Rejected(rejected));
        }
    } else {
        log::error!("Order-list submit failed, awaiting reconciliation: {error}");
    }

    Err(error.into())
}

fn build_cancel_order_params(cmd: &CancelOrder) -> CancelOrderParams {
    let order_id = cmd
        .venue_order_id
        .and_then(|id| id.inner().parse::<i64>().ok());

    if let Some(order_id) = order_id {
        CancelOrderParams::by_order_id(cmd.instrument_id.symbol.to_string(), order_id)
    } else {
        let client_id_str = encode_broker_id(&cmd.client_order_id, BINANCE_NAUTILUS_SPOT_BROKER_ID);
        CancelOrderParams::by_client_order_id(cmd.instrument_id.symbol.to_string(), client_id_str)
    }
}

fn build_cancel_replace_params(
    cmd: &ModifyOrder,
    order: &impl Order,
    quantity: Quantity,
    use_gtd: bool,
) -> anyhow::Result<CancelReplaceOrderParams> {
    let binance_side = BinanceSide::try_from(order.order_side())?;
    let binance_order_type = order_type_to_binance_spot(order.order_type(), false)?;
    let binance_tif = time_in_force_to_binance_spot(order.time_in_force(), use_gtd)?;

    let cancel_order_id: Option<i64> = cmd
        .venue_order_id
        .map(|id| {
            id.inner()
                .parse::<i64>()
                .map_err(|_| anyhow::anyhow!("Invalid venue order ID: {id}"))
        })
        .transpose()?;

    let client_id_str = encode_broker_id(&cmd.client_order_id, BINANCE_NAUTILUS_SPOT_BROKER_ID);

    Ok(CancelReplaceOrderParams {
        symbol: cmd.instrument_id.symbol.to_string(),
        side: binance_side,
        order_type: binance_order_type,
        cancel_replace_mode: BinanceCancelReplaceMode::StopOnFailure,
        time_in_force: Some(binance_tif),
        quantity: Some(quantity.to_string()),
        quote_order_qty: None,
        price: cmd.price.map(|p| p.to_string()),
        cancel_order_id,
        cancel_orig_client_order_id: if cancel_order_id.is_none() {
            Some(client_id_str.clone())
        } else {
            None
        },
        new_client_order_id: Some(client_id_str),
        stop_price: None,
        trailing_delta: None,
        iceberg_qty: None,
        new_order_resp_type: Some(BinanceOrderResponseType::Full),
        self_trade_prevention_mode: None,
    })
}

/// Dispatches a Spot execution report with tracked/untracked routing.
///
/// Tracked orders (with registered identity) produce proper order events.
/// Untracked orders fall back to execution reports for reconciliation.
#[expect(clippy::too_many_arguments)]
fn dispatch_execution_report(
    report: &BinanceSpotExecutionReport,
    emitter: &ExecutionEventEmitter,
    http_client: &BinanceSpotHttpClient,
    account_id: AccountId,
    treat_expired_as_canceled: bool,
    dispatch_state: &WsDispatchState,
    seen_trade_ids: &std::sync::Arc<Mutex<FifoCache<(Ustr, i64), 10_000>>>,
    ts_init: UnixNanos,
) {
    let symbol = report.symbol;
    let instrument_id = InstrumentId::new(symbol.into(), *BINANCE_VENUE);
    let (price_precision, size_precision) = http_client
        .get_instrument(&symbol)
        .map_or((8, 8), |i| (i.price_precision(), i.size_precision()));

    let client_order_id =
        match decode_client_order_id(&report.client_order_id, BINANCE_NAUTILUS_SPOT_BROKER_ID) {
            Ok(client_order_id) => client_order_id,
            Err(e) => {
                log::warn!("Skipping Spot execution report with invalid client order ID: {e}");
                return;
            }
        };

    let identity = dispatch_state
        .order_identities
        .get(&client_order_id)
        .map(|r| r.clone());

    if let Some(identity) = identity {
        dispatch_tracked_execution_report(
            report,
            emitter,
            account_id,
            treat_expired_as_canceled,
            dispatch_state,
            seen_trade_ids,
            client_order_id,
            &identity,
            instrument_id,
            price_precision,
            size_precision,
            ts_init,
        );
    } else {
        dispatch_untracked_execution_report(
            report,
            emitter,
            http_client,
            account_id,
            treat_expired_as_canceled,
            seen_trade_ids,
            instrument_id,
            price_precision,
            size_precision,
            ts_init,
        );
    }
}

/// Dispatches a tracked execution report as proper order events.
#[expect(clippy::too_many_arguments)]
fn dispatch_tracked_execution_report(
    report: &BinanceSpotExecutionReport,
    emitter: &ExecutionEventEmitter,
    account_id: AccountId,
    treat_expired_as_canceled: bool,
    state: &WsDispatchState,
    seen_trade_ids: &std::sync::Arc<Mutex<FifoCache<(Ustr, i64), 10_000>>>,
    client_order_id: ClientOrderId,
    identity: &OrderIdentity,
    instrument_id: InstrumentId,
    price_precision: u8,
    size_precision: u8,
    ts_init: UnixNanos,
) {
    let venue_order_id = VenueOrderId::new(report.order_id.to_string());
    let ts_event = parse_millis_or_init(report.event_time, "Spot execution event time", ts_init);

    match report.execution_type {
        BinanceSpotExecutionType::New => {
            if state.has_filled(&client_order_id) {
                log::debug!("Skipping New for already-filled {client_order_id}");
                return;
            }

            if state.has_emitted_accepted(&client_order_id) {
                // Already accepted: this New is a cancel-replace result
                let Some(price) = parse_spot_execution_report_price(
                    report,
                    &report.price,
                    price_precision,
                    "price",
                ) else {
                    return;
                };
                let Some(quantity) = parse_spot_execution_report_quantity(
                    report,
                    &report.original_qty,
                    size_precision,
                    "original_qty",
                ) else {
                    return;
                };
                let Some(stop_price) =
                    parse_spot_execution_report_decimal(report, &report.stop_price, "stop_price")
                else {
                    return;
                };
                let trigger = if stop_price > Decimal::ZERO {
                    let Some(trigger_price) = parse_spot_execution_report_price(
                        report,
                        &report.stop_price,
                        price_precision,
                        "stop_price",
                    ) else {
                        return;
                    };
                    Some(trigger_price)
                } else {
                    None
                };
                let updated = OrderUpdated::new(
                    emitter.trader_id(),
                    identity.strategy_id,
                    identity.instrument_id,
                    client_order_id,
                    quantity,
                    UUID4::new(),
                    ts_event,
                    ts_init,
                    false,
                    Some(venue_order_id),
                    Some(account_id),
                    Some(price),
                    trigger,
                    None,  // protection_price
                    false, // is_quote_quantity
                );
                emitter.send_order_event(OrderEventAny::Updated(updated));
                return;
            }
            state.insert_accepted(client_order_id);
            let accepted = OrderAccepted::new(
                emitter.trader_id(),
                identity.strategy_id,
                identity.instrument_id,
                client_order_id,
                venue_order_id,
                account_id,
                UUID4::new(),
                ts_event,
                ts_init,
                false,
            );
            emitter.send_order_event(OrderEventAny::Accepted(accepted));
        }
        BinanceSpotExecutionType::Trade => {
            let dedup_key = (report.symbol, report.trade_id);
            let mut guard = seen_trade_ids.lock();
            let is_duplicate = guard.contains(&dedup_key);
            guard.add(dedup_key);
            drop(guard);

            if is_duplicate {
                log::debug!(
                    "Duplicate trade_id={} for {}, skipping",
                    report.trade_id,
                    report.symbol
                );
                return;
            }

            ensure_accepted_emitted(
                client_order_id,
                account_id,
                venue_order_id,
                identity,
                emitter,
                state,
                ts_init,
            );

            let Some(last_qty) = parse_spot_execution_report_quantity(
                report,
                &report.last_filled_qty,
                size_precision,
                "last_filled_qty",
            ) else {
                return;
            };
            let Some(last_px) = parse_spot_execution_report_price(
                report,
                &report.last_filled_price,
                price_precision,
                "last_filled_price",
            ) else {
                return;
            };
            let Some(commission) =
                parse_spot_execution_report_decimal(report, &report.commission, "commission")
            else {
                return;
            };
            let commission_currency = report
                .commission_asset
                .as_ref()
                .map_or_else(Currency::USDT, |a| {
                    Currency::get_or_create_crypto(a.as_str())
                });
            let commission_money = match Money::from_decimal(commission, commission_currency) {
                Ok(money) => money,
                Err(e) => {
                    log::warn!(
                        "Failed to build Spot commission money for symbol={}, order_id={}, \
                        trade_id={}: {e}",
                        report.symbol,
                        report.order_id,
                        report.trade_id,
                    );
                    return;
                }
            };

            let liquidity_side = if report.is_maker {
                LiquiditySide::Maker
            } else {
                LiquiditySide::Taker
            };

            let filled = OrderFilled::new(
                emitter.trader_id(),
                identity.strategy_id,
                instrument_id,
                client_order_id,
                venue_order_id,
                account_id,
                TradeId::new(report.trade_id.to_string()),
                identity.order_side,
                identity.order_type,
                last_qty,
                last_px,
                commission_currency,
                liquidity_side,
                UUID4::new(),
                ts_event,
                ts_init,
                false,
                None,
                Some(commission_money),
                None,
            );

            state.insert_filled(client_order_id);
            emitter.send_order_event(OrderEventAny::Filled(filled));

            let cumulative_qty = parse_spot_execution_report_decimal(
                report,
                &report.cumulative_filled_qty,
                "cumulative_filled_qty",
            );
            let original_qty =
                parse_spot_execution_report_decimal(report, &report.original_qty, "original_qty");
            if let (Some(original_qty), Some(cumulative_qty)) = (original_qty, cumulative_qty)
                && original_qty <= cumulative_qty
            {
                state.cleanup_terminal(client_order_id);
            }
        }
        BinanceSpotExecutionType::Replaced => {
            // Cancel-replace succeeded: the old order is being replaced.
            // The replacement NEW event follows with the new price/qty.
            log::debug!(
                "Order replaced: client_order_id={client_order_id}, venue_order_id={venue_order_id}"
            );
        }
        BinanceSpotExecutionType::Canceled | BinanceSpotExecutionType::TradePrevention => {
            ensure_accepted_emitted(
                client_order_id,
                account_id,
                venue_order_id,
                identity,
                emitter,
                state,
                ts_init,
            );
            let canceled = OrderCanceled::new(
                emitter.trader_id(),
                identity.strategy_id,
                identity.instrument_id,
                client_order_id,
                UUID4::new(),
                ts_event,
                ts_init,
                false,
                Some(venue_order_id),
                Some(account_id),
            );
            state.cleanup_terminal(client_order_id);
            emitter.send_order_event(OrderEventAny::Canceled(canceled));
        }
        BinanceSpotExecutionType::Expired => {
            ensure_accepted_emitted(
                client_order_id,
                account_id,
                venue_order_id,
                identity,
                emitter,
                state,
                ts_init,
            );
            state.cleanup_terminal(client_order_id);

            if treat_expired_as_canceled {
                let canceled = OrderCanceled::new(
                    emitter.trader_id(),
                    identity.strategy_id,
                    identity.instrument_id,
                    client_order_id,
                    UUID4::new(),
                    ts_event,
                    ts_init,
                    false,
                    Some(venue_order_id),
                    Some(account_id),
                );
                emitter.send_order_event(OrderEventAny::Canceled(canceled));
            } else {
                let expired = OrderExpired::new(
                    emitter.trader_id(),
                    identity.strategy_id,
                    identity.instrument_id,
                    client_order_id,
                    UUID4::new(),
                    ts_event,
                    ts_init,
                    false,
                    Some(venue_order_id),
                    Some(account_id),
                );
                emitter.send_order_event(OrderEventAny::Expired(expired));
            }
        }
        BinanceSpotExecutionType::Rejected => {
            let reason = if report.reject_reason.is_empty() {
                Ustr::from("Order rejected by venue")
            } else {
                Ustr::from(&report.reject_reason)
            };
            let due_post_only = report.time_in_force == BinanceTimeInForce::Gtx
                || (report.order_type == "LIMIT_MAKER"
                    && (report.reject_reason.is_empty() || report.reject_reason == "NONE"));
            state.cleanup_terminal(client_order_id);
            emitter.emit_order_rejected_event(
                identity.strategy_id,
                identity.instrument_id,
                client_order_id,
                reason.as_str(),
                ts_init,
                due_post_only,
            );
        }
    }
}

fn parse_spot_execution_report_quantity(
    report: &BinanceSpotExecutionReport,
    raw: &str,
    precision: u8,
    field: &str,
) -> Option<Quantity> {
    match parse_required_quantity_at_precision(raw, precision, field) {
        Ok(value) => Some(value),
        Err(e) => {
            warn_invalid_spot_execution_report_field(report, field, &e);
            None
        }
    }
}

fn parse_spot_execution_report_price(
    report: &BinanceSpotExecutionReport,
    raw: &str,
    precision: u8,
    field: &str,
) -> Option<Price> {
    match parse_required_price_at_precision(raw, precision, field) {
        Ok(value) => Some(value),
        Err(e) => {
            warn_invalid_spot_execution_report_field(report, field, &e);
            None
        }
    }
}

fn parse_spot_execution_report_decimal(
    report: &BinanceSpotExecutionReport,
    raw: &str,
    field: &str,
) -> Option<Decimal> {
    match parse_required_decimal(raw, field) {
        Ok(value) => Some(value),
        Err(e) => {
            warn_invalid_spot_execution_report_field(report, field, &e);
            None
        }
    }
}

fn warn_invalid_spot_execution_report_field(
    report: &BinanceSpotExecutionReport,
    field: &str,
    error: &anyhow::Error,
) {
    log::warn!(
        "Failed to parse Spot execution report {field} for symbol={}, order_id={}, \
        trade_id={}, client_order_id={}: {error}",
        report.symbol,
        report.order_id,
        report.trade_id,
        report.client_order_id,
    );
}

/// Dispatches an untracked execution report as execution reports for reconciliation.
#[expect(clippy::too_many_arguments)]
fn dispatch_untracked_execution_report(
    report: &BinanceSpotExecutionReport,
    emitter: &ExecutionEventEmitter,
    _http_client: &BinanceSpotHttpClient,
    account_id: AccountId,
    treat_expired_as_canceled: bool,
    seen_trade_ids: &std::sync::Arc<Mutex<FifoCache<(Ustr, i64), 10_000>>>,
    instrument_id: InstrumentId,
    price_precision: u8,
    size_precision: u8,
    ts_init: UnixNanos,
) {
    match report.execution_type {
        BinanceSpotExecutionType::Trade => {
            let dedup_key = (report.symbol, report.trade_id);
            let mut guard = seen_trade_ids.lock();
            let is_duplicate = guard.contains(&dedup_key);
            guard.add(dedup_key);
            drop(guard);

            if is_duplicate {
                log::debug!(
                    "Duplicate trade_id={} for {}, skipping",
                    report.trade_id,
                    report.symbol
                );
                return;
            }

            match parse_spot_exec_report_to_order_status(
                report,
                instrument_id,
                price_precision,
                size_precision,
                account_id,
                treat_expired_as_canceled,
                ts_init,
            ) {
                Ok(status) => emitter.send_order_status_report(status),
                Err(e) => log::error!("Failed to parse order status report: {e}"),
            }

            match parse_spot_exec_report_to_fill(
                report,
                instrument_id,
                price_precision,
                size_precision,
                account_id,
                ts_init,
            ) {
                Ok(fill) => emitter.send_fill_report(fill),
                Err(e) => log::error!("Failed to parse fill report: {e}"),
            }
        }
        BinanceSpotExecutionType::New
        | BinanceSpotExecutionType::Canceled
        | BinanceSpotExecutionType::Replaced
        | BinanceSpotExecutionType::Rejected
        | BinanceSpotExecutionType::Expired
        | BinanceSpotExecutionType::TradePrevention => {
            match parse_spot_exec_report_to_order_status(
                report,
                instrument_id,
                price_precision,
                size_precision,
                account_id,
                treat_expired_as_canceled,
                ts_init,
            ) {
                Ok(status) => emitter.send_order_status_report(status),
                Err(e) => log::error!("Failed to parse order status report: {e}"),
            }
        }
    }
}

// Checks for GTX (-5022) and spot LIMIT_MAKER (-2010 + specific message)
fn is_spot_post_only_rejection(error: &BinanceSpotHttpError) -> bool {
    match error {
        BinanceSpotHttpError::BinanceError { code, message } => {
            *code == BINANCE_GTX_ORDER_REJECT_CODE
                || (*code == BINANCE_NEW_ORDER_REJECTED_CODE
                    && message == BINANCE_SPOT_POST_ONLY_REJECT_MSG)
        }
        _ => false,
    }
}

fn is_structured_venue_rejection(err: &anyhow::Error) -> bool {
    err.downcast_ref::<BinanceSpotHttpError>()
        .is_some_and(|be| matches!(be, BinanceSpotHttpError::BinanceError { .. }))
}

fn is_ambiguous_submit_error(err: &anyhow::Error) -> bool {
    err.downcast_ref::<BinanceSpotHttpError>()
        .is_some_and(|be| {
            matches!(
                be,
                BinanceSpotHttpError::BinanceError {
                    code: BINANCE_UNEXPECTED_RESPONSE_CODE | BINANCE_STATUS_UNKNOWN_CODE,
                    ..
                }
            )
        })
}

fn is_local_command_failure(err: &anyhow::Error) -> bool {
    err.downcast_ref::<BinanceSpotHttpError>()
        .is_some_and(is_local_http_command_failure)
}

fn is_local_http_command_failure(err: &BinanceSpotHttpError) -> bool {
    matches!(
        err,
        BinanceSpotHttpError::MissingCredentials | BinanceSpotHttpError::ValidationError(_)
    )
}

#[cfg(test)]
mod tests {
    use nautilus_common::messages::ExecutionEvent;
    use nautilus_core::time::get_atomic_clock_realtime;
    use nautilus_model::{
        enums::{AccountType, LiquiditySide, OrderSide},
        identifiers::{StrategyId, TraderId},
    };
    use rstest::rstest;

    use super::*;
    use crate::common::enums::BinanceEnvironment;

    #[rstest]
    #[case::live(BinanceEnvironment::Live, BINANCE_SPOT_SBE_WS_API_URL)]
    #[case::testnet(BinanceEnvironment::Testnet, BINANCE_SPOT_SBE_WS_API_TESTNET_URL)]
    #[case::demo(BinanceEnvironment::Demo, BINANCE_SPOT_SBE_WS_API_DEMO_URL)]
    fn test_resolve_ws_trading_url_uses_environment_default(
        #[case] environment: BinanceEnvironment,
        #[case] expected: &str,
    ) {
        assert_eq!(
            BinanceSpotExecutionClient::resolve_ws_trading_url(None, environment),
            expected
        );
    }

    #[rstest]
    fn test_resolve_ws_trading_url_preserves_override() {
        let expected = "wss://example.com/ws-api/v3";

        assert_eq!(
            BinanceSpotExecutionClient::resolve_ws_trading_url(
                Some(expected.to_string()),
                BinanceEnvironment::Testnet,
            ),
            expected
        );
    }

    #[rstest]
    fn test_dispatch_ws_trading_message_emits_cancel_rejected_and_clears_pending_request() {
        let tasks = TaskGroup::new();
        let task_spawner = tasks.spawner().expect("task spawner");
        let clock = get_atomic_clock_realtime();
        let (emitter, mut rx) = create_test_emitter(clock);
        let http_client = create_test_http_client(clock);
        let dispatch_state = create_tracked_dispatch_state(
            ClientOrderId::from("TEST"),
            InstrumentId::from("BTCUSDT.BINANCE"),
        );
        let ws_authenticated = tokio::sync::Notify::new();
        let ws_user_data_subscribed = tokio::sync::Notify::new();
        let (ws_setup_error_tx, _ws_setup_error_rx) = tokio::sync::mpsc::unbounded_channel();
        let seen_trade_ids = Arc::new(Mutex::new(FifoCache::new()));

        dispatch_state.pending_requests.insert(
            "req-cancel".to_string(),
            PendingRequest {
                client_order_id: ClientOrderId::from("TEST"),
                venue_order_id: Some(VenueOrderId::from("12345")),
                operation: PendingOperation::Cancel,
            },
        );

        dispatch_ws_trading_message(
            BinanceSpotWsTradingMessage::CancelRejected {
                request_id: "req-cancel".to_string(),
                code: -2011,
                msg: "Unknown order sent".to_string(),
            },
            &emitter,
            &http_client,
            AccountId::from("BINANCE-001"),
            false,
            clock,
            &dispatch_state,
            &ws_authenticated,
            &ws_user_data_subscribed,
            &ws_setup_error_tx,
            &seen_trade_ids,
            &task_spawner,
        );

        assert!(dispatch_state.pending_requests.get("req-cancel").is_none());

        match rx
            .try_recv()
            .expect("Cancel rejection event should be emitted")
        {
            ExecutionEvent::Order(OrderEventAny::CancelRejected(event)) => {
                assert_eq!(event.client_order_id, ClientOrderId::from("TEST"));
                assert_eq!(event.account_id, Some(AccountId::from("BINANCE-001")));
                assert!(event.reason.as_str().contains("code=-2011"));
            }
            other => panic!("Expected CancelRejected event, was {other:?}"),
        }
    }

    #[rstest]
    #[case(
        BINANCE_UNEXPECTED_RESPONSE_CODE,
        "An unexpected response was received from the message bus"
    )]
    #[case(
        BINANCE_STATUS_UNKNOWN_CODE,
        "Timeout waiting for response from backend server"
    )]
    fn test_dispatch_ws_trading_message_unknown_status_keeps_order_registered(
        #[case] code: i64,
        #[case] msg: &str,
    ) {
        let tasks = TaskGroup::new();
        let task_spawner = tasks.spawner().expect("task spawner");
        let clock = get_atomic_clock_realtime();
        let (emitter, mut rx) = create_test_emitter(clock);
        let http_client = create_test_http_client(clock);
        let client_order_id = ClientOrderId::from("TEST");
        let dispatch_state =
            create_tracked_dispatch_state(client_order_id, InstrumentId::from("BTCUSDT.BINANCE"));
        let ws_authenticated = tokio::sync::Notify::new();
        let ws_user_data_subscribed = tokio::sync::Notify::new();
        let (ws_setup_error_tx, _ws_setup_error_rx) = tokio::sync::mpsc::unbounded_channel();
        let seen_trade_ids = Arc::new(Mutex::new(FifoCache::new()));

        dispatch_state.pending_requests.insert(
            "req-submit".to_string(),
            PendingRequest {
                client_order_id,
                venue_order_id: None,
                operation: PendingOperation::Place,
            },
        );

        dispatch_ws_trading_message(
            BinanceSpotWsTradingMessage::OrderRejected {
                request_id: "req-submit".to_string(),
                code: code as i32,
                msg: msg.to_string(),
            },
            &emitter,
            &http_client,
            AccountId::from("BINANCE-001"),
            false,
            clock,
            &dispatch_state,
            &ws_authenticated,
            &ws_user_data_subscribed,
            &ws_setup_error_tx,
            &seen_trade_ids,
            &task_spawner,
        );

        assert!(dispatch_state.pending_requests.get("req-submit").is_none());
        assert!(
            dispatch_state
                .order_identities
                .get(&client_order_id)
                .is_some()
        );
        assert!(rx.try_recv().is_err());
    }

    #[rstest]
    fn test_dispatch_ws_trading_message_definite_submit_rejection_emits_order_rejected() {
        let tasks = TaskGroup::new();
        let task_spawner = tasks.spawner().expect("task spawner");
        let clock = get_atomic_clock_realtime();
        let (emitter, mut rx) = create_test_emitter(clock);
        let http_client = create_test_http_client(clock);
        let client_order_id = ClientOrderId::from("TEST");
        let dispatch_state =
            create_tracked_dispatch_state(client_order_id, InstrumentId::from("BTCUSDT.BINANCE"));
        let ws_authenticated = tokio::sync::Notify::new();
        let ws_user_data_subscribed = tokio::sync::Notify::new();
        let (ws_setup_error_tx, _ws_setup_error_rx) = tokio::sync::mpsc::unbounded_channel();
        let seen_trade_ids = Arc::new(Mutex::new(FifoCache::new()));

        dispatch_state.pending_requests.insert(
            "req-submit".to_string(),
            PendingRequest {
                client_order_id,
                venue_order_id: None,
                operation: PendingOperation::Place,
            },
        );

        dispatch_ws_trading_message(
            BinanceSpotWsTradingMessage::OrderRejected {
                request_id: "req-submit".to_string(),
                code: BINANCE_NEW_ORDER_REJECTED_CODE as i32,
                msg: BINANCE_SPOT_POST_ONLY_REJECT_MSG.to_string(),
            },
            &emitter,
            &http_client,
            AccountId::from("BINANCE-001"),
            false,
            clock,
            &dispatch_state,
            &ws_authenticated,
            &ws_user_data_subscribed,
            &ws_setup_error_tx,
            &seen_trade_ids,
            &task_spawner,
        );

        assert!(dispatch_state.pending_requests.get("req-submit").is_none());
        assert!(
            dispatch_state
                .order_identities
                .get(&client_order_id)
                .is_none()
        );

        match rx
            .try_recv()
            .expect("OrderRejected event should be emitted")
        {
            ExecutionEvent::Order(OrderEventAny::Rejected(event)) => {
                assert_eq!(event.client_order_id, client_order_id);
                assert_eq!(event.account_id, AccountId::from("BINANCE-001"));
                assert!(event.reason.as_str().contains("code=-2010"));
                assert!(event.due_post_only);
            }
            other => panic!("Expected OrderRejected event, was {other:?}"),
        }
    }

    #[rstest]
    fn test_dispatch_ws_trading_message_emits_modify_rejected_and_clears_pending_request() {
        let tasks = TaskGroup::new();
        let task_spawner = tasks.spawner().expect("task spawner");
        let clock = get_atomic_clock_realtime();
        let (emitter, mut rx) = create_test_emitter(clock);
        let http_client = create_test_http_client(clock);
        let dispatch_state = create_tracked_dispatch_state(
            ClientOrderId::from("TEST"),
            InstrumentId::from("BTCUSDT.BINANCE"),
        );
        let ws_authenticated = tokio::sync::Notify::new();
        let ws_user_data_subscribed = tokio::sync::Notify::new();
        let (ws_setup_error_tx, _ws_setup_error_rx) = tokio::sync::mpsc::unbounded_channel();
        let seen_trade_ids = Arc::new(Mutex::new(FifoCache::new()));

        dispatch_state.pending_requests.insert(
            "req-modify".to_string(),
            PendingRequest {
                client_order_id: ClientOrderId::from("TEST"),
                venue_order_id: Some(VenueOrderId::from("12345")),
                operation: PendingOperation::Modify,
            },
        );

        dispatch_ws_trading_message(
            BinanceSpotWsTradingMessage::CancelReplaceRejected {
                request_id: "req-modify".to_string(),
                code: -2021,
                msg: "Order cancel-replace partially failed".to_string(),
            },
            &emitter,
            &http_client,
            AccountId::from("BINANCE-001"),
            false,
            clock,
            &dispatch_state,
            &ws_authenticated,
            &ws_user_data_subscribed,
            &ws_setup_error_tx,
            &seen_trade_ids,
            &task_spawner,
        );

        assert!(dispatch_state.pending_requests.get("req-modify").is_none());

        match rx
            .try_recv()
            .expect("Modify rejection event should be emitted")
        {
            ExecutionEvent::Order(OrderEventAny::ModifyRejected(event)) => {
                assert_eq!(event.client_order_id, ClientOrderId::from("TEST"));
                assert_eq!(event.account_id, Some(AccountId::from("BINANCE-001")));
                assert!(event.reason.as_str().contains("code=-2021"));
            }
            other => panic!("Expected ModifyRejected event, was {other:?}"),
        }
    }

    fn create_test_emitter(
        clock: &'static AtomicTime,
    ) -> (
        ExecutionEventEmitter,
        tokio::sync::mpsc::UnboundedReceiver<ExecutionEvent>,
    ) {
        let mut emitter = ExecutionEventEmitter::new(
            clock,
            TraderId::from("TESTER-001"),
            AccountId::from("BINANCE-001"),
            AccountType::Cash,
            None,
        );
        let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
        emitter.set_sender(tx);
        (emitter, rx)
    }

    fn create_test_http_client(clock: &'static AtomicTime) -> BinanceSpotHttpClient {
        BinanceSpotHttpClient::new(
            BinanceEnvironment::Live,
            clock,
            None,
            None,
            None,
            None,
            None,
            None,
        )
        .expect("Test HTTP client should be created")
    }

    fn create_tracked_dispatch_state(
        client_order_id: ClientOrderId,
        instrument_id: InstrumentId,
    ) -> WsDispatchState {
        let dispatch_state = WsDispatchState::default();
        dispatch_state.order_identities.insert(
            client_order_id,
            OrderIdentity {
                instrument_id,
                strategy_id: StrategyId::from("TEST-STRATEGY"),
                order_side: OrderSide::Buy,
                order_type: OrderType::Limit,
                price: None,
                quantity: Quantity::from("1"),
                venue_position_id: None,
            },
        );
        dispatch_state
    }

    #[rstest]
    fn test_http_submit_success_defers_acceptance_to_user_stream() {
        let clock = get_atomic_clock_realtime();
        let (emitter, mut rx) = create_test_emitter(clock);
        let client_order_id = ClientOrderId::from("O-20200101-000000-000-000-0");
        let instrument_id = InstrumentId::from("ETHUSDT.BINANCE");
        let dispatch_state = Arc::new(create_tracked_dispatch_state(
            client_order_id,
            instrument_id,
        ));
        handle_spot_order_submit_success(client_order_id, VenueOrderId::from("12345678"));

        assert!(!dispatch_state.has_emitted_accepted(&client_order_id));
        assert!(rx.try_recv().is_err());
        let new_json = crate::common::testing::load_fixture_string(
            "spot/user_data_json/execution_report_new.json",
        );
        let report: BinanceSpotExecutionReport = serde_json::from_str(&new_json).unwrap();
        let identity = dispatch_state
            .order_identities
            .get(&client_order_id)
            .unwrap()
            .clone();
        dispatch_tracked_execution_report(
            &report,
            &emitter,
            AccountId::from("BINANCE-001"),
            false,
            &dispatch_state,
            &Arc::new(Mutex::new(FifoCache::new())),
            client_order_id,
            &identity,
            instrument_id,
            2,
            8,
            clock.get_time_ns(),
        );

        assert!(matches!(
            rx.try_recv(),
            Ok(ExecutionEvent::Order(OrderEventAny::Accepted(event)))
                if event.client_order_id == client_order_id
                    && event.venue_order_id == VenueOrderId::from("12345678")
        ));
        assert!(rx.try_recv().is_err());
    }

    #[rstest]
    #[case::gtx(
        BinanceSpotHttpError::BinanceError {
            code: BINANCE_GTX_ORDER_REJECT_CODE,
            message: "Order would immediately trigger.".to_string(),
        },
        true,
    )]
    #[case::spot_post_only(
        BinanceSpotHttpError::BinanceError {
            code: BINANCE_NEW_ORDER_REJECTED_CODE,
            message: BINANCE_SPOT_POST_ONLY_REJECT_MSG.to_string(),
        },
        true,
    )]
    #[case::new_order_rejected_other_message(
        BinanceSpotHttpError::BinanceError {
            code: BINANCE_NEW_ORDER_REJECTED_CODE,
            message: "Insufficient balance.".to_string(),
        },
        false,
    )]
    #[case::unrelated_code(
        BinanceSpotHttpError::BinanceError {
            code: -2011,
            message: "Unknown order sent.".to_string(),
        },
        false,
    )]
    #[case::non_binance_error(
        BinanceSpotHttpError::NetworkError("connection reset".to_string()),
        false,
    )]
    fn test_is_spot_post_only_rejection(
        #[case] error: BinanceSpotHttpError,
        #[case] expected: bool,
    ) {
        assert_eq!(is_spot_post_only_rejection(&error), expected);
    }

    #[rstest]
    #[case(BINANCE_UNEXPECTED_RESPONSE_CODE)]
    #[case(BINANCE_STATUS_UNKNOWN_CODE)]
    fn test_unknown_status_submit_error_is_ambiguous(#[case] code: i64) {
        let err = anyhow::Error::new(BinanceSpotHttpError::BinanceError {
            code,
            message: "test error".to_string(),
        });
        assert!(is_ambiguous_submit_error(&err));
        assert!(is_structured_venue_rejection(&err));
    }

    #[rstest]
    fn test_other_structured_submit_error_is_not_ambiguous() {
        let err = anyhow::Error::new(BinanceSpotHttpError::BinanceError {
            code: BINANCE_GTX_ORDER_REJECT_CODE,
            message: "test error".to_string(),
        });
        assert!(!is_ambiguous_submit_error(&err));
        assert!(is_structured_venue_rejection(&err));
    }

    #[rstest]
    fn test_dispatch_tracked_execution_report_trade_dedup() {
        let tasks = TaskGroup::new();
        let task_spawner = tasks.spawner().expect("task spawner");
        let clock = get_atomic_clock_realtime();
        let (emitter, mut rx) = create_test_emitter(clock);
        let http_client = create_test_http_client(clock);
        let client_order_id = ClientOrderId::from("x-TD67BGP9-T0000000000000");
        let dispatch_state = create_tracked_dispatch_state(
            ClientOrderId::from("O-20200101-000000-000-000-0"),
            InstrumentId::from("ETHUSDT.BINANCE"),
        );
        let ws_authenticated = tokio::sync::Notify::new();
        let ws_user_data_subscribed = tokio::sync::Notify::new();
        let (ws_setup_error_tx, _ws_setup_error_rx) = tokio::sync::mpsc::unbounded_channel();
        let seen_trade_ids = Arc::new(Mutex::new(FifoCache::new()));

        let trade_json = crate::common::testing::load_fixture_string(
            "spot/user_data_json/execution_report_trade.json",
        );
        let report: BinanceSpotExecutionReport = serde_json::from_str(&trade_json).unwrap();

        dispatch_ws_trading_message(
            BinanceSpotWsTradingMessage::ExecutionReport(Box::new(report.clone())),
            &emitter,
            &http_client,
            AccountId::from("BINANCE-001"),
            false,
            clock,
            &dispatch_state,
            &ws_authenticated,
            &ws_user_data_subscribed,
            &ws_setup_error_tx,
            &seen_trade_ids,
            &task_spawner,
        );
        dispatch_ws_trading_message(
            BinanceSpotWsTradingMessage::ExecutionReport(Box::new(report)),
            &emitter,
            &http_client,
            AccountId::from("BINANCE-001"),
            false,
            clock,
            &dispatch_state,
            &ws_authenticated,
            &ws_user_data_subscribed,
            &ws_setup_error_tx,
            &seen_trade_ids,
            &task_spawner,
        );

        let mut events = Vec::new();
        while let Ok(event) = rx.try_recv() {
            events.push(event);
        }

        let fills: Vec<_> = events
            .iter()
            .filter(|e| matches!(e, ExecutionEvent::Order(OrderEventAny::Filled(_))))
            .collect();
        assert_eq!(fills.len(), 1, "duplicate trade should be deduped");

        match fills[0] {
            ExecutionEvent::Order(OrderEventAny::Filled(fill)) => {
                assert_eq!(
                    fill.client_order_id,
                    ClientOrderId::from("O-20200101-000000-000-000-0"),
                );
                assert_eq!(fill.trade_id, TradeId::new("98765432"));
                assert_eq!(fill.liquidity_side, LiquiditySide::Maker);
            }
            _ => unreachable!(),
        }
        let _ = client_order_id;
    }

    #[rstest]
    fn test_dispatch_tracked_execution_report_invalid_fill_qty_skips_filled_event() {
        let tasks = TaskGroup::new();
        let task_spawner = tasks.spawner().expect("task spawner");
        let clock = get_atomic_clock_realtime();
        let (emitter, mut rx) = create_test_emitter(clock);
        let http_client = create_test_http_client(clock);
        let dispatch_state = create_tracked_dispatch_state(
            ClientOrderId::from("O-20200101-000000-000-000-0"),
            InstrumentId::from("ETHUSDT.BINANCE"),
        );
        let ws_authenticated = tokio::sync::Notify::new();
        let ws_user_data_subscribed = tokio::sync::Notify::new();
        let (ws_setup_error_tx, _ws_setup_error_rx) = tokio::sync::mpsc::unbounded_channel();
        let seen_trade_ids = Arc::new(Mutex::new(FifoCache::new()));

        let trade_json = crate::common::testing::load_fixture_string(
            "spot/user_data_json/execution_report_trade.json",
        );
        let mut report: BinanceSpotExecutionReport = serde_json::from_str(&trade_json).unwrap();
        report.last_filled_qty = "not-a-number".to_string();

        dispatch_ws_trading_message(
            BinanceSpotWsTradingMessage::ExecutionReport(Box::new(report)),
            &emitter,
            &http_client,
            AccountId::from("BINANCE-001"),
            false,
            clock,
            &dispatch_state,
            &ws_authenticated,
            &ws_user_data_subscribed,
            &ws_setup_error_tx,
            &seen_trade_ids,
            &task_spawner,
        );

        let mut events = Vec::new();
        while let Ok(event) = rx.try_recv() {
            events.push(event);
        }

        assert!(
            events
                .iter()
                .all(|e| !matches!(e, ExecutionEvent::Order(OrderEventAny::Filled(_)))),
            "invalid fill quantity must not emit OrderFilled",
        );
    }

    #[rstest]
    fn test_dispatch_execution_report_invalid_client_order_id_emits_nothing() {
        let clock = get_atomic_clock_realtime();
        let (emitter, mut rx) = create_test_emitter(clock);
        let http_client = create_test_http_client(clock);
        let dispatch_state = WsDispatchState::default();
        let seen_trade_ids = Arc::new(Mutex::new(FifoCache::new()));
        let json = crate::common::testing::load_fixture_string(
            "spot/user_data_json/execution_report_new.json",
        );
        let mut report: BinanceSpotExecutionReport = serde_json::from_str(&json).unwrap();
        report.client_order_id = "x-TD67BGP9-R".to_string();

        dispatch_execution_report(
            &report,
            &emitter,
            &http_client,
            AccountId::from("BINANCE-001"),
            false,
            &dispatch_state,
            &seen_trade_ids,
            clock.get_time_ns(),
        );

        assert!(rx.try_recv().is_err());
        assert!(dispatch_state.order_identities.is_empty());
    }

    #[rstest]
    #[case::as_expired(false, OrderStatus::Expired)]
    #[case::as_canceled(true, OrderStatus::Canceled)]
    fn test_normalize_spot_order_status_report_expired_respects_config(
        #[case] treat_expired_as_canceled: bool,
        #[case] expected: OrderStatus,
    ) {
        let clock = get_atomic_clock_realtime();
        let json = crate::common::testing::load_fixture_string(
            "spot/user_data_json/execution_report_expired.json",
        );
        let msg: BinanceSpotExecutionReport = serde_json::from_str(&json).unwrap();
        let mut report = parse_spot_exec_report_to_order_status(
            &msg,
            InstrumentId::from("ETHUSDT.BINANCE"),
            2,
            5,
            AccountId::from("BINANCE-001"),
            false,
            clock.get_time_ns(),
        )
        .unwrap();
        let mut reports = vec![report.clone()];

        normalize_spot_order_status_report(&mut report, treat_expired_as_canceled);
        normalize_spot_order_status_reports(&mut reports, treat_expired_as_canceled);

        assert_eq!(report.order_status, expected);
        assert_eq!(reports[0].order_status, expected);
    }

    #[rstest]
    #[case::as_expired(false)]
    #[case::as_canceled(true)]
    fn test_dispatch_tracked_execution_report_expired_respects_config(
        #[case] treat_expired_as_canceled: bool,
    ) {
        let clock = get_atomic_clock_realtime();
        let (emitter, mut rx) = create_test_emitter(clock);
        let client_order_id = ClientOrderId::from("O-20200101-000000-000-000-0");
        let instrument_id = InstrumentId::from("ETHUSDT.BINANCE");
        let dispatch_state = WsDispatchState::default();
        dispatch_state.insert_accepted(client_order_id);
        let seen_trade_ids = Arc::new(Mutex::new(FifoCache::new()));
        let identity = OrderIdentity {
            instrument_id,
            strategy_id: StrategyId::from("TEST-STRATEGY"),
            order_side: OrderSide::Buy,
            order_type: OrderType::Limit,
            price: None,
            quantity: Quantity::from("1"),
            venue_position_id: None,
        };

        let json = crate::common::testing::load_fixture_string(
            "spot/user_data_json/execution_report_expired.json",
        );
        let report: BinanceSpotExecutionReport = serde_json::from_str(&json).unwrap();

        dispatch_tracked_execution_report(
            &report,
            &emitter,
            AccountId::from("BINANCE-001"),
            treat_expired_as_canceled,
            &dispatch_state,
            &seen_trade_ids,
            client_order_id,
            &identity,
            instrument_id,
            2,
            5,
            clock.get_time_ns(),
        );

        let event = rx.try_recv().expect("terminal order event expected");
        match (treat_expired_as_canceled, event) {
            (true, ExecutionEvent::Order(OrderEventAny::Canceled(event))) => {
                assert_eq!(event.client_order_id, client_order_id);
            }
            (false, ExecutionEvent::Order(OrderEventAny::Expired(event))) => {
                assert_eq!(event.client_order_id, client_order_id);
            }
            (_, other) => panic!("Expected terminal expired/canceled event, was {other:?}"),
        }
        assert!(rx.try_recv().is_err());
    }

    #[rstest]
    fn test_dispatch_tracked_execution_report_rejected_gtx_sets_post_only() {
        let tasks = TaskGroup::new();
        let task_spawner = tasks.spawner().expect("task spawner");
        let clock = get_atomic_clock_realtime();
        let (emitter, mut rx) = create_test_emitter(clock);
        let http_client = create_test_http_client(clock);
        let client_order_id = ClientOrderId::from("O-20200101-000000-000-000-1");
        let dispatch_state =
            create_tracked_dispatch_state(client_order_id, InstrumentId::from("ETHUSDT.BINANCE"));
        let ws_authenticated = tokio::sync::Notify::new();
        let ws_user_data_subscribed = tokio::sync::Notify::new();
        let (ws_setup_error_tx, _ws_setup_error_rx) = tokio::sync::mpsc::unbounded_channel();
        let seen_trade_ids = Arc::new(Mutex::new(FifoCache::new()));

        let encoded = encode_broker_id(&client_order_id, BINANCE_NAUTILUS_SPOT_BROKER_ID);
        let report_json = format!(
            r#"{{
                "e":"executionReport","E":1709654400000,"s":"ETHUSDT",
                "c":"{encoded}","S":"BUY","o":"LIMIT","f":"GTX",
                "q":"1.00000000","p":"2500.00000000","P":"0.00000000",
                "x":"REJECTED","X":"REJECTED","r":"NONE","i":12345678,
                "l":"0.00000000","z":"0.00000000","L":"0.00000000",
                "n":"0","N":null,"T":1709654400000,"t":-1,"w":false,"m":false,
                "O":1709654400000,"Z":"0.00000000","C":""
            }}"#,
        );
        let report: BinanceSpotExecutionReport = serde_json::from_str(&report_json).unwrap();

        dispatch_ws_trading_message(
            BinanceSpotWsTradingMessage::ExecutionReport(Box::new(report)),
            &emitter,
            &http_client,
            AccountId::from("BINANCE-001"),
            false,
            clock,
            &dispatch_state,
            &ws_authenticated,
            &ws_user_data_subscribed,
            &ws_setup_error_tx,
            &seen_trade_ids,
            &task_spawner,
        );

        match rx.try_recv().expect("OrderRejected event expected") {
            ExecutionEvent::Order(OrderEventAny::Rejected(event)) => {
                assert_eq!(event.client_order_id, client_order_id);
                assert_eq!(event.account_id, AccountId::from("BINANCE-001"));
                assert!(event.due_post_only);
            }
            other => panic!("Expected OrderRejected event, was {other:?}"),
        }
    }
}