//! Trading service gRPC implementation with full business logic

use crate::streaming::create_monitored_channel;
use common::OrderSide;
use num_traits::ToPrimitive;
use std::pin::Pin;
use std::sync::Arc;
use tokio::sync::mpsc;
use tokio_stream::{wrappers::ReceiverStream, Stream};
use tonic::{Request, Response, Result as TonicResult, Status};
use tracing::{debug, error, info, instrument, warn};

use crate::error::TradingServiceResult;
use crate::latency_recorder::{time_async, LatencyCategory};
use crate::proto::trading::{
    trading_service_server, CancelOrderRequest, CancelOrderResponse, ExecutionEvent,
    GetExecutionHistoryRequest, GetExecutionHistoryResponse, GetOrderBookRequest,
    GetOrderBookResponse, GetOrderStatusRequest, GetOrderStatusResponse,
    GetPortfolioSummaryRequest, GetPortfolioSummaryResponse, GetPositionsRequest,
    GetPositionsResponse, GetRegimeStateRequest, GetRegimeStateResponse,
    GetRegimeTransitionsRequest, GetRegimeTransitionsResponse, MarketDataEvent, Order, OrderBook,
    OrderBookLevel, OrderEvent, OrderEventType, OrderStatus, Position, PositionEvent,
    RegimeTransition, StreamExecutionsRequest, StreamMarketDataRequest, StreamOrderBookRequest,
    StreamOrdersRequest, StreamPortfolioSummaryRequest, StreamPositionsRequest,
    SubmitOrderRequest, SubmitOrderResponse,
};
use crate::state::TradingServiceState;

/// Trading service implementation with complete business logic
#[derive(Debug)]
pub struct TradingServiceImpl {
    state: Arc<TradingServiceState>,
}

impl TradingServiceImpl {
    /// Create new trading service implementation
    pub fn new(state: Arc<TradingServiceState>) -> Self {
        Self { state }
    }
}

#[tonic::async_trait]
impl trading_service_server::TradingService for TradingServiceImpl {
    #[instrument(skip_all)]
    async fn submit_order(
        &self,
        request: Request<SubmitOrderRequest>,
    ) -> TonicResult<Response<SubmitOrderResponse>> {
        let req = request.into_inner();
        info!("Submit order request for symbol: {}", req.symbol);

        // KILL SWITCH CHECK - Must be first for regulatory compliance
        if let Some(ref kill_switch) = self.state.kill_switch_system {
            if let Err(e) = kill_switch.check_trading_allowed(&req.symbol, Some(&req.account_id)) {
                warn!("Order rejected by kill switch: {}", e);
                return Err(Status::failed_precondition(format!(
                    "Trading blocked by kill switch: {}",
                    e
                )));
            }
        }

        // Validate order request
        if req.symbol.is_empty() {
            return Err(Status::invalid_argument("Symbol cannot be empty"));
        }

        // Validate symbol format (uppercase letters, digits, '/', or '-')
        if !req
            .symbol
            .chars()
            .all(|c| c.is_ascii_uppercase() || c == '/' || c == '-' || c.is_ascii_digit())
        {
            return Err(Status::invalid_argument(format!(
                "Invalid symbol format: {} (must be uppercase letters, digits, '/', or '-')",
                req.symbol
            )));
        }
        if req.symbol.is_empty() || req.symbol.len() > 10 {
            return Err(Status::invalid_argument(format!(
                "Invalid symbol length: {} (must be 1-10 characters)",
                req.symbol
            )));
        }

        if req.quantity <= 0.0 {
            return Err(Status::invalid_argument("Quantity must be positive"));
        }
        // Risk validation with timing
        let risk_result = time_async(LatencyCategory::RiskValidation, async {
            let risk_engine = self.state.risk_engine.read().await;
            let result = self.validate_order_risk(&req).await;
            drop(risk_engine);
            result
        })
        .await;

        if let Err(e) = risk_result {
            warn!("Order rejected due to risk violation: {}", e);
            return Err(Status::failed_precondition(format!(
                "Risk violation: {}",
                e
            )));
        }

        // Submit order through order manager with timing
        // Submit order through trading repository with timing
        let order_result = time_async(LatencyCategory::OrderProcessing, async {
            // Convert request to repository format
            let trading_order = crate::repositories::TradingOrder {
                id: uuid::Uuid::new_v4().to_string(),
                account_id: req.account_id.clone(),
                symbol: req.symbol.clone(),
                side: match req.side {
                    1 => common::types::OrderSide::Buy,
                    2 => common::types::OrderSide::Sell,
                    _ => common::types::OrderSide::Buy,
                },
                order_type: match req.order_type {
                    1 => common::types::OrderType::Market,
                    2 => common::types::OrderType::Limit,
                    _ => common::types::OrderType::Market,
                },
                quantity: req.quantity,
                filled_quantity: 0.0,
                price: req.price.unwrap_or(0.0),
                stop_price: req.stop_price,
                status: common::types::OrderStatus::Pending,
                timestamp: chrono::Utc::now().timestamp(),
            };

            self.state
                .trading_repository
                .store_order(&trading_order)
                .await
        })
        .await;

        match order_result {
            Ok(order_id) => {
                info!("Order submitted successfully: {}", order_id);

                // Persist order event to trading_events table for compliance
                let event_data = crate::event_persistence::TradingEventData {
                    event_type: "order_submitted".to_string(),
                    symbol: req.symbol.clone(),
                    event_data: serde_json::json!({
                        "order_id": &order_id,
                        "account_id": &req.account_id,
                        "quantity": req.quantity,
                        "price": req.price,
                        "side": req.side,
                        "order_type": req.order_type,
                    }),
                    metadata: Some(serde_json::json!({
                        "client_id": req.account_id.clone(),
                    })),
                };

                // Fire and forget - log error but don't fail order submission
                let _ = self.state.event_persistence.write_event(event_data).await;

                // Publish order event
                self.publish_order_event(&order_id, OrderEventType::Created)
                    .await;

                Ok(Response::new(SubmitOrderResponse {
                    order_id,
                    status: OrderStatus::Submitted as i32,
                    message: "Order submitted successfully".to_string(),
                    timestamp: chrono::Utc::now().timestamp(),
                }))
            },
            Err(e) => {
                error!("Failed to submit order: {}", e);
                Err(Status::internal(format!("Failed to submit order: {}", e)))
            },
        }
    }

