//! Ensemble Coordinator for Production Trading
//!
//! This module implements the core ensemble coordinator that aggregates predictions
//! from multiple ML models (DQN, PPO, TFT) for production trading decisions.
//!
//! ## Architecture
//!
//! ```text
//! ┌────────────────────────────────────────────────┐
//! │       Ensemble Coordinator                      │
//! │                                                 │
//! │  ┌──────────────────────────────────────┐     │
//! │  │        Model Registry                 │     │
//! │  │  (Active Buffer - Hot-swappable)     │     │
//! │  │                                       │     │
//! │  │  ┌─────┐  ┌─────┐  ┌─────┐          │     │
//! │  │  │ DQN │  │ PPO │  │ TFT │          │     │
//! │  │  └─────┘  └─────┘  └─────┘          │     │
//! │  └──────────────────────────────────────┘     │
//! │                  │                              │
//! │                  ▼                              │
//! │  ┌──────────────────────────────────────┐     │
//! │  │      Signal Aggregator                │     │
//! │  │  - Weighted voting                    │     │
//! │  │  - Confidence calculation             │     │
//! │  │  - Disagreement detection             │     │
//! │  └──────────────────────────────────────┘     │
//! │                  │                              │
//! │                  ▼                              │
//! │  ┌──────────────────────────────────────┐     │
//! │  │      Ensemble Decision                │     │
//! │  │  - Action: Buy/Sell/Hold              │     │
//! │  │  - Confidence: 0.0-1.0                │     │
//! │  │  - Disagreement rate                  │     │
//! │  └──────────────────────────────────────┘     │
//! └────────────────────────────────────────────────┘
//! ```

use anyhow::{Context, Result};
use chrono::{DateTime, Utc};
use ml::ensemble::{EnsembleDecision, ModelVote, ModelWeight, TradingAction};
use ml::features::ProductionFeatureExtractorAdapter;
use ml::{Features, MLError, MLModel, MLResult, ModelPrediction};
use serde::{Deserialize, Serialize};
use sqlx::PgPool;
use std::collections::HashMap;
use std::sync::Arc;
use std::time::Instant;
use tokio::sync::RwLock;
use tracing::{debug, info, warn};
use uuid::Uuid;

/// Canonical feature names for the 42-dimension production feature vector.
///
/// Layout matches `ml::features::extraction::FeatureExtractor::extract_current_features_v2`:
/// - 0..4: OHLCV (5)
/// - 5..9: Technical indicators (5)
/// - 10..15: Price patterns (6)
/// - 16..21: Volume features (6)
/// - 22..26: Time features (5)
/// - 27..39: Statistical features (13)
/// - 40..41: Regime detection (2)
const FEATURE_NAMES_42: &[&str] = &[
    // OHLCV (5)
    "open_return", "high_return", "low_return", "close_return", "volume_norm",
    // Technical indicators (5)
    "rsi_14", "macd_histogram", "bb_upper", "bb_lower", "atr_14",
    // Price patterns (6)
    "simple_return", "intraday_return", "overnight_return",
    "close_sma20_ratio", "close_sma50_ratio", "linreg_slope_20",
    // Volume features (6)
    "vol_ratio_sma20", "vol_spike", "vwap_20", "vwap_deviation",
    "price_vol_product", "price_vol_corr_20",
    // Time features (5)
    "hour_of_day", "day_of_week", "is_market_open",
    "minutes_since_open", "minutes_to_close",
    // Statistical features (13)
    "zscore_10", "zscore_20", "percentile_10", "percentile_20",
    "autocorr_lag1", "autocorr_lag5", "autocorr_lag10",
    "skewness_5", "skewness_10", "skewness_20",
    "kurtosis_5", "kurtosis_10", "kurtosis_20",
    // Regime detection (2)
    "adx_strength", "cusum_direction",
];

/// Minimum number of market data updates required before reliable feature extraction.
/// Mirrors the warmup period in `ml::features::extraction::FeatureExtractor`.
/// Exposed as a public constant for callers that want to pre-check warmup status.
pub const FEATURE_WARMUP_BARS: usize = 51;

use crate::ensemble_metrics::{EnsemblePredictionMetrics, ModelPnLAttribution, ModelWeightUpdate};

/// Ensemble prediction record for database persistence
#[derive(Debug, Clone, sqlx::FromRow, Serialize, Deserialize)]
pub struct EnsemblePrediction {
    // Primary identifiers
    pub id: Uuid,
    pub prediction_timestamp: DateTime<Utc>,

    // Trading context
    pub symbol: String,
    pub account_id: Option<String>,
    pub strategy_id: Option<String>,

    // Ensemble decision
    pub ensemble_action: String, // "BUY", "SELL", "HOLD"
    pub ensemble_signal: f64,
    pub ensemble_confidence: f64,
    pub disagreement_rate: f64,

