//! Model Performance Tracking Repository //! //! Tracks model performance metrics, A/B testing results, and performance //! degradation detection for HFT ML models with PostgreSQL integration. use chrono::{DateTime, Utc}; use serde::{Deserialize, Serialize}; use std::collections::HashMap; use uuid::Uuid; use crate::{MlDataError, PerformanceConfig, Result}; use database::{Database, DatabaseTransaction}; use sqlx::Row; /// Performance tracking repository for ML models #[derive(Clone)] pub struct PerformanceRepository { db: Database, config: PerformanceConfig, } impl PerformanceRepository { pub async fn new(db: Database, config: PerformanceConfig) -> Result { let repo = Self { db, config }; repo.initialize_schema().await?; Ok(repo) } /// Initialize database schema for performance tracking pub async fn initialize_schema(&self) -> Result<()> { // Initialize schema // Model performance metrics table self.db .execute( r#" CREATE TABLE IF NOT EXISTS ml_model_performance ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), model_id UUID NOT NULL, model_name VARCHAR NOT NULL, model_version VARCHAR NOT NULL, environment VARCHAR NOT NULL, timestamp TIMESTAMPTZ NOT NULL, metric_name VARCHAR NOT NULL, metric_value DOUBLE PRECISION NOT NULL, metric_metadata JSONB DEFAULT '{}', created_at TIMESTAMPTZ DEFAULT NOW() ) "#, ) .await?; // Performance benchmarks table self.db .execute( r#" CREATE TABLE IF NOT EXISTS ml_performance_benchmarks ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), benchmark_name VARCHAR NOT NULL, model_id UUID NOT NULL, model_name VARCHAR NOT NULL, model_version VARCHAR NOT NULL, environment VARCHAR NOT NULL, started_at TIMESTAMPTZ NOT NULL, completed_at TIMESTAMPTZ, duration_ms BIGINT, status benchmark_status DEFAULT 'running', results JSONB DEFAULT '{}', error_message TEXT, created_by VARCHAR NOT NULL, metadata JSONB DEFAULT '{}' ) "#, ) .await?; // A/B testing experiments table self.db .execute( r#" CREATE TABLE IF NOT EXISTS ml_ab_experiments ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), experiment_name VARCHAR NOT NULL, description TEXT, control_model_id UUID NOT NULL, treatment_model_id UUID NOT NULL, started_at TIMESTAMPTZ NOT NULL, ended_at TIMESTAMPTZ, status experiment_status DEFAULT 'running', traffic_split DOUBLE PRECISION NOT NULL, confidence_level DOUBLE PRECISION NOT NULL, results JSONB DEFAULT '{}', created_by VARCHAR NOT NULL, metadata JSONB DEFAULT '{}' ) "#, ) .await?; // A/B experiment metrics table self.db .execute( r#" CREATE TABLE IF NOT EXISTS ml_ab_experiment_metrics ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), experiment_id UUID NOT NULL REFERENCES ml_ab_experiments(id) ON DELETE CASCADE, model_variant ab_variant NOT NULL, timestamp TIMESTAMPTZ NOT NULL, metric_name VARCHAR NOT NULL, metric_value DOUBLE PRECISION NOT NULL, sample_size INTEGER NOT NULL, metadata JSONB DEFAULT '{}' ) "#, ) .await?; // Performance alerts table self.db .execute( r#" CREATE TABLE IF NOT EXISTS ml_performance_alerts ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), model_id UUID NOT NULL, model_name VARCHAR NOT NULL, alert_type alert_type NOT NULL, severity alert_severity NOT NULL, metric_name VARCHAR NOT NULL, threshold_value DOUBLE PRECISION NOT NULL, actual_value DOUBLE PRECISION NOT NULL, triggered_at TIMESTAMPTZ NOT NULL, resolved_at TIMESTAMPTZ, status alert_status DEFAULT 'active', message TEXT NOT NULL, metadata JSONB DEFAULT '{}' ) "#, ) .await?; // Create enums self.db.execute(r#" DO $$ BEGIN CREATE TYPE benchmark_status AS ENUM ('running', 'completed', 'failed', 'cancelled'); EXCEPTION WHEN duplicate_object THEN null; END $$; "#).await?; self.db.execute(r#" DO $$ BEGIN CREATE TYPE experiment_status AS ENUM ('draft', 'running', 'paused', 'completed', 'failed'); EXCEPTION WHEN duplicate_object THEN null; END $$; "#).await?; self.db .execute( r#" DO $$ BEGIN CREATE TYPE ab_variant AS ENUM ('control', 'treatment'); EXCEPTION WHEN duplicate_object THEN null; END $$; "#, ) .await?; self.db .execute( r#" DO $$ BEGIN CREATE TYPE alert_type AS ENUM ('degradation', 'anomaly', 'threshold', 'drift'); EXCEPTION WHEN duplicate_object THEN null; END $$; "#, ) .await?; self.db .execute( r#" DO $$ BEGIN CREATE TYPE alert_severity AS ENUM ('low', 'medium', 'high', 'critical'); EXCEPTION WHEN duplicate_object THEN null; END $$; "#, ) .await?; self.db .execute( r#" DO $$ BEGIN CREATE TYPE alert_status AS ENUM ('active', 'acknowledged', 'resolved'); EXCEPTION WHEN duplicate_object THEN null; END $$; "#, ) .await?; // Indexes for performance self.db.execute("CREATE INDEX IF NOT EXISTS idx_performance_model_timestamp ON ml_model_performance(model_id, timestamp DESC)").await?; self.db.execute("CREATE INDEX IF NOT EXISTS idx_performance_metric_name ON ml_model_performance(metric_name, timestamp DESC)").await?; self.db.execute("CREATE INDEX IF NOT EXISTS idx_benchmarks_model ON ml_performance_benchmarks(model_id, started_at DESC)").await?; self.db.execute("CREATE INDEX IF NOT EXISTS idx_experiments_status ON ml_ab_experiments(status, started_at DESC)").await?; self.db.execute("CREATE INDEX IF NOT EXISTS idx_alerts_model ON ml_performance_alerts(model_id, triggered_at DESC)").await?; Ok(()) } /// Record model performance metrics pub async fn record_metrics(&self, request: RecordMetricsRequest) -> Result<()> { let mut tx = self.db.begin_transaction().await?; let metric_count = request.metrics.len(); for metric in &request.metrics { let query = format!( r#"INSERT INTO ml_model_performance (model_id, model_name, model_version, environment, timestamp, metric_name, metric_value, metric_metadata) VALUES ('{}', '{}', '{}', '{}', '{}', '{}', {}, '{}')"#, request.model_id, request.model_name, request.model_version, request.environment, metric.timestamp.to_rfc3339(), metric.name, metric.value, metric.metadata.to_string().replace("'", "''") ); tx.execute(&query).await?; // Check for performance alerts self.check_performance_threshold(&mut tx, &request, metric) .await?; } tx.commit().await?; tracing::info!( "Recorded {} metrics for model {} in {}", metric_count, request.model_name, request.environment ); Ok(()) } /// Get performance metrics for a model pub async fn get_performance( &self, model_id: Uuid, metric_names: Option>, time_range: Option<(DateTime, DateTime)>, limit: Option, ) -> Result { let mut conn = self.db.acquire().await?; let _query = r#" SELECT timestamp, metric_name, metric_value, metric_metadata FROM ml_model_performance WHERE model_id = $1 "# .to_string(); // Using sqlx query builder pattern let mut query_builder = sqlx::QueryBuilder::new( "SELECT timestamp, metric_name, metric_value, metric_metadata FROM ml_model_performance WHERE model_id = " ); query_builder.push_bind(model_id); // Add metric name filter if let Some(ref names) = metric_names { query_builder.push(" AND metric_name = ANY("); query_builder.push_bind(names); query_builder.push(")"); } // Add time range filter if let Some((start, end)) = time_range { query_builder.push(" AND timestamp >= "); query_builder.push_bind(start); query_builder.push(" AND timestamp <= "); query_builder.push_bind(end); } query_builder.push(" ORDER BY timestamp DESC"); // Add limit if let Some(limit_val) = limit { query_builder.push(" LIMIT "); query_builder.push_bind(limit_val as i64); } let query = query_builder.build(); let rows = query.fetch_all(&mut *conn).await?; let mut metrics = Vec::new(); for row in rows { metrics.push(PerformanceMetric { timestamp: row.get("timestamp"), name: row.get("metric_name"), value: row.get("metric_value"), metadata: row.get("metric_metadata"), }); } // Calculate summary statistics let summary = self.calculate_performance_summary(&metrics); let last_updated = metrics.first().map(|m| m.timestamp); Ok(ModelPerformance { model_id, metrics, summary, last_updated, }) } /// Start a performance benchmark pub async fn start_benchmark(&self, request: StartBenchmarkRequest) -> Result { let benchmark_id = Uuid::new_v4(); let mut conn = self.db.acquire().await?; sqlx::query( r#"INSERT INTO ml_performance_benchmarks (id, benchmark_name, model_id, model_name, model_version, environment, started_at, created_by, metadata) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)"#, ) .bind(benchmark_id) .bind(&request.benchmark_name) .bind(request.model_id) .bind(&request.model_name) .bind(&request.model_version) .bind(&request.environment) .bind(request.started_at) .bind(&request.created_by) .bind(&request.metadata) .execute(conn.as_mut()) .await?; let benchmark = BenchmarkResult { id: benchmark_id, benchmark_name: request.benchmark_name, model_id: request.model_id, model_name: request.model_name, model_version: request.model_version, environment: request.environment, started_at: request.started_at, completed_at: None, duration_ms: None, status: BenchmarkStatus::Running, results: serde_json::Value::Object(serde_json::Map::new()), error_message: None, created_by: request.created_by, metadata: request.metadata, }; tracing::info!( "Started benchmark {} for model {}", benchmark.benchmark_name, benchmark.model_name ); Ok(benchmark) } /// Complete a performance benchmark pub async fn complete_benchmark( &self, benchmark_id: Uuid, results: serde_json::Value, error_message: Option, ) -> Result<()> { let mut conn = self.db.acquire().await?; let completed_at = Utc::now(); // Get start time to calculate duration let start_time: DateTime = sqlx::query_scalar("SELECT started_at FROM ml_performance_benchmarks WHERE id = $1") .bind(benchmark_id) .fetch_one(conn.as_mut()) .await?; let duration_ms = (completed_at - start_time).num_milliseconds(); let status = if error_message.is_some() { BenchmarkStatus::Failed } else { BenchmarkStatus::Completed }; sqlx::query( r#"UPDATE ml_performance_benchmarks SET completed_at = $1, duration_ms = $2, status = $3, results = $4, error_message = $5 WHERE id = $6"#, ) .bind(completed_at) .bind(duration_ms) .bind(status.to_string()) .bind(&results) .bind(&error_message) .bind(benchmark_id) .execute(conn.as_mut()) .await?; tracing::info!("Completed benchmark {} in {}ms", benchmark_id, duration_ms); Ok(()) } /// Create A/B testing experiment pub async fn create_experiment( &self, request: CreateExperimentRequest, ) -> Result { let mut conn = self.db.acquire().await?; let experiment_id = Uuid::new_v4(); sqlx::query( r#"INSERT INTO ml_ab_experiments (id, experiment_name, description, control_model_id, treatment_model_id, started_at, traffic_split, confidence_level, created_by, metadata) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)"#, ) .bind(experiment_id) .bind(&request.experiment_name) .bind(&request.description) .bind(request.control_model_id) .bind(request.treatment_model_id) .bind(request.started_at) .bind(request.traffic_split) .bind(request.confidence_level) .bind(&request.created_by) .bind(&request.metadata) .execute(conn.as_mut()) .await?; let experiment = AbTestExperiment { id: experiment_id, experiment_name: request.experiment_name, description: request.description, control_model_id: request.control_model_id, treatment_model_id: request.treatment_model_id, started_at: request.started_at, ended_at: None, status: ExperimentStatus::Running, traffic_split: request.traffic_split, confidence_level: request.confidence_level, results: serde_json::Value::Object(serde_json::Map::new()), created_by: request.created_by, metadata: request.metadata, metrics: Vec::new(), }; tracing::info!("Created A/B experiment: {}", experiment.experiment_name); Ok(experiment) } /// Record A/B experiment metrics pub async fn record_experiment_metrics( &self, experiment_id: Uuid, metrics: Vec, ) -> Result<()> { let mut tx = self.db.begin_transaction().await?; let metric_count = metrics.len(); for metric in metrics { let query = format!( r#"INSERT INTO ml_ab_experiment_metrics (experiment_id, model_variant, timestamp, metric_name, metric_value, sample_size, metadata) VALUES ('{}', '{}', '{}', '{}', {}, {}, '{}')"#, experiment_id, metric.variant.to_string(), metric.timestamp.to_rfc3339(), metric.metric_name, metric.metric_value, metric.sample_size, metric.metadata.to_string().replace("'", "''") ); tx.execute(&query).await?; } tx.commit().await?; tracing::info!("Recorded {} experiment metrics", metric_count); Ok(()) } /// Get active performance alerts pub async fn get_active_alerts(&self, model_id: Option) -> Result> { let mut conn = self.db.acquire().await?; // Using sqlx query builder pattern let rows = if let Some(model_id) = model_id { sqlx::query_as::< _, ( Uuid, Uuid, String, String, String, String, f64, f64, DateTime, String, serde_json::Value, ), >( r#"SELECT id, model_id, model_name, alert_type, severity, metric_name, threshold_value, actual_value, triggered_at, message, metadata FROM ml_performance_alerts WHERE model_id = $1 AND status = 'active' ORDER BY triggered_at DESC"#, ) .bind(model_id) .fetch_all(conn.as_mut()) .await? } else { sqlx::query_as::< _, ( Uuid, Uuid, String, String, String, String, f64, f64, DateTime, String, serde_json::Value, ), >( r#"SELECT id, model_id, model_name, alert_type, severity, metric_name, threshold_value, actual_value, triggered_at, message, metadata FROM ml_performance_alerts WHERE status = 'active' ORDER BY triggered_at DESC"#, ) .fetch_all(conn.as_mut()) .await? }; let mut alerts = Vec::new(); for ( id, model_id, model_name, alert_type, severity, metric_name, threshold_value, actual_value, triggered_at, message, metadata, ) in rows { alerts.push(PerformanceAlert { id, model_id, model_name, alert_type: self.parse_alert_type(&alert_type)?, severity: self.parse_alert_severity(&severity)?, metric_name, threshold_value, actual_value, triggered_at, resolved_at: None, status: AlertStatus::Active, message, metadata, }); } Ok(alerts) } /// Check performance thresholds and create alerts async fn check_performance_threshold( &self, tx: &mut DatabaseTransaction, request: &RecordMetricsRequest, metric: &PerformanceMetric, ) -> Result<()> { // Check for degradation based on configuration threshold if let Some(threshold) = self.get_metric_threshold(&metric.name) { let degradation_detected = match metric.name.as_str() { "accuracy" | "precision" | "recall" | "f1_score" => { metric.value < threshold - self.config.degradation_threshold }, "latency_ms" | "error_rate" => { metric.value > threshold + self.config.degradation_threshold }, _ => false, }; if degradation_detected { let alert_id = Uuid::new_v4(); let message = format!( "Performance degradation detected for {}: {} (threshold: {})", metric.name, metric.value, threshold ); let query = format!( r#"INSERT INTO ml_performance_alerts (id, model_id, model_name, alert_type, severity, metric_name, threshold_value, actual_value, triggered_at, message) VALUES ('{}', '{}', '{}', 'degradation', 'medium', '{}', {}, {}, '{}', '{}')"#, alert_id, request.model_id, request.model_name, metric.name, threshold, metric.value, metric.timestamp.to_rfc3339(), message.replace("'", "''") ); tx.execute(&query).await?; tracing::warn!( "Performance alert triggered for model {} metric {}", request.model_name, metric.name ); } } Ok(()) } /// Get metric threshold (placeholder - would be configurable) fn get_metric_threshold(&self, metric_name: &str) -> Option { match metric_name { "accuracy" => Some(0.95), "precision" => Some(0.90), "recall" => Some(0.90), "f1_score" => Some(0.90), "latency_ms" => Some(100.0), "error_rate" => Some(0.01), _ => None, } } /// Calculate performance summary statistics fn calculate_performance_summary(&self, metrics: &[PerformanceMetric]) -> PerformanceSummary { let mut metric_groups: HashMap> = HashMap::new(); for metric in metrics { metric_groups .entry(metric.name.clone()) .or_default() .push(metric.value); } let mut summaries = HashMap::new(); for (name, values) in metric_groups { if !values.is_empty() { let sum: f64 = values.iter().sum(); let mean = sum / values.len() as f64; let min = values.iter().fold(f64::INFINITY, |a, &b| a.min(b)); let max = values.iter().fold(f64::NEG_INFINITY, |a, &b| a.max(b)); summaries.insert( name, MetricSummary { count: values.len(), mean, min, max, latest: values[0], // First value is latest due to DESC order }, ); } } PerformanceSummary { total_metrics: metrics.len(), metric_summaries: summaries, time_range: if let (Some(first), Some(last)) = (metrics.first(), metrics.last()) { Some((last.timestamp, first.timestamp)) } else { None }, } } /// Parse alert type from database fn parse_alert_type(&self, type_str: &str) -> Result { match type_str { "degradation" => Ok(AlertType::Degradation), "anomaly" => Ok(AlertType::Anomaly), "threshold" => Ok(AlertType::Threshold), "drift" => Ok(AlertType::Drift), _ => Err(MlDataError::Validation { message: format!("Invalid alert type: {}", type_str), }), } } /// Parse alert severity from database fn parse_alert_severity(&self, severity_str: &str) -> Result { match severity_str { "low" => Ok(AlertSeverity::Low), "medium" => Ok(AlertSeverity::Medium), "high" => Ok(AlertSeverity::High), "critical" => Ok(AlertSeverity::Critical), _ => Err(MlDataError::Validation { message: format!("Invalid alert severity: {}", severity_str), }), } } /// Health check for performance repository pub async fn health_check(&self) -> Result { let mut conn = self.db.acquire().await?; let _: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM ml_model_performance") .fetch_one(conn.as_mut()) .await?; Ok(true) } } /// Request to record performance metrics #[derive(Debug)] pub struct RecordMetricsRequest { pub model_id: Uuid, pub model_name: String, pub model_version: String, pub environment: String, pub metrics: Vec, } /// Individual performance metric #[derive(Debug, Clone, Serialize, Deserialize)] pub struct PerformanceMetric { pub timestamp: DateTime, pub name: String, pub value: f64, pub metadata: serde_json::Value, } /// Model performance data #[derive(Debug, Clone, Serialize, Deserialize)] pub struct ModelPerformance { pub model_id: Uuid, pub metrics: Vec, pub summary: PerformanceSummary, pub last_updated: Option>, } /// Performance summary statistics #[derive(Debug, Clone, Serialize, Deserialize)] pub struct PerformanceSummary { pub total_metrics: usize, pub metric_summaries: HashMap, pub time_range: Option<(DateTime, DateTime)>, } /// Summary statistics for a specific metric #[derive(Debug, Clone, Serialize, Deserialize)] pub struct MetricSummary { pub count: usize, pub mean: f64, pub min: f64, pub max: f64, pub latest: f64, } /// Performance benchmark request #[derive(Debug)] pub struct StartBenchmarkRequest { pub benchmark_name: String, pub model_id: Uuid, pub model_name: String, pub model_version: String, pub environment: String, pub started_at: DateTime, pub created_by: String, pub metadata: serde_json::Value, } /// Benchmark result #[derive(Debug, Clone, Serialize, Deserialize)] pub struct BenchmarkResult { pub id: Uuid, pub benchmark_name: String, pub model_id: Uuid, pub model_name: String, pub model_version: String, pub environment: String, pub started_at: DateTime, pub completed_at: Option>, pub duration_ms: Option, pub status: BenchmarkStatus, pub results: serde_json::Value, pub error_message: Option, pub created_by: String, pub metadata: serde_json::Value, } /// Benchmark status #[derive(Debug, Clone, Serialize, Deserialize)] pub enum BenchmarkStatus { Running, Completed, Failed, Cancelled, } impl ToString for BenchmarkStatus { fn to_string(&self) -> String { match self { BenchmarkStatus::Running => "running".to_string(), BenchmarkStatus::Completed => "completed".to_string(), BenchmarkStatus::Failed => "failed".to_string(), BenchmarkStatus::Cancelled => "cancelled".to_string(), } } } /// Create A/B experiment request #[derive(Debug)] pub struct CreateExperimentRequest { pub experiment_name: String, pub description: Option, pub control_model_id: Uuid, pub treatment_model_id: Uuid, pub started_at: DateTime, pub traffic_split: f64, pub confidence_level: f64, pub created_by: String, pub metadata: serde_json::Value, } /// A/B testing experiment #[derive(Debug, Clone, Serialize, Deserialize)] pub struct AbTestExperiment { pub id: Uuid, pub experiment_name: String, pub description: Option, pub control_model_id: Uuid, pub treatment_model_id: Uuid, pub started_at: DateTime, pub ended_at: Option>, pub status: ExperimentStatus, pub traffic_split: f64, pub confidence_level: f64, pub results: serde_json::Value, pub created_by: String, pub metadata: serde_json::Value, pub metrics: Vec, } /// Experiment status #[derive(Debug, Clone, Serialize, Deserialize)] pub enum ExperimentStatus { Draft, Running, Paused, Completed, Failed, } /// Experiment metric data #[derive(Debug, Clone, Serialize, Deserialize)] pub struct ExperimentMetric { pub variant: AbVariant, pub timestamp: DateTime, pub metric_name: String, pub metric_value: f64, pub sample_size: i32, pub metadata: serde_json::Value, } /// A/B test variant #[derive(Debug, Clone, Serialize, Deserialize)] pub enum AbVariant { Control, Treatment, } impl ToString for AbVariant { fn to_string(&self) -> String { match self { AbVariant::Control => "control".to_string(), AbVariant::Treatment => "treatment".to_string(), } } } /// Performance alert #[derive(Debug, Clone, Serialize, Deserialize)] pub struct PerformanceAlert { pub id: Uuid, pub model_id: Uuid, pub model_name: String, pub alert_type: AlertType, pub severity: AlertSeverity, pub metric_name: String, pub threshold_value: f64, pub actual_value: f64, pub triggered_at: DateTime, pub resolved_at: Option>, pub status: AlertStatus, pub message: String, pub metadata: serde_json::Value, } /// Alert type #[derive(Debug, Clone, Serialize, Deserialize)] pub enum AlertType { Degradation, Anomaly, Threshold, Drift, } /// Alert severity #[derive(Debug, Clone, Serialize, Deserialize)] pub enum AlertSeverity { Low, Medium, High, Critical, } /// Alert status #[derive(Debug, Clone, Serialize, Deserialize)] pub enum AlertStatus { Active, Acknowledged, Resolved, } // TECHNICAL DEBT ELIMINATED - Use HashMap> directly