rusty_backtest/

parallel.rs

//! Parallel L2 Backtesting - Multi-symbol parallel execution with rayon
//!
//! Provides efficient parallel backtesting across multiple symbols using
//! rayon's work-stealing thread pool for optimal CPU utilization.

use crate::engine::{
    BacktestConfig, BacktestStats, DefaultBacktestEngine, MarketDataEvent, Strategy,
};
use crate::latency::LatencyModel;
use crate::matching::QueueModel;
use parking_lot::{Mutex, RwLock};
use rayon::prelude::*;
use rust_decimal::Decimal;
use rusty_common::SmartString;
use rusty_common::collections::{FxHashMap, SmallStrategyVec, SmallSymbolVec};
use smallvec::SmallVec;
// HashMap import removed - using FxHashMap
use std::sync::Arc;

/// Parallel strategy wrapper that clones strategies for each symbol
pub trait ParallelStrategy: Send + Sync {
    /// Clone the strategy for a specific symbol
    fn clone_for_symbol(&self, symbol: &str) -> Box<dyn Strategy>;

    /// Called after all backtests complete with results
    fn on_completion(&mut self, results: &FxHashMap<SmartString, BacktestStats>);
}

/// Configuration for parallel L2 backtesting
pub struct ParallelConfig {
    /// Start time of the backtest in nanoseconds
    pub start_time_ns: u64,
    /// End time of the backtest in nanoseconds
    pub end_time_ns: u64,
    /// Queue position model to use (FIFO, ProRata, etc.)
    pub queue_model: QueueModel,
    /// Whether to allow partial order fills
    pub allow_partial_fills: bool,
    /// Latency model for order submission and responses
    pub order_latency: Box<dyn LatencyModel>,
    /// Latency model for market data delivery
    pub market_data_latency: Box<dyn LatencyModel>,
}

/// Symbol-specific configuration
#[derive(Clone)]
pub struct SymbolConfig {
    /// Trading symbol/instrument identifier
    pub symbol: SmartString,
    /// Minimum price increment for this symbol
    pub tick_size: Decimal,
    /// Minimum order quantity increment for this symbol
    pub lot_size: Decimal,
}

/// Market data provider trait
pub trait MarketDataProvider: Send + Sync {
    /// Get market data events for a symbol
    fn get_events(
        &self,
        symbol: &str,
        start_ns: u64,
        end_ns: u64,
    ) -> SmallVec<[(u64, MarketDataEvent); 256]>;
}

/// Parallel L2 backtesting engine
pub struct ParallelEngine {
    config: ParallelConfig,
    symbols: SmallSymbolVec<SymbolConfig>,
    strategies: Arc<RwLock<SmallStrategyVec<Box<dyn ParallelStrategy>>>>,
    data_provider: Arc<dyn MarketDataProvider>,
}

impl ParallelEngine {
    /// Create a new parallel L2 engine
    #[must_use]
    pub fn new(
        config: ParallelConfig,
        symbols: SmallSymbolVec<SymbolConfig>,
        data_provider: Arc<dyn MarketDataProvider>,
    ) -> Self {
        Self {
            config,
            symbols,
            strategies: Arc::new(RwLock::new(SmallStrategyVec::new())),
            data_provider,
        }
    }

    /// Add a parallel strategy
    pub fn add_strategy(&self, strategy: Box<dyn ParallelStrategy>) {
        self.strategies.write().push(strategy);
    }

    /// Run parallel backtests
    #[must_use]
    pub fn run(&self) -> FxHashMap<SmartString, BacktestStats> {
        // Create engines for each symbol in parallel
        let results: Vec<(SmartString, BacktestStats)> = self
            .symbols
            .par_iter()
            .map(|symbol_config| self.run_single_symbol(symbol_config))
            .collect();

        // Convert to hashmap
        let results_map: FxHashMap<SmartString, BacktestStats> = results.into_iter().collect();

        // Notify strategies of completion
        let mut strategies = self.strategies.write();
        for strategy in strategies.iter_mut() {
            strategy.on_completion(&results_map);
        }

        results_map
    }

