"""Impact estimation for AI Insights.""" from typing import Dict, Any, Optional, Tuple from decimal import Decimal from datetime import datetime, timedelta class ImpactEstimator: """ Estimate potential impact of recommendations. Calculates expected business value in terms of: - Cost savings (euros) - Revenue increase (euros) - Waste reduction (euros or percentage) - Efficiency gains (hours or percentage) - Quality improvements (units or percentage) """ def estimate_procurement_savings( self, current_price: Decimal, predicted_price: Decimal, order_quantity: Decimal, timeframe_days: int = 30 ) -> Tuple[Decimal, str, str]: """ Estimate savings from opportunistic buying. Args: current_price: Current unit price predicted_price: Predicted future price order_quantity: Quantity to order timeframe_days: Time horizon for prediction Returns: tuple: (impact_value, impact_unit, impact_type) """ savings_per_unit = predicted_price - current_price if savings_per_unit > 0: total_savings = savings_per_unit * order_quantity return ( round(total_savings, 2), 'euros', 'cost_savings' ) return (Decimal('0.0'), 'euros', 'cost_savings') def estimate_waste_reduction_savings( self, current_waste_rate: float, optimized_waste_rate: float, monthly_volume: Decimal, avg_cost_per_unit: Decimal ) -> Tuple[Decimal, str, str]: """ Estimate savings from waste reduction. Args: current_waste_rate: Current waste rate (0-1) optimized_waste_rate: Optimized waste rate (0-1) monthly_volume: Monthly volume avg_cost_per_unit: Average cost per unit Returns: tuple: (impact_value, impact_unit, impact_type) """ waste_reduction_rate = current_waste_rate - optimized_waste_rate units_saved = monthly_volume * Decimal(str(waste_reduction_rate)) savings = units_saved * avg_cost_per_unit return ( round(savings, 2), 'euros/month', 'waste_reduction' ) def estimate_forecast_improvement_value( self, current_mape: float, improved_mape: float, avg_monthly_revenue: Decimal ) -> Tuple[Decimal, str, str]: """ Estimate value from forecast accuracy improvement. Better forecasts reduce: - Stockouts (lost sales) - Overproduction (waste) - Emergency orders (premium costs) Args: current_mape: Current forecast MAPE improved_mape: Improved forecast MAPE avg_monthly_revenue: Average monthly revenue Returns: tuple: (impact_value, impact_unit, impact_type) """ # Rule of thumb: 1% MAPE improvement = 0.5% revenue impact mape_improvement = current_mape - improved_mape revenue_impact_pct = mape_improvement * 0.5 / 100 revenue_increase = avg_monthly_revenue * Decimal(str(revenue_impact_pct)) return ( round(revenue_increase, 2), 'euros/month', 'revenue_increase' ) def estimate_production_efficiency_gain( self, time_saved_minutes: int, batches_per_month: int, labor_cost_per_hour: Decimal = Decimal('15.0') ) -> Tuple[Decimal, str, str]: """ Estimate value from production efficiency improvements. Args: time_saved_minutes: Minutes saved per batch batches_per_month: Number of batches per month labor_cost_per_hour: Labor cost per hour Returns: tuple: (impact_value, impact_unit, impact_type) """ hours_saved_per_month = (time_saved_minutes * batches_per_month) / 60 cost_savings = Decimal(str(hours_saved_per_month)) * labor_cost_per_hour return ( round(cost_savings, 2), 'euros/month', 'efficiency_gain' ) def estimate_safety_stock_optimization( self, current_safety_stock: Decimal, optimal_safety_stock: Decimal, holding_cost_per_unit_per_day: Decimal, stockout_cost_reduction: Decimal = Decimal('0.0') ) -> Tuple[Decimal, str, str]: """ Estimate impact of safety stock optimization. Args: current_safety_stock: Current safety stock level optimal_safety_stock: Optimal safety stock level holding_cost_per_unit_per_day: Daily holding cost stockout_cost_reduction: Reduction in stockout costs Returns: tuple: (impact_value, impact_unit, impact_type) """ stock_reduction = current_safety_stock - optimal_safety_stock if stock_reduction > 0: # Savings from reduced holding costs daily_savings = stock_reduction * holding_cost_per_unit_per_day monthly_savings = daily_savings * 30 total_savings = monthly_savings + stockout_cost_reduction return ( round(total_savings, 2), 'euros/month', 'cost_savings' ) elif stock_reduction < 0: # Cost increase but reduces stockouts daily_cost = abs(stock_reduction) * holding_cost_per_unit_per_day monthly_cost = daily_cost * 30 net_savings = stockout_cost_reduction - monthly_cost if net_savings > 0: return ( round(net_savings, 2), 'euros/month', 'cost_savings' ) return (Decimal('0.0'), 'euros/month', 'cost_savings') def estimate_supplier_switch_savings( self, current_supplier_price: Decimal, alternative_supplier_price: Decimal, monthly_order_quantity: Decimal, quality_difference_score: float = 0.0 # -1 to 1 ) -> Tuple[Decimal, str, str]: """ Estimate savings from switching suppliers. Args: current_supplier_price: Current supplier unit price alternative_supplier_price: Alternative supplier unit price monthly_order_quantity: Monthly order quantity quality_difference_score: Quality difference (-1=worse, 0=same, 1=better) Returns: tuple: (impact_value, impact_unit, impact_type) """ price_savings = (current_supplier_price - alternative_supplier_price) * monthly_order_quantity # Adjust for quality difference # If quality is worse, reduce estimated savings quality_adjustment = 1 + (quality_difference_score * 0.1) # ±10% max adjustment adjusted_savings = price_savings * Decimal(str(quality_adjustment)) return ( round(adjusted_savings, 2), 'euros/month', 'cost_savings' ) def estimate_yield_improvement_value( self, current_yield_rate: float, predicted_yield_rate: float, production_volume: Decimal, product_price: Decimal ) -> Tuple[Decimal, str, str]: """ Estimate value from production yield improvements. Args: current_yield_rate: Current yield rate (0-1) predicted_yield_rate: Predicted yield rate (0-1) production_volume: Monthly production volume product_price: Product selling price Returns: tuple: (impact_value, impact_unit, impact_type) """ yield_improvement = predicted_yield_rate - current_yield_rate if yield_improvement > 0: additional_units = production_volume * Decimal(str(yield_improvement)) revenue_increase = additional_units * product_price return ( round(revenue_increase, 2), 'euros/month', 'revenue_increase' ) return (Decimal('0.0'), 'euros/month', 'revenue_increase') def estimate_demand_pattern_value( self, pattern_strength: float, # 0-1 potential_revenue_increase: Decimal, implementation_cost: Decimal = Decimal('0.0') ) -> Tuple[Decimal, str, str]: """ Estimate value from acting on demand patterns. Args: pattern_strength: Strength of detected pattern (0-1) potential_revenue_increase: Potential monthly revenue increase implementation_cost: One-time implementation cost Returns: tuple: (impact_value, impact_unit, impact_type) """ # Discount by pattern strength (confidence) expected_value = potential_revenue_increase * Decimal(str(pattern_strength)) # Amortize implementation cost over 6 months monthly_cost = implementation_cost / 6 net_value = expected_value - monthly_cost return ( round(max(Decimal('0.0'), net_value), 2), 'euros/month', 'revenue_increase' ) def estimate_composite_impact( self, impacts: list[Dict[str, Any]] ) -> Tuple[Decimal, str, str]: """ Combine multiple impact estimations. Args: impacts: List of impact dicts with 'value', 'unit', 'type' Returns: tuple: (total_impact_value, impact_unit, impact_type) """ total_savings = Decimal('0.0') total_revenue = Decimal('0.0') for impact in impacts: value = Decimal(str(impact['value'])) impact_type = impact['type'] if impact_type == 'cost_savings': total_savings += value elif impact_type == 'revenue_increase': total_revenue += value # Combine both types total_impact = total_savings + total_revenue if total_impact > 0: # Determine primary type primary_type = 'cost_savings' if total_savings > total_revenue else 'revenue_increase' return ( round(total_impact, 2), 'euros/month', primary_type ) return (Decimal('0.0'), 'euros/month', 'cost_savings')