//! DBN Data Loader for ML Training
//!
//! Loads real market data from Databento (DBN) files and converts it to FinancialFeatures
//! for ML model training. Replaces synthetic/mock data with production-quality market data.
//!
//! ## Features
//!
//! - Load OHLCV bars from DBN files (ES.FUT futures data)
//! - Calculate technical indicators from real price data
//! - Compute microstructure features from real volume/spread
//! - Calculate risk metrics from historical price movements
//! - Train/validation split with proper time-series ordering
//!
//! ## Usage
//!
//! ```rust,no_run
//! use ml_training_service::dbn_data_loader::load_real_training_data;
//!
//! # async fn example() -> anyhow::Result<()> {
//! // Load real market data from DBN file
//! let (training_data, validation_data) = load_real_training_data(
//!     "test_data/real/databento/ES.FUT_ohlcv-1m_2024-01-02.dbn",
//!     0.8  // 80% training, 20% validation
//! ).await?;
//!
//! println!("Loaded {} training samples", training_data.len());
//! # Ok(())
//! # }
//! ```

use anyhow::{Context, Result};
use chrono::{DateTime, TimeZone, Utc};
use dbn::decode::{DbnDecoder, DecodeRecordRef};
use dbn::{OhlcvMsg, VersionUpgradePolicy};
use std::collections::{HashMap, VecDeque};
use std::path::Path;
use tracing::{debug, info, warn};

use common::Price;
use ml::training_pipeline::{FinancialFeatures, MicrostructureFeatures, RiskFeatures};

/// Convert DBN fixed-point price to f64
/// DBN stores prices as i64 with 9 decimal places precision
fn dbn_price_to_f64(price: i64) -> f64 {
    price as f64 / 1_000_000_000.0
}

/// Technical indicator calculator for RSI, SMA, EMA
struct TechnicalIndicatorCalculator {
    price_history: VecDeque<f64>,
    window_size: usize,
}

impl TechnicalIndicatorCalculator {
    fn new(window_size: usize) -> Self {
        Self {
            price_history: VecDeque::with_capacity(window_size),
            window_size,
        }
    }

    fn update(&mut self, price: f64) {
        self.price_history.push_back(price);
        if self.price_history.len() > self.window_size {
            self.price_history.pop_front();
        }
    }

    /// Calculate Relative Strength Index (RSI)
    fn calculate_rsi(&self, period: usize) -> f64 {
        if self.price_history.len() < period + 1 {
            return 50.0; // Neutral RSI if insufficient data
        }

        let prices: Vec<f64> = self
            .price_history
            .iter()
            .rev()
            .take(period + 1)
            .rev()
            .copied()
            .collect();

        let mut gains = 0.0;
        let mut losses = 0.0;

        for i in 1..prices.len() {
            let change = prices[i] - prices[i - 1];
            if change > 0.0 {
                gains += change;
            } else {
                losses += -change;
            }
        }

        let avg_gain = gains / period as f64;
        let avg_loss = losses / period as f64;

        if avg_loss < 1e-10 {
            return 100.0; // All gains, max RSI
        }

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

    /// Calculate Simple Moving Average (SMA)
    fn calculate_sma(&self) -> f64 {
        if self.price_history.is_empty() {
            return 0.0;
        }

        self.price_history.iter().sum::<f64>() / self.price_history.len() as f64
    }

    /// Calculate Exponential Moving Average (EMA)
    fn calculate_ema(&self, alpha: f64) -> f64 {
        if self.price_history.is_empty() {
            return 0.0;
        }

        let mut ema = self.price_history.front().copied().unwrap_or(0.0);
        for &price in self.price_history.iter().skip(1) {
            ema = alpha * price + (1.0 - alpha) * ema;
        }
        ema
    }
}

/// Risk metrics calculator for VaR, ES, drawdown, Sharpe
struct RiskMetricsCalculator {
    price_history: VecDeque<f64>,
    window_size: usize,
}

impl RiskMetricsCalculator {
    fn new(window_size: usize) -> Self {
        Self {
            price_history: VecDeque::with_capacity(window_size),
            window_size,
        }
    }

    fn update(&mut self, price: f64) {
        if !price.is_finite() || price <= 0.0 {
            return;
        }

        self.price_history.push_back(price);
        if self.price_history.len() > self.window_size {
            self.price_history.pop_front();
        }
    }

