430 lines
14 KiB
Python
430 lines
14 KiB
Python
"""
|
|
Inventory Projector
|
|
|
|
Projects future inventory levels day-by-day to identify coverage gaps
|
|
and stockout risks before they occur.
|
|
"""
|
|
|
|
from datetime import date, timedelta
|
|
from decimal import Decimal
|
|
from typing import List, Dict, Optional, Tuple
|
|
from dataclasses import dataclass, field
|
|
import logging
|
|
|
|
from shared.utils.time_series_utils import generate_future_dates
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@dataclass
|
|
class DailyDemand:
|
|
"""Daily demand forecast for an ingredient"""
|
|
ingredient_id: str
|
|
date: date
|
|
quantity: Decimal
|
|
|
|
|
|
@dataclass
|
|
class ScheduledReceipt:
|
|
"""Planned receipt (PO, production, etc.)"""
|
|
ingredient_id: str
|
|
date: date
|
|
quantity: Decimal
|
|
source: str # 'purchase_order', 'production', 'transfer'
|
|
reference_id: Optional[str] = None
|
|
|
|
|
|
@dataclass
|
|
class InventoryLevel:
|
|
"""Current inventory level"""
|
|
ingredient_id: str
|
|
quantity: Decimal
|
|
unit_of_measure: str
|
|
|
|
|
|
@dataclass
|
|
class DailyProjection:
|
|
"""Daily inventory projection"""
|
|
date: date
|
|
starting_stock: Decimal
|
|
forecasted_consumption: Decimal
|
|
scheduled_receipts: Decimal
|
|
projected_ending_stock: Decimal
|
|
is_stockout: bool
|
|
coverage_gap: Decimal # Negative amount if stockout
|
|
|
|
|
|
@dataclass
|
|
class IngredientProjection:
|
|
"""Complete projection for one ingredient"""
|
|
ingredient_id: str
|
|
ingredient_name: str
|
|
current_stock: Decimal
|
|
unit_of_measure: str
|
|
projection_horizon_days: int
|
|
daily_projections: List[DailyProjection] = field(default_factory=list)
|
|
total_consumption: Decimal = Decimal('0')
|
|
total_receipts: Decimal = Decimal('0')
|
|
stockout_days: int = 0
|
|
stockout_risk: str = "low" # low, medium, high
|
|
|
|
|
|
class InventoryProjector:
|
|
"""
|
|
Projects inventory levels over time to identify coverage gaps.
|
|
|
|
Algorithm:
|
|
For each day in horizon:
|
|
Starting Stock = Previous Day's Ending Stock
|
|
Consumption = Forecasted Demand
|
|
Receipts = Scheduled Deliveries + Production
|
|
Ending Stock = Starting Stock - Consumption + Receipts
|
|
|
|
Identifies:
|
|
- Days when stock goes negative (stockouts)
|
|
- Coverage gaps (how much short)
|
|
- Stockout risk level
|
|
"""
|
|
|
|
def __init__(self, projection_horizon_days: int = 7):
|
|
"""
|
|
Initialize inventory projector.
|
|
|
|
Args:
|
|
projection_horizon_days: Number of days to project
|
|
"""
|
|
self.projection_horizon_days = projection_horizon_days
|
|
|
|
def project_inventory(
|
|
self,
|
|
ingredient_id: str,
|
|
ingredient_name: str,
|
|
current_stock: Decimal,
|
|
unit_of_measure: str,
|
|
daily_demand: List[DailyDemand],
|
|
scheduled_receipts: List[ScheduledReceipt],
|
|
start_date: Optional[date] = None
|
|
) -> IngredientProjection:
|
|
"""
|
|
Project inventory levels for one ingredient.
|
|
|
|
Args:
|
|
ingredient_id: Ingredient ID
|
|
ingredient_name: Ingredient name
|
|
current_stock: Current inventory level
|
|
unit_of_measure: Unit of measure
|
|
daily_demand: List of daily demand forecasts
|
|
scheduled_receipts: List of scheduled receipts
|
|
start_date: Starting date (defaults to today)
|
|
|
|
Returns:
|
|
IngredientProjection with daily projections
|
|
"""
|
|
if start_date is None:
|
|
start_date = date.today()
|
|
|
|
# Generate projection dates
|
|
projection_dates = generate_future_dates(start_date, self.projection_horizon_days)
|
|
|
|
# Build demand lookup
|
|
demand_by_date = {d.date: d.quantity for d in daily_demand}
|
|
|
|
# Build receipts lookup
|
|
receipts_by_date: Dict[date, Decimal] = {}
|
|
for receipt in scheduled_receipts:
|
|
if receipt.date not in receipts_by_date:
|
|
receipts_by_date[receipt.date] = Decimal('0')
|
|
receipts_by_date[receipt.date] += receipt.quantity
|
|
|
|
# Project day by day
|
|
daily_projections = []
|
|
running_stock = current_stock
|
|
total_consumption = Decimal('0')
|
|
total_receipts = Decimal('0')
|
|
stockout_days = 0
|
|
|
|
for projection_date in projection_dates:
|
|
starting_stock = running_stock
|
|
|
|
# Get consumption for this day
|
|
consumption = demand_by_date.get(projection_date, Decimal('0'))
|
|
|
|
# Get receipts for this day
|
|
receipts = receipts_by_date.get(projection_date, Decimal('0'))
|
|
|
|
# Calculate ending stock
|
|
ending_stock = starting_stock - consumption + receipts
|
|
|
|
# Check for stockout
|
|
is_stockout = ending_stock < Decimal('0')
|
|
coverage_gap = min(Decimal('0'), ending_stock)
|
|
|
|
if is_stockout:
|
|
stockout_days += 1
|
|
|
|
# Create daily projection
|
|
daily_proj = DailyProjection(
|
|
date=projection_date,
|
|
starting_stock=starting_stock,
|
|
forecasted_consumption=consumption,
|
|
scheduled_receipts=receipts,
|
|
projected_ending_stock=ending_stock,
|
|
is_stockout=is_stockout,
|
|
coverage_gap=coverage_gap
|
|
)
|
|
|
|
daily_projections.append(daily_proj)
|
|
|
|
# Update running totals
|
|
total_consumption += consumption
|
|
total_receipts += receipts
|
|
running_stock = ending_stock
|
|
|
|
# Calculate stockout risk
|
|
stockout_risk = self._calculate_stockout_risk(
|
|
stockout_days=stockout_days,
|
|
total_days=len(projection_dates),
|
|
final_stock=running_stock
|
|
)
|
|
|
|
return IngredientProjection(
|
|
ingredient_id=ingredient_id,
|
|
ingredient_name=ingredient_name,
|
|
current_stock=current_stock,
|
|
unit_of_measure=unit_of_measure,
|
|
projection_horizon_days=self.projection_horizon_days,
|
|
daily_projections=daily_projections,
|
|
total_consumption=total_consumption,
|
|
total_receipts=total_receipts,
|
|
stockout_days=stockout_days,
|
|
stockout_risk=stockout_risk
|
|
)
|
|
|
|
def project_multiple_ingredients(
|
|
self,
|
|
ingredients_data: List[Dict]
|
|
) -> List[IngredientProjection]:
|
|
"""
|
|
Project inventory for multiple ingredients.
|
|
|
|
Args:
|
|
ingredients_data: List of dicts with ingredient data
|
|
|
|
Returns:
|
|
List of ingredient projections
|
|
"""
|
|
projections = []
|
|
|
|
for data in ingredients_data:
|
|
projection = self.project_inventory(
|
|
ingredient_id=data['ingredient_id'],
|
|
ingredient_name=data['ingredient_name'],
|
|
current_stock=data['current_stock'],
|
|
unit_of_measure=data['unit_of_measure'],
|
|
daily_demand=data.get('daily_demand', []),
|
|
scheduled_receipts=data.get('scheduled_receipts', []),
|
|
start_date=data.get('start_date')
|
|
)
|
|
|
|
projections.append(projection)
|
|
|
|
return projections
|
|
|
|
def identify_coverage_gaps(
|
|
self,
|
|
projection: IngredientProjection
|
|
) -> List[Dict]:
|
|
"""
|
|
Identify all coverage gaps in projection.
|
|
|
|
Args:
|
|
projection: Ingredient projection
|
|
|
|
Returns:
|
|
List of coverage gap details
|
|
"""
|
|
gaps = []
|
|
|
|
for daily_proj in projection.daily_projections:
|
|
if daily_proj.is_stockout:
|
|
gap = {
|
|
'date': daily_proj.date,
|
|
'shortage_quantity': abs(daily_proj.coverage_gap),
|
|
'starting_stock': daily_proj.starting_stock,
|
|
'consumption': daily_proj.forecasted_consumption,
|
|
'receipts': daily_proj.scheduled_receipts
|
|
}
|
|
gaps.append(gap)
|
|
|
|
if gaps:
|
|
logger.warning(
|
|
f"{projection.ingredient_name}: {len(gaps)} stockout days detected"
|
|
)
|
|
|
|
return gaps
|
|
|
|
def calculate_required_order_quantity(
|
|
self,
|
|
projection: IngredientProjection,
|
|
target_coverage_days: int = 7
|
|
) -> Decimal:
|
|
"""
|
|
Calculate how much to order to achieve target coverage.
|
|
|
|
Args:
|
|
projection: Ingredient projection
|
|
target_coverage_days: Target days of coverage
|
|
|
|
Returns:
|
|
Required order quantity
|
|
"""
|
|
# Calculate average daily consumption
|
|
if projection.daily_projections:
|
|
avg_daily_consumption = projection.total_consumption / len(projection.daily_projections)
|
|
else:
|
|
return Decimal('0')
|
|
|
|
# Target stock level
|
|
target_stock = avg_daily_consumption * Decimal(str(target_coverage_days))
|
|
|
|
# Calculate shortfall
|
|
final_projected_stock = projection.daily_projections[-1].projected_ending_stock if projection.daily_projections else Decimal('0')
|
|
|
|
required_order = max(Decimal('0'), target_stock - final_projected_stock)
|
|
|
|
return required_order
|
|
|
|
def _calculate_stockout_risk(
|
|
self,
|
|
stockout_days: int,
|
|
total_days: int,
|
|
final_stock: Decimal
|
|
) -> str:
|
|
"""
|
|
Calculate stockout risk level.
|
|
|
|
Args:
|
|
stockout_days: Number of stockout days
|
|
total_days: Total projection days
|
|
final_stock: Final projected stock
|
|
|
|
Returns:
|
|
Risk level: 'low', 'medium', 'high', 'critical'
|
|
"""
|
|
if stockout_days == 0 and final_stock > Decimal('0'):
|
|
return "low"
|
|
|
|
stockout_ratio = stockout_days / total_days if total_days > 0 else 0
|
|
|
|
if stockout_ratio >= 0.5 or final_stock < Decimal('-100'):
|
|
return "critical"
|
|
elif stockout_ratio >= 0.3 or final_stock < Decimal('-50'):
|
|
return "high"
|
|
elif stockout_ratio > 0 or final_stock < Decimal('0'):
|
|
return "medium"
|
|
else:
|
|
return "low"
|
|
|
|
def get_high_risk_ingredients(
|
|
self,
|
|
projections: List[IngredientProjection]
|
|
) -> List[IngredientProjection]:
|
|
"""
|
|
Filter to high/critical risk ingredients.
|
|
|
|
Args:
|
|
projections: List of ingredient projections
|
|
|
|
Returns:
|
|
List of high-risk projections
|
|
"""
|
|
high_risk = [
|
|
p for p in projections
|
|
if p.stockout_risk in ['high', 'critical']
|
|
]
|
|
|
|
if high_risk:
|
|
logger.warning(f"Found {len(high_risk)} high-risk ingredients")
|
|
for proj in high_risk:
|
|
logger.warning(
|
|
f" - {proj.ingredient_name}: {proj.stockout_days} stockout days, "
|
|
f"risk={proj.stockout_risk}"
|
|
)
|
|
|
|
return high_risk
|
|
|
|
def get_summary_statistics(
|
|
self,
|
|
projections: List[IngredientProjection]
|
|
) -> Dict:
|
|
"""
|
|
Get summary statistics across all projections.
|
|
|
|
Args:
|
|
projections: List of ingredient projections
|
|
|
|
Returns:
|
|
Summary statistics
|
|
"""
|
|
total_ingredients = len(projections)
|
|
stockout_ingredients = sum(1 for p in projections if p.stockout_days > 0)
|
|
|
|
risk_breakdown = {
|
|
'low': sum(1 for p in projections if p.stockout_risk == 'low'),
|
|
'medium': sum(1 for p in projections if p.stockout_risk == 'medium'),
|
|
'high': sum(1 for p in projections if p.stockout_risk == 'high'),
|
|
'critical': sum(1 for p in projections if p.stockout_risk == 'critical')
|
|
}
|
|
|
|
total_stockout_days = sum(p.stockout_days for p in projections)
|
|
total_consumption = sum(p.total_consumption for p in projections)
|
|
total_receipts = sum(p.total_receipts for p in projections)
|
|
|
|
return {
|
|
'total_ingredients': total_ingredients,
|
|
'stockout_ingredients': stockout_ingredients,
|
|
'stockout_percentage': (stockout_ingredients / total_ingredients * 100) if total_ingredients > 0 else 0,
|
|
'risk_breakdown': risk_breakdown,
|
|
'total_stockout_days': total_stockout_days,
|
|
'total_consumption': float(total_consumption),
|
|
'total_receipts': float(total_receipts),
|
|
'projection_horizon_days': self.projection_horizon_days
|
|
}
|
|
|
|
def export_projection_to_dict(
|
|
self,
|
|
projection: IngredientProjection
|
|
) -> Dict:
|
|
"""
|
|
Export projection to dictionary for API response.
|
|
|
|
Args:
|
|
projection: Ingredient projection
|
|
|
|
Returns:
|
|
Dictionary representation
|
|
"""
|
|
return {
|
|
'ingredient_id': projection.ingredient_id,
|
|
'ingredient_name': projection.ingredient_name,
|
|
'current_stock': float(projection.current_stock),
|
|
'unit_of_measure': projection.unit_of_measure,
|
|
'projection_horizon_days': projection.projection_horizon_days,
|
|
'total_consumption': float(projection.total_consumption),
|
|
'total_receipts': float(projection.total_receipts),
|
|
'stockout_days': projection.stockout_days,
|
|
'stockout_risk': projection.stockout_risk,
|
|
'daily_projections': [
|
|
{
|
|
'date': dp.date.isoformat(),
|
|
'starting_stock': float(dp.starting_stock),
|
|
'forecasted_consumption': float(dp.forecasted_consumption),
|
|
'scheduled_receipts': float(dp.scheduled_receipts),
|
|
'projected_ending_stock': float(dp.projected_ending_stock),
|
|
'is_stockout': dp.is_stockout,
|
|
'coverage_gap': float(dp.coverage_gap)
|
|
}
|
|
for dp in projection.daily_projections
|
|
]
|
|
}
|