Imported from majiayu000/claude-skill-registry (skills/data/enterprise-risk-aggregator/SKILL.md). Install upstream with npx skills add majiayu000/claude-skill-registry --skill enterprise-risk-aggregator. Copyright stays with the author.

Enterprise Risk Aggregator

Overview

Aggregate individual project risks into a portfolio-level view. Identify correlated risks across projects, calculate enterprise risk exposure, and develop portfolio-wide mitigation strategies.

Risk Aggregation Framework

┌─────────────────────────────────────────────────────────────────┐
│                ENTERPRISE RISK AGGREGATION                       │
├─────────────────────────────────────────────────────────────────┤
│                                                                  │
│  PROJECT RISKS           CORRELATION            PORTFOLIO VIEW  │
│  ─────────────           ───────────            ──────────────  │
│                                                                  │
│  Project A:              Market risks ←→        Total Exposure: │
│  • Material cost ↗       affect all             $45M            │
│  • Labor shortage        projects               ─────────────── │
│                                    ↓            Risk Categories:│
│  Project B:              Weather impacts        • Market: 35%   │
│  • Weather delay         multiple sites         • Schedule: 25% │
│  • Permit issue                    ↓            • Safety: 15%   │
│                          Supply chain           • Regulatory:15%│
│  Project C:              affects                • Technical:10% │
│  • Subcontractor ↗       entire region          ─────────────── │
│  • Design change                                Top 5 Risks:    │
│                                                 1. Steel prices │
│                                                 2. Labor market │
│                                                 3. Supply chain │
│                                                                  │
└─────────────────────────────────────────────────────────────────┘

Technical Implementation

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

class RiskCategory(Enum):
    MARKET = "market"
    SCHEDULE = "schedule"
    SAFETY = "safety"
    REGULATORY = "regulatory"
    TECHNICAL = "technical"
    FINANCIAL = "financial"
    ENVIRONMENTAL = "environmental"
    SUPPLY_CHAIN = "supply_chain"
    LABOR = "labor"
    WEATHER = "weather"

class RiskLevel(Enum):
    LOW = 1
    MEDIUM = 2
    HIGH = 3
    CRITICAL = 4

class CorrelationType(Enum):
    POSITIVE = "positive"      # Risks tend to occur together
    NEGATIVE = "negative"      # One risk may offset another
    INDEPENDENT = "independent"

@dataclass
class ProjectRisk:
    id: str
    project_id: str
    project_name: str
    category: RiskCategory
    description: str
    probability: float  # 0-1
    impact: float       # Dollar amount
    score: float = 0.0  # P x I
    level: RiskLevel = RiskLevel.MEDIUM
    status: str = "open"
    mitigation: str = ""
    triggers: List[str] = field(default_factory=list)

    def __post_init__(self):
        self.score = self.probability * self.impact
        if self.score > 5000000:
            self.level = RiskLevel.CRITICAL
        elif self.score > 1000000:
            self.level = RiskLevel.HIGH
        elif self.score > 250000:
            self.level = RiskLevel.MEDIUM
        else:
            self.level = RiskLevel.LOW

@dataclass
class RiskCorrelation:
    risk1_id: str
    risk2_id: str
    correlation_type: CorrelationType
    strength: float  # 0-1
    shared_triggers: List[str]
    notes: str = ""

@dataclass
class AggregatedRisk:
    category: RiskCategory
    total_exposure: float
    expected_loss: float
    worst_case: float
    risk_count: int
    projects_affected: int
    mitigation_cost: float
    residual_exposure: float

@dataclass
class PortfolioRiskProfile:
    report_date: datetime
    total_projects: int
    total_risks: int
    total_exposure: float
    expected_loss: float
    var_95: float  # Value at Risk at 95% confidence
    by_category: Dict[str, AggregatedRisk]
    top_risks: List[ProjectRisk]
    correlations: List[RiskCorrelation]
    systemic_risks: List[str]

