539 lines
19 KiB
Python
539 lines
19 KiB
Python
"""
|
|
Supplier Selector
|
|
|
|
Intelligently selects suppliers based on multi-criteria optimization including
|
|
price, lead time, quality, reliability, and risk diversification.
|
|
"""
|
|
|
|
from decimal import Decimal
|
|
from typing import List, Dict, Optional, Tuple
|
|
from dataclasses import dataclass
|
|
from datetime import date
|
|
import logging
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@dataclass
|
|
class SupplierOption:
|
|
"""Supplier option for an ingredient"""
|
|
supplier_id: str
|
|
supplier_name: str
|
|
unit_price: Decimal
|
|
lead_time_days: int
|
|
min_order_quantity: Optional[Decimal] = None
|
|
max_capacity: Optional[Decimal] = None
|
|
quality_score: float = 0.85 # 0-1
|
|
reliability_score: float = 0.90 # 0-1
|
|
on_time_delivery_rate: float = 0.95 # 0-1
|
|
current_allocation_percentage: float = 0.0 # Current % of total orders
|
|
|
|
|
|
@dataclass
|
|
class SupplierAllocation:
|
|
"""Allocation of quantity to a supplier"""
|
|
supplier_id: str
|
|
supplier_name: str
|
|
allocated_quantity: Decimal
|
|
allocation_percentage: float
|
|
allocation_type: str # 'primary', 'backup', 'diversification'
|
|
unit_price: Decimal
|
|
total_cost: Decimal
|
|
lead_time_days: int
|
|
supplier_score: float
|
|
score_breakdown: Dict[str, float]
|
|
allocation_reason: str
|
|
|
|
|
|
@dataclass
|
|
class SupplierSelectionResult:
|
|
"""Complete supplier selection result"""
|
|
ingredient_id: str
|
|
ingredient_name: str
|
|
required_quantity: Decimal
|
|
allocations: List[SupplierAllocation]
|
|
total_cost: Decimal
|
|
weighted_lead_time: float
|
|
risk_score: float # Lower is better
|
|
diversification_applied: bool
|
|
selection_strategy: str
|
|
|
|
|
|
class SupplierSelector:
|
|
"""
|
|
Selects optimal suppliers using multi-criteria decision analysis.
|
|
|
|
Scoring Factors:
|
|
1. Price (lower is better)
|
|
2. Lead time (shorter is better)
|
|
3. Quality score (higher is better)
|
|
4. Reliability (higher is better)
|
|
5. Diversification (balance across suppliers)
|
|
|
|
Strategies:
|
|
- Single source: Best overall supplier
|
|
- Dual source: Primary + backup
|
|
- Multi-source: Split across 2-3 suppliers for large orders
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
price_weight: float = 0.40,
|
|
lead_time_weight: float = 0.20,
|
|
quality_weight: float = 0.20,
|
|
reliability_weight: float = 0.20,
|
|
diversification_threshold: Decimal = Decimal('1000'),
|
|
max_single_supplier_percentage: float = 0.70
|
|
):
|
|
"""
|
|
Initialize supplier selector.
|
|
|
|
Args:
|
|
price_weight: Weight for price (0-1)
|
|
lead_time_weight: Weight for lead time (0-1)
|
|
quality_weight: Weight for quality (0-1)
|
|
reliability_weight: Weight for reliability (0-1)
|
|
diversification_threshold: Quantity above which to diversify
|
|
max_single_supplier_percentage: Max % to single supplier
|
|
"""
|
|
self.price_weight = price_weight
|
|
self.lead_time_weight = lead_time_weight
|
|
self.quality_weight = quality_weight
|
|
self.reliability_weight = reliability_weight
|
|
self.diversification_threshold = diversification_threshold
|
|
self.max_single_supplier_percentage = max_single_supplier_percentage
|
|
|
|
# Validate weights sum to 1
|
|
total_weight = (
|
|
price_weight + lead_time_weight + quality_weight + reliability_weight
|
|
)
|
|
if abs(total_weight - 1.0) > 0.01:
|
|
logger.warning(
|
|
f"Supplier selection weights don't sum to 1.0 (sum={total_weight}), normalizing"
|
|
)
|
|
self.price_weight /= total_weight
|
|
self.lead_time_weight /= total_weight
|
|
self.quality_weight /= total_weight
|
|
self.reliability_weight /= total_weight
|
|
|
|
def select_suppliers(
|
|
self,
|
|
ingredient_id: str,
|
|
ingredient_name: str,
|
|
required_quantity: Decimal,
|
|
supplier_options: List[SupplierOption]
|
|
) -> SupplierSelectionResult:
|
|
"""
|
|
Select optimal supplier(s) for an ingredient.
|
|
|
|
Args:
|
|
ingredient_id: Ingredient ID
|
|
ingredient_name: Ingredient name
|
|
required_quantity: Quantity needed
|
|
supplier_options: List of available suppliers
|
|
|
|
Returns:
|
|
SupplierSelectionResult with allocations
|
|
"""
|
|
if not supplier_options:
|
|
raise ValueError(f"No supplier options available for {ingredient_name}")
|
|
|
|
logger.info(
|
|
f"Selecting suppliers for {ingredient_name}: "
|
|
f"{required_quantity} units from {len(supplier_options)} options"
|
|
)
|
|
|
|
# Score all suppliers
|
|
scored_suppliers = self._score_suppliers(supplier_options)
|
|
|
|
# Determine selection strategy
|
|
strategy = self._determine_strategy(required_quantity, supplier_options)
|
|
|
|
# Select suppliers based on strategy
|
|
if strategy == 'single_source':
|
|
allocations = self._select_single_source(
|
|
required_quantity,
|
|
scored_suppliers
|
|
)
|
|
elif strategy == 'dual_source':
|
|
allocations = self._select_dual_source(
|
|
required_quantity,
|
|
scored_suppliers
|
|
)
|
|
else: # multi_source
|
|
allocations = self._select_multi_source(
|
|
required_quantity,
|
|
scored_suppliers
|
|
)
|
|
|
|
# Calculate result metrics
|
|
total_cost = sum(alloc.total_cost for alloc in allocations)
|
|
weighted_lead_time = sum(
|
|
alloc.lead_time_days * alloc.allocation_percentage
|
|
for alloc in allocations
|
|
)
|
|
risk_score = self._calculate_risk_score(allocations)
|
|
diversification_applied = len(allocations) > 1
|
|
|
|
result = SupplierSelectionResult(
|
|
ingredient_id=ingredient_id,
|
|
ingredient_name=ingredient_name,
|
|
required_quantity=required_quantity,
|
|
allocations=allocations,
|
|
total_cost=total_cost,
|
|
weighted_lead_time=weighted_lead_time,
|
|
risk_score=risk_score,
|
|
diversification_applied=diversification_applied,
|
|
selection_strategy=strategy
|
|
)
|
|
|
|
logger.info(
|
|
f"{ingredient_name}: Selected {len(allocations)} supplier(s) "
|
|
f"(strategy={strategy}, total_cost=${total_cost:.2f})"
|
|
)
|
|
|
|
return result
|
|
|
|
def _score_suppliers(
|
|
self,
|
|
suppliers: List[SupplierOption]
|
|
) -> List[Tuple[SupplierOption, float, Dict[str, float]]]:
|
|
"""
|
|
Score all suppliers using weighted criteria.
|
|
|
|
Args:
|
|
suppliers: List of supplier options
|
|
|
|
Returns:
|
|
List of (supplier, score, score_breakdown) tuples
|
|
"""
|
|
if not suppliers:
|
|
return []
|
|
|
|
# Normalize factors for comparison
|
|
prices = [s.unit_price for s in suppliers]
|
|
lead_times = [s.lead_time_days for s in suppliers]
|
|
|
|
min_price = min(prices)
|
|
max_price = max(prices)
|
|
min_lead_time = min(lead_times)
|
|
max_lead_time = max(lead_times)
|
|
|
|
scored = []
|
|
|
|
for supplier in suppliers:
|
|
# Price score (normalized, lower is better)
|
|
if max_price > min_price:
|
|
price_score = 1.0 - float((supplier.unit_price - min_price) / (max_price - min_price))
|
|
else:
|
|
price_score = 1.0
|
|
|
|
# Lead time score (normalized, shorter is better)
|
|
if max_lead_time > min_lead_time:
|
|
lead_time_score = 1.0 - (supplier.lead_time_days - min_lead_time) / (max_lead_time - min_lead_time)
|
|
else:
|
|
lead_time_score = 1.0
|
|
|
|
# Quality and reliability scores (already 0-1)
|
|
quality_score = supplier.quality_score
|
|
reliability_score = supplier.reliability_score
|
|
|
|
# Calculate weighted total score
|
|
total_score = (
|
|
self.price_weight * price_score +
|
|
self.lead_time_weight * lead_time_score +
|
|
self.quality_weight * quality_score +
|
|
self.reliability_weight * reliability_score
|
|
)
|
|
|
|
score_breakdown = {
|
|
'price_score': price_score,
|
|
'lead_time_score': lead_time_score,
|
|
'quality_score': quality_score,
|
|
'reliability_score': reliability_score,
|
|
'total_score': total_score
|
|
}
|
|
|
|
scored.append((supplier, total_score, score_breakdown))
|
|
|
|
# Sort by score (descending)
|
|
scored.sort(key=lambda x: x[1], reverse=True)
|
|
|
|
return scored
|
|
|
|
def _determine_strategy(
|
|
self,
|
|
required_quantity: Decimal,
|
|
suppliers: List[SupplierOption]
|
|
) -> str:
|
|
"""
|
|
Determine selection strategy based on quantity and options.
|
|
|
|
Args:
|
|
required_quantity: Quantity needed
|
|
suppliers: Available suppliers
|
|
|
|
Returns:
|
|
Strategy: 'single_source', 'dual_source', or 'multi_source'
|
|
"""
|
|
if len(suppliers) == 1:
|
|
return 'single_source'
|
|
|
|
# Large orders should be diversified
|
|
if required_quantity >= self.diversification_threshold:
|
|
return 'multi_source' if len(suppliers) >= 3 else 'dual_source'
|
|
|
|
# Small orders: single source unless quality/reliability concerns
|
|
avg_reliability = sum(s.reliability_score for s in suppliers) / len(suppliers)
|
|
if avg_reliability < 0.85:
|
|
return 'dual_source' # Use backup for unreliable suppliers
|
|
|
|
return 'single_source'
|
|
|
|
def _select_single_source(
|
|
self,
|
|
required_quantity: Decimal,
|
|
scored_suppliers: List[Tuple[SupplierOption, float, Dict[str, float]]]
|
|
) -> List[SupplierAllocation]:
|
|
"""
|
|
Select single best supplier.
|
|
|
|
Args:
|
|
required_quantity: Quantity needed
|
|
scored_suppliers: Scored suppliers
|
|
|
|
Returns:
|
|
List with single allocation
|
|
"""
|
|
best_supplier, score, score_breakdown = scored_suppliers[0]
|
|
|
|
# Check capacity
|
|
if best_supplier.max_capacity and required_quantity > best_supplier.max_capacity:
|
|
logger.warning(
|
|
f"{best_supplier.supplier_name}: Required quantity {required_quantity} "
|
|
f"exceeds capacity {best_supplier.max_capacity}, will need to split"
|
|
)
|
|
# Fall back to dual source
|
|
return self._select_dual_source(required_quantity, scored_suppliers)
|
|
|
|
allocation = SupplierAllocation(
|
|
supplier_id=best_supplier.supplier_id,
|
|
supplier_name=best_supplier.supplier_name,
|
|
allocated_quantity=required_quantity,
|
|
allocation_percentage=1.0,
|
|
allocation_type='primary',
|
|
unit_price=best_supplier.unit_price,
|
|
total_cost=best_supplier.unit_price * required_quantity,
|
|
lead_time_days=best_supplier.lead_time_days,
|
|
supplier_score=score,
|
|
score_breakdown=score_breakdown,
|
|
allocation_reason='Best overall score (single source strategy)'
|
|
)
|
|
|
|
return [allocation]
|
|
|
|
def _select_dual_source(
|
|
self,
|
|
required_quantity: Decimal,
|
|
scored_suppliers: List[Tuple[SupplierOption, float, Dict[str, float]]]
|
|
) -> List[SupplierAllocation]:
|
|
"""
|
|
Select primary supplier + backup.
|
|
|
|
Args:
|
|
required_quantity: Quantity needed
|
|
scored_suppliers: Scored suppliers
|
|
|
|
Returns:
|
|
List with two allocations
|
|
"""
|
|
if len(scored_suppliers) < 2:
|
|
return self._select_single_source(required_quantity, scored_suppliers)
|
|
|
|
primary_supplier, primary_score, primary_breakdown = scored_suppliers[0]
|
|
backup_supplier, backup_score, backup_breakdown = scored_suppliers[1]
|
|
|
|
# Primary gets 70%, backup gets 30%
|
|
primary_percentage = self.max_single_supplier_percentage
|
|
backup_percentage = 1.0 - primary_percentage
|
|
|
|
primary_qty = required_quantity * Decimal(str(primary_percentage))
|
|
backup_qty = required_quantity * Decimal(str(backup_percentage))
|
|
|
|
# Check capacities
|
|
if primary_supplier.max_capacity and primary_qty > primary_supplier.max_capacity:
|
|
# Rebalance
|
|
primary_qty = primary_supplier.max_capacity
|
|
backup_qty = required_quantity - primary_qty
|
|
primary_percentage = float(primary_qty / required_quantity)
|
|
backup_percentage = float(backup_qty / required_quantity)
|
|
|
|
allocations = [
|
|
SupplierAllocation(
|
|
supplier_id=primary_supplier.supplier_id,
|
|
supplier_name=primary_supplier.supplier_name,
|
|
allocated_quantity=primary_qty,
|
|
allocation_percentage=primary_percentage,
|
|
allocation_type='primary',
|
|
unit_price=primary_supplier.unit_price,
|
|
total_cost=primary_supplier.unit_price * primary_qty,
|
|
lead_time_days=primary_supplier.lead_time_days,
|
|
supplier_score=primary_score,
|
|
score_breakdown=primary_breakdown,
|
|
allocation_reason=f'Primary supplier ({primary_percentage*100:.0f}% allocation)'
|
|
),
|
|
SupplierAllocation(
|
|
supplier_id=backup_supplier.supplier_id,
|
|
supplier_name=backup_supplier.supplier_name,
|
|
allocated_quantity=backup_qty,
|
|
allocation_percentage=backup_percentage,
|
|
allocation_type='backup',
|
|
unit_price=backup_supplier.unit_price,
|
|
total_cost=backup_supplier.unit_price * backup_qty,
|
|
lead_time_days=backup_supplier.lead_time_days,
|
|
supplier_score=backup_score,
|
|
score_breakdown=backup_breakdown,
|
|
allocation_reason=f'Backup supplier ({backup_percentage*100:.0f}% allocation for risk mitigation)'
|
|
)
|
|
]
|
|
|
|
return allocations
|
|
|
|
def _select_multi_source(
|
|
self,
|
|
required_quantity: Decimal,
|
|
scored_suppliers: List[Tuple[SupplierOption, float, Dict[str, float]]]
|
|
) -> List[SupplierAllocation]:
|
|
"""
|
|
Split across multiple suppliers for large orders.
|
|
|
|
Args:
|
|
required_quantity: Quantity needed
|
|
scored_suppliers: Scored suppliers
|
|
|
|
Returns:
|
|
List with multiple allocations
|
|
"""
|
|
if len(scored_suppliers) < 3:
|
|
return self._select_dual_source(required_quantity, scored_suppliers)
|
|
|
|
# Use top 3 suppliers
|
|
top_3 = scored_suppliers[:3]
|
|
|
|
# Allocate proportionally to scores
|
|
total_score = sum(score for _, score, _ in top_3)
|
|
|
|
allocations = []
|
|
remaining_qty = required_quantity
|
|
|
|
for i, (supplier, score, score_breakdown) in enumerate(top_3):
|
|
if i == len(top_3) - 1:
|
|
# Last supplier gets remainder
|
|
allocated_qty = remaining_qty
|
|
else:
|
|
# Allocate based on score proportion
|
|
proportion = score / total_score
|
|
allocated_qty = required_quantity * Decimal(str(proportion))
|
|
|
|
# Check capacity
|
|
if supplier.max_capacity and allocated_qty > supplier.max_capacity:
|
|
allocated_qty = supplier.max_capacity
|
|
|
|
allocation_percentage = float(allocated_qty / required_quantity)
|
|
|
|
allocation = SupplierAllocation(
|
|
supplier_id=supplier.supplier_id,
|
|
supplier_name=supplier.supplier_name,
|
|
allocated_quantity=allocated_qty,
|
|
allocation_percentage=allocation_percentage,
|
|
allocation_type='diversification',
|
|
unit_price=supplier.unit_price,
|
|
total_cost=supplier.unit_price * allocated_qty,
|
|
lead_time_days=supplier.lead_time_days,
|
|
supplier_score=score,
|
|
score_breakdown=score_breakdown,
|
|
allocation_reason=f'Multi-source diversification ({allocation_percentage*100:.0f}%)'
|
|
)
|
|
|
|
allocations.append(allocation)
|
|
remaining_qty -= allocated_qty
|
|
|
|
if remaining_qty <= 0:
|
|
break
|
|
|
|
return allocations
|
|
|
|
def _calculate_risk_score(
|
|
self,
|
|
allocations: List[SupplierAllocation]
|
|
) -> float:
|
|
"""
|
|
Calculate overall risk score (lower is better).
|
|
|
|
Args:
|
|
allocations: List of allocations
|
|
|
|
Returns:
|
|
Risk score (0-1)
|
|
"""
|
|
if not allocations:
|
|
return 1.0
|
|
|
|
# Single source = higher risk
|
|
diversification_risk = 1.0 / len(allocations)
|
|
|
|
# Concentration risk (how much in single supplier)
|
|
max_allocation = max(alloc.allocation_percentage for alloc in allocations)
|
|
concentration_risk = max_allocation
|
|
|
|
# Reliability risk (average of supplier reliability)
|
|
# Note: We don't have reliability in SupplierAllocation, estimate from score
|
|
avg_supplier_score = sum(alloc.supplier_score for alloc in allocations) / len(allocations)
|
|
reliability_risk = 1.0 - avg_supplier_score
|
|
|
|
# Combined risk (weighted)
|
|
risk_score = (
|
|
0.4 * diversification_risk +
|
|
0.3 * concentration_risk +
|
|
0.3 * reliability_risk
|
|
)
|
|
|
|
return risk_score
|
|
|
|
def export_result_to_dict(self, result: SupplierSelectionResult) -> Dict:
|
|
"""
|
|
Export result to dictionary for API response.
|
|
|
|
Args:
|
|
result: Supplier selection result
|
|
|
|
Returns:
|
|
Dictionary representation
|
|
"""
|
|
return {
|
|
'ingredient_id': result.ingredient_id,
|
|
'ingredient_name': result.ingredient_name,
|
|
'required_quantity': float(result.required_quantity),
|
|
'total_cost': float(result.total_cost),
|
|
'weighted_lead_time': result.weighted_lead_time,
|
|
'risk_score': result.risk_score,
|
|
'diversification_applied': result.diversification_applied,
|
|
'selection_strategy': result.selection_strategy,
|
|
'allocations': [
|
|
{
|
|
'supplier_id': alloc.supplier_id,
|
|
'supplier_name': alloc.supplier_name,
|
|
'allocated_quantity': float(alloc.allocated_quantity),
|
|
'allocation_percentage': alloc.allocation_percentage,
|
|
'allocation_type': alloc.allocation_type,
|
|
'unit_price': float(alloc.unit_price),
|
|
'total_cost': float(alloc.total_cost),
|
|
'lead_time_days': alloc.lead_time_days,
|
|
'supplier_score': alloc.supplier_score,
|
|
'score_breakdown': alloc.score_breakdown,
|
|
'allocation_reason': alloc.allocation_reason
|
|
}
|
|
for alloc in result.allocations
|
|
]
|
|
}
|