//! Data Loading for DQN Training
//!
//! Methods for loading market data from DBN files,
//! including feature extraction, preprocessing, and train/val splitting.

use std::path::Path;
use std::sync::Arc;

use anyhow::Result;
use tracing::{debug, info, warn};

use crate::features::extraction::OHLCVBar;
use crate::training_pipeline::FinancialFeatures;

use crate::features::extraction::FeatureVector;
use super::trainer::DQNTrainer;

/// OFI dimension — mirrors `ml_core::state_layout::OFI_DIM`.
const OFI_DIM: usize = ml_core::state_layout::OFI_DIM;

/// Resolve a data directory path — handles relative paths by checking against
/// the workspace root (via CARGO_MANIFEST_DIR) when the raw path doesn't exist.
/// Used for all data paths: dbn_data_dir, mbp10_data_dir, trades_data_dir.
pub(crate) fn resolve_data_dir(raw_path: &str) -> std::path::PathBuf {
    let raw = Path::new(raw_path);
    if raw.is_absolute() || raw.exists() {
        return raw.to_path_buf();
    }
    let manifest = std::env::var("CARGO_MANIFEST_DIR").unwrap_or_default();
    let workspace = Path::new(&manifest)
        .parent().and_then(|p| p.parent())
        .unwrap_or(Path::new(""));
    let resolved = workspace.join(raw);
    if resolved.exists() { resolved } else { raw.to_path_buf() }
}

/// Detect zstd-compressed files by magic bytes (0x28B52FFD).
/// File extension is unreliable — Databento downloads often use `.dbn` for zstd-compressed data.
fn is_zstd_file(path: &Path) -> Result<bool> {
    let mut file = std::fs::File::open(path)?;
    let mut magic = [0_u8; 4];
    use std::io::Read;
    if file.read_exact(&mut magic).is_ok() {
        Ok(magic == [0x28, 0xB5, 0x2F, 0xFD])
    } else {
        Ok(false)
    }
}

/// Recursively collect all .dbn and .dbn.zst files from a directory and its subdirectories.
pub fn collect_dbn_files_recursive(dir: &Path) -> Vec<std::path::PathBuf> {
    let mut files = Vec::new();
    if let Ok(entries) = std::fs::read_dir(dir) {
        for entry in entries.flatten() {
            let path = entry.path();
            if path.is_dir() {
                files.extend(collect_dbn_files_recursive(&path));
            } else if path.extension().and_then(|s| s.to_str()) == Some("dbn")
                || path.to_string_lossy().ends_with(".dbn.zst")
            {
                files.push(path);
            } else {
                // Skip non-DBN files
            }
        }
    }
    files
}

/// Collect DBN files, filtering to a specific symbol subdirectory.
/// If `symbol` is `Some("ES.FUT")`, only loads files from `dir/ES.FUT/`.
/// If the symbol subdirectory doesn't exist, falls back to filename matching.
pub fn collect_dbn_files_filtered(dir: &Path, symbol: Option<&str>) -> Vec<std::path::PathBuf> {
    match symbol {
        Some(sym) => {
            let symbol_dir = dir.join(sym);
            if symbol_dir.is_dir() {
                collect_dbn_files_recursive(&symbol_dir)
            } else {
                // No subdirectory — filter files by name containing symbol
                collect_dbn_files_recursive(dir)
                    .into_iter()
                    .filter(|p| {
                        p.file_name()
                            .and_then(|n| n.to_str())
                            .map(|n| n.contains(sym))
                            .unwrap_or(false)
                    })
                    .collect()
            }
        }
        None => collect_dbn_files_recursive(dir),
    }
}

/// Extract OHLCV bars from a single DBN file — no DQNTrainer required.
pub fn extract_ohlcv_bars_from_dbn(file_path: &Path) -> Result<Vec<OHLCVBar>> {
    if is_zstd_file(file_path)? {
        let mut decoder = dbn::decode::DbnDecoder::from_zstd_file(file_path)
            .map_err(|e| anyhow::anyhow!("Failed to create zstd DBN decoder for {}: {}", file_path.display(), e))?;
        DQNTrainer::decode_ohlcv_records(&mut decoder, file_path)
    } else {
        let mut decoder = dbn::decode::DbnDecoder::from_file(file_path)
            .map_err(|e| anyhow::anyhow!("Failed to create DBN decoder for {}: {}", file_path.display(), e))?;
        DQNTrainer::decode_ohlcv_records(&mut decoder, file_path)
    }
}

