foxhunt/backtesting/tests/dbn_backtest_integration.rs

//! End-to-end integration tests for the backtesting vertical slice.
//!
//! Validates the full pipeline: synthetic data -> model loading -> replay -> prediction.
//! No real .dbn files, no PostgreSQL, no trained checkpoints required.

// Integration tests live outside the crate, so the crate-level deny lints don't apply.
// Tests are allowed to use .unwrap().

use backtesting::dbn_converter::processed_to_market_event;
use backtesting::dbn_replay::DbnReplayEngine;
use backtesting::model_loader::{load_models_for_backtest, BacktestMLModel, ModelSpec};
use backtesting::replay_engine::ReplayEvent;

use chrono::{DateTime, TimeDelta, Utc};
use common::{OrderSide, Price, Quantity, Symbol};
use data::providers::databento::dbn_parser::ProcessedMessage;
use ml::dqn::dqn::DQNConfig;
use ml::ensemble::adapters::dqn::DqnInferenceAdapter;
use ml::features::ProductionFeatureExtractorAdapter;
use ml::{get_global_registry, Features, MLModel, ModelType};
use rust_decimal::Decimal;
use trading_engine::timing::{HardwareTimestamp, TimingSource};
use trading_engine::types::events::MarketEvent;

// ---------------------------------------------------------------------------
// Helpers
// ---------------------------------------------------------------------------

/// Build a `HardwareTimestamp` from a `DateTime<Utc>`.
fn ts_from_dt(dt: DateTime<Utc>) -> HardwareTimestamp {
    #[allow(clippy::cast_sign_loss)]
    let nanos = dt.timestamp_nanos_opt().unwrap_or(0) as u64;
    HardwareTimestamp {
        cycles: 0,
        nanos,
        source: TimingSource::SystemClock,
        validation_passed: true,
    }
}

/// Create a synthetic `MarketEvent::Trade`.
fn make_trade_event(symbol: &str, price_f64: f64, volume: f64, ts: DateTime<Utc>) -> MarketEvent {
    MarketEvent::Trade {
        symbol: Symbol::new(symbol.to_string()),
        price: Price::from_f64(price_f64).unwrap_or(Price::ZERO),
        size: Quantity::from_f64(volume).unwrap_or(Quantity::ZERO),
        timestamp: ts,
        side: Some(OrderSide::Buy),
        venue: None,
        trade_id: None,
    }
}

/// Wrap a `MarketEvent` into a `ReplayEvent` with a given sequence and timestamp.
fn make_replay_event(event: MarketEvent, sequence: u64, ts: DateTime<Utc>) -> ReplayEvent {
    ReplayEvent {
        event,
        original_timestamp: ts,
        replay_timestamp: ts,
        source_id: "synthetic".to_string(),
        sequence,
    }
}

// ---------------------------------------------------------------------------
// Test 1: DBN converter roundtrip
// ---------------------------------------------------------------------------

#[test]
fn test_dbn_converter_roundtrip() {
    let now = Utc::now();
    let hw_ts = ts_from_dt(now);
    let price = Price::from_f64(150.25).unwrap();

    let msg = ProcessedMessage::Trade {
        symbol: "AAPL".to_string(),
        timestamp: hw_ts,
        price,
        size: Decimal::new(200, 0), // 200 shares
        side: OrderSide::Buy,
        trade_id: Some("T42".to_string()),
        conditions: vec![],
    };

    let event = processed_to_market_event(&msg);
    assert!(event.is_some(), "Trade ProcessedMessage should convert to MarketEvent");

    let event = event.unwrap();
    if let MarketEvent::Trade {
        symbol,
        price: p,
        size,
        side,
        trade_id,
        ..
    } = &event
    {
        assert_eq!(symbol.as_str(), "AAPL");
        assert_eq!(*p, price);
        assert!(
            (size.to_f64() - 200.0).abs() < f64::EPSILON,
            "size should be 200, got {}",
            size.to_f64()
        );
        assert_eq!(*side, Some(OrderSide::Buy));
        assert_eq!(*trade_id, Some("T42".to_string()));
    } else {
        panic!("Expected MarketEvent::Trade variant");
    }
}

// ---------------------------------------------------------------------------
// Test 2: Replay engine streams events in timestamp order
// ---------------------------------------------------------------------------