    /// Run backtest for a single symbol
    fn run_single_symbol(&self, symbol_config: &SymbolConfig) -> (SmartString, BacktestStats) {
        // Create config for this symbol
        let mut config = BacktestConfig {
            start_time_ns: self.config.start_time_ns,
            end_time_ns: self.config.end_time_ns,
            #[allow(clippy::disallowed_names)]
            symbols: smallvec::smallvec![symbol_config.symbol.clone()],
            tick_sizes: FxHashMap::default(),
            lot_sizes: FxHashMap::default(),
            queue_model: self.config.queue_model,
            allow_partial_fills: self.config.allow_partial_fills,
            order_latency: self.config.order_latency.clone_box(),
            market_data_latency: self.config.market_data_latency.clone_box(),
            conservative_mode: false,
            market_impact: None,
            conservative_params: None,
        };

        config
            .tick_sizes
            .insert(symbol_config.symbol.clone(), symbol_config.tick_size);
        config
            .lot_sizes
            .insert(symbol_config.symbol.clone(), symbol_config.lot_size);

        // Create engine
        let engine = DefaultBacktestEngine::new(config);

        // Clone strategies for this symbol
        let strategies = self.strategies.read();
        for strategy in strategies.iter() {
            let symbol_strategy = strategy.clone_for_symbol(&symbol_config.symbol);
            engine.add_strategy(symbol_strategy);
        }
        drop(strategies);

        // Load market data
        let events = self.data_provider.get_events(
            &symbol_config.symbol,
            self.config.start_time_ns,
            self.config.end_time_ns,
        );

        // Add events to engine
        for (timestamp_ns, event) in events {
            engine.add_market_data(timestamp_ns, symbol_config.symbol.clone(), event);
        }

        // Run backtest
        let stats = engine.run();

        (symbol_config.symbol.clone(), stats)
    }

    /// Run with progress tracking
    #[must_use]
    pub fn run_with_progress<F>(&self, progress_fn: F) -> FxHashMap<SmartString, BacktestStats>
    where
        F: Fn(&str, f64) + Sync + Send,
    {
        let total_symbols = self.symbols.len();
        let completed = Arc::new(Mutex::new(0usize));

        let results: Vec<(SmartString, BacktestStats)> = self
            .symbols
            .par_iter()
            .map(|symbol_config| {
                let result = self.run_single_symbol(symbol_config);

                // Update progress
                let mut count = completed.lock();
                *count += 1;
                let progress = (*count as f64) / (total_symbols as f64);
                progress_fn(&symbol_config.symbol, progress);

                result
            })
            .collect();

        // Convert to hashmap
        let results_map: FxHashMap<SmartString, BacktestStats> = results.into_iter().collect();

        // Notify strategies of completion
        let mut strategies = self.strategies.write();
        for strategy in strategies.iter_mut() {
            strategy.on_completion(&results_map);
        }

        results_map
    }
}

/// Multi-symbol backtesting with shared state
pub struct MultiSymbolBacktest {
    /// Map of symbol to its dedicated backtest engine
    engines: FxHashMap<SmartString, Arc<DefaultBacktestEngine>>,
    /// Shared state accessible by all symbol engines
    shared_state: Arc<RwLock<SharedState>>,
}

/// Shared state between symbols
#[derive(Default)]
pub struct SharedState {
    /// Current position sizes by symbol
    pub positions: FxHashMap<SmartString, f64>,
    /// Total profit and loss across all symbols
    pub total_pnl: f64,
    /// Total traded volume across all symbols
    pub total_volume: f64,
}

impl Default for MultiSymbolBacktest {
    fn default() -> Self {
        Self::new()
    }
}

impl MultiSymbolBacktest {
    /// Create a new multi-symbol backtest
    #[must_use]
    pub fn new() -> Self {
        Self {
            engines: FxHashMap::default(),
            shared_state: Arc::new(RwLock::new(SharedState::default())),
        }
    }

    /// Add a symbol with its engine
    pub fn add_symbol(&mut self, symbol: SmartString, engine: Arc<DefaultBacktestEngine>) {
        self.engines.insert(symbol, engine);
    }

    /// Get shared state
    pub fn shared_state(&self) -> Arc<RwLock<SharedState>> {
        self.shared_state.clone()
    }

    /// Run all symbols in parallel
    #[must_use]
    pub fn run_parallel(&self) -> FxHashMap<SmartString, BacktestStats> {
        let results: FxHashMap<SmartString, BacktestStats> = self
            .engines
            .par_iter()
            .map(|(symbol, engine)| {
                let stats = engine.run();
                (symbol.clone(), stats)
            })
            .collect();

        results.into_iter().collect()
    }
}

/// Example market data provider for testing
pub struct SimpleMarketDataProvider {
    data: Arc<RwLock<MarketDataStore>>,
}

type MarketDataStore = FxHashMap<SmartString, SmallVec<[(u64, MarketDataEvent); 256]>>;

impl Default for SimpleMarketDataProvider {
    fn default() -> Self {
        Self::new()
    }
}

impl SimpleMarketDataProvider {
    /// Create a new SimpleMarketDataProvider
    #[must_use]
    pub fn new() -> Self {
        Self {
            data: Arc::new(RwLock::new(FxHashMap::default())),
        }
    }

    /// Add market data events for a symbol
    pub fn add_data(&self, symbol: SmartString, events: SmallVec<[(u64, MarketDataEvent); 256]>) {
        self.data.write().insert(symbol, events);
    }
}

