""" Trend Analyzer for AegisLM Experiment Analysis. Provides comprehensive trend analysis capabilities including improvement rates, degradation detection, stability analysis, and performance forecasting. """ import uuid from typing import Dict, List, Any, Optional, Tuple from dataclasses import dataclass, field from datetime import datetime, timedelta from enum import Enum import logging import statistics from collections import defaultdict from experiments.experiment_manager import get_experiment_manager from schemas.experiment_schema import Experiment, ResultSummary, ExperimentStatus logger = logging.getLogger(__name__) class TrendDirection(str, Enum): """Trend direction enumeration.""" INCREASING = "increasing" DECREASING = "decreasing" STABLE = "stable" VOLATILE = "volatile" class TrendStrength(str, Enum): """Trend strength enumeration.""" STRONG = "strong" MODERATE = "moderate" WEAK = "weak" NONE = "none" @dataclass class TrendPoint: """Single data point in trend analysis.""" timestamp: datetime value: float run_id: str experiment_name: Optional[str] = None @dataclass class TrendMetrics: """Trend analysis metrics.""" direction: TrendDirection strength: TrendStrength slope: float # Rate of change correlation: float # Correlation coefficient volatility: float # Standard deviation improvement_rate: float # Percentage improvement over period stability_score: float # 0-1 stability score # Statistical measures mean: float median: float min_value: float max_value: float std_deviation: float # Trend-specific metrics momentum: float # Recent trend momentum acceleration: float # Rate of change of slope @dataclass class MetricTrend: """Trend analysis for a specific metric.""" metric_name: str data_points: List[TrendPoint] metrics: TrendMetrics forecast: Optional[List[Tuple[datetime, float]]] = None # (timestamp, predicted_value) anomalies: List[TrendPoint] = field(default_factory=list) # Outlier points # Change points significant_changes: List[Tuple[datetime, float]] = field(default_factory=list) @dataclass class TrendAnalysisResult: """Complete trend analysis result.""" run_ids: List[str] analysis_date: datetime time_period_days: int total_runs: int # Per-metric trends metric_trends: Dict[str, MetricTrend] = field(default_factory=dict) # Overall assessment overall_direction: TrendDirection = TrendDirection.STABLE overall_health_score: float = 0.0 # 0-1 overall health # Key insights key_insights: List[str] = field(default_factory=list) recommendations: List[str] = field(default_factory=list) warning_indicators: List[str] = field(default_factory=list) # Performance evolution improvement_summary: Dict[str, float] = field(default_factory=dict) degradation_summary: Dict[str, float] = field(default_factory=dict) # Visualization data chart_data: Dict[str, Any] = field(default_factory=dict) class TrendAnalyzer: """ Analyzer for experiment performance trends. Provides comprehensive trend analysis including direction detection, strength assessment, stability analysis, and forecasting. """ def __init__(self): """Initialize trend analyzer.""" self.experiment_manager = get_experiment_manager() async def analyze_trend(self, run_ids: List[str]) -> TrendAnalysisResult: """ Analyze performance trends across multiple runs. Args: run_ids: List of run IDs to analyze Returns: TrendAnalysisResult: Comprehensive trend analysis Raises: ValueError: If insufficient valid runs provided """ if len(run_ids) < 3: raise ValueError("At least 3 runs required for trend analysis") # Fetch experiment data experiments = await self._fetch_experiments(run_ids) if len(experiments) < 3: raise ValueError("Insufficient valid experiments for trend analysis") # Filter to completed experiments only completed_experiments = [ exp for exp in experiments if exp.status == ExperimentStatus.COMPLETED and exp.result_summary ] if len(completed_experiments) < 3: raise ValueError("At least 3 completed experiments required for trend analysis") # Sort by creation time completed_experiments.sort(key=lambda x: x.created_at) logger.info(f"Analyzing trends for {len(completed_experiments)} experiments") # Calculate time period time_period = completed_experiments[-1].created_at - completed_experiments[0].created_at time_period_days = max(1, time_period.days) # Create result result = TrendAnalysisResult( run_ids=[exp.run_id.hex for exp in completed_experiments], analysis_date=datetime.utcnow(), time_period_days=time_period_days, total_runs=len(completed_experiments) ) # Analyze trends for each metric metrics_to_analyze = [ 'robustness_score', 'risk_score', 'success_rate', 'confidence_score', 'execution_time_ms' ] for metric in metrics_to_analyze: trend = await self._analyze_metric_trend(completed_experiments, metric) if trend: result.metric_trends[metric] = trend # Generate overall assessment await self._generate_overall_assessment(result) # Generate insights and recommendations await self._generate_insights(result) # Create visualization data await self._create_chart_data(result) return result async def _fetch_experiments(self, run_ids: List[str]) -> List[Experiment]: """ Fetch experiments by run IDs. Args: run_ids: List of run IDs Returns: List[Experiment]: Valid experiments """ experiments = [] for run_id in run_ids: try: # Convert string to UUID if needed if isinstance(run_id, str): try: run_uuid = uuid.UUID(run_id) except ValueError: logger.warning(f"Invalid run ID format: {run_id}") continue else: run_uuid = run_id # Fetch experiment experiment = self.experiment_manager.store.get_experiment(run_uuid) if experiment: experiments.append(experiment) else: logger.warning(f"Experiment not found: {run_id}") except Exception as e: logger.error(f"Error fetching experiment {run_id}: {e}") continue return experiments async def _analyze_metric_trend(self, experiments: List[Experiment], metric_name: str) -> Optional[MetricTrend]: """ Analyze trend for a specific metric. Args: experiments: List of experiments to analyze metric_name: Name of metric to analyze Returns: MetricTrend: Trend analysis for the metric """ # Extract data points data_points = [] for exp in experiments: if exp.result_summary and hasattr(exp.result_summary, metric_name): value = getattr(exp.result_summary, metric_name) if value is not None: data_points.append(TrendPoint( timestamp=exp.created_at, value=float(value), run_id=exp.run_id.hex, experiment_name=exp.experiment_name )) if len(data_points) < 3: return None # Sort by timestamp data_points.sort(key=lambda x: x.timestamp) # Calculate trend metrics trend_metrics = await self._calculate_trend_metrics(data_points) # Detect anomalies anomalies = await self._detect_anomalies(data_points) # Find significant changes significant_changes = await self._find_change_points(data_points) # Generate forecast forecast = await self._generate_forecast(data_points, trend_metrics) return MetricTrend( metric_name=metric_name, data_points=data_points, metrics=trend_metrics, forecast=forecast, anomalies=anomalies, significant_changes=significant_changes ) async def _calculate_trend_metrics(self, data_points: List[TrendPoint]) -> TrendMetrics: """ Calculate comprehensive trend metrics. Args: data_points: List of data points Returns: TrendMetrics: Calculated metrics """ values = [point.value for point in data_points] timestamps = [point.timestamp.timestamp() for point in data_points] # Basic statistics mean_val = statistics.mean(values) median_val = statistics.median(values) min_val = min(values) max_val = max(values) std_dev = statistics.stdev(values) if len(values) > 1 else 0.0 # Calculate slope (linear regression) if len(timestamps) > 1: # Normalize timestamps to start from 0 normalized_times = [(t - timestamps[0]) for t in timestamps] n = len(normalized_times) # Calculate slope using linear regression sum_x = sum(normalized_times) sum_y = sum(values) sum_xy = sum(x * y for x, y in zip(normalized_times, values)) sum_x2 = sum(x * x for x in normalized_times) denominator = n * sum_x2 - sum_x * sum_x slope = (n * sum_xy - sum_x * sum_y) / denominator if denominator != 0 else 0.0 # Calculate correlation coefficient try: correlation = (n * sum_xy - sum_x * sum_y) / ( (n * sum_x2 - sum_x * sum_x) * (n * sum_y2 - sum_y * sum_y) ) ** 0.5 except: correlation = 0.0 else: slope = 0.0 correlation = 0.0 # Determine trend direction and strength direction, strength = await self._determine_trend_direction(slope, correlation, std_dev, mean_val) # Calculate improvement rate if len(values) >= 2: improvement_rate = ((values[-1] - values[0]) / values[0]) * 100 if values[0] != 0 else 0.0 else: improvement_rate = 0.0 # Calculate stability score (inverse of volatility) volatility = std_dev / mean_val if mean_val != 0 else 0.0 stability_score = max(0.0, 1.0 - volatility) # Calculate momentum (recent trend) if len(values) >= 3: recent_values = values[-3:] recent_slope = (recent_values[-1] - recent_values[0]) / 2 # Slope over last 3 points momentum = recent_slope / mean_val if mean_val != 0 else 0.0 else: momentum = 0.0 # Calculate acceleration (change in slope) if len(values) >= 4: mid_point = len(values) // 2 early_slope = (values[mid_point] - values[0]) / mid_point late_slope = (values[-1] - values[mid_point]) / (len(values) - mid_point - 1) acceleration = (late_slope - early_slope) / mean_val if mean_val != 0 else 0.0 else: acceleration = 0.0 return TrendMetrics( direction=direction, strength=strength, slope=slope, correlation=correlation, volatility=volatility, improvement_rate=improvement_rate, stability_score=stability_score, mean=mean_val, median=median_val, min_value=min_val, max_value=max_val, std_deviation=std_dev, momentum=momentum, acceleration=acceleration ) async def _determine_trend_direction( self, slope: float, correlation: float, std_dev: float, mean_val: float ) -> Tuple[TrendDirection, TrendStrength]: """ Determine trend direction and strength. Args: slope: Calculated slope correlation: Correlation coefficient std_dev: Standard deviation mean_val: Mean value Returns: Tuple[TrendDirection, TrendStrength]: Direction and strength """ # Determine direction based on slope if abs(slope) < 0.01: # Very small slope if std_dev / mean_val > 0.2: # High volatility direction = TrendDirection.VOLATILE else: direction = TrendDirection.STABLE elif slope > 0: direction = TrendDirection.INCREASING else: direction = TrendDirection.DECREASING # Determine strength based on correlation and slope magnitude correlation_strength = abs(correlation) slope_strength = abs(slope) / mean_val if mean_val != 0 else 0.0 combined_strength = (correlation_strength + slope_strength) / 2 if combined_strength > 0.8: strength = TrendStrength.STRONG elif combined_strength > 0.5: strength = TrendStrength.MODERATE elif combined_strength > 0.2: strength = TrendStrength.WEAK else: strength = TrendStrength.NONE return direction, strength async def _detect_anomalies(self, data_points: List[TrendPoint]) -> List[TrendPoint]: """ Detect anomalous data points using statistical methods. Args: data_points: List of data points Returns: List[TrendPoint]: Anomalous points """ if len(data_points) < 5: return [] values = [point.value for point in data_points] mean_val = statistics.mean(values) std_dev = statistics.stdev(values) # Points beyond 2 standard deviations are considered anomalies threshold = 2 * std_dev anomalies = [ point for point in data_points if abs(point.value - mean_val) > threshold ] return anomalies async def _find_change_points(self, data_points: List[TrendPoint]) -> List[Tuple[datetime, float]]: """ Find significant change points in the trend. Args: data_points: List of data points Returns: List[Tuple[datetime, float]]: Change points with magnitudes """ if len(data_points) < 5: return [] change_points = [] values = [point.value for point in data_points] # Use simple change point detection based on mean shifts window_size = max(3, len(values) // 4) for i in range(window_size, len(values) - window_size): # Calculate mean before and after point before_mean = statistics.mean(values[i-window_size:i]) after_mean = statistics.mean(values[i+1:i+window_size+1]) # Calculate change magnitude change_magnitude = abs(after_mean - before_mean) relative_change = change_magnitude / before_mean if before_mean != 0 else 0.0 # Mark as change point if significant (>20% change) if relative_change > 0.2: change_points.append((data_points[i].timestamp, relative_change)) return change_points async def _generate_forecast( self, data_points: List[TrendPoint], trend_metrics: TrendMetrics ) -> Optional[List[Tuple[datetime, float]]]: """ Generate simple forecast based on trend. Args: data_points: Historical data points trend_metrics: Calculated trend metrics Returns: List[Tuple[datetime, float]]: Forecast points """ if len(data_points) < 3 or trend_metrics.strength == TrendStrength.NONE: return None # Simple linear forecast last_timestamp = data_points[-1].timestamp last_value = data_points[-1].value forecast_points = [] forecast_days = 7 # Forecast 7 days ahead for day in range(1, forecast_days + 1): future_timestamp = last_timestamp + timedelta(days=day) # Use slope to predict future value predicted_value = last_value + (trend_metrics.slope * day * 86400) # Convert days to seconds forecast_points.append((future_timestamp, predicted_value)) return forecast_points async def _generate_overall_assessment(self, result: TrendAnalysisResult): """ Generate overall trend assessment. Args: result: Trend analysis result to update """ if not result.metric_trends: return # Calculate overall health score health_scores = [] for metric_name, trend in result.metric_trends.items(): # Higher health score for stable/increasing robustness/success # Lower health score for increasing risk if metric_name in ['robustness_score', 'success_rate', 'confidence_score']: if trend.metrics.direction == TrendDirection.INCREASING: health_scores.append(0.9) elif trend.metrics.direction == TrendDirection.STABLE: health_scores.append(0.7) else: health_scores.append(0.4) elif metric_name in ['risk_score', 'execution_time_ms']: if trend.metrics.direction == TrendDirection.DECREASING: health_scores.append(0.9) elif trend.metrics.direction == TrendDirection.STABLE: health_scores.append(0.7) else: health_scores.append(0.4) else: health_scores.append(0.6) # Neutral for other metrics result.overall_health_score = sum(health_scores) / len(health_scores) if health_scores else 0.5 # Determine overall direction directions = [trend.metrics.direction for trend in result.metric_trends.values()] direction_counts = defaultdict(int) for direction in directions: direction_counts[direction] += 1 if direction_counts[TrendDirection.INCREASING] > len(directions) / 2: result.overall_direction = TrendDirection.INCREASING elif direction_counts[TrendDirection.DECREASING] > len(directions) / 2: result.overall_direction = TrendDirection.DECREASING elif direction_counts[TrendDirection.VOLATILE] > len(directions) / 2: result.overall_direction = TrendDirection.VOLATILE else: result.overall_direction = TrendDirection.STABLE async def _generate_insights(self, result: TrendAnalysisResult): """ Generate insights and recommendations. Args: result: Trend analysis result to update """ for metric_name, trend in result.metric_trends.items(): metric_display = metric_name.replace('_', ' ').title() # Generate insights based on trend if trend.metrics.direction == TrendDirection.INCREASING: if metric_name in ['robustness_score', 'success_rate', 'confidence_score']: result.key_insights.append( f"{metric_display} is improving with {trend.metrics.improvement_rate:.1f}% increase" ) result.improvement_summary[metric_name] = trend.metrics.improvement_rate else: result.key_insights.append( f"{metric_display} is increasing by {trend.metrics.improvement_rate:.1f}% (may need attention)" ) result.degradation_summary[metric_name] = trend.metrics.improvement_rate elif trend.metrics.direction == TrendDirection.DECREASING: if metric_name in ['risk_score', 'execution_time_ms']: result.key_insights.append( f"{metric_display} is decreasing by {abs(trend.metrics.improvement_rate):.1f}% (good)" ) result.improvement_summary[metric_name] = abs(trend.metrics.improvement_rate) else: result.key_insights.append( f"{metric_display} is declining by {abs(trend.metrics.improvement_rate):.1f}% (concerning)" ) result.degradation_summary[metric_name] = abs(trend.metrics.improvement_rate) # Check for volatility if trend.metrics.direction == TrendDirection.VOLATILE: result.warning_indicators.append( f"{metric_display} shows high volatility (std dev: {trend.metrics.std_deviation:.3f})" ) # Check for low stability if trend.metrics.stability_score < 0.5: result.warning_indicators.append( f"{metric_display} has low stability score ({trend.metrics.stability_score:.2f})" ) # Generate recommendations if trend.metrics.strength == TrendStrength.STRONG: if metric_name == 'robustness_score' and trend.metrics.direction == TrendDirection.INCREASING: result.recommendations.append( f"Continue current approach for {metric_display} - showing strong improvement" ) elif metric_name == 'risk_score' and trend.metrics.direction == TrendDirection.INCREASING: result.recommendations.append( f"Urgent attention needed for {metric_display} - showing strong degradation" ) async def _create_chart_data(self, result: TrendAnalysisResult): """ Create visualization-ready chart data. Args: result: Trend analysis result to update """ # Line chart data for trends line_data = { 'metrics': {} } for metric_name, trend in result.metric_trends.items(): # Historical data historical = { 'timestamps': [point.timestamp.isoformat() for point in trend.data_points], 'values': [point.value for point in trend.data_points], 'labels': [point.experiment_name or point.run_id[:8] for point in trend.data_points] } # Forecast data forecast = None if trend.forecast: forecast = { 'timestamps': [point[0].isoformat() for point in trend.forecast], 'values': [point[1] for point in trend.forecast] } # Anomalies anomalies = { 'timestamps': [point.timestamp.isoformat() for point in trend.anomalies], 'values': [point.value for point in trend.anomalies], 'labels': [point.experiment_name or point.run_id[:8] for point in trend.anomalies] } line_data['metrics'][metric_name] = { 'historical': historical, 'forecast': forecast, 'anomalies': anomalies, 'trend_direction': trend.metrics.direction.value, 'trend_strength': trend.metrics.strength.value } # Summary chart data summary_data = { 'health_score': result.overall_health_score, 'overall_direction': result.overall_direction.value, 'improvements': result.improvement_summary, 'degradations': result.degradation_summary, 'warnings': len(result.warning_indicators) } result.chart_data = { 'trends': line_data, 'summary': summary_data } # Global trend analyzer instance trend_analyzer = TrendAnalyzer() async def get_trend_analyzer() -> TrendAnalyzer: """ Get the global trend analyzer instance. Returns: TrendAnalyzer: Global instance """ return trend_analyzer