//! # Benzinga HFT Integration Module
//!
//! This module provides the complete integration between Benzinga news/sentiment data
//! and the Foxhunt HFT trading system, including ML model integration and event-driven
//! trading signals.
//!
//! ## Architecture
//!
//! This integration module orchestrates:
//! - **Real-time Data Ingestion**: WebSocket streaming from Benzinga Pro API
//! - **Historical Data Backfill**: REST API for historical news and sentiment
//! - **ML Feature Pipeline**: Real-time feature extraction for TFT and Liquid Networks
//! - **Trading Signal Generation**: Event-driven signals from news impact analysis
//! - **Configuration Management**: Integration with Foxhunt config system
//! - **Performance Optimization**: HFT-optimized processing with sub-millisecond latency
//!
//! ## Usage
//!
//! ```rust,no_run
//! use data::providers::benzinga::integration::BenzingaHFTIntegration;
//! use config::ConfigManager;
//! use common::Symbol;
//!
//! # async fn example() -> anyhow::Result<()> {
//! // Initialize with configuration
//! let config_manager = ConfigManager::new().await?;
//! let mut integration = BenzingaHFTIntegration::new(config_manager).await?;
//!
//! // Start the integration
//! integration.start().await?;
//!
//! // Subscribe to symbols for news/sentiment analysis
//! integration.subscribe_symbols(vec![
//!     Symbol::from("AAPL"),
//!     Symbol::from("SPY"),
//!     Symbol::from("TSLA"),
//! ]).await?;
//!
//! // Process real-time events
//! let mut signal_stream = integration.get_trading_signals().await?;
//! while let Some(signal) = signal_stream.next().await {
//!     // Process trading signals derived from news/sentiment
//!     match signal {
//!         TradingSignal::NewsImpact { symbol, impact, confidence } => {
//!             println!("News impact for {}: {} (confidence: {})", symbol, impact, confidence);
//!         }
//!         TradingSignal::SentimentShift { symbol, sentiment_change, momentum } => {
//!             println!("Sentiment shift for {}: {} (momentum: {})", symbol, sentiment_change, momentum);
//!         }
//!         TradingSignal::AnalystAction { symbol, action, price_target_change } => {
//!             println!("Analyst action for {}: {:?} (PT change: {:?})", symbol, action, price_target_change);
//!         }
//!     }
//! }
//! # Ok(())
//! # }
//! ```

use crate::error::Result;
use crate::providers::benzinga::ml_integration::{
    BenzingaFeatureVector, BenzingaMLConfig, BenzingaMLExtractor,
};
use crate::providers::benzinga::production_historical::{
    ProductionBenzingaHistoricalConfig, ProductionBenzingaHistoricalProvider,
};
use crate::providers::benzinga::production_streaming::{
    ProductionBenzingaConfig, ProductionBenzingaProvider,
};
use crate::types::ExtendedMarketDataEvent;
// use crate::providers::traits::RealTimeProvider;
use common::Symbol;
use config::{data_config::TrainingBenzingaConfig, manager::ConfigManager};
use rust_decimal::Decimal;
// use tokio_stream::StreamExt;
use chrono::{DateTime, Duration as ChronoDuration, Utc};
use futures_util::stream::BoxStream;
use serde::{Deserialize, Serialize};
use std::collections::{HashMap, VecDeque};
use std::sync::Arc;
use tokio::sync::{mpsc, Mutex, RwLock};
use tracing::{debug, info, instrument};

/// Trading signals generated from Benzinga data analysis
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum TradingSignal {
    /// News impact signal with calculated market effect
    NewsImpact {
        symbol: Symbol,
        impact: f64,
        confidence: f64,
        category: String,
        headline: String,
        timestamp: DateTime<Utc>,
    },

    /// Sentiment shift signal based on momentum analysis
    SentimentShift {
        symbol: Symbol,
        sentiment_change: f64,
        momentum: f64,
        sample_size: u32,
        timestamp: DateTime<Utc>,
    },

    /// Analyst rating action with price target implications
    AnalystAction {
        symbol: Symbol,
        action: String,
        price_target_change: Option<Decimal>,
        firm: String,
        confidence: f64,
        timestamp: DateTime<Utc>,
    },

    /// Unusual options activity with directional bias
    OptionsFlow {
        symbol: Symbol,
        activity_type: String,
        sentiment: String,
        volume_impact: f64,
        confidence: f64,
        timestamp: DateTime<Utc>,
    },
}