    // Per-model votes (DQN)
    pub dqn_signal: Option<f64>,
    pub dqn_confidence: Option<f64>,
    pub dqn_weight: Option<f64>,
    pub dqn_vote: Option<String>,

    // Per-model votes (PPO)
    pub ppo_signal: Option<f64>,
    pub ppo_confidence: Option<f64>,
    pub ppo_weight: Option<f64>,
    pub ppo_vote: Option<String>,

    // Per-model votes (MAMBA-2)
    pub mamba2_signal: Option<f64>,
    pub mamba2_confidence: Option<f64>,
    pub mamba2_weight: Option<f64>,
    pub mamba2_vote: Option<String>,

    // Per-model votes (TFT)
    pub tft_signal: Option<f64>,
    pub tft_confidence: Option<f64>,
    pub tft_weight: Option<f64>,
    pub tft_vote: Option<String>,

    // Execution tracking
    pub order_id: Option<Uuid>,
    pub executed_price: Option<i64>,
    pub position_size: Option<i64>,
    pub pnl: Option<i64>,
    pub commission: Option<i64>,
    pub slippage_bps: Option<i32>,

    // Feature snapshot
    pub feature_snapshot: Option<serde_json::Value>,

    // System context
    pub node_id: Option<String>,
    pub inference_latency_us: Option<i32>,
    pub aggregation_latency_us: Option<i32>,

    // Metadata
    pub metadata: Option<serde_json::Value>,
}

impl EnsemblePrediction {
    /// Create from EnsembleDecision
    pub fn from_decision(
        decision: &EnsembleDecision,
        symbol: String,
        account_id: Option<String>,
    ) -> Self {
        let ensemble_action = format!("{:?}", decision.action).to_uppercase();

        // Extract per-model votes
        let (dqn_signal, dqn_confidence, dqn_weight, dqn_vote) =
            extract_model_vote(&decision.model_votes, "DQN");
        let (ppo_signal, ppo_confidence, ppo_weight, ppo_vote) =
            extract_model_vote(&decision.model_votes, "PPO");
        let (mamba2_signal, mamba2_confidence, mamba2_weight, mamba2_vote) =
            extract_model_vote(&decision.model_votes, "MAMBA2");
        let (tft_signal, tft_confidence, tft_weight, tft_vote) =
            extract_model_vote(&decision.model_votes, "TFT");

        Self {
            id: Uuid::new_v4(),
            prediction_timestamp: Utc::now(),
            symbol,
            account_id,
            strategy_id: None,
            ensemble_action,
            ensemble_signal: decision.signal,
            ensemble_confidence: decision.confidence,
            disagreement_rate: decision.disagreement_rate,
            dqn_signal,
            dqn_confidence,
            dqn_weight,
            dqn_vote,
            ppo_signal,
            ppo_confidence,
            ppo_weight,
            ppo_vote,
            mamba2_signal,
            mamba2_confidence,
            mamba2_weight,
            mamba2_vote,
            tft_signal,
            tft_confidence,
            tft_weight,
            tft_vote,
            order_id: None,
            executed_price: None,
            position_size: None,
            pnl: None,
            commission: None,
            slippage_bps: None,
            feature_snapshot: None,
            node_id: Some(get_node_id()),
            inference_latency_us: None,
            aggregation_latency_us: None,
            metadata: None,
        }
    }
}

/// Helper: Extract model vote from HashMap
fn extract_model_vote(
    votes: &HashMap<String, ModelVote>,
    model_id: &str,
) -> (Option<f64>, Option<f64>, Option<f64>, Option<String>) {
    votes
        .get(model_id)
        .map(|vote| {
            let action = format!("{:?}", vote.action(0.3)).to_uppercase();
            (
                Some(vote.signal),
                Some(vote.confidence),
                Some(vote.weight),
                Some(action),
            )
        })
        .unwrap_or((None, None, None, None))
}

/// Get current node ID (hostname or environment variable)
fn get_node_id() -> String {
    std::env::var("NODE_ID").unwrap_or_else(|_| {
        hostname::get()
            .unwrap_or_else(|_| "unknown".into())
            .to_string_lossy()
            .to_string()
    })
}

/// Ensemble coordinator configuration
#[derive(Debug, Clone)]
pub struct EnsembleConfig {
    /// Symbols to generate predictions for
    pub symbols: Vec<String>,

    /// Prediction generation interval (seconds)
    pub prediction_interval_secs: u64,
}

impl Default for EnsembleConfig {
    fn default() -> Self {
        Self {
            symbols: vec![
                "ES.FUT".to_string(),
                "NQ.FUT".to_string(),
                "ZN.FUT".to_string(),
                "6E.FUT".to_string(),
            ],
            prediction_interval_secs: 60, // Every 60 seconds
        }
    }
}

/// Ensemble coordinator for aggregating model predictions
#[derive(Debug)]
pub struct EnsembleCoordinator {
    /// Active model registry (dual-buffer for hot-swapping)
    active_models: Arc<RwLock<ModelRegistry>>,

