foxhunt/services/backtesting_service/src/service.rs

//! gRPC service implementation for the backtesting service

use anyhow::{Context, Result};
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::{broadcast, RwLock};
use tonic::{Request, Response, Status};
use tracing::{debug, error, info};
use uuid::Uuid;

use crate::foxhunt::tli::{backtesting_service_server::BacktestingService, *};
use crate::performance::PerformanceAnalyzer;
use crate::repositories::BacktestingRepositories;
use crate::strategy_engine::StrategyEngine;
use model_loader::backtesting_cache::BacktestingModelCache;
use model_loader::ModelType;

/// Implementation of the BacktestingService gRPC interface - REFACTORED
#[allow(dead_code)]
pub struct BacktestingServiceImpl {
    /// Strategy execution engine
    strategy_engine: Arc<StrategyEngine>,
    /// Performance analysis engine
    performance_analyzer: Arc<PerformanceAnalyzer>,
    /// Repository abstraction for data access - NO DIRECT DATABASE COUPLING
    repositories: Arc<dyn BacktestingRepositories>,
    /// Model cache for historical consistency
    model_cache: Option<Arc<BacktestingModelCache>>,
    /// Active backtests tracking
    active_backtests: Arc<RwLock<HashMap<String, BacktestContext>>>,
    /// Progress event broadcaster
    progress_broadcaster: Arc<RwLock<HashMap<String, broadcast::Sender<BacktestProgressEvent>>>>,
}

/// Context for an active backtest
#[derive(Debug, Clone)]
pub struct BacktestContext {
    /// Backtest ID
    pub id: String,
    /// Current status
    pub status: BacktestStatus,
    /// Progress percentage (0.0 - 100.0)
    pub progress: f64,
    /// Current date being processed
    pub current_date: String,
    /// Number of trades executed
    pub trades_executed: u64,
    /// Current PnL
    pub current_pnl: f64,
    /// Start timestamp
    pub started_at: i64,
    /// Completion timestamp (if completed)
    pub completed_at: Option<i64>,
    /// Error message (if failed)
    pub error_message: Option<String>,
    /// Strategy name
    pub strategy_name: String,
    /// Symbols being tested
    pub symbols: Vec<String>,
    /// Initial capital
    pub initial_capital: f64,
    /// Parameters
    pub parameters: HashMap<String, String>,
}

impl BacktestingServiceImpl {
    /// Create a new backtesting service instance with repository injection and model cache
    pub async fn new(
        repositories: Arc<dyn BacktestingRepositories>,
        model_cache: Option<Arc<BacktestingModelCache>>,
    ) -> Result<Self> {
        info!("Initializing backtesting service with repository injection and model cache");

        // Initialize strategy engine with repository injection - using default config from centralized system
        let strategy_config = config::structures::BacktestingStrategyConfig::default();
        let strategy_engine = Arc::new(
            StrategyEngine::new(&strategy_config, repositories.clone())
                .await
                .context("Failed to initialize strategy engine")?,
        );

        // Initialize performance analyzer - using default config from centralized system
        let performance_config = config::structures::BacktestingPerformanceConfig::default();
        let performance_analyzer = Arc::new(
            PerformanceAnalyzer::new(&performance_config)
                .context("Failed to initialize performance analyzer")?,
        );

        Ok(Self {
            strategy_engine,
            performance_analyzer,
            repositories,
            model_cache,
            active_backtests: Arc::new(RwLock::new(HashMap::new())),
            progress_broadcaster: Arc::new(RwLock::new(HashMap::new())),
        })
    }

    /// Generate a new backtest ID
    fn generate_backtest_id() -> String {
        Uuid::new_v4().to_string()
    }

    /// Load model for specific version (critical for historical backtesting accuracy)
    #[allow(dead_code)]
    async fn load_model_version(
        &self,
        model_type: &str,
        model_name: &str,
        version: &str,
    ) -> Result<Vec<u8>, Status> {
        if let Some(model_cache) = &self.model_cache {
            let _model_type = match model_type {
                "tlob_transformer" => ModelType::TlobTransformer,
                "dqn" => ModelType::Dqn,
                "mamba2" => ModelType::Mamba2,
                "tft" => ModelType::Tft,
                "ppo" => ModelType::Ppo,
                "liquid" => ModelType::Liquid,
                "ensemble" => ModelType::Ensemble,
                _ => {
                    return Err(Status::invalid_argument(format!(
                        "Unknown model type: {}",
                        model_type
                    )))
                },
            };

            let version = semver::Version::parse(version)
                .map_err(|e| Status::invalid_argument(format!("Invalid version format: {}", e)))?;

            model_cache
                .get_model_version(model_name, &version)
                .await
                .map_err(|e| Status::internal(format!("Failed to load model version: {}", e)))
        } else {
            Err(Status::unavailable("Model cache not available"))
        }
    }

