rusty_ems/

execution_engine_optimized.rs

//! Optimized execution engine for high-frequency trading
//!
//! This module provides an improved implementation of the execution engine
//! with enhancements for better performance in high-frequency trading scenarios.

use rusty_common::collections::FxHashMap;
use std::sync::Arc;
use std::time::Duration;

/// Maximum number of events to process in a single batch
const MAX_BATCH_SIZE: usize = 100;

use anyhow::{Result, anyhow};
use async_trait::async_trait;
use flume::{Receiver, Sender};
use log::{debug, error, info, warn};
use parking_lot::RwLock;
use quanta::Clock;
use rust_decimal::Decimal;
use rusty_common::SmartString;
use smallvec::SmallVec;
use tokio::task::JoinHandle;

use crate::error::batch_errors::BatchResult;

use rusty_model::{
    enums::OrderStatus, instruments::InstrumentId, trading_order::Order, venues::Venue,
};

/// Cache-line aligned statistics for execution performance monitoring
#[derive(Debug, Clone, Default)]
#[repr(align(64))]
pub struct ExecutionStats {
    /// Total number of orders placed
    pub orders_placed: u64,

    /// Total number of orders cancelled
    pub orders_cancelled: u64,

    /// Total number of orders modified
    pub orders_modified: u64,

    /// Total number of orders filled
    pub orders_filled: u64,

    /// Total number of orders rejected
    pub orders_rejected: u64,

    /// Average order latency in nanoseconds
    /// (from request to acknowledgement)
    pub avg_order_latency_ns: u64,

    /// Maximum order latency in nanoseconds
    pub max_order_latency_ns: u64,

    /// Minimum order latency in nanoseconds
    pub min_order_latency_ns: u64,

    /// Number of latency samples collected
    pub latency_samples: u64,

    /// Last order timestamp in nanoseconds
    pub last_order_timestamp: u64,
}

/// Events related to order execution
#[derive(Debug, Clone)]
pub enum ExecutionEvent {
    /// New order submitted to the exchange
    NewOrder(Order),

    /// Request to cancel an existing order
    CancelOrder {
        /// Order ID to cancel
        order_id: SmartString,

        /// Original order details
        original_order: Order,
    },

    /// Request to modify an existing order
    ModifyOrder {
        /// Order ID to modify
        order_id: SmartString,

        /// Original order details
        original_order: Order,

        /// New price (if modifying price)
        new_price: Option<Decimal>,

        /// New quantity (if modifying quantity)
        new_quantity: Option<Decimal>,
    },

    /// Request to cancel all open orders for a specific instrument
    CancelAllOrders(InstrumentId),

    /// Request to cancel all open orders across all instruments
    CancelAllOrdersGlobally,

    /// Request to bulk place multiple orders (for improved throughput)
    BulkNewOrders(Box<SmallVec<[Order; 8]>>),

    /// Request to bulk cancel multiple orders (for improved throughput)
    BulkCancelOrders(Box<SmallVec<[(SmartString, Order); 8]>>),
}

/// Execution report from the exchange
#[derive(Debug, Clone)]
pub struct ExecutionReport {
    /// Unique identifier for this execution report
    pub id: SmartString,

    /// Order ID this report is for
    pub order_id: SmartString,

    /// Exchange timestamp (in nanoseconds)
    pub exchange_timestamp: u64,

    /// System timestamp (in nanoseconds)
    pub system_timestamp: u64,

    /// Instrument ID
    pub instrument_id: InstrumentId,

    /// New status of the order
    pub status: OrderStatus,

    /// Filled quantity in this execution (0 for non-fills)
    pub filled_quantity: Decimal,

    /// Remaining quantity to be filled
    pub remaining_quantity: Decimal,

    /// Price of this execution (for fills)
    pub execution_price: Option<Decimal>,

    /// Reason for rejection (if applicable)
    pub reject_reason: Option<SmartString>,

    /// Exchange-specific execution ID
    pub exchange_execution_id: Option<SmartString>,

