foxhunt/crates/backtesting/src/replay_engine.rs

//! Market data replay engine for historical backtesting
//!
//! Provides tick-by-tick historical market data replay with configurable speed,
//! filtering, and synchronization capabilities for strategy testing.

use std::{
    sync::Arc,
    time::{Duration as StdDuration, Instant},
};

use anyhow::{Context, Result};
use chrono::{DateTime, TimeDelta, Utc};
use common::{Price, Quantity, Symbol, Timestamp};
use rust_decimal::Decimal;
use trading_engine::types::events::MarketEvent;
// Channel imports removed as not used
use dashmap::DashMap;
use serde::{Deserialize, Serialize};
use tokio::{
    fs::File,
    io::{AsyncBufReadExt, BufReader},
    sync::{mpsc, RwLock},
    time::sleep,
};
use tracing::{error, info, warn};

/// Configuration for market data replay
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ReplayConfig {
    /// Speed multiplier for replay (1.0 = real-time, 0.0 = maximum speed)
    pub speed_multiplier: f64,
    /// Start time for replay
    pub start_time: DateTime<Utc>,
    /// End time for replay
    pub end_time: DateTime<Utc>,
    /// Symbols to include in replay
    pub symbols: Vec<Symbol>,
    /// Data sources to replay from
    pub data_sources: Vec<DataSource>,
    /// Buffer size for event queue
    pub buffer_size: usize,
    /// Enable tick-by-tick replay (vs aggregated bars)
    pub tick_by_tick: bool,
    /// Filter configuration
    pub filters: ReplayFilters,
}

impl Default for ReplayConfig {
    fn default() -> Self {
        Self {
            speed_multiplier: 1.0,
            start_time: Utc::now() - TimeDelta::days(1),
            end_time: Utc::now(),
            symbols: Vec::new(),
            data_sources: vec![DataSource::default()],
            buffer_size: 10000,
            tick_by_tick: true,
            filters: ReplayFilters::default(),
        }
    }
}

/// Data source configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DataSource {
    /// Source type (file, database, etc.)
    pub source_type: SourceType,
    /// Path or connection string
    pub path: String,
    /// Data format
    pub format: DataFormat,
    /// Priority for conflicting data
    pub priority: u8,
}

impl Default for DataSource {
    fn default() -> Self {
        Self {
            source_type: SourceType::CsvFile,
            path: "data/market_data.csv".to_string(),
            format: DataFormat::OhlcvTicks,
            priority: 1,
        }
    }
}

/// Supported data source types
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum SourceType {
    CsvFile,
    ParquetFile,
    Database,
    BinaryFile,
}

/// Data format specifications
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum DataFormat {
    OhlcvTicks,
    L1BookUpdates,
    L2BookUpdates,
    TradeUpdates,
    Custom(String),
}

/// Replay filtering configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ReplayFilters {
    /// Minimum price change to include
    pub min_price_change: Option<Decimal>,
    /// Minimum volume to include
    pub min_volume: Option<Quantity>,
    /// Include only market hours
    pub market_hours_only: bool,
    /// Custom filter expressions
    pub custom_filters: Vec<String>,
}

impl Default for ReplayFilters {
    fn default() -> Self {
        Self {
            min_price_change: None,
            min_volume: None,
            market_hours_only: false,
            custom_filters: Vec::new(),
        }
    }
}

/// Historical market event with metadata
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ReplayEvent {
    /// The market event
    pub event: MarketEvent,
    /// Original timestamp from data source
    pub original_timestamp: Timestamp,
    /// Replay timestamp (adjusted for speed)
    pub replay_timestamp: Timestamp,
    /// Data source identifier
    pub source_id: String,
    /// Event sequence number
    pub sequence: u64,
}

/// Market data replay engine
pub struct MarketReplay {
    /// Replay configuration
    config: ReplayConfig,
    /// Event output channel
    _event_sender: mpsc::UnboundedSender<ReplayEvent>,
    /// Event receiver for consumers
    event_receiver: Arc<RwLock<Option<mpsc::UnboundedReceiver<ReplayEvent>>>>,
    /// Current replay state
    state: Arc<RwLock<ReplayState>>,
    /// Symbol-specific order books
    order_books: Arc<DashMap<Symbol, std::collections::HashMap<String, String>>>,
    /// Performance metrics
    metrics: Arc<ReplayMetrics>,
    /// Event sequence counter
    sequence_counter: Arc<std::sync::atomic::AtomicU64>,
}