/// ML model integration for feature processing
#[derive(Debug)]
pub struct MLModelIntegration {
    /// `TFT` model for temporal sequence prediction
    tft_features: Arc<Mutex<VecDeque<BenzingaFeatureVector>>>,

    /// Liquid Networks for adaptive learning
    liquid_features: Arc<Mutex<VecDeque<BenzingaFeatureVector>>>,

    /// Feature extraction pipeline
    feature_extractor: Arc<Mutex<BenzingaMLExtractor>>,

    /// Model prediction cache
    prediction_cache: Arc<RwLock<HashMap<Symbol, ModelPredictions>>>,
}

/// Model predictions for a symbol
#[derive(Debug, Clone)]
struct ModelPredictions {
    /// `TFT` predictions (price movement probability)
    tft_prediction: Option<f64>,

    /// Liquid Networks prediction (adaptive sentiment)
    liquid_prediction: Option<f64>,

    /// Ensemble confidence score
    ensemble_confidence: f64,

    /// Prediction timestamp
    timestamp: DateTime<Utc>,
}

/// Signal generation configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SignalConfig {
    /// Minimum news importance to generate signal
    pub min_news_importance: f64,

    /// Minimum sentiment change to generate signal
    pub min_sentiment_change: f64,

    /// Minimum confidence for signal generation
    pub min_confidence: f64,

    /// Signal cooldown period (prevent spam)
    pub signal_cooldown_secs: u64,

    /// Enable ML-enhanced signals
    pub enable_ml_signals: bool,

    /// Maximum signals per symbol per minute
    pub max_signals_per_minute: u32,
}

impl Default for SignalConfig {
    fn default() -> Self {
        Self {
            min_news_importance: 0.6,
            min_sentiment_change: 0.1,
            min_confidence: 0.7,
            signal_cooldown_secs: 30,
            enable_ml_signals: true,
            max_signals_per_minute: 10,
        }
    }
}

/// Comprehensive Benzinga HFT Integration
pub struct BenzingaHFTIntegration {
    /// Configuration manager
    config_manager: Arc<ConfigManager>,

    /// Real-time streaming provider
    streaming_provider: Arc<Mutex<Option<ProductionBenzingaProvider>>>,

    /// Historical data provider
    historical_provider: Arc<ProductionBenzingaHistoricalProvider>,

    /// ML model integration
    ml_integration: Arc<MLModelIntegration>,

    /// Signal generation configuration
    signal_config: SignalConfig,

    /// Trading signal sender
    signal_tx: Arc<Mutex<Option<mpsc::UnboundedSender<TradingSignal>>>>,

    /// Signal rate limiting
    signal_rate_limiter: Arc<RwLock<HashMap<Symbol, VecDeque<DateTime<Utc>>>>>,

    /// Active subscriptions
    subscribed_symbols: Arc<RwLock<Vec<Symbol>>>,

    /// Integration metrics
    metrics: Arc<RwLock<IntegrationMetrics>>,

    /// Shutdown signal
    shutdown_tx: Arc<Mutex<Option<mpsc::UnboundedSender<()>>>>,
}

/// Integration performance metrics
#[derive(Debug, Default)]
pub struct IntegrationMetrics {
    /// Total events processed
    events_processed: u64,

    /// Trading signals generated
    signals_generated: u64,

    /// ML features extracted
    features_extracted: u64,

    /// Model predictions made
    predictions_made: u64,

    /// Average processing latency (microseconds)
    avg_processing_latency_us: u64,

    /// Error count
    error_count: u64,

    /// Last activity timestamp
    last_activity: Option<DateTime<Utc>>,
}

