605 lines
19 KiB
Python
605 lines
19 KiB
Python
"""
|
|
Tests for Safety Stock Optimizer
|
|
"""
|
|
|
|
import pytest
|
|
import pandas as pd
|
|
import numpy as np
|
|
from datetime import datetime, timedelta
|
|
from app.ml.safety_stock_optimizer import SafetyStockOptimizer
|
|
|
|
|
|
@pytest.fixture
|
|
def stable_demand_history():
|
|
"""Generate demand history with stable, predictable demand."""
|
|
dates = pd.date_range(start='2024-01-01', end='2024-12-31', freq='D')
|
|
|
|
# Stable demand: mean=100, low variance
|
|
demands = np.random.normal(100, 10, len(dates))
|
|
demands = np.maximum(demands, 0) # No negative demand
|
|
|
|
data = {
|
|
'date': dates,
|
|
'demand_quantity': demands,
|
|
'lead_time_days': [3] * len(dates),
|
|
'stockout': [False] * len(dates)
|
|
}
|
|
|
|
return pd.DataFrame(data)
|
|
|
|
|
|
@pytest.fixture
|
|
def variable_demand_history():
|
|
"""Generate demand history with high variability."""
|
|
dates = pd.date_range(start='2024-01-01', end='2024-12-31', freq='D')
|
|
|
|
# Variable demand: mean=100, high variance
|
|
demands = np.random.normal(100, 40, len(dates))
|
|
demands = np.maximum(demands, 0)
|
|
|
|
# Add some stockouts (10%)
|
|
stockouts = np.random.random(len(dates)) < 0.1
|
|
|
|
data = {
|
|
'date': dates,
|
|
'demand_quantity': demands,
|
|
'lead_time_days': np.random.normal(3, 1, len(dates)),
|
|
'stockout': stockouts
|
|
}
|
|
|
|
return pd.DataFrame(data)
|
|
|
|
|
|
@pytest.fixture
|
|
def high_criticality_product():
|
|
"""Product characteristics for high criticality product."""
|
|
return {
|
|
'shelf_life_days': 7,
|
|
'criticality': 'high',
|
|
'unit_cost': 15.0,
|
|
'avg_daily_demand': 100
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def low_criticality_product():
|
|
"""Product characteristics for low criticality product."""
|
|
return {
|
|
'shelf_life_days': 30,
|
|
'criticality': 'low',
|
|
'unit_cost': 2.0,
|
|
'avg_daily_demand': 50
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def cost_parameters():
|
|
"""Standard cost parameters."""
|
|
return {
|
|
'holding_cost_per_unit_per_day': 0.01,
|
|
'stockout_cost_per_unit': 10.0
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_optimize_stable_demand(stable_demand_history, high_criticality_product):
|
|
"""Test optimization with stable demand."""
|
|
optimizer = SafetyStockOptimizer()
|
|
|
|
results = await optimizer.optimize_safety_stock(
|
|
tenant_id='test-tenant',
|
|
inventory_product_id='stable-product',
|
|
demand_history=stable_demand_history,
|
|
product_characteristics=high_criticality_product,
|
|
min_history_days=90
|
|
)
|
|
|
|
# Check structure
|
|
assert 'tenant_id' in results
|
|
assert 'inventory_product_id' in results
|
|
assert 'optimal_result' in results
|
|
assert 'hardcoded_result' in results
|
|
assert 'comparison' in results
|
|
assert 'insights' in results
|
|
|
|
# Stable demand should have lower safety stock
|
|
optimal = results['optimal_result']
|
|
assert optimal['safety_stock'] > 0
|
|
assert 0.90 <= optimal['service_level'] <= 0.99
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_optimize_variable_demand(variable_demand_history, high_criticality_product):
|
|
"""Test optimization with variable demand."""
|
|
optimizer = SafetyStockOptimizer()
|
|
|
|
results = await optimizer.optimize_safety_stock(
|
|
tenant_id='test-tenant',
|
|
inventory_product_id='variable-product',
|
|
demand_history=variable_demand_history,
|
|
product_characteristics=high_criticality_product,
|
|
min_history_days=90
|
|
)
|
|
|
|
optimal = results['optimal_result']
|
|
|
|
# Variable demand should require higher safety stock
|
|
assert optimal['safety_stock'] > 0
|
|
|
|
# Should achieve high service level for high criticality
|
|
assert optimal['service_level'] >= 0.95
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_demand_statistics_calculation(stable_demand_history):
|
|
"""Test demand statistics calculation."""
|
|
optimizer = SafetyStockOptimizer()
|
|
|
|
results = await optimizer.optimize_safety_stock(
|
|
tenant_id='test-tenant',
|
|
inventory_product_id='test-product',
|
|
demand_history=stable_demand_history,
|
|
product_characteristics={'criticality': 'medium', 'avg_daily_demand': 100}
|
|
)
|
|
|
|
stats = results['demand_stats']
|
|
|
|
# Check all required statistics present
|
|
required_stats = [
|
|
'mean_demand', 'std_demand', 'cv_demand',
|
|
'min_demand', 'max_demand',
|
|
'mean_lead_time', 'std_lead_time',
|
|
'stockout_rate', 'data_points'
|
|
]
|
|
|
|
for stat in required_stats:
|
|
assert stat in stats, f"Missing statistic: {stat}"
|
|
|
|
# Check values are reasonable
|
|
assert stats['mean_demand'] > 0
|
|
assert stats['std_demand'] >= 0
|
|
assert 0 <= stats['cv_demand'] <= 2
|
|
assert stats['mean_lead_time'] > 0
|
|
assert 0 <= stats['stockout_rate'] <= 1
|
|
assert stats['data_points'] == len(stable_demand_history)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_statistical_safety_stock_calculation():
|
|
"""Test statistical safety stock calculation method."""
|
|
optimizer = SafetyStockOptimizer()
|
|
|
|
demand_stats = {
|
|
'mean_demand': 100,
|
|
'std_demand': 20,
|
|
'mean_lead_time': 3,
|
|
'std_lead_time': 0.5,
|
|
'cv_demand': 0.2
|
|
}
|
|
|
|
product_chars = {
|
|
'criticality': 'high',
|
|
'avg_daily_demand': 100
|
|
}
|
|
|
|
result = optimizer._calculate_statistical_safety_stock(
|
|
demand_stats, product_chars, supplier_reliability=0.95
|
|
)
|
|
|
|
# Check structure
|
|
assert 'method' in result
|
|
assert result['method'] == 'statistical'
|
|
assert 'safety_stock' in result
|
|
assert 'service_level' in result
|
|
assert 'z_score' in result
|
|
|
|
# High criticality should get 98% service level
|
|
assert result['service_level'] == 0.98
|
|
assert result['safety_stock'] > 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_criticality_affects_service_level():
|
|
"""Test that product criticality affects target service level."""
|
|
optimizer = SafetyStockOptimizer()
|
|
|
|
demand_stats = {
|
|
'mean_demand': 100,
|
|
'std_demand': 20,
|
|
'mean_lead_time': 3,
|
|
'std_lead_time': 0.5
|
|
}
|
|
|
|
# High criticality
|
|
high_result = optimizer._calculate_statistical_safety_stock(
|
|
demand_stats,
|
|
{'criticality': 'high', 'avg_daily_demand': 100},
|
|
None
|
|
)
|
|
|
|
# Low criticality
|
|
low_result = optimizer._calculate_statistical_safety_stock(
|
|
demand_stats,
|
|
{'criticality': 'low', 'avg_daily_demand': 100},
|
|
None
|
|
)
|
|
|
|
# High criticality should have higher service level and safety stock
|
|
assert high_result['service_level'] > low_result['service_level']
|
|
assert high_result['safety_stock'] > low_result['safety_stock']
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cost_based_optimization(stable_demand_history, high_criticality_product, cost_parameters):
|
|
"""Test cost-based safety stock optimization."""
|
|
optimizer = SafetyStockOptimizer()
|
|
|
|
results = await optimizer.optimize_safety_stock(
|
|
tenant_id='test-tenant',
|
|
inventory_product_id='test-product',
|
|
demand_history=stable_demand_history,
|
|
product_characteristics=high_criticality_product,
|
|
cost_parameters=cost_parameters
|
|
)
|
|
|
|
optimal = results['optimal_result']
|
|
|
|
# Should use cost optimization method for high-value product
|
|
# (unit_cost > 5)
|
|
assert optimal['method'] == 'cost_optimization'
|
|
assert 'annual_total_cost' in optimal
|
|
assert optimal['annual_total_cost'] > 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_comparison_with_hardcoded(stable_demand_history, high_criticality_product, cost_parameters):
|
|
"""Test comparison between optimal and hardcoded approaches."""
|
|
optimizer = SafetyStockOptimizer()
|
|
|
|
results = await optimizer.optimize_safety_stock(
|
|
tenant_id='test-tenant',
|
|
inventory_product_id='test-product',
|
|
demand_history=stable_demand_history,
|
|
product_characteristics=high_criticality_product,
|
|
cost_parameters=cost_parameters
|
|
)
|
|
|
|
comparison = results['comparison']
|
|
|
|
# Check comparison metrics present
|
|
assert 'stock_difference' in comparison
|
|
assert 'stock_difference_pct' in comparison
|
|
assert 'optimal_service_level' in comparison
|
|
assert 'hardcoded_service_level' in comparison
|
|
|
|
# Should have cost savings calculation
|
|
if cost_parameters:
|
|
assert 'annual_holding_cost_savings' in comparison
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_over_stocking_insight_generation(stable_demand_history, low_criticality_product, cost_parameters):
|
|
"""Test insight generation when product is over-stocked."""
|
|
optimizer = SafetyStockOptimizer()
|
|
|
|
# Low criticality product with stable demand should recommend reduction
|
|
results = await optimizer.optimize_safety_stock(
|
|
tenant_id='test-tenant',
|
|
inventory_product_id='overstocked-product',
|
|
demand_history=stable_demand_history,
|
|
product_characteristics=low_criticality_product,
|
|
cost_parameters=cost_parameters
|
|
)
|
|
|
|
insights = results['insights']
|
|
|
|
# Should generate insights
|
|
assert len(insights) > 0
|
|
|
|
# Check for reduction insight (if optimal is significantly lower)
|
|
comparison = results['comparison']
|
|
if comparison['stock_difference_pct'] < -10:
|
|
reduction_insights = [i for i in insights if 'reduce' in i.get('title', '').lower()]
|
|
assert len(reduction_insights) > 0
|
|
|
|
insight = reduction_insights[0]
|
|
assert insight['type'] == 'optimization'
|
|
assert insight['impact_type'] == 'cost_savings'
|
|
assert 'actionable' in insight
|
|
assert insight['actionable'] is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_under_stocking_insight_generation(variable_demand_history, high_criticality_product):
|
|
"""Test insight generation when product is under-stocked."""
|
|
optimizer = SafetyStockOptimizer()
|
|
|
|
# High criticality with variable demand should need more stock
|
|
results = await optimizer.optimize_safety_stock(
|
|
tenant_id='test-tenant',
|
|
inventory_product_id='understocked-product',
|
|
demand_history=variable_demand_history,
|
|
product_characteristics=high_criticality_product
|
|
)
|
|
|
|
insights = results['insights']
|
|
|
|
# Check for increase recommendation
|
|
comparison = results['comparison']
|
|
if comparison['stock_difference_pct'] > 10:
|
|
increase_insights = [i for i in insights if 'increase' in i.get('title', '').lower()]
|
|
|
|
if increase_insights:
|
|
insight = increase_insights[0]
|
|
assert insight['type'] in ['alert', 'recommendation']
|
|
assert 'recommendation_actions' in insight
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_high_variability_insight(variable_demand_history, high_criticality_product):
|
|
"""Test insight generation for high demand variability."""
|
|
optimizer = SafetyStockOptimizer()
|
|
|
|
results = await optimizer.optimize_safety_stock(
|
|
tenant_id='test-tenant',
|
|
inventory_product_id='variable-product',
|
|
demand_history=variable_demand_history,
|
|
product_characteristics=high_criticality_product
|
|
)
|
|
|
|
insights = results['insights']
|
|
stats = results['demand_stats']
|
|
|
|
# If CV > 0.5, should generate variability insight
|
|
if stats['cv_demand'] > 0.5:
|
|
variability_insights = [i for i in insights if 'variability' in i.get('title', '').lower()]
|
|
assert len(variability_insights) > 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stockout_alert_generation():
|
|
"""Test alert generation for frequent stockouts."""
|
|
# Create demand history with frequent stockouts
|
|
dates = pd.date_range(start='2024-01-01', end='2024-12-31', freq='D')
|
|
|
|
data = {
|
|
'date': dates,
|
|
'demand_quantity': np.random.normal(100, 20, len(dates)),
|
|
'lead_time_days': [3] * len(dates),
|
|
'stockout': np.random.random(len(dates)) < 0.15 # 15% stockout rate
|
|
}
|
|
|
|
demand_history = pd.DataFrame(data)
|
|
|
|
optimizer = SafetyStockOptimizer()
|
|
|
|
results = await optimizer.optimize_safety_stock(
|
|
tenant_id='test-tenant',
|
|
inventory_product_id='stockout-product',
|
|
demand_history=demand_history,
|
|
product_characteristics={'criticality': 'high', 'avg_daily_demand': 100}
|
|
)
|
|
|
|
insights = results['insights']
|
|
|
|
# Should generate stockout alert
|
|
stockout_insights = [i for i in insights if 'stockout' in i.get('title', '').lower()]
|
|
assert len(stockout_insights) > 0
|
|
|
|
insight = stockout_insights[0]
|
|
assert insight['priority'] in ['high', 'critical']
|
|
assert insight['type'] == 'alert'
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_shelf_life_constraint():
|
|
"""Test that shelf life constrains safety stock."""
|
|
dates = pd.date_range(start='2024-01-01', end='2024-12-31', freq='D')
|
|
|
|
demand_history = pd.DataFrame({
|
|
'date': dates,
|
|
'demand_quantity': np.random.normal(100, 30, len(dates)), # High variance
|
|
'lead_time_days': [3] * len(dates),
|
|
'stockout': [False] * len(dates)
|
|
})
|
|
|
|
# Product with short shelf life
|
|
product_chars = {
|
|
'shelf_life_days': 3, # Very short shelf life
|
|
'criticality': 'high',
|
|
'unit_cost': 5.0,
|
|
'avg_daily_demand': 100
|
|
}
|
|
|
|
optimizer = SafetyStockOptimizer()
|
|
|
|
results = await optimizer.optimize_safety_stock(
|
|
tenant_id='test-tenant',
|
|
inventory_product_id='perishable-product',
|
|
demand_history=demand_history,
|
|
product_characteristics=product_chars
|
|
)
|
|
|
|
optimal = results['optimal_result']
|
|
|
|
# Safety stock should be constrained by shelf life
|
|
# Max allowed = avg_daily_demand * (shelf_life * 0.5)
|
|
max_allowed = product_chars['avg_daily_demand'] * (product_chars['shelf_life_days'] * 0.5)
|
|
|
|
assert optimal['safety_stock'] <= max_allowed + 1 # Allow small rounding
|
|
|
|
# Should have constraint flag
|
|
if optimal['safety_stock'] >= max_allowed - 1:
|
|
assert optimal.get('constrained_by') == 'shelf_life'
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_supplier_reliability_adjustment():
|
|
"""Test that low supplier reliability increases safety stock."""
|
|
dates = pd.date_range(start='2024-01-01', end='2024-12-31', freq='D')
|
|
|
|
demand_history = pd.DataFrame({
|
|
'date': dates,
|
|
'demand_quantity': np.random.normal(100, 15, len(dates)),
|
|
'lead_time_days': [3] * len(dates),
|
|
'stockout': [False] * len(dates)
|
|
})
|
|
|
|
product_chars = {
|
|
'criticality': 'medium',
|
|
'avg_daily_demand': 100
|
|
}
|
|
|
|
optimizer = SafetyStockOptimizer()
|
|
|
|
# Good supplier
|
|
results_good = await optimizer.optimize_safety_stock(
|
|
tenant_id='test-tenant',
|
|
inventory_product_id='test-product',
|
|
demand_history=demand_history,
|
|
product_characteristics=product_chars,
|
|
supplier_reliability=0.98
|
|
)
|
|
|
|
# Poor supplier
|
|
results_poor = await optimizer.optimize_safety_stock(
|
|
tenant_id='test-tenant',
|
|
inventory_product_id='test-product',
|
|
demand_history=demand_history,
|
|
product_characteristics=product_chars,
|
|
supplier_reliability=0.85
|
|
)
|
|
|
|
# Poor supplier should require higher safety stock
|
|
assert results_poor['optimal_result']['safety_stock'] >= results_good['optimal_result']['safety_stock']
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_insufficient_data_handling():
|
|
"""Test handling of insufficient demand history."""
|
|
# Only 30 days (less than min_history_days=90)
|
|
dates = pd.date_range(start='2024-01-01', end='2024-01-30', freq='D')
|
|
|
|
small_history = pd.DataFrame({
|
|
'date': dates,
|
|
'demand_quantity': np.random.normal(100, 15, len(dates)),
|
|
'lead_time_days': [3] * len(dates),
|
|
'stockout': [False] * len(dates)
|
|
})
|
|
|
|
product_chars = {
|
|
'criticality': 'high',
|
|
'avg_daily_demand': 100,
|
|
'shelf_life_days': 7
|
|
}
|
|
|
|
optimizer = SafetyStockOptimizer()
|
|
|
|
results = await optimizer.optimize_safety_stock(
|
|
tenant_id='test-tenant',
|
|
inventory_product_id='new-product',
|
|
demand_history=small_history,
|
|
product_characteristics=product_chars,
|
|
min_history_days=90
|
|
)
|
|
|
|
# Should return fallback response
|
|
assert results['history_days'] == 0
|
|
assert results['optimal_result']['method'] == 'fallback_heuristic'
|
|
|
|
# Should use simple heuristic (7 days for high criticality)
|
|
expected_safety_stock = product_chars['avg_daily_demand'] * 7
|
|
assert results['optimal_result']['safety_stock'] == expected_safety_stock
|
|
|
|
|
|
def test_get_optimal_safety_stock():
|
|
"""Test retrieval of cached optimal safety stock."""
|
|
optimizer = SafetyStockOptimizer()
|
|
|
|
# Initially no cached value
|
|
assert optimizer.get_optimal_safety_stock('product-1') is None
|
|
|
|
# Set a value
|
|
optimizer.optimal_stocks['product-1'] = 150.5
|
|
|
|
# Should retrieve it
|
|
assert optimizer.get_optimal_safety_stock('product-1') == 150.5
|
|
|
|
|
|
def test_get_learned_service_level():
|
|
"""Test retrieval of learned service levels."""
|
|
optimizer = SafetyStockOptimizer()
|
|
|
|
# Initially no learned level
|
|
assert optimizer.get_learned_service_level('product-1') is None
|
|
|
|
# Set a level
|
|
optimizer.learned_service_levels['product-1'] = 0.96
|
|
|
|
# Should retrieve it
|
|
assert optimizer.get_learned_service_level('product-1') == 0.96
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_hardcoded_comparison():
|
|
"""Test comparison specifically highlights hardcoded vs optimal."""
|
|
dates = pd.date_range(start='2024-01-01', end='2024-12-31', freq='D')
|
|
|
|
demand_history = pd.DataFrame({
|
|
'date': dates,
|
|
'demand_quantity': np.random.normal(100, 15, len(dates)),
|
|
'lead_time_days': [3] * len(dates),
|
|
'stockout': [False] * len(dates)
|
|
})
|
|
|
|
optimizer = SafetyStockOptimizer()
|
|
|
|
results = await optimizer.optimize_safety_stock(
|
|
tenant_id='test-tenant',
|
|
inventory_product_id='test-product',
|
|
demand_history=demand_history,
|
|
product_characteristics={'criticality': 'medium', 'avg_daily_demand': 100}
|
|
)
|
|
|
|
# Hardcoded should always be 95% service level
|
|
assert results['hardcoded_result']['service_level'] == 0.95
|
|
assert results['hardcoded_result']['method'] == 'hardcoded_95_service_level'
|
|
|
|
# Comparison should show difference
|
|
comparison = results['comparison']
|
|
assert 'stock_difference' in comparison
|
|
assert 'service_level_difference' in comparison
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_multiple_products_caching():
|
|
"""Test that optimizer caches results for multiple products."""
|
|
optimizer = SafetyStockOptimizer()
|
|
|
|
dates = pd.date_range(start='2024-01-01', end='2024-12-31', freq='D')
|
|
|
|
# Optimize multiple products
|
|
for i in range(3):
|
|
demand_history = pd.DataFrame({
|
|
'date': dates,
|
|
'demand_quantity': np.random.normal(100 + i*10, 15, len(dates)),
|
|
'lead_time_days': [3] * len(dates),
|
|
'stockout': [False] * len(dates)
|
|
})
|
|
|
|
await optimizer.optimize_safety_stock(
|
|
tenant_id='test-tenant',
|
|
inventory_product_id=f'product-{i}',
|
|
demand_history=demand_history,
|
|
product_characteristics={'criticality': 'medium', 'avg_daily_demand': 100 + i*10}
|
|
)
|
|
|
|
# Should have cached all three
|
|
assert len(optimizer.optimal_stocks) == 3
|
|
assert len(optimizer.learned_service_levels) == 3
|
|
|
|
# Each should be retrievable
|
|
for i in range(3):
|
|
assert optimizer.get_optimal_safety_stock(f'product-{i}') is not None
|
|
assert optimizer.get_learned_service_level(f'product-{i}') is not None
|