/// Current state of replay engine
#[derive(Debug, Clone)]
pub struct ReplayState {
    /// Current replay time
    pub current_time: DateTime<Utc>,
    /// Is replay active
    pub is_active: bool,
    /// Is replay paused
    pub is_paused: bool,
    /// Events processed count
    pub events_processed: u64,
    /// Replay start time (wall clock)
    pub replay_start: Option<Instant>,
    /// Last event timestamp
    pub last_event_time: Option<DateTime<Utc>>,
}

impl Default for ReplayState {
    fn default() -> Self {
        Self {
            current_time: Utc::now(),
            is_active: false,
            is_paused: false,
            events_processed: 0,
            replay_start: None,
            last_event_time: None,
        }
    }
}

/// Performance metrics for replay engine
#[derive(Debug, Default)]
pub struct ReplayMetrics {
    /// Events per second
    pub events_per_second: Arc<std::sync::atomic::AtomicU64>,
    /// Total events processed
    pub total_events: Arc<std::sync::atomic::AtomicU64>,
    /// Latency distribution
    pub latency_histogram: Arc<RwLock<Vec<StdDuration>>>,
    /// Memory usage
    pub memory_usage: Arc<std::sync::atomic::AtomicUsize>,
    /// Error count
    pub error_count: Arc<std::sync::atomic::AtomicU64>,
}

impl MarketReplay {
    /// Create new market replay engine
    ///
    /// # Arguments
    /// * `config` - Configuration for market data replay
    ///
    /// # Returns
    /// * `Self` - New market replay engine instance
    pub fn new(config: ReplayConfig) -> Self {
        let (_event_sender, event_receiver) = mpsc::unbounded_channel();

        // Initialize state with config.start_time instead of Utc::now()
        // to ensure consistent timestamps before start_replay() is called
        let initial_state = ReplayState {
            current_time: config.start_time,
            is_active: false,
            is_paused: false,
            events_processed: 0,
            replay_start: None,
            last_event_time: None,
        };

        Self {
            config,
            _event_sender,
            event_receiver: Arc::new(RwLock::new(Some(event_receiver))),
            state: Arc::new(RwLock::new(initial_state)),
            order_books: Arc::new(DashMap::new()),
            metrics: Arc::new(ReplayMetrics::default()),
            sequence_counter: Arc::new(std::sync::atomic::AtomicU64::new(0)),
        }
    }

    /// Take the event receiver (can only be called once)
    ///
    /// # Returns
    /// * `Option<mpsc::UnboundedReceiver<ReplayEvent>>` - Event receiver for consuming replay events
    ///
    /// # Note
    ///
    /// This method can only be called once as it moves the receiver out of the engine
    pub async fn take_receiver(&self) -> Option<mpsc::UnboundedReceiver<ReplayEvent>> {
        self.event_receiver.write().await.take()
    }

    /// Start the replay process
    ///
    /// # Returns
    /// * `Result<()>` - Success or error from replay process
    ///
    /// # Errors
    ///
    /// Returns error if data loading or replay fails
    pub async fn start_replay(&self) -> Result<()> {
        info!("Starting market data replay");

        // Update state
        {
            let mut state = self.state.write().await;
            state.is_active = true;
            state.is_paused = false;
            state.current_time = self.config.start_time;
            state.replay_start = Some(Instant::now());
        }

        // Load and sort all data sources
        let mut all_events = self.load_all_events().await?;
        all_events.sort_by_key(|event| event.original_timestamp);

        info!("Loaded {} events for replay", all_events.len());

        // Start replay loop
        self.replay_events(all_events).await?;

        Ok(())
    }

    /// Load events from all configured data sources
    ///
    /// # Returns
    /// * `Result<Vec<ReplayEvent>>` - All loaded and filtered events sorted by timestamp
    ///
    /// # Errors
    ///
    /// Returns error if any data source fails to load
    async fn load_all_events(&self) -> Result<Vec<ReplayEvent>> {
        let mut all_events = Vec::new();

        for (source_idx, source) in self.config.data_sources.iter().enumerate() {
            match source.source_type {
                SourceType::CsvFile => {
                    let events = self.load_csv_events(&source, source_idx).await?;
                    all_events.extend(events);
                },
                SourceType::ParquetFile => {
                    warn!("Parquet files not yet implemented");
                },
                SourceType::Database => {
                    warn!("Database sources not yet implemented");
                },
                SourceType::BinaryFile => {
                    warn!("Binary files not yet implemented");
                },
            }
        }

        // Apply filters
        let filtered_events = self.apply_filters(all_events).await?;

        Ok(filtered_events)
    }

