Food & Beverage Supply Chain

• 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?

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 optimization
• scipy: Statistical methods
• networkx: Supply network modeling

Data Analysis:

• pandas: Data manipulation and inventory management
• numpy: Numerical computations
• matplotlib, seaborn: Visualization

Forecasting:

• statsmodels: Time series forecasting
• prophet: Demand forecasting with seasonality
• sklearn: 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:

Supply Chain Performance:

Food Safety Status:

Promotional Performance:

Action Items:

1. Reduce dairy waste from 2.8% to <2.0% - implement dynamic pricing
2. Improve promotional forecast accuracy - refine lift models
3. Complete cold chain upgrade at DC3 - install backup refrigeration
4. Launch traceability system digitization - complete by Q2
5. Resolve produce stockouts - add backup supplier

Questions to Ask

If you need more context:

1. What product categories? (fresh, frozen, shelf-stable, beverages)
2. What are the shelf lives? (days, weeks, months)
3. What temperature requirements? (ambient, refrigerated, frozen)
4. What distribution channels? (retail, foodservice, DTC, export)
5. What are current waste/spoilage rates?
6. What food safety certifications are needed? (HACCP, GFSI, organic)
7. How promotional is the business? (% sales on promotion)
8. 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