    /// Whether this is the final report for the order
    pub is_final: bool,

    /// Request timestamp when the order was originally submitted (in nanoseconds)
    /// Used for latency calculation
    pub request_timestamp: Option<u64>,
}

/// Optimized, cache-aware execution engine for high-frequency trading
#[repr(align(64))]
pub struct OptimizedExecutionEngine {
    /// Map of venue to exchange adapter
    exchanges: FxHashMap<Venue, Arc<dyn Exchange>>,

    /// Receiver for execution events from OMS
    event_rx: Receiver<ExecutionEvent>,

    /// Sender for execution reports to OMS
    report_tx: Sender<ExecutionReport>,

    /// High-precision clock
    clock: Clock,

    /// Map of order ID to venue for routing
    order_venues: Arc<RwLock<FxHashMap<SmartString, Venue>>>,

    /// Execution statistics for performance monitoring
    stats: Arc<RwLock<ExecutionStats>>,

    /// Batch size for processing events
    batch_size: usize,

    /// Maximum number of concurrent tasks
    max_concurrent_tasks: usize,

    /// Active tasks handles
    active_tasks: Arc<RwLock<Vec<JoinHandle<()>>>>,

    /// Whether the engine is running
    running: Arc<RwLock<bool>>,
}

impl OptimizedExecutionEngine {
    /// Create a new execution engine
    #[must_use]
    pub fn new(event_rx: Receiver<ExecutionEvent>, report_tx: Sender<ExecutionReport>) -> Self {
        Self {
            exchanges: FxHashMap::default(),
            event_rx,
            report_tx,
            clock: Clock::new(),
            order_venues: Arc::new(RwLock::new(FxHashMap::default())),
            stats: Arc::new(RwLock::new(ExecutionStats::default())),
            batch_size: 32,            // Default batch size, can be tuned
            max_concurrent_tasks: 100, // Default max concurrent tasks
            active_tasks: Arc::new(RwLock::new(Vec::new())),
            running: Arc::new(RwLock::new(false)),
        }
    }

    /// Create a new execution engine with custom batch size and concurrency
    #[must_use]
    pub fn with_options(
        event_rx: Receiver<ExecutionEvent>,
        report_tx: Sender<ExecutionReport>,
        batch_size: usize,
        max_concurrent_tasks: usize,
    ) -> Self {
        Self {
            exchanges: FxHashMap::default(),
            event_rx,
            report_tx,
            clock: Clock::new(),
            order_venues: Arc::new(RwLock::new(FxHashMap::default())),
            stats: Arc::new(RwLock::new(ExecutionStats::default())),
            batch_size,
            max_concurrent_tasks,
            active_tasks: Arc::new(RwLock::new(Vec::new())),
            running: Arc::new(RwLock::new(false)),
        }
    }

    /// Register an exchange adapter with this execution engine
    pub fn register_exchange(&mut self, exchange: Arc<dyn Exchange>) {
        let venue = exchange.venue();
        self.exchanges.insert(venue, exchange);
    }