    /// Load events from CSV file
    ///
    /// # Arguments
    /// * `source` - Data source configuration specifying the CSV file
    ///
    /// * `source_idx` - Index of the data source for identification
    ///
    /// # Returns
    /// * `Result<Vec<ReplayEvent>>` - Events loaded from the CSV file
    ///
    /// # Errors
    ///
    /// Returns error if file cannot be opened or parsed
    async fn load_csv_events(
        &self,
        source: &DataSource,
        source_idx: usize,
    ) -> Result<Vec<ReplayEvent>> {
        let file = File::open(&source.path)
            .await
            .with_context(|| format!("Failed to open CSV file: {}", source.path))?;

        let mut reader = BufReader::new(file);
        let mut line = String::new();
        let mut events = Vec::new();
        let mut line_number = 0;

        // Skip header
        reader.read_line(&mut line).await?;
        line.clear();

        while reader.read_line(&mut line).await? > 0 {
            line_number += 1;

            if let Ok(event) = self.parse_csv_line(&line, source, source_idx).await {
                events.push(event);
            } else {
                warn!("Failed to parse line {}: {}", line_number, line.trim());
            }

            line.clear();
        }

        Ok(events)
    }

    /// Parse a single CSV line into a replay event
    ///
    /// # Arguments
    /// * `line` - CSV line to parse
    ///
    /// * `source` - Data source configuration for format specification
    /// * `source_idx` - Index of the data source for identification
    ///
    /// # Returns
    /// * `Result<ReplayEvent>` - Parsed replay event
    ///
    /// # Errors
    ///
    /// Returns error if line format is invalid or cannot be parsed
    async fn parse_csv_line(
        &self,
        line: &str,
        source: &DataSource,
        source_idx: usize,
    ) -> Result<ReplayEvent> {
        let fields: Vec<&str> = line.trim().split(',').collect();

        match source.format {
            DataFormat::OhlcvTicks => {
                if fields.len() < 7 {
                    return Err(anyhow::anyhow!(
                        "Invalid OHLCV format: need at least 7 fields"
                    ));
                }

                let timestamp: i64 = fields[0].parse()?;
                let symbol = Symbol::new(fields[1].to_string());
                let _open: Decimal = fields[2].parse()?;
                let _high: Decimal = fields[3].parse()?;
                let _low: Decimal = fields[4].parse()?;
                let close: Decimal = fields[5].parse()?;
                let volume: Decimal = fields[6].parse()?;

                let market_event = MarketEvent::Trade {
                    symbol: symbol.clone(),
                    price: Price::from_f64(close.try_into().unwrap_or(0.0))?,
                    size: Quantity::from_f64(volume.try_into().unwrap_or(0.0))?,
                    timestamp: DateTime::from_timestamp(
                        timestamp / 1000,
                        ((timestamp % 1000) * 1_000_000) as u32,
                    )
                    .unwrap_or_default(),
                    side: None,
                    venue: None,
                    trade_id: None,
                };

                let sequence = self
                    .sequence_counter
                    .fetch_add(1, std::sync::atomic::Ordering::Relaxed);

                Ok(ReplayEvent {
                    event: market_event,
                    original_timestamp: DateTime::from_timestamp(
                        timestamp / 1000,
                        ((timestamp % 1000) * 1_000_000) as u32,
                    )
                    .unwrap_or_default(),
                    replay_timestamp: Utc::now(),
                    source_id: format!("source_{}", source_idx),
                    sequence,
                })
            },
            _ => Err(anyhow::anyhow!(
                "Unsupported data format: {:?}",
                source.format
            )),
        }
    }

    /// Apply configured filters to events
    ///
    /// # Arguments
    /// * `events` - Vector of events to filter
    ///
    /// # Returns
    /// * `Result<Vec<ReplayEvent>>` - Filtered events that meet filter criteria
    async fn apply_filters(&self, events: Vec<ReplayEvent>) -> Result<Vec<ReplayEvent>> {
        let mut filtered = Vec::new();
        let total_events = events.len(); // Store length before moving events

        for event in events {
            if self.should_include_event(&event).await {
                filtered.push(event);
            }
        }

        info!(
            "Filtered {} events from {} total",
            filtered.len(),
            total_events
        );
        Ok(filtered)
    }