class EnterpriseRiskAggregator:
    """Aggregate risks across project portfolio."""

    # Common triggers that create correlation
    SYSTEMIC_TRIGGERS = [
        "steel_price_increase",
        "labor_shortage",
        "supply_chain_disruption",
        "interest_rate_change",
        "regulatory_change",
        "weather_event",
        "economic_downturn",
        "pandemic",
        "trade_restrictions"
    ]

    def __init__(self, portfolio_name: str):
        self.portfolio_name = portfolio_name
        self.risks: Dict[str, ProjectRisk] = {}
        self.correlations: List[RiskCorrelation] = []
        self.projects: Set[str] = set()

    def add_risk(self, project_id: str, project_name: str,
                category: RiskCategory, description: str,
                probability: float, impact: float,
                triggers: List[str] = None,
                mitigation: str = "") -> ProjectRisk:
        """Add project risk to portfolio."""
        risk_id = f"RISK-{project_id}-{len(self.risks)+1:04d}"

        risk = ProjectRisk(
            id=risk_id,
            project_id=project_id,
            project_name=project_name,
            category=category,
            description=description,
            probability=probability,
            impact=impact,
            triggers=triggers or [],
            mitigation=mitigation
        )

        self.risks[risk_id] = risk
        self.projects.add(project_id)

        return risk

    def import_project_risks(self, project_id: str, project_name: str,
                            risks: List[Dict]) -> int:
        """Import risks from project risk register."""
        count = 0
        for r in risks:
            self.add_risk(
                project_id=project_id,
                project_name=project_name,
                category=RiskCategory(r['category']),
                description=r['description'],
                probability=r['probability'],
                impact=r['impact'],
                triggers=r.get('triggers', []),
                mitigation=r.get('mitigation', '')
            )
            count += 1
        return count

    def detect_correlations(self) -> List[RiskCorrelation]:
        """Automatically detect correlated risks."""
        self.correlations = []
        risks = list(self.risks.values())

        for i, risk1 in enumerate(risks):
            for risk2 in risks[i+1:]:
                # Check for shared triggers
                shared = set(risk1.triggers) & set(risk2.triggers)

                if shared:
                    # Calculate correlation strength
                    total_triggers = len(set(risk1.triggers) | set(risk2.triggers))
                    strength = len(shared) / total_triggers if total_triggers > 0 else 0

                    correlation = RiskCorrelation(
                        risk1_id=risk1.id,
                        risk2_id=risk2.id,
                        correlation_type=CorrelationType.POSITIVE,
                        strength=strength,
                        shared_triggers=list(shared)
                    )
                    self.correlations.append(correlation)

                # Check for same category across projects
                elif (risk1.category == risk2.category and
                      risk1.project_id != risk2.project_id):
                    correlation = RiskCorrelation(
                        risk1_id=risk1.id,
                        risk2_id=risk2.id,
                        correlation_type=CorrelationType.POSITIVE,
                        strength=0.3,  # Weak assumed correlation
                        shared_triggers=[],
                        notes=f"Same category: {risk1.category.value}"
                    )
                    self.correlations.append(correlation)

        return self.correlations

    def identify_systemic_risks(self) -> List[Dict]:
        """Identify systemic risks affecting multiple projects."""
        systemic = []

        # Count triggers across all risks
        trigger_count: Dict[str, Set[str]] = {}
        for risk in self.risks.values():
            for trigger in risk.triggers:
                if trigger not in trigger_count:
                    trigger_count[trigger] = set()
                trigger_count[trigger].add(risk.project_id)

        # Systemic if affects multiple projects
        for trigger, projects in trigger_count.items():
            if len(projects) > 1:
                # Calculate total exposure
                affected_risks = [r for r in self.risks.values()
                                 if trigger in r.triggers]
                total_exposure = sum(r.score for r in affected_risks)

                systemic.append({
                    "trigger": trigger,
                    "projects_affected": len(projects),
                    "risks_affected": len(affected_risks),
                    "total_exposure": total_exposure,
                    "is_systemic": trigger in self.SYSTEMIC_TRIGGERS
                })

        return sorted(systemic, key=lambda x: -x['total_exposure'])

    def aggregate_by_category(self) -> Dict[RiskCategory, AggregatedRisk]:
        """Aggregate risks by category."""
        by_category = {}

        for category in RiskCategory:
            cat_risks = [r for r in self.risks.values() if r.category == category]

            if not cat_risks:
                continue

            projects = set(r.project_id for r in cat_risks)

            # Simple aggregation (no correlation adjustment)
            total_exposure = sum(r.impact for r in cat_risks)
            expected_loss = sum(r.score for r in cat_risks)

            # Worst case assuming all materialize
            worst_case = total_exposure

            by_category[category] = AggregatedRisk(
                category=category,
                total_exposure=total_exposure,
                expected_loss=expected_loss,
                worst_case=worst_case,
                risk_count=len(cat_risks),
                projects_affected=len(projects),
                mitigation_cost=0,
                residual_exposure=expected_loss
            )

        return by_category

    def calculate_var(self, confidence: float = 0.95,
                     simulations: int = 10000) -> float:
        """Calculate Value at Risk using Monte Carlo simulation."""
        import random

        losses = []
        risks = list(self.risks.values())

        for _ in range(simulations):
            sim_loss = 0
            for risk in risks:
                if random.random() < risk.probability:
                    sim_loss += risk.impact
            losses.append(sim_loss)

        losses.sort()
        var_index = int(simulations * confidence)
        return losses[var_index]

    def generate_portfolio_profile(self) -> PortfolioRiskProfile:
        """Generate comprehensive portfolio risk profile."""
        if not self.correlations:
            self.detect_correlations()

        total_exposure = sum(r.impact for r in self.risks.values())
        expected_loss = sum(r.score for r in self.risks.values())

        by_category = self.aggregate_by_category()

        # Top risks by score
        top_risks = sorted(self.risks.values(), key=lambda x: -x.score)[:10]

        # Systemic risks
        systemic = self.identify_systemic_risks()
        systemic_triggers = [s['trigger'] for s in systemic if s['is_systemic']]

        # VaR calculation
        var_95 = self.calculate_var(0.95)

        return PortfolioRiskProfile(
            report_date=datetime.now(),
            total_projects=len(self.projects),
            total_risks=len(self.risks),
            total_exposure=total_exposure,
            expected_loss=expected_loss,
            var_95=var_95,
            by_category={k.value: v for k, v in by_category.items()},
            top_risks=top_risks,
            correlations=self.correlations,
            systemic_risks=systemic_triggers
        )

    def suggest_mitigation_priorities(self) -> List[Dict]:
        """Suggest prioritized mitigation actions."""
        priorities = []

        # Systemic risks first
        systemic = self.identify_systemic_risks()
        for s in systemic[:5]:
            if s['is_systemic']:
                priorities.append({
                    "priority": 1,
                    "type": "systemic",
                    "target": s['trigger'],
                    "exposure": s['total_exposure'],
                    "projects": s['projects_affected'],
                    "recommendation": f"Portfolio-wide mitigation for {s['trigger']}"
                })

        # High-correlation risks
        high_corr = [c for c in self.correlations if c.strength > 0.5]
        for corr in high_corr[:5]:
            r1 = self.risks.get(corr.risk1_id)
            r2 = self.risks.get(corr.risk2_id)
            if r1 and r2:
                priorities.append({
                    "priority": 2,
                    "type": "correlated",
                    "target": f"{r1.description[:30]} / {r2.description[:30]}",
                    "exposure": r1.score + r2.score,
                    "projects": 2,
                    "recommendation": f"Joint mitigation via {corr.shared_triggers}"
                })

        # Individual high-impact risks
        top_risks = sorted(self.risks.values(), key=lambda x: -x.score)[:10]
        for risk in top_risks:
            if not any(p['target'].startswith(risk.description[:20]) for p in priorities):
                priorities.append({
                    "priority": 3,
                    "type": "individual",
                    "target": risk.description[:50],
                    "exposure": risk.score,
                    "projects": 1,
                    "recommendation": risk.mitigation or "Develop mitigation plan"
                })

        return sorted(priorities, key=lambda x: (x['priority'], -x['exposure']))

    def generate_report(self) -> str:
        """Generate enterprise risk report."""
        profile = self.generate_portfolio_profile()

        lines = [
            "# Enterprise Risk Aggregation Report",
            "",
            f"**Portfolio:** {self.portfolio_name}",
            f"**Report Date:** {profile.report_date.strftime('%Y-%m-%d')}",
            "",
            "## Executive Summary",
            "",
            f"| Metric | Value |",
            f"|--------|-------|",
            f"| Total Projects | {profile.total_projects} |",
            f"| Total Risks | {profile.total_risks} |",
            f"| Total Exposure | ${profile.total_exposure:,.0f} |",
            f"| Expected Loss | ${profile.expected_loss:,.0f} |",
            f"| VaR (95%) | ${profile.var_95:,.0f} |",
            "",
            "## Risk Distribution by Category",
            "",
            "| Category | Risks | Projects | Expected Loss | % of Total |",
            "|----------|-------|----------|---------------|------------|"
        ]

        for cat, agg in profile.by_category.items():
            pct = (agg.expected_loss / profile.expected_loss * 100) if profile.expected_loss > 0 else 0
            lines.append(
                f"| {cat} | {agg.risk_count} | {agg.projects_affected} | "
                f"${agg.expected_loss:,.0f} | {pct:.1f}% |"
            )

        # Systemic risks
        if profile.systemic_risks:
            lines.extend([
                "",
                "## Systemic Risks (Portfolio-Wide)",
                ""
            ])
            for trigger in profile.systemic_risks[:5]:
                lines.append(f"- **{trigger}**")

        # Top individual risks
        lines.extend([
            "",
            "## Top 10 Individual Risks",
            "",
            "| Project | Risk | Prob | Impact | Score |",
            "|---------|------|------|--------|-------|"
        ])

        for risk in profile.top_risks:
            lines.append(
                f"| {risk.project_name} | {risk.description[:30]} | "
                f"{risk.probability:.0%} | ${risk.impact:,.0f} | ${risk.score:,.0f} |"
            )

        # Correlations
        high_corr = [c for c in profile.correlations if c.strength > 0.3]
        if high_corr:
            lines.extend([
                "",
                f"## Risk Correlations ({len(high_corr)} significant)",
                "",
                "| Strength | Shared Triggers |",
                "|----------|-----------------|"
            ])
            for c in high_corr[:10]:
                lines.append(
                    f"| {c.strength:.0%} | {', '.join(c.shared_triggers[:3])} |"
                )

        return "\n".join(lines)