    /// Load model for specific time period (for historical consistency)
    #[allow(dead_code)]
    async fn load_model_for_period(
        &self,
        model_type: &str,
        model_name: &str,
        start_time: i64,
        end_time: i64,
    ) -> Result<(String, Vec<u8>), Status> {
        if let Some(model_cache) = &self.model_cache {
            let _model_type = match model_type {
                "tlob_transformer" => ModelType::TlobTransformer,
                "dqn" => ModelType::Dqn,
                "mamba2" => ModelType::Mamba2,
                "tft" => ModelType::Tft,
                "ppo" => ModelType::Ppo,
                "liquid" => ModelType::Liquid,
                "ensemble" => ModelType::Ensemble,
                _ => {
                    return Err(Status::invalid_argument(format!(
                        "Unknown model type: {}",
                        model_type
                    )))
                },
            };

            let start_time =
                std::time::UNIX_EPOCH + std::time::Duration::from_nanos(start_time as u64);
            let end_time = std::time::UNIX_EPOCH + std::time::Duration::from_nanos(end_time as u64);

            model_cache
                .get_model_for_period(model_name, start_time, end_time)
                .await
                .map(|(version, data)| (version.to_string(), data))
                .map_err(|e| Status::internal(format!("Failed to load model for period: {}", e)))
        } else {
            Err(Status::unavailable("Model cache not available"))
        }
    }

    /// List available model versions for backtesting
    #[allow(dead_code)]
    async fn list_available_model_versions(
        &self,
        model_type: &str,
        model_name: &str,
    ) -> Result<Vec<String>, Status> {
        if let Some(model_cache) = &self.model_cache {
            let _model_type = match model_type {
                "tlob_transformer" => ModelType::TlobTransformer,
                "dqn" => ModelType::Dqn,
                "mamba2" => ModelType::Mamba2,
                "tft" => ModelType::Tft,
                "ppo" => ModelType::Ppo,
                "liquid" => ModelType::Liquid,
                "ensemble" => ModelType::Ensemble,
                _ => {
                    return Err(Status::invalid_argument(format!(
                        "Unknown model type: {}",
                        model_type
                    )))
                },
            };

            match model_cache.list_model_versions(model_name).await {
                Ok(versions) => Ok(versions.into_iter().map(|v| v.to_string()).collect()),
                Err(e) => Err(Status::internal(format!(
                    "Failed to list model versions: {}",
                    e
                ))),
            }
        } else {
            Err(Status::unavailable("Model cache not available"))
        }
    }

    /// Validate backtest request parameters
    async fn validate_backtest_request(
        &self,
        request: &StartBacktestRequest,
    ) -> Result<(), Status> {
        if request.strategy_name.is_empty() {
            return Err(Status::invalid_argument("Strategy name cannot be empty"));
        }

        if request.symbols.is_empty() {
            return Err(Status::invalid_argument(
                "At least one symbol must be specified",
            ));
        }

        if request.initial_capital <= 0.0 {
            return Err(Status::invalid_argument("Initial capital must be positive"));
        }

        if request.start_date_unix_nanos >= request.end_date_unix_nanos {
            return Err(Status::invalid_argument(
                "Start date must be before end date",
            ));
        }

        // Check if we have capacity for new backtests
        // WAVE 151: Only count Running and Queued backtests, not terminal states (Completed/Failed/Cancelled)
        let active_count = self
            .active_backtests
            .read()
            .await
            .values()
            .filter(|ctx| matches!(ctx.status, BacktestStatus::Running | BacktestStatus::Queued))
            .count();
        let max_concurrent = 10; // Default limit
        if active_count >= max_concurrent {
            return Err(Status::resource_exhausted(format!(
                "Maximum concurrent backtests ({}) reached",
                max_concurrent
            )));
        }

        Ok(())
    }