    /// Check if event should be included based on filters
    ///
    /// # Arguments
    /// * `event` - Event to check against filter criteria
    ///
    /// # Returns
    /// * `bool` - True if event should be included, false otherwise
    async fn should_include_event(&self, event: &ReplayEvent) -> bool {
        // Time range filter
        let event_time = DateTime::from_timestamp(
            event.original_timestamp.timestamp(),
            event.original_timestamp.timestamp_subsec_nanos(),
        )
        .unwrap_or_default();

        if event_time < self.config.start_time || event_time > self.config.end_time {
            return false;
        }

        // Symbol filter
        if !self.config.symbols.is_empty() {
            let event_symbol = match &event.event {
                MarketEvent::Trade { symbol, .. } => symbol,
                MarketEvent::Quote { symbol, .. } => symbol,
                MarketEvent::OrderBookUpdate { symbol, .. } => symbol,
                MarketEvent::Bar { symbol, .. } => symbol,
                MarketEvent::OrderBook { symbol, .. } => symbol,
                MarketEvent::Sentiment { .. } => return true, // Include sentiment events for now
                MarketEvent::Control { .. } => return true,   // Include control events for now
            };

            if !self.config.symbols.contains(event_symbol) {
                return false;
            }
        }

        // Volume filter
        if let Some(min_volume) = &self.config.filters.min_volume {
            let event_volume = match &event.event {
                MarketEvent::Trade { size, .. } => Some(size),
                _ => None,
            };

            if let Some(volume) = event_volume {
                if volume < min_volume {
                    return false;
                }
            }
        }

        true
    }

    /// Replay events with timing control
    ///
    /// # Arguments
    /// * `events` - Vector of events to replay in chronological order
    ///
    /// # Returns
    /// * `Result<()>` - Success or error from replay process
    ///
    /// # Note
    ///
    /// Respects speed multiplier for timing and handles pause/resume functionality
    async fn replay_events(&self, events: Vec<ReplayEvent>) -> Result<()> {
        let mut last_event_time: Option<DateTime<Utc>> = None;
        let _replay_start = Instant::now();

        for event in events {
            // Check if replay should continue
            {
                let state = self.state.read().await;
                if !state.is_active {
                    break;
                }

                // Handle pause
                while state.is_paused {
                    sleep(StdDuration::from_millis(100)).await;
                }
            }

            // Calculate timing
            let event_time = DateTime::from_timestamp(
                event.original_timestamp.timestamp(),
                event.original_timestamp.timestamp_subsec_nanos(),
            )
            .unwrap_or_default();

            if let Some(last_time) = last_event_time {
                let time_diff = event_time.signed_duration_since(last_time);
                if time_diff > TimeDelta::zero() && self.config.speed_multiplier > 0.0 {
                    let sleep_duration = StdDuration::from_millis(
                        ((time_diff.num_milliseconds() as f64) / self.config.speed_multiplier)
                            as u64,
                    );
                    sleep(sleep_duration).await;
                }
            }

            // Update order book if applicable
            self.update_order_book(&event).await;

            // Send event
            if let Err(e) = self._event_sender.send(event.clone()) {
                error!("Failed to send replay event: {}", e);
                break;
            }

            // Update metrics and state
            self.update_metrics(&event).await;
            self.update_state(event_time).await;

            last_event_time = Some(event_time);
        }

        // Mark replay as complete
        {
            let mut state = self.state.write().await;
            state.is_active = false;
        }

        info!("Market data replay completed");
        Ok(())
    }

    /// Update order book with new event
    ///
    /// # Arguments
    /// * `event` - Replay event that may contain order book updates
    ///
    /// # Note
    ///
    /// Currently handles basic order book tracking for trade and order book events
    async fn update_order_book(&self, event: &ReplayEvent) {
        match &event.event {
            MarketEvent::OrderBookUpdate { symbol, .. } => {
                // Update order book logic would go here
                // For now, just ensure the symbol exists
                self.order_books
                    .entry(symbol.clone())
                    .or_insert_with(std::collections::HashMap::new);
            },
            MarketEvent::Trade {
                symbol,
                price: _price,
                ..
            } => {
                // Update last trade price in order book
                if let Some(mut _book) = self.order_books.get_mut(symbol) {
                    // Update last trade price logic
                }
            },
            _ => {},
        }
    }