impl DQNTrainer {
    /// Load training data from DBN files using official dbn crate decoder
    /// Returns (train_data, val_data) with 80/20 split
    pub async fn load_training_data(
        &mut self,
        dbn_data_dir: &str,
        symbol: &str,
    ) -> Result<(
        Vec<(FeatureVector, Vec<f64>)>,
        Vec<(FeatureVector, Vec<f64>)>,
    )> {
        // ── fxcache discovery (strict key match via shared function) ────────
        let cache_dir_override = self.feature_cache_dir.as_deref();
        // Resolve mbp10/trades paths — try as-is first, then relative to workspace root
        // (CARGO_MANIFEST_DIR points to crates/ml, workspace root is 2 levels up).
        let resolve_data_path = |raw: &str| -> Option<std::path::PathBuf> {
            if raw.is_empty() { return None; }
            let p = std::path::Path::new(raw);
            if p.exists() { return Some(p.to_path_buf()); }
            // Try relative to workspace root
            if let Ok(manifest) = std::env::var("CARGO_MANIFEST_DIR") {
                if let Some(root) = std::path::Path::new(&manifest).parent().and_then(|p| p.parent()) {
                    let abs = root.join(raw);
                    if abs.exists() { return Some(abs); }
                }
            }
            // Try relative to dbn_data_dir's parent (e.g. test_data/futures-baseline → test_data/)
            let parent = std::path::Path::new(dbn_data_dir).parent().unwrap_or(std::path::Path::new("."));
            let sibling = parent.join(std::path::Path::new(raw).file_name().unwrap_or_default());
            if sibling.exists() { return Some(sibling); }
            None
        };
        let mbp10_resolved = resolve_data_path(&self.hyperparams.mbp10_data_dir);
        let trades_resolved = resolve_data_path(&self.hyperparams.trades_data_dir);
        let mbp10_path = mbp10_resolved.as_deref();
        let trades_path = trades_resolved.as_deref();

        if let Some(cached) = crate::fxcache::discover_and_load(
            std::path::Path::new(dbn_data_dir),
            symbol,
            mbp10_path,
            trades_path,
            &self.hyperparams.data_source,
            cache_dir_override,
        ) {
            // OFI is unconditional — the model requires order flow data.
            if !cached.has_ofi {
                anyhow::bail!(
                    "fxcache has_ofi=false — delete stale cache and rebuild with --mbp10-data-dir. \
                     Model requires OFI features from MBP-10 order book data."
                );
            }
            let ofi_nonzero = cached.ofi.iter().filter(|r| r.iter().any(|&v| v != 0.0)).count();
            tracing::info!("OFI loaded from fxcache: {} bars, {} non-zero rows", cached.ofi.len(), ofi_nonzero);

            let mut ofi_data = cached.ofi;
            let mut all_data: Vec<(FeatureVector, Vec<f64>)> = cached.features
                .into_iter()
                .zip(cached.targets.into_iter())
                .map(|(f, t)| {
                    (f, t.to_vec())
                })
                .collect();

            // Truncate to max_bars if configured (0 = unlimited).
            // OFI must be truncated in lockstep — downstream GPU upload asserts
            // ofi.len() == num_bars and the desync previously surfaced as a
            // misleading "fxcache is corrupt" error.
            if self.hyperparams.max_bars > 0 && all_data.len() > self.hyperparams.max_bars {
                tracing::info!("max_bars={}: truncating {} → {} bars (incl. OFI)", self.hyperparams.max_bars, all_data.len(), self.hyperparams.max_bars);
                all_data.truncate(self.hyperparams.max_bars);
                ofi_data.truncate(self.hyperparams.max_bars);
            }
            debug_assert_eq!(ofi_data.len(), all_data.len(),
                "OFI and market-data lengths must match after truncation");
            self.ofi_features = Some(Arc::from(ofi_data));

            let split = (all_data.len() * 80) / 100;
            let train = all_data[..split].to_vec();
            let val = all_data[split..].to_vec();
            return Ok((train, val));
        }

        // ── DBN fallback: load bars scoped to symbol ─────────────────────────
        let symbol_data_dir = std::path::Path::new(dbn_data_dir).join(symbol);
        let effective_dir = if symbol_data_dir.exists() {
            info!("Symbol filter: loading {} from {}", symbol, symbol_data_dir.display());
            symbol_data_dir.to_string_lossy().to_string()
        } else {
            dbn_data_dir.to_string()
        };
        let dbn_data_dir = &effective_dir;

        // ── Load bars: MBP-10 imbalance bars or OHLCV candles ─────────────
        let all_ohlcv_bars = match self.hyperparams.data_source.as_str() {
            "mbp10" => {
                if self.hyperparams.mbp10_data_dir.is_empty() {
                    anyhow::bail!("data_source='mbp10' requires mbp10_data_dir to be set in [training] config");
                }
                let mbp10_path = resolve_data_dir(&self.hyperparams.mbp10_data_dir);
                // Extract symbol from dbn_data_dir (last path component, e.g. "ES.FUT")
                let symbol = Path::new(dbn_data_dir)
                    .file_name()
                    .and_then(|s| s.to_str())
                    .unwrap_or("ES.FUT");
                info!(
                    "Loading MBP-10 imbalance bars: dir={}, symbol={}, threshold={}, alpha={}",
                    mbp10_path.display(), symbol,
                    self.hyperparams.imbalance_bar_threshold,
                    self.hyperparams.imbalance_bar_ewma_alpha,
                );
                let bars = crate::features::mbp10_loader::mbp10_to_imbalance_bars(
                    &mbp10_path,
                    symbol,
                    f64::from(self.hyperparams.imbalance_bar_threshold),
                    f64::from(self.hyperparams.imbalance_bar_ewma_alpha),
                ).map_err(|e| anyhow::anyhow!("MBP-10 imbalance bar loading failed: {}", e))?;
                info!("Loaded {} imbalance bars from MBP-10 data", bars.len());
                bars
            }
            _ => {
                // Default: load trades → front-month filter → volume bars
                let trades_dir_path = if !self.hyperparams.trades_data_dir.is_empty() {
                    resolve_data_dir(&self.hyperparams.trades_data_dir)
                } else {
                    Path::new(dbn_data_dir).to_path_buf()
                };

                if !trades_dir_path.exists() {
                    return Err(anyhow::anyhow!(
                        "Data directory not found: {}",
                        trades_dir_path.display()
                    ));
                }

                let mut trade_files = collect_dbn_files_filtered(&trades_dir_path, Some(symbol));
                trade_files.sort();

                if trade_files.is_empty() {
                    return Err(anyhow::anyhow!(
                        "No trades DBN files found for symbol '{}' in: {}",
                        symbol, trades_dir_path.display()
                    ));
                }

                info!("Found {} trades DBN files for '{}'", trade_files.len(), symbol);

                // Filter front-month PER FILE (each quarterly DBN has different front-month)
                let mut all_front_month = Vec::new();
                let mut total_raw = 0_usize;
                for file in &trade_files {
                    match crate::features::load_trades_sync(file) {
                        Ok(trades) => {
                            total_raw += trades.len();
                            let filtered = crate::features::filter_front_month(&trades);
                            debug!("  {} -> {} trades, {} front-month",
                                file.file_name().unwrap_or_default().to_string_lossy(),
                                trades.len(), filtered.len());
                            all_front_month.extend(filtered);
                        }
                        Err(e) => {
                            warn!("  Failed to load {:?}: {e}", file.file_name());
                        }
                    }
                }
                all_front_month.sort_by_key(|t| t.timestamp);

                let bars = crate::features::build_volume_bars(
                    &all_front_month, crate::features::DEFAULT_VOLUME_BAR_SIZE,
                );
                info!("Volume bars: {} bars from {} trades ({} front-month)",
                    bars.len(), total_raw, all_front_month.len());

                if bars.is_empty() {
                    return Err(anyhow::anyhow!(
                        "No volume bars built from trades. Check if files contain trade messages."
                    ));
                }

                bars
            }
        };

        // Sort bars by timestamp (critical for rolling window feature extraction)
        let mut all_ohlcv_bars = all_ohlcv_bars;
        debug!("Sorting bars chronologically by timestamp...");
        all_ohlcv_bars.sort_by_key(|bar| bar.timestamp);
        debug!("Bars sorted successfully");

        // Extract feature vectors (technical indicators, time, statistical features)
        info!("Extracting feature vectors from OHLCV bars...");
        let feature_vectors = self.extract_full_features(&all_ohlcv_bars, None)?;

        info!(
            "Extracted {} feature vectors ({} dimensions: technical indicators, time, statistical, regime features)",
            feature_vectors.len(),
            if feature_vectors.is_empty() { 42 } else { feature_vectors[0].len() }
        );

        // Compute per-bar OFI features from MBP-10 snapshots (8 features per bar)
        // Uses rayon par_iter to load files in parallel (~7-9x speedup over sequential)
        if !self.hyperparams.mbp10_data_dir.is_empty() {
            use crate::features::ofi_calculator::OFICalculator;
            use crate::features::trades_loader::get_trades_for_bar;
            use data::providers::databento::dbn_parser::DbnParser;

            let mbp10_dir = Path::new(&self.hyperparams.mbp10_data_dir);
            if mbp10_dir.exists() {
                let mut dbn_files: Vec<_> = collect_dbn_files_recursive(mbp10_dir);
                dbn_files.sort();

                if !dbn_files.is_empty() {
                    let start_mbp10 = std::time::Instant::now();
                    let n_files = dbn_files.len();
                    info!("Loading MBP-10 snapshots from {n_files} files (parallel)...");

                    // Parse all MBP-10 files in parallel with rayon.
                    // Each file produces a sorted Vec of snapshots; we merge after.
                    let per_file: Vec<Vec<_>> = {
                        use rayon::prelude::*;
                        dbn_files.par_iter().filter_map(|file| {
                            let parser = match DbnParser::new() {
                                Ok(p) => p,
                                Err(e) => { warn!("  DBN parser init failed: {e}"); return None; }
                            };
                            let mut snapshots = Vec::new();
                            match parser.parse_mbp10_streaming(file, 100, |snap| {
                                snapshots.push(snap.clone());
                            }) {
                                Ok(_) => {
                                    info!("  {} -> {} snapshots",
                                        file.file_name().unwrap_or_default().to_string_lossy(),
                                        snapshots.len());
                                    Some(snapshots)
                                }
                                Err(e) => {
                                    warn!("  Failed to load {:?}: {e}", file.file_name());
                                    None
                                }
                            }
                        }).collect()
                    };

                    let total_snaps: usize = per_file.iter().map(|v| v.len()).sum();
                    let mut all_snapshots = Vec::with_capacity(total_snaps);
                    for snaps in per_file {
                        all_snapshots.extend(snaps);
                    }
                    all_snapshots.sort_by_key(|s| s.timestamp);
                    let mbp10_secs = start_mbp10.elapsed().as_secs_f64();
                    info!("Loaded {total_snaps} MBP-10 snapshots from {n_files} files in {mbp10_secs:.1}s");

                    if !all_snapshots.is_empty() {
                        // Load trade data in parallel for VPIN/Kyle's Lambda enrichment
                        let trades = if !self.hyperparams.trades_data_dir.is_empty() {
                            use crate::features::trades_loader::load_trades_sync;
                            let trades_dir = Path::new(&self.hyperparams.trades_data_dir);
                            if trades_dir.exists() {
                                let mut trade_files: Vec<_> = collect_dbn_files_recursive(trades_dir);
                                trade_files.sort();
                                let per_file_trades: Vec<Vec<_>> = {
                                    use rayon::prelude::*;
                                    trade_files.par_iter().filter_map(|path| {
                                        load_trades_sync(path).ok()
                                    }).collect()
                                };
                                let mut all_trades = Vec::new();
                                for t in per_file_trades {
                                    all_trades.extend(t);
                                }
                                (!all_trades.is_empty()).then(|| {
                                    all_trades.sort_by_key(|t| t.timestamp);
                                    all_trades
                                })
                            } else {
                                None
                            }
                        } else {
                            None
                        };

                        info!("Computing per-bar OFI from {} MBP-10 snapshots...", all_snapshots.len());
                        let mut ofi_calculator = OFICalculator::new();
                        let mut ofi_per_bar = Vec::with_capacity(feature_vectors.len());
                        const WARMUP: usize = 50;

                        for i in 0..feature_vectors.len() {
                            let bar = &all_ohlcv_bars[i + WARMUP];
                            // Bar timestamp = close-time of bar (volume bars). The OFI
                            // window for bar t is the formation interval
                            // `(close(bar_{t-1}), close(bar_t)]` — strictly historical at
                            // the policy's decision time. Previous implementation used
                            // `[close(bar_t), close(bar_{t+1}))` which leaks bar t+1's
                            // microstructure (audit `docs/lookahead-bias-audit-2026-04-28.md`
                            // §3). Migrated in lockstep with the canonical precompute
                            // path at `crates/ml/examples/precompute_features.rs:441-475`.
                            let bar_ts = bar.timestamp.timestamp_nanos_opt().unwrap_or(0) as u64;
                            let bar_start_ts = if i + WARMUP >= 1 {
                                all_ohlcv_bars[i + WARMUP - 1].timestamp.timestamp_nanos_opt().unwrap_or(0) as u64
                            } else {
                                bar_ts
                            };

                            if let Some(ref all_trades) = trades {
                                // Half-open at start, closed at end: trade-at-close
                                // included in bar t (which it formed).
                                let bar_trades = get_trades_for_bar(all_trades, bar_start_ts, bar_ts.saturating_add(1));
                                for trade in bar_trades {
                                    ofi_calculator.feed_trade(trade.price, trade.volume, trade.is_buy);
                                }
                            }

                            // Formation-interval snapshot: latest snapshot at-or-before
                            // close(bar_t) AND strictly after close(bar_{t-1}).
                            let snap_end = all_snapshots.partition_point(|s| s.timestamp <= bar_ts);
                            let snap_start = all_snapshots.partition_point(|s| s.timestamp <= bar_start_ts);
                            let window = if snap_end > snap_start {
                                &all_snapshots[snap_start..snap_end]
                            } else {
                                &all_snapshots[0..0]
                            };
                            if let Some(snap) = window.last() {
                                match ofi_calculator.calculate(snap) {
                                    Ok(features) => {
                                        if features.is_valid() {
                                            let arr8 = features.to_array();
                                            let mut ofi_row = [0.0_f64; OFI_DIM];
                                            ofi_row[..8].copy_from_slice(&arr8);
                                            // Book aggression from MBP-10 depth
                                            let mut buy_ws = 0.0_f64;
                                            let mut buy_t = 0.0_f64;
                                            let mut sell_ws = 0.0_f64;
                                            let mut sell_t = 0.0_f64;
                                            for (lv, pair) in snap.levels.iter().enumerate().take(10) {
                                                let bs = pair.bid_sz as f64;
                                                let as_ = pair.ask_sz as f64;
                                                buy_ws += (lv + 1) as f64 * bs;
                                                buy_t += bs;
                                                sell_ws += (lv + 1) as f64 * as_;
                                                sell_t += as_;
                                            }
                                            let buy_com = if buy_t > 0.0 { buy_ws / buy_t } else { 5.5 };
                                            let sell_com = if sell_t > 0.0 { sell_ws / sell_t } else { 5.5 };
                                            ofi_row[16] = (sell_com - buy_com) / 10.0;
                                            // Slots [17..OFI_DIM) stay zero in this DBN-fallback path;
                                            // the canonical signal source is precompute_features
                                            // which fills every slot from full MBP-10 history.
                                            ofi_per_bar.push(ofi_row);
                                        } else {
                                            ofi_per_bar.push([0.0; OFI_DIM]);
                                        }
                                    }
                                    Err(_) => ofi_per_bar.push([0.0; OFI_DIM]),
                                }
                            } else {
                                ofi_per_bar.push([0.0; OFI_DIM]);
                            }
                        }

                        // Compute OFI temporal deltas: delta[bar] = ofi[bar] - ofi[bar-1]
                        for i in (1..ofi_per_bar.len()).rev() {
                            for k in 0..8 {
                                ofi_per_bar[i][8 + k] = ofi_per_bar[i][k] - ofi_per_bar[i - 1][k];
                            }
                        }

                        // Compute log bar duration
                        for i in 0..ofi_per_bar.len() {
                            let duration_secs = if i > 0 {
                                let ts_cur = all_ohlcv_bars[i + WARMUP].timestamp;
                                let ts_prev = all_ohlcv_bars[i + WARMUP - 1].timestamp;
                                (ts_cur - ts_prev).num_seconds() as f64
                            } else {
                                60.0
                            };
                            ofi_per_bar[i][17] = duration_secs.max(0.1).ln() / 10.0;
                        }

                        let non_zero = ofi_per_bar.iter().filter(|f| f.iter().any(|&v| v != 0.0)).count();
                        info!("Per-bar OFI computed: {} total, {} non-zero ({:.1}%)",
                            ofi_per_bar.len(), non_zero,
                            if ofi_per_bar.is_empty() { 0.0 } else { non_zero as f64 / ofi_per_bar.len() as f64 * 100.0 });
                        self.ofi_features = Some(Arc::from(ofi_per_bar));
                    }
                }
            }
        }

        // Create training data pairs (features, target)
        // Target layout: [preproc_close, preproc_next, raw_close, raw_next, raw_open, mid_price_open]
        //   [0:1] = log-return-normalized close/next (network input; matches contract
        //           in experience_kernels.cu:1556 + cuda_pipeline/mod.rs:508)
        //   [2:3] = raw dollar prices for portfolio simulation P&L + tx costs
        //   [4]   = raw open price
        //   [5]   = mid-price at bar open (MBP-10 midpoint; fallback to raw_open)
        //
        // Prior implementation wrote raw prices to all 6 columns (target[0:2] should
        // have been preproc_close / preproc_next per documented contract). The
        // raw-price values at target[0:1] were never consumed — production aux head
        // reads next_states[i][0] (= MARKET feat[0] = log_return), not targets.
        // Fixed here so the on-disk fxcache matches its own documented schema.
        let mut training_data = Vec::new();
        for i in 0..feature_vectors.len().saturating_sub(1) {
            let raw_current = all_ohlcv_bars[i + 50].close;
            let raw_next = all_ohlcv_bars[i + 1 + 50].close;
            let raw_open = all_ohlcv_bars[i + 50].open;
            // prev_close: 1-bar lag for log-return computation. Within the 50-bar
            // warmup window we are guaranteed to have all_ohlcv_bars[i + 49] valid
            // (extraction needs ≥50 bars of history before emitting feat 0).
            let prev_close = all_ohlcv_bars[i + 49].close;
            let preproc_close = if prev_close > 0.0 {
                (raw_current / prev_close).ln()
            } else { 0.0 };
            let preproc_next = if raw_current > 0.0 {
                (raw_next / raw_current).ln()
            } else { 0.0 };
            training_data.push((
                feature_vectors[i],
                vec![preproc_close, preproc_next, raw_current, raw_next, raw_open, raw_open],
            ));
        }
        // Last sample targets itself — preproc returns are 0.0 (no future bar).
        if !feature_vectors.is_empty() {
            let idx = all_ohlcv_bars.len() - 1;
            let raw_close = all_ohlcv_bars[idx].close;
            let raw_open = all_ohlcv_bars[idx].open;
            training_data.push((
                feature_vectors[feature_vectors.len() - 1],
                vec![0.0, 0.0, raw_close, raw_close, raw_open, raw_open],
            ));
        }

        let feature_dim = if feature_vectors.is_empty() { 42 } else { feature_vectors[0].len() };
        info!(
            "Created {} total samples with {}-dim features",
            training_data.len(),
            feature_dim
        );

        // Truncate to max_bars if configured (0 = unlimited).
        // OFI (if present from the DBN path above) must be truncated in lockstep —
        // downstream GPU upload asserts ofi.len() == num_bars.
        if self.hyperparams.max_bars > 0 && training_data.len() > self.hyperparams.max_bars {
            tracing::info!("max_bars={}: truncating {} → {} bars (incl. OFI)", self.hyperparams.max_bars, training_data.len(), self.hyperparams.max_bars);
            training_data.truncate(self.hyperparams.max_bars);
            if let Some(ofi_arc) = self.ofi_features.take() {
                let mut ofi_vec: Vec<[f64; OFI_DIM]> = ofi_arc.iter().copied().collect();
                ofi_vec.truncate(self.hyperparams.max_bars);
                self.ofi_features = Some(Arc::from(ofi_vec));
            }
        }
        debug_assert!(
            self.ofi_features.as_ref().map_or(true, |o| o.len() == training_data.len()),
            "OFI and training-data lengths must match after truncation",
        );

        // Split training data 80/20 for train/validation
        let split_idx = (training_data.len() * 80) / 100;
        let train_data = training_data[..split_idx].to_vec(); // cpu-side split
        let val_data = training_data[split_idx..].to_vec(); // cpu-side split

        info!(
            "Split data - Training samples: {}, Validation samples: {}",
            train_data.len(),
            val_data.len()
        );

        Ok((train_data, val_data))
    }

