foxhunt/data/tests/test_event_conversion_streaming.rs

//! Comprehensive tests for event conversion and streaming
//!
//! This module contains extensive tests for market data event conversion
//! between different provider formats, streaming performance, event
//! aggregation, filtering, and real-time processing pipelines.

use data::providers::common::{
    MarketDataEvent, TradeEvent, QuoteEvent, OrderBookSnapshot, OrderBookUpdate,
    PriceLevel, PriceLevelChange, PriceLevelChangeType, OrderBookSide, AggregateEvent,
    NewsEvent, SentimentEvent, AnalystRatingEvent, UnusualOptionsEvent,
    ConnectionStatusEvent, ErrorEvent, MarketStatusEvent, ErrorCategory, MarketState,
    NewsEventType, SentimentPeriod, RatingAction, OptionsContract, OptionsType,
    UnusualOptionsType, OptionsSentiment
};
use data::providers::databento_streaming::{
    DatabentoStreamingProvider, DatabentoMessage, DatabentoTrade, DatabentoQuote, DatabentoOrderBook
};
use data::providers::benzinga::{BenzingaNewsArticle, BenzingaEarnings, BenzingaRating, NewsEvent as BenzingaNewsEvent};
use core::types::{Symbol, Price, Quantity, Decimal};
use core::trading::data_interface::{MarketDataEvent as CoreMarketDataEvent, TradeEvent as CoreTradeEvent};
use tokio::sync::{broadcast, mpsc};
use tokio::time::{sleep, timeout, Duration, Instant};
use tokio_stream::{Stream, StreamExt};
use futures::stream;
use std::collections::{HashMap, VecDeque};
use std::sync::Arc;
use std::pin::Pin;
use chrono::{DateTime, Utc, Duration as ChronoDuration};
use rust_decimal_macros::dec;
use tokio_test;

/// Event aggregator for combining multiple data sources
struct EventAggregator {
    trade_buffer: VecDeque<TradeEvent>,
    quote_buffer: VecDeque<QuoteEvent>,
    news_buffer: VecDeque<NewsEvent>,
    event_sender: broadcast::Sender<MarketDataEvent>,
    max_buffer_size: usize,
}

impl EventAggregator {
    fn new(max_buffer_size: usize) -> Self {
        let (event_sender, _) = broadcast::channel(10000);
        Self {
            trade_buffer: VecDeque::with_capacity(max_buffer_size),
            quote_buffer: VecDeque::with_capacity(max_buffer_size),
            news_buffer: VecDeque::with_capacity(max_buffer_size),
            event_sender,
            max_buffer_size,
        }
    }

    fn add_trade(&mut self, trade: TradeEvent) -> Result<(), &'static str> {
        if self.trade_buffer.len() >= self.max_buffer_size {
            self.trade_buffer.pop_front();
        }
        self.trade_buffer.push_back(trade.clone());
        
        let event = MarketDataEvent::Trade(trade);
        self.event_sender.send(event).map_err(|_| "Failed to send trade event")?;
        Ok(())
    }

    fn add_quote(&mut self, quote: QuoteEvent) -> Result<(), &'static str> {
        if self.quote_buffer.len() >= self.max_buffer_size {
            self.quote_buffer.pop_front();
        }
        self.quote_buffer.push_back(quote.clone());
        
        let event = MarketDataEvent::Quote(quote);
        self.event_sender.send(event).map_err(|_| "Failed to send quote event")?;
        Ok(())
    }

    fn add_news(&mut self, news: NewsEvent) -> Result<(), &'static str> {
        if self.news_buffer.len() >= self.max_buffer_size {
            self.news_buffer.pop_front();
        }
        self.news_buffer.push_back(news.clone());
        
        let event = MarketDataEvent::NewsAlert(news);
        self.event_sender.send(event).map_err(|_| "Failed to send news event")?;
        Ok(())
    }

    fn get_trade_count(&self) -> usize {
        self.trade_buffer.len()
    }

    fn get_quote_count(&self) -> usize {
        self.quote_buffer.len()
    }

    fn get_news_count(&self) -> usize {
        self.news_buffer.len()
    }

    fn subscribe(&self) -> broadcast::Receiver<MarketDataEvent> {
        self.event_sender.subscribe()
    }

    fn get_latest_trade_for_symbol(&self, symbol: &Symbol) -> Option<&TradeEvent> {
        self.trade_buffer.iter().rev().find(|trade| &trade.symbol == symbol)
    }

    fn get_latest_quote_for_symbol(&self, symbol: &Symbol) -> Option<&QuoteEvent> {
        self.quote_buffer.iter().rev().find(|quote| &quote.symbol == symbol)
    }
}