    #[instrument(skip_all)]
    async fn cancel_order(
        &self,
        request: Request<CancelOrderRequest>,
    ) -> TonicResult<Response<CancelOrderResponse>> {
        let req = request.into_inner();
        info!("Cancel order request for order_id: {}", req.order_id);

        // Parse UUID from string
        let order_id = uuid::Uuid::parse_str(&req.order_id)
            .map_err(|e| Status::invalid_argument(format!("Invalid order ID: {}", e)))?
            .to_string();

        match self
            .state
            .trading_repository
            .update_order_status(&order_id, common::types::OrderStatus::Cancelled)
            .await
        {
            Ok(()) => {
                info!("Order cancelled successfully: {}", order_id);

                // Persist cancellation event to trading_events table for compliance
                let event_data = crate::event_persistence::TradingEventData {
                    event_type: "order_cancelled".to_string(),
                    symbol: "UNKNOWN".to_string(), // Symbol not available in cancel request
                    event_data: serde_json::json!({
                        "order_id": &order_id,
                        "cancelled_by": &req.account_id,
                    }),
                    metadata: Some(serde_json::json!({
                        "client_id": req.account_id.clone(),
                    })),
                };

                // Fire and forget - log error but don't fail cancellation
                if let Err(e) = self.state.event_persistence.write_event(event_data).await {
                    warn!(
                        "Failed to persist cancellation event for order {}: {}",
                        order_id, e
                    );
                }

                // Publish order cancellation event
                self.publish_order_event(&order_id, OrderEventType::Cancelled)
                    .await;

                Ok(Response::new(CancelOrderResponse {
                    success: true,
                    message: "Order cancelled successfully".to_string(),
                    timestamp: chrono::Utc::now().timestamp(),
                }))
            },
            Err(e) => {
                error!("Failed to cancel order: {}", e);
                Err(Status::internal(format!("Failed to cancel order: {}", e)))
            },
        }
    }

    #[instrument(skip_all)]
    async fn get_order_status(
        &self,
        request: Request<GetOrderStatusRequest>,
    ) -> TonicResult<Response<GetOrderStatusResponse>> {
        let req = request.into_inner();
        debug!("Get order status for order_id: {}", req.order_id);

        // Parse UUID from string
        let order_id = uuid::Uuid::parse_str(&req.order_id)
            .map_err(|e| Status::invalid_argument(format!("Invalid order ID: {}", e)))?
            .to_string();

        match self.state.trading_repository.get_order(&order_id).await {
            Ok(Some(trading_order)) => {
                // Convert repository order to proto order
                let proto_order = Order {
                    order_id: trading_order.id,
                    symbol: trading_order.symbol,
                    side: trading_order.side as i32,
                    quantity: trading_order.quantity,
                    filled_quantity: trading_order.filled_quantity,
                    order_type: trading_order.order_type as i32,
                    price: Some(trading_order.price),
                    stop_price: trading_order.stop_price,
                    status: trading_order.status as i32,
                    created_at: trading_order.timestamp,
                    updated_at: Some(trading_order.timestamp),
                    account_id: trading_order.account_id,
                    metadata: std::collections::HashMap::new(),
                };
                Ok(Response::new(GetOrderStatusResponse {
                    order: Some(proto_order),
                }))
            },
            Ok(None) => Err(Status::not_found(format!("Order {} not found", order_id))),
            Err(e) => {
                error!("Failed to get order status: {}", e);
                Err(Status::internal(format!(
                    "Failed to get order status: {}",
                    e
                )))
            },
        }
    }

    // Order Streaming Implementation
    type StreamOrdersStream = Pin<Box<dyn Stream<Item = Result<OrderEvent, Status>> + Send>>;

    #[instrument(skip_all)]
    async fn stream_orders(
        &self,
        request: Request<StreamOrdersRequest>,
    ) -> TonicResult<Response<Self::StreamOrdersStream>> {
        let req = request.into_inner();
        info!("Stream orders request for account: {:?}", req.account_id);

        // Wave 67 Agent 6: Monitored channel with backpressure monitoring
        use crate::streaming::StreamType;
        let buffer_size = StreamType::MediumFrequency.buffer_size();
        let (_tx, rx, _monitor, _metrics) = create_monitored_channel(buffer_size, "orders");

        // Subscribe to order events and forward to stream
        let event_publisher = Arc::clone(&self.state.event_publisher);
        let account_id_filter = req.account_id;

        tokio::spawn(async move {
            // Subscribe to trading events and filter for order events
            let mut subscription = match event_publisher.subscribe() {
                Ok(sub) => sub,
                Err(e) => {
                    error!("Failed to subscribe to events: {}", e);
                    return;
                },
            };

            while let Ok(event) = subscription.recv().await {
                // Filter for order events matching account_id if specified
                if event.is_order_event() {
                    if let Some(ref account_filter) = account_id_filter {
                        if !event.matches_account(account_filter) {
                            continue;
                        }
                    }

                    // Convert TradingEvent to OrderEvent proto and send
                    if let Err(e) = _tx
                        .send_monitored(Ok(Self::convert_to_order_event(&event)))
                        .await
                    {
                        warn!("Order stream send failed: {}", e);
                        break;
                    }
                }
            }
        });

        let stream = ReceiverStream::new(rx);
        Ok(Response::new(Box::pin(stream)))
    }

    #[instrument(skip_all)]
    async fn get_positions(
        &self,
        request: Request<GetPositionsRequest>,
    ) -> TonicResult<Response<GetPositionsResponse>> {
        let req = request.into_inner();
        debug!("Get positions request for account: {:?}", req.account_id);

        match self
            .state
            .trading_repository
            .get_positions(req.account_id.as_deref(), req.symbol.as_deref())
            .await
        {
            Ok(repo_positions) => {
                // Convert repository positions to proto positions
                let positions = repo_positions
                    .into_iter()
                    .map(|pos| Position {
                        symbol: pos.symbol,
                        quantity: pos.quantity,
                        average_price: pos.average_price,
                        market_value: pos.market_value,
                        unrealized_pnl: pos.unrealized_pnl,
                        // BLOCKER: TradingPosition lacks a realized_pnl field.
                        // The TradingRepository trait exposes get_realized_pnl(account, symbol)
                        // but it returns a single f64 per account/symbol pair, requiring an
                        // async call per position which cannot run inside Iterator::map.
                        // Fix: either add realized_pnl to TradingPosition so the repository
                        // populates it in one query, or pre-fetch a HashMap<symbol, f64> before
                        // the map closure and look up each symbol.
                        realized_pnl: 0.0,
                        account_id: pos.account_id,
                        updated_at: pos.timestamp,
                    })
                    .collect();
                Ok(Response::new(GetPositionsResponse { positions }))
            },
            Err(e) => {
                error!("Failed to get positions: {}", e);
                Err(Status::internal(format!("Failed to get positions: {}", e)))
            },
        }
    }

    // Position Streaming Implementation
    type StreamPositionsStream = Pin<Box<dyn Stream<Item = Result<PositionEvent, Status>> + Send>>;