    /// Signal aggregator
    aggregator: Arc<SignalAggregator>,

    /// Model weights configuration
    model_weights: Arc<RwLock<HashMap<String, ModelWeight>>>,

    /// Database connection pool
    db_pool: Option<PgPool>,

    /// Configuration
    config: Arc<RwLock<EnsembleConfig>>,

    /// Per-symbol production feature extractors (51-dim vectors from ml crate).
    /// Each extractor maintains rolling state (windows, indicators) for its symbol.
    feature_extractors: Arc<RwLock<HashMap<String, ProductionFeatureExtractorAdapter>>>,
}

impl EnsembleCoordinator {
    /// Create new ensemble coordinator
    pub fn new() -> Self {
        Self {
            active_models: Arc::new(RwLock::new(ModelRegistry::new())),
            aggregator: Arc::new(SignalAggregator::new()),
            model_weights: Arc::new(RwLock::new(HashMap::new())),
            db_pool: None,
            config: Arc::new(RwLock::new(EnsembleConfig::default())),
            feature_extractors: Arc::new(RwLock::new(HashMap::new())),
        }
    }

    /// Add database pool to coordinator
    pub fn with_db_pool(mut self, db_pool: PgPool) -> Self {
        self.db_pool = Some(db_pool);
        self
    }

    /// Set configuration
    pub async fn set_config(&self, config: EnsembleConfig) {
        let mut cfg = self.config.write().await;
        *cfg = config;
    }

    /// Register a model in the ensemble (without loaded model instance)
    pub async fn register_model(&self, model_id: String, weight: f64) -> MLResult<()> {
        let model_weight = ModelWeight::new(model_id.clone(), weight);

        let mut weights = self.model_weights.write().await;
        weights.insert(model_id.clone(), model_weight);

        info!("Registered model {} with weight {}", model_id, weight);
        Ok(())
    }

    /// Register a loaded model in the ensemble with model instance
    pub async fn register_loaded_model(
        &self,
        model_id: String,
        model: Arc<dyn MLModel>,
        weight: f64,
    ) -> MLResult<()> {
        // Register weight
        let model_weight = ModelWeight::new(model_id.clone(), weight);
        let mut weights = self.model_weights.write().await;
        weights.insert(model_id.clone(), model_weight);

        // Store model in active registry
        let mut registry = self.active_models.write().await;
        registry.register_active(model_id.clone(), model);

        info!(
            "Registered loaded model {} with weight {} (model instance active)",
            model_id, weight
        );
        Ok(())
    }

    /// Make ensemble prediction from features using real model inference
    pub async fn predict(&self, features: &Features) -> MLResult<EnsembleDecision> {
        debug!(
            "Making ensemble prediction with {} features",
            features.values.len()
        );

        // Start timing aggregation latency
        let start_time = Instant::now();

        // Get real model predictions
        let predictions = self.generate_real_predictions(features).await?;

        // Aggregate predictions
        let decision = self
            .aggregator
            .aggregate(predictions, &*self.model_weights.read().await)
            .await?;

        // Calculate aggregation latency
        let aggregation_latency_us = start_time.elapsed().as_micros() as f64;

        info!(
            "Ensemble decision: {:?}, confidence: {:.3}, disagreement: {:.3}, latency: {:.1}μs",
            decision.action,
            decision.confidence,
            decision.disagreement_rate,
            aggregation_latency_us
        );

        // Record ensemble prediction metrics
        let metrics = EnsemblePredictionMetrics {
            symbol: features
                .symbol
                .clone()
                .unwrap_or_else(|| "UNKNOWN".to_string()),
            action: format!("{:?}", decision.action).to_lowercase(),
            confidence: decision.confidence,
            disagreement_rate: decision.disagreement_rate,
            aggregation_latency_us,
            aggregation_method: "weighted_average".to_string(),
        };
        metrics.record();

        Ok(decision)
    }

    /// Generate real predictions from loaded models
    async fn generate_real_predictions(
        &self,
        features: &Features,
    ) -> MLResult<Vec<ModelPrediction>> {
        let registry = self.active_models.read().await;
        let weights = self.model_weights.read().await;

        let mut predictions = Vec::new();

        // Get active models from registry
        let active_models = registry.get_active_models();

        for (model_id, model) in active_models.iter() {
            // Verify model is registered in weights
            if !weights.contains_key(model_id) {
                warn!(
                    "Model {} in registry but not in weights, skipping",
                    model_id
                );
                continue;
            }

            // Call real model inference
            match model.predict(features).await {
                Ok(prediction) => {
                    debug!(
                        "Model {} predicted: value={:.3}, confidence={:.3}",
                        model_id, prediction.value, prediction.confidence
                    );
                    predictions.push(prediction);
                },
                Err(e) => {
                    warn!("Model {} prediction failed: {}", model_id, e);
                    // Continue with other models (ensemble degradation handling)
                },
            }
        }

        if predictions.is_empty() {
            return Err(MLError::InferenceError(
                "No successful predictions from any model".to_string(),
            ));
        }

        Ok(predictions)
    }