impl MarketDataProvider for SimpleMarketDataProvider {
    fn get_events(
        &self,
        symbol: &str,
        start_ns: u64,
        end_ns: u64,
    ) -> SmallVec<[(u64, MarketDataEvent); 256]> {
        self.data
            .read()
            .get(symbol)
            .map(|events| {
                events
                    .iter()
                    .filter(|(ts, _)| *ts >= start_ns && *ts <= end_ns)
                    .cloned()
                    .collect()
            })
            .unwrap_or_default()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::engine::Strategy;
    use crate::latency::FixedLatency;
    use crate::matching::{Order, OrderSide, OrderType, QueuePosition, TimeInForce};
    use crate::orderbook::OrderBook;
    use rust_decimal_macros::dec;
    use rusty_common::SmallOrderVec;

    struct TestParallelStrategy {}

    impl ParallelStrategy for TestParallelStrategy {
        fn clone_for_symbol(&self, symbol: &str) -> Box<dyn Strategy> {
            Box::new(TestSymbolStrategy {
                symbol: SmartString::from(symbol),
                orders_sent: false,
            })
        }

        fn on_completion(&mut self, results: &FxHashMap<SmartString, BacktestStats>) {
            println!("Backtest complete: {results:?}");
        }
    }

    struct TestSymbolStrategy {
        symbol: SmartString,
        orders_sent: bool,
    }

    impl Strategy for TestSymbolStrategy {
        fn on_market_data(&mut self, _symbol: &str, _event: &MarketDataEvent, _book: &OrderBook) {
            // No action
        }

        fn on_order_response(&mut self, _response: &crate::engine::OrderResponse) {
            // No action
        }

        fn on_timer(&mut self, _timer_id: u64) {
            // No action
        }

        fn get_orders(&mut self) -> SmallOrderVec<Order> {
            if !self.orders_sent && self.symbol == "BTC-USD" {
                self.orders_sent = true;
                smallvec::smallvec![Order {
                    id: 1,
                    symbol: self.symbol.clone(),
                    side: OrderSide::Buy,
                    order_type: OrderType::Limit,
                    price: dec!(50000),
                    quantity: Decimal::ONE,
                    remaining_quantity: Decimal::ONE,
                    time_in_force: TimeInForce::GTC,
                    timestamp_ns: 1_000_000,
                    queue_position: QueuePosition::RiskAverse,
                }]
            } else {
                SmallOrderVec::new()
            }
        }

        fn get_cancels(&mut self) -> SmallOrderVec<u64> {
            SmallOrderVec::new()
        }
    }

    #[test]
    fn test_parallel_engine() {
        let config = ParallelConfig {
            start_time_ns: 0,
            end_time_ns: 10_000_000_000,
            queue_model: QueueModel::FIFO,
            allow_partial_fills: true,
            order_latency: Box::new(FixedLatency::new(100_000)),
            market_data_latency: Box::new(FixedLatency::new(50_000)),
        };

        let symbols = smallvec::smallvec![
            SymbolConfig {
                symbol: "BTC-USD".into(),
                tick_size: dec!(0.01),
                lot_size: Decimal::new(1, 3), // 0.001
            },
            SymbolConfig {
                symbol: "ETH-USD".into(),
                tick_size: dec!(0.01),
                lot_size: Decimal::new(1, 3), // 0.001
            },
        ];

        // Create data provider
        let provider = Arc::new(SimpleMarketDataProvider::new());

        // Add test data for BTC
        provider.add_data(
            "BTC-USD".into(),
            smallvec::smallvec![
                (
                    1_000_000,
                    MarketDataEvent::DepthUpdate {
                        side: OrderSide::Sell,
                        price: dec!(50001),
                        quantity: Decimal::TWO,
                        order_count: 1,
                    },
                ),
                (
                    2_000_000,
                    MarketDataEvent::Trade {
                        side: OrderSide::Sell,
                        price: dec!(50000),
                        quantity: Decimal::new(5, 1), // 0.5
                    },
                ),
            ],
        );

        // Add test data for ETH
        provider.add_data(
            "ETH-USD".into(),
            smallvec::smallvec![(
                1_000_000,
                MarketDataEvent::DepthUpdate {
                    side: OrderSide::Buy,
                    price: dec!(3000),
                    quantity: Decimal::from(10),
                    order_count: 2,
                },
            )],
        );

        let engine = ParallelEngine::new(config, symbols, provider);
        engine.add_strategy(Box::new(TestParallelStrategy {}));

        // Run with progress tracking
        let results = engine.run_with_progress(|symbol, progress| {
            println!("{}: {:.1}%", symbol, progress * 100.0);
        });

        assert_eq!(results.len(), 2);
        assert!(results.contains_key("BTC-USD"));
        assert!(results.contains_key("ETH-USD"));

        // BTC should have processed orders
        let btc_stats = &results["BTC-USD"];
        assert_eq!(btc_stats.orders_submitted, 1);
        assert_eq!(btc_stats.orders_filled, 1);
    }
}