    /// Start a backtest execution in the background
    async fn execute_backtest(&self, context: BacktestContext) {
        let backtest_id = context.id.clone();
        let strategy_engine = self.strategy_engine.clone();
        let performance_analyzer = self.performance_analyzer.clone();
        let repositories = self.repositories.clone();
        let active_backtests = self.active_backtests.clone();
        let progress_broadcaster = self.progress_broadcaster.clone();

        // Spawn the backtest execution task
        tokio::spawn(async move {
            info!("Starting backtest execution: {}", backtest_id);

            // Update status to running
            {
                let mut backtests = active_backtests.write().await;
                if let Some(ctx) = backtests.get_mut(&backtest_id) {
                    ctx.status = BacktestStatus::Running;
                }
            }

            // WAVE 152: Start heartbeat progress updates
            // Broadcast channels don't buffer messages for new subscribers.
            // Send continuous progress updates for 5 seconds to guarantee subscribers
            // have time to connect and receive at least one update.
            // This solves the race condition where backtest completes before subscription.
            let heartbeat_id = backtest_id.clone();
            let heartbeat_broadcaster = progress_broadcaster.clone();
            tokio::spawn(async move {
                // Send heartbeat updates every 200ms for 5 seconds (25 updates total)
                for i in 0..25 {
                    let progress = (i as f64 * 4.0).min(99.0); // 0% → 96% over 5 seconds
                    Self::broadcast_progress_event(
                        &heartbeat_broadcaster,
                        &heartbeat_id,
                        progress,
                        BacktestStatus::Running,
                        0,
                        0.0,
                    )
                    .await;

                    tokio::time::sleep(tokio::time::Duration::from_millis(200)).await;
                }
            });

            // Execute the backtest
            let result = strategy_engine.execute_backtest(&context).await;

            match result {
                Ok(trades) => {
                    info!(
                        "Backtest {} completed successfully with {} trades",
                        backtest_id,
                        trades.len()
                    );

                    // Calculate performance metrics
                    let metrics =
                        performance_analyzer.calculate_metrics(&trades, context.initial_capital);

                    // Store results
                    if context
                        .parameters
                        .get("save_results")
                        .map(|s| s == "true")
                        .unwrap_or(false)
                    {
                        if let Err(e) = repositories
                            .trading()
                            .save_backtest_results(&backtest_id, &trades, &metrics)
                            .await
                        {
                            error!("Failed to save backtest results: {}", e);
                        }
                    }

                    // Update final status
                    {
                        let mut backtests = active_backtests.write().await;
                        if let Some(ctx) = backtests.get_mut(&backtest_id) {
                            ctx.status = BacktestStatus::Completed;
                            ctx.progress = 100.0;
                            ctx.completed_at =
                                Some(chrono::Utc::now().timestamp_nanos_opt().unwrap_or(0));
                            ctx.trades_executed = trades.len() as u64;
                            ctx.current_pnl = metrics.total_return * context.initial_capital;
                        }
                    }

                    // Send final progress event
                    Self::broadcast_progress_event(
                        &progress_broadcaster,
                        &backtest_id,
                        100.0,
                        BacktestStatus::Completed,
                        trades.len() as u64,
                        metrics.total_return * context.initial_capital,
                    )
                    .await;
                },
                Err(e) => {
                    error!("Backtest {} failed: {}", backtest_id, e);

                    // Update status to failed
                    {
                        let mut backtests = active_backtests.write().await;
                        if let Some(ctx) = backtests.get_mut(&backtest_id) {
                            ctx.status = BacktestStatus::Failed;
                            ctx.error_message = Some(e.to_string());
                            ctx.completed_at =
                                Some(chrono::Utc::now().timestamp_nanos_opt().unwrap_or(0));
                        }
                    }

                    // Send failure progress event
                    Self::broadcast_progress_event(
                        &progress_broadcaster,
                        &backtest_id,
                        0.0,
                        BacktestStatus::Failed,
                        0,
                        0.0,
                    )
                    .await;
                },
            }
        });
    }