    #[instrument(skip_all)]
    async fn stream_positions(
        &self,
        request: Request<StreamPositionsRequest>,
    ) -> TonicResult<Response<Self::StreamPositionsStream>> {
        let req = request.into_inner();
        info!("Stream positions request for account: {:?}", req.account_id);

        // Wave 67 Agent 3: Use StreamType-specific buffer size for positions (medium frequency)
        use crate::streaming::StreamType;
        let buffer_size = StreamType::MediumFrequency.buffer_size();
        let (_tx, rx) = mpsc::channel(buffer_size);

        // Subscribe to position events
        let event_publisher = Arc::clone(&self.state.event_publisher);
        let account_id_filter = req.account_id;

        tokio::spawn(async move {
            let mut subscription = match event_publisher.subscribe() {
                Ok(sub) => sub,
                Err(e) => {
                    error!("Failed to subscribe to position events: {}", e);
                    return;
                },
            };

            while let Ok(event) = subscription.recv().await {
                if event.is_position_event()
                    && event.matches_account(account_id_filter.as_deref().unwrap_or(""))
                {
                    let _ = _tx.send(Ok(Self::convert_to_position_event(&event))).await;
                }
            }
        });

        let stream = ReceiverStream::new(rx);
        Ok(Response::new(Box::pin(stream)))
    }

    #[instrument(skip_all)]
    async fn get_portfolio_summary(
        &self,
        request: Request<GetPortfolioSummaryRequest>,
    ) -> TonicResult<Response<GetPortfolioSummaryResponse>> {
        let req = request.into_inner();
        debug!("Get portfolio summary for account: {}", req.account_id);

        match self
            .state
            .trading_repository
            .get_portfolio_summary(&req.account_id)
            .await
        {
            Ok(repo_summary) => {
                // Convert repository summary to proto summary
                let positions = self
                    .state
                    .trading_repository
                    .get_positions(Some(&req.account_id), None)
                    .await
                    .unwrap_or_default()
                    .into_iter()
                    .map(|pos| Position {
                        symbol: pos.symbol,
                        quantity: pos.quantity,
                        average_price: pos.average_price,
                        market_value: pos.market_value,
                        unrealized_pnl: pos.unrealized_pnl,
                        realized_pnl: 0.0, // Same blocker as get_positions — see comment there
                        account_id: pos.account_id.clone(),
                        updated_at: pos.timestamp,
                    })
                    .collect();

                let summary = GetPortfolioSummaryResponse {
                    total_value: repo_summary.total_value,
                    unrealized_pnl: repo_summary.unrealized_pnl,
                    realized_pnl: repo_summary.realized_pnl,
                    day_pnl: self
                        .state
                        .trading_repository
                        .get_day_pnl(&req.account_id)
                        .await
                        .unwrap_or(0.0),
                    buying_power: repo_summary.cash_balance, // Use cash balance as buying power
                    margin_used: self
                        .state
                        .risk_repository
                        .calculate_margin_used(&req.account_id)
                        .await
                        .unwrap_or(0.0),
                    positions,
                };
                Ok(Response::new(summary))
            },
            Err(e) => {
                error!("Failed to get portfolio summary: {}", e);
                Err(Status::internal(format!(
                    "Failed to get portfolio summary: {}",
                    e
                )))
            },
        }
    }

    // Market Data Streaming Implementation
    type StreamMarketDataStream =
        Pin<Box<dyn Stream<Item = Result<MarketDataEvent, Status>> + Send>>;

    #[instrument(skip_all)]
    async fn stream_market_data(
        &self,
        request: Request<StreamMarketDataRequest>,
    ) -> TonicResult<Response<Self::StreamMarketDataStream>> {
        let req = request.into_inner();
        info!("Stream market data for symbols: {:?}", req.symbols);

        // Wave 67 Agent 3: Use HIGH FREQUENCY buffer for market data (critical for HFT)
        use crate::streaming::StreamType;
        let buffer_size = StreamType::HighFrequency.buffer_size(); // 100K buffer
        let (tx, rx) = mpsc::channel(buffer_size);

        // Subscribe to market data events
        let event_publisher = Arc::clone(&self.state.event_publisher);
        let _symbols_filter = req.symbols;

        tokio::spawn(async move {
            let mut subscription = match event_publisher.subscribe() {
                Ok(sub) => sub,
                Err(e) => {
                    error!("Failed to subscribe to market data events: {}", e);
                    return;
                },
            };

            while let Ok(event) = subscription.recv().await {
                if event.event_type.is_market_data_event() {
                    // Send market data event (filtering by symbols can be added if needed)
                    let _ = tx
                        .send(Ok(Self::convert_to_market_data_event(&event)))
                        .await;
                }
            }
        });

        let stream = ReceiverStream::new(rx);
        Ok(Response::new(Box::pin(stream)))
    }

    #[instrument(skip_all)]
    async fn get_order_book(
        &self,
        request: Request<GetOrderBookRequest>,
    ) -> TonicResult<Response<GetOrderBookResponse>> {
        let req = request.into_inner();
        debug!("Get order book for symbol: {}", req.symbol);

        match self
            .state
            .market_data_repository
            .get_order_book(&req.symbol, req.depth.unwrap_or(10))
            .await
        {
            Ok(repo_order_book) => {
                // Convert repository order book to proto order book
                // Fetch order counts for all bid levels
                let mut bid_levels = Vec::new();
                for level in repo_order_book.bids {
                    let price_f64 = level.price.to_f64().unwrap_or(0.0);
                    let order_count = self
                        .state
                        .market_data_repository
                        .get_order_book_level_count(&req.symbol, price_f64, OrderSide::Buy)
                        .await
                        .unwrap_or(1);
                    bid_levels.push(OrderBookLevel {
                        price: price_f64,
                        quantity: level.size.to_f64().unwrap_or(0.0),
                        order_count,
                    });
                }

                // Fetch order counts for all ask levels
                let mut ask_levels = Vec::new();
                for level in repo_order_book.asks {
                    let price_f64 = level.price.to_f64().unwrap_or(0.0);
                    let order_count = self
                        .state
                        .market_data_repository
                        .get_order_book_level_count(&req.symbol, price_f64, OrderSide::Sell)
                        .await
                        .unwrap_or(1);
                    ask_levels.push(OrderBookLevel {
                        price: price_f64,
                        quantity: level.size.to_f64().unwrap_or(0.0),
                        order_count,
                    });
                }

                let order_book = OrderBook {
                    symbol: repo_order_book.symbol,
                    bids: bid_levels,
                    asks: ask_levels,
                    timestamp: repo_order_book.timestamp,
                };
                Ok(Response::new(GetOrderBookResponse {
                    order_book: Some(order_book),
                }))
            },
            Err(e) => {
                error!("Failed to get order book: {}", e);
                Err(Status::internal(format!("Failed to get order book: {}", e)))
            },
        }
    }

    // Execution Streaming Implementation
    type StreamExecutionsStream =
        Pin<Box<dyn Stream<Item = Result<ExecutionEvent, Status>> + Send>>;

