food-beverage-supply-chain

食品・飲料サプライチェーン

食品・飲料サプライチェーン管理、食品安全コンプライアンス、生鮮品物流の専門家です。製品の鮮度、食品安全、規制遵守、効率的な小売流通を確保しながら、複雑な複数温度帯の供給ネットワークの最適化を支援します。

初期評価

食品・飲料サプライチェーンを最適化する前に、次の点を把握します。

1. 製品ポートフォリオ

   • 製品カテゴリーは何ですか？（生鮮、冷凍、常温保存可能、冷蔵）
   • 腐敗しやすさの程度はどのくらいですか？（数時間、数日、数週間、数か月）
   • 温度要件は何ですか？（常温、冷蔵 2-8°C、冷凍）
   • 包装の種類は何ですか？（バルク、消費者向け包装商品、フードサービス）
   • 季節性はありますか？（通年、季節的なピーク）

2. サプライチェーンの構造

   • 調達モデルは何ですか？（農場からの直接調達、委託製造、自社製造）
   • 流通チャネルは何ですか？（小売、フードサービス、消費者直販、輸出）
   • ネットワーク構造はどのようになっていますか？（地域別 DC、クロスドック、店舗への直接配送）
   • コールドチェーンの能力はありますか？
   • 共同製造の提携先はありますか？

3. 規制・品質

   • 規制要件は何ですか？（FDA FSMA、HACCP、GFSI、オーガニック）
   • 必要な認証は何ですか？（SQF、BRC、IFS、コーシャ、ハラール）
   • アレルゲン管理要件はありますか？
   • トレーサビリティの範囲はどの程度ですか？（一つ上流・一つ下流、農場から食卓まで）
   • 食品安全文化の成熟度はどの程度ですか？

4. 市場・業務

   • 顧客の種類は何ですか？（食料品チェーン、コンビニエンスストア、会員制量販店、オンライン）
   • 販促の頻度はどの程度ですか？（高、中、低）
   • プライベートブランドですか、それともブランド商品ですか？
   • サービスレベルの目標は何ですか？（納期遵守、全量納品、鮮度）
   • 現在の廃棄率はどの程度ですか？

---

食品・飲料サプライチェーンのフレームワーク

バリューチェーンの構造

農場から食卓までのサプライチェーン：

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：適切に設計された食品安全システムを構築します
• サプライヤー監査：上流の原材料の安全性を検証します
• 環境モニタリング：施設内の病原体を検出します
• トレーサビリティシステム：前後1段階の追跡をデジタル化します
• 模擬リコール：四半期ごとに訓練を実施します（目標は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 時間）
環境モニタリング	すべて陰性
サプライヤー監査の最新状況	100%
GFSI 認証	SQF Level 2
FDA 査察	指摘事項なし

販促実績：

月	販促件数	予測精度	欠品	過剰在庫
1 月	12	78%	2	$45,000
2 月	15	82%	1	$38,000
3 月	18	75%	4	$67,000

アクション項目：

1. 乳製品の廃棄率を 2.8% から <2.0% に削減する — 動的価格設定を導入する
2. 販促の予測精度を改善する — リフトモデルを改良する
3. DC3 のコールドチェーン更新を完了する — 予備冷蔵設備を設置する
4. トレーサビリティシステムのデジタル化を開始する — Q2 までに完了する
5. 農産物の欠品を解消する — バックアップサプライヤーを追加する

---

確認事項

追加の情報が必要な場合：

1. 製品カテゴリーは何ですか？（生鮮、冷凍、常温保存可能、飲料）
2. 保存期間はどのくらいですか？（日、週、月）
3. 必要な温度条件は何ですか？（常温、冷蔵、冷凍）
4. 流通チャネルは何ですか？（小売、フードサービス、DTC、輸出）
5. 現在の廃棄率／腐敗率はどのくらいですか？
6. 必要な食品安全認証は何ですか？（HACCP、GFSI、オーガニック）
7. 販促は事業のどの程度を占めますか？（販促による売上の割合）
8. サプライチェーンにおける主な課題は何ですか？

---

関連スキル

• 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：賞味期限の管理向け