""" Episode grading module for the Cloud Resource Allocation RL environment. This module implements the EpisodeGrader class that evaluates complete episode trajectories based on stability, efficiency, and overall performance. The grader calculates a composite score and determines pass/fail status against thresholds. """ import numpy as np class EpisodeGrader: """ Evaluates agent performance across complete episodes. Metrics: - Average reward (overall performance) - Stability score (utilization variance) - Efficiency score (resource usage patterns) The grader combines these metrics into a weighted composite score and determines whether the episode meets minimum performance thresholds. """ def __init__( self, pass_threshold: float = 0.0, stability_weight: float = 0.3, efficiency_weight: float = 0.3, reward_weight: float = 0.4 ): """ Initialize grader with scoring weights and thresholds. Args: pass_threshold: Minimum score to pass evaluation stability_weight: Weight for stability component (0-1) efficiency_weight: Weight for efficiency component (0-1) reward_weight: Weight for cumulative reward component (0-1) """ self.pass_threshold = pass_threshold self.stability_weight = stability_weight self.efficiency_weight = efficiency_weight self.reward_weight = reward_weight def grade_episode( self, states: list[np.ndarray], actions: list[int], rewards: list[float] ) -> dict: """ Evaluate a complete episode trajectory. The grader analyzes the full trajectory to assess: 1. Stability: How consistent were utilization levels? 2. Efficiency: How well were resources managed? 3. Performance: What was the average reward? Args: states: List of state observations throughout episode actions: List of actions taken throughout episode rewards: List of rewards received throughout episode Returns: Dictionary containing: - score: Overall performance score - passed: Boolean pass/fail status - stability_score: System stability metric - efficiency_score: Resource efficiency metric - avg_reward: Average reward per step - avg_cpu: Average CPU utilization - avg_memory: Average memory utilization - avg_latency: Average latency - final_score: Alias for score (compatibility) Raises: ValueError: If trajectory data is empty or mismatched lengths """ # Validate inputs if not states or not actions or not rewards: raise ValueError("Trajectory data cannot be empty") if not (len(states) == len(actions) == len(rewards)): raise ValueError( f"Mismatched trajectory lengths: states={len(states)}, " f"actions={len(actions)}, rewards={len(rewards)}" ) # Calculate component metrics stability_score = self._calculate_stability(states) efficiency_score = self._calculate_efficiency(states) avg_reward = float(np.mean(rewards)) # Calculate average CPU and memory utilization cpu_utils = [state[0] for state in states] memory_utils = [state[1] for state in states] avg_cpu = float(np.mean(cpu_utils)) avg_memory = float(np.mean(memory_utils)) # Calculate average latency from states # Latency = request_rate / allocated_resources request_rates = [state[2] for state in states] allocated_resources = [state[3] for state in states] latencies = [req / max(res, 1) for req, res in zip(request_rates, allocated_resources)] avg_latency = float(np.mean(latencies)) # Normalize average reward to [0, 1] range for scoring # Assuming rewards typically range from -5 to +2 based on reward module # Map this to [0, 1] where 0 is worst (-5) and 1 is best (+2) normalized_reward = (avg_reward + 5.0) / 7.0 normalized_reward = max(0.0, min(1.0, normalized_reward)) # Calculate weighted composite score score = ( self.stability_weight * stability_score + self.efficiency_weight * efficiency_score + self.reward_weight * normalized_reward ) # Determine pass/fail status passed = score >= self.pass_threshold return { 'score': score, 'passed': passed, 'stability_score': stability_score, 'efficiency_score': efficiency_score, 'avg_reward': avg_reward, 'avg_cpu': avg_cpu, 'avg_memory': avg_memory, 'avg_latency': avg_latency, 'final_score': score # Alias for compatibility } def _calculate_stability(self, states: list[np.ndarray]) -> float: """ Calculate stability score based on utilization variance. Stability is measured by how consistent CPU and memory utilization levels are throughout the episode. Lower variance indicates better stability and receives a higher score. Args: states: List of state observations throughout episode Returns: Stability score in range [0, 1] where 1 is most stable """ # Extract CPU and memory utilization from states # State format: [cpu_util, memory_util, request_rate, allocated_resources] cpu_utils = [state[0] for state in states] memory_utils = [state[1] for state in states] # Calculate variance for both metrics cpu_variance = float(np.var(cpu_utils)) memory_variance = float(np.var(memory_utils)) # Average variance avg_variance = (cpu_variance + memory_variance) / 2.0 # Convert variance to stability score [0, 1] # Lower variance = higher stability # Use exponential decay: score = exp(-variance / scale) # Scale factor of 100 means variance of 100 gives score ~0.37 stability_score = np.exp(-avg_variance / 100.0) return float(stability_score) def _calculate_efficiency(self, states: list[np.ndarray]) -> float: """ Calculate efficiency score based on resource allocation patterns. Efficiency is measured by how well resources are utilized. Lower average resource allocation (while maintaining acceptable utilization) indicates better efficiency and receives a higher score. Args: states: List of state observations throughout episode Returns: Efficiency score in range [0, 1] where 1 is most efficient """ # Extract allocated resources from states # State format: [cpu_util, memory_util, request_rate, allocated_resources] allocated_resources = [state[3] for state in states] # Calculate average resource allocation avg_resources = float(np.mean(allocated_resources)) # Convert to efficiency score [0, 1] # Lower average resources = higher efficiency # Assume typical range is 1-10 instances # Map: 1 instance -> 1.0, 10 instances -> 0.1 # Use formula: score = 1 / avg_resources, capped at [0, 1] efficiency_score = 1.0 / avg_resources efficiency_score = min(1.0, efficiency_score) return float(efficiency_score)