//! # Benzinga ML Integration Module
//!
//! This module provides ML integration capabilities for Benzinga news/sentiment data,
//! specifically designed to work with TFT (Temporal Fusion Transformer) and
//! Liquid Networks for adaptive learning in HFT systems.
//!
//! ## Features
//!
//! - **News Feature Extraction**: Convert news events to numerical features for ML models
//! - **Sentiment Analysis Integration**: Feed sentiment scores into TFT and Liquid Networks  
//! - **Real-time Feature Streaming**: Continuous feature extraction for live trading
//! - **Backtesting Support**: Historical feature generation for model training
//! - **Adaptive Feature Engineering**: Dynamic feature selection based on market conditions
//! - **Time Series Preparation**: Format data for temporal ML models

use crate::error::{DataError, Result};
use crate::providers::common::{
    AnalystRatingEvent, NewsEvent, OptionsSentiment, RatingAction, SentimentEvent,
    UnusualOptionsEvent,
};
use chrono::{DateTime, Datelike, Duration as ChronoDuration, Timelike, Utc};
use common::Symbol;
use num_traits::ToPrimitive;
use rust_decimal::Decimal;
use rust_decimal_macros::dec;
use serde::{Deserialize, Serialize};
use std::collections::{HashMap, VecDeque};
use std::sync::{
    atomic::{AtomicU64, Ordering},
    Arc,
};
use tokio::sync::RwLock;
use tracing::{debug, info, instrument};

/// Configuration for ML integration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct BenzingaMLConfig {
    /// Feature extraction window in minutes
    pub feature_window_minutes: u64,

    /// Maximum historical lookback in hours
    pub max_lookback_hours: u64,

    /// Minimum news importance for feature extraction
    pub min_news_importance: f64,

    /// Sentiment smoothing factor (0.0 to 1.0)
    pub sentiment_smoothing: f64,

    /// Enable advanced NLP features
    pub enable_nlp_features: bool,

    /// Enable technical indicators on sentiment
    pub enable_sentiment_indicators: bool,

    /// Feature normalization method
    pub normalization_method: NormalizationMethod,

    /// Enable real-time feature streaming
    pub enable_realtime_streaming: bool,

    /// Batch size for feature extraction
    pub feature_batch_size: usize,

    /// Enable adaptive feature selection
    pub enable_adaptive_features: bool,
}

/// Normalization methods for ML features
#[derive(Debug, Clone, Copy, Serialize, Deserialize)]
pub enum NormalizationMethod {
    /// No normalization
    None,
    /// Z-score normalization (mean=0, std=1)
    ZScore,
    /// Min-Max normalization (0 to 1)
    MinMax,
    /// Robust scaling using median and IQR
    Robust,
}

impl Default for BenzingaMLConfig {
    fn default() -> Self {
        Self {
            feature_window_minutes: 60,
            max_lookback_hours: 24 * 7, // 1 week
            min_news_importance: 0.3,
            sentiment_smoothing: 0.8,
            enable_nlp_features: true,
            enable_sentiment_indicators: true,
            normalization_method: NormalizationMethod::ZScore,
            enable_realtime_streaming: true,
            feature_batch_size: 100,
            enable_adaptive_features: true,
        }
    }
}