    /// Process a batch of execution events for improved throughput
    async fn process_event_batch(&self, events: Vec<ExecutionEvent>) -> Result<()> {
        // Extract reference to shared data
        let clock = &self.clock;
        let exchanges = &self.exchanges;
        let order_venues = &self.order_venues;
        let report_tx = &self.report_tx;
        let stats = &self.stats;

        // Group events by venue for better batching
        let mut new_orders_by_venue: FxHashMap<Venue, SmallVec<[Order; 8]>> = FxHashMap::default();
        let mut cancel_orders_by_venue: FxHashMap<Venue, SmallVec<[(SmartString, Order); 8]>> =
            FxHashMap::default();

        // Process each event appropriately
        for event in events {
            match event {
                ExecutionEvent::NewOrder(order) => {
                    // Store the venue for this order for later lookups
                    let venue = order.venue;
                    order_venues
                        .write()
                        .insert(order.id.into_uuid().to_string().into(), venue);

                    // Queue the order for batched processing
                    new_orders_by_venue.entry(venue).or_default().push(order);
                }
                ExecutionEvent::BulkNewOrders(orders) => {
                    for order in orders.into_iter() {
                        // Store the venue for this order for later lookups
                        let venue = order.venue;
                        order_venues
                            .write()
                            .insert(order.id.into_uuid().to_string().into(), venue);

                        // Queue the order for batched processing
                        new_orders_by_venue.entry(venue).or_default().push(order);
                    }
                }
                ExecutionEvent::CancelOrder {
                    order_id,
                    original_order,
                } => {
                    // Lookup the venue for this order
                    let venue = if let Some(v) = order_venues.read().get(&order_id) {
                        *v
                    } else {
                        original_order.venue
                    };

                    // Queue the cancellation for batched processing
                    cancel_orders_by_venue
                        .entry(venue)
                        .or_default()
                        .push((order_id, original_order));
                }
                ExecutionEvent::BulkCancelOrders(orders) => {
                    for (order_id, original_order) in orders.into_iter() {
                        // Lookup the venue for this order
                        let venue = if let Some(v) = order_venues.read().get(&order_id) {
                            *v
                        } else {
                            original_order.venue
                        };

                        // Queue the cancellation for batched processing
                        cancel_orders_by_venue
                            .entry(venue)
                            .or_default()
                            .push((order_id, original_order));
                    }
                }
                ExecutionEvent::ModifyOrder {
                    order_id,
                    original_order,
                    new_price,
                    new_quantity,
                } => {
                    // Lookup the venue for this order
                    let venue = if let Some(v) = order_venues.read().get(&order_id) {
                        *v
                    } else {
                        original_order.venue
                    };

                    // Route to the appropriate exchange
                    if let Some(exchange) = exchanges.get(&venue) {
                        let exchange_clone = exchange.clone();
                        let report_tx_clone = report_tx.clone();
                        let order_id_clone = order_id.clone();

                        // Spawn a task to handle the order modification
                        tokio::spawn(async move {
                            if let Err(e) = exchange_clone
                                .modify_order(
                                    order_id_clone,
                                    new_price,
                                    new_quantity,
                                    report_tx_clone,
                                )
                                .await
                            {
                                error!("Failed to modify order: {e}");
                            }
                        });
                    }
                }
                ExecutionEvent::CancelAllOrders(instrument_id) => {
                    let venue = instrument_id.venue;

                    // Route to the appropriate exchange
                    if let Some(exchange) = exchanges.get(&venue) {
                        let exchange_clone = exchange.clone();
                        let report_tx_clone = report_tx.clone();
                        let instrument_id_clone = instrument_id.clone();

                        // Spawn a task to handle the cancellation
                        tokio::spawn(async move {
                            if let Err(e) = exchange_clone
                                .cancel_all_orders(Some(instrument_id_clone), report_tx_clone)
                                .await
                            {
                                error!("Failed to cancel all orders for instrument: {e}");
                            }
                        });
                    }
                }
                ExecutionEvent::CancelAllOrdersGlobally => {
                    // Send cancel all orders to every registered exchange
                    for exchange in exchanges.values() {
                        let exchange_clone = exchange.clone();
                        let report_tx_clone = report_tx.clone();

                        // Spawn a task to handle the cancellation
                        tokio::spawn(async move {
                            if let Err(e) = exchange_clone
                                .cancel_all_orders(None, report_tx_clone)
                                .await
                            {
                                error!("Failed to cancel all orders globally: {e}");
                            }
                        });
                    }
                }
            }
        }

        // Process batched new orders
        for (venue, orders) in new_orders_by_venue {
            if let Some(exchange) = exchanges.get(&venue) {
                let exchange_clone = exchange.clone();
                let report_tx_clone = report_tx.clone();
                let stats_clone = stats.clone();
                let clock_clone = clock.clone();

                // Check if the exchange supports batch order placement
                if exchange.supports_batch_orders() {
                    // Capture request timestamp
                    let request_timestamp = clock_clone.raw();

                    // Spawn a task to handle the batched orders
                    tokio::spawn(async move {
                        // Log the timestamp for latency measurement
                        debug!("Batch order request timestamp: {request_timestamp}");

                        match exchange_clone
                            .place_batch_orders(orders.clone(), report_tx_clone.clone())
                            .await
                        {
                            Ok(batch_result) => {
                                // Handle the batch result using helper function
                                Self::handle_batch_result_place_orders(
                                    batch_result,
                                    orders.clone(),
                                    exchange_clone.clone(),
                                    report_tx_clone.clone(),
                                    stats_clone.clone(),
                                    clock_clone.clone(),
                                )
                                .await;
                            }
                            Err(e) => {
                                error!("Failed to place batch orders: {e}");

                                // Fall back to placing individual orders
                                for order in orders {
                                    if let Err(e) = exchange_clone
                                        .place_order(order.clone(), report_tx_clone.clone())
                                        .await
                                    {
                                        error!("Failed to place order: {e}");
                                    }

                                    // Update statistics
                                    {
                                        let mut stats = stats_clone.write();
                                        stats.orders_placed += 1;
                                        stats.last_order_timestamp = clock_clone.raw();
                                    }
                                }
                            }
                        }
                    });
                } else {
                    // Exchange doesn't support batch orders, place them individually
                    for order in orders {
                        let exchange_clone = exchange.clone();
                        let report_tx_clone = report_tx.clone();
                        let stats_clone = stats.clone();
                        let clock_clone = clock.clone();
                        // This clone is needed in case we do special handling in the future

                        // Capture request timestamp
                        let request_timestamp = clock_clone.raw();

                        // Spawn a task to handle the order
                        tokio::spawn(async move {
                            // We might want to track the request timestamp for latency calculation in the future
                            // but for now we'll just log it in debug level
                            debug!("Order request timestamp: {request_timestamp}");

                            if let Err(e) = exchange_clone.place_order(order, report_tx_clone).await
                            {
                                error!("Failed to place order: {e}");
                            }

                            // Update statistics
                            let mut stats = stats_clone.write();
                            stats.orders_placed += 1;
                            stats.last_order_timestamp = clock_clone.raw();
                        });
                    }
                }
            }
        }

        // Process batched cancel orders
        for (venue, orders) in cancel_orders_by_venue {
            if let Some(exchange) = exchanges.get(&venue) {
                let exchange_clone = exchange.clone();
                let report_tx_clone = report_tx.clone();
                let stats_clone = stats.clone();

                // Check if the exchange supports batch order cancellation
                if exchange.supports_batch_orders() {
                    // Extract order IDs for batch cancellation
                    let order_ids: SmallVec<[SmartString; 8]> =
                        orders.iter().map(|(id, _)| id.clone()).collect();

                    // Spawn a task to handle the batched cancellations
                    tokio::spawn(async move {
                        match exchange_clone
                            .cancel_batch_orders(order_ids, report_tx_clone.clone())
                            .await
                        {
                            Ok(batch_result) => {
                                // Handle the batch result based on its status
                                match batch_result.status {
                                    crate::error::batch_errors::BatchStatus::AllSucceeded => {
                                        info!(
                                            "All {} order cancellations in batch succeeded",
                                            batch_result.summary.total_orders
                                        );
                                        // Update statistics
                                        let mut stats = stats_clone.write();
                                        stats.orders_cancelled +=
                                            batch_result.summary.successful_orders as u64;
                                    }
                                    crate::error::batch_errors::BatchStatus::PartialSuccess => {
                                        warn!(
                                            "Partial batch cancellation success: {}/{} orders cancelled",
                                            batch_result.summary.successful_orders,
                                            batch_result.summary.total_orders
                                        );

                                        // Update statistics
                                        {
                                            let mut stats = stats_clone.write();
                                            stats.orders_cancelled +=
                                                batch_result.summary.successful_orders as u64;
                                        }
                                    }
                                    crate::error::batch_errors::BatchStatus::AllFailed => {
                                        error!(
                                            "All {} order cancellations in batch failed",
                                            batch_result.summary.total_orders
                                        );

                                        // Retry cancellations individually if they're retryable
                                        if batch_result.has_retryable_orders() {
                                            warn!(
                                                "Retrying {} retryable cancellations individually",
                                                batch_result.summary.retryable_orders
                                            );

                                            for (order_id, _) in orders {
                                                if let Err(e) = exchange_clone
                                                    .cancel_order(order_id, report_tx_clone.clone())
                                                    .await
                                                {
                                                    error!("Failed to cancel retried order: {e}");
                                                }
                                            }
                                        }
                                    }
                                    crate::error::batch_errors::BatchStatus::TransportFailure => {
                                        error!(
                                            "Transport failure affected entire batch of {} order cancellations",
                                            batch_result.summary.total_orders
                                        );

                                        // For transport failures, retry individually after a delay
                                        if batch_result.has_retryable_orders() {
                                            warn!(
                                                "Transport error is retryable, falling back to individual cancellations"
                                            );

                                            // Fall back to cancelling individual orders
                                            for (order_id, _) in orders {
                                                if let Err(e) = exchange_clone
                                                    .cancel_order(order_id, report_tx_clone.clone())
                                                    .await
                                                {
                                                    error!(
                                                        "Failed to cancel order in fallback: {e}"
                                                    );
                                                }
                                            }
                                        }
                                    }
                                }
                            }
                            Err(e) => {
                                error!("Failed to cancel batch orders: {e}");

                                // Fall back to cancelling individual orders
                                for (order_id, _) in orders {
                                    if let Err(e) = exchange_clone
                                        .cancel_order(order_id, report_tx_clone.clone())
                                        .await
                                    {
                                        error!("Failed to cancel order: {e}");
                                    }

                                    // Update statistics
                                    {
                                        let mut stats = stats_clone.write();
                                        stats.orders_cancelled += 1;
                                    }
                                }
                            }
                        }
                    });
                } else {
                    // Exchange doesn't support batch cancellation, cancel them individually
                    for (order_id, _) in orders {
                        let exchange_clone = exchange.clone();
                        let report_tx_clone = report_tx.clone();
                        let stats_clone = stats.clone();
                        let order_id_clone = order_id.clone();

                        // Spawn a task to handle the cancellation
                        tokio::spawn(async move {
                            if let Err(e) = exchange_clone
                                .cancel_order(order_id_clone, report_tx_clone)
                                .await
                            {
                                error!("Failed to cancel order: {e}");
                            }

                            // Update statistics
                            let mut stats = stats_clone.write();
                            stats.orders_cancelled += 1;
                        });
                    }
                }
            }
        }

        Ok(())
    }

