hse-risk-analyzer

name	hse-risk-analyzer
description	Analyze BSEE HSE (Health, Safety, Environment) incident data for risk assessment. Use for operator safety scoring, incident trend analysis, compliance tracking, and ESG-integrated economic evaluation.

HSE Risk Analyzer

Analyze BSEE HSE incident data to assess operational risk, operator safety performance, and integrate safety metrics into economic analysis for Gulf of Mexico operations.

When to Use

• Assessing operator safety performance before investment decisions
• Analyzing incident trends for specific fields, facilities, or operators
• Calculating risk-adjusted economic metrics (NPV with safety factors)
• Supporting ESG (Environmental, Social, Governance) compliance requirements
• Benchmarking operator safety records across similar assets
• Identifying high-risk operators or facilities for due diligence
• Generating safety-integrated investment analysis reports

Core Pattern

Query Parameters → HSE Database → Aggregate → Score → Integrate with Economics → Report

Implementation

Data Models

from dataclasses import dataclass, field
from datetime import datetime, date
from typing import Optional, List, Dict, Any
from enum import Enum
import pandas as pd
import numpy as np

class IncidentType(Enum):
    """HSE incident classification types."""
    INJURY = "injury"
    SPILL = "spill"
    EQUIPMENT_FAILURE = "equipment_failure"
    VIOLATION = "violation"

class SeverityLevel(Enum):
    """Incident severity classification."""
    FATALITY = "fatality"
    LOST_TIME = "lost_time"
    RECORDABLE = "recordable"
    NEAR_MISS = "near_miss"
    MINOR = "minor"

@dataclass
class HSEIncidentRecord:
    """Single HSE incident record."""
    bsee_incident_id: str
    incident_date: datetime
    operator: str
    incident_type: IncidentType
    severity: SeverityLevel
    facility_name: Optional[str] = None
    lease_number: Optional[str] = None
    block_number: Optional[str] = None
    field_name: Optional[str] = None
    latitude: Optional[float] = None
    longitude: Optional[float] = None
    description: Optional[str] = None
    penalty_amount: Optional[float] = None
    spill_volume_bbls: Optional[float] = None
    days_away_from_work: Optional[int] = None

    @property
    def severity_weight(self) -> float:
        """Numeric weight for severity calculations."""
        weights = {
            SeverityLevel.FATALITY: 100.0,
            SeverityLevel.LOST_TIME: 25.0,
            SeverityLevel.RECORDABLE: 10.0,
            SeverityLevel.NEAR_MISS: 5.0,
            SeverityLevel.MINOR: 1.0
        }
        return weights.get(self.severity, 1.0)

@dataclass
class OperatorSafetyProfile:
    """Safety profile for an operator."""
    operator_name: str
    total_incidents: int = 0
    fatalities: int = 0
    lost_time_incidents: int = 0
    recordable_incidents: int = 0
    total_penalties: float = 0.0
    total_spill_volume: float = 0.0
    exposure_hours: Optional[float] = None
    years_analyzed: int = 0

    @property
    def trir(self) -> Optional[float]:
        """Total Recordable Incident Rate (per 200,000 hours)."""
        if self.exposure_hours and self.exposure_hours > 0:
            recordable_count = self.fatalities + self.lost_time_incidents + self.recordable_incidents
            return (recordable_count / self.exposure_hours) * 200000
        return None

    @property
    def safety_score(self) -> float:
        """
        Calculated safety score (0-100, higher is safer).

        Based on weighted incident counts and severity.
        """
        if self.total_incidents == 0:
            return 100.0

        weighted_incidents = (
            self.fatalities * 100 +
            self.lost_time_incidents * 25 +
            self.recordable_incidents * 10
        )

        # Normalize to 0-100 scale (inverse - lower incidents = higher score)
        # Assumes max weighted score of 500 per year is "very bad"
        max_weighted = 500 * max(self.years_analyzed, 1)
        score = 100 - min(100, (weighted_incidents / max_weighted) * 100)
        return round(score, 1)

    @property
    def risk_category(self) -> str:
        """Risk category based on safety score."""
        if self.safety_score >= 90:
            return "LOW"
        elif self.safety_score >= 70:
            return "MODERATE"
        elif self.safety_score >= 50:
            return "ELEVATED"
        else:
            return "HIGH"