/// Feature vector for ML models (compatible with `TFT` and Liquid Networks)
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct BenzingaFeatureVector {
    /// Symbol
    pub symbol: Symbol,

    /// Timestamp
    pub timestamp: DateTime<Utc>,

    /// Sequence ID for temporal models
    pub sequence_id: u64,

    // === NEWS FEATURES ===
    /// News volume (count of news items in window)
    pub news_volume: f64,

    /// Average news importance score
    pub news_importance_avg: f64,

    /// Maximum news importance in window
    pub news_importance_max: f64,

    /// News sentiment aggregate
    pub news_sentiment: f64,

    /// News category distribution (encoded)
    pub news_category_encoding: Vec<f64>,

    /// Breaking news indicator (1.0 if breaking news, 0.0 otherwise)
    pub breaking_news_indicator: f64,

    // === SENTIMENT FEATURES ===
    /// Current sentiment score
    pub sentiment_score: f64,

    /// Sentiment momentum (rate of change)
    pub sentiment_momentum: f64,

    /// Sentiment volatility (rolling standard deviation)
    pub sentiment_volatility: f64,

    /// Bullish ratio
    pub bullish_ratio: f64,

    /// Bearish ratio
    pub bearish_ratio: f64,

    /// Sentiment confidence
    pub sentiment_confidence: f64,

    /// Sentiment sample size (log-normalized)
    pub sentiment_sample_size_log: f64,

    // === ANALYST RATING FEATURES ===
    /// Rating change indicator (1=upgrade, -1=downgrade, 0=no change)
    pub rating_change: f64,

    /// Price target change (percentage)
    pub price_target_change_pct: f64,

    /// Analyst consensus score (derived)
    pub analyst_consensus: f64,

    /// Rating volume (count of ratings in window)
    pub rating_volume: f64,

    // === OPTIONS FEATURES ===
    /// Unusual options activity indicator
    pub unusual_options_activity: f64,

    /// Options flow sentiment (bullish/bearish bias)
    pub options_flow_sentiment: f64,

    /// Options volume (normalized)
    pub options_volume_normalized: f64,

    /// Implied volatility signal
    pub iv_signal: f64,

    // === TIME-BASED FEATURES ===
    /// Hour of day (cyclical encoding)
    pub hour_sin: f64,
    pub hour_cos: f64,

    /// Day of week (cyclical encoding)  
    pub day_sin: f64,
    pub day_cos: f64,

    /// Market session indicator (0=closed, 1=pre, 2=regular, 3=post)
    pub market_session: f64,

    // === ADVANCED FEATURES (NLP) ===
    /// Keyword sentiment scores (top 10 financial keywords)
    pub keyword_sentiments: Vec<f64>,

    /// Topic modeling scores (LDA topics)
    pub topic_scores: Vec<f64>,

    /// Named entity sentiment (companies, people, etc.)
    pub entity_sentiments: Vec<f64>,

    // === TECHNICAL INDICATORS ON SENTIMENT ===
    /// Sentiment RSI
    pub sentiment_rsi: f64,

    /// Sentiment moving average (short-term)
    pub sentiment_ma_short: f64,

    /// Sentiment moving average (long-term)
    pub sentiment_ma_long: f64,

    /// Sentiment Bollinger Band position
    pub sentiment_bb_position: f64,

    // === META FEATURES ===
    /// Data quality score (0.0 to 1.0)
    pub data_quality_score: f64,

    /// Feature completeness ratio
    pub feature_completeness: f64,

    /// Regime indicator (bull/bear/neutral market)
    pub market_regime: f64,
}

/// Feature statistics for normalization
#[derive(Debug, Clone)]
struct FeatureStats {
    mean: f64,
    std: f64,
    min: f64,
    max: f64,
    median: f64,
    q25: f64,
    q75: f64,
}

/// Historical data buffer for feature computation
#[derive(Debug)]
struct HistoricalBuffer {
    news_events: VecDeque<NewsEvent>,
    sentiment_events: VecDeque<SentimentEvent>,
    rating_events: VecDeque<AnalystRatingEvent>,
    options_events: VecDeque<UnusualOptionsEvent>,
    last_cleanup: DateTime<Utc>,
}

impl HistoricalBuffer {
    fn new() -> Self {
        Self {
            news_events: VecDeque::new(),
            sentiment_events: VecDeque::new(),
            rating_events: VecDeque::new(),
            options_events: VecDeque::new(),
            last_cleanup: Utc::now(),
        }
    }

    /// Clean expired events
    fn cleanup(&mut self, cutoff_time: DateTime<Utc>) {
        self.news_events.retain(|e| e.timestamp > cutoff_time);
        self.sentiment_events.retain(|e| e.timestamp > cutoff_time);
        self.rating_events.retain(|e| e.timestamp > cutoff_time);
        self.options_events.retain(|e| e.timestamp > cutoff_time);
        self.last_cleanup = Utc::now();
    }
}

/// Benzinga ML feature extractor
#[derive(Debug)]
pub struct BenzingaMLExtractor {
    /// Configuration
    config: BenzingaMLConfig,

    /// Per-symbol historical buffers
    buffers: Arc<RwLock<HashMap<Symbol, HistoricalBuffer>>>,

    /// Feature statistics for normalization  
    feature_stats: Arc<RwLock<HashMap<String, FeatureStats>>>,

    /// Sequence counter for temporal models
    sequence_counter: Arc<AtomicU64>,

    /// Category encoding map (news categories to numerical)
    category_encoding: Arc<RwLock<HashMap<String, usize>>>,

    /// Financial keywords for NLP features
    financial_keywords: Vec<String>,

    /// Cached intermediate calculations
    calculation_cache: Arc<RwLock<HashMap<String, (DateTime<Utc>, f64)>>>,
}