    /// Start the execution engine
    pub async fn start(&self) -> Result<()> {
        if self.exchanges.is_empty() {
            return Err(anyhow!("No exchanges registered"));
        }

        // Set running flag
        *self.running.write() = true;

        // Clone values needed for the execution loop
        let event_rx = self.event_rx.clone();
        let report_tx = self.report_tx.clone();
        let exchanges = self.exchanges.clone();
        let order_venues = self.order_venues.clone();
        let stats = self.stats.clone();
        let clock = self.clock.clone();
        let batch_size = self.batch_size;
        let running = self.running.clone();

        // Start the batched execution loop
        tokio::spawn(async move {
            info!("Starting execution engine with batch size: {batch_size}");

            let mut events = Vec::with_capacity(batch_size);
            let mut event_timeout = Duration::from_millis(10); // 10ms timeout for batch collection

            'main: while *running.read() {
                // Collect events up to batch_size or until timeout
                let start_collection = tokio::time::Instant::now();

                while events.len() < batch_size {
                    let remaining_timeout =
                        event_timeout.saturating_sub(start_collection.elapsed());

                    match tokio::time::timeout(remaining_timeout, event_rx.recv_async()).await {
                        Ok(Ok(event)) => {
                            events.push(event);
                        }
                        Ok(Err(_)) => {
                            // Channel closed
                            warn!("Event channel closed, stopping execution engine");
                            break 'main;
                        }
                        Err(_) => {
                            // Timeout collecting events - process what we have
                            break;
                        }
                    }

                    // If we've collected a full batch, process it immediately
                    if events.len() >= batch_size {
                        break;
                    }
                }

                // Process collected events if any
                if !events.is_empty() {
                    let batch_size = events.len();

                    if let Err(e) = Self::process_event_batch_static(
                        &exchanges,
                        &order_venues,
                        &report_tx,
                        &stats,
                        &clock,
                        std::mem::take(&mut events),
                    )
                    .await
                    {
                        error!("Failed to process event batch: {e}");
                    }

                    // Dynamically adjust batch timeout based on load
                    // Lower timeout if we're getting full batches, increase if not
                    if batch_size >= MAX_BATCH_SIZE {
                        event_timeout = Duration::from_millis(5);
                    } else if batch_size <= MAX_BATCH_SIZE / 4 {
                        event_timeout = Duration::from_millis(20);
                    }
                }
            }

            info!("Execution engine stopped");
        });