    #[instrument(skip_all)]
    async fn stream_executions(
        &self,
        request: Request<StreamExecutionsRequest>,
    ) -> TonicResult<Response<Self::StreamExecutionsStream>> {
        let req = request.into_inner();
        info!("Stream executions for account: {:?}", req.account_id);

        // Wave 67 Agent 3: Use StreamType-specific buffer size for executions (medium frequency)
        use crate::streaming::StreamType;
        let buffer_size = StreamType::MediumFrequency.buffer_size();
        let (_tx, rx) = mpsc::channel(buffer_size);

        // Subscribe to execution events
        let event_publisher = Arc::clone(&self.state.event_publisher);
        let account_id_filter = req.account_id;

        tokio::spawn(async move {
            let mut subscription = match event_publisher.subscribe() {
                Ok(sub) => sub,
                Err(e) => {
                    error!("Failed to subscribe to execution events: {}", e);
                    return;
                },
            };

            while let Ok(event) = subscription.recv().await {
                if event.is_execution_event()
                    && event.matches_account(account_id_filter.as_deref().unwrap_or(""))
                {
                    let _ = _tx.send(Ok(Self::convert_to_execution_event(&event))).await;
                }
            }
        });

        let stream = ReceiverStream::new(rx);
        Ok(Response::new(Box::pin(stream)))
    }

    #[instrument(skip_all)]
    async fn get_execution_history(
        &self,
        request: Request<GetExecutionHistoryRequest>,
    ) -> TonicResult<Response<GetExecutionHistoryResponse>> {
        let req = request.into_inner();
        debug!("Get execution history for account: {:?}", req.account_id);

        match self
            .state
            .trading_repository
            .get_execution_history(&req)
            .await
        {
            Ok(repo_executions) => {
                // Convert repository executions to proto executions
                let executions = repo_executions
                    .into_iter()
                    .map(|exec| crate::proto::trading::Execution {
                        execution_id: exec.id,
                        order_id: exec.order_id,
                        symbol: exec.symbol,
                        side: exec.side as i32,
                        quantity: exec.quantity,
                        price: exec.price,
                        timestamp: exec.timestamp,
                        account_id: exec.account_id,
                        metadata: std::collections::HashMap::new(),
                    })
                    .collect();
                Ok(Response::new(GetExecutionHistoryResponse { executions }))
            },
            Err(e) => {
                error!("Failed to get execution history: {}", e);
                Err(Status::internal(format!(
                    "Failed to get execution history: {}",
                    e
                )))
            },
        }
    }

    #[instrument(skip_all)]
    async fn submit_ml_order(
        &self,
        request: Request<crate::proto::trading::MlOrderRequest>,
    ) -> TonicResult<Response<crate::proto::trading::MlOrderResponse>> {
        let req = request.into_inner();
        info!("Submit ML order for symbol: {}", req.symbol);

        // Validate feature vector (require 26 features: 5 OHLCV + 21 technical indicators)
        if req.features.len() != 26 {
            return Err(Status::invalid_argument(format!(
                            "Invalid feature count: {} (expected 26 features: 5 OHLCV + 21 technical indicators)",
                            req.features.len()
                        )));
        }

        // Use ensemble coordinator if available
        if let Some(ref ensemble_coordinator) = self.state.ensemble_coordinator {
            // Generate ensemble prediction (features are generated internally for now)
            // ROADMAP: Pass req.features to EnsembleCoordinator once feature pipeline
            // is integrated. Currently, generate_and_save_prediction() calls
            // extract_features_for_symbol() internally which builds a tick-based
            // approximation. Once the OHLCV bar pipeline is live (see state.rs
            // extract_features_for_symbol roadmap), the EnsembleCoordinator API should
            // accept an optional Features parameter to avoid redundant extraction.
            match ensemble_coordinator
                .generate_and_save_prediction(&req.symbol)
                .await
            {
                Ok(prediction) => {
                    let confidence = prediction.ensemble_confidence;
                    let action = prediction.ensemble_action.clone();
                    let prediction_id = prediction.id.to_string();

                    // Check confidence threshold (60%)
                    if confidence < 0.60 {
                        return Ok(Response::new(crate::proto::trading::MlOrderResponse {
                            order_id: String::new(),
                            prediction_id,
                            action: "HOLD".to_string(),
                            confidence,
                            message: format!(
                                "Confidence {:.2}% below 60% threshold - no order executed",
                                confidence * 100.0
                            ),
                            executed: false,
                        }));
                    }

                    // Execute order if BUY or SELL
                    if action == "BUY" || action == "SELL" {
                        // Create order through paper trading executor or direct order submission
                        let side = if action == "BUY" { 1 } else { 2 };
                        let submit_req = SubmitOrderRequest {
                            symbol: req.symbol.clone(),
                            side,
                            quantity: 1.0, // 1 contract for ML orders
                            order_type: 1, // Market order
                            price: None,
                            stop_price: None,
                            account_id: req.account_id.clone(),
                            metadata: std::collections::HashMap::from([
                                ("ml_prediction_id".to_string(), prediction_id.clone()),
                                ("confidence".to_string(), confidence.to_string()),
                            ]),
                        };

                        match self.submit_order(Request::new(submit_req)).await {
                            Ok(order_response) => {
                                let order = order_response.into_inner();
                                Ok(Response::new(crate::proto::trading::MlOrderResponse {
                                    order_id: order.order_id,
                                    prediction_id,
                                    action,
                                    confidence,
                                    message: format!(
                                        "ML order executed with {:.2}% confidence",
                                        confidence * 100.0
                                    ),
                                    executed: true,
                                }))
                            },
                            Err(e) => {
                                warn!("Failed to execute ML order: {}", e);
                                Ok(Response::new(crate::proto::trading::MlOrderResponse {
                                    order_id: String::new(),
                                    prediction_id,
                                    action,
                                    confidence,
                                    message: format!("Order submission failed: {}", e),
                                    executed: false,
                                }))
                            },
                        }
                    } else {
                        // HOLD action
                        Ok(Response::new(crate::proto::trading::MlOrderResponse {
                            order_id: String::new(),
                            prediction_id,
                            action: "HOLD".to_string(),
                            confidence,
                            message: "ML prediction: HOLD - no order executed".to_string(),
                            executed: false,
                        }))
                    }
                },
                Err(e) => {
                    error!("Failed to generate ML prediction: {}", e);
                    Err(Status::internal(format!(
                        "Failed to generate ML prediction: {}",
                        e
                    )))
                },
            }
        } else {
            Err(Status::unavailable("Ensemble coordinator not available"))
        }
    }