impl BenzingaMLExtractor {
    /// Create new ML feature extractor
    pub fn new(config: BenzingaMLConfig) -> Self {
        let financial_keywords = vec![
            "earnings".to_string(),
            "revenue".to_string(),
            "profit".to_string(),
            "merger".to_string(),
            "acquisition".to_string(),
            "dividend".to_string(),
            "buyback".to_string(),
            "guidance".to_string(),
            "forecast".to_string(),
            "upgrade".to_string(),
            "downgrade".to_string(),
            "target".to_string(),
            "fda".to_string(),
            "approval".to_string(),
            "patent".to_string(),
            "lawsuit".to_string(),
            "regulation".to_string(),
            "competition".to_string(),
            "innovation".to_string(),
            "partnership".to_string(),
        ];

        Self {
            config,
            buffers: Arc::new(RwLock::new(HashMap::new())),
            feature_stats: Arc::new(RwLock::new(HashMap::new())),
            sequence_counter: Arc::new(AtomicU64::new(1)),
            category_encoding: Arc::new(RwLock::new(HashMap::new())),
            financial_keywords,
            calculation_cache: Arc::new(RwLock::new(HashMap::new())),
        }
    }

    /// Process a market data event and update internal state
    #[instrument(skip(self))]
    pub async fn process_event(&self, event: &crate::types::ExtendedMarketDataEvent) -> Result<()> {
        let symbol = Symbol::from(event.symbol());

        let mut buffers = self.buffers.write().await;
        let buffer = buffers.entry(symbol).or_insert_with(HistoricalBuffer::new);

        // Clean up old events periodically
        let cutoff = Utc::now() - ChronoDuration::hours(self.config.max_lookback_hours as i64);
        if buffer.last_cleanup < Utc::now() - ChronoDuration::minutes(5) {
            buffer.cleanup(cutoff);
        }

        // Add new event to appropriate buffer
        match event {
            crate::types::ExtendedMarketDataEvent::NewsAlert(news) => {
                if news.impact_score.unwrap_or(0.0) >= self.config.min_news_importance {
                    buffer.news_events.push_back(news.clone());
                    self.update_category_encoding(&news.category).await;
                }
            },
            crate::types::ExtendedMarketDataEvent::SentimentUpdate(sentiment) => {
                buffer.sentiment_events.push_back(sentiment.clone());
            },
            crate::types::ExtendedMarketDataEvent::AnalystRating(rating) => {
                buffer.rating_events.push_back(rating.clone());
            },
            crate::types::ExtendedMarketDataEvent::UnusualOptions(options) => {
                buffer.options_events.push_back(options.clone());
            },
            crate::types::ExtendedMarketDataEvent::Core(_) => {}, // Ignore core market data events
        }

        Ok(())
    }

    /// Update category encoding for news categorization
    async fn update_category_encoding(&self, category: &str) {
        let mut encoding = self.category_encoding.write().await;
        if !encoding.contains_key(category) {
            let index = encoding.len();
            encoding.insert(category.to_string(), index);
        }
    }