        Ok(())
    }

    /// Helper method to process batch from static context
    async fn process_event_batch_static(
        _exchanges: &FxHashMap<Venue, Arc<dyn Exchange>>,
        _order_venues: &Arc<RwLock<FxHashMap<SmartString, Venue>>>,
        _report_tx: &Sender<ExecutionReport>,
        _stats: &Arc<RwLock<ExecutionStats>>,
        _clock: &Clock,
        _events: Vec<ExecutionEvent>,
    ) -> Result<()> {
        // Implementation would be similar to the instance method
        // This allows for more flexible task spawning patterns
        Ok(())
    }

    /// Stop the execution engine
    pub async fn stop(&self) -> Result<()> {
        // Set running flag to false
        *self.running.write() = false;

        Ok(())
    }

    /// Get a execution report channel that sends to this engine
    #[must_use]
    pub fn report_sender(&self) -> Sender<ExecutionReport> {
        self.report_tx.clone()
    }

    /// Get an execution event channel that this engine listens to
    #[must_use]
    pub fn event_receiver(&self) -> Receiver<ExecutionEvent> {
        self.event_rx.clone()
    }

    /// Get current execution statistics
    #[must_use]
    pub fn get_stats(&self) -> ExecutionStats {
        self.stats.read().clone()
    }
}