    fn calculate_log_returns(&self) -> Vec<f64> {
        if self.price_history.len() < 2 {
            return Vec::new();
        }

        self.price_history
            .iter()
            .zip(self.price_history.iter().skip(1))
            .map(|(prev, curr)| (curr / prev).ln())
            .filter(|r| r.is_finite())
            .collect()
    }

    /// Calculate Value at Risk at 5% confidence
    fn calculate_var(&self) -> f64 {
        let mut returns = self.calculate_log_returns();
        if returns.is_empty() {
            return -0.02; // Default -2%
        }

        returns.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal));
        let index = (returns.len() as f64 * 0.05).floor() as usize;
        let index = index.min(returns.len().saturating_sub(1));

        returns[index]
    }

    /// Calculate Expected Shortfall (CVaR)
    fn calculate_expected_shortfall(&self) -> f64 {
        let var = self.calculate_var();
        let returns = self.calculate_log_returns();

        if returns.is_empty() {
            return -0.03; // Default -3%
        }

        let tail_returns: Vec<f64> = returns.iter().filter(|&&r| r <= var).copied().collect();

        if tail_returns.is_empty() {
            return var;
        }

        tail_returns.iter().sum::<f64>() / tail_returns.len() as f64
    }

    /// Calculate maximum drawdown
    fn calculate_max_drawdown(&self) -> f64 {
        if self.price_history.len() < 2 {
            return -0.05; // Default -5%
        }

        let mut max_price = self.price_history.front().copied().unwrap_or(0.0);
        let mut max_drawdown = 0.0;

        for &price in self.price_history.iter().skip(1) {
            if price > max_price {
                max_price = price;
            } else {
                let drawdown = (price - max_price) / max_price;
                if drawdown < max_drawdown {
                    max_drawdown = drawdown;
                }
            }
        }

        max_drawdown
    }

    /// Calculate Sharpe ratio (annualized, risk-free rate = 0)
    fn calculate_sharpe_ratio(&self) -> f64 {
        let returns = self.calculate_log_returns();
        if returns.len() < 2 {
            return 1.0; // Neutral Sharpe
        }

        let mean_return = returns.iter().sum::<f64>() / returns.len() as f64;
        let variance = returns
            .iter()
            .map(|r| (r - mean_return).powi(2))
            .sum::<f64>()
            / returns.len() as f64;
        let std_dev = variance.sqrt();

        if std_dev < 1e-10 {
            return 0.0;
        }

        // Annualize (252 trading days, 1440 minutes per day)
        let annualized_return = mean_return * 252.0 * 1440.0;
        let annualized_volatility = std_dev * (252.0 * 1440.0_f64).sqrt();

        annualized_return / annualized_volatility
    }
}

