//! Trade Data Loader for DBN `Schema::Trades`
//!
//! Loads Databento trade records from `.dbn` or `.dbn.zst` files and provides
//! time-windowed access for feeding into `OFICalculator::feed_trade()` (VPIN,
//! Kyle's Lambda).

use std::fs::File;
use std::io::BufReader;
use std::path::Path;

use dbn::decode::{DbnDecoder, DecodeRecordRef};
use dbn::RecordRefEnum;

use crate::MLError;

/// A single parsed trade from a DBN file.
#[derive(Debug, Clone, Copy)]
pub struct DbnTrade {
    /// Event timestamp in nanoseconds since Unix epoch
    pub timestamp: u64,
    /// Trade price (f64, converted from i64 scaled by 1e-9)
    pub price: f64,
    /// Trade size in contracts
    pub volume: u64,
    /// True if buyer-initiated (side == 'B'), false if seller-initiated
    pub is_buy: bool,
}

/// Load all trades from a `.dbn` or `.dbn.zst` file (synchronous).
///
/// Auto-detects zstd compression by file extension. Returns trades sorted
/// by timestamp (as they appear in the file — DBN files are already ordered).
pub fn load_trades_sync(file_path: &Path) -> Result<Vec<DbnTrade>, MLError> {
    let file = File::open(file_path).map_err(|e| {
        MLError::InsufficientData(format!("Failed to open trade file {:?}: {}", file_path, e))
    })?;

    let is_zstd = file_path
        .to_string_lossy()
        .ends_with(".dbn.zst");

    let reader: Box<dyn std::io::Read> = if is_zstd {
        Box::new(
            zstd::Decoder::new(BufReader::new(file)).map_err(|e| {
                MLError::InsufficientData(format!("Failed to create zstd decoder: {}", e))
            })?,
        )
    } else {
        Box::new(BufReader::new(file))
    };

    let mut decoder = DbnDecoder::new(reader).map_err(|e| {
        MLError::InsufficientData(format!("Failed to create DBN decoder: {}", e))
    })?;

    let mut trades = Vec::new();

    loop {
        match decoder.decode_record_ref() {
            Ok(Some(record)) => {
                let record_enum = record.as_enum().map_err(|e| {
                    MLError::InsufficientData(format!("Failed to convert DBN record: {}", e))
                })?;

                if let RecordRefEnum::Trade(trade) = record_enum {
                    // Prices are i64 scaled by 1e-9 per DBN specification
                    let price_f64 = trade.price as f64 * 1e-9;
                    // Side: trade.side == b'B' as i8 => Buy, b'A' as i8 => Sell
                    let is_buy = trade.side == b'B' as i8;

                    trades.push(DbnTrade {
                        timestamp: trade.hd.ts_event,
                        price: price_f64,
                        volume: u64::from(trade.size),
                        is_buy,
                    });
                }
            }
            Ok(None) => break,
            Err(e) => {
                return Err(MLError::InsufficientData(format!(
                    "DBN decode error: {}",
                    e
                )));
            }
        }
    }

    // Sort by timestamp (should already be ordered, but ensure correctness)
    trades.sort_by_key(|t| t.timestamp);

    Ok(trades)
}

/// Get the sub-slice of trades within `[bar_start_ns, bar_end_ns)`.
///
/// Uses binary search for O(log n) lookup on the pre-sorted trades slice.
pub fn get_trades_for_bar(trades: &[DbnTrade], bar_start_ns: u64, bar_end_ns: u64) -> &[DbnTrade] {
    if trades.is_empty() {
        return &[];
    }

    // First trade with timestamp >= bar_start_ns
    let start = trades.partition_point(|t| t.timestamp < bar_start_ns);
    // First trade with timestamp >= bar_end_ns
    let end = trades.partition_point(|t| t.timestamp < bar_end_ns);

    trades.get(start..end).unwrap_or(&[])
}

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

    fn make_trade(timestamp: u64, price: f64, volume: u64, is_buy: bool) -> DbnTrade {
        DbnTrade {
            timestamp,
            price,
            volume,
            is_buy,
        }
    }

    #[test]
    fn test_get_trades_for_bar_empty() {
        let trades: &[DbnTrade] = &[];
        let result = get_trades_for_bar(trades, 1000, 2000);
        assert!(result.is_empty());
    }

    #[test]
    fn test_get_trades_for_bar_binary_search() {
        let trades = vec![
            make_trade(1000, 100.0, 10, true),
            make_trade(2000, 101.0, 20, false),
            make_trade(3000, 102.0, 30, true),
            make_trade(4000, 103.0, 40, false),
            make_trade(5000, 104.0, 50, true),
        ];

        // Window [2000, 4000) should return trades at timestamps 2000 and 3000
        let result = get_trades_for_bar(&trades, 2000, 4000);
        assert_eq!(result.len(), 2);
        assert_eq!(result.first().map(|t| t.timestamp), Some(2000));
        assert_eq!(result.last().map(|t| t.timestamp), Some(3000));

        // Window [1000, 6000) should return all 5 trades
        let all = get_trades_for_bar(&trades, 1000, 6000);
        assert_eq!(all.len(), 5);

        // Window [5001, 6000) should return nothing (past all trades)
        let none = get_trades_for_bar(&trades, 5001, 6000);
        assert!(none.is_empty());

        // Window [0, 1000) should return nothing (before all trades)
        let before = get_trades_for_bar(&trades, 0, 1000);
        assert!(before.is_empty());
    }
}
