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

refactor: Convert internal services to structured JSON reasoning

Browse Source 
Convert pipe-separated reasoning codes to structured JSON format for:
- Safety stock calculator (statistical calculations, errors)
- Price forecaster (procurement recommendations, volatility)
- Order optimization (EOQ, tier pricing)

This enables i18n translation of internal calculation reasoning
and provides structured data for frontend AI insights display.

Benefits:
- Consistent with PO/Batch reasoning_data format
- Frontend can translate using same i18n infrastructure
- Structured parameters enable rich UI visualization
- No legacy string parsing needed

Changes:
- safety_stock_calculator.py: Replace reasoning str with reasoning_data dict
- price_forecaster.py: Convert recommendation reasoning to structured format
- optimization.py: Update EOQ and tier pricing to use reasoning_data

Part of complete i18n implementation for AI insights.
This commit is contained in:
Claude
2025-11-07 19:22:02 +00:00
parent be8cb20b18
commit 9d284cae46
3 changed files with 162 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

65
services/procurement/app/ml/price_forecaster.py
View File

@@ -502,32 +502,69 @@ class PriceForecaster:
if expected_change_pct < -5:
# Price expected to drop >5%
action = 'wait'
reasoning = f'PRICE_FORECAST:DECREASE_EXPECTED|change_pct={abs(expected_change_pct):.1f}|days=30'
reasoning_data = {
'type': 'decrease_expected',
'parameters': {
'change_pct': round(abs(expected_change_pct), 1),
'forecast_days': 30,
'current_price': round(current_price, 2),
'forecast_mean': round(forecast_mean, 2)
}
}
urgency = 'low'
elif expected_change_pct > 5:
# Price expected to increase >5%
action = 'buy_now'
reasoning = f'PRICE_FORECAST:INCREASE_EXPECTED|change_pct={expected_change_pct:.1f}|days=30'
reasoning_data = {
'type': 'increase_expected',
'parameters': {
'change_pct': round(expected_change_pct, 1),
'forecast_days': 30,
'current_price': round(current_price, 2),
'forecast_mean': round(forecast_mean, 2)
}
}
urgency = 'high'
elif volatility['volatility_level'] == 'high':
# High volatility - wait for dip
action = 'wait_for_dip'
reasoning = f'PRICE_FORECAST:HIGH_VOLATILITY|coefficient={volatility["coefficient_of_variation"]:.2f}'
reasoning_data = {
'type': 'high_volatility',
'parameters': {
'coefficient': round(volatility['coefficient_of_variation'], 2),
'volatility_level': volatility['volatility_level'],
'avg_daily_change_pct': round(volatility['avg_daily_change_pct'], 2)
}
}
urgency = 'medium'
elif current_price < price_stats['mean_price'] * 0.95:
# Currently below average
below_avg_pct = ((price_stats["mean_price"] - current_price) / price_stats["mean_price"] * 100)
action = 'buy_now'
reasoning = f'PRICE_FORECAST:BELOW_AVERAGE|current_price={current_price:.2f}|below_avg_pct={below_avg_pct:.1f}'
reasoning_data = {
'type': 'below_average',
'parameters': {
'current_price': round(current_price, 2),
'mean_price': round(price_stats['mean_price'], 2),
'below_avg_pct': round(below_avg_pct, 1)
}
}
urgency = 'medium'
else:
# Neutral
action = 'normal_purchase'
reasoning = 'PRICE_FORECAST:STABLE'
reasoning_data = {
'type': 'stable',
'parameters': {
'current_price': round(current_price, 2),
'forecast_mean': round(forecast_mean, 2),
'expected_change_pct': round(expected_change_pct, 2)
}
}
urgency = 'low'
# Optimal purchase timing
@@ -536,7 +573,7 @@ class PriceForecaster:
return {
'action': action,
'reasoning': reasoning,
'reasoning_data': reasoning_data,
'urgency': urgency,
'expected_price_change_pct': round(expected_change_pct, 2),
'current_price': current_price,
@@ -617,7 +654,7 @@ class PriceForecaster:
'priority': recommendations['urgency'],
'category': 'procurement',
'title': f'Buy Now: Price Increasing {recommendations["expected_price_change_pct"]:.1f}%',
'description': recommendations['reasoning'],
'reasoning_data': recommendations['reasoning_data'],
'impact_type': 'cost_avoidance',
'impact_value': abs(recommendations['expected_price_change_pct']),
'impact_unit': 'percentage',
@@ -651,7 +688,11 @@ class PriceForecaster:
'priority': 'medium',
'category': 'procurement',
'title': f'Wait to Buy: Price Decreasing {abs(recommendations["expected_price_change_pct"]):.1f}%',
'description': recommendations['reasoning'] + f' Optimal purchase date: {recommendations["optimal_purchase_date"]}.',
'reasoning_data': {
**recommendations['reasoning_data'],
'optimal_purchase_date': recommendations['optimal_purchase_date'],
'days_until_optimal': recommendations['days_until_optimal']
},
'impact_type': 'cost_savings',
'impact_value': abs(recommendations['expected_price_change_pct']),
'impact_unit': 'percentage',
@@ -788,7 +829,13 @@ class PriceForecaster:
'volatility': {},
'recommendations': {
'action': 'insufficient_data',
'reasoning': 'Not enough price history for reliable forecast. Need at least 180 days.',
'reasoning_data': {
'type': 'insufficient_data',
'parameters': {
'history_days': len(price_history),
'min_required_days': 180
}
},
'urgency': 'low'
},
'bulk_opportunities': {'has_bulk_opportunity': False},

70
services/procurement/app/services/safety_stock_calculator.py
View File

@@ -25,7 +25,7 @@ class SafetyStockResult:
lead_time_days: int
calculation_method: str
confidence: str # 'high', 'medium', 'low'
reasoning: str
reasoning_data: dict
@dataclass
@@ -108,7 +108,13 @@ class SafetyStockCalculator:
lead_time_days=lead_time_days,
calculation_method='zero_due_to_invalid_inputs',
confidence='low',
reasoning='ERROR:LEAD_TIME_INVALID' # Error code for i18n translation
reasoning_data={
'type': 'error_lead_time_invalid',
'parameters': {
'lead_time_days': lead_time_days,
'demand_std_dev': demand_std_dev
}
}
)
# Safety Stock = Z × σ × √L
@@ -117,9 +123,6 @@ class SafetyStockCalculator:
# Determine confidence
confidence = self._determine_confidence(demand_std_dev, lead_time_days)
# Use calculation method as reasoning code with parameters
reasoning = f"CALC:STATISTICAL_Z_SCORE|service_level={service_level*100:.1f}|z_score={z_score:.2f}|demand_std={demand_std_dev:.2f}|lead_time={lead_time_days}"
return SafetyStockResult(
safety_stock_quantity=Decimal(str(round(safety_stock, 2))),
service_level=service_level,
@@ -128,7 +131,18 @@ class SafetyStockCalculator:
lead_time_days=lead_time_days,
calculation_method='statistical_z_score',
confidence=confidence,
reasoning=reasoning
reasoning_data={
'type': 'statistical_z_score',
'calculation_method': 'statistical_z_score',
'parameters': {
'service_level': round(service_level * 100, 1),
'z_score': round(z_score, 2),
'demand_std_dev': round(demand_std_dev, 2),
'lead_time_days': lead_time_days,
'safety_stock': round(safety_stock, 2)
},
'confidence': confidence
}
)
def calculate_from_demand_history(
@@ -158,7 +172,13 @@ class SafetyStockCalculator:
lead_time_days=lead_time_days,
calculation_method='insufficient_data',
confidence='low',
reasoning='ERROR:INSUFFICIENT_DATA' # Error code for i18n translation
reasoning_data={
'type': 'error_insufficient_data',
'parameters': {
'data_points': len(daily_demands) if daily_demands else 0,
'min_required': 2
}
}
)
# Calculate standard deviation
@@ -209,9 +229,6 @@ class SafetyStockCalculator:
confidence = 'high' if lead_time_std_dev > 0 else 'medium'
# Use calculation method as reasoning code with parameters
reasoning = f"CALC:ADVANCED_VARIABILITY|demand_std={demand_std_dev:.2f}|lead_time_std={lead_time_std_dev:.1f}"
return SafetyStockResult(
safety_stock_quantity=Decimal(str(round(safety_stock, 2))),
service_level=service_level,
@@ -220,7 +237,20 @@ class SafetyStockCalculator:
lead_time_days=lead_time_mean,
calculation_method='statistical_with_lead_time_variability',
confidence=confidence,
reasoning=reasoning
reasoning_data={
'type': 'advanced_variability',
'calculation_method': 'statistical_with_lead_time_variability',
'parameters': {
'service_level': round(service_level * 100, 1),
'z_score': round(z_score, 2),
'demand_mean': round(demand_mean, 2),
'demand_std_dev': round(demand_std_dev, 2),
'lead_time_mean': lead_time_mean,
'lead_time_std_dev': round(lead_time_std_dev, 1),
'safety_stock': round(safety_stock, 2)
},
'confidence': confidence
}
)
def calculate_using_fixed_percentage(
@@ -245,9 +275,6 @@ class SafetyStockCalculator:
lead_time_demand = average_demand * lead_time_days
safety_stock = lead_time_demand * percentage
# Use calculation method as reasoning code with parameters
reasoning = f"CALC:FIXED_PERCENTAGE|percentage={percentage*100:.0f}|lead_time_demand={lead_time_demand:.2f}"
return SafetyStockResult(
safety_stock_quantity=Decimal(str(round(safety_stock, 2))),
service_level=0.0, # Not based on service level
@@ -256,7 +283,18 @@ class SafetyStockCalculator:
lead_time_days=lead_time_days,
calculation_method='fixed_percentage',
confidence='low',
reasoning=reasoning
reasoning_data={
'type': 'fixed_percentage',
'calculation_method': 'fixed_percentage',
'parameters': {
'percentage': round(percentage * 100, 0),
'average_demand': round(average_demand, 2),
'lead_time_days': lead_time_days,
'lead_time_demand': round(lead_time_demand, 2),
'safety_stock': round(safety_stock, 2)
},
'confidence': 'low'
}
)
def calculate_batch_safety_stock(
@@ -432,5 +470,5 @@ class SafetyStockCalculator:
'lead_time_days': result.lead_time_days,
'calculation_method': result.calculation_method,
'confidence': result.confidence,
'reasoning': result.reasoning
'reasoning_data': result.reasoning_data
}

65
shared/utils/optimization.py
View File

@@ -19,7 +19,7 @@ class OrderOptimizationResult:
holding_cost: Decimal
total_cost: Decimal
orders_per_year: float
reasoning: str
reasoning_data: dict
def calculate_economic_order_quantity(
@@ -87,20 +87,36 @@ def optimize_order_quantity(
# Start with EOQ or required quantity, whichever is larger
optimal_qty = max(float(required_quantity), eoq)
# Build structured reasoning code with parameters
reasoning_parts = [f"EOQ:BASE|eoq={eoq:.2f}|required={required_quantity}"]
# Build structured reasoning data
reasoning_data = {
'type': 'eoq_base',
'parameters': {
'eoq': round(eoq, 2),
'required_quantity': float(required_quantity),
'annual_demand': round(annual_demand, 2),
'ordering_cost': ordering_cost,
'holding_cost_rate': holding_cost_rate
},
'constraints_applied': []
}
# Apply minimum order quantity
if min_order_qty and Decimal(optimal_qty) < min_order_qty:
optimal_qty = float(min_order_qty)
reasoning_parts.append(f"EOQ:MOQ_APPLIED|moq={min_order_qty}")
reasoning_data['constraints_applied'].append({
'type': 'moq_applied',
'moq': float(min_order_qty)
})
# Apply maximum order quantity
if max_order_qty and Decimal(optimal_qty) > max_order_qty:
optimal_qty = float(max_order_qty)
reasoning_parts.append(f"EOQ:MAX_APPLIED|max={max_order_qty}")
reasoning_data['constraints_applied'].append({
'type': 'max_applied',
'max_qty': float(max_order_qty)
})
reasoning = "|".join(reasoning_parts)
reasoning_data['parameters']['optimal_quantity'] = round(optimal_qty, 2)
# Calculate costs
orders_per_year = annual_demand / optimal_qty if optimal_qty > 0 else 0
@@ -114,7 +130,7 @@ def optimize_order_quantity(
holding_cost=Decimal(str(annual_holding_cost)),
total_cost=Decimal(str(total_annual_cost)),
orders_per_year=orders_per_year,
reasoning=reasoning
reasoning_data=reasoning_data
)
@@ -180,7 +196,7 @@ def apply_price_tier_optimization(
base_quantity: Decimal,
unit_price: Decimal,
price_tiers: List[Dict]
) -> Tuple[Decimal, Decimal, str]:
) -> Tuple[Decimal, Decimal, dict]:
"""
Optimize quantity to take advantage of price tiers.
@@ -190,10 +206,16 @@ def apply_price_tier_optimization(
price_tiers: List of dicts with 'min_quantity' and 'unit_price'
Returns:
Tuple of (optimized_quantity, unit_price, reasoning)
Tuple of (optimized_quantity, unit_price, reasoning_data)
"""
if not price_tiers:
return base_quantity, unit_price, "No price tiers available"
return base_quantity, unit_price, {
'type': 'no_tiers',
'parameters': {
'base_quantity': float(base_quantity),
'unit_price': float(unit_price)
}
}
# Sort tiers by min_quantity
sorted_tiers = sorted(price_tiers, key=lambda x: x['min_quantity'])
@@ -211,7 +233,14 @@ def apply_price_tier_optimization(
best_quantity = base_quantity
best_price = current_tier_price
best_savings = Decimal('0')
reasoning = f"TIER_PRICING:CURRENT_TIER|price={current_tier_price}"
reasoning_data = {
'type': 'current_tier',
'parameters': {
'base_quantity': float(base_quantity),
'current_tier_price': float(current_tier_price),
'base_cost': float(base_cost)
}
}
for tier in sorted_tiers:
tier_min_qty = Decimal(str(tier['min_quantity']))
@@ -235,9 +264,19 @@ def apply_price_tier_optimization(
best_quantity = tier_min_qty
best_price = tier_price
best_savings = savings
reasoning = f"TIER_PRICING:UPGRADED|tier_min={tier_min_qty}|savings={savings:.2f}"
reasoning_data = {
'type': 'tier_upgraded',
'parameters': {
'base_quantity': float(base_quantity),
'tier_min_qty': float(tier_min_qty),
'base_price': float(current_tier_price),
'tier_price': float(tier_price),
'savings': round(float(savings), 2),
'additional_qty': float(additional_qty)
}
}
return best_quantity, best_price, reasoning
return best_quantity, best_price, reasoning_data
def aggregate_requirements_for_moq(