#[tokio::test]
async fn test_replay_engine_streams_events() {
    let base = DateTime::from_timestamp(1_700_000_000, 0).unwrap();

    // Create 10 synthetic events with trending price 100 -> 109
    let events: Vec<ReplayEvent> = (0..10)
        .map(|i| {
            let ts = base + TimeDelta::seconds(i as i64);
            let price = 100.0 + i as f64;
            let event = make_trade_event("SYNTH", price, 50.0, ts);
            make_replay_event(event, i as u64, ts)
        })
        .collect();

    let engine = DbnReplayEngine::from_events(events);
    assert_eq!(engine.event_count(), 10);

    let mut rx = engine.start_replay();

    let mut received = Vec::new();
    while let Some(ev) = rx.recv().await {
        received.push(ev);
    }

    assert_eq!(received.len(), 10, "Should receive all 10 events");

    // Verify chronological order
    for pair in received.windows(2) {
        let a = pair.first().unwrap();
        let b = pair.get(1).unwrap();
        assert!(
            a.original_timestamp <= b.original_timestamp,
            "Events not in chronological order: {:?} > {:?}",
            a.original_timestamp,
            b.original_timestamp,
        );
    }

    // Verify price trend
    if let MarketEvent::Trade { price: first_price, .. } = &received.first().unwrap().event {
        if let MarketEvent::Trade { price: last_price, .. } = &received.last().unwrap().event {
            assert!(
                last_price > first_price,
                "Price should trend upward: first={}, last={}",
                first_price, last_price
            );
        }
    }
}

// ---------------------------------------------------------------------------
// Test 3: DQN model loads with random weights and predicts
// ---------------------------------------------------------------------------

#[tokio::test]
async fn test_model_loads_and_predicts() {
    let config = DQNConfig {
        state_dim: 51,
        num_actions: 45,
        hidden_dims: vec![64, 64],
        ..Default::default()
    };

    let adapter = DqnInferenceAdapter::new(config).unwrap();
    let model = BacktestMLModel::new(Box::new(adapter), ModelType::DQN);

    assert!(model.is_ready(), "Model should be ready after creation");
    assert_eq!(model.name(), "DQN");
    assert_eq!(model.model_type(), ModelType::DQN);

    // Build a 51-dim feature vector
    let features = Features::new(
        vec![0.1; 51],
        (0..51).map(|i| format!("f{i}")).collect(),
    );

    let prediction = model.predict(&features).await.unwrap();

    // Direction must be in [-1, 1]
    assert!(
        prediction.value >= -1.0 && prediction.value <= 1.0,
        "direction {} out of [-1, 1]",
        prediction.value,
    );
    // Confidence must be in [0, 1]
    assert!(
        prediction.confidence >= 0.0 && prediction.confidence <= 1.0,
        "confidence {} out of [0, 1]",
        prediction.confidence,
    );
    // Model id should be "DQN"
    assert_eq!(prediction.model_id, "DQN");
}

// ---------------------------------------------------------------------------
// Test 4: Production feature extractor produces exactly 51 dimensions
// ---------------------------------------------------------------------------

#[test]
fn test_production_features_51_dim() {
    use common::ml_strategy::ProductionFeatureExtractor225;

    let mut extractor = ProductionFeatureExtractorAdapter::new();

    // Feed 55 price updates (past the warmup period of 50)
    for i in 0..55 {
        let price = 100.0 + i as f64 * 0.1;
        let volume = 1000.0;
        let timestamp = Utc::now();
        extractor
            .update(price, volume, timestamp)
            .unwrap_or_else(|e| {
                panic!("Production extractor update failed at step {}: {}", i, e)
            });
    }

    let features = extractor
        .extract_features()
        .unwrap_or_else(|e| panic!("extract_features failed: {}", e));

    assert_eq!(
        features.len(),
        51,
        "Production extractor should produce exactly 51 features, got {}",
        features.len()
    );

    // Validate all features are finite
    for (i, val) in features.iter().enumerate() {
        assert!(
            val.is_finite(),
            "Feature {} should be finite, found {}",
            i, val
        );
    }
}

// ---------------------------------------------------------------------------
// Test 5: Full pipeline -- events -> feature extractor -> model prediction
// ---------------------------------------------------------------------------