@dataclass
class RiskAdjustedMetrics:
    """Economic metrics adjusted for HSE risk."""
    base_npv: float
    risk_adjusted_npv: float
    risk_discount_factor: float
    safety_score: float
    risk_category: str
    potential_penalty_exposure: float
    insurance_adjustment: float

    @property
    def npv_impact(self) -> float:
        """NPV reduction due to safety risk."""
        return self.base_npv - self.risk_adjusted_npv

    @property
    def npv_impact_percent(self) -> float:
        """Percentage NPV impact from safety risk."""
        if self.base_npv != 0:
            return (self.npv_impact / abs(self.base_npv)) * 100
        return 0.0

HSE Risk Analyzer

from pathlib import Path
from typing import Optional, List, Dict, Generator
import pandas as pd
import logging
from datetime import datetime, timedelta

logger = logging.getLogger(__name__)

class HSERiskAnalyzer:
    """
    Analyzer for BSEE HSE incident data with risk scoring.

    Provides operator safety assessment, incident trend analysis,
    and risk-adjusted economic calculations.
    """

    # Industry benchmark TRIR (2023 GOM average)
    INDUSTRY_TRIR_BENCHMARK = 0.8

    # Risk discount factors by category
    RISK_DISCOUNT_FACTORS = {
        "LOW": 0.0,
        "MODERATE": 0.05,
        "ELEVATED": 0.10,
        "HIGH": 0.20
    }

    def __init__(self, data_path: Path = None):
        """
        Initialize HSE risk analyzer.

        Args:
            data_path: Path to HSE data directory or database
        """
        self.data_path = data_path or Path("data/modules/hse")
        self._incidents_cache: Optional[pd.DataFrame] = None
        self._penalties_cache: Optional[pd.DataFrame] = None

    def load_incidents(self, csv_path: Path = ) -> pd.DataFrame:
        
         csv_path  csv_path.exists():
            df = pd.read_csv(csv_path, parse_dates=[])
            ._incidents_cache = df
             df

        
        default_path = .data_path / 
         default_path.exists():
            df = pd.read_csv(default_path, parse_dates=[])
            ._incidents_cache = df
             df

         FileNotFoundError()

     () -> OperatorSafetyProfile:
        
         ._incidents_cache  :
            .load_incidents()

        df = ._incidents_cache.copy()

        
        df = df[df[]..contains(operator, =, na=)]

        
         start_date:
            df = df[df[] >= start_date]
         end_date:
            df = df[df[] <= end_date]

        
         (df) > :
            date_range = (df[].() - df[].()).days
            years = (, date_range / )
        :
            years = 

        profile = OperatorSafetyProfile(
            operator_name=operator,
            total_incidents=(df),
            fatalities=(df[df[] == ]),
            lost_time_incidents=(df[df[] == ]),
            recordable_incidents=(df[df[] == ]),
            total_penalties=df[].()    df.columns  ,
            total_spill_volume=df[].()    df.columns  ,
            years_analyzed=(years)
        )

         profile

     () -> [, ]:
        
         ._incidents_cache  :
            .load_incidents()

        df = ._incidents_cache.copy()
        cutoff = datetime.now() - timedelta(days=years * )

        
        field_df = df[
            (df[]..contains(field_name, =, na=)) &
            (df[] >= cutoff)
        ]

        
        operators = field_df[].unique().tolist()  (field_df) >   []

        
        type_counts = field_df[].value_counts().to_dict()  (field_df) >   {}
        severity_counts = field_df[].value_counts().to_dict()  (field_df) >   {}

        
        weighted_sum = (
            severity_counts.get(, ) *  +
            severity_counts.get(, ) *  +
            severity_counts.get(, ) *  +
            severity_counts.get(, ) *  +
            severity_counts.get(, ) * 
        )

        max_expected =  * years  
        field_score =  - (, (weighted_sum / max_expected) * )

         {
            : field_name,
            : years,
            : (field_df),
            : operators,
            : type_counts,
            : severity_counts,
            : (field_score, ),
            : ._score_to_category(field_score),
            : field_df[].()    field_df.columns  ,
            : field_df[].()    field_df.columns  
        }

     () -> :
        
         score >= :
             
         score >= :
             
         score >= :
             
         

     () -> RiskAdjustedMetrics:
        
        profile = .get_operator_profile(operator)

        
        risk_factor = .RISK_DISCOUNT_FACTORS.get(profile.risk_category, )

        
        penalty_exposure = 
         include_penalty_exposure  profile.years_analyzed > :
            annual_penalties = profile.total_penalties / profile.years_analyzed
            penalty_exposure = annual_penalties *   

        
        insurance_adj = 
         include_insurance_adjustment  profile.trir:
            
            trir_ratio = profile.trir / .INDUSTRY_TRIR_BENCHMARK
            insurance_adj = base_npv *  * (, trir_ratio - )  

        
        risk_adjusted = base_npv * ( - risk_factor) - penalty_exposure - insurance_adj

         RiskAdjustedMetrics(
            base_npv=base_npv,
            risk_adjusted_npv=(risk_adjusted, ),
            risk_discount_factor=risk_factor,
            safety_score=profile.safety_score,
            risk_category=profile.risk_category,
            potential_penalty_exposure=(penalty_exposure, ),
            insurance_adjustment=(insurance_adj, )
        )

     () -> pd.DataFrame:
        
         ._incidents_cache  :
            .load_incidents()

        df = ._incidents_cache.copy()
        cutoff = datetime.now() - timedelta(days=years * )
        df = df[df[] >= cutoff]

         operator:
            df = df[df[]..contains(operator, =, na=)]
         field:
            df = df[df[]..contains(field, =, na=)]

        
        df[] = df[].dt.to_period(frequency)

        
        trends = df.groupby().agg({
            : ,
            :     df.columns   x: ,
            :     df.columns   x: 
        }).reset_index()

        trends.columns = [, , , ]
        trends[] = trends[].astype()

         trends

     () -> pd.DataFrame:
        
        start_date = datetime.now() - timedelta(days=years * )

        results = []
         op  operators:
            profile = .get_operator_profile(op, start_date=start_date)
            results.append({
                : profile.operator_name,
                : profile.total_incidents,
                : profile.fatalities,
                : profile.lost_time_incidents,
                : profile.recordable_incidents,
                : profile.safety_score,
                : profile.risk_category,
                : profile.total_penalties,
                : profile.total_spill_volume
            })

         pd.DataFrame(results).sort_values(, ascending=)

