#!/usr/bin/env python3 """ Performance Metrics and KPI Tracking System Comprehensive performance monitoring and KPI calculation framework """ import json import math import base64 import hashlib import datetime import re import string import itertools import collections import dataclasses import random from typing import Dict, List, Optional, Any, Union, Tuple from dataclasses import dataclass, field from collections import defaultdict from enum import Enum class KPIType(Enum): EFFICIENCY = "efficiency" FINANCIAL = "financial" OPERATIONAL = "operational" CUSTOMER_SATISFACTION = "customer_satisfaction" SAFETY = "safety" QUALITY = "quality" class MetricCategory(Enum): REALTIME = "realtime" CUMULATIVE = "cumulative" PERCENTAGE = "percentage" RATIO = "ratio" AVERAGE = "average" class TrendDirection(Enum): INCREASING = "increasing" DECREASING = "decreasing" STABLE = "stable" VOLATILE = "volatile" @dataclass class KPI: """Key Performance Indicator definition""" kpi_id: str name: str description: str kpi_type: KPIType category: MetricCategory unit: str target_value: Optional[float] = None min_acceptable: Optional[float] = None max_acceptable: Optional[float] = None weight: float = 1.0 formula: Optional[str] = None @dataclass class MetricValue: """Metric value with metadata""" metric_id: str value: float timestamp: datetime.datetime tags: Dict[str, str] = field(default_factory=dict) quality_score: float = 1.0 @dataclass class KPICalculation: """KPI calculation result""" kpi_id: str calculated_value: float target_achievement: float # Percentage of target achieved trend_direction: TrendDirection confidence_score: float calculation_timestamp: datetime.datetime = field(default_factory=datetime.datetime.now) contributing_metrics: List[str] = field(default_factory=list) class PerformanceMetricsTracker: """ Performance metrics and KPI tracking system Calculates, monitors, and reports on concert performance indicators """ def __init__(self): self.kpis = {} self.metric_values = defaultdict(list) self.kpi_history = defaultdict(list) self.thresholds = {} self.benchmarks = {} self.performance_score = 0.0 # Initialize core concert KPIs self._initialize_concert_kpis() def _initialize_concert_kpis(self): """Initialize core concert KPIs""" core_kpis = [ KPI("occupancy_rate", "Venue Occupancy Rate", "Percentage of venue capacity utilized", KPIType.OPERATIONAL, MetricCategory.PERCENTAGE, "%", target_value=85.0, min_acceptable=70.0), KPI("revenue_per_attendee", "Revenue Per Attendee", "Average revenue generated per concert attendee", KPIType.FINANCIAL, MetricCategory.AVERAGE, "$", target_value=75.0, min_acceptable=50.0), KPI("ticket_conversion_rate", "Ticket Conversion Rate", "Percentage of website visitors who purchase tickets", KPIType.EFFICIENCY, MetricCategory.PERCENTAGE, "%", target_value=15.0, min_acceptable=10.0), KPI("customer_satisfaction_score", "Customer Satisfaction Score", "Overall customer satisfaction rating", KPIType.CUSTOMER_SATISFACTION, MetricCategory.AVERAGE, "score", target_value=4.5, min_acceptable=4.0), KPI("incident_rate", "Safety Incident Rate", "Number of incidents per 1000 attendees", KPIType.SAFETY, MetricCategory.RATIO, "incidents/1k", target_value=2.0, max_acceptable=5.0), KPI("staff_efficiency", "Staff Efficiency Score", "Ratio of attendees to staff members", KPIType.OPERATIONAL, MetricCategory.RATIO, "attendees/staff", target_value=50.0, min_acceptable=30.0), KPI("audio_quality_score", "Audio Quality Score", "Technical audio system performance rating", KPIType.QUALITY, MetricCategory.AVERAGE, "score", target_value=0.9, min_acceptable=0.8), KPI("social_engagement_rate", "Social Media Engagement Rate", "Social media engagement per 1000 attendees", KPIType.CUSTOMER_SATISFACTION, MetricCategory.RATIO, "engagements/1k", target_value=100.0, min_acceptable=50.0), KPI("operational_cost_per_attendee", "Operational Cost Per Attendee", "Cost to serve each attendee", KPIType.FINANCIAL, MetricCategory.AVERAGE, "$", max_acceptable=25.0, target_value=15.0), KPI("entry_wait_time", "Average Entry Wait Time", "Average time attendees wait to enter venue", KPIType.OPERATIONAL, MetricCategory.AVERAGE, "minutes", max_acceptable=15.0, target_value=10.0) ] for kpi in core_kpis: self.kpis[kpi.kpi_id] = kpi def add_metric_value(self, metric_id: str, value: float, tags: Optional[Dict[str, str]] = None, quality_score: float = 1.0) -> bool: """Add a new metric value""" try: metric_value = MetricValue( metric_id=metric_id, value=value, timestamp=datetime.datetime.now(), tags=tags or {}, quality_score=quality_score ) self.metric_values[metric_id].append(metric_value) return True except Exception as e: return False def calculate_kpi(self, kpi_id: str, time_window: str = "1h") -> Optional[KPICalculation]: """Calculate KPI based on underlying metrics""" if kpi_id not in self.kpis: return None kpi = self.kpis[kpi_id] # Get relevant metrics for the time window relevant_metrics = self._get_relevant_metrics(kpi_id, time_window) if not relevant_metrics: return None # Calculate KPI value based on type calculated_value = self._calculate_kpi_value(kpi, relevant_metrics) # Calculate target achievement target_achievement = self._calculate_target_achievement(kpi, calculated_value) # Determine trend direction trend_direction = self._calculate_trend_direction(kpi_id, calculated_value) # Calculate confidence score based on data quality confidence_score = self._calculate_confidence_score(relevant_metrics) kpi_calc = KPICalculation( kpi_id=kpi_id, calculated_value=calculated_value, target_achievement=target_achievement, trend_direction=trend_direction, confidence_score=confidence_score, contributing_metrics=list(relevant_metrics.keys()) ) # Store in history self.kpi_history[kpi_id].append(kpi_calc) return kpi_calc def _get_relevant_metrics(self, kpi_id: str, time_window: str) -> Dict[str, List[MetricValue]]: """Get relevant metrics for KPI calculation""" relevant = defaultdict(list) # Define metric mappings for each KPI kpi_metrics = { "occupancy_rate": ["attendance_count", "venue_capacity"], "revenue_per_attendee": ["total_revenue", "attendance_count"], "ticket_conversion_rate": ["tickets_sold", "website_visitors"], "customer_satisfaction_score": ["satisfaction_ratings"], "incident_rate": ["incident_count", "attendance_count"], "staff_efficiency": ["attendance_count", "staff_count"], "audio_quality_score": ["audio_quality_ratings"], "social_engagement_rate": ["social_engagements", "attendance_count"], "operational_cost_per_attendee": ["operational_cost", "attendance_count"], "entry_wait_time": ["wait_times"] } if kpi_id in kpi_metrics: cutoff_time = self._parse_time_window(time_window) for metric_id in kpi_metrics[kpi_id]: if metric_id in self.metric_values: # Filter by time window recent_values = [ mv for mv in self.metric_values[metric_id] if mv.timestamp >= cutoff_time ] if recent_values: relevant[metric_id] = recent_values return dict(relevant) def _parse_time_window(self, time_window: str) -> datetime.datetime: """Parse time window string to datetime cutoff""" now = datetime.datetime.now() if time_window.endswith("h"): hours = int(time_window[:-1]) return now - datetime.timedelta(hours=hours) elif time_window.endswith("d"): days = int(time_window[:-1]) return now - datetime.timedelta(days=days) elif time_window.endswith("m"): minutes = int(time_window[:-1]) return now - datetime.timedelta(minutes=minutes) else: return now - datetime.timedelta(hours=1) # Default to 1 hour def _calculate_kpi_value(self, kpi: KPI, metrics: Dict[str, List[MetricValue]]) -> float: """Calculate KPI value from metrics""" if kpi.kpi_id == "occupancy_rate": attendance = sum(mv.value for mv in metrics.get("attendance_count", [])) capacity = sum(mv.value for mv in metrics.get("venue_capacity", [])) return (attendance / capacity * 100) if capacity > 0 else 0.0 elif kpi.kpi_id == "revenue_per_attendee": revenue = sum(mv.value for mv in metrics.get("total_revenue", [])) attendance = sum(mv.value for mv in metrics.get("attendance_count", [])) return (revenue / attendance) if attendance > 0 else 0.0 elif kpi.kpi_id == "ticket_conversion_rate": tickets = sum(mv.value for mv in metrics.get("tickets_sold", [])) visitors = sum(mv.value for mv in metrics.get("website_visitors", [])) return (tickets / visitors * 100) if visitors > 0 else 0.0 elif kpi.kpi_id == "customer_satisfaction_score": ratings = [mv.value for mv in metrics.get("satisfaction_ratings", [])] return sum(ratings) / len(ratings) if ratings else 0.0 elif kpi.kpi_id == "incident_rate": incidents = sum(mv.value for mv in metrics.get("incident_count", [])) attendance = sum(mv.value for mv in metrics.get("attendance_count", [])) return (incidents / attendance * 1000) if attendance > 0 else 0.0 elif kpi.kpi_id == "staff_efficiency": attendance = sum(mv.value for mv in metrics.get("attendance_count", [])) staff = sum(mv.value for mv in metrics.get("staff_count", [])) return (attendance / staff) if staff > 0 else 0.0 elif kpi.kpi_id == "audio_quality_score": ratings = [mv.value for mv in metrics.get("audio_quality_ratings", [])] return sum(ratings) / len(ratings) if ratings else 0.0 elif kpi.kpi_id == "social_engagement_rate": engagements = sum(mv.value for mv in metrics.get("social_engagements", [])) attendance = sum(mv.value for mv in metrics.get("attendance_count", [])) return (engagements / attendance * 1000) if attendance > 0 else 0.0 elif kpi.kpi_id == "operational_cost_per_attendee": cost = sum(mv.value for mv in metrics.get("operational_cost", [])) attendance = sum(mv.value for mv in metrics.get("attendance_count", [])) return (cost / attendance) if attendance > 0 else 0.0 elif kpi.kpi_id == "entry_wait_time": wait_times = [mv.value for mv in metrics.get("wait_times", [])] return sum(wait_times) / len(wait_times) if wait_times else 0.0 else: # Default to average if specific calculation not defined all_values = [] for metric_values in metrics.values(): all_values.extend([mv.value for mv in metric_values]) return sum(all_values) / len(all_values) if all_values else 0.0 def _calculate_target_achievement(self, kpi: KPI, value: float) -> float: """Calculate percentage of target achieved""" if kpi.target_value is None: return 100.0 # No target defined, assume 100% achievement if kpi.kpi_type == KPIType.SAFETY: # For safety metrics (lower is better), achievement is inverse if kpi.target_value and kpi.target_value > 0 and value > 0: achievement = max(0, min(100, (kpi.target_value / value) * 100)) else: achievement = 100.0 if value == 0 else 0.0 else: # For other metrics (higher is better) if kpi.target_value and kpi.target_value > 0: achievement = min(100, (value / kpi.target_value) * 100) else: achievement = 100.0 return achievement def _calculate_trend_direction(self, kpi_id: str, current_value: float) -> TrendDirection: """Calculate trend direction based on historical values""" history = self.kpi_history[kpi_id] if len(history) < 2: return TrendDirection.STABLE # Compare with recent values recent_values = [calc.calculated_value for calc in history[-5:]] recent_avg = sum(recent_values) / len(recent_values) # Calculate percentage change if recent_avg != 0: change_percent = ((current_value - recent_avg) / recent_avg) * 100 else: change_percent = 0.0 # Determine trend if abs(change_percent) < 5: return TrendDirection.STABLE elif change_percent > 10: return TrendDirection.INCREASING elif change_percent < -10: return TrendDirection.DECREASING else: return TrendDirection.VOLATILE def _calculate_confidence_score(self, metrics: Dict[str, List[MetricValue]]) -> float: """Calculate confidence score based on data quality and quantity""" if not metrics: return 0.0 total_quality_score = 0.0 total_values = 0 for metric_values in metrics.values(): for mv in metric_values: total_quality_score += mv.quality_score total_values += 1 if total_values == 0: return 0.0 base_confidence = total_quality_score / total_values # Adjust for data quantity (more data = higher confidence) data_factor = min(1.0, total_values / 10.0) # Normalize to max at 10 values return base_confidence * data_factor def calculate_overall_performance_score(self) -> Dict[str, Any]: """Calculate overall performance score across all KPIs""" kpi_scores = {} total_weighted_score = 0.0 total_weight = 0.0 for kpi_id, kpi in self.kpis.items(): latest_calc = self.calculate_kpi(kpi_id) if latest_calc: # Use target achievement as the score score = latest_calc.target_achievement weighted_score = score * kpi.weight kpi_scores[kpi_id] = { "name": kpi.name, "score": score, "weighted_score": weighted_score, "trend": latest_calc.trend_direction.value, "confidence": latest_calc.confidence_score } total_weighted_score += weighted_score total_weight += kpi.weight overall_score = (total_weighted_score / total_weight) if total_weight > 0 else 0.0 self.performance_score = overall_score # Categorize performance if overall_score >= 90: performance_level = "Excellent" elif overall_score >= 75: performance_level = "Good" elif overall_score >= 60: performance_level = "Average" elif overall_score >= 40: performance_level = "Below Average" else: performance_level = "Poor" return { "overall_score": overall_score, "performance_level": performance_level, "kpi_breakdown": kpi_scores, "total_kpis_evaluated": len(kpi_scores), "calculation_timestamp": datetime.datetime.now().isoformat() } def get_kpi_dashboard_data(self) -> Dict[str, Any]: """Get comprehensive KPI dashboard data""" dashboard_data = { "timestamp": datetime.datetime.now().isoformat(), "overall_performance": self.calculate_overall_performance_score(), "kpi_details": {}, "alerts": [], "trends": {} } # Get detailed KPI data for kpi_id, kpi in self.kpis.items(): latest_calc = self.calculate_kpi(kpi_id) if latest_calc: kpi_data = { "name": kpi.name, "current_value": latest_calc.calculated_value, "unit": kpi.unit, "target": kpi.target_value, "achievement": latest_calc.target_achievement, "trend": latest_calc.trend_direction.value, "confidence": latest_calc.confidence_score, "status": self._get_kpi_status(kpi, latest_calc) } dashboard_data["kpi_details"][kpi_id] = kpi_data # Add alerts for underperforming KPIs if latest_calc.target_achievement < 70: dashboard_data["alerts"].append({ "kpi_id": kpi_id, "name": kpi.name, "severity": "high" if latest_calc.target_achievement < 50 else "medium", "message": f"{kpi.name} is at {latest_calc.target_achievement:.1f}% of target" }) # Add trend data dashboard_data["trends"][kpi_id] = { "direction": latest_calc.trend_direction.value, "recent_values": [calc.calculated_value for calc in self.kpi_history[kpi_id][-10:]] } return dashboard_data def _get_kpi_status(self, kpi: KPI, calculation: KPICalculation) -> str: """Get KPI status based on performance""" achievement = calculation.target_achievement if achievement >= 100: return "excellent" elif achievement >= 90: return "good" elif achievement >= 75: return "acceptable" elif achievement >= 60: return "needs_improvement" else: return "critical" def simulate_concert_metrics(self, duration_minutes: int = 180) -> Dict[str, Any]: """Simulate realistic concert metrics for testing""" simulation_results = { "duration_minutes": duration_minutes, "metrics_generated": 0, "kpi_calculations": 0 } # Simulate base metrics over time for minute in range(duration_minutes): progress = minute / duration_minutes # Attendance builds up then stabilizes attendance = min(5000, progress * 6000 + random.uniform(-200, 200)) # Revenue correlates with attendance revenue = attendance * 75 + random.uniform(-5000, 5000) # Various other metrics wait_time = max(5, 15 - progress * 5 + random.uniform(-3, 3)) satisfaction = 4.0 + random.uniform(-0.5, 1.0) incidents = random.randint(0, 2) if random.random() < 0.1 else 0 # Add metric values self.add_metric_value("attendance_count", attendance) self.add_metric_value("total_revenue", revenue) self.add_metric_value("wait_times", wait_time) self.add_metric_value("satisfaction_ratings", satisfaction) self.add_metric_value("incident_count", incidents) simulation_results["metrics_generated"] += 5 # Calculate KPIs periodically if minute % 10 == 0: # Every 10 minutes for kpi_id in self.kpis.keys(): self.calculate_kpi(kpi_id, "30m") simulation_results["kpi_calculations"] += 1 return simulation_results def export_kpi_configuration(self) -> Dict[str, Any]: """Export KPI configuration and definitions""" config = { "kpis": {}, "performance_thresholds": self.thresholds, "benchmarks": self.benchmarks, "export_timestamp": datetime.datetime.now().isoformat() } for kpi_id, kpi in self.kpis.items(): config["kpis"][kpi_id] = { "kpi_id": kpi.kpi_id, "name": kpi.name, "description": kpi.description, "kpi_type": kpi.kpi_type.value, "category": kpi.category.value, "unit": kpi.unit, "target_value": kpi.target_value, "min_acceptable": kpi.min_acceptable, "max_acceptable": kpi.max_acceptable, "weight": kpi.weight, "formula": kpi.formula } return config if __name__ == "__main__": # Initialize performance tracker tracker = PerformanceMetricsTracker() print("=== Performance Metrics and KPI Tracking System ===") # Simulate concert data print("\nSimulating concert metrics...") simulation = tracker.simulate_concert_metrics(60) # 1 hour print(json.dumps(simulation, indent=2)) # Calculate all KPIs print("\n=== KPI Calculations ===") for kpi_id in tracker.kpis.keys(): calculation = tracker.calculate_kpi(kpi_id) if calculation: print(f"\n{kpi_id}:") print(f" Value: {calculation.calculated_value:.2f}") print(f" Target Achievement: {calculation.target_achievement:.1f}%") print(f" Trend: {calculation.trend_direction.value}") print(f" Confidence: {calculation.confidence_score:.2f}") # Get overall performance score print("\n=== Overall Performance ===") overall = tracker.calculate_overall_performance_score() print(json.dumps(overall, indent=2)) # Get dashboard data print("\n=== Dashboard Data ===") dashboard = tracker.get_kpi_dashboard_data() print(json.dumps(dashboard, indent=2)) # Export configuration print("\n=== KPI Configuration ===") config = tracker.export_kpi_configuration() print(json.dumps(config, indent=2))