Portfolio Dashboard

Overview

Aggregate and analyze data across multiple construction projects for portfolio-level visibility. Track KPIs, identify trends, compare project performance, and support strategic resource allocation decisions.

Portfolio Analytics Framework

┌─────────────────────────────────────────────────────────────────┐
│                  PORTFOLIO DASHBOARD                             │
├─────────────────────────────────────────────────────────────────┤
│                                                                  │
│  PROJECT A    PROJECT B    PROJECT C    PROJECT D               │
│     ↓             ↓            ↓            ↓                   │
│  ┌─────────────────────────────────────────────┐                │
│  │           DATA AGGREGATION                   │                │
│  │  Cost | Schedule | Safety | Quality | Risk   │                │
│  └─────────────────────────────────────────────┘                │
│                         ↓                                        │
│  ┌─────────────────────────────────────────────┐                │
│  │          PORTFOLIO KPIs                      │                │
│  │  📊 Total Value    📈 On-Schedule %          │                │
│  │  💰 On-Budget %    🛡️ Safety Rate            │                │
│  │  ⚠️ Risk Score     📋 Resource Util          │                │
│  └─────────────────────────────────────────────┘                │
│                                                                  │
└─────────────────────────────────────────────────────────────────┘

Technical Implementation

from dataclasses import dataclass, field
from typing import List, Dict, Optional, Tuple
from datetime import datetime, timedelta
from enum import Enum
import statistics

class ProjectStatus(Enum):
    PLANNING = "planning"
    ACTIVE = "active"
    ON_HOLD = "on_hold"
    COMPLETE = "complete"
    CANCELLED = "cancelled"

class HealthStatus(Enum):
    GREEN = "green"       # On track
    YELLOW = "yellow"     # At risk
    RED = "red"           # Critical
    GREY = "grey"         # Not started/on hold

@dataclass
class ProjectMetrics:
    project_id: str
    project_name: str
    status: ProjectStatus
    contract_value: float
    percent_complete: float

    # Schedule
    planned_start: datetime
    planned_end: datetime
    actual_start: Optional[datetime]
    forecast_end: datetime
    schedule_variance_days: int = 0

    # Cost
    budget: float
    actual_cost: float
    forecast_cost: float
    cost_variance: float = 0.0
    cpi: float = 1.0
    spi: float = 1.0

    # Safety
    recordable_incidents: int = 0
    total_hours: float = 0
    trir: float = 0.0

    # Quality
    defects_open: int = 0
    rework_cost: float = 0.0

    # Risk
    risk_score: float = 0.0
    critical_risks: int = 0

    @property
    def health(self) -> HealthStatus:
        """Determine overall project health."""
        if self.status in [ProjectStatus.ON_HOLD, ProjectStatus.CANCELLED]:
            return HealthStatus.GREY

        # Critical if significantly over budget/schedule
        if self.cpi < 0.85 or self.spi < 0.85 or self.critical_risks > 3:
            return HealthStatus.RED

        # At risk if moderately off track
        if self.cpi < 0.95 or self.spi < 0.95 or self.critical_risks > 0:
            return HealthStatus.YELLOW

        return HealthStatus.GREEN

@dataclass
class PortfolioSummary:
    report_date: datetime
    total_projects: int
    active_projects: int
    total_contract_value: float
    total_budget: float
    total_actual_cost: float
    total_forecast_cost: float

    # Performance
    avg_cpi: float
    avg_spi: float
    on_budget_pct: float
    on_schedule_pct: float

    # Safety
    portfolio_trir: float
    total_incidents: int

    # Health distribution
    green_count: int
    yellow_count: int
    red_count: int

    # Trends
    cost_trend: str
    schedule_trend: str

@dataclass
class ProjectComparison:
    metric: str
    projects: Dict[str, float]
    avg: float
    best: Tuple[str, float]
    worst: Tuple[str, float]

class PortfolioDashboard:
    """Multi-project portfolio analytics."""

    # Health thresholds
    THRESHOLDS = {
        "cpi_warning": 0.95,
        "cpi_critical": 0.85,
        "spi_warning": 0.95,
        "spi_critical": 0.85,
        "trir_warning": 2.0,
        "risk_score_warning": 7.0
    }

    def __init__(self, portfolio_name: str):
        self.portfolio_name = portfolio_name
        self.projects: Dict[str, ProjectMetrics] = {}
        self.snapshots: List[Dict] = []  # Historical data

    def add_project(self, metrics: ProjectMetrics):
        """Add or update project in portfolio."""
        self.projects[metrics.project_id] = metrics

    def import_projects(self, projects_data: List[Dict]) -> int:
        """Import multiple projects from data."""
        count = 0
        for p in projects_data:
            metrics = ProjectMetrics(
                project_id=p['id'],
                project_name=p['name'],
                status=ProjectStatus(p.get('status', 'active')),
                contract_value=p['contract_value'],
                percent_complete=p.get('percent_complete', 0),
                planned_start=p['planned_start'],
                planned_end=p['planned_end'],
                actual_start=p.get('actual_start'),
                forecast_end=p.get('forecast_end', p['planned_end']),
                budget=p['budget'],
                actual_cost=p.get('actual_cost', 0),
                forecast_cost=p.get('forecast_cost', p['budget']),
                cpi=p.get('cpi', 1.0),
                spi=p.get('spi', 1.0),
                recordable_incidents=p.get('incidents', 0),
                total_hours=p.get('total_hours', 0),
                risk_score=p.get('risk_score', 0),
                critical_risks=p.get('critical_risks', 0)
            )

            # Calculate derived metrics
            metrics.cost_variance = metrics.budget - metrics.actual_cost
            metrics.schedule_variance_days = (metrics.planned_end - metrics.forecast_end).days

            if metrics.total_hours > 0:
                metrics.trir = (metrics.recordable_incidents * 200000) / metrics.total_hours

            self.add_project(metrics)
            count += 1

        return count

    def get_active_projects(self) -> List[ProjectMetrics]:
        """Get list of active projects."""
        return [p for p in self.projects.values()
                if p.status == ProjectStatus.ACTIVE]

    def calculate_portfolio_summary(self) -> PortfolioSummary:
        """Calculate portfolio-level summary metrics."""
        active = self.get_active_projects()
        all_projects = list(self.projects.values())

        if not all_projects:
            return None

        # Totals
        total_contract = sum(p.contract_value for p in all_projects)
        total_budget = sum(p.budget for p in all_projects)
        total_actual = sum(p.actual_cost for p in all_projects)
        total_forecast = sum(p.forecast_cost for p in all_projects)

        # Performance averages (weighted by budget)
        if total_budget > 0:
            avg_cpi = sum(p.cpi * p.budget for p in active) / sum(p.budget for p in active) if active else 1.0
            avg_spi = sum(p.spi * p.budget for p in active) / sum(p.budget for p in active) if active else 1.0
        else:
            avg_cpi = avg_spi = 1.0

        # On budget/schedule percentages
        on_budget = len([p for p in active if p.cpi >= 0.95])
        on_schedule = len([p for p in active if p.spi >= 0.95])

        on_budget_pct = (on_budget / len(active) * 100) if active else 100
        on_schedule_pct = (on_schedule / len(active) * 100) if active else 100

        # Safety metrics
        total_incidents = sum(p.recordable_incidents for p in all_projects)
        total_hours = sum(p.total_hours for p in all_projects)
        portfolio_trir = (total_inciden