157 lines
5.6 KiB
Python
157 lines
5.6 KiB
Python
"""
|
|
Business impact analyzer for alerts.
|
|
|
|
Calculates financial impact, affected orders, customer impact, and other
|
|
business metrics from event metadata.
|
|
"""
|
|
|
|
from typing import Dict, Any
|
|
import structlog
|
|
|
|
logger = structlog.get_logger()
|
|
|
|
|
|
class BusinessImpactAnalyzer:
|
|
"""Analyze business impact from event metadata"""
|
|
|
|
def analyze(self, event_type: str, metadata: Dict[str, Any]) -> dict:
|
|
"""
|
|
Analyze business impact for an event.
|
|
|
|
Returns dict with:
|
|
- financial_impact_eur: Direct financial cost
|
|
- affected_orders: Number of orders impacted
|
|
- affected_customers: List of customer names
|
|
- production_delay_hours: Hours of production delay
|
|
- estimated_revenue_loss_eur: Potential revenue loss
|
|
- customer_impact: high/medium/low
|
|
- waste_risk_kg: Potential waste in kg
|
|
"""
|
|
|
|
impact = {
|
|
"financial_impact_eur": 0,
|
|
"affected_orders": 0,
|
|
"affected_customers": [],
|
|
"production_delay_hours": 0,
|
|
"estimated_revenue_loss_eur": 0,
|
|
"customer_impact": "low",
|
|
"waste_risk_kg": 0
|
|
}
|
|
|
|
# Stock-related impacts
|
|
if "stock" in event_type or "shortage" in event_type:
|
|
impact.update(self._analyze_stock_impact(metadata))
|
|
|
|
# Production-related impacts
|
|
elif "production" in event_type or "delay" in event_type or "equipment" in event_type:
|
|
impact.update(self._analyze_production_impact(metadata))
|
|
|
|
# Procurement-related impacts
|
|
elif "po_" in event_type or "delivery" in event_type:
|
|
impact.update(self._analyze_procurement_impact(metadata))
|
|
|
|
# Quality-related impacts
|
|
elif "quality" in event_type or "expired" in event_type:
|
|
impact.update(self._analyze_quality_impact(metadata))
|
|
|
|
return impact
|
|
|
|
def _analyze_stock_impact(self, metadata: Dict[str, Any]) -> dict:
|
|
"""Analyze impact of stock-related alerts"""
|
|
impact = {}
|
|
|
|
# Calculate financial impact
|
|
shortage_amount = metadata.get("shortage_amount", 0)
|
|
unit_cost = metadata.get("unit_cost", 5) # Default €5/kg
|
|
impact["financial_impact_eur"] = float(shortage_amount) * unit_cost
|
|
|
|
# Affected orders from metadata
|
|
impact["affected_orders"] = metadata.get("affected_orders", 0)
|
|
|
|
# Customer impact based on affected orders
|
|
if impact["affected_orders"] > 5:
|
|
impact["customer_impact"] = "high"
|
|
elif impact["affected_orders"] > 2:
|
|
impact["customer_impact"] = "medium"
|
|
|
|
# Revenue loss (estimated)
|
|
avg_order_value = 50 # €50 per order
|
|
impact["estimated_revenue_loss_eur"] = impact["affected_orders"] * avg_order_value
|
|
|
|
return impact
|
|
|
|
def _analyze_production_impact(self, metadata: Dict[str, Any]) -> dict:
|
|
"""Analyze impact of production-related alerts"""
|
|
impact = {}
|
|
|
|
# Delay minutes to hours
|
|
delay_minutes = metadata.get("delay_minutes", 0)
|
|
impact["production_delay_hours"] = round(delay_minutes / 60, 1)
|
|
|
|
# Affected orders and customers
|
|
impact["affected_orders"] = metadata.get("affected_orders", 0)
|
|
|
|
customer_names = metadata.get("customer_names", [])
|
|
impact["affected_customers"] = customer_names
|
|
|
|
# Customer impact based on delay
|
|
if delay_minutes > 120: # 2+ hours
|
|
impact["customer_impact"] = "high"
|
|
elif delay_minutes > 60: # 1+ hours
|
|
impact["customer_impact"] = "medium"
|
|
|
|
# Financial impact: hourly production cost
|
|
hourly_cost = 100 # €100/hour operational cost
|
|
impact["financial_impact_eur"] = impact["production_delay_hours"] * hourly_cost
|
|
|
|
# Revenue loss
|
|
if impact["affected_orders"] > 0:
|
|
avg_order_value = 50
|
|
impact["estimated_revenue_loss_eur"] = impact["affected_orders"] * avg_order_value
|
|
|
|
return impact
|
|
|
|
def _analyze_procurement_impact(self, metadata: Dict[str, Any]) -> dict:
|
|
"""Analyze impact of procurement-related alerts"""
|
|
impact = {}
|
|
|
|
# Extract potential_loss_eur from reasoning_data.parameters
|
|
reasoning_data = metadata.get("reasoning_data", {})
|
|
parameters = reasoning_data.get("parameters", {})
|
|
potential_loss_eur = parameters.get("potential_loss_eur")
|
|
|
|
# Use potential loss from reasoning as financial impact (what's at risk)
|
|
# Fallback to PO amount only if reasoning data is not available
|
|
if potential_loss_eur is not None:
|
|
impact["financial_impact_eur"] = float(potential_loss_eur)
|
|
else:
|
|
po_amount = metadata.get("po_amount", metadata.get("total_amount", 0))
|
|
impact["financial_impact_eur"] = float(po_amount)
|
|
|
|
# Days overdue affects customer impact
|
|
days_overdue = metadata.get("days_overdue", 0)
|
|
if days_overdue > 3:
|
|
impact["customer_impact"] = "high"
|
|
elif days_overdue > 1:
|
|
impact["customer_impact"] = "medium"
|
|
|
|
return impact
|
|
|
|
def _analyze_quality_impact(self, metadata: Dict[str, Any]) -> dict:
|
|
"""Analyze impact of quality-related alerts"""
|
|
impact = {}
|
|
|
|
# Expired products
|
|
expired_count = metadata.get("expired_count", 0)
|
|
total_value = metadata.get("total_value", 0)
|
|
|
|
impact["financial_impact_eur"] = float(total_value)
|
|
impact["waste_risk_kg"] = metadata.get("total_quantity_kg", 0)
|
|
|
|
if expired_count > 5:
|
|
impact["customer_impact"] = "high"
|
|
elif expired_count > 2:
|
|
impact["customer_impact"] = "medium"
|
|
|
|
return impact
|