HSE Report Generator

import plotly.express as px
import plotly.graph_objects as go
from plotly.subplots import make_subplots
from pathlib import Path
from datetime import datetime

class HSEReportGenerator:
    """
    Generate interactive HTML reports for HSE risk analysis.
    """

    def __init__(self, analyzer: HSERiskAnalyzer):
        """
        Initialize report generator.

        Args:
            analyzer: HSERiskAnalyzer instance
        """
        self.analyzer = analyzer

    def generate_operator_report(
        self,
        operator: str,
        output_path: Path,
        years: int = 5
    ) -> Path:
        """
        Generate comprehensive operator safety report.

        Args:
            operator: Operator name
            output_path: Output HTML file path
            years: Years to analyze

        Returns:
            Path to generated report
        """
        profile = self.analyzer.get_operator_profile(
            operator,
            start_date=datetime.now() - timedelta(days=years * 365)
        )
        trends = self.analyzer.get_incident_trends(operator=operator, years=years)

        # Create subplots
        fig = make_subplots(
            rows=2, cols=2,
            subplot_titles=[
                'Incident Trends', 'Severity Breakdown',
                'Safety Score Gauge', 'Penalty History'
            ],
            specs=[
                [{"type": }, {: }],
                [{: }, {: }]
            ]
        )

        
        fig.add_trace(
            go.Scatter(
                x=trends[],
                y=trends[],
                mode=,
                name=
            ),
            row=, col=
        )

        
        severity_data = [
            profile.fatalities,
            profile.lost_time_incidents,
            profile.recordable_incidents,
            profile.total_incidents - profile.fatalities - profile.lost_time_incidents - profile.recordable_incidents
        ]
        severity_labels = [, , , ]

        fig.add_trace(
            go.Pie(
                labels=severity_labels,
                values=severity_data,
                hole=
            ),
            row=, col=
        )

        
        fig.add_trace(
            go.Indicator(
                mode=,
                value=profile.safety_score,
                title={: },
                gauge={
                    : {: [, ]},
                    : {: },
                    : [
                        {: [, ], : },
                        {: [, ], : },
                        {: [, ], : },
                        {: [, ], : }
                    ]
                }
            ),
            row=, col=
        )

        
        fig.add_trace(
            go.Bar(
                x=trends[],
                y=trends[],
                name=
            ),
            row=, col=
        )

        fig.update_layout(
            title=,
            height=,
            showlegend=
        )

        
        output_path = Path(output_path)
        output_path.parent.mkdir(parents=, exist_ok=)
        fig.write_html((output_path))

         output_path

     () -> Path:
        
        comparison = .analyzer.compare_operators(operators, years)

        fig = make_subplots(
            rows=, cols=,
            subplot_titles=[, ],
            specs=[[{: }], [{: }]]
        )

        
        colors = comparison[].({
            : ,
            : ,
            : ,
            : 
        })

        fig.add_trace(
            go.Bar(
                x=comparison[],
                y=comparison[],
                marker_color=colors,
                name=
            ),
            row=, col=
        )

        
        fig.add_trace(
            go.Bar(x=comparison[], y=comparison[], name=),
            row=, col=
        )
        fig.add_trace(
            go.Bar(x=comparison[], y=comparison[], name=),
            row=, col=
        )
        fig.add_trace(
            go.Bar(x=comparison[], y=comparison[], name=),
            row=, col=
        )

        fig.update_layout(
            title=,
            height=,
            barmode=
        )

        output_path = Path(output_path)
        output_path.parent.mkdir(parents=, exist_ok=)
        fig.write_html((output_path))

         output_path