impl BenzingaHFTIntegration {
    /// Create new Benzinga HFT integration
    #[instrument(skip(config_manager))]
    pub async fn new(config_manager: ConfigManager) -> Result<Self> {
        info!("Initializing Benzinga HFT Integration");

        let config_manager = Arc::new(config_manager);

        // Get Benzinga configuration from config manager or use default
        let training_config = TrainingBenzingaConfig::default();

        // Create streaming provider configuration
        let streaming_config = ProductionBenzingaConfig {
            api_key: std::env::var(&training_config.api_key_env).unwrap_or_default(),
            enable_news: training_config.data_types.contains(&"news".to_string()),
            enable_sentiment: training_config
                .data_types
                .contains(&"sentiment".to_string()),
            enable_ratings: training_config.data_types.contains(&"ratings".to_string()),
            enable_options: training_config.data_types.contains(&"options".to_string()),
            rate_limit_per_second: training_config.rate_limit as u32,
            enable_ml_integration: true,
            enable_smart_categorization: true,
            ..Default::default()
        };

        // Create historical provider configuration
        let historical_config = ProductionBenzingaHistoricalConfig {
            api_key: std::env::var(&training_config.api_key_env).unwrap_or_default(),
            rate_limit_per_second: (training_config.rate_limit / 2) as u32, // More conservative for historical
            enable_caching: true,
            enable_bulk_download: true,
            ..Default::default()
        };

        // Create ML configuration
        let ml_config = BenzingaMLConfig {
            feature_window_minutes: 60,
            enable_nlp_features: true,
            enable_sentiment_indicators: true,
            enable_adaptive_features: true,
            ..Default::default()
        };

        // Initialize providers
        let streaming_provider = ProductionBenzingaProvider::new(streaming_config)?;
        let historical_provider = Arc::new(ProductionBenzingaHistoricalProvider::new(
            historical_config,
        )?);

        // Initialize ML integration
        let ml_integration = Arc::new(MLModelIntegration {
            tft_features: Arc::new(Mutex::new(VecDeque::new())),
            liquid_features: Arc::new(Mutex::new(VecDeque::new())),
            feature_extractor: Arc::new(Mutex::new(BenzingaMLExtractor::new(ml_config))),
            prediction_cache: Arc::new(RwLock::new(HashMap::new())),
        });

        let (signal_tx, _signal_rx) = mpsc::unbounded_channel();

        Ok(Self {
            config_manager,
            streaming_provider: Arc::new(Mutex::new(Some(streaming_provider))),
            historical_provider,
            ml_integration,
            signal_config: SignalConfig::default(),
            signal_tx: Arc::new(Mutex::new(Some(signal_tx))),
            signal_rate_limiter: Arc::new(RwLock::new(HashMap::new())),
            subscribed_symbols: Arc::new(RwLock::new(Vec::new())),
            metrics: Arc::new(RwLock::new(IntegrationMetrics::default())),
            shutdown_tx: Arc::new(Mutex::new(None)),
        })
    }

    /// Start the integration
    #[instrument(skip(self))]
    pub async fn start(&mut self) -> Result<()> {
        info!("Starting Benzinga HFT Integration");

        // Start streaming provider - comment out for now until provider is fully implemented
        /*
        {
            let mut provider_guard = self.streaming_provider.lock().await;
            if let Some(provider) = provider_guard.as_mut() {
                provider.connect().await?;
            }
        }
        */

        // Start event processing loop
        // self.start_event_processing().await?;

        // Start ML feature processing
        // self.start_ml_processing().await?;

        info!("Benzinga HFT Integration started successfully (streaming disabled)");
        Ok(())
    }

    /// Subscribe to symbols for news/sentiment analysis
    #[instrument(skip(self))]
    pub async fn subscribe_symbols(&mut self, symbols: Vec<Symbol>) -> Result<()> {
        info!("Subscribing to {} symbols for Benzinga data", symbols.len());

        // Subscribe to streaming data - commented out until provider is fully implemented
        /*
        {
            let mut provider_guard = self.streaming_provider.lock().await;
            if let Some(provider) = provider_guard.as_mut() {
                provider.subscribe(symbols.clone()).await?;
            }
        }
        */

        // Update subscriptions
        {
            let mut subscriptions = self.subscribed_symbols.write().await;
            for symbol in symbols {
                if !subscriptions.contains(&symbol) {
                    subscriptions.push(symbol);
                }
            }
        }

        info!("Successfully subscribed to symbols (streaming disabled)");
        Ok(())
    }

    /// Get trading signals stream
    pub async fn get_trading_signals(&self) -> Result<BoxStream<'static, TradingSignal>> {
        let (tx, rx) = mpsc::unbounded_channel();