/// Event filter for processing specific types of market data
struct EventFilter {
    allowed_symbols: Option<Vec<Symbol>>,
    allowed_event_types: Vec<String>,
    min_trade_size: Option<Decimal>,
    min_news_importance: Option<f64>,
}

impl EventFilter {
    fn new() -> Self {
        Self {
            allowed_symbols: None,
            allowed_event_types: vec![],
            min_trade_size: None,
            min_news_importance: None,
        }
    }

    fn with_symbols(mut self, symbols: Vec<Symbol>) -> Self {
        self.allowed_symbols = Some(symbols);
        self
    }

    fn with_event_types(mut self, event_types: Vec<String>) -> Self {
        self.allowed_event_types = event_types;
        self
    }

    fn with_min_trade_size(mut self, min_size: Decimal) -> Self {
        self.min_trade_size = Some(min_size);
        self
    }

    fn with_min_news_importance(mut self, min_importance: f64) -> Self {
        self.min_news_importance = Some(min_importance);
        self
    }

    fn should_process_event(&self, event: &MarketDataEvent) -> bool {
        // Check symbol filter
        if let Some(ref allowed_symbols) = self.allowed_symbols {
            if let Some(symbol) = event.symbol() {
                if !allowed_symbols.contains(symbol) {
                    return false;
                }
            }
        }

        // Check event type filters
        if !self.allowed_event_types.is_empty() {
            let event_type = match event {
                MarketDataEvent::Trade(_) => "trade",
                MarketDataEvent::Quote(_) => "quote",
                MarketDataEvent::OrderBookL2Snapshot(_) => "orderbook",
                MarketDataEvent::NewsAlert(_) => "news",
                _ => "other",
            };

            if !self.allowed_event_types.contains(&event_type.to_string()) {
                return false;
            }
        }

        // Check trade size filter
        if let Some(min_size) = self.min_trade_size {
            if let MarketDataEvent::Trade(trade) = event {
                if trade.size < min_size {
                    return false;
                }
            }
        }

        // Check news importance filter
        if let Some(min_importance) = self.min_news_importance {
            if let MarketDataEvent::NewsAlert(news) = event {
                if news.importance < min_importance {
                    return false;
                }
            }
        }

        true
    }
}

/// Stream processor for real-time event handling
struct StreamProcessor {
    processed_count: u64,
    filtered_count: u64,
    error_count: u64,
    filter: Option<EventFilter>,
}

impl StreamProcessor {
    fn new() -> Self {
        Self {
            processed_count: 0,
            filtered_count: 0,
            error_count: 0,
            filter: None,
        }
    }

    fn with_filter(mut self, filter: EventFilter) -> Self {
        self.filter = Some(filter);
        self
    }

