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 ide