kpi-dashboard について
kpi-dashboardは、建設プロジェクト向けの統合KPIダッシュボードを作成するスキルです。データ収集の自動化、可視化、アラート、経営層向けレポートに対応します。
主な機能
- 01KPIデータの自動収集
- 02プロジェクト指標の可視化
- 03HTMLダッシュボードの生成
- 04カスタムKPIの追加
- 05KPI詳細の確認
活用シーン
- 01複数のプロジェクト指標を一元管理
- 02レポート用HTMLダッシュボードの作成
- 03プロジェクト固有の持続可能性KPIの定義
インストール
コマンドをコピーしてターミナルで実行します。
$
npx skills add https:/インストール後、Claude Code で/kpi-dashboardを入力して読み込まれたか確認します。
同梱ファイルclaw.jsoninstructions.md
詳細
KPI ダッシュボードビルダー
ビジネスケース
問題点
プロジェクトのモニタリングにおける課題：
- 複数の指標を追跡する必要がある
- さまざまなソースからデータを取得する
- リアルタイムで状況を把握する必要がある
- 経営陣向けのレポート作成
解決策
建設プロジェクト向けの統合 KPI ダッシュボードシステムです。データ収集、可視化、アラート通知を自動化します。
技術的な実装
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,
'target': kpi.target,
'formula': kpi.formula
},
'current': {
'value': history[-1].value if history else None,
'status': history[-1].status.value if history else None,
'trend': history[-1].trend.value if history else None
},
'history': [
{'date': h.date.isoformat(), 'value': h.value, 'status': h.status.value}
for h in history
]
}
def generate_html_dashboard(self) -> str:
"""Generate HTML dashboard."""
summary = self.get_dashboard_summary()
current = self.get_current_values()
html = f"""
<!DOCTYPE html>
<html>
<head>
<title>KPI Dashboard - {self.project_name}</title>
<style>
body {{ font-family: Arial, sans-serif; margin: 20px; }}
.header {{ background: #2196F3; color: white; padding: 20px; margin-bottom: 20px; }}
.kpi-grid {{ display: grid; grid-template-columns: repeat(4, 1fr); gap: 15px; }}
.kpi-card {{ border: 1px solid #ddd; padding: 15px; border-radius: 5px; }}
.on_target {{ border-left: 4px solid #4CAF50; }}
.at_risk {{ border-left: 4px solid #FF9800; }}
.critical {{ border-left: 4px solid #F44336; }}
.kpi-value {{ font-size: 24px; font-weight: bold; }}
.kpi-name {{ color: #666; font-size: 14px; }}
</style>
</head>
<body>
<div class="header">
<h1>{self.project_name} - KPI Dashboard</h1>
<p>Last updated: {summary['date']}</p>
</div>
<div class="kpi-grid">
"""
for kpi_id, value in current.items():
kpi = self.kpis[kpi_id]
html += f"""
<div class="kpi-card {value.status.value}">
<div class="kpi-name">{kpi.name}</div>
<div class="kpi-value">{value.value:.2f} {kpi.unit}</div>
<div>Target: {kpi.target} | Trend: {value.trend.value}</div>
</div>
"""
html += "</div></body></html>"
return html
def export_to_excel(self, output_path: str) -> str:
"""Export dashboard to Excel."""
with pd.ExcelWriter(output_path, engine='openpyxl') as writer:
# Summary
summary = self.get_dashboard_summary()
summary_df = pd.DataFrame([{
'Project': summary['project'],
'Date': summary['date'],
'On Target': summary['by_status']['on_target'],
'At Risk': summary['by_status']['at_risk'],
'Critical': summary['by_status']['critical']
}])
summary_df.to_excel(writer, sheet_name='Summary', index=False)
# Current values
current = self.get_current_values()
current_data = []
for kpi_id, value in current.items():
kpi = self.kpis[kpi_id]
current_data.append({
'KPI': kpi.name,
'Category': kpi.category.value,
'Value': value.value,
'Unit': kpi.unit,
'Target': kpi.target,
'Status': value.status.value,
'Trend': value.trend.value
})
current_df = pd.DataFrame(current_data)
current_df.to_excel(writer, sheet_name='Current KPIs', index=False)
return output_path
クイックスタート
# Create dashboard
dashboard = KPIDashboard("Office Building A")
# Record KPI values
dashboard.record_value("CPI", 0.95)
dashboard.record_value("SPI", 1.02)
dashboard.record_value("DEFECT_RATE", 3.5)
dashboard.record_value("TRIR", 1.8)
dashboard.record_value("LABOR_PROD", 92)
# Get summary
summary = dashboard.get_dashboard_summary()
print(f"On Target: {summary['by_status']['on_target']}")
print(f"Critical: {summary['by_status']['critical']}")
一般的なユースケース
1. HTML ダッシュボード
html = dashboard.generate_html_dashboard()
with open("dashboard.html", "w") as f:
f.write(html)
2. KPI の詳細
details = dashboard.get_kpi_details("CPI")
print(f"Current CPI: {details['current']['value']}")
3. カスタム KPI
dashboard.add_custom_kpi(KPIDefinition(
"WASTE_DIVERSION", "Waste Diversion Rate",
KPICategory.SUSTAINABILITY, "%", 75, 60, 50, True
))
リソース
- DDC Book：第 4.1 章 - データ分析と意思決定
- Web サイト：https://datadrivenconstruction.io
コメント
Markdown 対応