Usage Examples

Basic Operator Safety Assessment

from worldenergydata.hse import HSERiskAnalyzer

# Initialize analyzer
analyzer = HSERiskAnalyzer()
analyzer.load_incidents(Path("data/hse/incidents.csv"))

# Get operator safety profile
profile = analyzer.get_operator_profile("Chevron")

print(f"Operator: {profile.operator_name}")
print(f"Safety Score: {profile.safety_score}/100 ({profile.risk_category})")
print(f"Total Incidents: {profile.total_incidents}")
print(f"Fatalities: {profile.fatalities}")
print(f"Total Penalties: ${profile.total_penalties:,.2f}")

Field Risk Analysis

# Analyze specific field
field_risk = analyzer.analyze_field_risk("Thunder Horse", years=5)

print(f"Field: {field_risk['field_name']}")
print(f"Risk Category: {field_risk['risk_category']}")
print(f"Safety Score: {field_risk['field_safety_score']}")
print(f"Incidents by Type: {field_risk['incident_types']}")

Risk-Adjusted NPV Calculation

# Calculate risk-adjusted NPV
base_npv = 150_000_000  # $150M base NPV

risk_metrics = analyzer.calculate_risk_adjusted_npv(
    base_npv=base_npv,
    operator="Shell",
    include_penalty_exposure=True,
    include_insurance_adjustment=True
)