    /// Extract features for a specific symbol at a given timestamp
    #[instrument(skip(self))]
    pub async fn extract_features(
        &self,
        symbol: &Symbol,
        timestamp: DateTime<Utc>,
    ) -> Result<BenzingaFeatureVector> {
        let buffers = self.buffers.read().await;
        let buffer = buffers.get(symbol).ok_or_else(|| {
            DataError::internal(format!("No data buffer found for symbol {}", symbol))
        })?;

        let window_start =
            timestamp - ChronoDuration::minutes(self.config.feature_window_minutes as i64);
        let sequence_id = self.sequence_counter.fetch_add(1, Ordering::Relaxed);

        // Extract features from each data type
        let news_features = self
            .extract_news_features(&buffer.news_events, window_start, timestamp)
            .await;
        let sentiment_features = self
            .extract_sentiment_features(&buffer.sentiment_events, window_start, timestamp)
            .await;
        let rating_features = self
            .extract_rating_features(&buffer.rating_events, window_start, timestamp)
            .await;
        let options_features = self
            .extract_options_features(&buffer.options_events, window_start, timestamp)
            .await;
        let time_features = self.extract_time_features(timestamp);

        let mut feature_vector = BenzingaFeatureVector {
            symbol: symbol.clone(),
            timestamp,
            sequence_id,

            // News features
            news_volume: news_features.0,
            news_importance_avg: news_features.1,
            news_importance_max: news_features.2,
            news_sentiment: news_features.3,
            news_category_encoding: news_features.4,
            breaking_news_indicator: news_features.5,

            // Sentiment features
            sentiment_score: sentiment_features.0,
            sentiment_momentum: sentiment_features.1,
            sentiment_volatility: sentiment_features.2,
            bullish_ratio: sentiment_features.3,
            bearish_ratio: sentiment_features.4,
            sentiment_confidence: sentiment_features.5,
            sentiment_sample_size_log: sentiment_features.6,

            // Rating features
            rating_change: rating_features.0,
            price_target_change_pct: rating_features.1,
            analyst_consensus: rating_features.2,
            rating_volume: rating_features.3,

            // Options features
            unusual_options_activity: options_features.0,
            options_flow_sentiment: options_features.1,
            options_volume_normalized: options_features.2,
            iv_signal: options_features.3,

            // Time features
            hour_sin: time_features.0,
            hour_cos: time_features.1,
            day_sin: time_features.2,
            day_cos: time_features.3,
            market_session: time_features.4,

            // Initialize advanced features
            keyword_sentiments: vec![0.0; self.financial_keywords.len()],
            topic_scores: vec![0.0; 5],      // 5 common financial topics
            entity_sentiments: vec![0.0; 3], // Company, person, location

            // Technical indicators (will be computed)
            sentiment_rsi: 50.0,
            sentiment_ma_short: sentiment_features.0,
            sentiment_ma_long: sentiment_features.0,
            sentiment_bb_position: 0.5,

            // Meta features
            data_quality_score: 1.0,
            feature_completeness: 1.0,
            market_regime: 0.0, // Neutral
        };

        // Add advanced features if enabled
        if self.config.enable_nlp_features {
            feature_vector.keyword_sentiments = self
                .extract_keyword_sentiments(&buffer.news_events, window_start, timestamp)
                .await;
        }

        if self.config.enable_sentiment_indicators {
            let indicators = self
                .compute_sentiment_indicators(&buffer.sentiment_events, timestamp)
                .await;
            feature_vector.sentiment_rsi = indicators.0;
            feature_vector.sentiment_ma_short = indicators.1;
            feature_vector.sentiment_ma_long = indicators.2;
            feature_vector.sentiment_bb_position = indicators.3;
        }

        // Normalize features if configured
        if !matches!(self.config.normalization_method, NormalizationMethod::None) {
            self.normalize_features(&mut feature_vector).await?;
        }

        Ok(feature_vector)
    }

    /// Extract news-related features
    async fn extract_news_features(
        &self,
        events: &VecDeque<NewsEvent>,
        start: DateTime<Utc>,
        end: DateTime<Utc>,
    ) -> (f64, f64, f64, f64, Vec<f64>, f64) {
        let relevant_events: Vec<_> = events
            .iter()
            .filter(|e| e.timestamp >= start && e.timestamp <= end)
            .collect();

        if relevant_events.is_empty() {
            let encoding = self.category_encoding.read().await;
            let category_features = vec![0.0; encoding.len().max(1)];
            return (0.0, 0.0, 0.0, 0.0, category_features, 0.0);
        }

        let volume = relevant_events.len() as f64;

        let importance_scores: Vec<f64> = relevant_events
            .iter()
            .filter_map(|e| e.impact_score)
            .collect();

        let importance_avg = if !importance_scores.is_empty() {
            importance_scores.iter().sum::<f64>() / importance_scores.len() as f64
        } else {
            0.0
        };

        let importance_max = importance_scores.iter().cloned().fold(0.0, f64::max);

        // Simple sentiment from impact scores (could be enhanced with NLP)
        let sentiment = importance_scores
            .iter()
            .map(|&score| if score > 0.5 { score } else { -score })
            .sum::<f64>()
            / importance_scores.len().max(1) as f64;

        // Category encoding
        let encoding = self.category_encoding.read().await;
        let mut category_features = vec![0.0; encoding.len().max(1)];
        for event in &relevant_events {
            if let Some(&index) = encoding.get(&event.category) {
                if index < category_features.len() {
                    category_features[index] += 1.0;
                }
            }
        }

        // Normalize category features
        let total_events = volume;
        if total_events > 0.0 {
            for feature in &mut category_features {
                *feature /= total_events;
            }
        }

        // Breaking news indicator (high importance + recent)
        let breaking_news = relevant_events.iter().any(|e| {
            e.impact_score.unwrap_or(0.0) > 0.8
                && e.tags
                    .iter()
                    .any(|tag| tag.contains("breaking") || tag.contains("urgent"))
        }) as u32 as f64;

        (
            volume,
            importance_avg,
            importance_max,
            sentiment,
            category_features,
            breaking_news,
        )
    }