    #[instrument(skip_all)]
    async fn get_ml_predictions(
        &self,
        request: Request<crate::proto::trading::MlPredictionsRequest>,
    ) -> TonicResult<Response<crate::proto::trading::MlPredictionsResponse>> {
        let req = request.into_inner();
        debug!(
            "Get ML predictions for symbol: {}, model: {:?}",
            req.symbol, req.model_name
        );

        // Validate and clamp limit (default 100, max 1000 for safety)
        let limit = if req.limit > 0 && req.limit <= 1000 {
            req.limit
        } else if req.limit > 1000 {
            warn!("Request limit {} exceeds max 1000, clamping", req.limit);
            1000
        } else {
            100
        };

        // Build model name filter condition (supports filtering by specific model)
        let model_filter = req
            .model_name
            .as_ref()
            .and_then(|m| match m.to_uppercase().as_str() {
                "DQN" | "PPO" | "MAMBA2" | "TFT" => Some(m.to_uppercase()),
                _ => {
                    warn!("Invalid model name filter: {}, ignoring", m);
                    None
                },
            });

        // Query ensemble_predictions with LEFT JOIN to orders for actual outcomes
        let predictions = sqlx::query!(
                        r#"
                        SELECT
                            ep.id,
                            ep.symbol,
                            ep.ensemble_action,
                            ep.ensemble_signal,
                            ep.ensemble_confidence,
                            ep.prediction_timestamp,
                            ep.order_id,
                            ep.pnl as actual_pnl,
                            ep.executed_price,
                            ep.position_size,
                            ep.dqn_signal,
                            ep.dqn_confidence,
                            ep.dqn_vote,
                            ep.mamba2_signal,
                            ep.mamba2_confidence,
                            ep.mamba2_vote,
                            ep.ppo_signal,
                            ep.ppo_confidence,
                            ep.ppo_vote,
                            ep.tft_signal,
                            ep.tft_confidence,
                            ep.tft_vote,
                            o.status::text as "order_status!",
                            o.filled_quantity
                        FROM ensemble_predictions ep
                        LEFT JOIN orders o ON ep.order_id = o.id
                        WHERE ep.symbol = $1
                            AND ($2::text IS NULL OR ep.prediction_timestamp >= to_timestamp($2::bigint / 1000000000.0))
                            AND ($3::text IS NULL OR ep.prediction_timestamp <= to_timestamp($3::bigint / 1000000000.0))
                            AND (
                                $4::text IS NULL OR
                                ($4 = 'DQN' AND ep.dqn_vote IS NOT NULL) OR
                                ($4 = 'PPO' AND ep.ppo_vote IS NOT NULL) OR
                                ($4 = 'MAMBA2' AND ep.mamba2_vote IS NOT NULL) OR
                                ($4 = 'TFT' AND ep.tft_vote IS NOT NULL)
                            )
                        ORDER BY ep.prediction_timestamp DESC
                        LIMIT $5
                        "#,
                        req.symbol,
                        req.start_time.map(|t| t.to_string()),
                        req.end_time.map(|t| t.to_string()),
                        model_filter,
                        limit as i64,
                    )
                    .fetch_all(&self.state.db_pool)
                    .await
                    .map_err(|e| {
                        error!("Failed to query ML predictions: {}", e);
                        Status::internal(format!("Failed to query predictions: {}", e))
                    })?;

        debug!(
            "Retrieved {} ML predictions for symbol {}",
            predictions.len(),
            req.symbol
        );

        let proto_predictions: Vec<crate::proto::trading::MlPrediction> = predictions
            .into_iter()
            .map(|p| {
                use crate::proto::trading::{MlPrediction, ModelPrediction};

                // Calculate actual P&L in dollars (pnl is stored in cents)
                let actual_pnl = p.actual_pnl.map(|pnl_cents| pnl_cents as f64 / 100.0);

                // Build model predictions list (only include models with votes)
                let mut model_predictions = Vec::with_capacity(4);

                if let (Some(signal), Some(conf)) = (p.dqn_signal, p.dqn_confidence) {
                    model_predictions.push(ModelPrediction {
                        model_name: "DQN".to_string(),
                        signal,
                        confidence: conf,
                    });
                }

                if let (Some(signal), Some(conf)) = (p.mamba2_signal, p.mamba2_confidence) {
                    model_predictions.push(ModelPrediction {
                        model_name: "MAMBA2".to_string(),
                        signal,
                        confidence: conf,
                    });
                }

                if let (Some(signal), Some(conf)) = (p.ppo_signal, p.ppo_confidence) {
                    model_predictions.push(ModelPrediction {
                        model_name: "PPO".to_string(),
                        signal,
                        confidence: conf,
                    });
                }

                if let (Some(signal), Some(conf)) = (p.tft_signal, p.tft_confidence) {
                    model_predictions.push(ModelPrediction {
                        model_name: "TFT".to_string(),
                        signal,
                        confidence: conf,
                    });
                }

                MlPrediction {
                    id: p.id.to_string(),
                    symbol: p.symbol,
                    ensemble_action: p.ensemble_action,
                    ensemble_signal: p.ensemble_signal,
                    ensemble_confidence: p.ensemble_confidence,
                    timestamp: p.prediction_timestamp.timestamp_nanos_opt().unwrap_or(0),
                    order_id: p.order_id.map(|id| id.to_string()),
                    actual_pnl,
                    model_predictions,
                }
            })
            .collect();

        info!(
            "Returning {} ML predictions for symbol {} (model filter: {:?})",
            proto_predictions.len(),
            req.symbol,
            model_filter
        );

        Ok(Response::new(
            crate::proto::trading::MlPredictionsResponse {
                predictions: proto_predictions,
            },
        ))
    }

    #[instrument(skip_all)]
    async fn get_ml_performance(
        &self,
        request: Request<crate::proto::trading::MlPerformanceRequest>,
    ) -> TonicResult<Response<crate::proto::trading::MlPerformanceResponse>> {
        let req = request.into_inner();
        debug!("Get ML performance metrics for model: {:?}", req.model_name);

        // Query ensemble_predictions table for real-time performance data
        // This provides per-model attribution from the production ensemble system
        let model_names = if let Some(ref name) = req.model_name {
            vec![name.clone()]
        } else {
            vec![
                "DQN".to_string(),
                "MAMBA2".to_string(),
                "PPO".to_string(),
                "TFT".to_string(),
            ]
        };

        let mut proto_models = Vec::new();

        for model_name in model_names {
            // Calculate metrics for each model from ensemble_predictions
            let metrics = self
                .calculate_model_performance_metrics(&model_name, req.start_time, req.end_time)
                .await?;
            proto_models.push(metrics);
        }

        Ok(Response::new(
            crate::proto::trading::MlPerformanceResponse {
                models: proto_models,
            },
        ))
    }

