┌─────────────────────────────────────────────────────────────────┐
│                  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_incidents * 200000 / total_hours) if total_hours > 0 else 0

        # Health distribution
        green = len([p for p in active if p.health == HealthStatus.GREEN])
        yellow = len([p for p in active if p.health == HealthStatus.YELLOW])
        red = len([p for p in active if p.health == HealthStatus.RED])

        # Trends (compare to previous snapshot if available)
        cost_trend = "stable"
        schedule_trend = "stable"

        if self.snapshots:
            prev = self.snapshots[-1]
            if avg_cpi > prev.get('avg_cpi', 1.0):
                cost_trend = "improving"
            elif avg_cpi < prev.get('avg_cpi', 1.0):
                cost_trend = "declining"

            if avg_spi > prev.get('avg_spi', 1.0):
                schedule_trend = "improving"
            elif avg_spi < prev.get('avg_spi', 1.0):
                schedule_trend = "declining"

        return PortfolioSummary(
            report_date=datetime.now(),
            total_projects=len(all_projects),
            active_projects=len(active),
            total_contract_value=total_contract,
            total_budget=total_budget,
            total_actual_cost=total_actual,
            total_forecast_cost=total_forecast,
            avg_cpi=avg_cpi,
            avg_spi=avg_spi,
            on_budget_pct=on_budget_pct,
            on_schedule_pct=on_schedule_pct,
            portfolio_trir=portfolio_trir,
            total_incidents=total_incidents,
            green_count=green,
            yellow_count=yellow,
            red_count=red,
            cost_trend=cost_trend,
            schedule_trend=schedule_trend
        )

    def compare_projects(self, metric: str) -> ProjectComparison:
        """Compare projects by specific metric."""
        active = self.get_active_projects()

        if not active:
            return None

        metric_map = {
            "cpi": lambda p: p.cpi,
            "spi": lambda p: p.spi,
            "percent_complete": lambda p: p.percent_complete,
            "cost_variance": lambda p: p.cost_variance,
            "trir": lambda p: p.trir,
            "risk_score": lambda p: p.risk_score
        }

        if metric not in metric_map:
            raise ValueError(f"Unknown metric: {metric}")

        getter = metric_map[metric]
        values = {p.project_name: getter(p) for p in active}

        avg = statistics.mean(values.values())

        # Best/worst depends on metric (higher CPI good, lower TRIR good)
        if metric in ["trir", "risk_score"]:
            best = min(values.items(), key=lambda x: x[1])
            worst = max(values.items(), key=lambda x: x[1])
        else:
            best = max(values.items(), key=lambda x: x[1])
            worst = min(values.items(), key=lambda x: x[1])

        return ProjectComparison(
            metric=metric,
            projects=values,
            avg=avg,
            best=best,
            worst=worst
        )

    def get_projects_at_risk(self) -> List[ProjectMetrics]:
        """Get projects that need attention."""
        return [p for p in self.get_active_projects()
                if p.health in [HealthStatus.YELLOW, HealthStatus.RED]]

    def get_top_risks(self, limit: int = 10) -> List[Dict]:
        """Get top risks across portfolio."""
        risks = []

        for p in self.get_active_projects():
            if p.risk_score > 0:
                risks.append({
                    "project": p.project_name,
                    "risk_score": p.risk_score,
                    "critical_risks": p.critical_risks,
                    "cpi": p.cpi,
                    "spi": p.spi
                })

        return sorted(risks, key=lambda x: -x['risk_score'])[:limit]

    def forecast_cash_needs(self, months: int = 6) -> List[Dict]:
        """Forecast cash needs across portfolio."""
        forecasts = []

        for month in range(1, months + 1):
            month_date = datetime.now() + timedelta(days=month * 30)

            month_spend = 0
            for p in self.get_active_projects():
                # Simple linear projection based on remaining work
                remaining = p.forecast_cost - p.actual_cost
                months_remaining = max(1, (p.forecast_end - datetime.now()).days / 30)
                monthly_burn = remaining / months_remaining
                month_spend += monthly_burn

            forecasts.append({
                "month": month_date.strftime("%Y-%m"),
                "projected_spend": month_spend
            })

        return forecasts

    def save_snapshot(self):
        """Save current state for trend analysis."""
        summary = self.calculate_portfolio_summary()
        if summary:
            self.snapshots.append({
                "date": datetime.now(),
                "avg_cpi": summary.avg_cpi,
                "avg_spi": summary.avg_spi,
                "on_budget_pct": summary.on_budget_pct,
                "on_schedule_pct": summary.on_schedule_pct,
                "total_forecast": summary.total_forecast_cost
            })

    def generate_report(self) -> str:
        """Generate portfolio dashboard report."""
        summary = self.calculate_portfolio_summary()

        if not summary:
            return "No projects in portfolio"

        lines = [
            "# Portfolio Dashboard",
            "",
            f"**Portfolio:** {self.portfolio_name}",
            f"**Report Date:** {summary.report_date.strftime('%Y-%m-%d')}",
            "",
            "## Executive Summary",
            "",
            f"| Metric | Value |",
            f"|--------|-------|",
            f"| Total Projects | {summary.total_projects} ({summary.active_projects} active) |",
            f"| Total Contract Value | ${summary.total_contract_value:,.0f} |",
            f"| Total Budget | ${summary.total_budget:,.0f} |",
            f"| Actual Cost to Date | ${summary.total_actual_cost:,.0f} |",
            f"| Forecast at Completion | ${summary.total_forecast_cost:,.0f} |",
            "",
            "## Performance Indicators",
            "",
            f"| KPI | Value | Trend |",
            f"|-----|-------|-------|",
            f"| Avg CPI | {summary.avg_cpi:.2f} | {summary.cost_trend} |",
            f"| Avg SPI | {summary.avg_spi:.2f} | {summary.schedule_trend} |",
            f"| On Budget | {summary.on_budget_pct:.0f}% | |",
            f"| On Schedule | {summary.on_schedule_pct:.0f}% | |",
            f"| Portfolio TRIR | {summary.portfolio_trir:.2f} | |",
            "",
            "## Health Distribution",
            "",
            f"🟢 Green: {summary.green_count} | 🟡 Yellow: {summary.yellow_count} | 🔴 Red: {summary.red_count}",
            ""
        ]

        # Projects at risk
        at_risk = self.get_projects_at_risk()
        if at_risk:
            lines.extend([
                "## Projects Requiring Attention",
                "",
                "| Project | Health | CPI | SPI | Critical Risks |",
                "|---------|--------|-----|-----|----------------|"
            ])
            for p in sorted(at_risk, key=lambda x: x.cpi):
                health_icon = "🟡" if p.health == HealthStatus.YELLOW else "🔴"
                lines.append(
                    f"| {p.project_name} | {health_icon} | {p.cpi:.2f} | {p.spi:.2f} | {p.critical_risks} |"
                )
            lines.append("")

        # Project comparison
        lines.extend([
            "## Project Comparison - CPI",
            "",
            "| Project | CPI |",
            "|---------|-----|"
        ])

        cpi_compare = self.compare_projects("cpi")
        if cpi_compare:
            for name, value in sorted(cpi_compare.projects.items(), key=lambda x: -x[1]):
                lines.append(f"| {name} | {value:.2f} |")

        return "\n".join(lines)

