foxhunt/tests/integration/broker_risk_integration.rs

//! Broker to Risk System Integration Tests
//! 
//! Tests comprehensive integration between broker operations and risk management systems.
//! Validates real-time risk assessment, emergency stops, and broker response coordination.
//! 
//! Coverage Areas:
//! - Broker connection to risk system integration
//! - Real-time risk assessment during order flow
//! - Emergency stop mechanisms
//! - Risk limit enforcement across brokers
//! - Position sizing validation
//! - Market data to risk calculation pipeline

use std::sync::Arc;
use std::time::Duration;
use tokio::time::timeout;

// Import core types and modules
use trading_engine::{
    timing::HardwareTimestamp,
    types::prelude::*,
    trading::{
        engine::TradingEngine,
        broker_client::BrokerClient,
        order_manager::OrderManager,
        position_manager::PositionManager,
    },
    brokers::{
        config::BrokerConnectorConfig,
        error::BrokerError,
    },
    simd::SimdPriceOps,
    lockfree::LockFreeRingBuffer,
};

/// Test result type for safe error handling (no panics)
type TestResult<T> = Result<T, Box<dyn std::error::Error + Send + Sync>>;

/// Integration test module configuration
#[derive(Debug, Clone)]
pub struct BrokerRiskTestConfig {
    pub broker_endpoints: Vec<String>,
    pub risk_limits: RiskLimits,
    pub max_position_size: Decimal,
    pub emergency_stop_threshold: Decimal,
    pub test_timeout_ms: u64,
}

impl Default for BrokerRiskTestConfig {
    fn default() -> Self {
        Self {
            broker_endpoints: vec![
                "localhost:8080".to_string(),  // Mock broker 1
                "localhost:8081".to_string(),  // Mock broker 2
            ],
            risk_limits: RiskLimits::default(),
            max_position_size: Decimal::new(100_000, 2), // $1000.00
            emergency_stop_threshold: Decimal::new(5, 2), // 5% loss
            test_timeout_ms: 30_000, // 30 seconds
        }
    }
}

#[derive(Debug, Clone)]
pub struct RiskLimits {
    pub max_order_value: Decimal,
    pub max_daily_loss: Decimal,
    pub max_position_concentration: Decimal,
    pub var_limit: Decimal,
}

impl Default for RiskLimits {
    fn default() -> Self {
        Self {
            max_order_value: Decimal::new(50_000, 2), // $500.00
            max_daily_loss: Decimal::new(1_000_00, 2), // $1000.00
            max_position_concentration: Decimal::new(25, 2), // 25%
            var_limit: Decimal::new(2_000_00, 2), // $2000.00 VaR
        }
    }
}

/// Mock risk engine for integration testing
#[derive(Clone)]
struct MockRiskEngine {
    risk_limits: RiskLimits,
    current_positions: Arc<std::sync::Mutex<Vec<Position>>>,    daily_pnl: Arc<std::sync::Mutex<Decimal>>,
    emergency_stop_active: Arc<std::sync::atomic::AtomicBool>,
}

impl MockRiskEngine {
    pub fn new(config: BrokerRiskTestConfig) -> Self {
        Self {
            risk_limits: config.risk_limits,
            current_positions: Arc::new(std::sync::Mutex::new(Vec::new())),
            daily_pnl: Arc::new(std::sync::Mutex::new(Decimal::ZERO)),
            emergency_stop_active: Arc::new(std::sync::atomic::AtomicBool::new(false)),
        }
    }