    #[instrument(skip_all)]
    async fn get_regime_state(
        &self,
        request: Request<GetRegimeStateRequest>,
    ) -> TonicResult<Response<GetRegimeStateResponse>> {
        let req = request.into_inner();
        debug!("Get regime state for symbol: {}", req.symbol);

        // Query database for regime state using the get_latest_regime stored function
        let record = sqlx::query!(
            r#"
                    SELECT
                        regime,
                        confidence,
                        event_timestamp,
                        cusum_s_plus,
                        cusum_s_minus,
                        adx,
                        stability
                    FROM get_latest_regime($1)
                    "#,
            req.symbol
        )
        .fetch_one(&self.state.db_pool)
        .await
        .map_err(|e| {
            error!("Failed to get regime state for {}: {}", req.symbol, e);
            Status::internal(format!("Database error: {}", e))
        })?;

        let response = GetRegimeStateResponse {
            symbol: req.symbol.clone(),
            current_regime: record.regime.unwrap_or_else(|| "Normal".to_string()),
            confidence: record.confidence.unwrap_or(0.0),
            cusum_s_plus: record.cusum_s_plus.unwrap_or(0.0),
            cusum_s_minus: record.cusum_s_minus.unwrap_or(0.0),
            adx: record.adx.unwrap_or(0.0),
            stability: record.stability.unwrap_or(0.0),
            entropy: 0.0, // Entropy not stored in DB yet - placeholder for Wave D Phase 4
            updated_at: record
                .event_timestamp
                .map(|ts| ts.timestamp_nanos_opt().unwrap_or(0))
                .unwrap_or(0),
        };

        Ok(Response::new(response))
    }

    #[instrument(skip_all)]
    async fn get_regime_transitions(
        &self,
        request: Request<GetRegimeTransitionsRequest>,
    ) -> TonicResult<Response<GetRegimeTransitionsResponse>> {
        let req = request.into_inner();
        let limit = if req.limit > 0 { req.limit } else { 100 };
        debug!(
            "Get regime transitions for symbol: {}, limit: {}",
            req.symbol, limit
        );

        // Query database for regime transitions
        let records = sqlx::query!(
            r#"
                    SELECT
                        from_regime,
                        to_regime,
                        event_timestamp,
                        duration_bars,
                        transition_probability
                    FROM regime_transitions
                    WHERE symbol = $1
                    ORDER BY event_timestamp DESC
                    LIMIT $2
                    "#,
            req.symbol,
            limit as i64
        )
        .fetch_all(&self.state.db_pool)
        .await
        .map_err(|e| {
            error!("Failed to get regime transitions for {}: {}", req.symbol, e);
            Status::internal(format!("Database error: {}", e))
        })?;

        let proto_transitions = records
            .into_iter()
            .map(|rec| RegimeTransition {
                from_regime: rec.from_regime,
                to_regime: rec.to_regime,
                duration_bars: rec.duration_bars.unwrap_or(0),
                transition_probability: rec.transition_probability.unwrap_or(0.0),
                timestamp: rec.event_timestamp.timestamp_nanos_opt().unwrap_or(0),
            })
            .collect();

        let response = GetRegimeTransitionsResponse {
            transitions: proto_transitions,
        };

        Ok(Response::new(response))
    }

    // Server-streaming: polls GetPortfolioSummary at configurable interval
    type StreamPortfolioSummaryStream =
        Pin<Box<dyn Stream<Item = Result<GetPortfolioSummaryResponse, Status>> + Send>>;

    #[instrument(skip_all)]
    async fn stream_portfolio_summary(
        &self,
        request: Request<StreamPortfolioSummaryRequest>,
    ) -> TonicResult<Response<Self::StreamPortfolioSummaryStream>> {
        let req = request.into_inner();
        let interval_secs = if req.interval_seconds > 0 {
            u64::from(req.interval_seconds.clamp(1, 60))
        } else {
            3
        };

        info!(
            interval_secs = interval_secs,
            account_id = %req.account_id,
            "StreamPortfolioSummary started"
        );

        let (tx, rx) = mpsc::channel(16);
        let state = Arc::clone(&self.state);
        let account_id = req.account_id;

        tokio::spawn(async move {
            let svc = TradingServiceImpl::new(state);
            let mut interval =
                tokio::time::interval(std::time::Duration::from_secs(interval_secs));

            loop {
                interval.tick().await;

                let summary_req = GetPortfolioSummaryRequest {
                    account_id: account_id.clone(),
                };

                match svc
                    .get_portfolio_summary(tonic::Request::new(summary_req))
                    .await
                {
                    Ok(resp) => {
                        if tx.send(Ok(resp.into_inner())).await.is_err() {
                            info!("StreamPortfolioSummary: client disconnected");
                            break;
                        }
                    },
                    Err(e) => {
                        warn!("StreamPortfolioSummary: poll failed: {}", e);
                        if tx.send(Err(e)).await.is_err() {
                            break;
                        }
                        break;
                    },
                }
            }
        });

        let stream = ReceiverStream::new(rx);
        Ok(Response::new(Box::pin(stream)))
    }

    // Server-streaming: polls GetOrderBook at configurable interval
    type StreamOrderBookStream =
        Pin<Box<dyn Stream<Item = Result<GetOrderBookResponse, Status>> + Send>>;

    #[instrument(skip_all)]
    async fn stream_order_book(
        &self,
        request: Request<StreamOrderBookRequest>,
    ) -> TonicResult<Response<Self::StreamOrderBookStream>> {
        let req = request.into_inner();
        let interval_secs = if req.interval_seconds > 0 {
            u64::from(req.interval_seconds.clamp(1, 60))
        } else {
            1
        };

        info!(
            interval_secs = interval_secs,
            symbol = %req.symbol,
            depth = req.depth,
            "StreamOrderBook started"
        );

        let (tx, rx) = mpsc::channel(16);
        let state = Arc::clone(&self.state);
        let symbol = req.symbol;
        let depth = req.depth;

        tokio::spawn(async move {
            let svc = TradingServiceImpl::new(state);
            let mut interval =
                tokio::time::interval(std::time::Duration::from_secs(interval_secs));

            loop {
                interval.tick().await;

                let ob_req = GetOrderBookRequest {
                    symbol: symbol.clone(),
                    depth: if depth > 0 { Some(depth) } else { None },
                };

                match svc
                    .get_order_book(tonic::Request::new(ob_req))
                    .await
                {
                    Ok(resp) => {
                        if tx.send(Ok(resp.into_inner())).await.is_err() {
                            info!("StreamOrderBook: client disconnected");
                            break;
                        }
                    },
                    Err(e) => {
                        warn!("StreamOrderBook: poll failed: {}", e);
                        if tx.send(Err(e)).await.is_err() {
                            break;
                        }
                        break;
                    },
                }
            }
        });

        let stream = ReceiverStream::new(rx);
        Ok(Response::new(Box::pin(stream)))
    }
}

impl TradingServiceImpl {
    /// Validate order against risk parameters
    async fn validate_order_risk(&self, order: &SubmitOrderRequest) -> TradingServiceResult<()> {
        // Basic risk validations that can be done without RiskManager

        // Validate quantity limits (from common sense limits, not config)
        const MAX_REASONABLE_QUANTITY: f64 = 100_000.0; // Reasonable max for single order
        if order.quantity > MAX_REASONABLE_QUANTITY {
            return Err(crate::error::TradingServiceError::RiskViolation {
                violation_type: "QuantityLimit".to_string(),
                message: "Order quantity exceeds maximum reasonable size".to_string(),
            });
        }

        // Validate notional value if price is available
        if let Some(price) = order.price {
            const MAX_NOTIONAL: f64 = 10_000_000.0; // $10M max notional
            let notional = order.quantity * price;

            // Check for overflow/infinity in notional calculation
            if !notional.is_finite() {
                return Err(crate::error::TradingServiceError::RiskViolation {
                    violation_type: "CalculationOverflow".to_string(),
                    message: format!(
                        "Notional calculation overflow: {} * {} = {}",
                        order.quantity, price, notional
                    ),
                });
            }

            if notional > MAX_NOTIONAL {
                return Err(crate::error::TradingServiceError::RiskViolation {
                    violation_type: "NotionalLimit".to_string(),
                    message: "Order notional value exceeds maximum limit".to_string(),
                });
            }
        }

        // NOTE: Full risk validation (VaR, position limits, etc.) requires
        // architectural refactoring to make RiskManager accessible from trading service
        // For now, these basic checks provide some risk protection

        Ok(())
    }