    /// Extract sentiment-related features
    async fn extract_sentiment_features(
        &self,
        events: &VecDeque<SentimentEvent>,
        start: DateTime<Utc>,
        end: DateTime<Utc>,
    ) -> (f64, f64, f64, f64, f64, f64, f64) {
        let relevant_events: Vec<_> = events
            .iter()
            .filter(|e| e.timestamp >= start && e.timestamp <= end)
            .collect();

        if relevant_events.is_empty() {
            return (0.0, 0.0, 0.0, 0.5, 0.5, 0.0, 0.0);
        }

        // Current sentiment (latest event)
        let current_sentiment = relevant_events
            .last()
            .map(|e| e.sentiment_score)
            .unwrap_or(0.0);

        // Sentiment momentum (rate of change)
        let momentum = if relevant_events.len() > 1 {
            // len() > 1 guarantees first() and last() are Some
            let first = relevant_events
                .first()
                .map(|e| e.sentiment_score)
                .unwrap_or(0.0);
            let last = relevant_events
                .last()
                .map(|e| e.sentiment_score)
                .unwrap_or(0.0);
            last - first
        } else {
            0.0
        };

        // Sentiment volatility (standard deviation)
        let scores: Vec<f64> = relevant_events.iter().map(|e| e.sentiment_score).collect();
        let mean_score = scores.iter().sum::<f64>() / scores.len() as f64;
        let variance = scores
            .iter()
            .map(|score| (score - mean_score).powi(2))
            .sum::<f64>()
            / scores.len() as f64;
        let volatility = variance.sqrt();

        // Average ratios
        let avg_bullish = relevant_events.iter().map(|e| e.bullish_ratio).sum::<f64>()
            / relevant_events.len() as f64;

        let avg_bearish = relevant_events.iter().map(|e| e.bearish_ratio).sum::<f64>()
            / relevant_events.len() as f64;

        // Average confidence
        let avg_confidence = relevant_events.iter().map(|e| e.confidence).sum::<f64>()
            / relevant_events.len() as f64;

        // Log-normalized sample size
        let total_sample_size = relevant_events
            .iter()
            .map(|e| e.sample_size as f64)
            .sum::<f64>();
        let log_sample_size = (total_sample_size + 1.0).ln();

        (
            current_sentiment,
            momentum,
            volatility,
            avg_bullish,
            avg_bearish,
            avg_confidence,
            log_sample_size,
        )
    }

    /// Extract rating-related features
    async fn extract_rating_features(
        &self,
        events: &VecDeque<AnalystRatingEvent>,
        start: DateTime<Utc>,
        end: DateTime<Utc>,
    ) -> (f64, f64, f64, f64) {
        let relevant_events: Vec<_> = events
            .iter()
            .filter(|e| e.timestamp >= start && e.timestamp <= end)
            .collect();

        if relevant_events.is_empty() {
            return (0.0, 0.0, 0.0, 0.0);
        }

        // Rating change indicator
        let rating_changes: Vec<f64> = relevant_events
            .iter()
            .map(|e| match e.action {
                RatingAction::Upgrade => 1.0,
                RatingAction::Downgrade => -1.0,
                RatingAction::Initiate => 0.5,
                RatingAction::Discontinue => -0.5,
                RatingAction::Maintain => 0.0,
                RatingAction::Suspend => -0.3, // Neutral-negative for suspended ratings
            })
            .collect();

        let avg_rating_change = rating_changes.iter().sum::<f64>() / rating_changes.len() as f64;

        // Price target changes
        let price_target_changes: Vec<f64> = relevant_events
            .iter()
            .filter_map(|e| {
                if let (Some(current), Some(previous)) = (e.price_target, e.previous_price_target) {
                    if previous.to_decimal().unwrap_or(Decimal::ZERO) > dec!(0) {
                        let current_dec = current.to_decimal().unwrap_or(Decimal::ZERO);
                        let previous_dec = previous.to_decimal().unwrap_or(Decimal::ZERO);
                        let change_pct = ((current_dec - previous_dec) / previous_dec * dec!(100))
                            .to_f64()
                            .unwrap_or(0.0);
                        Some(change_pct)
                    } else {
                        None
                    }
                } else {
                    None
                }
            })
            .collect();

        let avg_price_target_change = if !price_target_changes.is_empty() {
            price_target_changes.iter().sum::<f64>() / price_target_changes.len() as f64
        } else {
            0.0
        };

        // Analyst consensus (simplified)
        let consensus = if rating_changes.iter().sum::<f64>() > 0.0 {
            0.75 // Positive consensus
        } else if rating_changes.iter().sum::<f64>() < 0.0 {
            0.25 // Negative consensus
        } else {
            0.5 // Neutral
        };

        let rating_volume = relevant_events.len() as f64;

        (
            avg_rating_change,
            avg_price_target_change,
            consensus,
            rating_volume,
        )
    }