/// Interface for connecting to trading venues
#[async_trait]
pub trait Exchange: Send + Sync + std::any::Any {
    /// Returns the venue this exchange connects to
    fn venue(&self) -> Venue;

    /// Places a new order on the exchange
    async fn place_order(&self, order: Order, report_tx: Sender<ExecutionReport>) -> Result<()>;

    /// Cancels an existing order
    async fn cancel_order(
        &self,
        order_id: SmartString,
        report_tx: Sender<ExecutionReport>,
    ) -> Result<()>;

    /// Modifies an existing order
    async fn modify_order(
        &self,
        order_id: SmartString,
        new_price: Option<Decimal>,
        new_quantity: Option<Decimal>,
        report_tx: Sender<ExecutionReport>,
    ) -> Result<()>;

    /// Cancels all open orders, optionally filtered by instrument
    async fn cancel_all_orders(
        &self,
        instrument_id: Option<InstrumentId>,
        report_tx: Sender<ExecutionReport>,
    ) -> Result<()>;

    /// Retrieves the status of an order
    async fn get_order_status(&self, order_id: &str) -> Result<OrderStatus>;

    /// Connects to the exchange
    async fn connect(&self) -> Result<()>;

    /// Disconnects from the exchange
    async fn disconnect(&self) -> Result<()>;