    async fn process_event(&mut self, event: MarketDataEvent) -> Result<Option<MarketDataEvent>, String> {
        // Apply filter if present
        if let Some(ref filter) = self.filter {
            if !filter.should_process_event(&event) {
                self.filtered_count += 1;
                return Ok(None);
            }
        }

        // Process the event (simulate some processing time)
        match &event {
            MarketDataEvent::Trade(trade) => {
                if trade.price <= dec!(0.0) {
                    self.error_count += 1;
                    return Err("Invalid trade price".to_string());
                }
            }
            MarketDataEvent::Quote(quote) => {
                if let (Some(bid), Some(ask)) = (quote.bid, quote.ask) {
                    if bid >= ask {
                        self.error_count += 1;
                        return Err("Invalid quote spread".to_string());
                    }
                }
            }
            _ => {} // Other event types pass through
        }

        self.processed_count += 1;
        Ok(Some(event))
    }

    fn get_stats(&self) -> (u64, u64, u64) {
        (self.processed_count, self.filtered_count, self.error_count)
    }
}

/// Test event aggregation with multiple event types
#[tokio::test]
async fn test_event_aggregation() {
    let mut aggregator = EventAggregator::new(100);
    let mut receiver = aggregator.subscribe();

    // Add some trades
    let trade1 = TradeEvent {
        symbol: Symbol::from("AAPL"),
        price: dec!(150.00),
        size: dec!(100),
        exchange: "NASDAQ".to_string(),
        conditions: vec![],
        trade_id: Some("trade1".to_string()),
        timestamp: Utc::now(),
        sequence: 1,
    };

    let trade2 = TradeEvent {
        symbol: Symbol::from("MSFT"),
        price: dec!(300.00),
        size: dec!(200),
        exchange: "NASDAQ".to_string(),
        conditions: vec![],
        trade_id: Some("trade2".to_string()),
        timestamp: Utc::now(),
        sequence: 2,
    };

    aggregator.add_trade(trade1.clone()).unwrap();
    aggregator.add_trade(trade2.clone()).unwrap();

    assert_eq!(aggregator.get_trade_count(), 2);

    // Verify events were sent
    let event1 = timeout(Duration::from_millis(100), receiver.recv()).await.unwrap().unwrap();
    let event2 = timeout(Duration::from_millis(100), receiver.recv()).await.unwrap().unwrap();

    match event1 {
        MarketDataEvent::Trade(t) => assert_eq!(t.trade_id, trade1.trade_id),
        _ => panic!("Expected trade event"),
    }

    match event2 {
        MarketDataEvent::Trade(t) => assert_eq!(t.trade_id, trade2.trade_id),
        _ => panic!("Expected trade event"),
    }
}

/// Test event aggregation with buffer overflow
#[tokio::test]
async fn test_event_aggregation_buffer_overflow() {
    let mut aggregator = EventAggregator::new(3); // Small buffer

    // Add more trades than buffer size
    for i in 1..=5 {
        let trade = TradeEvent {
            symbol: Symbol::from("TEST"),
            price: dec!(100.00),
            size: dec!(100),
            exchange: "TEST".to_string(),
            conditions: vec![],
            trade_id: Some(format!("trade{}", i)),
            timestamp: Utc::now(),
            sequence: i,
        };
        aggregator.add_trade(trade).unwrap();
    }

    assert_eq!(aggregator.get_trade_count(), 3); // Should be capped at buffer size

    // Latest trades should be preserved
    let latest = aggregator.get_latest_trade_for_symbol(&Symbol::from("TEST")).unwrap();
    assert_eq!(latest.sequence, 5);
}

/// Test event filtering by symbol
#[tokio::test]
async fn test_event_filter_by_symbol() {
    let filter = EventFilter::new()
        .with_symbols(vec![Symbol::from("AAPL"), Symbol::from("MSFT")]);

    let trade_aapl = MarketDataEvent::Trade(TradeEvent {
        symbol: Symbol::from("AAPL"),
        price: dec!(150.00),
        size: dec!(100),
        exchange: "NASDAQ".to_string(),
        conditions: vec![],
        trade_id: None,
        timestamp: Utc::now(),
        sequence: 1,
    });

    let trade_googl = MarketDataEvent::Trade(TradeEvent {
        symbol: Symbol::from("GOOGL"),
        price: dec!(2800.00),
        size: dec!(50),
        exchange: "NASDAQ".to_string(),
        conditions: vec![],
        trade_id: None,
        timestamp: Utc::now(),
        sequence: 2,
    });

    assert!(filter.should_process_event(&trade_aapl));
    assert!(!filter.should_process_event(&trade_googl));
}