    /// Update model weights based on performance
    pub async fn update_model_weights(&self) -> MLResult<()> {
        let mut weights = self.model_weights.write().await;

        for weight in weights.values_mut() {
            weight.update_dynamic_weight();

            // Record updated weight metric
            let weight_update = ModelWeightUpdate {
                model_id: weight.model_id.clone(),
                // ROADMAP: Per-symbol weight tracking
                // Currently all models share a single global weight. To support per-symbol
                // weights:
                //   1. Change model_weights from HashMap<ModelId, ModelWeight> to
                //      HashMap<(ModelId, Symbol), ModelWeight>.
                //   2. Call update_model_weights(symbol) from the prediction path so each
                //      symbol's performance history independently adjusts model weights.
                //   3. Record per-symbol Prometheus metrics via the ModelWeightUpdate label.
                //   4. Requires the PnL attribution (record_pnl_attribution) to track symbol.
                // Tracking issue: per-symbol ensemble weight specialization.
                symbol: "ALL".to_string(),
                weight: weight.effective_weight(),
            };
            weight_update.record();
        }

        debug!("Updated dynamic weights for {} models", weights.len());
        Ok(())
    }

    /// Record P&L attribution for a model
    pub fn record_model_pnl(&self, model_id: &str, symbol: &str, pnl_dollars: f64) {
        let pnl = ModelPnLAttribution {
            model_id: model_id.to_string(),
            symbol: symbol.to_string(),
            pnl_dollars,
        };
        pnl.record();
    }

    /// Get model count
    pub async fn model_count(&self) -> usize {
        self.model_weights.read().await.len()
    }

    /// Save prediction to database
    pub async fn save_prediction_to_db(&self, prediction: &EnsemblePrediction) -> Result<Uuid> {
        let db_pool = self
            .db_pool
            .as_ref()
            .context("Database pool not configured")?;

        let start = Instant::now();

        // Insert prediction
        let row = sqlx::query!(
            r#"
            INSERT INTO ensemble_predictions (
                id, prediction_timestamp, symbol, account_id, strategy_id,
                ensemble_action, ensemble_signal, ensemble_confidence, disagreement_rate,
                dqn_signal, dqn_confidence, dqn_weight, dqn_vote,
                ppo_signal, ppo_confidence, ppo_weight, ppo_vote,
                mamba2_signal, mamba2_confidence, mamba2_weight, mamba2_vote,
                tft_signal, tft_confidence, tft_weight, tft_vote,
                feature_snapshot, node_id, inference_latency_us, aggregation_latency_us, metadata
            ) VALUES (
                $1, $2, $3, $4, $5,
                $6, $7, $8, $9,
                $10, $11, $12, $13,
                $14, $15, $16, $17,
                $18, $19, $20, $21,
                $22, $23, $24, $25,
                $26, $27, $28, $29, $30
            )
            RETURNING id
            "#,
            prediction.id,
            prediction.prediction_timestamp,
            prediction.symbol,
            prediction.account_id,
            prediction.strategy_id,
            prediction.ensemble_action,
            prediction.ensemble_signal,
            prediction.ensemble_confidence,
            prediction.disagreement_rate,
            prediction.dqn_signal,
            prediction.dqn_confidence,
            prediction.dqn_weight,
            prediction.dqn_vote,
            prediction.ppo_signal,
            prediction.ppo_confidence,
            prediction.ppo_weight,
            prediction.ppo_vote,
            prediction.mamba2_signal,
            prediction.mamba2_confidence,
            prediction.mamba2_weight,
            prediction.mamba2_vote,
            prediction.tft_signal,
            prediction.tft_confidence,
            prediction.tft_weight,
            prediction.tft_vote,
            prediction.feature_snapshot,
            prediction.node_id,
            prediction.inference_latency_us,
            prediction.aggregation_latency_us,
            prediction.metadata,
        )
        .fetch_one(db_pool)
        .await
        .context("Failed to insert ensemble prediction")?;

        let latency_us = start.elapsed().as_micros() as i64;

        info!(
            "Saved prediction {} to database: {} {} (latency: {}μs)",
            row.id, prediction.ensemble_action, prediction.symbol, latency_us
        );

        Ok(row.id)
    }

    /// Start background prediction loop
    pub async fn populate_predictions_continuously(
        self: Arc<Self>,
        interval_secs: u64,
    ) -> Result<()> {
        let mut interval = tokio::time::interval(std::time::Duration::from_secs(interval_secs));

        info!(
            "Starting background prediction loop (interval: {}s)",
            interval_secs
        );

        loop {
            interval.tick().await;

            // Generate prediction for configured symbols
            let config = self.config.read().await;
            let symbols = config.symbols.clone();
            drop(config);

            for symbol in &symbols {
                match self.generate_and_save_prediction(symbol).await {
                    Ok(prediction) => {
                        debug!("Generated prediction {} for {}", prediction.id, symbol);
                    },
                    Err(e) => {
                        warn!("Failed to generate prediction for {}: {}", symbol, e);
                    },
                }
            }
        }
    }

