Project KPI Dashboard
Business Case
Problem Statement
Project stakeholders struggle with:
- Scattered data across multiple systems
- Delayed reporting on project health
- No real-time visibility into KPIs
- Inconsistent metric definitions
Solution
Centralized KPI dashboard that aggregates data from multiple sources and presents key metrics with drill-down capabilities.
Business Value
- Real-time visibility - Live project health status
- Data-driven decisions - Actionable insights
- Stakeholder alignment - Single source of truth
- Early warning - Proactive issue detection
Technical Implementation
import pandas as pd
from datetime import datetime, date, timedelta
from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from enum import Enum
class KPIStatus(Enum):
"""KPI health status."""
ON_TRACK = "on_track"
AT_RISK = "at_risk"
CRITICAL = "critical"
UNKNOWN = "unknown"
class KPICategory(Enum):
"""KPI categories."""
SCHEDULE = "schedule"
COST = "cost"
QUALITY = "quality"
SAFETY = "safety"
PRODUCTIVITY = "productivity"
SUSTAINABILITY = "sustainability"
@dataclass
class KPIMetric:
"""Single KPI metric."""
name: str
category: KPICategory
current_value: float
target_value: float
unit: str
status: KPIStatus
trend: str # up, down, stable
last_updated: datetime
description: str = ""
@property
def variance(self) -> float:
"""Calculate variance from target."""
if self.target_value == 0:
return 0
return ((self.current_value - self.target_value) / self.target_value) * 100
@property
def achievement(self) -> float:
"""Calculate achievement percentage."""
if self.target_value == 0:
return 0
return (self.current_value / self.target_value) * 100
@dataclass
class DashboardConfig:
"""Dashboard configuration."""
project_name: str
project_code: str
start_date: date
end_date: date
budget: float
currency: str = "USD"
refresh_interval_minutes: int = 15
class ProjectKPIDashboard:
"""Construction project KPI dashboard."""
# Standard thresholds for RAG status
THRESHOLDS = {
'schedule': {'green': 0.95, 'amber': 0.85},
'cost': {'green': 1.05, 'amber': 1.15},
'quality': {'green': 0.98, 'amber': 0.95},
'safety': {'green': 0, 'amber': 1} # incident count
}
def __init__(self, config: DashboardConfig):
self.config = config
self.metrics: Dict[str, KPIMetric] = {}
self.history: List[Dict[str, Any]] = []
def add_metric(self, metric: KPIMetric):
"""Add or update a KPI metric."""
self.metrics[metric.name] = metric
self._record_history(metric)
def _record_history(self, metric: KPIMetric):
"""Record metric history for trending."""
self.history.append({
'name': metric.name,
'value': metric.current_value,
'timestamp': metric.last_updated,
'status': metric.status.value
})
def calculate_schedule_kpis(self,
planned_activities: int,
completed_activities: int,
planned_duration_days: int,
actual_duration_days: int) -> List[KPIMetric]:
"""Calculate schedule-related KPIs."""
# Schedule Performance Index (SPI)
spi = completed_activities / planned_activities if planned_activities > 0 else 0
spi_status = self._get_status(spi, 'schedule')
# Schedule Variance
sv = completed_activities - planned_activities
# Percent Complete
pct_complete = (completed_activities / planned_activities * 100) if planned_activities > 0 else 0
metrics = [
KPIMetric(
name="Schedule Performance Index",
category=KPICategory.SCHEDULE,
current_value=round(spi, 2),
target_value=1.0,
unit="ratio",
status=spi_status,
trend=self._calculate_trend("Schedule Performance Index"),
last_updated=datetime.now(),
description="SPI = Earned Value / Planned Value"
),
KPIMetric(
name="Percent Complete",
category=KPICategory.SCHEDULE,
current_value=round(pct_complete, 1),
target_value=100,
unit="%",
status=spi_status,
trend=self._calculate_trend("Percent Complete"),
last_updated=datetime.now()
),
KPIMetric(
name="Schedule Variance",
category=KPICategory.SCHEDULE,
current_value=sv,
target_value=0,
unit="activities",
status=spi_status,
trend=self._calculate_trend("Schedule Variance"),
last_updated=datetime.now()
)
]
for m in metrics:
self.add_metric(m)
return metrics
def calculate_cost_kpis(self,
budgeted_cost: float,
actual_cost: float,
earned_value: float) -> List[KPIMetric]:
"""Calculate cost-related KPIs."""
# Cost Performance Index (CPI)
cpi = earned_value / actual_cost if actual_cost > 0 else 0
cpi_status = self._get_status(cpi, 'cost', inverse=True)
# Cost Variance
cv = earned_value - actual_cost
# Budget utilization
budget_used = (actual_cost / budgeted_cost * 100) if budgeted_cost > 0 else 0
metrics = [
KPIMetric(
name="Cost Performance Index",
category=KPICategory.COST,
current_value=round(cpi, 2),
target_value=1.0,
unit="ratio",
status=cpi_status,
trend=self._calculate_trend("Cost Performance Index"),
last_updated=datetime.now(),
description="CPI = Earned Value / Actual Cost"
),
KPIMetric(
name="Cost Variance",
category=KPICategory.COST,
current_value=round(cv, 2),
target_value=0,
unit=self.config.currency,
status=cpi_status,
trend=self._calculate_trend("Cost Variance"),
last_updated=datetime.now()
),
KPIMetric(
name="Budget Utilization",
category=KPICategory.COST,
current_value=round(budget_used, 1),
target_value=100,
unit="%",
status=cpi_status,
trend=self._calculate_trend("Budget Utilization"),
last_updated=datetime.now()
)
]
for m in metrics:
self.add_metric(m)
return metrics
def calculate_quality_kpis(self,
total_inspections: int,
passed_inspections: int,
rework_items: int,
total_items: int) -> List[KPIMetric]:
"""Calculate quality-related KPIs."""
# First Pass Yield
fpy = passed_inspections / total_inspections if total_inspections > 0 else 0
fpy_status = self._get_status(fpy, 'quality')
# Rework Rate
rework_rate = rework_items / total_items * 100 if total_items > 0 else 0
metrics = [
KPIMetric(
name="First Pass Yield",
category=KPICategory.QUALITY,
current_value=round(fpy * 100, 1),
target_value=98,
unit="%",