        // Store the new sender
        {
            let mut signal_tx_guard = self.signal_tx.lock().await;
            *signal_tx_guard = Some(tx);
        }

        let stream = tokio_stream::wrappers::UnboundedReceiverStream::new(rx);
        Ok(Box::pin(stream))
    }

    /// Start event processing loop
    async fn start_event_processing(&self) -> Result<()> {
        // Commented out until streaming provider is fully implemented
        /*
        let (shutdown_tx, mut shutdown_rx) = mpsc::unbounded_channel();

        // Store shutdown sender
        {
            let mut tx = self.shutdown_tx.lock().await;
            *tx = Some(shutdown_tx);
        }

        let streaming_provider = self.streaming_provider.clone();
        let ml_integration = self.ml_integration.clone();
        let signal_tx = self.signal_tx.clone();
        let signal_config = self.signal_config.clone();
        let signal_rate_limiter = self.signal_rate_limiter.clone();
        let metrics = self.metrics.clone();

        tokio::spawn(async move {
            // Get event stream from provider
            let mut event_stream = {
                let mut provider_guard = streaming_provider.lock().await;
                if let Some(provider) = provider_guard.as_mut() {
                    match provider.stream().await {
                        Ok(stream) => stream,
                        Err(e) => {
                            error!("Failed to get event stream: {}", e);
                            return;
                        }
                    }
                } else {
                    error!("Streaming provider not available");
                    return;
                }
            };

            loop {
                tokio::select! {
                    // Handle shutdown
                    _ = shutdown_rx.recv() => {
                        info!("Received shutdown signal for event processing");
                        break;
                    }

                    // Process events
                    event = event_stream.next() => {
                        if let Some(event) = event {
                            let start_time = std::time::Instant::now();

                            // Update metrics
                            {
                                let mut m = metrics.write().await;
                                m.events_processed += 1;
                                m.last_activity = Some(Utc::now());
                            }

                            // Process event for ML features
                            {
                                let feature_extractor = ml_integration.feature_extractor.clone();
                                let extractor = feature_extractor.lock().await;
                                let extended_event = ExtendedMarketDataEvent::Core(event.clone());
                                if let Err(e) = extractor.process_event(&extended_event).await {
                                    error!("Failed to process event for ML: {}", e);
                                }
                            }

                            // Generate trading signals
                            let extended_event = ExtendedMarketDataEvent::Core(event.clone());
                            if let Some(signal) = Self::generate_trading_signal(
                                &extended_event,
                                &signal_config,
                                &signal_rate_limiter,
                            ).await {
                                // Send signal
                                if let Some(tx) = signal_tx.lock().await.as_ref() {
                                    if let Err(e) = tx.send(signal) {
                                        error!("Failed to send trading signal: {}", e);
                                    } else {
                                        // Update metrics
                                        let mut m = metrics.write().await;
                                        m.signals_generated += 1;
                                    }
                                }
                            }

                            // Update processing latency metrics
                            {
                                let mut m = metrics.write().await;
                                let latency = start_time.elapsed().as_micros() as u64;
                                m.avg_processing_latency_us =
                                    (m.avg_processing_latency_us + latency) / 2;
                            }
                        }
                    }
                }
            }

            info!("Event processing loop ended");
        });
        */

        Ok(())
    }

    /// Start ML feature processing
    async fn start_ml_processing(&self) -> Result<()> {
        // Commented out until ML integration is fully implemented
        /*
        let ml_integration = self.ml_integration.clone();
        let subscribed_symbols = self.subscribed_symbols.clone();
        let metrics = self.metrics.clone();

        tokio::spawn(async move {
            let mut interval = tokio::time::interval(std::time::Duration::from_secs(60));

            loop {
                interval.tick().await;

                let symbols = {
                    let subs = subscribed_symbols.read().await;
                    subs.clone()
                };

                for symbol in symbols {
                    // Extract features for this symbol
                    let features_result = {
                        let feature_extractor = ml_integration.feature_extractor.lock().await;
                        feature_extractor.extract_features(&symbol, Utc::now()).await
                    };

                    match features_result {
                        Ok(feature_vector) => {
                            // Add to TFT feature queue
                            {
                                let mut tft_features = ml_integration.tft_features.lock().await;
                                tft_features.push_back(feature_vector.clone());

                                // Limit queue size
                                if tft_features.len() > 1000 {
                                    tft_features.pop_front();
                                }
                            }

                            // Add to Liquid Networks feature queue
                            {
                                let mut liquid_features = ml_integration.liquid_features.lock().await;
                                liquid_features.push_back(feature_vector.clone());

                                // Limit queue size
                                if liquid_features.len() > 500 {
                                    liquid_features.pop_front();
                                }
                            }

                            // Update metrics
                            {
                                let mut m = metrics.write().await;
                                m.features_extracted += 1;
                            }

                            debug!("Extracted ML features for symbol: {}", symbol);
                        }
                        Err(e) => {
                            debug!("Failed to extract features for {}: {}", symbol, e);
                        }
                    }
                }
            }
        });
        */

        Ok(())
    }

    /// Generate trading signal from market data event
    async fn generate_trading_signal(
        event: &ExtendedMarketDataEvent,
        signal_config: &SignalConfig,
        rate_limiter: &Arc<RwLock<HashMap<Symbol, VecDeque<DateTime<Utc>>>>>,
    ) -> Option<TradingSignal> {
        let symbol = event.symbol();
        let now = Utc::now();

        // Check rate limiting
        {
            let mut limiter = rate_limiter.write().await;
            let signal_times = limiter.entry(symbol.into()).or_insert_with(VecDeque::new);

            // Clean old signals
            let cutoff = now - ChronoDuration::seconds(60);
            signal_times.retain(|&time| time > cutoff);

            // Check if we've hit the rate limit
            if signal_times.len() >= signal_config.max_signals_per_minute as usize {
                debug!("Rate limit reached for symbol: {}", symbol);
                return None;
            }

            // Record this signal time
            signal_times.push_back(now);
        }

        match event {
            ExtendedMarketDataEvent::NewsAlert(news) => {
                if let Some(impact_score) = news.impact_score {
                    if impact_score.abs() >= signal_config.min_news_importance {
                        let confidence = impact_score.abs().min(1.0);

                        if confidence >= signal_config.min_confidence {
                            return Some(TradingSignal::NewsImpact {
                                symbol: symbol.into(),
                                impact: impact_score,
                                confidence,
                                category: news.category.clone(),
                                headline: news.headline.clone(),
                                timestamp: news.timestamp,
                            });
                        }
                    }
                }
            },

            ExtendedMarketDataEvent::SentimentUpdate(sentiment) => {
                // Calculate sentiment momentum (simplified)
                let sentiment_momentum = sentiment.sentiment_score * 0.5; // Placeholder calculation

                if sentiment_momentum.abs() >= signal_config.min_sentiment_change {
                    let confidence = sentiment.confidence;

                    if confidence >= signal_config.min_confidence {
                        return Some(TradingSignal::SentimentShift {
                            symbol: symbol.into(),
                            sentiment_change: sentiment.sentiment_score,
                            momentum: sentiment_momentum,
                            sample_size: sentiment.sample_size,
                            timestamp: sentiment.timestamp,
                        });
                    }
                }
            },

            ExtendedMarketDataEvent::AnalystRating(rating) => {
                let action_score: f64 = match rating.action.to_string().as_str() {
                    "Upgrade" => 1.0,
                    "Downgrade" => -1.0,
                    "Initiate" => 0.5,
                    _ => 0.0,
                };

                if action_score.abs() >= 0.5 {
                    return Some(TradingSignal::AnalystAction {
                        symbol: symbol.into(),
                        action: rating.action.to_string(),
                        price_target_change: rating
                            .price_target
                            .map(|p| p.to_decimal().unwrap_or(Decimal::ZERO)),
                        firm: rating.firm.clone(),
                        confidence: 0.8, // Default confidence for analyst actions
                        timestamp: rating.timestamp,
                    });
                }
            },

            ExtendedMarketDataEvent::UnusualOptions(options) => {
                if options.confidence >= signal_config.min_confidence {
                    let volume_impact = (options.volume.as_f64()).ln() / 10.0; // Log-normalized volume impact

                    return Some(TradingSignal::OptionsFlow {
                        symbol: symbol.into(),
                        activity_type: format!("{:?}", options.activity_type),
                        sentiment: format!("{:?}", options.sentiment),
                        volume_impact,
                        confidence: options.confidence,
                        timestamp: options.timestamp,
                    });
                }
            },

            _ => {}, // Other event types don't generate signals
        }

        None
    }

    /// Get ML features for `TFT` model
    pub async fn get_tft_features(&self, limit: usize) -> Vec<BenzingaFeatureVector> {
        let tft_features = self.ml_integration.tft_features.lock().await;
        tft_features.iter().rev().take(limit).cloned().collect()
    }

    /// Get ML features for Liquid Networks
    pub async fn get_liquid_features(&self, limit: usize) -> Vec<BenzingaFeatureVector> {
        let liquid_features = self.ml_integration.liquid_features.lock().await;
        liquid_features.iter().rev().take(limit).cloned().collect()
    }

    /// Get historical data for backtesting
    #[instrument(skip(self))]
    pub async fn get_historical_events(
        &self,
        symbols: &[Symbol],
        start: DateTime<Utc>,
        end: DateTime<Utc>,
    ) -> Result<Vec<ExtendedMarketDataEvent>> {
        let symbol_strs: Vec<&str> = symbols.iter().map(|s| s.as_str()).collect();

        let events = self
            .historical_provider
            .get_all_events(Some(&symbol_strs), start, end)
            .await?;

        info!("Retrieved {} historical events from Benzinga", events.len());
        Ok(events)
    }

    /// Get integration metrics
    pub async fn get_metrics(&self) -> IntegrationMetrics {
        let metrics = self.metrics.read().await;
        IntegrationMetrics {
            events_processed: metrics.events_processed,
            signals_generated: metrics.signals_generated,
            features_extracted: metrics.features_extracted,
            predictions_made: metrics.predictions_made,
            avg_processing_latency_us: metrics.avg_processing_latency_us,
            error_count: metrics.error_count,
            last_activity: metrics.last_activity,
        }
    }

    /// Stop the integration
    pub async fn stop(&mut self) -> Result<()> {
        info!("Stopping Benzinga HFT Integration");

        // Send shutdown signal
        if let Some(tx) = self.shutdown_tx.lock().await.take() {
            let _ = tx.send(());
        }

        // Disconnect streaming provider - commented out until provider is fully implemented
        /*
        {
            let mut provider_guard = self.streaming_provider.lock().await;
            if let Some(provider) = provider_guard.as_mut() {
                provider.disconnect().await?;
            }
        }
        */

        info!("Benzinga HFT Integration stopped");
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    // use config::ConfigManager;

    #[test]
    fn test_signal_config_default() {
        let config = SignalConfig::default();
        assert!(config.min_news_importance > 0.0);
        assert!(config.min_sentiment_change > 0.0);
        assert!(config.min_confidence > 0.0);
        assert!(config.signal_cooldown_secs > 0);
        assert!(config.max_signals_per_minute > 0);
    }

    #[test]
    fn test_trading_signal_serialization() {
        let signal = TradingSignal::NewsImpact {
            symbol: Symbol::from("AAPL"),
            impact: 0.75,
            confidence: 0.85,
            category: "earnings".to_string(),
            headline: "Apple beats earnings expectations".to_string(),
            timestamp: Utc::now(),
        };

        let json = serde_json::to_string(&signal).unwrap();
        let deserialized: TradingSignal = serde_json::from_str(&json).unwrap();

        match deserialized {
            TradingSignal::NewsImpact {
                symbol,
                impact,
                confidence,
                ..
            } => {
                assert_eq!(symbol, Symbol::from("AAPL"));
                assert!((impact - 0.75).abs() < 0.001);
                assert!((confidence - 0.85).abs() < 0.001);
            },
            _ => panic!("Expected NewsImpact signal"),
        }
    }

    #[test]
    fn test_integration_metrics_default() {
        let metrics = IntegrationMetrics::default();
        assert_eq!(metrics.events_processed, 0);
        assert_eq!(metrics.signals_generated, 0);
        assert_eq!(metrics.features_extracted, 0);
        assert_eq!(metrics.predictions_made, 0);
        assert_eq!(metrics.error_count, 0);
    }

    // Note: Full integration tests would require API keys and actual Benzinga access
    // These would be run in a separate integration test suite
}