    /// Validate order against risk limits
    pub async fn validate_order(&self, order: &Order) -> TestResult<RiskAssessment> {
        let start_time = HardwareTimestamp::now();
        
        // Check if emergency stop is active
        if self.emergency_stop_active.load(std::sync::atomic::Ordering::Acquire) {
            return Ok(RiskAssessment {
                approved: false,
                reason: "Emergency stop active".to_string(),
                risk_score: Decimal::ONE,
                validation_latency_ns: HardwareTimestamp::now().latency_ns(&start_time),
            });
        }
        
        // Validate order value
        let order_value = order.price * order.quantity;
        if order_value > self.risk_limits.max_order_value {
            return Ok(RiskAssessment {
                approved: false,
                reason: format!("Order value {} exceeds limit {}", 
                    order_value, self.risk_limits.max_order_value),
                risk_score: Decimal::new(8, 1), // 0.8
                validation_latency_ns: HardwareTimestamp::now().latency_ns(&start_time),
            });
        }
        
        // Simulate VaR calculation with SIMD optimization
        let portfolio_values = vec![order_value.to_f64().unwrap_or(0.0); 4]; // Simulate portfolio positions
        let portfolio_quantities = vec![order.quantity; 4];
        let simd_start = HardwareTimestamp::now();
        
        let simd_ops = if std::arch::is_x86_feature_detected!("avx2") {
            unsafe { Some(SimdPriceOps::new()) }
        } else {
            None
        };
        let _total_exposure = if let Some(ops) = simd_ops {
            let total_value = unsafe { ops.calculate_vwap(
                    &portfolio_values,
                    &vec![1.0; portfolio_values.len()]
                ) };
                total_value * portfolio_values.len() as f64
            } else {
                portfolio_values.iter().sum::<f64>()
            };
        
        let simd_latency = HardwareTimestamp::now().latency_ns(&simd_start);
        
        // VaR calculation should be sub-microsecond with SIMD
        if simd_latency > 1_000 { // 1μs
            eprintln!("WARNING: SIMD VaR calculation took {}ns, expected <1000ns", simd_latency);
        }
        
        let validation_latency = HardwareTimestamp::now().latency_ns(&start_time);
        
        Ok(RiskAssessment {
            approved: true,
            reason: "Order passes risk checks".to_string(),
            risk_score: Decimal::new(2, 1), // 0.2 (low risk)
            validation_latency_ns: validation_latency,
        })
    }
    
    /// Activate emergency stop mechanism
    pub fn trigger_emergency_stop(&self, reason: &str) -> TestResult<()> {
        self.emergency_stop_active.store(true, std::sync::atomic::Ordering::Release);
        eprintln!("EMERGENCY STOP ACTIVATED: {}", reason);
        Ok(())
    }
    
    /// Check if emergency stop should be triggered based on PnL
    pub async fn monitor_pnl(&self, current_pnl: Decimal) -> TestResult<bool> {
        let mut daily_pnl = self.daily_pnl.lock()
            .map_err(|e| format!("Failed to acquire PnL lock: {}", e))?;
        
        *daily_pnl += current_pnl;
        
        let loss_threshold = Decimal::from(-10000); // Emergency stop threshold
        
        if *daily_pnl < loss_threshold {
            self.trigger_emergency_stop(&format!(
                "Daily PnL {} exceeds loss threshold {}", 
                *daily_pnl, loss_threshold
            ))?;
            return Ok(true);
        }
        
        Ok(false)
    }
}

#[derive(Debug, Clone)]
pub struct RiskAssessment {
    pub approved: bool,
    pub reason: String,
    pub risk_score: Decimal,
    pub validation_latency_ns: u64,
}

#[derive(Debug, Clone)]
pub struct Order {
    pub symbol: String,
    pub side: OrderSide,
    pub quantity: Decimal,
    pub price: Decimal,
    pub order_type: OrderType,
    pub timestamp: HardwareTimestamp,
}

#[derive(Debug, Clone)]
// OrderSide now imported from canonical source
use trading_engine::types::prelude::OrderSide;

// OrderType now imported via trading_engine::types::prelude::* (line 21)

#[derive(Debug, Clone)]
pub struct Position {
    pub symbol: String,
    pub quantity: Decimal,
    pub average_price: Decimal,
    pub market_value: Decimal,
    pub unrealized_pnl: Decimal,
}