#[tokio::test]
async fn test_full_pipeline_synthetic_backtest() {
    use common::ml_strategy::ProductionFeatureExtractor225;

    // --- Step 1: Register a DQN model in the global registry ---
    let specs = vec![ModelSpec {
        model_type: "DQN".to_string(),
        checkpoint_path: String::new(), // random weights
        weight: 1.0,
    }];

    load_models_for_backtest(&specs)
        .await
        .unwrap_or_else(|e| panic!("load_models_for_backtest failed: {}", e));

    let registry = get_global_registry();
    let model = registry.get("DQN").await;
    assert!(model.is_some(), "DQN should be registered in the global registry");

    // --- Step 2: Create 100 synthetic ReplayEvents with trending prices ---
    let base = DateTime::from_timestamp(1_700_000_000, 0).unwrap();
    let events: Vec<ReplayEvent> = (0..100)
        .map(|i| {
            let ts = base + TimeDelta::seconds(i as i64);
            let price = 100.0 + (i as f64); // 100 -> 199
            let volume = 500.0 + (i as f64) * 2.0;
            let event = make_trade_event("SYNTH", price, volume, ts);
            make_replay_event(event, i as u64, ts)
        })
        .collect();

    // --- Step 3: Create replay engine and stream events ---
    let engine = DbnReplayEngine::from_events(events);
    assert_eq!(engine.event_count(), 100);

    let mut rx = engine.start_replay();

    // --- Step 4: Consume events, update feature extractor, run predictions ---
    let mut extractor = ProductionFeatureExtractorAdapter::new();
    let mut events_processed: u64 = 0;
    let mut predictions_made: u64 = 0;
    let warmup_period: u64 = 55; // extractor needs ~50 bars for warmup

    while let Some(replay_event) = rx.recv().await {
        events_processed += 1;

        // Extract price and volume from the MarketEvent
        if let MarketEvent::Trade {
            price, size, timestamp, ..
        } = &replay_event.event
        {
            let price_f64 = price.to_decimal().unwrap_or(Decimal::ZERO);
            let volume_f64 = size.to_f64();

            // Update the feature extractor
            let price_f64_val: f64 = price_f64.try_into().unwrap_or(0.0);
            let _ = extractor.update(price_f64_val, volume_f64, *timestamp);

            // After warmup, extract features and predict
            if events_processed >= warmup_period {
                match extractor.extract_features() {
                    Ok(feat_values) if feat_values.len() == 51 => {
                        // Verify exactly 51 dimensions
                        assert_eq!(feat_values.len(), 51, "Feature vector should be 51-dim");

                        let feature_names: Vec<String> = (0..51)
                            .map(|i| format!("prod_feature_{}", i))
                            .collect();

                        let features = Features {
                            values: feat_values,
                            names: feature_names,
                            timestamp: timestamp.timestamp_micros() as u64,
                            symbol: Some("SYNTH".to_string()),
                        };

                        // Run prediction through the global registry
                        let results = registry
                            .predict_selected(&["DQN".to_string()], &features)
                            .await;

                        for result in results {
                            match result {
                                Ok(pred) => {
                                    predictions_made += 1;
                                    // Validate prediction bounds
                                    assert!(
                                        pred.value >= -1.0 && pred.value <= 1.0,
                                        "Prediction value {} out of [-1, 1]",
                                        pred.value
                                    );
                                    assert!(
                                        pred.confidence >= 0.0 && pred.confidence <= 1.0,
                                        "Prediction confidence {} out of [0, 1]",
                                        pred.confidence
                                    );
                                }
                                Err(e) => {
                                    panic!("DQN prediction failed: {:?}", e);
                                }
                            }
                        }
                    }
                    Ok(feat) => {
                        // Feature extractor not yet warmed up, fewer than 51 features
                        eprintln!(
                            "Event {}: extractor returned {} features (expected 51, still warming up)",
                            events_processed,
                            feat.len()
                        );
                    }
                    Err(e) => {
                        // Expected during warmup
                        eprintln!(
                            "Event {}: extractor not ready yet: {}",
                            events_processed, e
                        );
                    }
                }
            }
        }
    }

    // --- Step 5: Verify results ---
    assert_eq!(events_processed, 100, "Should have processed all 100 events");
    assert!(
        predictions_made > 0,
        "Should have made at least one prediction after warmup (made {})",
        predictions_made,
    );

    println!(
        "Full pipeline test passed: {} events processed, {} predictions made",
        events_processed, predictions_made
    );
}