kpi-dashboard

What this skill does

# KPI Dashboard Builder

## Business Case

### Problem Statement
Project monitoring challenges:
- Multiple metrics to track
- Data from various sources
- Real-time visibility needed
- Executive reporting

### Solution
Unified KPI dashboard system for construction projects with automated data collection, visualization, and alerting.

## Technical Implementation

```python
import pandas as pd
from typing import Dict, Any, List, Optional, Callable
from dataclasses import dataclass, field
from datetime import date, datetime
from enum import Enum


class KPICategory(Enum):
    COST = "cost"
    SCHEDULE = "schedule"
    QUALITY = "quality"
    SAFETY = "safety"
    PRODUCTIVITY = "productivity"
    SUSTAINABILITY = "sustainability"


class KPIStatus(Enum):
    ON_TARGET = "on_target"
    AT_RISK = "at_risk"
    CRITICAL = "critical"


class TrendDirection(Enum):
    IMPROVING = "improving"
    STABLE = "stable"
    DECLINING = "declining"


@dataclass
class KPIDefinition:
    kpi_id: str
    name: str
    category: KPICategory
    unit: str
    target: float
    warning_threshold: float
    critical_threshold: float
    higher_is_better: bool = True
    formula: str = ""


@dataclass
class KPIValue:
    kpi_id: str
    value: float
    date: date
    status: KPIStatus
    trend: TrendDirection


class KPIDashboard:
    """Build and manage KPI dashboards for construction projects."""

    def __init__(self, project_name: str):
        self.project_name = project_name
        self.kpis: Dict[str, KPIDefinition] = {}
        self.history: Dict[str, List[KPIValue]] = {}
        self._define_standard_kpis()

    def _define_standard_kpis(self):
        """Define standard construction KPIs."""

        standard_kpis = [
            # Cost KPIs
            KPIDefinition("CPI", "Cost Performance Index", KPICategory.COST,
                         "ratio", 1.0, 0.95, 0.90, True, "BCWP / ACWP"),
            KPIDefinition("CV", "Cost Variance", KPICategory.COST,
                         "$", 0, -50000, -100000, True, "BCWP - ACWP"),
            KPIDefinition("BUDGET_USED", "Budget Utilization", KPICategory.COST,
                         "%", 100, 105, 110, False),

            # Schedule KPIs
            KPIDefinition("SPI", "Schedule Performance Index", KPICategory.SCHEDULE,
                         "ratio", 1.0, 0.95, 0.90, True, "BCWP / BCWS"),
            KPIDefinition("SV", "Schedule Variance", KPICategory.SCHEDULE,
                         "days", 0, -7, -14, True),
            KPIDefinition("COMPLETION", "Project Completion", KPICategory.SCHEDULE,
                         "%", 100, 95, 90, True),

            # Quality KPIs
            KPIDefinition("DEFECT_RATE", "Defect Rate", KPICategory.QUALITY,
                         "per 1000 units", 0, 5, 10, False),
            KPIDefinition("FIRST_PASS", "First Pass Yield", KPICategory.QUALITY,
                         "%", 95, 90, 85, True),
            KPIDefinition("REWORK", "Rework Percentage", KPICategory.QUALITY,
                         "%", 0, 3, 5, False),

            # Safety KPIs
            KPIDefinition("TRIR", "Total Recordable Incident Rate", KPICategory.SAFETY,
                         "per 200k hours", 0, 2, 4, False),
            KPIDefinition("LOST_DAYS", "Lost Time Injuries", KPICategory.SAFETY,
                         "incidents", 0, 1, 3, False),
            KPIDefinition("SAFETY_OBSERVATIONS", "Safety Observations", KPICategory.SAFETY,
                         "count", 50, 30, 20, True),

            # Productivity KPIs
            KPIDefinition("LABOR_PROD", "Labor Productivity", KPICategory.PRODUCTIVITY,
                         "%", 100, 90, 80, True),
            KPIDefinition("EQUIP_UTIL", "Equipment Utilization", KPICategory.PRODUCTIVITY,
                         "%", 85, 70, 60, True),
        ]

        for kpi in standard_kpis:
            self.kpis[kpi.kpi_id] = kpi
            self.history[kpi.kpi_id] = []

    def add_custom_kpi(self, kpi: KPIDefinition):
        """Add custom KPI definition."""
        self.kpis[kpi.kpi_id] = kpi
        self.history[kpi.kpi_id] = []

    def record_value(self, kpi_id: str, value: float, record_date: date = None):
        """Record KPI value."""

        if kpi_id not in self.kpis:
            return

        kpi = self.kpis[kpi_id]
        record_date = record_date or date.today()

        # Calculate status
        status = self._calculate_status(kpi, value)

        # Calculate trend
        trend = self._calculate_trend(kpi_id, value)

        kpi_value = KPIValue(
            kpi_id=kpi_id,
            value=value,
            date=record_date,
            status=status,
            trend=trend
        )

        self.history[kpi_id].append(kpi_value)

    def _calculate_status(self, kpi: KPIDefinition, value: float) -> KPIStatus:
        """Calculate KPI status based on thresholds."""

        if kpi.higher_is_better:
            if value >= kpi.target:
                return KPIStatus.ON_TARGET
            elif value >= kpi.warning_threshold:
                return KPIStatus.AT_RISK
            else:
                return KPIStatus.CRITICAL
        else:
            if value <= kpi.target:
                return KPIStatus.ON_TARGET
            elif value <= kpi.warning_threshold:
                return KPIStatus.AT_RISK
            else:
                return KPIStatus.CRITICAL

    def _calculate_trend(self, kpi_id: str, current_value: float) -> TrendDirection:
        """Calculate trend direction."""

        history = self.history.get(kpi_id, [])
        if len(history) < 2:
            return TrendDirection.STABLE

        # Compare with average of last 3 values
        recent_values = [h.value for h in history[-3:]]
        avg = sum(recent_values) / len(recent_values)

        kpi = self.kpis[kpi_id]
        diff = current_value - avg

        if abs(diff) < avg * 0.05:  # Within 5%
            return TrendDirection.STABLE
        elif (diff > 0 and kpi.higher_is_better) or (diff < 0 and not kpi.higher_is_better):
            return TrendDirection.IMPROVING
        else:
            return TrendDirection.DECLINING

    def get_current_values(self) -> Dict[str, KPIValue]:
        """Get most recent value for each KPI."""

        current = {}
        for kpi_id, history in self.history.items():
            if history:
                current[kpi_id] = history[-1]
        return current

    def get_dashboard_summary(self) -> Dict[str, Any]:
        """Get dashboard summary."""

        current = self.get_current_values()

        summary = {
            'project': self.project_name,
            'date': date.today().isoformat(),
            'total_kpis': len(self.kpis),
            'by_status': {s.value: 0 for s in KPIStatus},
            'by_category': {},
            'alerts': []
        }

        for kpi_id, value in current.items():
            summary['by_status'][value.status.value] += 1

            category = self.kpis[kpi_id].category.value
            if category not in summary['by_category']:
                summary['by_category'][category] = {'on_target': 0, 'at_risk': 0, 'critical': 0}
            summary['by_category'][category][value.status.value] += 1

            if value.status == KPIStatus.CRITICAL:
                summary['alerts'].append({
                    'kpi': self.kpis[kpi_id].name,
                    'value': value.value,
                    'target': self.kpis[kpi_id].target,
                    'status': 'critical'
                })

        return summary

    def get_kpi_details(self, kpi_id: str) -> Dict[str, Any]:
        """Get detailed KPI information."""

        if kpi_id not in self.kpis:
            return {}

        kpi = self.kpis[kpi_id]
        history = self.history.get(kpi_id, [])

        return {
            'definition': {
                'id': kpi.kpi_id,
                'name': kpi.name,
                'category': kpi.category.value,
                'unit': kpi.unit,