foxhunt/tli/src/client/stream_manager.rs

//! Data Stream Manager for Real-time Dashboard Updates
//!
//! Manages real-time data streams from gRPC services to dashboard components

use crate::dashboard::events::{
    DashboardEvent, MLPredictionEvent, MarketDataDisplayEvent, PredictionType, RiskMetricsEvent,
    SystemStatusEvent,
};
use anyhow::Result;
use rand::Rng;
use std::collections::HashMap;
use tokio::sync::mpsc;
use tokio::time::{interval, Duration};

pub struct DataStreamManager {
    _event_sender: mpsc::Sender<DashboardEvent>,
    is_running: bool,
}

impl DataStreamManager {
    pub const fn new(_event_sender: mpsc::Sender<DashboardEvent>) -> Self {
        Self {
            _event_sender,
            is_running: false,
        }
    }

    /// Start all data streams for real-time dashboard updates
    pub async fn start_streams(&mut self) -> Result<()> {
        if self.is_running {
            return Ok(());
        }

        self.is_running = true;

        // Spawn individual stream tasks
        let market_data_task = self.spawn_market_data_stream();
        let risk_metrics_task = self.spawn_risk_metrics_stream();
        let ml_predictions_task = self.spawn_ml_predictions_stream();
        let system_status_task = self.spawn_system_status_stream();

        // Start all streams concurrently
        tokio::try_join!(
            market_data_task,
            risk_metrics_task,
            ml_predictions_task,
            system_status_task
        )?;

        Ok(())
    }

    /// Generate mock market data stream for demo purposes
    async fn spawn_market_data_stream(&self) -> Result<()> {
        let mut ticker = interval(Duration::from_millis(1000));
        let sender = self._event_sender.clone();
        let symbols = vec!["AAPL", "TSLA", "SPY", "QQQ", "NVDA"];
        let mut prices: HashMap<&str, f64> = HashMap::new();

        // Initialize prices
        prices.insert("AAPL", 150.25);
        prices.insert("TSLA", 800.50);
        prices.insert("SPY", 420.10);
        prices.insert("QQQ", 350.75);
        prices.insert("NVDA", 450.30);

        tokio::spawn(async move {
            loop {
                ticker.tick().await;

                // Generate all updates in a single scope without holding RNG across await
                let updates: Vec<MarketDataDisplayEvent> = {
                    let mut rng = rand::thread_rng();
                    let mut updates = Vec::new();

                    for symbol in &symbols {
                        if let Some(current_price) = prices.get_mut(symbol) {
                            // Simulate price movement (±0.5%)
                            let change_pct = (rng.gen::<f64>() - 0.5) * 0.01;
                            *current_price *= 1.0 + change_pct;

                            let market_data = MarketDataDisplayEvent {
                                symbol: symbol.to_string(),
                                price: *current_price,
                                volume: rng.gen_range(500_000..1_500_000),
                                timestamp: chrono::Utc::now().timestamp(),
                                bid: Some(*current_price - 0.01),
                                ask: Some(*current_price + 0.01),
                                change: Some(change_pct * *current_price),
                                change_percent: Some(change_pct * 100.0),
                            };

                            updates.push(market_data);
                        }
                    }
                    updates
                };

                // Send all updates
                for market_data in updates {
                    let _ = sender
                        .send(DashboardEvent::MarketDataUpdate(market_data))
                        .await;
                }
            }
        });
        Ok(())
    }

    /// Generate mock risk metrics stream
    async fn spawn_risk_metrics_stream(&self) -> Result<()> {
        let mut ticker = interval(Duration::from_millis(5000));
        let sender = self._event_sender.clone();

        tokio::spawn(async move {
            let mut portfolio_value = 1_000_000.0;

            loop {
                ticker.tick().await;

                // Generate risk metrics in a single scope
                let risk_metrics = {
                    let mut rng = rand::thread_rng();
                    let pnl_change = (rng.gen::<f64>() - 0.5) * 5000.0;
                    let var_1d_rand = rng.gen::<f64>() * 1000.0;
                    let var_5d_rand = rng.gen::<f64>() * 1500.0;
                    let dd_rand = (rng.gen::<f64>() - 0.5) * 0.02;
                    let risk_rand = rng.gen::<f64>() * 0.4;

                    portfolio_value += pnl_change;

                    RiskMetricsEvent {
                        portfolio_value,
                        daily_pnl: pnl_change,
                        total_pnl: portfolio_value - 1_000_000.0,
                        var_1d: 5000.0 + var_1d_rand,
                        var_5d: 8000.0 + var_5d_rand,
                        max_drawdown: -0.15,
                        current_drawdown: -0.025 + dd_rand,
                        risk_score: 0.3 + risk_rand,
                        timestamp: chrono::Utc::now().timestamp(),
                    }
                };

                let _ = sender
                    .send(DashboardEvent::RiskMetricsUpdate(risk_metrics))
                    .await;
            }
        });
        Ok(())
    }

    /// Generate mock ML predictions stream
    async fn spawn_ml_predictions_stream(&self) -> Result<()> {
        let mut ticker = interval(Duration::from_millis(3000));
        let sender = self._event_sender.clone();
        let models = vec!["DQN", "MAMBA", "TFT", "LIQUID", "TLOB", "PPO"];
        let symbols = vec!["AAPL", "TSLA", "SPY"];

        tokio::spawn(async move {
            loop {
                ticker.tick().await;

                // Generate all predictions in a single scope
                let predictions: Vec<MLPredictionEvent> = {
                    let mut rng = rand::thread_rng();
                    let mut predictions = Vec::new();

                    for model in &models {
                        for symbol in &symbols {
                            let prediction_type = match rng.gen_range(0..3) {
                                0 => PredictionType::Buy,
                                1 => PredictionType::Sell,
                                _ => PredictionType::Hold,
                            };

                            let ml_prediction = MLPredictionEvent {
                                model_name: model.to_string(),
                                symbol: symbol.to_string(),
                                prediction: prediction_type,
                                confidence: 0.5 + (rng.gen::<f64>() * 0.4),
                                signal_strength: rng.gen::<f64>(),
                                features: (0..10).map(|_| rng.gen::<f64>()).collect(),
                                timestamp: chrono::Utc::now().timestamp(),
                            };

                            predictions.push(ml_prediction);
                        }
                    }
                    predictions
                };

                // Send all predictions
                for ml_prediction in predictions {
                    let _ = sender
                        .send(DashboardEvent::MLPredictionUpdate(ml_prediction))
                        .await;
                }
            }
        });
        Ok(())
    }

    /// Generate mock system status updates
    async fn spawn_system_status_stream(&self) -> Result<()> {
        let mut ticker = interval(Duration::from_millis(2000));
        let sender = self._event_sender.clone();

        tokio::spawn(async move {
            loop {
                ticker.tick().await;

                // Generate system status in a single scope
                let system_status = {
                    let mut rng = rand::thread_rng();
                    SystemStatusEvent {
                        trading_enabled: true,
                        risk_controls_active: true,
                        ml_models_online: 6,
                        total_ml_models: 6,
                        active_positions: rng.gen_range(1..11),
                        pending_orders: rng.gen_range(0..5),
                        timestamp: chrono::Utc::now().timestamp(),
                    }
                };

                let _ = sender
                    .send(DashboardEvent::SystemStatus(system_status))
                    .await;
            }
        });
        Ok(())
    }

    pub fn stop(&mut self) {
        self.is_running = false;
    }
}