/// Mock broker client for testing
#[derive(Clone)]
pub struct MockBrokerClient {
    pub endpoint: String,
    pub connected: Arc<std::sync::atomic::AtomicBool>,
    pub order_queue: Arc<std::sync::Mutex<Vec<Order>>>,
    pub latency_stats: Arc<std::sync::Mutex<Vec<u64>>>,
}

impl MockBrokerClient {
    pub fn new(endpoint: String) -> Self {
        Self {
            endpoint,
            connected: Arc::new(std::sync::atomic::AtomicBool::new(false)),
            order_queue: Arc::new(std::sync::Mutex::new(Vec::new())),
            latency_stats: Arc::new(std::sync::Mutex::new(Vec::new())),
        }
    }
    
    pub async fn connect(&self) -> TestResult<()> {
        // Simulate broker connection with realistic latency
        tokio::time::sleep(Duration::from_millis(100)).await;
        self.connected.store(true, std::sync::atomic::Ordering::Release);
        Ok(())
    }
    
    pub async fn submit_order(&self, order: Order) -> TestResult<OrderResponse> {
        let start_time = HardwareTimestamp::now();
        
        if !self.connected.load(std::sync::atomic::Ordering::Acquire) {
            return Err("Broker not connected".into());
        }
        
        // Simulate order processing latency
        tokio::time::sleep(Duration::from_micros(50)).await; // 50μs broker latency
        
        let latency = HardwareTimestamp::now().latency_ns(&start_time);
        
        // Record latency statistics
        if let Ok(mut stats) = self.latency_stats.lock() {
            stats.push(latency);
        }
        
        // Mock implementation - in real system would push to lock-free queue
        Ok(())
            .map_err(|e: &str| format!("Failed to queue order: {}", e))?;
        
        Ok(OrderResponse {
            order_id: format!("{}_{}", self.endpoint, order.symbol),
            status: OrderStatus::Submitted,
            fill_price: None,
            fill_quantity: None,
            execution_latency_ns: latency,
        })
    }
    
    pub fn get_average_latency(&self) -> TestResult<u64> {
        let stats = self.latency_stats.lock()
            .map_err(|e| format!("Failed to acquire latency stats: {}", e))?;
        
        if stats.is_empty() {
            return Ok(0);
        }
        
        let sum: u64 = stats.iter().sum();
        Ok(sum / stats.len() as u64)
    }
}

#[derive(Debug, Clone)]
pub struct OrderResponse {
    pub order_id: String,
    pub status: OrderStatus,
    pub fill_price: Option<Decimal>,
    pub fill_quantity: Option<Decimal>,
    pub execution_latency_ns: u64,
}

// OrderStatus now imported from canonical source
use trading_engine::types::prelude::OrderStatus;

// =============================================================================
// INTEGRATION TESTS
// =============================================================================

#[tokio::test]
async fn test_broker_risk_order_validation_integration() -> TestResult<()> {
    let config = BrokerRiskTestConfig::default();
    let risk_engine = MockRiskEngine::new(config.clone());
    let broker = MockBrokerClient::new("test_broker:8080".to_string());
    
    // Connect to broker
    broker.connect().await?;
    
    // Test 1: Valid order should pass risk checks and execute
    let valid_order = Order {
        symbol: "AAPL".to_string(),
        side: OrderSide::Buy,
        quantity: Decimal::new(100, 0), // 100 shares
        price: Decimal::new(150_00, 2), // $150.00
        order_type: OrderType::Market,
        timestamp: HardwareTimestamp::now(),
    };
    
    let risk_assessment = risk_engine.validate_order(&valid_order).await?;
    assert!(risk_assessment.approved, "Valid order should pass risk checks");
    assert!(risk_assessment.validation_latency_ns < 50_000, 
        "Risk validation should be <50μs, got {}ns", risk_assessment.validation_latency_ns);
    
    if risk_assessment.approved {
        let order_response = broker.submit_order(valid_order).await?;
        assert!(matches!(order_response.status, OrderStatus::Submitted));
        assert!(order_response.execution_latency_ns < 100_000, 
            "Broker execution should be <100μs, got {}ns", order_response.execution_latency_ns);
    }
    
    // Test 2: Order exceeding risk limits should be rejected
    let risky_order = Order {
        symbol: "TSLA".to_string(),
        side: OrderSide::Buy,
        quantity: Decimal::new(1000, 0), // 1000 shares
        price: Decimal::new(800_00, 2), // $800.00 (exceeds max order value)
        order_type: OrderType::Market,
        timestamp: HardwareTimestamp::now(),
    };
    
    let risk_assessment = risk_engine.validate_order(&risky_order).await?;
    assert!(!risk_assessment.approved, "Risky order should be rejected");
    assert!(risk_assessment.reason.contains("exceeds limit"));
    
    println!("✓ Broker-Risk order validation integration test passed");
    Ok(())
}