/// Load real training data from DBN file
///
/// # Arguments
///
/// * `dbn_file_path` - Path to DBN file (e.g., "test_data/real/databento/ES.FUT_ohlcv-1m_2024-01-02.dbn")
/// * `train_split` - Fraction of data for training (e.g., 0.8 = 80% training, 20% validation)
///
/// # Returns
///
/// Tuple of (training_data, validation_data) with FinancialFeatures and price targets
pub async fn load_real_training_data(
    dbn_file_path: &str,
    train_split: f64,
) -> Result<(
    Vec<(FinancialFeatures, Vec<f64>)>,
    Vec<(FinancialFeatures, Vec<f64>)>,
)> {
    info!("Loading real market data from DBN file: {}", dbn_file_path);

    // Validate inputs
    if !Path::new(dbn_file_path).exists() {
        return Err(anyhow::anyhow!("DBN file not found: {}", dbn_file_path));
    }

    if !(0.0..=1.0).contains(&train_split) {
        return Err(anyhow::anyhow!(
            "train_split must be between 0.0 and 1.0, got: {}",
            train_split
        ));
    }

    // Load OHLCV bars from DBN file
    let bars = load_dbn_ohlcv_bars(dbn_file_path).await?;

    if bars.is_empty() {
        return Err(anyhow::anyhow!("No data loaded from DBN file"));
    }

    info!("Loaded {} OHLCV bars from DBN file", bars.len());

    // Convert bars to FinancialFeatures with technical indicators
    let mut features_with_targets = Vec::new();
    let mut tech_calc = TechnicalIndicatorCalculator::new(50);
    let mut risk_calc = RiskMetricsCalculator::new(100);

    for i in 0..bars.len() {
        let bar = &bars[i];

        // Update calculators
        tech_calc.update(bar.close);
        risk_calc.update(bar.close);

        // Skip first few bars until we have enough history
        if i < 20 {
            continue;
        }

        // Calculate technical indicators
        let mut indicators = HashMap::new();
        indicators.insert("rsi_14".to_string(), tech_calc.calculate_rsi(14));
        indicators.insert("sma_20".to_string(), tech_calc.calculate_sma());
        indicators.insert("ema_12".to_string(), tech_calc.calculate_ema(0.15)); // alpha = 2/(period+1)

        // Calculate VWAP (simplified: use close as approximation)
        let vwap = Price::from_f64(bar.close)
            .or_else(|_| Price::new(bar.close))
            .unwrap_or_default();

        // Calculate spread (use high-low as proxy)
        let spread_bps = ((bar.high - bar.low) / bar.close * 10_000.0) as i32;

        // Calculate order imbalance (simplified: use volume change)
        let imbalance = if i > 0 {
            let vol_change = (bars[i].volume - bars[i - 1].volume) / bars[i - 1].volume;
            vol_change.clamp(-1.0, 1.0)
        } else {
            0.0
        };

        // Calculate trade intensity (volume per minute)
        let trade_intensity = bar.volume / 60.0;

        // Create microstructure features
        let microstructure = MicrostructureFeatures {
            spread_bps,
            imbalance,
            trade_intensity,
            vwap,
        };

        // Calculate risk metrics
        let risk_metrics = RiskFeatures {
            var_5pct: risk_calc.calculate_var(),
            expected_shortfall: risk_calc.calculate_expected_shortfall(),
            max_drawdown: risk_calc.calculate_max_drawdown(),
            sharpe_ratio: risk_calc.calculate_sharpe_ratio(),
        };

        // Create FinancialFeatures
        let features = FinancialFeatures {
            prices: vec![
                Price::from_f64(bar.open).or_else(|_| Price::new(bar.open)).unwrap_or_default(),
                Price::from_f64(bar.high).or_else(|_| Price::new(bar.high)).unwrap_or_default(),
                Price::from_f64(bar.low).or_else(|_| Price::new(bar.low)).unwrap_or_default(),
                Price::from_f64(bar.close).or_else(|_| Price::new(bar.close)).unwrap_or_default(),
            ],
            volumes: vec![bar.volume as i64],
            technical_indicators: indicators,
            microstructure,
            risk_metrics,
            timestamp: bar.timestamp,
        };

        // Target: next bar's close (for price prediction)
        let target = if i + 1 < bars.len() {
            vec![bars[i + 1].close]
        } else {
            vec![bar.close] // Last bar: use current close
        };

        features_with_targets.push((features, target));
    }

    info!(
        "Converted {} bars to FinancialFeatures",
        features_with_targets.len()
    );

    // Split into training and validation (time-series split, no shuffle)
    let split_idx = (features_with_targets.len() as f64 * train_split) as usize;
    let training_data = features_with_targets[..split_idx].to_vec();
    let validation_data = features_with_targets[split_idx..].to_vec();

    info!(
        "Split data: {} training samples, {} validation samples",
        training_data.len(),
        validation_data.len()
    );

    Ok((training_data, validation_data))
}

/// OHLCV bar structure (intermediate format)
struct OhlcvBar {
    timestamp: DateTime<Utc>,
    open: f64,
    high: f64,
    low: f64,
    close: f64,
    volume: f64,
}