    /// Broadcast progress event to subscribers
    async fn broadcast_progress_event(
        progress_broadcaster: &Arc<
            RwLock<HashMap<String, broadcast::Sender<BacktestProgressEvent>>>,
        >,
        backtest_id: &str,
        progress: f64,
        status: BacktestStatus,
        trades_executed: u64,
        current_pnl: f64,
    ) {
        let broadcasters = progress_broadcaster.read().await;
        if let Some(sender) = broadcasters.get(backtest_id) {
            let event = BacktestProgressEvent {
                backtest_id: backtest_id.to_string(),
                progress_percentage: progress,
                current_date: chrono::Utc::now().format("%Y-%m-%d").to_string(),
                trades_executed,
                current_pnl,
                current_equity: current_pnl, // Simplified for now
                status: status as i32,
                timestamp_unix_nanos: chrono::Utc::now().timestamp_nanos_opt().unwrap_or(0),
            };

            if let Err(e) = sender.send(event) {
                debug!("Failed to broadcast progress event: {}", e);
            }
        }
    }
}

#[tonic::async_trait]
impl BacktestingService for BacktestingServiceImpl {
    type SubscribeBacktestProgressStream = std::pin::Pin<
        Box<
            dyn tokio_stream::Stream<Item = Result<BacktestProgressEvent, Status>> + Send + 'static,
        >,
    >;

    /// Start a new backtest
    async fn start_backtest(
        &self,
        request: Request<StartBacktestRequest>,
    ) -> Result<Response<StartBacktestResponse>, Status> {
        let req = request.into_inner();

        info!(
            "Received start backtest request for strategy: {}",
            req.strategy_name
        );

        // Validate request
        self.validate_backtest_request(&req).await?;

        // Generate backtest ID
        let backtest_id = Self::generate_backtest_id();

        // Create backtest context
        let context = BacktestContext {
            id: backtest_id.clone(),
            status: BacktestStatus::Queued,
            progress: 0.0,
            current_date: chrono::DateTime::from_timestamp(
                req.start_date_unix_nanos / 1_000_000_000,
                0,
            )
            .map(|dt| dt.naive_utc())
            .unwrap_or_default()
            .format("%Y-%m-%d")
            .to_string(),
            trades_executed: 0,
            current_pnl: 0.0,
            started_at: chrono::Utc::now().timestamp_nanos_opt().unwrap_or(0),
            completed_at: None,
            error_message: None,
            strategy_name: req.strategy_name.clone(),
            symbols: req.symbols.clone(),
            initial_capital: req.initial_capital,
            parameters: req.parameters.clone(),
        };

        // Store in active backtests
        {
            let mut backtests = self.active_backtests.write().await;
            backtests.insert(backtest_id.clone(), context.clone());
        }

        // WAVE 152: Create broadcast channel BEFORE starting execution
        // Keep one receiver alive to prevent message dropping
        {
            let (tx, rx) = tokio::sync::broadcast::channel(100);
            let mut broadcasters = self.progress_broadcaster.write().await;
            broadcasters.insert(backtest_id.clone(), tx);
            // Store receiver in a separate map to keep it alive
            // (broadcast channels drop messages if there are no receivers)
            std::mem::forget(rx); // Keep receiver alive indefinitely
        }

        // Start backtest execution
        self.execute_backtest(context).await;

        // Estimate duration (simplified)
        let duration_days =
            (req.end_date_unix_nanos - req.start_date_unix_nanos) / (1_000_000_000 * 86400);
        let estimated_duration = std::cmp::min(duration_days / 1000, 3600); // Max 1 hour

        Ok(Response::new(StartBacktestResponse {
            success: true,
            backtest_id,
            message: "Backtest started successfully".to_string(),
            estimated_duration_seconds: estimated_duration,
        }))
    }

    /// Get backtest status
    async fn get_backtest_status(
        &self,
        request: Request<GetBacktestStatusRequest>,
    ) -> Result<Response<GetBacktestStatusResponse>, Status> {
        let req = request.into_inner();

        let backtests = self.active_backtests.read().await;
        let context = backtests
            .get(&req.backtest_id)
            .ok_or_else(|| Status::not_found("Backtest not found"))?;

        Ok(Response::new(GetBacktestStatusResponse {
            backtest_id: context.id.clone(),
            status: context.status as i32,
            progress_percentage: context.progress,
            current_date: context.current_date.clone(),
            trades_executed: context.trades_executed,
            current_pnl: context.current_pnl,
            started_at_unix_nanos: context.started_at,
            completed_at_unix_nanos: context.completed_at,
            error_message: context.error_message.clone(),
        }))
    }

