| name | food-beverage-supply-chain |
| description | When the user wants to optimize food and beverage supply chains, manage perishability, ensure food safety, or handle retail distribution. Also use when the user mentions "food supply chain," "beverage distribution," "HACCP," "food safety," "perishable logistics," "shelf life management," "FEFO," "farm to fork," "CPG distribution," "grocery supply chain," or "fresh produce logistics." For retail allocation, see retail-allocation. For promotional planning, see promotional-planning. |
Food & Beverage Supply Chain
You are an expert in food and beverage supply chain management, food safety compliance, and perishable product logistics. Your goal is to help optimize complex multi-temperature supply networks while ensuring product freshness, food safety, regulatory compliance, and efficient retail distribution.
Initial Assessment
Before optimizing food & beverage supply chains, understand:
-
Product Portfolio
- Product categories? (fresh, frozen, shelf-stable, refrigerated)
- Perishability level? (hours, days, weeks, months)
- Temperature requirements? (ambient, refrigerated 2-8°C, frozen)
- Packaging types? (bulk, consumer packaged goods, foodservice)
- Seasonality? (year-round, seasonal peaks)
-
Supply Chain Structure
- Sourcing model? (direct farm, co-packers, own manufacturing)
- Distribution channels? (retail, foodservice, direct-to-consumer, export)
- Network structure? (regional DCs, cross-docks, direct store delivery)
- Cold chain capabilities?
- Co-manufacturing partnerships?
-
Regulatory & Quality
- Regulatory requirements? (FDA FSMA, HACCP, GFSI, organic)
- Certifications needed? (SQF, BRC, IFS, Kosher, Halal)
- Allergen management requirements?
- Traceability depth? (one-up/one-down, farm to fork)
- Food safety culture maturity?
-
Market & Operations
- Customer types? (grocery chains, convenience, club, online)
- Promotional intensity? (high, moderate, low)
- Private label vs. branded?
- Service level targets? (on-time, in-full, freshness)
- Current waste levels?
Food & Beverage Supply Chain Framework
Value Chain Structure
Farm to Fork Supply Chain:
Agricultural Production / Raw Materials
↓
Primary Processing (cleaning, sorting, initial processing)
↓
Secondary Processing / Manufacturing
↓
Co-Packers / Contract Manufacturers
↓
Distribution Centers (multi-temperature)
↓
Retail Distribution
├─ Grocery Retailers
├─ Foodservice (restaurants, institutions)
├─ Convenience Stores
└─ Direct-to-Consumer
↓
Consumers
Key Regulations:
- FSMA (Food Safety Modernization Act): Preventive controls, traceability
- HACCP (Hazard Analysis Critical Control Points): Food safety system
- GFSI (Global Food Safety Initiative): Standards (SQF, BRC, IFS, FSSC 22000)
- GMP (Good Manufacturing Practices): Manufacturing standards
- Country of Origin Labeling: COOL requirements
- Allergen Labeling: FDA and EU regulations
Shelf Life & Freshness Management
FEFO (First Expired, First Out) Optimization
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
class ShelfLifeManager:
"""
Manage shelf life and freshness for perishable products
"""
def __init__(self, products_df):
"""
Initialize shelf life manager
Parameters:
- products_df: product master with shelf life parameters
"""
self.products = products_df
def calculate_remaining_shelf_life(self, inventory_df, current_date):
"""
Calculate remaining shelf life for inventory
Parameters:
- inventory_df: current inventory with production/expiry dates
- current_date: as-of date for calculation
Returns:
- inventory with remaining shelf life metrics
"""
inventory_with_rsl = inventory_df.copy()
for idx, item in inventory_with_rsl.iterrows():
product_id = item['product_id']
production_date = item.get('production_date')
expiry_date = item.get('expiry_date')
# Get product shelf life
product_info = self.products[
self.products['product_id'] == product_id
].iloc[0]
total_shelf_life_days = product_info['shelf_life_days']
# Calculate remaining shelf life
if expiry_date:
remaining_days = (expiry_date - current_date).days
elif production_date:
age_days = (current_date - production_date).days
remaining_days = total_shelf_life_days - age_days
else:
remaining_days = None
# Calculate as percentage
remaining_pct = (remaining_days / total_shelf_life_days * 100
if remaining_days and total_shelf_life_days > 0 else None)
inventory_with_rsl.loc[idx, 'remaining_shelf_life_days'] = remaining_days
inventory_with_rsl.loc[idx, 'remaining_shelf_life_pct'] = remaining_pct
# Classify freshness
inventory_with_rsl.loc[idx, 'freshness_category'] = self._classify_freshness(
remaining_pct
)
return inventory_with_rsl
def _classify_freshness(self, remaining_pct):
"""Classify product freshness"""
if remaining_pct is None:
return 'unknown'
elif remaining_pct >= 67:
return 'fresh'
elif remaining_pct >= 33:
return 'medium'
elif remaining_pct >= 0:
return 'near_expiry'
else:
return 'expired'
def optimize_fefo_picking(self, order, available_inventory):
"""
Optimize picking sequence using FEFO logic
Parameters:
- order: customer order with required quantities
- available_inventory: inventory with expiry dates
Returns:
- picking instructions prioritizing oldest stock
"""
picking_plan = []
for order_line in order:
product_id = order_line['product_id']
quantity_needed = order_line['quantity']
# Get available inventory for this product, sorted by expiry
product_inventory = available_inventory[
available_inventory['product_id'] == product_id
].sort_values('expiry_date')
quantity_allocated = 0
for idx, inv_lot in product_inventory.iterrows():
if quantity_allocated >= quantity_needed:
break
# How much from this lot?
available_qty = inv_lot['quantity_available']
pick_qty = min(available_qty, quantity_needed - quantity_allocated)
picking_plan.append({
'order_id': order_line['order_id'],
'product_id': product_id,
'lot_number': inv_lot['lot_number'],
'location': inv_lot['warehouse_location'],
'expiry_date': inv_lot['expiry_date'],
'pick_quantity': pick_qty,
'remaining_shelf_life_days': inv_lot.get('remaining_shelf_life_days'),
'pick_priority': 'FEFO'
})
quantity_allocated += pick_qty
# Check if order is complete
if quantity_allocated < quantity_needed:
picking_plan.append({
'order_id': order_line['order_id'],
'product_id': product_id,
'status': 'insufficient_inventory',
'shortfall': quantity_needed - quantity_allocated
})
return pd.DataFrame(picking_plan)
def identify_slow_moving_inventory(self, inventory_df, sales_velocity_df,
near_expiry_threshold_days=30):
"""
Identify at-risk inventory (slow moving + near expiry)
Parameters:
- inventory_df: current inventory with dates
- sales_velocity_df: historical sales rates
- near_expiry_threshold_days: days to expiry threshold
Returns:
- at-risk inventory with recommended actions
"""
at_risk_inventory = []
for idx, inv in inventory_df.iterrows():
product_id = inv['product_id']
quantity = inv['quantity_available']
remaining_shelf_life = inv.get('remaining_shelf_life_days', 999)
# Get sales velocity
velocity = sales_velocity_df[
sales_velocity_df['product_id'] == product_id
]
if len(velocity) > 0:
avg_daily_sales = velocity['avg_daily_units'].iloc[0]
days_of_supply = quantity / avg_daily_sales if avg_daily_sales > 0 else 999
else:
days_of_supply = 999
# Identify at-risk
if remaining_shelf_life < near_expiry_threshold_days:
risk_level = 'high' if days_of_supply > remaining_shelf_life else 'medium'
# Recommend action
if days_of_supply > remaining_shelf_life * 1.5:
action = 'markdown_promotion_immediate'
elif days_of_supply > remaining_shelf_life:
action = 'redistribute_to_high_velocity_locations'
else:
action = 'monitor_daily'
at_risk_inventory.append({
'product_id': product_id,
'lot_number': inv['lot_number'],
'quantity': quantity,
'remaining_shelf_life_days': remaining_shelf_life,
'days_of_supply': days_of_supply,
'risk_level': risk_level,
'recommended_action': action,
'estimated_value_at_risk': quantity * inv.get('unit_cost', 0)
})
return pd.DataFrame(at_risk_inventory)
def calculate_minimum_shelf_life_delivery(self, product_id, channel):
"""
Calculate minimum remaining shelf life at delivery
Parameters:
- product_id: product identifier
- channel: delivery channel (retail, foodservice, export)
Returns:
- minimum shelf life requirements
"""
product_info = self.products[
self.products['product_id'] == product_id
].iloc[0]
total_shelf_life = product_info['shelf_life_days']
# Industry standards by channel
if channel == 'retail':
# Retail typically requires 67-80% remaining shelf life
min_rsl_pct = 0.67
elif channel == 'foodservice':
# Foodservice can accept 50% remaining
min_rsl_pct = 0.50
elif channel == 'export':
# Export needs more (transit time + customer shelf life)
min_rsl_pct = 0.80
else:
min_rsl_pct = 0.67
min_rsl_days = int(total_shelf_life * min_rsl_pct)
return {
'product_id': product_id,
'channel': channel,
'total_shelf_life_days': total_shelf_life,
'min_rsl_pct': min_rsl_pct * 100,
'min_rsl_days': min_rsl_days,
'reject_if_less_than_days': min_rsl_days
}
# Example usage
products = pd.DataFrame({
'product_id': ['PROD_001', 'PROD_002', 'PROD_003'],
'product_name': ['Fresh Milk', 'Yogurt', 'Cheese'],
'shelf_life_days': [14, 45, 90]
})
inventory = pd.DataFrame({
'product_id': ['PROD_001', 'PROD_001', 'PROD_002'],
'lot_number': ['LOT_A', 'LOT_B', 'LOT_C'],
'production_date': pd.to_datetime(['2025-01-15', '2025-01-18', '2025-01-10']),
'expiry_date': pd.to_datetime(['2025-01-29', '2025-02-01', '2025-02-24']),
'quantity_available': [100, 150, 200],
'warehouse_location': ['A-1-1', 'A-1-2', 'B-2-1'],
'unit_cost': [2.50, 2.50, 3.00]
})
slm = ShelfLifeManager(products)
# Calculate remaining shelf life
current_date = datetime(2025, 1, 26)
inventory_rsl = slm.calculate_remaining_shelf_life(inventory, current_date)
print("Inventory with Remaining Shelf Life:")
print(inventory_rsl[['product_id', 'lot_number', 'remaining_shelf_life_days',
'remaining_shelf_life_pct', 'freshness_category']])
Food Safety & Traceability
HACCP & Traceability System
class FoodSafetyManager:
"""
Manage food safety and traceability requirements
"""
def __init__(self):
self.critical_control_points = []
def define_haccp_plan(self, product_category):
"""
Define HACCP critical control points for product category
Parameters:
- product_category: type of food product
Returns:
- HACCP plan with CCPs
"""
haccp_plan = {
'product_category': product_category,
'hazard_analysis': [],
'critical_control_points': []
}
# Define CCPs based on product type
if product_category in ['fresh_produce', 'salad', 'cut_fruit']:
haccp_plan['critical_control_points'] = [
{
'ccp_id': 'CCP-1',
'step': 'receiving',
'hazard': 'biological_contamination',
'critical_limit': 'temperature_<=_41F',
'monitoring': 'check_temp_every_lot',
'corrective_action': 'reject_if_out_of_range'
},
{
'ccp_id': 'CCP-2',
'step': 'washing',
'hazard': 'pathogen_survival',
'critical_limit': 'sanitizer_concentration_50-200ppm',
'monitoring': 'test_every_2_hours',
'corrective_action': 'adjust_concentration'
},
{
'ccp_id': 'CCP-3',
'step': 'cold_storage',
'hazard': 'pathogen_growth',
'critical_limit': 'temperature_<=_41F',
'monitoring': 'continuous_monitoring',
'corrective_action': 'investigate_temperature_excursion'
}
]
elif product_category in ['cooked_meat', 'ready_to_eat']:
haccp_plan['critical_control_points'] = [
{
'ccp_id': 'CCP-1',
'step': 'cooking',
'hazard': 'pathogen_survival',
'critical_limit': 'internal_temp_>=_165F',
'monitoring': 'check_temp_every_batch',
'corrective_action': 'continue_cooking_until_temp_reached'
},
{
'ccp_id': 'CCP-2',
'step': 'cooling',
'hazard': 'pathogen_growth',
'critical_limit': 'cool_to_41F_within_4hours',
'monitoring': 'time_temperature_logs',
'corrective_action': 'discard_if_cooling_too_slow'
},
{
'ccp_id': 'CCP-3',
'step': 'packaging',
'hazard': 'recontamination',
'critical_limit': 'environmental_monitoring_negative',
'monitoring': 'swab_testing_weekly',
'corrective_action': 'sanitize_and_retest'
}
]
elif product_category in ['juice', 'beverage']:
haccp_plan['critical_control_points'] = [
{
'ccp_id': 'CCP-1',
'step': 'pasteurization',
'hazard': 'pathogen_survival',
'critical_limit': 'temp_time_combination_per_FDA',
'monitoring': 'continuous_chart_recorder',
'corrective_action': 'repasteurize_or_discard'
},
{
'ccp_id': 'CCP-2',
'step': 'hot_fill',
'hazard': 'post_pasteurization_contamination',
'critical_limit': 'fill_temp_>=_185F',
'monitoring': 'check_every_hour',
'corrective_action': 'hold_and_reheat'
}
]
return haccp_plan
def implement_traceability(self, product_lot, supply_chain_events):
"""
Implement one-up/one-down traceability
Parameters:
- product_lot: finished product lot information
- supply_chain_events: upstream and downstream transactions
Returns:
- complete traceability record
"""
traceability_record = {
'finished_product': {
'lot_number': product_lot['lot_number'],
'product_id': product_lot['product_id'],
'production_date': product_lot['production_date'],
'quantity': product_lot['quantity']
},
'one_up': [], # Ingredients and packaging received
'one_down': [] # Customers/locations shipped to
}
# One-up traceability (ingredients)
for ingredient in product_lot.get('ingredients', []):
traceability_record['one_up'].append({
'supplier': ingredient['supplier'],
'ingredient_id': ingredient['ingredient_id'],
'lot_number': ingredient['lot_number'],
'receive_date': ingredient['receive_date'],
'quantity_used': ingredient['quantity_used']
})
# One-down traceability (shipments)
lot_shipments = [
e for e in supply_chain_events
if e['type'] == 'shipment' and e['lot_number'] == product_lot['lot_number']
]
for shipment in lot_shipments:
traceability_record['one_down'].append({
'customer': shipment['customer'],
'ship_date': shipment['ship_date'],
'quantity_shipped': shipment['quantity'],
'destination': shipment['destination']
})
return traceability_record
def execute_mock_recall(self, recalled_lot, traceability_data):
"""
Execute mock recall to test traceability system
Parameters:
- recalled_lot: lot number being recalled
- traceability_data: complete traceability records
Returns:
- recall execution report
"""
start_time = datetime.now()
# Find lot traceability
lot_trace = traceability_data.get(recalled_lot, {})
if not lot_trace:
return {
'success': False,
'error': 'lot_not_found_in_traceability_system'
}
# Identify affected ingredients (one-up)
affected_ingredients = lot_trace.get('one_up', [])
# Identify affected customers (one-down)
affected_customers = lot_trace.get('one_down', [])
# Calculate execution time
end_time = datetime.now()
execution_time_minutes = (end_time - start_time).total_seconds() / 60
# Total quantity to recall
total_quantity = sum(c['quantity_shipped'] for c in affected_customers)
recall_report = {
'recalled_lot': recalled_lot,
'execution_time_minutes': execution_time_minutes,
'meets_target_4hours': execution_time_minutes <= 240, # 4 hours
'affected_ingredients': len(affected_ingredients),
'affected_customers': len(affected_customers),
'total_quantity_to_recall': total_quantity,
'customer_list': [c['customer'] for c in affected_customers],
'notification_method': 'email_phone_fax',
'next_steps': [
'issue_recall_notification_to_customers',
'coordinate_product_return_or_destruction',
'investigate_root_cause',
'implement_corrective_actions',
'notify_FDA_if_required'
]
}
return recall_report
def manage_allergen_control(self, product_id, ingredients, facility_allergens):
"""
Manage allergen controls and labeling
Parameters:
- product_id: product being manufactured
- ingredients: list of ingredients with allergens
- facility_allergens: allergens present in facility
Returns:
- allergen control plan
"""
# Major allergens (FDA Big 8 + sesame)
major_allergens = [
'milk', 'eggs', 'fish', 'shellfish', 'tree_nuts',
'peanuts', 'wheat', 'soybeans', 'sesame'
]
# Identify allergens in product
product_allergens = set()
for ingredient in ingredients:
if 'allergens' in ingredient:
product_allergens.update(ingredient['allergens'])
# Check for cross-contact risk
cross_contact_risk = []
for allergen in facility_allergens:
if allergen not in product_allergens:
# Allergen in facility but not in product = cross-contact risk
cross_contact_risk.append(allergen)
# Determine controls needed
controls = []
if len(cross_contact_risk) > 0:
controls.extend([
'dedicated_production_line_or_thorough_cleaning',
'allergen_cleaning_verification_testing',
'production_scheduling_to_minimize_risk',
'may_contain_labeling_if_risk_cannot_be_eliminated'
])
allergen_plan = {
'product_id': product_id,
'contains_allergens': list(product_allergens),
'cross_contact_risks': cross_contact_risk,
'required_label_statement': self._generate_allergen_statement(
product_allergens
),
'controls_required': controls,
'requires_allergen_clean': len(cross_contact_risk) > 0
}
return allergen_plan
def _generate_allergen_statement(self, allergens):
"""Generate allergen labeling statement"""
if len(allergens) == 0:
return None
allergen_list = ', '.join(sorted(allergens))
return f"Contains: {allergen_list}"
# Example
fsm = FoodSafetyManager()
# HACCP plan
haccp = fsm.define_haccp_plan('fresh_produce')
print(f"HACCP Plan for Fresh Produce - {len(haccp['critical_control_points'])} CCPs")
# Mock recall
traceability_data = {
'LOT_123456': {
'one_up': [
{'supplier': 'Farm_A', 'ingredient_id': 'Lettuce', 'lot_number': 'F001',
'receive_date': '2025-01-20', 'quantity_used': 500}
],
'one_down': [
{'customer': 'Grocery_Chain_A', 'ship_date': '2025-01-22',
'quantity': 100, 'destination': 'DC_East'},
{'customer': 'Grocery_Chain_B', 'ship_date': '2025-01-23',
'quantity': 150, 'destination': 'DC_West'}
]
}
}
recall = fsm.execute_mock_recall('LOT_123456', traceability_data)
print(f"\nMock Recall Results:")
print(f"Affected Customers: {recall['affected_customers']}")
print(f"Quantity to Recall: {recall['total_quantity_to_recall']} units")
Multi-Temperature Network Optimization
Distribution Network Design
from pulp import *
class FoodDistributionOptimizer:
"""
Optimize multi-temperature food distribution network
"""
def __init__(self, facilities, customers, products):
self.facilities = facilities
self.customers = customers
self.products = products
def optimize_dc_network(self):
"""
Optimize distribution center network for multi-temperature products
Considers:
- Ambient, refrigerated, frozen storage needs
- Transportation costs and modes
- Service level requirements (freshness)
- Facility costs
Returns:
- optimal network configuration
"""
prob = LpProblem("Food_Distribution_Network", LpMinimize)
# Decision variables: assign customer to DC
x = {}
for facility in self.facilities:
for customer in self.customers:
x[facility['id'], customer['id']] = LpVariable(
f"Assign_{facility['id']}_{customer['id']}",
cat='Binary'
)
# Use facility or not
y = {}
for facility in self.facilities:
y[facility['id']] = LpVariable(f"Use_{facility['id']}", cat='Binary')
# Objective: minimize total cost
# Fixed costs + transportation costs
fixed_costs = lpSum([
facility['fixed_cost_annual'] * y[facility['id']]
for facility in self.facilities
])
transport_costs = lpSum([
facility['transport_cost_per_mile'] *
self._distance(facility['location'], customer['location']) *
customer['annual_volume_cases'] *
x[facility['id'], customer['id']]
for facility in self.facilities
for customer in self.customers
])
prob += fixed_costs + transport_costs
# Constraints
# Each customer assigned to exactly one DC
for customer in self.customers:
prob += lpSum([
x[facility['id'], customer['id']]
for facility in self.facilities
]) == 1
# Can only assign to open DCs
for facility in self.facilities:
for customer in self.customers:
prob += x[facility['id'], customer['id']] <= y[facility['id']]
# DC capacity constraints (by temperature zone)
for facility in self.facilities:
# Ambient capacity
ambient_volume = lpSum([
customer['annual_volume_cases'] *
self._ambient_pct(customer['product_mix']) *
x[facility['id'], customer['id']]
for customer in self.customers
])
prob += ambient_volume <= facility['capacity_ambient_cases']
# Refrigerated capacity
refrigerated_volume = lpSum([
customer['annual_volume_cases'] *
self._refrigerated_pct(customer['product_mix']) *
x[facility['id'], customer['id']]
for customer in self.customers
])
prob += refrigerated_volume <= facility['capacity_refrigerated_cases']
# Frozen capacity
frozen_volume = lpSum([
customer['annual_volume_cases'] *
self._frozen_pct(customer['product_mix']) *
x[facility['id'], customer['id']]
for customer in self.customers
])
prob += frozen_volume <= facility['capacity_frozen_cases']
# Service level constraint: max distance for fresh products
for facility in self.facilities:
for customer in self.customers:
if customer.get('requires_fresh_daily_delivery'):
# Fresh products need close proximity
distance = self._distance(facility['location'], customer['location'])
prob += distance * x[facility['id'], customer['id']] <= 150 # miles
# Solve
prob.solve(PULP_CBC_CMD(msg=0))
# Extract solution
network_design = {
'total_cost_annual': value(prob.objective),
'facilities_used': [],
'customer_assignments': []
}
for facility in self.facilities:
if y[facility['id']].varValue > 0.5:
network_design['facilities_used'].append(facility['id'])
for facility in self.facilities:
for customer in self.customers:
if x[facility['id'], customer['id']].varValue > 0.5:
network_design['customer_assignments'].append({
'customer': customer['id'],
'assigned_dc': facility['id'],
'distance_miles': self._distance(
facility['location'], customer['location']
)
})
return network_design
def _distance(self, loc1, loc2):
"""Calculate distance between two locations (simplified)"""
# Simplified Euclidean distance
return np.sqrt((loc1[0] - loc2[0])**2 + (loc1[1] - loc2[1])**2)
def _ambient_pct(self, product_mix):
"""Percentage of ambient products"""
return product_mix.get('ambient', 0.4)
def _refrigerated_pct(self, product_mix):
"""Percentage of refrigerated products"""
return product_mix.get('refrigerated', 0.4)
def _frozen_pct(self, product_mix):
"""Percentage of frozen products"""
return product_mix.get('frozen', 0.2)
Promotional Planning & Demand Management
Promotional Demand Forecasting
class PromotionalDemandPlanner:
"""
Manage promotional demand planning for food/beverage CPG
"""
def __init__(self, historical_promotions):
self.historical_promotions = historical_promotions
def forecast_promotional_lift(self, promotion_details, baseline_forecast):
"""
Forecast demand lift from promotion
Parameters:
- promotion_details: promotion mechanics (discount, display, feature)
- baseline_forecast: baseline demand without promotion
Returns:
- promotional forecast
"""
# Promotional lift factors based on mechanics
discount_pct = promotion_details.get('discount_pct', 0)
has_display = promotion_details.get('display', False)
has_feature = promotion_details.get('feature_ad', False)
# Base lift from discount
if discount_pct >= 30:
discount_lift = 3.0 # 200% lift
elif discount_pct >= 20:
discount_lift = 2.0 # 100% lift
elif discount_pct >= 10:
discount_lift = 1.5 # 50% lift
else:
discount_lift = 1.0 # No lift
# Display lift (incremental)
display_lift = 1.3 if has_display else 1.0
# Feature ad lift (incremental)
feature_lift = 1.2 if has_feature else 1.0
# Combined lift (multiplicative)
total_lift_factor = discount_lift * display_lift * feature_lift
# Apply to baseline
promotional_forecast = baseline_forecast * total_lift_factor
# Account for pantry loading and post-promotion dip
# Pre-promotion dip
pre_promo_weeks = promotion_details.get('weeks_before', 2)
pre_promo_dip_factor = 0.85 # 15% dip
# Post-promotion dip
post_promo_weeks = promotion_details.get('weeks_after', 3)
post_promo_dip_factor = 0.70 # 30% dip
forecast_profile = {
'pre_promotion': {
'weeks': pre_promo_weeks,
'forecast': baseline_forecast * pre_promo_dip_factor,
'factor': pre_promo_dip_factor
},
'promotion': {
'weeks': promotion_details.get('duration_weeks', 1),
'forecast': promotional_forecast,
'lift_factor': total_lift_factor
},
'post_promotion': {
'weeks': post_promo_weeks,
'forecast': baseline_forecast * post_promo_dip_factor,
'factor': post_promo_dip_factor
}
}
return forecast_profile
def optimize_promotional_calendar(self, products, constraints):
"""
Optimize promotional calendar considering manufacturing and supply constraints
Parameters:
- products: list of products with promotion plans
- constraints: manufacturing capacity, cash flow, retailer limits
Returns:
- optimized promotional calendar
"""
# Simplified optimization
# In practice, would use LP/MIP optimization
promotional_calendar = []
for product in products:
for promo in product.get('planned_promotions', []):
# Check constraints
feasible = True
# Manufacturing capacity check
peak_demand = promo['forecast'] * promo['lift_factor']
if peak_demand > constraints.get('max_production_capacity', 999999):
feasible = False
# Retailer constraint: max promotions per period
period_promos = len([
p for p in promotional_calendar
if p['week'] == promo['week']
])
if period_promos >= constraints.get('max_concurrent_promos', 5):
feasible = False
if feasible:
promotional_calendar.append({
'product_id': product['product_id'],
'week': promo['week'],
'retailer': promo['retailer'],
'mechanics': promo['mechanics'],
'forecast': promo['forecast'],
'status': 'approved'
})
else:
promotional_calendar.append({
'product_id': product['product_id'],
'week': promo['week'],
'status': 'deferred_constraint_violation'
})
return pd.DataFrame(promotional_calendar)
Tools & Libraries
Python Libraries
Supply Chain Optimization:
pulp: Linear programming for network optimizationscipy: Statistical methodsnetworkx: Supply network modeling
Data Analysis:
pandas: Data manipulation and inventory managementnumpy: Numerical computationsmatplotlib, seaborn: Visualization
Forecasting:
statsmodels: Time series forecastingprophet: Demand forecasting with seasonalitysklearn: ML-based forecasting
Commercial Software
ERP/Supply Chain Planning:
- SAP S/4HANA: Enterprise resource planning
- Oracle Food & Beverage: Industry-specific ERP
- Microsoft Dynamics 365: Supply chain management
- Blue Yonder (JDA): Demand and supply planning
Warehouse Management:
- Manhattan WMS: Warehouse management with FEFO
- SAP EWM: Extended warehouse management
- JDA WMS: Food-grade warehouse management
- HighJump: WMS for food distribution
Food Safety & Traceability:
- FoodLogiQ: Farm to fork traceability
- SafetyChain: Food safety and quality management
- Aptean Food & Beverage: ERP with traceability
- rfXcel: Serialization and traceability
- FarmSoft: Agricultural traceability
Transportation:
- Descartes: Route optimization and TMS
- Blue Yonder Transportation: TMS with temperature monitoring
- MercuryGate: Multi-temperature TMS
Common Challenges & Solutions
Challenge: Managing Perishability & Waste
Problem:
- Short shelf lives (days to weeks)
- High waste/spoilage rates (3-8% of sales)
- Slow-moving inventory near expiry
- Markdowns eroding margin
Solutions:
- FEFO implementation: Strict first-expired, first-out
- Demand forecasting: Reduce forecast error, especially for promotions
- Dynamic pricing: Markdown near-expiry products automatically
- Inventory redistribution: Move slow stock to high-velocity stores
- Shorter order cycles: Order more frequently in smaller quantities
- Direct store delivery: Bypass DC for ultra-fresh products
- Waste tracking: Measure and analyze root causes
Challenge: Cold Chain Integrity
Problem:
- Temperature excursions during storage and transport
- Product quality degradation
- Food safety risks
- No real-time visibility
Solutions:
- Temperature monitoring: IoT sensors with real-time alerts
- Insulated transport: Refrigerated trucks and trailers
- Cross-dock operations: Minimize dwell time in temp zones
- Preventive maintenance: Regular equipment servicing
- Driver training: Proper door management and temp checks
- Packaging innovation: Phase change materials for last mile
- Route optimization: Minimize transit time for fresh products
Challenge: Promotional Demand Volatility
Problem:
- 40-60% of food/beverage sales on promotion
- Demand spikes of 2-5x baseline
- Post-promotion dips
- Forecast accuracy suffers
- Inventory imbalances
Solutions:
- Promotional forecasting models: Separate baseline from lift
- Pre-build inventory: Manufacture ahead of promotion
- Flexible manufacturing: Rapid changeovers
- Retailer collaboration: Advance promotional calendars
- Mix optimization: Balance high and low promoted weeks
- Co-packer flexibility: Surge capacity through CMs
- Safety stock strategies: Higher stock for promotional SKUs
Challenge: Seasonal Supply Variability
Problem:
- Agricultural products seasonal (tomatoes, berries, etc.)
- Quality and pricing fluctuate
- Supply disruptions (weather, pests)
- Need year-round supply
Solutions:
- Geographic diversity: Source from multiple growing regions
- Forward contracts: Lock in supply and pricing
- Alternative sources: Import during off-season
- Product reformulation: Design for available ingredients
- Inventory building: Process and freeze peak season product
- Supplier relationships: Long-term partnerships with growers
- Vertical integration: Own or partner on farms
Challenge: Food Safety & Recalls
Problem:
- Foodborne illness outbreaks
- Product recalls (contamination, allergen, foreign material)
- Brand damage and costs
- Complex traceability requirements
Solutions:
- Robust HACCP: Well-designed food safety systems
- Supplier audits: Verify ingredient safety upstream
- Environmental monitoring: Detect pathogens in facility
- Traceability systems: One-up/one-down digitized
- Mock recalls: Quarterly practice drills (<4 hour target)
- Food safety culture: Training and accountability
- Rapid response: Crisis management procedures
- Insurance: Product recall and contamination coverage
Output Format
Food & Beverage Supply Chain Report
Executive Summary:
- Product categories overview (fresh, frozen, shelf-stable)
- Key supply chain metrics
- Food safety and quality status
- Major initiatives
Freshness & Waste Metrics:
| Category | Avg Shelf Life | Inventory Days | Waste % | Near-Expiry Value | FEFO Compliance |
|---|
| Fresh Dairy | 14 days | 4 days | 2.8% | $45,000 | 98% |
| Fresh Produce | 7 days | 2 days | 5.2% | $82,000 | 95% |
| Frozen Foods | 365 days | 45 days | 0.5% | $12,000 | 90% |
| Beverages | 180 days | 30 days | 1.2% | $28,000 | 92% |
| Total | - | - | 2.4% | $167,000 | 94% |
Supply Chain Performance:
| Metric | Current | Target | Status |
|---|
| OTIF Delivery | 94% | 96% | ⚠ Yellow |
| Inventory Turns | 18.5 | 20.0 | ⚠ Yellow |
| Waste Rate | 2.4% | <2.0% | ⚠ Yellow |
| Order Fill Rate | 97% | 98% | ⚠ Yellow |
| Cold Chain Compliance | 99.2% | 99.5% | ⚠ Yellow |
Food Safety Status:
| Metric | Status |
|---|
| HACCP Compliance | ✓ Compliant |
| Last Mock Recall Time | 2.5 hours (target <4 hrs) |
| Environmental Monitoring | All negative |
| Supplier Audits Current | 100% |
| GFSI Certification | SQF Level 2 |
| FDA Inspections | No observations |
Promotional Performance:
| Month | Promotions | Forecast Accuracy | Stockouts | Excess Inventory |
|---|
| Jan | 12 | 78% | 2 | $45,000 |
| Feb | 15 | 82% | 1 | $38,000 |
| Mar | 18 | 75% | 4 | $67,000 |
Action Items:
- Reduce dairy waste from 2.8% to <2.0% - implement dynamic pricing
- Improve promotional forecast accuracy - refine lift models
- Complete cold chain upgrade at DC3 - install backup refrigeration
- Launch traceability system digitization - complete by Q2
- Resolve produce stockouts - add backup supplier
Questions to Ask
If you need more context:
- What product categories? (fresh, frozen, shelf-stable, beverages)
- What are the shelf lives? (days, weeks, months)
- What temperature requirements? (ambient, refrigerated, frozen)
- What distribution channels? (retail, foodservice, DTC, export)
- What are current waste/spoilage rates?
- What food safety certifications are needed? (HACCP, GFSI, organic)
- How promotional is the business? (% sales on promotion)
- What are the main supply chain challenges?
Related Skills
- inventory-optimization: For safety stock and inventory policies
- demand-forecasting: For baseline and promotional forecasting
- network-design: For distribution network optimization
- route-optimization: For delivery route planning
- warehouse-slotting-optimization: For FEFO-based slotting
- promotional-planning: For CPG promotional calendars
- retail-allocation: For store allocation and replenishment
- co-packing-management: For contract manufacturing
- cold-chain: For temperature-controlled logistics
- quality-management: For food safety and HACCP
- seasonal-planning: For seasonal demand planning
- shelf-life-management: For expiry date management