/// Load OHLCV bars from DBN file
async fn load_dbn_ohlcv_bars(file_path: &str) -> Result<Vec<OhlcvBar>> {
    debug!("Loading DBN file: {}", file_path);

    let mut decoder = DbnDecoder::from_file(file_path).context(format!(
        "Failed to create DBN decoder for file: {}",
        file_path
    ))?;

    decoder
        .set_upgrade_policy(VersionUpgradePolicy::UpgradeToV3)
        .context("Failed to set upgrade policy")?;

    let mut bars = Vec::new();
    let mut prev_close: Option<f64> = None;
    let mut corrections_applied = 0;

    while let Some(record_ref) = decoder
        .decode_record_ref()
        .context("Failed to decode DBN record")?
    {
        if let Some(ohlcv) = record_ref.get::<OhlcvMsg>() {
            // Convert timestamp
            let ts_nanos = ohlcv.hd.ts_event as i64;
            let secs = ts_nanos / 1_000_000_000;
            let nanos = (ts_nanos % 1_000_000_000) as u32;
            let timestamp = Utc
                .timestamp_opt(secs, nanos)
                .single()
                .ok_or_else(|| anyhow::anyhow!("Invalid timestamp: {}", ts_nanos))?;

            // Convert prices with anomaly correction
            let mut open_f64 = dbn_price_to_f64(ohlcv.open);
            let mut high_f64 = dbn_price_to_f64(ohlcv.high);
            let mut low_f64 = dbn_price_to_f64(ohlcv.low);
            let mut close_f64 = dbn_price_to_f64(ohlcv.close);

            // Price anomaly detection (same logic as backtesting service)
            if let Some(prev) = prev_close {
                let pct_change = ((close_f64 - prev) / prev).abs();

                if pct_change > 0.5 && close_f64 < 1000.0 {
                    let corrected_close = close_f64 * 100.0;

                    if (3000.0..=6000.0).contains(&corrected_close) {
                        open_f64 *= 100.0;
                        high_f64 *= 100.0;
                        low_f64 *= 100.0;
                        close_f64 = corrected_close;
                        corrections_applied += 1;

                        if corrections_applied <= 5 {
                            debug!(
                                "Applied 100x price correction at bar {} ({}% change)",
                                bars.len() + 1,
                                pct_change * 100.0
                            );
                        }
                    } else {
                        warn!(
                            "Skipping corrupted bar at index {} (timestamp: {})",
                            bars.len() + 1,
                            timestamp
                        );
                        prev_close = Some(prev);
                        continue;
                    }
                }
            }

            prev_close = Some(close_f64);

            let bar = OhlcvBar {
                timestamp,
                open: open_f64,
                high: high_f64,
                low: low_f64,
                close: close_f64,
                volume: ohlcv.volume as f64,
            };

            bars.push(bar);
        }
    }

    if corrections_applied > 0 {
        info!(
            "Applied {} automatic price corrections for encoding inconsistencies",
            corrections_applied
        );
    }

    Ok(bars)
}

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

    #[tokio::test]
    async fn test_load_real_training_data() {
        let dbn_file = "test_data/real/databento/ES.FUT_ohlcv-1m_2024-01-02.dbn";

        // Skip test if file doesn't exist
        if !Path::new(dbn_file).exists() {
            eprintln!("Skipping test: DBN file not found at {}", dbn_file);
            return;
        }

        let result = load_real_training_data(dbn_file, 0.8).await;
        assert!(result.is_ok(), "Failed to load data: {:?}", result.err());

        let (training, validation) = result.unwrap();

        // Verify data loaded
        assert!(!training.is_empty(), "Training data should not be empty");
        assert!(
            !validation.is_empty(),
            "Validation data should not be empty"
        );

        // Verify split ratio is approximately correct
        let total = training.len() + validation.len();
        let train_ratio = training.len() as f64 / total as f64;
        assert!(
            (train_ratio - 0.8).abs() < 0.05,
            "Training split should be ~80%, got {:.1}%",
            train_ratio * 100.0
        );

        // Verify features are populated
        if let Some((features, target)) = training.first() {
            assert!(!features.prices.is_empty(), "Prices should not be empty");
            assert!(!features.volumes.is_empty(), "Volumes should not be empty");
            assert!(
                !features.technical_indicators.is_empty(),
                "Indicators should not be empty"
            );
            assert!(!target.is_empty(), "Target should not be empty");
        }

        println!(
            "✅ Loaded {} training samples, {} validation samples",
            training.len(),
            validation.len()
        );
    }

    #[tokio::test]
    async fn test_technical_indicators() {
        let mut calc = TechnicalIndicatorCalculator::new(50);

        // Feed some price data
        for i in 0..50 {
            calc.update(4000.0 + (i as f64 * 10.0));
        }

        let rsi = calc.calculate_rsi(14);
        assert!(
            (0.0..=100.0).contains(&rsi),
            "RSI should be between 0 and 100 (inclusive)"
        );

        let sma = calc.calculate_sma();
        assert!(sma > 0.0, "SMA should be positive");

        let ema = calc.calculate_ema(0.15);
        assert!(ema > 0.0, "EMA should be positive");

        println!("✅ RSI={:.2}, SMA={:.2}, EMA={:.2}", rsi, sma, ema);
    }
}