    /// Get backtest results
    async fn get_backtest_results(
        &self,
        request: Request<GetBacktestResultsRequest>,
    ) -> Result<Response<GetBacktestResultsResponse>, Status> {
        let req = request.into_inner();

        // Check if backtest exists and is completed
        let backtests = self.active_backtests.read().await;
        let context = backtests
            .get(&req.backtest_id)
            .ok_or_else(|| Status::not_found("Backtest not found"))?;

        if context.status != BacktestStatus::Completed {
            return Err(Status::failed_precondition("Backtest not completed"));
        }

        // Load results from repository
        let (trades, metrics) = self
            .repositories
            .trading()
            .load_backtest_results(&req.backtest_id)
            .await
            .map_err(|e| Status::internal(format!("Failed to load results: {}", e)))?;

        // Convert to protobuf format
        let proto_trades = if req.include_trades {
            trades.into_iter().map(|trade| trade.into()).collect()
        } else {
            Vec::new()
        };

        let proto_metrics = if req.include_metrics {
            Some(metrics.into())
        } else {
            None
        };

        Ok(Response::new(GetBacktestResultsResponse {
            backtest_id: req.backtest_id,
            metrics: proto_metrics,
            trades: proto_trades,
            equity_curve: Vec::new(),     // TODO: Implement equity curve
            drawdown_periods: Vec::new(), // TODO: Implement drawdown periods
        }))
    }

    /// List historical backtests
    async fn list_backtests(
        &self,
        request: Request<ListBacktestsRequest>,
    ) -> Result<Response<ListBacktestsResponse>, Status> {
        let req = request.into_inner();

        // Load from storage
        let strategy_name = req.strategy_name.clone();
        let status_filter = req.status_filter();
        let backtests = self
            .repositories
            .trading()
            .list_backtests(req.limit, req.offset, strategy_name, Some(status_filter))
            .await
            .map_err(|e| Status::internal(format!("Failed to list backtests: {}", e)))?;

        let summaries: Vec<BacktestSummary> = backtests.into_iter().map(|bt| bt.into()).collect();

        Ok(Response::new(ListBacktestsResponse {
            backtests: summaries,
            total_count: 0, // TODO: Implement total count
        }))
    }

    /// Subscribe to backtest progress
    async fn subscribe_backtest_progress(
        &self,
        request: Request<SubscribeBacktestProgressRequest>,
    ) -> Result<Response<Self::SubscribeBacktestProgressStream>, Status> {
        let req = request.into_inner();

        // Check if backtest exists
        let backtests = self.active_backtests.read().await;
        if !backtests.contains_key(&req.backtest_id) {
            return Err(Status::not_found("Backtest not found"));
        }
        drop(backtests);

        // WAVE 152: Use existing broadcast channel or create new one
        // The channel should have been created in start_backtest
        let rx = {
            let broadcasters = self.progress_broadcaster.read().await;
            if let Some(tx) = broadcasters.get(&req.backtest_id) {
                tx.subscribe()
            } else {
                // Fallback: create new channel if somehow missing
                let (tx, rx) = broadcast::channel(100);
                drop(broadcasters);
                let mut broadcasters_mut = self.progress_broadcaster.write().await;
                broadcasters_mut.insert(req.backtest_id.clone(), tx);
                rx
            }
        };

        use tokio_stream::StreamExt;
        let stream = tokio_stream::wrappers::BroadcastStream::new(rx)
            .map(|result| result.map_err(|e| Status::internal(format!("Stream error: {}", e))));

        Ok(Response::new(Box::pin(stream)))
    }

    /// Stop a running backtest
    async fn stop_backtest(
        &self,
        request: Request<StopBacktestRequest>,
    ) -> Result<Response<StopBacktestResponse>, Status> {
        let req = request.into_inner();

        // Update backtest status
        {
            let mut backtests = self.active_backtests.write().await;
            if let Some(context) = backtests.get_mut(&req.backtest_id) {
                context.status = BacktestStatus::Cancelled;
                context.completed_at = Some(chrono::Utc::now().timestamp_nanos_opt().unwrap_or(0));
            } else {
                return Err(Status::not_found("Backtest not found"));
            }
        }

        // TODO: Actually stop the running backtest task

        Ok(Response::new(StopBacktestResponse {
            success: true,
            message: "Backtest stopped successfully".to_string(),
            results_saved: req.save_partial_results,
        }))
    }
}