bakery-admin/bakery-ia
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Improve the frontend modals

Browse Source
This commit is contained in:
Urtzi Alfaro
2025-10-27 16:33:26 +01:00
parent 61376b7a9f
commit 858d985c92
143 changed files with 9289 additions and 2306 deletions
Download Patch File Download Diff File Expand all files Collapse all files

339
services/orders/app/services/smart_procurement_calculator.py Normal file
View File

@@ -0,0 +1,339 @@
# services/orders/app/services/smart_procurement_calculator.py
"""
Smart Procurement Calculator
Implements multi-constraint procurement quantity optimization combining:
- AI demand forecasting
- Ingredient reorder rules (reorder_point, reorder_quantity)
- Supplier constraints (minimum_order_quantity, minimum_order_amount)
- Storage limits (max_stock_level)
- Price tier optimization
"""
import math
from decimal import Decimal
from typing import Dict, Any, List, Tuple, Optional
import structlog
logger = structlog.get_logger()
class SmartProcurementCalculator:
"""
Smart procurement quantity calculator with multi-tier constraint optimization
"""
def __init__(self, procurement_settings: Dict[str, Any]):
"""
Initialize calculator with tenant procurement settings
Args:
procurement_settings: Tenant settings dict with flags:
- use_reorder_rules: bool
- economic_rounding: bool
- respect_storage_limits: bool
- use_supplier_minimums: bool
- optimize_price_tiers: bool
"""
self.use_reorder_rules = procurement_settings.get('use_reorder_rules', True)
self.economic_rounding = procurement_settings.get('economic_rounding', True)
self.respect_storage_limits = procurement_settings.get('respect_storage_limits', True)
self.use_supplier_minimums = procurement_settings.get('use_supplier_minimums', True)
self.optimize_price_tiers = procurement_settings.get('optimize_price_tiers', True)
def calculate_procurement_quantity(
self,
ingredient: Dict[str, Any],
supplier: Optional[Dict[str, Any]],
price_list_entry: Optional[Dict[str, Any]],
ai_forecast_quantity: Decimal,
current_stock: Decimal,
safety_stock_percentage: Decimal = Decimal('20.0')
) -> Dict[str, Any]:
"""
Calculate optimal procurement quantity using smart hybrid approach
Args:
ingredient: Ingredient data with reorder_point, reorder_quantity, max_stock_level
supplier: Supplier data with minimum_order_amount
price_list_entry: Price list with minimum_order_quantity, tier_pricing
ai_forecast_quantity: AI-predicted demand quantity
current_stock: Current stock level
safety_stock_percentage: Safety stock buffer percentage
Returns:
Dict with:
- order_quantity: Final calculated quantity to order
- calculation_method: Method used (e.g., 'REORDER_POINT_TRIGGERED')
- ai_suggested_quantity: Original AI forecast
- adjusted_quantity: Final quantity after constraints
- adjustment_reason: Human-readable explanation
- warnings: List of warnings/notes
- supplier_minimum_applied: bool
- storage_limit_applied: bool
- reorder_rule_applied: bool
- price_tier_applied: Dict or None
"""
warnings = []
result = {
'ai_suggested_quantity': ai_forecast_quantity,
'supplier_minimum_applied': False,
'storage_limit_applied': False,
'reorder_rule_applied': False,
'price_tier_applied': None
}
# Extract ingredient parameters
reorder_point = Decimal(str(ingredient.get('reorder_point', 0)))
reorder_quantity = Decimal(str(ingredient.get('reorder_quantity', 0)))
low_stock_threshold = Decimal(str(ingredient.get('low_stock_threshold', 0)))
max_stock_level = Decimal(str(ingredient.get('max_stock_level') or 'Infinity'))
# Extract supplier/price list parameters
supplier_min_qty = Decimal('0')
supplier_min_amount = Decimal('0')
tier_pricing = []
if price_list_entry:
supplier_min_qty = Decimal(str(price_list_entry.get('minimum_order_quantity', 0)))
tier_pricing = price_list_entry.get('tier_pricing') or []
if supplier:
supplier_min_amount = Decimal(str(supplier.get('minimum_order_amount', 0)))
# Calculate AI-based net requirement with safety stock
safety_stock = ai_forecast_quantity * (safety_stock_percentage / Decimal('100'))
total_needed = ai_forecast_quantity + safety_stock
ai_net_requirement = max(Decimal('0'), total_needed - current_stock)
# TIER 1: Critical Safety Check (Emergency Override)
if self.use_reorder_rules and current_stock <= low_stock_threshold:
base_order = max(reorder_quantity, ai_net_requirement)
result['calculation_method'] = 'CRITICAL_STOCK_EMERGENCY'
result['reorder_rule_applied'] = True
warnings.append(f"CRITICAL: Stock ({current_stock}) below threshold ({low_stock_threshold})")
order_qty = base_order
# TIER 2: Reorder Point Triggered
elif self.use_reorder_rules and current_stock <= reorder_point:
base_order = max(reorder_quantity, ai_net_requirement)
result['calculation_method'] = 'REORDER_POINT_TRIGGERED'
result['reorder_rule_applied'] = True
warnings.append(f"Reorder point triggered: stock ({current_stock}) ≤ reorder point ({reorder_point})")
order_qty = base_order
# TIER 3: Forecast-Driven (Above reorder point, no immediate need)
elif ai_net_requirement > 0:
order_qty = ai_net_requirement
result['calculation_method'] = 'FORECAST_DRIVEN_PROACTIVE'
warnings.append(f"AI forecast suggests ordering {ai_net_requirement} units")
# TIER 4: No Order Needed
else:
result['order_quantity'] = Decimal('0')
result['adjusted_quantity'] = Decimal('0')
result['calculation_method'] = 'SUFFICIENT_STOCK'
result['adjustment_reason'] = f"Current stock ({current_stock}) is sufficient. No order needed."
result['warnings'] = warnings
return result
# Apply Economic Rounding (reorder_quantity multiples)
if self.economic_rounding and reorder_quantity > 0:
multiples = math.ceil(float(order_qty / reorder_quantity))
rounded_qty = Decimal(multiples) * reorder_quantity
if rounded_qty > order_qty:
warnings.append(f"Rounded to {multiples}× reorder quantity ({reorder_quantity}) = {rounded_qty}")
order_qty = rounded_qty
# Apply Supplier Minimum Quantity Constraint
if self.use_supplier_minimums and supplier_min_qty > 0:
if order_qty < supplier_min_qty:
warnings.append(f"Increased from {order_qty} to supplier minimum ({supplier_min_qty})")
order_qty = supplier_min_qty
result['supplier_minimum_applied'] = True
else:
# Round to multiples of minimum_order_quantity (packaging constraint)
multiples = math.ceil(float(order_qty / supplier_min_qty))
rounded_qty = Decimal(multiples) * supplier_min_qty
if rounded_qty > order_qty:
warnings.append(f"Rounded to {multiples}× supplier packaging ({supplier_min_qty}) = {rounded_qty}")
result['supplier_minimum_applied'] = True
order_qty = rounded_qty
# Apply Price Tier Optimization
if self.optimize_price_tiers and tier_pricing and price_list_entry:
unit_price = Decimal(str(price_list_entry.get('unit_price', 0)))
tier_result = self._optimize_price_tier(
order_qty,
unit_price,
tier_pricing,
current_stock,
max_stock_level
)
if tier_result['tier_applied']:
order_qty = tier_result['optimized_quantity']
result['price_tier_applied'] = tier_result['tier_info']
warnings.append(tier_result['message'])
# Apply Storage Capacity Constraint
if self.respect_storage_limits and max_stock_level != Decimal('Infinity'):
if (current_stock + order_qty) > max_stock_level:
capped_qty = max(Decimal('0'), max_stock_level - current_stock)
warnings.append(f"Capped from {order_qty} to {capped_qty} due to storage limit ({max_stock_level})")
order_qty = capped_qty
result['storage_limit_applied'] = True
result['calculation_method'] += '_STORAGE_LIMITED'
# Check supplier minimum_order_amount (total order value constraint)
if self.use_supplier_minimums and supplier_min_amount > 0 and price_list_entry:
unit_price = Decimal(str(price_list_entry.get('unit_price', 0)))
order_value = order_qty * unit_price
if order_value < supplier_min_amount:
warnings.append(
f"⚠️ Order value €{order_value:.2f} < supplier minimum €{supplier_min_amount:.2f}. "
"This item needs to be combined with other products in the same PO."
)
result['calculation_method'] += '_NEEDS_CONSOLIDATION'
# Build final result
result['order_quantity'] = order_qty
result['adjusted_quantity'] = order_qty
result['adjustment_reason'] = self._build_adjustment_reason(
ai_forecast_quantity,
ai_net_requirement,
order_qty,
warnings,
result
)
result['warnings'] = warnings
return result
def _optimize_price_tier(
self,
current_qty: Decimal,
base_unit_price: Decimal,
tier_pricing: List[Dict[str, Any]],
current_stock: Decimal,
max_stock_level: Decimal
) -> Dict[str, Any]:
"""
Optimize order quantity to capture volume discount tiers if beneficial
Args:
current_qty: Current calculated order quantity
base_unit_price: Base unit price without tiers
tier_pricing: List of tier dicts with 'quantity' and 'price'
current_stock: Current stock level
max_stock_level: Maximum storage capacity
Returns:
Dict with tier_applied (bool), optimized_quantity, tier_info, message
"""
if not tier_pricing:
return {'tier_applied': False, 'optimized_quantity': current_qty}
# Sort tiers by quantity
sorted_tiers = sorted(tier_pricing, key=lambda x: x['quantity'])
best_tier = None
best_savings = Decimal('0')
for tier in sorted_tiers:
tier_qty = Decimal(str(tier['quantity']))
tier_price = Decimal(str(tier['price']))
# Skip if tier quantity is below current quantity (already captured)
if tier_qty <= current_qty:
continue
# Skip if tier would exceed storage capacity
if self.respect_storage_limits and (current_stock + tier_qty) > max_stock_level:
continue
# Skip if tier is more than 50% above current quantity (too much excess)
if tier_qty > current_qty * Decimal('1.5'):
continue
# Calculate savings
current_cost = current_qty * base_unit_price
tier_cost = tier_qty * tier_price
savings = current_cost - tier_cost
if savings > best_savings:
best_savings = savings
best_tier = {
'quantity': tier_qty,
'price': tier_price,
'savings': savings
}
if best_tier:
return {
'tier_applied': True,
'optimized_quantity': best_tier['quantity'],
'tier_info': best_tier,
'message': (
f"Upgraded to {best_tier['quantity']} units "
f"@ €{best_tier['price']}/unit "
f"(saves €{best_tier['savings']:.2f})"
)
}
return {'tier_applied': False, 'optimized_quantity': current_qty}
def _build_adjustment_reason(
self,
ai_forecast: Decimal,
ai_net_requirement: Decimal,
final_quantity: Decimal,
warnings: List[str],
result: Dict[str, Any]
) -> str:
"""
Build human-readable explanation of quantity adjustments
Args:
ai_forecast: Original AI forecast
ai_net_requirement: AI forecast + safety stock - current stock
final_quantity: Final order quantity after all adjustments
warnings: List of warning messages
result: Calculation result dict
Returns:
Human-readable adjustment explanation
"""
parts = []
# Start with calculation method
method = result.get('calculation_method', 'UNKNOWN')
parts.append(f"Method: {method.replace('_', ' ').title()}")
# AI forecast base
parts.append(f"AI Forecast: {ai_forecast} units, Net Requirement: {ai_net_requirement} units")
# Adjustments applied
adjustments = []
if result.get('reorder_rule_applied'):
adjustments.append("reorder rules")
if result.get('supplier_minimum_applied'):
adjustments.append("supplier minimums")
if result.get('storage_limit_applied'):
adjustments.append("storage limits")
if result.get('price_tier_applied'):
adjustments.append("price tier optimization")
if adjustments:
parts.append(f"Adjustments: {', '.join(adjustments)}")
# Final quantity
parts.append(f"Final Quantity: {final_quantity} units")
# Key warnings
if warnings:
key_warnings = [w for w in warnings if '⚠️' in w or 'CRITICAL' in w or 'saves €' in w]
if key_warnings:
parts.append(f"Notes: {'; '.join(key_warnings)}")
return " | ".join(parts)