    /// Generate prediction and save to database.
    ///
    /// Returns the saved prediction record which contains the database `id`,
    /// `ensemble_action` (BUY/SELL/HOLD), `ensemble_confidence`, and all
    /// per-model vote details.
    pub async fn generate_and_save_prediction(
        &self,
        symbol: &str,
    ) -> Result<EnsemblePrediction> {
        // 1. Fetch latest market features (uses ProductionFeatureExtractorAdapter with zero-fallback)
        let features = self.fetch_features_for_symbol(symbol).await?;

        // 2. Make ensemble prediction
        let decision = self
            .predict(&features)
            .await
            .map_err(|e| anyhow::anyhow!("Ensemble prediction failed: {}", e))?;

        // 3. Convert to database record
        let mut prediction = EnsemblePrediction::from_decision(
            &decision,
            symbol.to_string(),
            Some("paper_trading_001".to_string()),
        );

        // 4. Save to database (updates the id to the DB-assigned value)
        let prediction_id = self.save_prediction_to_db(&prediction).await?;
        prediction.id = prediction_id;

        Ok(prediction)
    }

    /// Feed a market data update (price tick) into the per-symbol feature extractor.
    ///
    /// Call this every time a new OHLCV bar (or tick approximation) arrives.
    /// After at least [`FEATURE_WARMUP_BARS`] updates, `fetch_features_for_symbol`
    /// will return real 51-dimensional feature vectors instead of zeros.
    ///
    /// # Arguments
    /// * `symbol`    - Trading symbol (e.g. "ES.FUT")
    /// * `price`     - Latest close/mid price
    /// * `volume`    - Bar or tick volume
    /// * `timestamp` - UTC timestamp of the observation
    pub async fn update_market_data(
        &self,
        symbol: &str,
        price: f64,
        volume: f64,
        timestamp: DateTime<Utc>,
    ) -> Result<()> {
        use common::ml_strategy::ProductionFeatureExtractor225;

        let mut extractors = self.feature_extractors.write().await;
        let extractor = extractors
            .entry(symbol.to_string())
            .or_insert_with(ProductionFeatureExtractorAdapter::new);
        extractor
            .update(price, volume, timestamp)
            .context("Failed to update feature extractor")?;

        debug!(
            "Updated feature extractor for {} (price={:.2}, volume={:.0})",
            symbol, price, volume
        );
        Ok(())
    }

    /// Return the number of market-data updates received for `symbol`.
    /// Useful for callers that want to check warmup progress.
    pub async fn feature_update_count(&self, symbol: &str) -> usize {
        let extractors = self.feature_extractors.read().await;
        // The extractor does not expose a bar count directly, but we track
        // presence: if it exists, it has been updated at least once.
        // For a finer count we would need to extend the adapter; returning
        // 0 or 1 is sufficient for the "is warmed up?" check via
        // `fetch_features_for_symbol` which tries extraction and catches errors.
        if extractors.contains_key(symbol) { 1 } else { 0 }
    }

    /// Extract real 42-dimensional features for `symbol` using the production
    /// feature extractor from the `ml` crate.
    ///
    /// If the extractor for this symbol has not received enough warmup bars
    /// (minimum ~50 updates), extraction will fail gracefully and a zero-filled
    /// feature vector is returned with a warning log so the ensemble can still
    /// produce a low-confidence prediction.
    pub async fn fetch_features_for_symbol(&self, symbol: &str) -> Result<Features> {
        use common::ml_strategy::ProductionFeatureExtractor225;

        let names: Vec<String> = FEATURE_NAMES_42.iter().map(|s| (*s).to_string()).collect();

        let mut extractors = self.feature_extractors.write().await;

        if let Some(extractor) = extractors.get_mut(symbol) {
            match extractor.extract_features() {
                Ok(values) => {
                    debug!(
                        "Extracted {} real features for {}",
                        values.len(),
                        symbol
                    );
                    return Ok(Features::new(values, names).with_symbol(symbol.to_string()));
                }
                Err(e) => {
                    warn!(
                        "Feature extraction failed for {} (likely warmup incomplete): {}. \
                         Falling back to zero features.",
                        symbol, e
                    );
                }
            }
        } else {
            warn!(
                "No feature extractor initialised for {} — call update_market_data() first. \
                 Returning zero features.",
                symbol
            );
        }

        // Fallback: zeros so the ensemble can still run (will produce low confidence).
        Ok(Features::new(vec![0.0; FEATURE_NAMES_42.len()], names)
            .with_symbol(symbol.to_string()))
    }
}

impl Default for EnsembleCoordinator {
    fn default() -> Self {
        Self::new()
    }
}