print(f"Base NPV: ${risk_metrics.base_npv:,.0f}")
print(f"Risk-Adjusted NPV: ${risk_metrics.risk_adjusted_npv:,.0f}")
print(f"NPV Impact: ${risk_metrics.npv_impact:,.0f} ({risk_metrics.npv_impact_percent:.1f}%)")
print(f"Risk Category: {risk_metrics.risk_category}")

Operator Comparison

# Compare multiple operators
operators = ["Shell", "Chevron", "BP", "ExxonMobil"]
comparison = analyzer.compare_operators(operators, years=5)

print(comparison[['operator', 'safety_score', 'risk_category', 'total_incidents']])

Generate Reports

from worldenergydata.hse import HSEReportGenerator

# Initialize report generator
reporter = HSEReportGenerator(analyzer)

# Generate operator report
report_path = reporter.generate_operator_report(
    operator="Shell",
    output_path=Path("reports/shell_hse_report.html"),
    years=5
)
print(f"Report generated: {report_path}")

# Generate comparison report
comparison_path = reporter.generate_comparison_report(
    operators=["Shell", "Chevron", "BP"],
    output_path=Path("reports/operator_comparison.html"),
    years=5
)

YAML Configuration

hse_analysis:
  data_source: "data/modules/hse"

  operator_analysis:
    operator: "Chevron"
    years: 5
    include_subsidiaries: true

  field_analysis:
    fields:
      - "Thunder Horse"
      - "Mars"
      - "Atlantis"
    years: 5

  risk_adjustment:
    include_penalty_exposure: true
    include_insurance_adjustment: true
    custom_discount_factors:
      LOW: 0.0
      MODERATE: 0.05
      ELEVATED: 0.10
      HIGH: 0.20

  reporting:
    output_dir: "reports/hse"
    formats:
      - html
      - csv
    include_charts: true

Integration with NPV Analysis

from worldenergydata.hse import HSERiskAnalyzer
from worldenergydata.economics import NPVCalculator

# Initialize components
hse_analyzer = HSERiskAnalyzer()
npv_calc = NPVCalculator()

# Calculate base NPV
base_result = npv_calc.calculate(
    production_profile=production_df,
    price_assumptions=prices,
    fiscal_terms=terms,
    discount_rate=0.10
)

# Apply HSE risk adjustment
risk_metrics = hse_analyzer.calculate_risk_adjusted_npv(
    base_npv=base_result.npv,
    operator="Shell"
)

print(f"Base NPV: ${base_result.npv:,.0f}")
print(f"Risk-Adjusted NPV: ${risk_metrics.risk_adjusted_npv:,.0f}")
print(f"Safety Risk Category: {risk_metrics.risk_category}")

ESG Compliance Output

# Generate ESG-compliant safety summary
def generate_esg_summary(analyzer: HSERiskAnalyzer, operator: str) -> Dict[str, Any]:
    """Generate ESG-compliant safety summary for reporting."""
    profile = analyzer.get_operator_profile(operator)

    return {
        "operator": operator,
        "reporting_period_years": profile.years_analyzed,
        "safety_metrics": {
            "total_recordable_incident_rate": profile.trir,
            "fatalities": profile.fatalities,
            "lost_time_incidents": profile.lost_time_incidents,
            "recordable_incidents": profile.recordable_incidents,
            "safety_score": profile.safety_score,
            "risk_classification": profile.risk_category
        },
        "environmental_metrics": {
            "total_spill_volume_bbls": profile.total_spill_volume,
            "regulatory_penalties_usd": profile.total_penalties
        },
        "governance_metrics": {
            "compliance_status": "COMPLIANT" if profile.safety_score >= 70 else "REVIEW_REQUIRED"
        }
    }

Notes

• Requires HSE incident data in CSV format or database connection
• Safety scores are normalized 0-100 (higher = safer)
• Risk discount factors are configurable for organization-specific policies
• Integrates with existing NPV and economic analysis modules
• Supports ESG reporting requirements for institutional investors
• TRIR calculations require exposure hours data for accuracy