/// Test event filtering by event type
#[tokio::test]
async fn test_event_filter_by_type() {
    let filter = EventFilter::new()
        .with_event_types(vec!["trade".to_string(), "news".to_string()]);

    let trade_event = MarketDataEvent::Trade(TradeEvent {
        symbol: Symbol::from("SPY"),
        price: dec!(400.00),
        size: dec!(100),
        exchange: "NYSE".to_string(),
        conditions: vec![],
        trade_id: None,
        timestamp: Utc::now(),
        sequence: 1,
    });

    let quote_event = MarketDataEvent::Quote(QuoteEvent {
        symbol: Symbol::from("SPY"),
        bid: Some(dec!(399.99)),
        ask: Some(dec!(400.01)),
        bid_size: Some(dec!(100)),
        ask_size: Some(dec!(100)),
        bid_exchange: None,
        ask_exchange: None,
        conditions: vec![],
        timestamp: Utc::now(),
        sequence: 2,
    });

    let news_event = MarketDataEvent::NewsAlert(NewsEvent {
        story_id: "news123".to_string(),
        headline: "Market Update".to_string(),
        summary: None,
        symbols: vec!["SPY".to_string()],
        category: "Markets".to_string(),
        tags: vec![],
        impact_score: None,
        author: None,
        source: "Test Source".to_string(),
        published_at: Utc::now(),
        timestamp: Utc::now(),
        url: None,
    });

    assert!(filter.should_process_event(&trade_event));
    assert!(!filter.should_process_event(&quote_event));
    assert!(filter.should_process_event(&news_event));
}

/// Test event filtering by trade size
#[tokio::test]
async fn test_event_filter_by_trade_size() {
    let filter = EventFilter::new()
        .with_min_trade_size(dec!(500));

    let large_trade = MarketDataEvent::Trade(TradeEvent {
        symbol: Symbol::from("TSLA"),
        price: dec!(250.00),
        size: dec!(1000),
        exchange: "NASDAQ".to_string(),
        conditions: vec![],
        trade_id: None,
        timestamp: Utc::now(),
        sequence: 1,
    });

    let small_trade = MarketDataEvent::Trade(TradeEvent {
        symbol: Symbol::from("TSLA"),
        price: dec!(250.00),
        size: dec!(100),
        exchange: "NASDAQ".to_string(),
        conditions: vec![],
        trade_id: None,
        timestamp: Utc::now(),
        sequence: 2,
    });

    assert!(filter.should_process_event(&large_trade));
    assert!(!filter.should_process_event(&small_trade));
}

/// Test event filtering by news importance
#[tokio::test]
async fn test_event_filter_by_news_importance() {
    let filter = EventFilter::new()
        .with_min_news_importance(0.7);

    let important_news = MarketDataEvent::NewsAlert(NewsEvent {
        story_id: "important123".to_string(),
        headline: "Breaking: Major Earnings Beat".to_string(),
        summary: None,
        symbols: vec!["AAPL".to_string()],
        category: "Earnings".to_string(),
        tags: vec![],
        impact_score: Some(0.8),
        author: None,
        source: "Reuters".to_string(),
        published_at: Utc::now(),
        timestamp: Utc::now(),
        url: None,
    });

    let minor_news = MarketDataEvent::NewsAlert(NewsEvent {
        story_id: "minor456".to_string(),
        headline: "Minor Company Update".to_string(),
        summary: None,
        symbols: vec!["AAPL".to_string()],
        category: "Company".to_string(),
        tags: vec![],
        impact_score: Some(0.3),
        author: None,
        source: "Blog".to_string(),
        published_at: Utc::now(),
        timestamp: Utc::now(),
        url: None,
    });

    assert!(filter.should_process_event(&important_news));
    assert!(!filter.should_process_event(&minor_news));
}