/// Model registry with dual-buffer support for hot-swapping
#[derive(Debug)]
pub struct ModelRegistry {
    /// Active models (currently serving predictions)
    active: HashMap<String, Arc<dyn ml::MLModel>>, // model_id -> model instance

    /// Shadow models (staged for hot-swap)
    shadow: HashMap<String, Arc<dyn ml::MLModel>>,
}

impl ModelRegistry {
    /// Create new model registry
    pub fn new() -> Self {
        Self {
            active: HashMap::new(),
            shadow: HashMap::new(),
        }
    }

    /// Stage a model in shadow buffer
    pub fn stage_model(&mut self, model_id: String, model: Arc<dyn MLModel>) {
        self.shadow.insert(model_id.clone(), model);
        info!("Staged model for hot-swap: {}", model_id);
    }

    /// Commit swap (shadow becomes active)
    pub fn commit_swap(&mut self, model_id: &str) -> MLResult<()> {
        if let Some(shadow_model) = self.shadow.remove(model_id) {
            let old_model = self.active.insert(model_id.to_string(), shadow_model);

            if let Some(old) = old_model {
                // Move old to shadow for potential rollback
                self.shadow.insert(model_id.to_string(), old);
            }

            info!("Committed model swap for {}", model_id);
            Ok(())
        } else {
            Err(MLError::ModelNotFound(format!(
                "No staged model for {}",
                model_id
            )))
        }
    }

    /// Rollback to previous model
    pub fn rollback(&mut self, model_id: &str) -> MLResult<()> {
        if let Some(previous_model) = self.shadow.remove(model_id) {
            self.active.insert(model_id.to_string(), previous_model);
            warn!("Rolled back model {} to previous version", model_id);
            Ok(())
        } else {
            Err(MLError::ModelNotFound(format!(
                "No previous model version for {}",
                model_id
            )))
        }
    }

    /// Register a loaded model directly in active registry
    pub fn register_active(&mut self, model_id: String, model: Arc<dyn MLModel>) {
        self.active.insert(model_id.clone(), model);
        info!("Registered model in active registry: {}", model_id);
    }

    /// Get active models for prediction
    pub fn get_active_models(&self) -> &HashMap<String, Arc<dyn MLModel>> {
        &self.active
    }
}

impl Default for ModelRegistry {
    fn default() -> Self {
        Self::new()
    }
}

/// Signal aggregator for ensemble predictions
#[derive(Debug)]
pub struct SignalAggregator {
    /// Signal threshold for Buy/Sell actions
    signal_threshold: f64,
}

impl SignalAggregator {
    /// Create new signal aggregator
    pub fn new() -> Self {
        Self {
            signal_threshold: 0.3,
        }
    }

    /// Aggregate model predictions into ensemble decision
    pub async fn aggregate(
        &self,
        predictions: Vec<ModelPrediction>,
        weights: &HashMap<String, ModelWeight>,
    ) -> MLResult<EnsembleDecision> {
        if predictions.is_empty() {
            return Err(MLError::ValidationError {
                message: "No predictions to aggregate".to_string(),
            });
        }

        // Calculate weighted average signal
        let (weighted_signal, _total_weight) =
            self.calculate_weighted_signal(&predictions, weights);

        // Calculate ensemble confidence
        let confidence = self.calculate_ensemble_confidence(&predictions, weights);

        // Calculate disagreement rate
        let disagreement_rate = self.calculate_disagreement_rate(&predictions);

        // Determine trading action
        let action = TradingAction::from_signal(weighted_signal, self.signal_threshold);

        // Build model votes
        let model_votes = self.build_model_votes(&predictions, weights);

        let decision = EnsembleDecision::new(
            action,
            confidence,
            weighted_signal,
            disagreement_rate,
            model_votes,
        );

        Ok(decision)
    }

    /// Calculate weighted average signal
    fn calculate_weighted_signal(
        &self,
        predictions: &[ModelPrediction],
        weights: &HashMap<String, ModelWeight>,
    ) -> (f64, f64) {
        let mut weighted_sum = 0.0;
        let mut total_weight = 0.0;

        for pred in predictions {
            let weight = weights
                .get(&pred.model_id)
                .map(|w| w.effective_weight())
                .unwrap_or(1.0 / predictions.len() as f64);

            weighted_sum += pred.value * pred.confidence * weight;
            total_weight += weight * pred.confidence;
        }

        let signal = if total_weight > 0.0 {
            weighted_sum / total_weight
        } else {
            0.0
        };

        (signal, total_weight)
    }

    /// Calculate ensemble confidence
    fn calculate_ensemble_confidence(
        &self,
        predictions: &[ModelPrediction],
        weights: &HashMap<String, ModelWeight>,
    ) -> f64 {
        let mut confidence_sum = 0.0;
        let mut weight_sum = 0.0;

        for pred in predictions {
            let weight = weights
                .get(&pred.model_id)
                .map(|w| w.effective_weight())
                .unwrap_or(1.0 / predictions.len() as f64);

            confidence_sum += pred.confidence * weight;
            weight_sum += weight;
        }

        if weight_sum > 0.0 {
            confidence_sum / weight_sum
        } else {
            0.0
        }
    }