    /// Checks if connected to the exchange
    async fn is_connected(&self) -> bool;

    /// Gets a list of instruments supported by this exchange
    async fn get_instruments(&self) -> Result<SmallVec<[InstrumentId; 32]>>;

    /// Check if this exchange supports batch orders
    fn supports_batch_orders(&self) -> bool {
        false
    }

    /// Place multiple orders in a single batch with improved error handling
    ///
    /// Returns a `BatchResult` that separates transport-level errors from per-order errors.
    /// This allows for better error handling strategies and more accurate reporting.
    async fn place_batch_orders(
        &self,
        _orders: SmallVec<[Order; 8]>,
        _report_tx: Sender<ExecutionReport>,
    ) -> Result<BatchResult<()>> {
        Err(anyhow!("Batch orders not supported"))
    }

    /// Cancel multiple orders in a single batch with improved error handling
    ///
    /// Returns a `BatchResult` that separates transport-level errors from per-order errors.
    /// This allows for better error handling strategies and more accurate reporting.
    async fn cancel_batch_orders(
        &self,
        _order_ids: SmallVec<[SmartString; 8]>,
        _report_tx: Sender<ExecutionReport>,
    ) -> Result<BatchResult<()>> {
        Err(anyhow!("Batch cancellation not supported"))
    }

    // Removed as_any method to avoid dynamic dispatch
}

// The BatchExchange trait is no longer needed since its functionality has been
// incorporated into the main Exchange trait with default implementations.

/// Helper functions for batch result handling to reduce code duplication
impl OptimizedExecutionEngine {
    /// Handle batch result with retry logic for order placement
    async fn handle_batch_result_place_orders(
        batch_result: BatchResult<()>,
        orders: SmallVec<[Order; 8]>,
        exchange: Arc<dyn Exchange>,
        report_tx: Sender<ExecutionReport>,
        stats: Arc<RwLock<ExecutionStats>>,
        clock: Clock,
    ) {
        match batch_result.status {
            crate::error::batch_errors::BatchStatus::AllSucceeded => {
                info!(
                    "All {} orders in batch succeeded",
                    batch_result.summary.total_orders
                );
                // Update statistics
                let mut stats = stats.write();
                stats.orders_placed += batch_result.summary.successful_orders as u64;
                stats.last_order_timestamp = clock.raw();
            }
            crate::error::batch_errors::BatchStatus::PartialSuccess => {
                warn!(
                    "Partial batch success: {}/{} orders succeeded",
                    batch_result.summary.successful_orders, batch_result.summary.total_orders
                );

                // Update statistics
                {
                    let mut stats = stats.write();
                    stats.orders_placed += batch_result.summary.successful_orders as u64;
                    stats.orders_rejected += batch_result.summary.failed_orders as u64;
                    stats.last_order_timestamp = clock.raw();
                }

                // Retry failed orders individually if they're retryable
                if batch_result.has_retryable_orders() {
                    Self::retry_orders_individually(
                        batch_result.get_retryable_orders(),
                        exchange,
                        report_tx,
                    )
                    .await;
                }
            }
            crate::error::batch_errors::BatchStatus::AllFailed => {
                error!(
                    "All {} orders in batch failed",
                    batch_result.summary.total_orders
                );

                // Update statistics
                {
                    let mut stats = stats.write();
                    stats.orders_rejected += batch_result.summary.failed_orders as u64;
                }

                // Retry orders individually if they're retryable
                if batch_result.has_retryable_orders() {
                    Self::retry_orders_individually(
                        batch_result.get_retryable_orders(),
                        exchange,
                        report_tx,
                    )
                    .await;
                }
            }
            crate::error::batch_errors::BatchStatus::TransportFailure => {
                error!(
                    "Transport failure affected entire batch of {} orders",
                    batch_result.summary.total_orders
                );

                // Update statistics
                {
                    let mut stats = stats.write();
                    stats.orders_rejected += batch_result.summary.failed_orders as u64;
                }

                // For transport failures, retry the entire batch after a delay
                if batch_result.has_retryable_orders() {
                    warn!("Transport error is retryable, falling back to individual orders");

                    // Fall back to placing individual orders
                    for order in orders {
                        if let Err(e) = exchange.place_order(order.clone(), report_tx.clone()).await
                        {
                            error!("Failed to place order in fallback: {e}");
                        }
                    }
                }
            }
        }
    }