Quick Start

# Initialize aggregator
aggregator = EnterpriseRiskAggregator("Regional Portfolio")

# Add risks from Project A
aggregator.add_risk(
    "PRJ-A", "Downtown Tower",
    RiskCategory.MARKET, "Steel price increase",
    probability=0.7, impact=2000000,
    triggers=["steel_price_increase", "trade_restrictions"]
)
aggregator.add_risk(
    "PRJ-A", "Downtown Tower",
    RiskCategory.LABOR, "Skilled labor shortage",
    probability=0.5, impact=1500000,
    triggers=["labor_shortage"]
)

# Add risks from Project B
aggregator.add_risk(
    "PRJ-B", "Hospital Wing",
    RiskCategory.MARKET, "Material cost escalation",
    probability=0.6, impact=1800000,
    triggers=["steel_price_increase", "supply_chain_disruption"]
)
aggregator.add_risk(
    "PRJ-B", "Hospital Wing",
    RiskCategory.SCHEDULE, "Weather delays",
    probability=0.4, impact=500000,
    triggers=["weather_event"]
)

# Detect correlations
correlations = aggregator.detect_correlations()
print(f"Found {len(correlations)} correlated risk pairs")

# Identify systemic risks
systemic = aggregator.identify_systemic_risks()
for s in systemic[:3]:
    print(f"Systemic: {s['trigger']} affects {s['projects_affected']} projects")

# Generate portfolio profile
profile = aggregator.generate_portfolio_profile()
print(f"Total Exposure: ${profile.total_exposure:,.0f}")
print(f"VaR (95%): ${profile.var_95:,.0f}")

# Get mitigation priorities
priorities = aggregator.suggest_mitigation_priorities()
for p in priorities[:5]:
    print(f"Priority {p['priority']}: {p['recommendation']}")

# Generate report
print(aggregator.generate_report())

Requirements

pip install (no external dependencies)