    /// Calculate disagreement rate (% models disagree with ensemble)
    fn calculate_disagreement_rate(&self, predictions: &[ModelPrediction]) -> f64 {
        if predictions.len() < 2 {
            return 0.0;
        }

        // Calculate mean signal
        let mean_signal: f64 =
            predictions.iter().map(|p| p.value).sum::<f64>() / predictions.len() as f64;

        // Count models with opposite sign from mean
        let disagreements = predictions
            .iter()
            .filter(|p| (p.value * mean_signal) < 0.0)
            .count();

        disagreements as f64 / predictions.len() as f64
    }

    /// Build model votes map
    fn build_model_votes(
        &self,
        predictions: &[ModelPrediction],
        weights: &HashMap<String, ModelWeight>,
    ) -> HashMap<String, ModelVote> {
        let mut votes = HashMap::new();

        for pred in predictions {
            let weight = weights
                .get(&pred.model_id)
                .map(|w| w.effective_weight())
                .unwrap_or(1.0 / predictions.len() as f64);

            let vote = ModelVote::new(pred.model_id.clone(), pred.value, pred.confidence, weight);

            votes.insert(pred.model_id.clone(), vote);
        }

        votes
    }
}

impl Default for SignalAggregator {
    fn default() -> Self {
        Self::new()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use ml::ensemble::TradingAction;

    #[tokio::test]
    async fn test_ensemble_coordinator_creation() {
        let coordinator = EnsembleCoordinator::new();
        assert_eq!(coordinator.model_count().await, 0);
    }

    #[tokio::test]
    async fn test_register_models() {
        let coordinator = EnsembleCoordinator::new();

        coordinator
            .register_model("DQN".to_string(), 0.33)
            .await
            .unwrap();
        coordinator
            .register_model("PPO".to_string(), 0.33)
            .await
            .unwrap();
        coordinator
            .register_model("TFT".to_string(), 0.34)
            .await
            .unwrap();

        assert_eq!(coordinator.model_count().await, 3);
    }

    #[tokio::test]
    async fn test_ensemble_prediction() {
        use ml::model_factory;

        let coordinator = EnsembleCoordinator::new();

        // Create and register LOADED models with model instances
        let dqn_model = model_factory::create_dqn_wrapper_with_id("DQN".to_string()).unwrap();
        let ppo_model = model_factory::create_ppo_wrapper_with_id("PPO".to_string()).unwrap();
        let tft_model = model_factory::create_tft_wrapper_with_id("TFT".to_string()).unwrap();

        coordinator
            .register_loaded_model("DQN".to_string(), dqn_model, 0.33)
            .await
            .unwrap();
        coordinator
            .register_loaded_model("PPO".to_string(), ppo_model, 0.33)
            .await
            .unwrap();
        coordinator
            .register_loaded_model("TFT".to_string(), tft_model, 0.34)
            .await
            .unwrap();

        // Create features
        let features = Features::new(
            vec![0.5, 0.6, 0.7, 0.8, 0.9],
            vec![
                "f1".to_string(),
                "f2".to_string(),
                "f3".to_string(),
                "f4".to_string(),
                "f5".to_string(),
            ],
        );

        // Make prediction
        let decision = coordinator.predict(&features).await.unwrap();

        assert!(decision.confidence >= 0.0 && decision.confidence <= 1.0);
        assert!(decision.signal >= -1.0 && decision.signal <= 1.0);
        assert_eq!(decision.model_count(), 3);
    }

    #[tokio::test]
    async fn test_disagreement_detection() {
        let aggregator = SignalAggregator::new();

        // High disagreement scenario
        let predictions = vec![
            ModelPrediction::new("DQN".to_string(), 0.8, 0.9),
            ModelPrediction::new("PPO".to_string(), -0.7, 0.85),
            ModelPrediction::new("TFT".to_string(), 0.1, 0.8),
        ];

        let weights = HashMap::new();
        let decision = aggregator.aggregate(predictions, &weights).await.unwrap();

        // Should detect disagreement (models have opposite signs)
        assert!(decision.disagreement_rate > 0.3);
    }

    #[tokio::test]
    async fn test_weighted_voting() {
        let aggregator = SignalAggregator::new();

        let predictions = vec![
            ModelPrediction::new("DQN".to_string(), 0.8, 0.9),
            ModelPrediction::new("PPO".to_string(), 0.7, 0.85),
            ModelPrediction::new("TFT".to_string(), 0.6, 0.8),
        ];

        let mut weights = HashMap::new();
        weights.insert("DQN".to_string(), ModelWeight::new("DQN".to_string(), 0.5));
        weights.insert("PPO".to_string(), ModelWeight::new("PPO".to_string(), 0.3));
        weights.insert("TFT".to_string(), ModelWeight::new("TFT".to_string(), 0.2));

        let decision = aggregator.aggregate(predictions, &weights).await.unwrap();

        // DQN has highest weight and signal, so ensemble should favor Buy
        assert_eq!(decision.action, TradingAction::Buy);
        assert!(decision.signal > 0.6); // Should be close to DQN's signal
    }

    #[tokio::test]
    async fn test_model_registry_operations() {
        use ml::model_factory;

        let mut registry = ModelRegistry::new();

        // Create a test model
        let test_model = model_factory::create_dqn_wrapper_with_id("DQN".to_string()).unwrap();

        // Stage model
        registry.stage_model("DQN".to_string(), test_model);

        // Commit swap
        registry.commit_swap("DQN").unwrap();

        assert!(registry.active.contains_key("DQN"));
        assert_eq!(registry.active.len(), 1);
    }

    #[tokio::test]
    async fn test_fetch_features_returns_zeros_without_warmup() {
        let coordinator = EnsembleCoordinator::new();

        // No market data fed yet — should return zero-filled 42-dim vector
        let features = coordinator
            .fetch_features_for_symbol("ES.FUT")
            .await
            .expect("fetch_features_for_symbol should not fail");

        assert_eq!(features.values.len(), 42, "Should return 42 features");
        assert_eq!(features.names.len(), 42, "Should have 42 feature names");
        assert_eq!(features.symbol, Some("ES.FUT".to_string()));

        // All values should be zero (no data fed)
        for &val in &features.values {
            assert!(
                val.abs() < f64::EPSILON,
                "Expected 0.0 without warmup, got {}",
                val,
            );
        }
    }

    #[tokio::test]
    async fn test_update_market_data_and_extract_real_features() {
        let coordinator = EnsembleCoordinator::new();
        let base_time = Utc::now();

        // Feed 60 bars to exceed the warmup period (50 bars)
        for i in 0..60 {
            let price = 4500.0 + (i as f64) * 0.25;
            let volume = 10_000.0 + (i as f64) * 100.0;
            let ts = base_time + chrono::Duration::seconds(i * 60);
            coordinator
                .update_market_data("ES.FUT", price, volume, ts)
                .await
                .expect("update_market_data should succeed");
        }

        // Now extraction should produce real features
        let features = coordinator
            .fetch_features_for_symbol("ES.FUT")
            .await
            .expect("fetch_features_for_symbol should succeed after warmup");

        assert_eq!(features.values.len(), 42, "Should return 42-dim vector");
        assert_eq!(features.symbol, Some("ES.FUT".to_string()));

        // After warmup, at least some features should be non-zero
        let non_zero_count = features.values.iter().filter(|v| v.abs() > 1e-12).count();
        assert!(
            non_zero_count > 5,
            "Expected many non-zero features after warmup, got {} non-zero out of 42",
            non_zero_count,
        );

        // All features must be finite (no NaN / Inf)
        for (i, &val) in features.values.iter().enumerate() {
            assert!(
                val.is_finite(),
                "Feature {} ({}) is not finite: {}",
                i,
                features.names.get(i).map(|s| s.as_str()).unwrap_or("?"),
                val,
            );
        }
    }

    #[tokio::test]
    async fn test_feature_names_match_42_dim_layout() {
        // Verify the constant array has exactly 42 entries
        assert_eq!(FEATURE_NAMES_42.len(), 42, "FEATURE_NAMES_42 must have 42 entries");

        // Spot-check a few known names
        assert_eq!(FEATURE_NAMES_42[0], "open_return");
        assert_eq!(FEATURE_NAMES_42[5], "rsi_14");
        assert_eq!(FEATURE_NAMES_42[40], "adx_strength");
        assert_eq!(FEATURE_NAMES_42[41], "cusum_direction");
    }

    #[tokio::test]
    async fn test_per_symbol_isolation() {
        let coordinator = EnsembleCoordinator::new();
        let base_time = Utc::now();

        // Feed data only for ES.FUT
        for i in 0..60 {
            let ts = base_time + chrono::Duration::seconds(i * 60);
            coordinator
                .update_market_data("ES.FUT", 4500.0 + i as f64, 10_000.0, ts)
                .await
                .expect("update_market_data should succeed");
        }

        // ES.FUT should have real features
        let es_features = coordinator
            .fetch_features_for_symbol("ES.FUT")
            .await
            .expect("Should succeed for ES.FUT");
        let es_nonzero = es_features.values.iter().filter(|v| v.abs() > 1e-12).count();
        assert!(es_nonzero > 5, "ES.FUT should have real features");

        // NQ.FUT was never fed data — should return zeros
        let nq_features = coordinator
            .fetch_features_for_symbol("NQ.FUT")
            .await
            .expect("Should succeed (zero fallback) for NQ.FUT");
        for &val in &nq_features.values {
            assert!(
                val.abs() < f64::EPSILON,
                "NQ.FUT should have zero features"
            );
        }
    }
}