    /// Extract raw OHLCV bars from DBN file using official dbn crate decoder
    ///
    /// This replaces the custom parser that only extracted 2 messages (header metadata).
    /// Now extracts all OHLCV bars (400-500+ records per file).
    ///
    /// Public for testing purposes.
    pub fn extract_ohlcv_bars_from_dbn(&self, file_path: &Path) -> Result<Vec<OHLCVBar>> {
        extract_ohlcv_bars_from_dbn(file_path)
    }

    fn decode_ohlcv_records<R: std::io::Read>(
        decoder: &mut dbn::decode::dbn::Decoder<R>,
        file_path: &Path,
    ) -> Result<Vec<OHLCVBar>> {
        use dbn::decode::{DbnMetadata, DecodeRecordRef};

        let mut ohlcv_bars = Vec::new();

        // Read metadata — reject non-OHLCV schemas at file level
        let metadata = decoder.metadata();
        debug!(
            "DBN file metadata: dataset={:?}, schema={:?}, symbols={:?}",
            metadata.dataset, metadata.schema, metadata.symbols
        );

        // Reject files with non-OHLCV schema upfront (MBP-10, trades, etc.)
        if let Some(schema) = metadata.schema {
            let schema_str = format!("{schema:?}");
            if !schema_str.contains("Ohlcv") {
                tracing::warn!(
                    schema = %schema_str,
                    file = %file_path.display(),
                    "Skipping non-OHLCV DBN file"
                );
                return Ok(Vec::new());
            }
        }

        // Decode all OHLCV records
        let mut ohlcv_count = 0;
        let mut other_count = 0;
        let mut idx = 0;

        loop {
            match decoder.decode_record_ref() {
                Ok(Some(record)) => {
                    idx += 1;

                    // Convert RecordRef to RecordRefEnum for pattern matching
                    let record_enum = record
                        .as_enum()
                        .map_err(|e| anyhow::anyhow!(
                            "Corrupt record at idx {} in {}: {}",
                            idx, file_path.display(), e
                        ))?;

                    match record_enum {
                        dbn::RecordRefEnum::Ohlcv(ohlcv) => {
                            ohlcv_count += 1;

                            // Extract OHLCV values (prices are i64 scaled by 1e-9 per DBN spec, volume is u64)
                            let open_f64 = ohlcv.open as f64 * 1e-9;
                            let high_f64 = ohlcv.high as f64 * 1e-9;
                            let low_f64 = ohlcv.low as f64 * 1e-9;
                            let close_f64 = ohlcv.close as f64 * 1e-9;
                            let volume_u64 = ohlcv.volume;

                            // WAVE 8 AGENT 36: Validate all price values are finite (not NaN/Inf)
                            // Skip bars with invalid data to prevent NaN propagation
                            if !open_f64.is_finite()
                                || !high_f64.is_finite()
                                || !low_f64.is_finite()
                                || !close_f64.is_finite()
                            {
                                debug!(
                                    "Skipping OHLCV bar {} with non-finite values: open={}, high={}, low={}, close={}",
                                    ohlcv_count, open_f64, high_f64, low_f64, close_f64
                                );
                                continue;
                            }

                            // Log first few records for validation
                            if ohlcv_count <= 5 {
                                debug!(
                                    "Raw OHLCV #{}: open={}, high={}, low={}, close={}",
                                    ohlcv_count, ohlcv.open, ohlcv.high, ohlcv.low, ohlcv.close
                                );
                                debug!(
                                    "Scaled OHLCV #{}: open={:.6}, high={:.6}, low={:.6}, close={:.6}",
                                    ohlcv_count, open_f64, high_f64, low_f64, close_f64
                                );
                            }

                            // Convert timestamp from nanoseconds since epoch to DateTime
                            let timestamp_nanos = ohlcv.hd.ts_event as i64;
                            let timestamp_secs = timestamp_nanos / 1_000_000_000;
                            let timestamp_nanos_remainder =
                                (timestamp_nanos.rem_euclid(1_000_000_000)) as u32;
                            let timestamp = chrono::DateTime::<chrono::Utc>::from_timestamp(
                                timestamp_secs,
                                timestamp_nanos_remainder,
                            )
                            .unwrap_or_else(|| chrono::Utc::now());

                            // Create OHLCVBar for feature extraction pipeline
                            let bar = OHLCVBar {
                                timestamp,
                                open: open_f64,
                                high: high_f64,
                                low: low_f64,
                                close: close_f64,
                                volume: volume_u64 as f64,
                            };

                            ohlcv_bars.push(bar);
                        },
                        dbn::RecordRefEnum::Mbo(_)
                        | dbn::RecordRefEnum::Trade(_)
                        | dbn::RecordRefEnum::Mbp1(_)
                        | dbn::RecordRefEnum::Mbp10(_)
                        | dbn::RecordRefEnum::Status(_)
                        | dbn::RecordRefEnum::InstrumentDef(_)
                        | dbn::RecordRefEnum::Imbalance(_)
                        | dbn::RecordRefEnum::Stat(_)
                        | dbn::RecordRefEnum::Error(_)
                        | dbn::RecordRefEnum::SymbolMapping(_)
                        | dbn::RecordRefEnum::System(_)
                        | dbn::RecordRefEnum::Cmbp1(_)
                        | dbn::RecordRefEnum::Bbo(_)
                        | dbn::RecordRefEnum::Cbbo(_) => {
                            other_count += 1;
                            if other_count <= 5 {
                                debug!("Skipping non-OHLCV record at index {}", idx);
                            }
                        },
                    }
                },
                Ok(None) => {
                    // End of stream
                    break;
                },
                Err(e) => {
                    return Err(anyhow::anyhow!(
                        "Failed to decode record {} in {}: {}",
                        idx, file_path.display(), e
                    ));
                },
            }
        }

        info!(
            "Extracted {} OHLCV bars from {:?} ({} other records skipped)",
            ohlcv_count,
            file_path.file_name().unwrap_or_default(),
            other_count
        );

        Ok(ohlcv_bars)
    }