    /// Extract options-related features
    async fn extract_options_features(
        &self,
        events: &VecDeque<UnusualOptionsEvent>,
        start: DateTime<Utc>,
        end: DateTime<Utc>,
    ) -> (f64, f64, f64, f64) {
        let relevant_events: Vec<_> = events
            .iter()
            .filter(|e| e.timestamp >= start && e.timestamp <= end)
            .collect();

        if relevant_events.is_empty() {
            return (0.0, 0.0, 0.0, 0.0);
        }

        // Unusual activity indicator
        let activity_strength = relevant_events.len() as f64
            * relevant_events.iter().map(|e| e.confidence).sum::<f64>()
            / relevant_events.len() as f64;

        // Options flow sentiment
        let sentiment_score = relevant_events
            .iter()
            .map(|e| match e.sentiment {
                OptionsSentiment::Bullish => 1.0,
                OptionsSentiment::Bearish => -1.0,
                OptionsSentiment::Neutral => 0.0,
            })
            .sum::<f64>()
            / relevant_events.len() as f64;

        // Normalized options volume
        let total_volume = relevant_events
            .iter()
            .map(|e| e.volume.as_f64())
            .sum::<f64>();
        let normalized_volume = (total_volume + 1.0).ln(); // Log normalization

        // Implied volatility signal (averaged)
        let iv_signal = relevant_events
            .iter()
            .filter_map(|e| e.implied_volatility)
            .sum::<f64>()
            / relevant_events.len() as f64;

        (
            activity_strength,
            sentiment_score,
            normalized_volume,
            iv_signal,
        )
    }

    /// Extract time-based cyclical features
    fn extract_time_features(&self, timestamp: DateTime<Utc>) -> (f64, f64, f64, f64, f64) {
        let hour = timestamp.hour() as f64;
        let day_of_week = timestamp.weekday().num_days_from_monday() as f64;

        // Cyclical encoding
        let hour_sin = (2.0 * std::f64::consts::PI * hour / 24.0).sin();
        let hour_cos = (2.0 * std::f64::consts::PI * hour / 24.0).cos();
        let day_sin = (2.0 * std::f64::consts::PI * day_of_week / 7.0).sin();
        let day_cos = (2.0 * std::f64::consts::PI * day_of_week / 7.0).cos();

        // Market session (simplified for US markets)
        let hour_int = hour as u8;
        let market_session = match hour_int {
            4..=8 => 1.0,   // Pre-market
            9..=15 => 2.0,  // Regular session
            16..=20 => 3.0, // After-hours
            _ => 0.0,       // Closed
        };

        (hour_sin, hour_cos, day_sin, day_cos, market_session)
    }

    /// Extract keyword-based sentiment features
    async fn extract_keyword_sentiments(
        &self,
        events: &VecDeque<NewsEvent>,
        start: DateTime<Utc>,
        end: DateTime<Utc>,
    ) -> Vec<f64> {
        let relevant_events: Vec<_> = events
            .iter()
            .filter(|e| e.timestamp >= start && e.timestamp <= end)
            .collect();

        let mut keyword_sentiments = vec![0.0; self.financial_keywords.len()];

        for (i, keyword) in self.financial_keywords.iter().enumerate() {
            let mut keyword_sentiment = 0.0;
            let mut keyword_count = 0;

            for event in &relevant_events {
                let text = format!("{} {}", event.headline, event.summary.as_str());
                let text_lower = text.to_lowercase();

                if text_lower.contains(keyword.as_str()) {
                    keyword_sentiment += event.impact_score.unwrap_or(0.0);
                    keyword_count += 1;
                }
            }

            if keyword_count > 0 {
                keyword_sentiments[i] = keyword_sentiment / keyword_count as f64;
            }
        }

        keyword_sentiments
    }

    /// Compute technical indicators on sentiment time series
    async fn compute_sentiment_indicators(
        &self,
        events: &VecDeque<SentimentEvent>,
        _current_time: DateTime<Utc>,
    ) -> (f64, f64, f64, f64) {
        // Get sentiment scores for the last N periods
        let lookback_period = 14; // Standard RSI period
        let recent_events: Vec<_> = events
            .iter()
            .rev()
            .take(lookback_period * 2) // Take more for longer MA
            .collect();

        if recent_events.len() < 3 {
            return (50.0, 0.0, 0.0, 0.5); // Neutral values
        }

        let scores: Vec<f64> = recent_events
            .iter()
            .rev()
            .map(|e| e.sentiment_score)
            .collect();

        // RSI calculation
        let rsi = self.calculate_rsi(&scores, lookback_period);

        // Moving averages
        let ma_short = self.calculate_ma(&scores, 5);
        let ma_long = self.calculate_ma(&scores, 10);

        // Bollinger Band position
        let bb_position = self.calculate_bollinger_position(&scores, 10, 2.0);

        (rsi, ma_short, ma_long, bb_position)
    }