/// Test stream processor with filtering
#[tokio::test]
async fn test_stream_processor_with_filtering() {
    let filter = EventFilter::new()
        .with_symbols(vec![Symbol::from("AAPL")])
        .with_event_types(vec!["trade".to_string()]);

    let mut processor = StreamProcessor::new().with_filter(filter);

    let allowed_event = MarketDataEvent::Trade(TradeEvent {
        symbol: Symbol::from("AAPL"),
        price: dec!(150.00),
        size: dec!(100),
        exchange: "NASDAQ".to_string(),
        conditions: vec![],
        trade_id: None,
        timestamp: Utc::now(),
        sequence: 1,
    });

    let filtered_event = MarketDataEvent::Trade(TradeEvent {
        symbol: Symbol::from("MSFT"),
        price: dec!(300.00),
        size: dec!(100),
        exchange: "NASDAQ".to_string(),
        conditions: vec![],
        trade_id: None,
        timestamp: Utc::now(),
        sequence: 2,
    });

    let result1 = processor.process_event(allowed_event).await.unwrap();
    assert!(result1.is_some());

    let result2 = processor.process_event(filtered_event).await.unwrap();
    assert!(result2.is_none());

    let (processed, filtered, errors) = processor.get_stats();
    assert_eq!(processed, 1);
    assert_eq!(filtered, 1);
    assert_eq!(errors, 0);
}

/// Test stream processor error handling
#[tokio::test]
async fn test_stream_processor_error_handling() {
    let mut processor = StreamProcessor::new();

    let invalid_trade = MarketDataEvent::Trade(TradeEvent {
        symbol: Symbol::from("TEST"),
        price: dec!(-10.00), // Invalid negative price
        size: dec!(100),
        exchange: "TEST".to_string(),
        conditions: vec![],
        trade_id: None,
        timestamp: Utc::now(),
        sequence: 1,
    });

    let result = processor.process_event(invalid_trade).await;
    assert!(result.is_err());
    assert_eq!(result.unwrap_err(), "Invalid trade price");

    let (processed, filtered, errors) = processor.get_stats();
    assert_eq!(processed, 0);
    assert_eq!(filtered, 0);
    assert_eq!(errors, 1);
}

/// Test invalid quote spread detection
#[tokio::test]
async fn test_stream_processor_invalid_quote() {
    let mut processor = StreamProcessor::new();

    let invalid_quote = MarketDataEvent::Quote(QuoteEvent {
        symbol: Symbol::from("TEST"),
        bid: Some(dec!(100.01)), // Bid higher than ask
        ask: Some(dec!(100.00)),
        bid_size: Some(dec!(100)),
        ask_size: Some(dec!(100)),
        bid_exchange: None,
        ask_exchange: None,
        conditions: vec![],
        timestamp: Utc::now(),
        sequence: 1,
    });

    let result = processor.process_event(invalid_quote).await;
    assert!(result.is_err());
    assert_eq!(result.unwrap_err(), "Invalid quote spread");

    let (processed, filtered, errors) = processor.get_stats();
    assert_eq!(processed, 0);
    assert_eq!(errors, 1);
}

/// Test databento event conversion to core events
#[tokio::test]
async fn test_databento_to_core_conversion() {
    let provider = DatabentoStreamingProvider::new("test-key".to_string()).unwrap();
    let mut receiver = provider.subscribe_market_events();

    let databento_trade = DatabentoTrade {
        symbol: "NVDA".to_string(),
        timestamp: Utc::now(),
        price: Price::from(875.50),
        size: Quantity::from(200),
        trade_id: Some("dt123".to_string()),
        exchange: Some("NASDAQ".to_string()),
        conditions: Some(vec!["Normal".to_string()]),
    };

    let message = DatabentoMessage::Trade(databento_trade.clone());
    provider.process_databento_message(message).await.unwrap();

    let core_event = timeout(Duration::from_millis(100), receiver.recv()).await.unwrap().unwrap();

    match core_event {
        CoreMarketDataEvent::Trade(trade) => {
            assert_eq!(trade.symbol, databento_trade.symbol);
            assert_eq!(trade.price, databento_trade.price);
            assert_eq!(trade.size, databento_trade.size);
            assert_eq!(trade.exchange, databento_trade.exchange);
        }
        _ => panic!("Expected trade event"),
    }
}