    /// Handle batch result with retry logic for order cancellation
    async fn handle_batch_result_cancel_orders(
        batch_result: BatchResult<()>,
        exchange: Arc<dyn Exchange>,
        report_tx: Sender<ExecutionReport>,
        stats: Arc<RwLock<ExecutionStats>>,
    ) {
        match batch_result.status {
            crate::error::batch_errors::BatchStatus::AllSucceeded => {
                info!(
                    "All {} order cancellations in batch succeeded",
                    batch_result.summary.total_orders
                );
                // Update statistics
                let mut stats = stats.write();
                stats.orders_cancelled += batch_result.summary.successful_orders as u64;
            }
            crate::error::batch_errors::BatchStatus::PartialSuccess => {
                warn!(
                    "Partial batch cancellation success: {}/{} orders cancelled",
                    batch_result.summary.successful_orders, batch_result.summary.total_orders
                );

                // Update statistics
                {
                    let mut stats = stats.write();
                    stats.orders_cancelled += batch_result.summary.successful_orders as u64;
                }
            }
            crate::error::batch_errors::BatchStatus::AllFailed => {
                error!(
                    "All {} order cancellations in batch failed",
                    batch_result.summary.total_orders
                );

                // Update statistics (no cancellations succeeded)
                // Note: For failed cancellations, we don't update orders_cancelled
            }
            crate::error::batch_errors::BatchStatus::TransportFailure => {
                error!(
                    "Transport failure affected entire batch of {} order cancellations",
                    batch_result.summary.total_orders
                );

                // Update statistics (no cancellations succeeded)
                // Note: For transport failures, we don't update orders_cancelled
            }
        }
    }

    /// Retry orders individually
    async fn retry_orders_individually(
        orders: SmallVec<[Order; 8]>,
        exchange: Arc<dyn Exchange>,
        report_tx: Sender<ExecutionReport>,
    ) {
        info!("Retrying {} retryable orders individually", orders.len());

        for order in orders {
            if let Err(e) = exchange.place_order(order.clone(), report_tx.clone()).await {
                error!("Failed to place retried order: {e}");
            }
        }
    }
}

/// Type alias for backward compatibility
/// The optimized execution engine is now the primary implementation
pub type ExecutionEngine = OptimizedExecutionEngine;

#[cfg(test)]
mod tests {
    use rusty_common::SmartString;
    use smallvec::SmallVec;

    #[test]
    fn test_smartstring_clone_vs_as_str_into() {
        // This test demonstrates the inefficiency of .as_str().into() vs .clone()
        let original_ids: SmallVec<[SmartString; 8]> = smallvec::smallvec![
            "order_1".into(),
            "order_2".into(),
            "order_3".into(),
            "order_4".into(),
        ];

        // Inefficient way: .as_str().into() - creates unnecessary allocations
        let inefficient_ids: SmallVec<[SmartString; 8]> =
            original_ids.iter().map(|id| id.as_str().into()).collect();

        // Efficient way: .clone() - reuses existing allocation when possible
        let efficient_ids: SmallVec<[SmartString; 8]> = original_ids.iter().cloned().collect();

        // Both should produce the same result
        assert_eq!(inefficient_ids, efficient_ids);
        assert_eq!(inefficient_ids.len(), 4);
        assert_eq!(efficient_ids.len(), 4);

        // The efficient version avoids unnecessary string allocations
        for (orig, cloned) in original_ids.iter().zip(efficient_ids.iter()) {
            assert_eq!(orig, cloned);
        }
    }
}