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
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,