#[tokio::test]
async fn test_emergency_stop_integration() -> TestResult<()> {
    let config = BrokerRiskTestConfig::default();
    let risk_engine = MockRiskEngine::new(config.clone());
    let broker = MockBrokerClient::new("emergency_test:8080".to_string());
    
    broker.connect().await?;
    
    // Test 1: Trigger emergency stop via PnL monitoring
    let large_loss = Decimal::new(-10_00, 2); // -$10.00 (exceeds 5% threshold)
    let emergency_triggered = risk_engine.monitor_pnl(large_loss).await?;
    assert!(emergency_triggered, "Emergency stop should be triggered on large loss");
    
    // Test 2: All subsequent orders should be rejected
    let order_after_stop = Order {
        symbol: "SPY".to_string(),
        side: OrderSide::Buy,
        quantity: Decimal::new(10, 0),
        price: Decimal::new(400_00, 2),
        order_type: OrderType::Market,
        timestamp: HardwareTimestamp::now(),
    };
    
    let risk_assessment = risk_engine.validate_order(&order_after_stop).await?;
    assert!(!risk_assessment.approved, "Orders should be rejected after emergency stop");
    assert!(risk_assessment.reason.contains("Emergency stop active"));
    
    println!("✓ Emergency stop integration test passed");
    Ok(())
}

#[tokio::test]
async fn test_multi_broker_risk_coordination() -> TestResult<()> {
    let config = BrokerRiskTestConfig::default();
    let risk_engine = Arc::new(MockRiskEngine::new(config.clone()));
    
    // Create multiple broker connections
    let brokers: Vec<MockBrokerClient> = config.broker_endpoints.iter()
        .map(|endpoint| MockBrokerClient::new(endpoint.clone()))
        .collect();
    
    // Connect all brokers
    for broker in &brokers {
        broker.connect().await?;
    }
    
    // Test concurrent order processing across brokers
    let mut handles = Vec::new();
    
    for (i, broker) in brokers.iter().enumerate() {
        let risk_engine = risk_engine.clone();
        let broker = broker.clone();
        
        let handle = tokio::spawn(async move {
            let order = Order {
                symbol: format!("STOCK_{}", i),
                side: OrderSide::Buy,
                quantity: Decimal::new(50, 0),
                price: Decimal::new(100_00, 2),
                order_type: OrderType::Market,
                timestamp: HardwareTimestamp::now(),
            };
            
            let risk_assessment = risk_engine.validate_order(&order).await?;
            
            if risk_assessment.approved {
                let order_response = broker.submit_order(order).await?;
                Ok::<_, Box<dyn std::error::Error + Send + Sync>>(
                    (risk_assessment.validation_latency_ns, order_response.execution_latency_ns)
                )
            } else {
                Err(format!("Order rejected: {}", risk_assessment.reason).into())
            }
        });
        
        handles.push(handle);
    }
    
    // Wait for all concurrent operations
    let mut results = Vec::new();
    for handle in handles {
        results.push(handle.await);
    }
    
    let mut successful_operations = 0;
    let mut total_risk_latency = 0u64;
    let mut total_execution_latency = 0u64;
    
    for result in results {
        match result {
            Ok(Ok((risk_latency, execution_latency))) => {
                successful_operations += 1;
                total_risk_latency += risk_latency;
                total_execution_latency += execution_latency;
            }
            Ok(Err(e)) => eprintln!("Order processing failed: {}", e),
            Err(e) => eprintln!("Task join failed: {}", e),
        }
    }
    
    assert!(successful_operations > 0, "At least one operation should succeed");
    
    if successful_operations > 0 {
        let avg_risk_latency = total_risk_latency / successful_operations;
        let avg_execution_latency = total_execution_latency / successful_operations;
        
        assert!(avg_risk_latency < 50_000, 
            "Average risk validation latency should be <50μs, got {}ns", avg_risk_latency);
        assert!(avg_execution_latency < 100_000, 
            "Average execution latency should be <100μs, got {}ns", avg_execution_latency);
    }
    
    println!("✓ Multi-broker risk coordination test passed ({} operations)", successful_operations);
    Ok(())
}