/// Test high-frequency event processing
#[tokio::test]
async fn test_high_frequency_event_processing() {
    let mut aggregator = EventAggregator::new(1000);
    let mut receiver = aggregator.subscribe();
    
    let start_time = Instant::now();
    let num_events = 500;

    // Generate high-frequency trade events
    for i in 0..num_events {
        let trade = TradeEvent {
            symbol: Symbol::from("SPY"),
            price: dec!(400.00) + dec!(0.01) * dec!(i % 100),
            size: dec!(100),
            exchange: "NYSE".to_string(),
            conditions: vec![],
            trade_id: Some(format!("hf_trade_{}", i)),
            timestamp: Utc::now(),
            sequence: i as u64,
        };
        
        aggregator.add_trade(trade).unwrap();
    }

    let processing_time = start_time.elapsed();
    
    assert_eq!(aggregator.get_trade_count(), num_events);
    
    // Should process events quickly (under 100ms for 500 events)
    assert!(processing_time < Duration::from_millis(100));
    
    // Verify events can be received
    let mut received_count = 0;
    while let Ok(Ok(_)) = timeout(Duration::from_millis(1), receiver.recv()).await {
        received_count += 1;
        if received_count >= num_events {
            break;
        }
    }
    
    assert_eq!(received_count, num_events);
}

/// Test event ordering preservation
#[tokio::test]
async fn test_event_ordering_preservation() {
    let mut aggregator = EventAggregator::new(100);
    let mut receiver = aggregator.subscribe();

    let symbols = vec!["AAPL", "MSFT", "GOOGL"];
    
    // Add events with increasing sequence numbers
    for (i, symbol) in symbols.iter().enumerate() {
        let trade = TradeEvent {
            symbol: Symbol::from(*symbol),
            price: dec!(100.00),
            size: dec!(100),
            exchange: "NASDAQ".to_string(),
            conditions: vec![],
            trade_id: Some(format!("ordered_trade_{}", i)),
            timestamp: Utc::now(),
            sequence: i as u64,
        };
        
        aggregator.add_trade(trade).unwrap();
    }

    // Receive events and verify order
    for i in 0..symbols.len() {
        let event = timeout(Duration::from_millis(100), receiver.recv()).await.unwrap().unwrap();
        
        match event {
            MarketDataEvent::Trade(trade) => {
                assert_eq!(trade.sequence, i as u64);
                assert_eq!(trade.symbol, Symbol::from(symbols[i]));
            }
            _ => panic!("Expected trade event"),
        }
    }
}

/// Test concurrent event processing
#[tokio::test]
async fn test_concurrent_event_processing() {
    let aggregator = Arc::new(tokio::sync::Mutex::new(EventAggregator::new(1000)));
    let mut handles = vec![];
    
    // Spawn multiple tasks adding events concurrently
    for task_id in 0..5 {
        let aggregator_clone = Arc::clone(&aggregator);
        let handle = tokio::spawn(async move {
            for i in 0..20 {
                let trade = TradeEvent {
                    symbol: Symbol::from(format!("SYM{}", task_id)),
                    price: dec!(100.00) + dec!(task_id) + dec!(i),
                    size: dec!(100),
                    exchange: "TEST".to_string(),
                    conditions: vec![],
                    trade_id: Some(format!("concurrent_{}_{}", task_id, i)),
                    timestamp: Utc::now(),
                    sequence: (task_id * 20 + i) as u64,
                };
                
                let mut agg = aggregator_clone.lock().await;
                agg.add_trade(trade).unwrap();
            }
        });
        handles.push(handle);
    }
    
    // Wait for all tasks to complete
    for handle in handles {
        handle.await.unwrap();
    }
    
    let final_aggregator = aggregator.lock().await;
    assert_eq!(final_aggregator.get_trade_count(), 100); // 5 tasks * 20 events each
}