    /// Publish order event to event stream
    async fn publish_order_event(&self, order_id: &str, event_type: OrderEventType) {
        use crate::event_streaming::events::{TradingEvent, TradingEventType};

        let event_type_internal = match event_type {
            OrderEventType::Created => TradingEventType::OrderSubmitted,
            OrderEventType::Filled => TradingEventType::OrderFilled,
            OrderEventType::Cancelled => TradingEventType::OrderCancelled,
            OrderEventType::Rejected => TradingEventType::OrderRejected,
            OrderEventType::Updated => TradingEventType::OrderModified,
            OrderEventType::PartiallyFilled => TradingEventType::PartialFill,
            OrderEventType::Unspecified => TradingEventType::OrderModified, // Default fallback
        };

        let payload = serde_json::json!({
            "order_id": order_id,
        })
        .to_string();

        let event = TradingEvent::new(event_type_internal, order_id.to_string(), payload);

        if let Err(e) = self.state.event_publisher.publish(event).await {
            warn!("Failed to publish event: {}", e);
        }
    }

    /// Convert TradingEvent to OrderEvent proto
    fn convert_to_order_event(event: &crate::event_streaming::events::TradingEvent) -> OrderEvent {
        // Parse payload JSON to extract order details
        let order_id = event.correlation_id.clone().unwrap_or_default();

        let event_type = match event.event_type {
            crate::event_streaming::events::TradingEventType::OrderSubmitted => {
                OrderEventType::Created
            },
            crate::event_streaming::events::TradingEventType::OrderFilled => OrderEventType::Filled,
            crate::event_streaming::events::TradingEventType::PartialFill => {
                OrderEventType::PartiallyFilled
            },
            crate::event_streaming::events::TradingEventType::OrderCancelled => {
                OrderEventType::Cancelled
            },
            crate::event_streaming::events::TradingEventType::OrderRejected => {
                OrderEventType::Rejected
            },
            _ => OrderEventType::Updated,
        };

        // Parse order details from JSON payload
        let order_data: serde_json::Value =
            serde_json::from_str(&event.payload).unwrap_or_default();

        let order = Some(Order {
            order_id: order_id.clone(),
            symbol: order_data.get("symbol").and_then(|v| v.as_str()).unwrap_or("").to_string(),
            side: order_data
                .get("side")
                .and_then(|v| v.as_i64())
                .unwrap_or(0) as i32,
            quantity: order_data.get("quantity").and_then(|v| v.as_f64()).unwrap_or(0.0),
            filled_quantity: order_data
                .get("filled_quantity")
                .and_then(|v| v.as_f64())
                .unwrap_or(0.0),
            order_type: order_data
                .get("order_type")
                .and_then(|v| v.as_i64())
                .unwrap_or(0) as i32,
            price: order_data.get("price").and_then(|v| v.as_f64()),
            stop_price: order_data.get("stop_price").and_then(|v| v.as_f64()),
            status: order_data
                .get("status")
                .and_then(|v| v.as_i64())
                .unwrap_or(0) as i32,
            created_at: order_data
                .get("created_at")
                .and_then(|v| v.as_i64())
                .unwrap_or(event.timestamp.timestamp()),
            updated_at: order_data.get("updated_at").and_then(|v| v.as_i64()),
            account_id: order_data
                .get("account_id")
                .and_then(|v| v.as_str())
                .unwrap_or("")
                .to_string(),
            metadata: std::collections::HashMap::new(),
        });

        OrderEvent {
            order_id,
            order,
            message: String::new(),
            event_type: event_type as i32,
            timestamp: event.timestamp.timestamp(),
        }
    }

    /// Convert TradingEvent to PositionEvent proto
    fn convert_to_position_event(
        event: &crate::event_streaming::events::TradingEvent,
    ) -> PositionEvent {
        // Parse payload to extract position details
        let position_data: serde_json::Value =
            serde_json::from_str(&event.payload).unwrap_or_default();

        let symbol = position_data.get("symbol").and_then(|v| v.as_str()).unwrap_or("").to_string();
        let quantity = position_data.get("quantity").and_then(|v| v.as_f64()).unwrap_or(0.0);
        let average_price = position_data
            .get("average_price")
            .and_then(|v| v.as_f64())
            .unwrap_or(0.0);
        let unrealized_pnl = position_data
            .get("unrealized_pnl")
            .and_then(|v| v.as_f64())
            .unwrap_or(0.0);
        let market_value = position_data
            .get("market_value")
            .and_then(|v| v.as_f64())
            .unwrap_or(0.0);
        let realized_pnl = position_data
            .get("realized_pnl")
            .and_then(|v| v.as_f64())
            .unwrap_or(0.0);
        let account_id = position_data
            .get("account_id")
            .and_then(|v| v.as_str())
            .unwrap_or("")
            .to_string();

        let position = Some(Position {
            symbol: symbol.clone(),
            quantity,
            average_price,
            market_value,
            unrealized_pnl,
            realized_pnl,
            account_id,
            updated_at: event.timestamp.timestamp(),
        });

        PositionEvent {
            symbol,
            position,
            quantity,
            average_price,
            unrealized_pnl,
            event_type: match event.event_type {
                crate::event_streaming::events::TradingEventType::PositionOpened => 1,
                crate::event_streaming::events::TradingEventType::PositionClosed => 2,
                crate::event_streaming::events::TradingEventType::PositionModified => 3,
                _ => 0,
            },
            timestamp: event.timestamp.timestamp(),
        }
    }