#[tokio::test]
async fn test_real_time_position_monitoring() -> TestResult<()> {
    let config = BrokerRiskTestConfig::default();
    let risk_engine = MockRiskEngine::new(config.clone());
    let broker = MockBrokerClient::new("position_monitor:8080".to_string());
    
    broker.connect().await?;
    
    // Simulate building up positions through multiple trades
    let symbols = vec!["AAPL", "GOOGL", "MSFT", "TSLA"];
    let mut total_exposure = Decimal::ZERO;
    
    for (i, &symbol) in symbols.iter().enumerate() {
        let order = Order {
            symbol: symbol.to_string(),
            side: OrderSide::Buy,
            quantity: Decimal::new((i + 1) as i64 * 10, 0), // Increasing position sizes
            price: Decimal::new(200_00 + (i as i64 * 50_00), 2), // Different prices
            order_type: OrderType::Market,
            timestamp: HardwareTimestamp::now(),
        };
        
        let order_value = order.price * order.quantity;
        total_exposure += order_value;
        
        // Risk assessment should consider cumulative exposure
        let risk_assessment = risk_engine.validate_order(&order).await?;
        
        if total_exposure <= config.max_position_size {
            assert!(risk_assessment.approved, 
                "Order should be approved when total exposure {} <= limit {}", 
                total_exposure, config.max_position_size);
                
            if risk_assessment.approved {
                let _order_response = broker.submit_order(order).await?;
            }
        } else {
            // Large positions should trigger additional risk checks
            println!("Large position detected: {} (limit: {})", total_exposure, config.max_position_size);
        }
        
        // Simulate market movement and PnL calculation
        let simulated_pnl = Decimal::new(-(i as i64 * 10), 2); // Gradual loss
        let emergency_triggered = risk_engine.monitor_pnl(simulated_pnl).await?;
        
        if emergency_triggered {
            println!("Emergency stop triggered after {} positions", i + 1);
            break;
        }
    }
    
    println!("✓ Real-time position monitoring test passed");
    Ok(())
}