    /// Calculate RSI for sentiment
    fn calculate_rsi(&self, values: &[f64], period: usize) -> f64 {
        if values.len() < period + 1 {
            return 50.0; // Neutral
        }

        let mut gains = Vec::new();
        let mut losses = Vec::new();

        // SAFETY: Loop starts at i=1, so i-1 is always valid.
        // Loop bound is values.len(), so i < values.len().
        #[allow(clippy::indexing_slicing)]
        for i in 1..values.len() {
            let change = values[i] - values[i - 1];
            if change > 0.0 {
                gains.push(change);
                losses.push(0.0);
            } else {
                gains.push(0.0);
                losses.push(-change);
            }
        }

        if gains.len() < period {
            return 50.0;
        }

        let avg_gain = gains.iter().take(period).sum::<f64>() / period as f64;
        let avg_loss = losses.iter().take(period).sum::<f64>() / period as f64;

        if avg_loss == 0.0 {
            return 100.0;
        }

        let rs = avg_gain / avg_loss;
        100.0 - (100.0 / (1.0 + rs))
    }

    /// Calculate moving average
    fn calculate_ma(&self, values: &[f64], period: usize) -> f64 {
        if values.len() < period {
            return values.iter().sum::<f64>() / values.len() as f64;
        }

        values.iter().rev().take(period).sum::<f64>() / period as f64
    }

    /// Calculate Bollinger Band position
    fn calculate_bollinger_position(&self, values: &[f64], period: usize, std_dev: f64) -> f64 {
        if values.len() < period {
            return 0.5; // Middle of band
        }

        let recent_values: Vec<f64> = values.iter().rev().take(period).cloned().collect();
        let mean = recent_values.iter().sum::<f64>() / recent_values.len() as f64;
        let variance = recent_values
            .iter()
            .map(|v| (v - mean).powi(2))
            .sum::<f64>()
            / recent_values.len() as f64;
        let std = variance.sqrt();

        let current_value = values.last().unwrap_or(&mean);
        let upper_band = mean + (std_dev * std);
        let lower_band = mean - (std_dev * std);

        if upper_band == lower_band {
            0.5
        } else {
            (current_value - lower_band) / (upper_band - lower_band)
        }
    }

    /// Normalize feature vector based on configuration
    async fn normalize_features(&self, features: &mut BenzingaFeatureVector) -> Result<()> {
        match self.config.normalization_method {
            NormalizationMethod::None => Ok(()),
            NormalizationMethod::ZScore => {
                // Z-score normalization would require historical statistics
                // For now, implement a simple version
                features.sentiment_score =
                    self.z_score_normalize(features.sentiment_score, 0.0, 0.5);
                features.news_importance_avg =
                    self.z_score_normalize(features.news_importance_avg, 0.5, 0.3);
                Ok(())
            },
            NormalizationMethod::MinMax => {
                features.sentiment_score =
                    self.min_max_normalize(features.sentiment_score, -1.0, 1.0);
                features.news_importance_avg =
                    self.min_max_normalize(features.news_importance_avg, 0.0, 1.0);
                Ok(())
            },
            NormalizationMethod::Robust => {
                // Robust scaling using median and IQR
                // Simplified implementation
                Ok(())
            },
        }
    }

    fn z_score_normalize(&self, value: f64, mean: f64, std: f64) -> f64 {
        if std == 0.0 {
            0.0
        } else {
            (value - mean) / std
        }
    }

    fn min_max_normalize(&self, value: f64, min: f64, max: f64) -> f64 {
        if max == min {
            0.0
        } else {
            (value - min) / (max - min)
        }
    }

    /// Extract features for multiple symbols (batch processing)
    #[instrument(skip(self))]
    pub async fn extract_features_batch(
        &self,
        symbols: &[Symbol],
        timestamp: DateTime<Utc>,
    ) -> Result<Vec<BenzingaFeatureVector>> {
        let mut features = Vec::new();

        for symbol in symbols {
            match self.extract_features(symbol, timestamp).await {
                Ok(feature_vector) => features.push(feature_vector),
                Err(e) => {
                    debug!("Failed to extract features for {}: {}", symbol, e);
                },
            }
        }

        info!("Extracted features for {} symbols", features.len());
        Ok(features)
    }

    /// Get feature vector size for ML model configuration
    pub fn get_feature_dimension(&self) -> usize {
        // Count all numerical features in BenzingaFeatureVector
        let base_features = 25; // Basic numerical features
        let category_encoding_size = 10; // Typical number of categories
        let keyword_features = self.financial_keywords.len();
        let topic_features = 5; // Number of topics
        let entity_features = 3; // Number of entity types

        base_features + category_encoding_size + keyword_features + topic_features + entity_features
    }