    /// Create features from OHLCV data
    pub(crate) fn create_ohlcv_features(
        &self,
        open: f64,
        high: f64,
        low: f64,
        close: f64,
        volume: u64,
    ) -> Result<FinancialFeatures> {
        use std::collections::HashMap;

        // Use absolute values for Price type (futures data can have negative values)
        // For ML training, the absolute magnitude is what matters for feature extraction
        let close_price =
            common::Price::from_f64(close.abs()).unwrap_or_else(|_| common::Price::ZERO);
        let open_price =
            common::Price::from_f64(open.abs()).unwrap_or_else(|_| common::Price::ZERO);
        let high_price =
            common::Price::from_f64(high.abs()).unwrap_or_else(|_| common::Price::ZERO);
        let low_price = common::Price::from_f64(low.abs()).unwrap_or_else(|_| common::Price::ZERO);

        // Calculate technical indicators
        let mut indicators = HashMap::new();

        // Price-based features
        let price_range = high - low;
        let body_size = (close - open).abs();
        let upper_shadow = high - close.max(open);
        let lower_shadow = close.min(open) - low;

        indicators.insert("price_range".to_owned(), price_range);
        indicators.insert("body_size".to_owned(), body_size);
        indicators.insert("upper_shadow".to_owned(), upper_shadow);
        indicators.insert("lower_shadow".to_owned(), lower_shadow);
        indicators.insert("close_to_high".to_owned(), (close - high).abs());
        indicators.insert("close_to_low".to_owned(), (close - low).abs());

        // Microstructure features
        let spread_bps = ((high - low) / close * 10000.0) as i32;
        let trade_intensity = volume as f64;

        Ok(FinancialFeatures {
            prices: vec![open_price, high_price, low_price, close_price],
            volumes: vec![volume as i64],
            technical_indicators: indicators,
            microstructure: crate::training_pipeline::MicrostructureFeatures {
                spread_bps,
                imbalance: 0.0, // Not available from OHLCV
                trade_intensity,
                vwap: close_price, // Approximate VWAP as close
            },
            risk_metrics: crate::training_pipeline::RiskFeatures {
                var_5pct: -0.02, // Placeholder
                expected_shortfall: -0.03,
                max_drawdown: -0.05,
                sharpe_ratio: 1.0,
            },
            timestamp: chrono::Utc::now(),
        })
    }
}
