| name | portfolio-dashboard |
| description | Multi-project portfolio analytics dashboard. Aggregate KPIs across projects, track portfolio health, compare performance, and support executive decision-making. |
| homepage | https://datadrivenconstruction.io |
| metadata | {"openclaw":{"emoji":"🚀","os":["darwin","linux","win32"],"homepage":"https://datadrivenconstruction.io","requires":{"bins":"[Truncated]"}}} |
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"
YELLOW = "yellow"
RED = "red"
GREY = "grey"
@dataclass
class ProjectMetrics:
project_id: str
project_name: str
status: ProjectStatus
contract_value: float
percent_complete: float
planned_start: datetime
planned_end: datetime
actual_start: Optional[datetime]
forecast_end: datetime
schedule_variance_days: int = 0
budget: float
actual_cost: float
forecast_cost: float
cost_variance: =
cpi: =
spi: =
recordable_incidents: =
total_hours: =
trir: =
defects_open: =
rework_cost: =
risk_score: =
critical_risks: =
() -> HealthStatus:
.status [ProjectStatus.ON_HOLD, ProjectStatus.CANCELLED]:
HealthStatus.GREY
.cpi < .spi < .critical_risks > :
HealthStatus.RED
.cpi < .spi < .critical_risks > :
HealthStatus.YELLOW
HealthStatus.GREEN
:
report_date: datetime
total_projects:
active_projects:
total_contract_value:
total_budget:
total_actual_cost:
total_forecast_cost:
avg_cpi:
avg_spi:
on_budget_pct:
on_schedule_pct:
portfolio_trir:
total_incidents:
green_count:
yellow_count:
red_count:
cost_trend:
schedule_trend:
:
metric:
projects: [, ]
avg:
best: [, ]
worst: [, ]
:
THRESHOLDS = {
: ,
: ,
: ,
: ,
: ,
:
}
():
.portfolio_name = portfolio_name
.projects: [, ProjectMetrics] = {}
.snapshots: [] = []
():
.projects[metrics.project_id] = metrics
() -> :
count =
p projects_data:
metrics = ProjectMetrics(
project_id=p[],
project_name=p[],
status=ProjectStatus(p.get(, )),
contract_value=p[],
percent_complete=p.get(, ),
planned_start=p[],
planned_end=p[],
actual_start=p.get(),
forecast_end=p.get(, p[]),
budget=p[],
actual_cost=p.get(, ),
forecast_cost=p.get(, p[]),
cpi=p.get(, ),
spi=p.get(, ),
recordable_incidents=p.get(, ),
total_hours=p.get(, ),
risk_score=p.get(, ),
critical_risks=p.get(, )
)
metrics.cost_variance = metrics.budget - metrics.actual_cost
metrics.schedule_variance_days = (metrics.planned_end - metrics.forecast_end).days
metrics.total_hours > :
metrics.trir = (metrics.recordable_incidents * ) / metrics.total_hours
.add_project(metrics)
count +=
count
() -> [ProjectMetrics]:
[p p .projects.values()
p.status == ProjectStatus.ACTIVE]
() -> PortfolioSummary:
active = .get_active_projects()
all_projects = (.projects.values())
all_projects:
total_contract = (p.contract_value p all_projects)
total_budget = (p.budget p all_projects)
total_actual = (p.actual_cost p all_projects)
total_forecast = (p.forecast_cost p all_projects)
total_budget > :
avg_cpi = (p.cpi * p.budget p active) / (p.budget p active) active
avg_spi = (p.spi * p.budget p active) / (p.budget p active) active
:
avg_cpi = avg_spi =
on_budget = ([p p active p.cpi >= ])
on_schedule = ([p p active p.spi >= ])
on_budget_pct = (on_budget / (active) * ) active
on_schedule_pct = (on_schedule / (active) * ) active
total_incidents = (p.recordable_incidents p all_projects)
total_hours = (p.total_hours p all_projects)
portfolio_trir = (total_incidents * / total_hours) total_hours >
green = ([p p active p.health == HealthStatus.GREEN])
yellow = ([p p active p.health == HealthStatus.YELLOW])
red = ([p p active p.health == HealthStatus.RED])
cost_trend =
schedule_trend =
.snapshots:
prev = .snapshots[-]
avg_cpi > prev.get(, ):
cost_trend =
avg_cpi < prev.get(, ):
cost_trend =
avg_spi > prev.get(, ):
schedule_trend =
avg_spi < prev.get(, ):
schedule_trend =
PortfolioSummary(
report_date=datetime.now(),
total_projects=(all_projects),
active_projects=(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
)
() -> ProjectComparison:
active = .get_active_projects()
active:
metric_map = {
: p: p.cpi,
: p: p.spi,
: p: p.percent_complete,
: p: p.cost_variance,
: p: p.trir,
: p: p.risk_score
}
metric metric_map:
ValueError()
getter = metric_map[metric]
values = {p.project_name: getter(p) p active}
avg = statistics.mean(values.values())
metric [, ]:
best = (values.items(), key= x: x[])
worst = (values.items(), key= x: x[])
:
best = (values.items(), key= x: x[])
worst = (values.items(), key= x: x[])
ProjectComparison(
metric=metric,
projects=values,
avg=avg,
best=best,
worst=worst
)
() -> [ProjectMetrics]:
[p p .get_active_projects()
p.health [HealthStatus.YELLOW, HealthStatus.RED]]
() -> []:
risks = []
p .get_active_projects():
p.risk_score > :
risks.append({
: p.project_name,
: p.risk_score,
: p.critical_risks,
: p.cpi,
: p.spi
})
(risks, key= x: -x[])[:limit]
() -> []:
forecasts = []
month (, months + ):
month_date = datetime.now() + timedelta(days=month * )
month_spend =
p .get_active_projects():
remaining = p.forecast_cost - p.actual_cost
months_remaining = (, (p.forecast_end - datetime.now()).days / )
monthly_burn = remaining / months_remaining
month_spend += monthly_burn
forecasts.append({
: month_date.strftime(),
: month_spend
})
forecasts
():
summary = .calculate_portfolio_summary()
summary:
.snapshots.append({
: datetime.now(),
: summary.avg_cpi,
: summary.avg_spi,
: summary.on_budget_pct,
: summary.on_schedule_pct,
: summary.total_forecast_cost
})
() -> :
summary = .calculate_portfolio_summary()
summary:
lines = [
,
,
,
,
,
,
,
,
,
,
,
,
,
,
,
,
,
,
,
,
,
,
,
,
,
,
,
,
]
at_risk = .get_projects_at_risk()
at_risk:
lines.extend([
,
,
,
])
p (at_risk, key= x: x.cpi):
health_icon = p.health == HealthStatus.YELLOW
lines.append(
)
lines.append()
lines.extend([
,
,
,
])
cpi_compare = .compare_projects()
cpi_compare:
name, value (cpi_compare.projects.items(), key= x: -x[]):
lines.append()
.join(lines)
Quick Start
from datetime import datetime, timedelta
dashboard = PortfolioDashboard("Regional Construction Portfolio")
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": ,
: ,
: ,
: ,
: ,
: ,
: datetime(, , ),
: datetime(, , ),
: datetime(, , ),
: ,
: ,
: ,
: ,
: ,
:
}
]
dashboard.import_projects(projects)
summary = dashboard.calculate_portfolio_summary()
()
()
()
at_risk = dashboard.get_projects_at_risk()
()
cpi_comparison = dashboard.compare_projects()
()
(dashboard.generate_report())
Requirements
pip install (no external dependencies)