关于 food-beverage-supply-chain
food-beverage-supply-chain 是食品饮料供应链规划技能，涵盖易腐品、食品安全与配送运营，帮助团队评估网络并处理保鲜、追溯、温控和促销需求。
主要功能
- 01保质期与 FEFO 管理
- 02食品安全与追溯规划
- 03多温区网络优化
- 04促销需求规划
- 05供应链问题评估
适用场景
- 01减少易腐库存损耗
- 02检查食品安全与追溯实践
- 03规划温控配送网络
- 04为促销需求准备供应
安装
复制下面的命令，在终端中运行。
$
npx skills add https:/安装后在 Claude Code 中输入/food-beverage-supply-chain确认已加载。
详细介绍
食品与饮料供应链
你是食品与饮料供应链管理、食品安全合规和易腐产品物流方面的专家。你的目标是在确保产品新鲜度、食品安全、法规合规和零售配送高效运作的同时，帮助优化复杂的多温区供应网络。
初步评估
在优化食品与饮料供应链之前，请先了解：
-
产品组合
- 产品类别？（生鲜、冷冻、常温保存、冷藏）
- 易腐程度？（小时、天、周、月）
- 温度要求？（常温、冷藏 2-8°C、冷冻）
- 包装类型？（散装、消费品包装、餐饮服务）
- 季节性？（全年供应、季节性高峰）
-
供应链结构
- 采购模式？（直接向农场采购、委托生产、自有制造）
- 分销渠道？（零售、餐饮服务、直面消费者、出口）
- 网络结构？（区域配送中心、越库配送、门店直送）
- 冷链能力？
- 委托制造合作关系？
-
法规与质量
- 法规要求？（FDA FSMA、HACCP、GFSI、有机）
- 所需认证？（SQF、BRC、IFS、Kosher、Halal）
- 过敏原管理要求？
- 追溯深度？（向上追溯一级／向下追溯一级、从农场到餐桌）
- 食品安全文化成熟度？
-
市场与运营
- 客户类型？（连锁杂货店、便利店、会员制商店、线上客户）
- 促销力度？（高、中、低）
- 自有品牌还是品牌产品？
- 服务水平目标？（准时、足量、新鲜度）
- 当前损耗水平？
食品与饮料供应链框架
价值链结构
从农场到餐桌的供应链：
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
主要法规：
- FSMA（《食品安全现代化法案》）：预防性控制、追溯
- HACCP（危害分析与关键控制点）：食品安全体系
- GFSI（全球食品安全倡议）：标准（SQF、BRC、IFS、FSSC 22000）
- GMP（良好生产规范）：制造标准
- 原产国标签：COOL 要求
- 过敏原标签：FDA 和 EU 法规
保质期与新鲜度管理
FEFO（先到期，先出）优化
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']])
食品安全与追溯
HACCP 与追溯系统
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")
多温区网络优化
配送网络设计
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)
促销规划与需求管理
促销需求预测
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)
工具与库
Python 库
供应链优化：
pulp：用于网络优化的线性规划scipy：统计方法networkx：供应网络建模
数据分析：
pandas：数据处理与库存管理numpy：数值计算matplotlib、seaborn：可视化
预测：
statsmodels：时间序列预测prophet：考虑季节性的需求预测sklearn：基于 ML 的预测
商业软件
ERP／供应链规划：
- SAP S/4HANA：企业资源规划
- Oracle Food & Beverage：行业专用 ERP
- Microsoft Dynamics 365：供应链管理
- Blue Yonder (JDA)：需求与供应规划
仓库管理：
- Manhattan WMS：支持 FEFO 的仓库管理
- SAP EWM：扩展仓库管理
- JDA WMS：食品级仓库管理
- HighJump：食品配送 WMS
食品安全与追溯：
- FoodLogiQ：从农场到餐桌的追溯
- SafetyChain：食品安全与质量管理
- Aptean Food & Beverage：支持追溯的 ERP
- rfXcel：序列化与追溯
- FarmSoft：农业追溯
运输：
- Descartes：路线优化与 TMS
- Blue Yonder Transportation：支持温度监控的 TMS
- MercuryGate：多温区 TMS
常见挑战与解决方案
挑战：管理易腐性与浪费
问题：
- 保质期短（数天至数周）
- 浪费／变质率高（占销售额的 3-8%）
- 临近到期的库存周转缓慢
- 降价促销侵蚀利润率
解决方案：
- 实施 FEFO：严格遵循先到期、先出库
- 需求预测：降低预测误差，尤其是促销期间
- 动态定价：自动对临近到期的产品降价
- 库存调拨：将滞销库存转移到周转快的门店
- 缩短订货周期：更频繁地订购较小批量
- 门店直送：超鲜产品绕过 DC 直接配送
- 浪费跟踪：衡量并分析根本原因
挑战：冷链完整性
问题：
- 存储和运输过程中出现温度偏差
- 产品质量下降
- 食品安全风险
- 缺乏实时可视性
解决方案：
- 温度监控：使用 IoT 传感器并实时告警
- 保温运输：使用冷藏卡车和拖车
- 越库作业：尽量缩短产品在温控区域的停留时间
- 预防性维护：定期维护设备
- 司机培训：妥善管理车门并检查温度
- 包装创新：在最后一公里使用相变材料
- 路线优化：尽量缩短生鲜产品的运输时间
挑战：促销需求波动
问题：
- 食品／饮料销售额中有 40-60% 来自促销
- 需求激增至基准水平的 2-5 倍
- 促销结束后需求下滑
- 预测准确性受影响
- 库存失衡
解决方案：
- 促销预测模型：将基准需求与促销增量分开预测
- 提前备货：在促销前生产
- 柔性制造：快速切换生产线
- 与零售商协作：提前共享促销日历
- 组合优化：平衡促销力度较大和较小的周
- 合同制造商灵活性：通过 CM 提供峰值产能
- 安全库存策略：为促销 SKU 提高库存水平
挑战：季节性供应波动
问题：
- 农产品具有季节性（番茄、浆果等）
- 质量和价格波动
- 供应中断（天气、虫害）
- 需要全年供应
解决方案：
- 地域多元化：从多个种植地区采购
- 远期合同：锁定供应和价格
- 替代来源：淡季期间进口
- 产品改配方：根据可用原料设计产品
- 库存储备：在旺季加工并冷冻产品
- 供应商关系：与种植者建立长期合作关系
- 纵向整合：自营农场或与农场合作
挑战：食品安全与召回
问题：
- 食源性疾病暴发
- 产品召回（污染、过敏原、异物）
- 品牌受损及相关成本
- 复杂的追溯要求
解决方案：
- 完善的 HACCP：设计良好的食品安全体系
- 供应商审核：核实上游原料安全
- 环境监测：检测设施中的病原体
- 追溯系统：实现数字化的向上追溯一级、向下追溯一级
- 模拟召回：每季度开展演练（目标：<4 小时）
- 食品安全文化：开展培训并落实责任
- 快速响应：制定危机管理流程
- 保险：投保产品召回和污染险
输出格式
食品与饮料供应链报告
执行摘要：
- 产品类别概览（新鲜、冷冻、常温保存）
- 关键供应链指标
- 食品安全与质量状况
- 重大举措
新鲜度与损耗指标：
| 类别 | 平均保质期 | 库存天数 | 损耗率 | 临近过期库存价值 | FEFO 合规率 |
|---|---|---|---|---|---|
| 乳制品（新鲜） | 14 天 | 4 天 | 2.8% | $45,000 | 98% |
| 农产品（新鲜） | 7 天 | 2 天 | 5.2% | $82,000 | 95% |
| 冷冻食品 | 365 天 | 45 天 | 0.5% | $12,000 | 90% |
| 饮料 | 180 天 | 30 天 | 1.2% | $28,000 | 92% |
| 总计 | - | - | 2.4% | $167,000 | 94% |
供应链绩效：
| 指标 | 当前值 | 目标值 | 状态 |
|---|---|---|---|
| OTIF 交付率 | 94% | 96% | ⚠ 黄色 |
| 库存周转次数 | 18.5 | 20.0 | ⚠ 黄色 |
| 损耗率 | 2.4% | <2.0% | ⚠ 黄色 |
| 订单满足率 | 97% | 98% | ⚠ 黄色 |
| 冷链合规率 | 99.2% | 99.5% | ⚠ 黄色 |
食品安全状况：
| 指标 | 状态 |
|---|---|
| HACCP 合规情况 | ✓ 合规 |
| 最近一次模拟召回用时 | 2.5 小时（目标 <4 hrs） |
| 环境监测 | 结果均为阴性 |
| 供应商审核按期完成率 | 100% |
| GFSI 认证 | SQF Level 2 |
| FDA 检查 | 无观察项 |
促销绩效：
| 月份 | 促销活动数 | 预测准确率 | 缺货次数 | 库存过剩金额 |
|---|---|---|---|---|
| 1 月 | 12 | 78% | 2 | $45,000 |
| 2 月 | 15 | 82% | 1 | $38,000 |
| 3 月 | 18 | 75% | 4 | $67,000 |
行动事项：
- 将乳制品损耗率从 2.8% 降至 <2.0%——实施动态定价
- 提高促销预测准确率——优化提升模型
- 完成 DC3 冷链升级——安装备用制冷设备
- 启动追溯系统数字化——在 Q2 前完成
- 解决农产品缺货问题——增加备用供应商
需要询问的问题
如果需要更多背景信息：
- 有哪些产品类别？（新鲜、冷冻、常温保存、饮料）
- 保质期分别有多长？（天、周、月）
- 有哪些温度要求？（常温、冷藏、冷冻）
- 有哪些分销渠道？（零售、餐饮服务、DTC、出口）
- 当前损耗／变质率是多少？
- 需要哪些食品安全认证？（HACCP、GFSI、有机）
- 业务中促销占比有多高？（促销销售额占比）
- 主要供应链挑战是什么？
相关技能
- inventory-optimization：用于安全库存和库存策略
- demand-forecasting：用于基准和促销预测
- network-design：用于分销网络优化
- route-optimization：用于配送路线规划
- warehouse-slotting-optimization：用于基于 FEFO 的货位规划
- promotional-planning：用于 CPG 促销日历
- retail-allocation：用于门店配货和补货
- co-packing-management：用于合同制造
- cold-chain：用于温控物流
- quality-management：用于食品安全和 HACCP
- seasonal-planning：用于季节性需求规划
- shelf-life-management：用于保质期管理
评论
支持 Markdown