    /// Update performance metrics
    ///
    /// # Arguments
    /// * `_event` - Replay event (currently unused but reserved for future metrics)
    ///
    /// # Note
    ///
    /// Updates event count and calculates events per second
    async fn update_metrics(&self, _event: &ReplayEvent) {
        self.metrics
            .total_events
            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);

        // Update events per second calculation
        // Implementation would track timing for EPS calculation
    }

    /// Update replay state
    ///
    /// # Arguments
    /// * `event_time` - Timestamp of the current event being processed
    ///
    /// # Note
    ///
    /// Updates current time, event count, and last event timestamp
    async fn update_state(&self, event_time: DateTime<Utc>) {
        let mut state = self.state.write().await;
        state.current_time = event_time;
        state.events_processed += 1;
        state.last_event_time = Some(event_time);
    }

    /// Pause the replay
    ///
    /// # Note
    ///
    /// Sets the replay state to paused, causing event processing to halt until resumed
    pub async fn pause(&self) {
        let mut state = self.state.write().await;
        state.is_paused = true;
        info!("Market replay paused");
    }

    /// Resume the replay
    ///
    /// # Note
    ///
    /// Clears the paused state, allowing event processing to continue
    pub async fn resume(&self) {
        let mut state = self.state.write().await;
        state.is_paused = false;
        info!("Market replay resumed");
    }

    /// Stop the replay
    ///
    /// # Note
    ///
    /// Completely stops the replay process and clears both active and paused states
    pub async fn stop(&self) {
        let mut state = self.state.write().await;
        state.is_active = false;
        state.is_paused = false;
        info!("Market replay stopped");
    }

    /// Get current replay state
    ///
    /// # Returns
    /// * `ReplayState` - Current state including timing, event counts, and status flags
    pub async fn get_state(&self) -> ReplayState {
        self.state.read().await.clone()
    }

    /// Get current order book for symbol
    ///
    /// # Arguments
    /// * `symbol` - Symbol to retrieve order book for
    ///
    /// # Returns
    /// * `Option<HashMap<String, String>>` - Order book data if available for the symbol
    pub async fn get_order_book(
        &self,
        symbol: &Symbol,
    ) -> Option<std::collections::HashMap<String, String>> {
        self.order_books.get(symbol).map(|book| book.clone())
    }

    /// Get performance metrics
    ///
    /// # Returns
    /// * `ReplayMetrics` - Current performance metrics including event rates and error counts
    pub async fn get_metrics(&self) -> ReplayMetrics {
        ReplayMetrics {
            events_per_second: Arc::clone(&self.metrics.events_per_second),
            total_events: Arc::clone(&self.metrics.total_events),
            latency_histogram: Arc::clone(&self.metrics.latency_histogram),
            memory_usage: Arc::clone(&self.metrics.memory_usage),
            error_count: Arc::clone(&self.metrics.error_count),
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::io::Write;
    use tempfile::NamedTempFile;

    #[tokio::test]
    async fn test_replay_engine_creation() {
        let config = ReplayConfig::default();
        let replay = MarketReplay::new(config);

        let state = replay.get_state().await;
        assert!(!state.is_active);
        assert!(!state.is_paused);
    }

    #[tokio::test]
    async fn test_csv_loading() {
        // Create test CSV file
        let mut temp_file = NamedTempFile::new().unwrap();
        writeln!(temp_file, "timestamp,symbol,open,high,low,close,volume").unwrap();
        writeln!(
            temp_file,
            "1609459200000,BTCUSD,29000.0,29100.0,28900.0,29050.0,1.5"
        )
        .unwrap();

        let config = ReplayConfig {
            data_sources: vec![DataSource {
                source_type: SourceType::CsvFile,
                path: temp_file.path().to_string_lossy().to_string(),
                format: DataFormat::OhlcvTicks,
                priority: 1,
            }],
            start_time: DateTime::from_timestamp(1609459200, 0).unwrap_or_default(),
            end_time: DateTime::from_timestamp(1609459300, 0).unwrap_or_default(),
            ..Default::default()
        };

        let replay = MarketReplay::new(config);
        let events = replay.load_all_events().await.unwrap();

        assert_eq!(events.len(), 1);
        assert_eq!(events[0].sequence, 0);
    }
}