Quick Start

from datetime import datetime, timedelta

# Initialize dashboard
dashboard = PortfolioDashboard("Regional Construction Portfolio")

# Import project data
projects = [
    {
        "id": "PRJ-001",
        "name": "Downtown Office Tower",
        "status": "active",
        "contract_value": 50000000,
        "budget": 48000000,
        "actual_cost": 25000000,
        "forecast_cost": 49000000,
        "percent_complete": 55,
        "planned_start": datetime(2024, 1, 1),
        "planned_end": datetime(2025, 6, 30),
        "forecast_end": datetime(2025, 7, 15),
        "cpi": 0.92,
        "spi": 0.95,
        "incidents": 2,
        "total_hours": 150000,
        "risk_score": 7.5,
        "critical_risks": 2
    },
    {
        "id": "PRJ-002",
        "name": "Hospital Expansion",
        "status": "active",
        "contract_value": 80000000,
        "budget": 75000000,
        "actual_cost": 30000000,
        "forecast_cost": 74000000,
        "percent_complete": 40,
        "planned_start": datetime(2024, 3, 1),
        "planned_end": datetime(2026, 2, 28),
        "forecast_end": datetime(2026, 2, 28),
        "cpi": 1.02,
        "spi": 1.00,
        "incidents": 0,
        "total_hours": 100000,
        "risk_score": 4.0,
        "critical_risks": 0
    }
]

dashboard.import_projects(projects)

# Get portfolio summary
summary = dashboard.calculate_portfolio_summary()
print(f"Portfolio Value: ${summary.total_contract_value:,.0f}")
print(f"Avg CPI: {summary.avg_cpi:.2f}")
print(f"On Budget: {summary.on_budget_pct:.0f}%")

# Find projects at risk
at_risk = dashboard.get_projects_at_risk()
print(f"Projects at risk: {len(at_risk)}")

# Compare projects
cpi_comparison = dashboard.compare_projects("cpi")
print(f"Best CPI: {cpi_comparison.best[0]} ({cpi_comparison.best[1]:.2f})")

# Generate report
print(dashboard.generate_report())

Requirements

pip install (no external dependencies)

portfolio-dashboard

Portfolio Dashboard

Overview

Portfolio Analytics Framework