/// Test memory usage with large event volumes
#[tokio::test]
async fn test_memory_usage_large_volumes() {
    let buffer_size = 10000;
    let mut aggregator = EventAggregator::new(buffer_size);
    
    // Add more events than buffer size to test memory bounds
    for i in 0..buffer_size * 2 {
        let trade = TradeEvent {
            symbol: Symbol::from("MEMORY_TEST"),
            price: dec!(100.00),
            size: dec!(100),
            exchange: "TEST".to_string(),
            conditions: vec![],
            trade_id: Some(format!("memory_trade_{}", i)),
            timestamp: Utc::now(),
            sequence: i as u64,
        };
        
        aggregator.add_trade(trade).unwrap();
    }
    
    // Should be capped at buffer size
    assert_eq!(aggregator.get_trade_count(), buffer_size);
}

/// Test event conversion accuracy
#[tokio::test]
async fn test_event_conversion_accuracy() {
    let original_trade = TradeEvent {
        symbol: Symbol::from("CONVERSION_TEST"),
        price: dec!(123.456789),
        size: dec!(987.654321),
        exchange: "ACCURACY_EXCHANGE".to_string(),
        conditions: vec![1, 2, 3, 4],
        trade_id: Some("precise_trade_id".to_string()),
        timestamp: Utc::now(),
        sequence: 999999999,
    };

    // Convert to MarketDataEvent and back
    let market_event = MarketDataEvent::Trade(original_trade.clone());
    
    match market_event {
        MarketDataEvent::Trade(converted_trade) => {
            assert_eq!(converted_trade.symbol, original_trade.symbol);
            assert_eq!(converted_trade.price, original_trade.price);
            assert_eq!(converted_trade.size, original_trade.size);
            assert_eq!(converted_trade.exchange, original_trade.exchange);
            assert_eq!(converted_trade.conditions, original_trade.conditions);
            assert_eq!(converted_trade.trade_id, original_trade.trade_id);
            assert_eq!(converted_trade.sequence, original_trade.sequence);
        }
        _ => panic!("Conversion failed"),
    }
}

/// Test stream processing with backpressure
#[tokio::test]
async fn test_stream_processing_with_backpressure() {
    let (tx, mut rx) = mpsc::channel::<MarketDataEvent>(10); // Small buffer for backpressure
    
    // Spawn a slow consumer
    let consumer_handle = tokio::spawn(async move {
        let mut received = 0;
        while let Some(_event) = rx.recv().await {
            sleep(Duration::from_millis(10)).await; // Slow processing
            received += 1;
            if received >= 5 {
                break;
            }
        }
        received
    });
    
    // Try to send many events quickly
    let mut sent = 0;
    for i in 0..20 {
        let trade = MarketDataEvent::Trade(TradeEvent {
            symbol: Symbol::from("BACKPRESSURE_TEST"),
            price: dec!(100.00),
            size: dec!(100),
            exchange: "TEST".to_string(),
            conditions: vec![],
            trade_id: Some(format!("bp_trade_{}", i)),
            timestamp: Utc::now(),
            sequence: i,
        });
        
        // Use try_send to detect backpressure
        match tx.try_send(trade) {
            Ok(_) => sent += 1,
            Err(_) => break, // Channel full, backpressure detected
        }
    }
    
    let received = consumer_handle.await.unwrap();
    
    // Should have hit backpressure before sending all events
    assert!(sent < 20);
    assert_eq!(received, 5);
}