    /// Convert TradingEvent to ExecutionEvent proto
    fn convert_to_execution_event(
        event: &crate::event_streaming::events::TradingEvent,
    ) -> ExecutionEvent {
        let execution_data: serde_json::Value =
            serde_json::from_str(&event.payload).unwrap_or_default();

        let execution_id = event.id.clone();
        let order_id = event.correlation_id.clone().unwrap_or_default();
        let symbol = execution_data.get("symbol").and_then(|v| v.as_str()).unwrap_or("").to_string();
        let quantity = execution_data.get("quantity").and_then(|v| v.as_f64()).unwrap_or(0.0);
        let price = execution_data.get("price").and_then(|v| v.as_f64()).unwrap_or(0.0);

        let execution = Some(crate::proto::trading::Execution {
            execution_id: execution_id.clone(),
            order_id: order_id.clone(),
            symbol: symbol.clone(),
            side: execution_data
                .get("side")
                .and_then(|v| v.as_i64())
                .unwrap_or(0) as i32,
            quantity,
            price,
            timestamp: event.timestamp.timestamp(),
            account_id: execution_data
                .get("account_id")
                .and_then(|v| v.as_str())
                .unwrap_or("")
                .to_string(),
            metadata: std::collections::HashMap::new(),
        });

        ExecutionEvent {
            execution_id,
            execution,
            timestamp: event.timestamp.timestamp(),
            order_id,
            symbol,
            quantity,
            price,
        }
    }

    /// Convert TradingEvent to MarketDataEvent proto
    fn convert_to_market_data_event(
        event: &crate::event_streaming::events::TradingEvent,
    ) -> MarketDataEvent {
        let market_data: serde_json::Value =
            serde_json::from_str(&event.payload).unwrap_or_default();

        use crate::proto::trading::market_data_event;

        MarketDataEvent {
            symbol: market_data["symbol"].as_str().unwrap_or("").to_string(),
            timestamp: event.timestamp.timestamp(),
            data_type: crate::proto::trading::MarketDataType::Trade as i32,
            data: Some(market_data_event::Data::Trade(
                crate::proto::trading::Trade {
                    price: market_data["price"].as_f64().unwrap_or(0.0),
                    volume: market_data["volume"].as_f64().unwrap_or(0.0),
                    timestamp: event.timestamp.timestamp(),
                },
            )),
        }
    }

    /// Calculate ML model performance metrics from ensemble_predictions table (UPDATED - Agent C7)
    ///
    /// This method queries the ensemble_predictions table for real-time performance data
    /// WITH OUTCOME TRACKING (actual_outcome, pnl, closed_at) and calculates comprehensive
    /// metrics including Sharpe ratio, win rate, and accuracy.
    ///
    /// CHANGE: Now filters by `actual_outcome IS NOT NULL` to only include closed positions
    async fn calculate_model_performance_metrics(
        &self,
        model_name: &str,
        start_time: Option<i64>,
        end_time: Option<i64>,
    ) -> TonicResult<crate::proto::trading::ModelPerformance> {
        use chrono::DateTime;

        // Convert timestamps to DateTime
        let start_dt = start_time.and_then(|ts| DateTime::from_timestamp(ts, 0));
        let end_dt = end_time.and_then(|ts| DateTime::from_timestamp(ts, 0));

        // Query predictions with P&L data AND OUTCOME TRACKING (UPDATED - Agent C7)
        // CHANGE: Filter by actual_outcome IS NOT NULL to only include closed positions
        let predictions = sqlx::query!(
            r#"
            SELECT
                dqn_signal, dqn_confidence, dqn_vote,
                mamba2_signal, mamba2_confidence, mamba2_vote,
                ppo_signal, ppo_confidence, ppo_vote,
                tft_signal, tft_confidence, tft_vote,
                pnl, ensemble_action, actual_outcome, closed_at
            FROM ensemble_predictions
            WHERE actual_outcome IS NOT NULL
              AND closed_at IS NOT NULL
              AND pnl IS NOT NULL
              AND ($1::timestamptz IS NULL OR prediction_timestamp >= $1)
              AND ($2::timestamptz IS NULL OR prediction_timestamp <= $2)
              AND (
                  ($3 = 'DQN' AND dqn_signal IS NOT NULL) OR
                  ($3 = 'MAMBA2' AND mamba2_signal IS NOT NULL) OR
                  ($3 = 'PPO' AND ppo_signal IS NOT NULL) OR
                  ($3 = 'TFT' AND tft_signal IS NOT NULL)
              )
            ORDER BY prediction_timestamp DESC
            "#,
            start_dt,
            end_dt,
            model_name,
        )
        .fetch_all(&self.state.db_pool)
        .await
        .map_err(|e| {
            Status::internal(format!(
                "Failed to query {} predictions with outcomes: {}",
                model_name, e
            ))
        })?;

        let total_predictions = predictions.len() as i64;
        if total_predictions == 0 {
            return Ok(crate::proto::trading::ModelPerformance {
                model_name: model_name.to_string(),
                total_predictions: 0,
                correct_predictions: 0,
                accuracy: 0.0,
                sharpe_ratio: 0.0,
                avg_pnl: 0.0,
            });
        }

        // Calculate P&L metrics
        let pnl_values: Vec<f64> = predictions
            .iter()
            .filter_map(|p| p.pnl.map(|pnl_cents| pnl_cents as f64 / 100.0))
            .collect();

        let avg_pnl = if !pnl_values.is_empty() {
            pnl_values.iter().sum::<f64>() / pnl_values.len() as f64
        } else {
            0.0
        };

        // Calculate Sharpe ratio (risk-adjusted returns)
        let sharpe_ratio = self.calculate_sharpe_ratio_from_pnl(&pnl_values)?;

        // Calculate accuracy (correct direction predictions)
        let correct_predictions = predictions
            .iter()
            .filter(|p| {
                let model_vote = match model_name {
                    "DQN" => p.dqn_vote.as_deref(),
                    "MAMBA2" => p.mamba2_vote.as_deref(),
                    "PPO" => p.ppo_vote.as_deref(),
                    "TFT" => p.tft_vote.as_deref(),
                    _ => None,
                };
                // Model is correct if its vote matches the ensemble action
                model_vote.map(|v| v == p.ensemble_action).unwrap_or(false)
            })
            .count() as i64;

        let accuracy = correct_predictions as f64 / total_predictions as f64;

        Ok(crate::proto::trading::ModelPerformance {
            model_name: model_name.to_string(),
            total_predictions,
            correct_predictions,
            accuracy,
            sharpe_ratio,
            avg_pnl,
        })
    }

    /// Calculate Sharpe ratio from P&L values
    ///
    /// Sharpe ratio = (Mean Return / Std Dev of Returns) * sqrt(252)
    /// Annualized for 252 trading days
    fn calculate_sharpe_ratio_from_pnl(&self, pnl_values: &[f64]) -> TonicResult<f64> {
        if pnl_values.len() < 2 {
            return Ok(0.0);
        }

        let mean = pnl_values.iter().sum::<f64>() / pnl_values.len() as f64;

        // Calculate standard deviation
        let variance = pnl_values
            .iter()
            .map(|x| {
                let diff = x - mean;
                diff * diff
            })
            .sum::<f64>()
            / (pnl_values.len() - 1) as f64;

        let std_dev = variance.sqrt();

        if std_dev == 0.0 {
            return Ok(0.0);
        }

        // Annualize Sharpe ratio (252 trading days per year)
        let sharpe = (mean / std_dev) * (252.0_f64).sqrt();

        Ok(sharpe)
    }
}