#[tokio::test]
async fn test_latency_under_stress() -> TestResult<()> {
    let config = BrokerRiskTestConfig::default();
    let risk_engine = Arc::new(MockRiskEngine::new(config.clone()));
    let broker = Arc::new(MockBrokerClient::new("stress_test:8080".to_string()));
    
    broker.connect().await?;
    
    // Generate high-frequency order flow
    let num_orders = 1000;
    let mut handles = Vec::new();
    let start_time = HardwareTimestamp::now();
    
    for i in 0..num_orders {
        let risk_engine = risk_engine.clone();
        let broker = broker.clone();
        
        let handle = tokio::spawn(async move {
            let order = Order {
                symbol: format!("STRESS_{}", i % 10), // 10 different symbols
                side: if i % 2 == 0 { OrderSide::Buy } else { OrderSide::Sell },
                quantity: Decimal::new(10 + (i % 50) as i64, 0),
                price: Decimal::new(100_00 + (i % 100) as i64, 2),
                order_type: OrderType::Market,
                timestamp: HardwareTimestamp::now(),
            };
            
            let risk_start = HardwareTimestamp::now();
            let risk_assessment = risk_engine.validate_order(&order).await?;
            let risk_latency = HardwareTimestamp::now().latency_ns(&risk_start);
            
            if risk_assessment.approved {
                let exec_start = HardwareTimestamp::now();
                let order_response = broker.submit_order(order).await?;
                let exec_latency = HardwareTimestamp::now().latency_ns(&exec_start);
                
                Ok::<_, Box<dyn std::error::Error + Send + Sync>>((risk_latency, exec_latency))
            } else {
                Ok((risk_latency, 0u64)) // Risk rejection is also a valid outcome
            }
        });
        
        handles.push(handle);
    }
    
    // Process all orders
    let mut results = Vec::new();
    for handle in handles {
        results.push(handle.await);
    }
    let total_time = HardwareTimestamp::now().latency_ns(&start_time);
    
    let mut successful_orders = 0;
    let mut risk_latencies = Vec::new();
    let mut exec_latencies = Vec::new();
    
    for result in results {
        match result {
            Ok(Ok((risk_latency, exec_latency))) => {
                successful_orders += 1;
                risk_latencies.push(risk_latency);
                if exec_latency > 0 {
                    exec_latencies.push(exec_latency);
                }
            }
            Ok(Err(e)) => eprintln!("Order failed: {}", e),
            Err(e) => eprintln!("Task failed: {}", e),
        }
    }
    
    // Calculate statistics
    let throughput = (successful_orders as f64 / (total_time as f64 / 1_000_000_000.0)) as u64;
    
    risk_latencies.sort_unstable();
    exec_latencies.sort_unstable();
    
    let p95_risk_latency = risk_latencies.get(risk_latencies.len() * 95 / 100).copied().unwrap_or(0);
    let p95_exec_latency = exec_latencies.get(exec_latencies.len() * 95 / 100).copied().unwrap_or(0);
    
    // Validate HFT performance requirements
    assert!(p95_risk_latency < 50_000, 
        "P95 risk validation latency should be <50μs, got {}ns", p95_risk_latency);
    assert!(p95_exec_latency < 100_000, 
        "P95 execution latency should be <100μs, got {}ns", p95_exec_latency);
    assert!(throughput > 1_000, 
        "Throughput should be >1000 orders/sec, got {} orders/sec", throughput);
    
    println!("✓ Stress test passed: {} orders/sec, P95 risk: {}ns, P95 exec: {}ns", 
        throughput, p95_risk_latency, p95_exec_latency);
    
    Ok(())
}

// =============================================================================
// INTEGRATION TEST RUNNER
// =============================================================================

#[tokio::test]
async fn run_all_broker_risk_integration_tests() -> TestResult<()> {
    println!("=== BROKER-RISK INTEGRATION TEST SUITE ===");
    
    let test_timeout = Duration::from_secs(60);
    
    // Run all integration tests with timeout protection
    let _result = timeout(test_timeout, async { test_broker_risk_order_validation_integration() }).await??;
    let _result = timeout(test_timeout, async { test_emergency_stop_integration() }).await??;
    let _result = timeout(test_timeout, async { test_multi_broker_risk_coordination() }).await??;
    let _result = timeout(test_timeout, async { test_real_time_position_monitoring() }).await??;
    let _result = timeout(test_timeout, async { test_latency_under_stress() }).await??;
    
    println!("=== ALL BROKER-RISK INTEGRATION TESTS PASSED ===");
    println!("✓ Order validation and risk assessment integration");
    println!("✓ Emergency stop mechanisms");
    println!("✓ Multi-broker coordination");
    println!("✓ Real-time position monitoring");
    println!("✓ High-frequency stress testing");
    println!("✓ Sub-50μs risk validation latency");
    println!("✓ Sub-100μs broker execution latency");
    println!("✓ >1000 orders/sec throughput");
    
    Ok(())
}