    /// Get feature names for interpretability
    pub fn get_feature_names(&self) -> Vec<String> {
        let mut names: Vec<String> = vec![
            "news_volume".to_string(),
            "news_importance_avg".to_string(),
            "news_importance_max".to_string(),
            "news_sentiment".to_string(),
            "breaking_news_indicator".to_string(),
            "sentiment_score".to_string(),
            "sentiment_momentum".to_string(),
            "sentiment_volatility".to_string(),
            "bullish_ratio".to_string(),
            "bearish_ratio".to_string(),
            "sentiment_confidence".to_string(),
            "sentiment_sample_size_log".to_string(),
            "rating_change".to_string(),
            "price_target_change_pct".to_string(),
            "analyst_consensus".to_string(),
            "rating_volume".to_string(),
            "unusual_options_activity".to_string(),
            "options_flow_sentiment".to_string(),
            "options_volume_normalized".to_string(),
            "iv_signal".to_string(),
            "hour_sin".to_string(),
            "hour_cos".to_string(),
            "day_sin".to_string(),
            "day_cos".to_string(),
            "market_session".to_string(),
            "sentiment_rsi".to_string(),
            "sentiment_ma_short".to_string(),
            "sentiment_ma_long".to_string(),
            "sentiment_bb_position".to_string(),
            "data_quality_score".to_string(),
            "feature_completeness".to_string(),
            "market_regime".to_string(),
        ];

        // Add category encoding features
        for i in 0..10 {
            names.push(format!("category_{}", i));
        }

        // Add keyword sentiment features
        for keyword in &self.financial_keywords {
            names.push(format!("keyword_{}_sentiment", keyword));
        }

        // Add topic features
        for i in 0..5 {
            names.push(format!("topic_{}", i));
        }

        // Add entity features
        for entity_type in &["company", "person", "location"] {
            names.push(format!("entity_{}_sentiment", entity_type));
        }

        names
    }
}

#[cfg(test)]
#[allow(clippy::assertions_on_result_states)]
mod tests {
    use super::*;

    #[test]
    fn test_ml_config_default() {
        let config = BenzingaMLConfig::default();
        assert!(config.feature_window_minutes > 0);
        assert!(config.max_lookback_hours > 0);
        assert!(config.feature_batch_size > 0);
    }

    #[tokio::test]
    async fn test_feature_extractor_creation() {
        let config = BenzingaMLConfig::default();
        let extractor = BenzingaMLExtractor::new(config);

        assert!(extractor.get_feature_dimension() > 0);
        assert!(!extractor.get_feature_names().is_empty());
    }

    #[test]
    fn test_rsi_calculation() {
        let extractor = BenzingaMLExtractor::new(BenzingaMLConfig::default());
        let values = vec![
            44.0, 44.25, 44.5, 43.75, 44.5, 44.75, 44.5, 45.0, 45.25, 45.5, 45.75, 46.0, 46.25,
            46.5,
        ];

        let rsi = extractor.calculate_rsi(&values, 14);
        assert!(rsi >= 0.0 && rsi <= 100.0);
    }

    #[test]
    fn test_moving_average_calculation() {
        let extractor = BenzingaMLExtractor::new(BenzingaMLConfig::default());
        let values = vec![1.0, 2.0, 3.0, 4.0, 5.0];

        let ma = extractor.calculate_ma(&values, 3);
        assert!((ma - 4.0).abs() < 0.001); // Average of [5, 4, 3] is 4
    }

    #[tokio::test]
    async fn test_process_event() {
        let config = BenzingaMLConfig::default();
        let extractor = BenzingaMLExtractor::new(config);

        let news_event = NewsEvent {
            symbol: Some(Symbol::from("AAPL")),
            symbols: vec![Symbol::from("AAPL")],
            story_id: "test123".to_string(),
            headline: "Test News".to_string(),
            content: String::new(),
            summary: String::new(),
            category: "earnings".to_string(),
            tags: vec!["test".to_string()],
            impact_score: Some(0.8),
            importance: 0.5,
            author: String::new(),
            timestamp: Utc::now(),
            published_at: Utc::now(),
            source: "Test".to_string(),
            url: String::new(),
            sentiment_score: None,
            #[allow(deprecated)]
            sentiment: None,
            event_type: crate::providers::common::NewsEventType::News,
        };

        let market_event = crate::types::ExtendedMarketDataEvent::NewsAlert(news_event);
        let result = extractor.process_event(&market_event).await;

        assert!(result.is_ok());
    }
}
