master-production-scheduling

マスタープロダクションスケジューリング（MPS）

マスタープロダクションスケジューリング（MPS）と生産計画の専門家として、需要要件と能力制約のバランスを取りながら、効率性とサービスレベルを最大化する、実行可能で最適化された生産スケジュールを組織が作成できるよう支援します。

初期評価

MPSを策定する前に、以下を把握します。

1. 計画環境

   • 製造形態は何ですか？（見込み生産、受注生産、受注組立、受注設計生産）
   • 生産プロセスは何ですか？（個別生産、プロセス生産、反復生産、バッチ生産）
   • 計画期間はどのくらいですか？（週、月）
   • 計画バケットのサイズはどのくらいですか？（日次、週次）

2. 製品と需要

   • 最終製品／SKUの数はいくつですか？
   • 製品構造の複雑さはどの程度ですか？（単一階層、複数階層のBOM）
   • 需要パターンはどのようなものですか？（安定、季節性、断続的）
   • 顧客注文のリードタイムと製造リードタイムはどのくらいですか？

3. 能力と制約

   • 重要なリソースやボトルネックは何ですか？
   • 段取り替え／セットアップ時間は重要ですか？
   • 従業員の稼働状況とスキルはどうなっていますか？
   • 資材に制約はありますか？

4. 現状

   • 既存のMPSプロセスはありますか？
   • ERP／MRPシステムを導入していますか？
   • スケジュール遵守率はどのくらいですか？
   • 在庫レベル（原材料、仕掛品、完成品）はどのくらいですか？

---

MPSフレームワーク

MPSの定義

**マスタープロダクションスケジュール（MPS）**とは、各最終製品をいつ、いくつ生産するかを指定する、期間別の計画です。S&OPによる総合生産計画を、個々の製品の具体的な生産スケジュールへと細分化します。

主な特徴：

• 期間別：各期間の具体的な数量
• 品目レベル：最終製品または計画品目
• 実行可能：能力と資材の制約を遵守
• 確定ゾーン：短期は固定、長期は柔軟
• MRPを駆動：資材所要量計画への入力

MPSと関連計画の比較

計画の種類	期間	レベル	頻度	目的
S&OP	12～18か月	製品ファミリー	月次	需要と供給の調整
MPS	3～6か月	最終製品	週次	詳細な生産計画
MRP	3～6か月	構成品	日次／週次	資材所要量
製造現場スケジュール	数日～数週間	作業	日次	詳細な順序付け

---

MPS計画プロセス

ステップ1：需要要件の確認

需要の発生源：

1. 需要予測

   • 需要計画による統計的な予測
   • マーケティング／営業からの情報
   • 季節性とトレンド

2. 顧客注文

   • 受注済みの注文
   • 確定した約束
   • 契約

3. 安全在庫要件

   • バッファ在庫の目標
   • サービスレベル方針

4. 工場間移送

   • 配送所要量
   • ネットワーク需要

5. サービスパーツ

   • アフターマーケット需要
   • 保証要件

需要の集約：

import pandas as pd
import numpy as np
from datetime import datetime, timedelta

class MpsDemandPlanning:
    """Aggregate demand sources for MPS"""

    def __init__(self, planning_horizon_weeks=26):
        self.horizon = planning_horizon_weeks
        self.demand_sources = []

    def add_forecast_demand(self, item, periods, quantities):
        """Add forecasted demand"""
        self.demand_sources.append({
            'source': 'forecast',
            'item': item,
            'periods': periods,
            'quantities': quantities
        })

    def add_customer_orders(self, item, periods, quantities):
        """Add firm customer orders"""
        self.demand_sources.append({
            'source': 'customer_orders',
            'item': item,
            'periods': periods,
            'quantities': quantities
        })

    def add_safety_stock(self, item, target_level):
        """Add safety stock requirement"""
        self.demand_sources.append({
            'source': 'safety_stock',
            'item': item,
            'target': target_level
        })

    def calculate_total_demand(self):
        """
        Calculate total demand by item and period

        Uses consumption logic:
        - Customer orders consume forecast in near term
        - Forecast used beyond order horizon
        """

        # Create time buckets
        start_date = datetime.now()
        periods = pd.date_range(start_date, periods=self.horizon, freq='W')

        all_items = set()
        for source in self.demand_sources:
            all_items.add(source['item'])

        demand_df = pd.DataFrame({
            'period': periods
        })

        for item in all_items:
            # Get forecast
            forecast = self._get_source_demand(item, 'forecast', periods)

            # Get customer orders
            orders = self._get_source_demand(item, 'customer_orders', periods)

            # Apply consumption logic: greater of forecast or orders
            total_demand = np.maximum(forecast, orders)

            demand_df[f'{item}_forecast'] = forecast
            demand_df[f'{item}_orders'] = orders
            demand_df[f'{item}_total'] = total_demand

        return demand_df

    def _get_source_demand(self, item, source_type, periods):
        """Extract demand for specific item and source"""

        quantities = np.zeros(len(periods))

        for source in self.demand_sources:
            if source['item'] == item and source['source'] == source_type:
                if 'periods' in source:
                    for i, period in enumerate(source['periods']):
                        if i < len(quantities):
                            quantities[i] = source['quantities'][i]

        return quantities

# Example usage
mps_demand = MpsDemandPlanning(planning_horizon_weeks=12)

# Add forecast
periods = list(range(12))
forecast_qty = [100, 110, 105, 120, 115, 125, 130, 120, 115, 110, 120, 125]
mps_demand.add_forecast_demand('Product_A', periods, forecast_qty)

# Add firm customer orders (first 4 weeks)
orders = [150, 130, 0, 0] + [0] * 8
mps_demand.add_customer_orders('Product_A', periods, orders)

# Calculate total demand
total_demand = mps_demand.calculate_total_demand()
print("MPS Demand by Week:")
print(total_demand[['period', 'Product_A_forecast', 'Product_A_orders', 'Product_A_total']])

ステップ2：暫定MPSの策定

MPSのロジック：

Projected Available Balance (PAB) formula:
PAB[t] = PAB[t-1] + MPS[t] - max(Forecast[t], Orders[t])

When PAB[t] < Safety Stock:
  Schedule MPS receipt

MPS計画表：

期間	予測	注文	合計	予測可能在庫	MPS	ATP
1	100	150	150	50	200	50
2	110	130	130	120	0	0
3	105	0	105	15	100	100

import pandas as pd
import numpy as np

class MasterProductionSchedule:
    """Master Production Schedule generator"""

    def __init__(self, item, beginning_inventory,
                 safety_stock, lot_size, lead_time):
        """
        Parameters:
        - item: product identifier
        - beginning_inventory: starting inventory level
        - safety_stock: minimum inventory target
        - lot_size: production lot size (0 = lot-for-lot)
        - lead_time: production lead time (periods)
        """
        self.item = item
        self.beginning_inventory = beginning_inventory
        self.safety_stock = safety_stock
        self.lot_size = lot_size
        self.lead_time = lead_time

    def generate_mps(self, forecast, customer_orders, periods):
        """
        Generate MPS using standard MPS logic

        Returns DataFrame with MPS planning table
        """

        mps_table = []

        projected_available = self.beginning_inventory
        cumulative_mps = 0

        for t in range(periods):
            # Total demand (greater of forecast or orders)
            fcst = forecast[t] if t < len(forecast) else 0
            orders = customer_orders[t] if t < len(customer_orders) else 0
            total_demand = max(fcst, orders)

            # Check if MPS receipt needed
            if projected_available - total_demand < self.safety_stock:
                # Schedule MPS receipt
                shortage = self.safety_stock - (projected_available - total_demand)

                if self.lot_size > 0:
                    # Fixed lot size
                    lots_needed = int(np.ceil(shortage / self.lot_size))
                    mps_quantity = lots_needed * self.lot_size
                else:
                    # Lot-for-lot
                    mps_quantity = shortage

                # Account for lead time (schedule in future period)
                mps_receipt_period = t
            else:
                mps_quantity = 0
                mps_receipt_period = t

            # Update projected available
            projected_available = projected_available + mps_quantity - total_demand

            # Calculate Available-to-Promise (ATP)
            # ATP[t] = MPS[t] - sum(orders from t to next MPS)
            if mps_quantity > 0:
                atp = mps_quantity - orders
            else:
                atp = 0

            mps_table.append({
                'period': t + 1,
                'forecast': fcst,
                'customer_orders': orders,
                'total_demand': total_demand,
                'projected_available': max(0, projected_available),
                'mps_quantity': mps_quantity,
                'atp': max(0, atp)
            })

        return pd.DataFrame(mps_table)

    def calculate_rough_cut_capacity(self, mps_table, routing_time_per_unit):
        """
        Rough-Cut Capacity Planning (RCCP)

        Check if MPS is feasible given capacity

        Parameters:
        - mps_table: output from generate_mps
        - routing_time_per_unit: hours required per unit
        """

        mps_table['required_hours'] = (
            mps_table['mps_quantity'] * routing_time_per_unit
        )

        return mps_table[['period', 'mps_quantity', 'required_hours']]

# Example
mps = MasterProductionSchedule(
    item='Product_A',
    beginning_inventory=200,
    safety_stock=50,
    lot_size=100,  # Fixed lot size of 100
    lead_time=2
)

# Demand data
forecast = [100, 110, 105, 120, 115, 125, 130, 120, 115, 110, 120, 125]
customer_orders = [150, 130, 80, 90] + [0] * 8

# Generate MPS
mps_plan = mps.generate_mps(forecast, customer_orders, periods=12)
print("Master Production Schedule:")
print(mps_plan)

# Rough-cut capacity
rccp = mps.calculate_rough_cut_capacity(mps_plan, routing_time_per_unit=2.5)
print("\nRough-Cut Capacity Requirements:")
print(rccp)

ステップ3：概略能力計画（RCCP）

目的： 能力制約を踏まえてMPSが実行可能か検証します。

プロセス：

1. MPSからリソース所要量を算出します
2. 利用可能な能力と比較します
3. 過負荷を特定します
4. MPSまたは能力を調整します

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt

class RoughCutCapacityPlanning:
    """RCCP - Validate MPS feasibility"""

    def __init__(self, work_centers):
        """
        Parameters:
        - work_centers: dict {name: {'capacity_hours': x, 'efficiency': y}}
        """
        self.work_centers = work_centers

    def validate_mps(self, mps_schedule, routings):
        """
        Check if MPS is feasible

        Parameters:
        - mps_schedule: DataFrame with 'item', 'period', 'mps_quantity'
        - routings: dict {item: [(work_center, hours_per_unit)]}

        Returns capacity analysis
        """

        requirements = []

        for idx, row in mps_schedule.iterrows():
            item = row['item']
            period = row['period']
            quantity = row['mps_quantity']

            if quantity == 0:
                continue

            # Get routing for item
            routing = routings.get(item, [])

            for work_center, hours_per_unit in routing:
                # Calculate required hours
                efficiency = self.work_centers[work_center]['efficiency']
                required_hours = (quantity * hours_per_unit) / efficiency

                requirements.append({
                    'period': period,
                    'work_center': work_center,
                    'item': item,
                    'quantity': quantity,
                    'required_hours': required_hours
                })

        req_df = pd.DataFrame(requirements)

        # Aggregate by work center and period
        capacity_analysis = req_df.groupby(
            ['period', 'work_center']
        )['required_hours'].sum().reset_index()

        # Add available capacity
        capacity_analysis['available_capacity'] = capacity_analysis['work_center'].map(
            lambda wc: self.work_centers[wc]['capacity_hours']
        )

        # Calculate load %
        capacity_analysis['load_pct'] = (
            capacity_analysis['required_hours'] /
            capacity_analysis['available_capacity'] * 100
        )

        capacity_analysis['status'] = capacity_analysis['load_pct'].apply(
            lambda x: 'Overload' if x > 100 else
                     'High' if x > 90 else
                     'OK'
        )

        return capacity_analysis

    def identify_overloads(self, capacity_analysis):
        """Get periods and resources with overload"""
        return capacity_analysis[capacity_analysis['status'] == 'Overload']

    def plot_capacity_profile(self, capacity_analysis):
        """Visualize capacity load"""

        work_centers = capacity_analysis['work_center'].unique()

        fig, axes = plt.subplots(len(work_centers), 1,
                                figsize=(12, 4 * len(work_centers)),
                                squeeze=False)

        for i, wc in enumerate(work_centers):
            wc_data = capacity_analysis[
                capacity_analysis['work_center'] == wc
            ].sort_values('period')

            ax = axes[i, 0]

            # Plot capacity line
            ax.axhline(y=wc_data['available_capacity'].iloc[0],
                      color='green', linestyle='--',
                      linewidth=2, label='Available Capacity')

            # Plot requirements
            bars = ax.bar(wc_data['period'], wc_data['required_hours'],
                         alpha=0.7)

            # Color overloads red
            for j, (idx, row) in enumerate(wc_data.iterrows()):
                if row['status'] == 'Overload':
                    bars[j].set_color('red')

            ax.set_title(f'{wc} - Capacity Load')
            ax.set_xlabel('Period')
            ax.set_ylabel('Hours')
            ax.legend()
            ax.grid(True, alpha=0.3)

        plt.tight_layout()
        return fig

# Example
work_centers = {
    'Assembly': {'capacity_hours': 160, 'efficiency': 0.90},
    'Testing': {'capacity_hours': 120, 'efficiency': 0.95},
    'Packaging': {'capacity_hours': 140, 'efficiency': 0.92}
}

rccp = RoughCutCapacityPlanning(work_centers)

# MPS schedule (from previous example)
mps_schedule = pd.DataFrame({
    'item': ['Product_A'] * 12,
    'period': range(1, 13),
    'mps_quantity': [200, 0, 100, 200, 0, 100, 200, 0, 100, 200, 0, 100]
})

# Routings
routings = {
    'Product_A': [
        ('Assembly', 1.5),
        ('Testing', 0.8),
        ('Packaging', 0.5)
    ]
}

# Validate MPS
capacity_check = rccp.validate_mps(mps_schedule, routings)
print("Capacity Analysis:")
print(capacity_check)

# Identify overloads
overloads = rccp.identify_overloads(capacity_check)
if not overloads.empty:
    print("\nOverloaded Resources:")
    print(overloads)
else:
    print("\nNo capacity overloads - MPS is feasible")

# Plot
rccp.plot_capacity_profile(capacity_check)

ステップ4：約束可能在庫（ATP）

ATPの定義： 在庫と計画生産のうち、未引当の部分です。

ATPのルール：

• 最初の期間：手持在庫 + MPS - 顧客注文
• 以降の期間：MPS - 顧客注文（次のMPSまで）

ATPの計算：

class AvailableToPromise:
    """Calculate ATP for order promising"""

    def __init__(self, mps_table, on_hand_inventory):
        self.mps = mps_table
        self.on_hand = on_hand_inventory

    def calculate_atp(self):
        """
        Calculate Available-to-Promise

        ATP provides visibility for order promising
        """

        atp_table = self.mps.copy()
        atp_table['atp'] = 0

        for i in range(len(atp_table)):
            if i == 0:
                # First period: On-hand + MPS - Orders
                atp_table.loc[i, 'atp'] = (
                    self.on_hand +
                    atp_table.loc[i, 'mps_quantity'] -
                    atp_table.loc[i, 'customer_orders']
                )
            else:
                # Check if MPS in this period
                if atp_table.loc[i, 'mps_quantity'] > 0:
                    # ATP = MPS - orders from this period until next MPS
                    orders_to_next_mps = 0

                    # Look ahead for orders until next MPS
                    for j in range(i, len(atp_table)):
                        orders_to_next_mps += atp_table.loc[j, 'customer_orders']

                        # Stop at next MPS
                        if j > i and atp_table.loc[j, 'mps_quantity'] > 0:
                            break

                    atp_table.loc[i, 'atp'] = (
                        atp_table.loc[i, 'mps_quantity'] - orders_to_next_mps
                    )

        return atp_table

    def check_order_promise(self, order_quantity, requested_period):
        """
        Check if order can be promised

        Returns: (can_promise, available_period)
        """

        atp_table = self.calculate_atp()

        # Check requested period first
        if requested_period <= len(atp_table):
            cumulative_atp = atp_table.loc[:requested_period-1, 'atp'].sum()

            if cumulative_atp >= order_quantity:
                return True, requested_period

        # Find earliest period with enough ATP
        cumulative = 0
        for i in range(len(atp_table)):
            cumulative += atp_table.loc[i, 'atp']

            if cumulative >= order_quantity:
                return True, i + 1

        # Cannot promise
        return False, None

# Example
mps_table = pd.DataFrame({
    'period': [1, 2, 3, 4, 5, 6],
    'forecast': [100, 110, 105, 120, 115, 125],
    'customer_orders': [80, 70, 50, 30, 0, 0],
    'mps_quantity': [200, 0, 150, 0, 0, 200],
    'projected_available': [120, 50, 195, 75, 75, 150]
})

atp = AvailableToPromise(mps_table, on_hand_inventory=100)

# Calculate ATP
atp_results = atp.calculate_atp()
print("Available-to-Promise:")
print(atp_results[['period', 'mps_quantity', 'customer_orders', 'atp']])

# Check if can promise order
can_promise, period = atp.check_order_promise(order_quantity=150, requested_period=2)
if can_promise:
    print(f"\n✓ Can promise 150 units by period {period}")
else:
    print("\n✗ Cannot promise order")

ステップ5：MPSの確定とリリース

タイムフェンス：

|<-- Frozen Zone -->|<-- Slushy Zone -->|<-- Liquid Zone -->|
       2-4 weeks          4-8 weeks           8+ weeks

Frozen: No changes without executive approval
Slushy: Changes with planner approval
Liquid: Flexible, planning purposes only

確定方針：

from datetime import datetime, timedelta

class MpsTimeFences:
    """Manage MPS time fences and change control"""

    def __init__(self, frozen_weeks, slushy_weeks):
        self.frozen_weeks = frozen_weeks
        self.slushy_weeks = slushy_weeks

    def get_zone(self, period_date):
        """Determine time fence zone for a period"""

        now = datetime.now()
        frozen_end = now + timedelta(weeks=self.frozen_weeks)
        slushy_end = now + timedelta(weeks=self.frozen_weeks + self.slushy_weeks)

        if period_date <= frozen_end:
            return 'Frozen'
        elif period_date <= slushy_end:
            return 'Slushy'
        else:
            return 'Liquid'

    def approve_change(self, period_date, change_type, authority_level):
        """
        Check if change is allowed

        Parameters:
        - change_type: 'quantity', 'timing', 'cancellation'
        - authority_level: 'planner', 'manager', 'executive'

        Returns: (approved, reason)
        """

        zone = self.get_zone(period_date)

        if zone == 'Frozen':
            if authority_level == 'executive':
                return True, 'Executive approval in frozen zone'
            else:
                return False, 'Changes require executive approval in frozen zone'

        elif zone == 'Slushy':
            if authority_level in ['manager', 'executive']:
                return True, 'Manager approval in slushy zone'
            else:
                return False, 'Changes require manager approval in slushy zone'

        else:  # Liquid
            return True, 'Free to change in liquid zone'

    def mps_release(self, mps_table):
        """
        Release MPS with time fence zones marked

        Returns MPS with zones and change authority
        """

        mps_released = mps_table.copy()

        # Assume period is datetime
        mps_released['zone'] = mps_released['period'].apply(self.get_zone)

        mps_released['change_authority'] = mps_released['zone'].map({
            'Frozen': 'Executive',
            'Slushy': 'Manager',
            'Liquid': 'Planner'
        })

        return mps_released

# Example
fences = MpsTimeFences(frozen_weeks=4, slushy_weeks=4)

# Check change approval
period = datetime.now() + timedelta(weeks=2)
approved, reason = fences.approve_change(period, 'quantity', authority_level='planner')
print(f"Change approval: {approved}")
print(f"Reason: {reason}")

---

製造環境別のMPS戦略

見込生産（MTS）

特徴：

• 需要予測に基づいて生産します
• 完成品在庫を維持します
• 顧客への納期が短くなります

MPS のアプローチ：

• 需要予測と安全在庫に基づいてスケジュールを作成します
• 平準化生産を優先します
• 補充のトリガー（最小／最大、EOQ）を設定します

def mts_mps_logic(forecast, current_inventory, safety_stock,
                  target_inventory, lot_size):
    """
    Make-to-Stock MPS logic

    Schedule production to maintain target inventory
    """

    mps_schedule = []

    inventory = current_inventory

    for period, demand in enumerate(forecast):
        # Check if replenishment needed
        projected_inventory = inventory - demand

        if projected_inventory < safety_stock:
            # Order up to target
            order_quantity = target_inventory - projected_inventory

            # Round to lot size
            if lot_size > 0:
                lots = int(np.ceil(order_quantity / lot_size))
                order_quantity = lots * lot_size

            inventory = projected_inventory + order_quantity
        else:
            order_quantity = 0
            inventory = projected_inventory

        mps_schedule.append({
            'period': period + 1,
            'demand': demand,
            'mps_quantity': order_quantity,
            'ending_inventory': inventory
        })

    return pd.DataFrame(mps_schedule)

# Example
forecast = [100, 110, 105, 120, 115, 125, 130, 120]
mts_plan = mts_mps_logic(
    forecast=forecast,
    current_inventory=200,
    safety_stock=50,
    target_inventory=250,
    lot_size=100
)
print("Make-to-Stock MPS:")
print(mts_plan)

受注生産（MTO）

特徴：

• 特定の顧客注文に対してのみ生産します
• 完成品在庫はありません
• 顧客へのリードタイムが長くなります

MPS のアプローチ：

• 顧客注文に基づいてスケジュールを作成します
• 納期から逆算してスケジュールを作成します
• ATP によって注文の納期回答を行います

def mto_mps_logic(customer_orders, production_lead_time):
    """
    Make-to-Order MPS logic

    Schedule production to meet order due dates
    """

    mps_schedule = []

    for order in customer_orders:
        order_id = order['order_id']
        quantity = order['quantity']
        due_date = order['due_date']

        # Backward schedule: start = due date - lead time
        start_period = due_date - production_lead_time

        mps_schedule.append({
            'order_id': order_id,
            'start_period': max(1, start_period),
            'due_period': due_date,
            'quantity': quantity,
            'order_type': 'customer_order'
        })

    return pd.DataFrame(mps_schedule)

# Example
orders = [
    {'order_id': 'ORD-001', 'quantity': 50, 'due_date': 5},
    {'order_id': 'ORD-002', 'quantity': 75, 'due_date': 7},
    {'order_id': 'ORD-003', 'quantity': 60, 'due_date': 10}
]

mto_plan = mto_mps_logic(orders, production_lead_time=3)
print("Make-to-Order MPS:")
print(mto_plan)

受注組立（ATO）

特徴：

• 部品を在庫し、受注後に最終組立を行います
• 共通部品を使って多様な製品を提供します
• リードタイムは中程度です

MPS のアプローチ：

• 2 段階の MPS：
  1. 部品／モジュールレベル（需要予測主導）
  2. 最終組立（注文主導）
• 計画用部品表
• 最終組立スケジュール（FAS）

class AssembleToOrderMPS:
    """ATO with planning bills and final assembly schedule"""

    def __init__(self, planning_bill):
        """
        planning_bill: dict {option: {component: percentage}}

        Example:
        {
            'Base_Model': {'Frame': 1.0, 'Engine_A': 0.6, 'Engine_B': 0.4},
            ...
        }
        """
        self.planning_bill = planning_bill

    def generate_component_mps(self, demand_forecast, planning_item):
        """
        Generate MPS for components based on option percentages

        Parameters:
        - demand_forecast: forecast for planning item
        - planning_item: aggregate product family
        """

        component_forecast = {}

        for component, percentage in self.planning_bill[planning_item].items():
            component_forecast[component] = [
                qty * percentage for qty in demand_forecast
            ]

        return component_forecast

    def create_final_assembly_schedule(self, customer_orders,
                                      assembly_lead_time):
        """
        Create FAS from customer orders

        FAS specifies exact configuration to build
        """

        fas = []

        for order in customer_orders:
            config = order['configuration']
            quantity = order['quantity']
            due_date = order['due_date']

            start_date = max(1, due_date - assembly_lead_time)

            fas.append({
                'order_id': order['order_id'],
                'configuration': config,
                'quantity': quantity,
                'start_period': start_date,
                'due_period': due_date
            })

        return pd.DataFrame(fas)

# Example
planning_bill = {
    'Laptop': {
        'Base_Unit': 1.0,
        'CPU_i5': 0.6,
        'CPU_i7': 0.4,
        'RAM_8GB': 0.5,
        'RAM_16GB': 0.5,
        'SSD_256GB': 0.7,
        'SSD_512GB': 0.3
    }
}

ato = AssembleToOrderMPS(planning_bill)

# Forecast for planning item
laptop_forecast = [100, 110, 105, 120, 115, 125]

# Generate component MPS
component_mps = ato.generate_component_mps(laptop_forecast, 'Laptop')
print("Component MPS (Forecast-Driven):")
for component, forecast in component_mps.items():
    print(f"  {component}: {forecast[:4]}")

# Customer orders with specific configurations
orders = [
    {
        'order_id': 'ORD-001',
        'configuration': 'i7/16GB/512GB',
        'quantity': 50,
        'due_date': 3
    },
    {
        'order_id': 'ORD-002',
        'configuration': 'i5/8GB/256GB',
        'quantity': 75,
        'due_date': 5
    }
]

# Final assembly schedule
fas = ato.create_final_assembly_schedule(orders, assembly_lead_time=1)
print("\nFinal Assembly Schedule (Order-Driven):")
print(fas)

---

MPS の最適化手法

平準化生産

目的： 変動を最小限に抑えるため、生産を平準化します

アプローチ：

• 平均需要を算出します
• 一定の生産率で生産します
• 在庫をバッファーとして使用します

def level_loading_mps(total_demand, periods, lot_size=0):
    """
    Level loading - produce at constant rate

    Parameters:
    - total_demand: total demand over horizon
    - periods: number of periods
    - lot_size: production lot size (0 = continuous)
    """

    # Calculate average per period
    avg_demand = total_demand / periods

    # Round to lot size if applicable
    if lot_size > 0:
        level_quantity = int(np.ceil(avg_demand / lot_size)) * lot_size
    else:
        level_quantity = avg_demand

    # Create level schedule
    mps = [level_quantity] * periods

    return mps

# Example
total_demand = 1200
periods = 10

level_mps = level_loading_mps(total_demand, periods, lot_size=50)
print(f"Level Loading MPS: {level_mps}")
print(f"Average production per period: {sum(level_mps)/len(level_mps):.0f}")

追従戦略

目的： 生産を需要に正確に合わせます

アプローチ：

• 需要に応じて生産率を変動させます
• 在庫を最小限に抑えます
• コストが増加します（採用／解雇、段取り）

def chase_strategy_mps(demand_forecast, lot_size=0):
    """
    Chase strategy - match production to demand

    Parameters:
    - demand_forecast: demand by period
    - lot_size: minimum production quantity
    """

    mps = []

    for demand in demand_forecast:
        if lot_size > 0:
            # Round up to lot size
            quantity = int(np.ceil(demand / lot_size)) * lot_size
        else:
            quantity = demand

        mps.append(quantity)

    return mps

# Example
demand = [100, 120, 90, 150, 110, 130, 95, 140]
chase_mps = chase_strategy_mps(demand, lot_size=50)
print(f"Chase Strategy MPS: {chase_mps}")

混合戦略（ハイブリッド）

目的： 在庫コストと生産変動のバランスを取ります

アプローチ：

• 基礎需要には平準化生産で対応します
• 需要のピークには、残業、外注、または在庫で対応します

from pulp import *

def mixed_strategy_mps(demand, costs, capacity_normal, capacity_overtime):
    """
    Mixed strategy optimization

    Minimize total cost of production + inventory + backorders

    Parameters:
    - demand: list of demand by period
    - costs: dict {'regular': x, 'overtime': y, 'inventory': z, 'backorder': w}
    - capacity_normal: regular capacity per period
    - capacity_overtime: max overtime capacity per period
    """

    periods = len(demand)

    # Create problem
    prob = LpProblem("Mixed_Strategy_MPS", LpMinimize)

    # Variables
    P = LpVariable.dicts("Regular", range(periods), lowBound=0)
    O = LpVariable.dicts("Overtime", range(periods), lowBound=0)
    I = LpVariable.dicts("Inventory", range(periods), lowBound=0)
    B = LpVariable.dicts("Backorder", range(periods), lowBound=0)

    # Objective
    prob += lpSum([
        costs['regular'] * P[t] +
        costs['overtime'] * O[t] +
        costs['inventory'] * I[t] +
        costs['backorder'] * B[t]
        for t in range(periods)
    ])

    # Constraints
    for t in range(periods):
        # Capacity
        prob += P[t] <= capacity_normal
        prob += O[t] <= capacity_overtime

        # Inventory balance
        if t == 0:
            prob += I[t] == P[t] + O[t] - demand[t] + B[t]
        else:
            prob += I[t] == I[t-1] + P[t] + O[t] - demand[t] + B[t] - B[t-1]

    # Solve
    prob.solve(PULP_CBC_CMD(msg=0))

    # Extract solution
    solution = {
        'regular': [P[t].varValue for t in range(periods)],
        'overtime': [O[t].varValue for t in range(periods)],
        'inventory': [I[t].varValue for t in range(periods)],
        'backorder': [B[t].varValue for t in range(periods)],
        'total_cost': value(prob.objective)
    }

    return solution

# Example
demand = [100, 120, 150, 180, 160, 140, 120, 110]

costs = {
    'regular': 50,
    'overtime': 75,
    'inventory': 5,
    'backorder': 100
}

mixed_mps = mixed_strategy_mps(
    demand=demand,
    costs=costs,
    capacity_normal=120,
    capacity_overtime=40
)

print("Mixed Strategy MPS:")
print(f"Total Cost: ${mixed_mps['total_cost']:,.0f}")
print(f"Regular Production: {[int(x) for x in mixed_mps['regular']]}")
print(f"Overtime Production: {[int(x) for x in mixed_mps['overtime']]}")
print(f"Ending Inventory: {[int(x) for x in mixed_mps['inventory']]}")

---

MPS のパフォーマンス指標

スケジュール遵守率

MPS 達成率：

Actual Production / Planned Production × 100%

目標：>95%

在庫指標

在庫回転率：

COGS / Average Inventory

在庫日数：

(Average Inventory / Daily Demand) × Days

顧客サービス

納期遵守率：

Orders Delivered On-Time / Total Orders × 100%

目標：>95%

注文充足率：

Orders Filled Complete / Total Orders × 100%

計画指標

計画の不安定性：

• スケジュールの不安定さを測定します
• 確定済み MPS の変更回数

ATP の精度：

• 約束した納期と実際の納入実績の比較

class MPSMetrics:
    """Track MPS performance metrics"""

    def __init__(self):
        self.periods = []

    def add_period(self, planned, actual, inventory, orders_on_time, total_orders):
        """Record period performance"""

        self.periods.append({
            'planned_production': planned,
            'actual_production': actual,
            'ending_inventory': inventory,
            'orders_on_time': orders_on_time,
            'total_orders': total_orders
        })

    def calculate_metrics(self):
        """Calculate summary metrics"""

        df = pd.DataFrame(self.periods)

        metrics = {
            'mps_attainment': (df['actual_production'].sum() /
                              df['planned_production'].sum() * 100),
            'avg_inventory': df['ending_inventory'].mean(),
            'on_time_delivery': (df['orders_on_time'].sum() /
                                df['total_orders'].sum() * 100)
        }

        return metrics

# Example
metrics = MPSMetrics()

# Add several periods
for i in range(6):
    metrics.add_period(
        planned=200,
        actual=np.random.randint(185, 205),
        inventory=np.random.randint(80, 120),
        orders_on_time=np.random.randint(18, 20),
        total_orders=20
    )

summary = metrics.calculate_metrics()
print("MPS Performance Metrics:")
for metric, value in summary.items():
    print(f"  {metric}: {value:.1f}")

---

ツールとライブラリ

Python ライブラリ

最適化：

• pulp：MPS 最適化のための線形計画法
• pyomo：高度な最適化
• scipy.optimize：最適化アルゴリズム

データ管理：

• pandas：時系列データとデータ操作
• numpy：数値計算

可視化：

• matplotlib、seaborn：ガントチャート、能力プロファイル
• plotly：インタラクティブなスケジュール

商用 MPS ソフトウェア

MPS を備えた ERP システム：

• SAP：MRP／MPS モジュール
• Oracle：高度なサプライチェーン計画
• Microsoft Dynamics：マスタープランニング
• Infor：生産計画

専門的な計画ソフトウェア：

• Kinaxis RapidResponse：リアルタイム計画
• Blue Yonder：需要主導型 MPS
• o9 Solutions：AI を活用した計画
• Anaplan：クラウドベースの計画

製造現場との連携：

• Plex：製造業向け MPS
• IQMS：製造業向け MRP／MPS
• Epicor：生産管理

---

よくある課題と解決策

課題：確定期間が短すぎる

問題：

• スケジュールが常に変更されます
• 生産が混乱します
• 達成率が低くなります

解決策：

• 確定期間を延長します（通常 4～6 週間）
• リードタイムを短縮し、必要な確定期間を短くします
• 需要予測の精度を向上させます
• 変更管理ポリシーを徹底します
• 安全在庫でバッファーを確保します

課題：能力の過負荷

問題：

• RCCP で能力不足が示されます
• MPS を達成できません

解決策：

• シフトや残業を追加します
• 外注／下請けを利用します
• 需要を削減します（割り当て）
• リードタイムを延長します
• 設備投資を行います（長期的な対策）

課題：スケジュール遵守率が低い

問題：

• 実際の生産が MPS と一致しません
• その場しのぎの対応や特急対応が発生します

解決策：

• 製造現場の実行力を向上させます
• 資材の可用性を高めます（MRP）
• 変動を抑えます（品質、稼働時間）
• 現実的な MPS を作成します（過剰な計画を避けます）
• 未達の根本原因を分析します

課題：MPS のロットサイズのばらつき

問題：

• ロットサイズが大きいと、需要のピークと谷が生じます
• 能力が大きく変動します
• 在庫が積み上がります

解決策：

• 段取り時間を短縮します（SMED）
• ロットサイズを小さくします
• 平準化生産方式を採用します
• 混合モデル生産スケジューリングを行います
• リーン生産方式の原則を適用します

課題：計画の不安定化（変更が多すぎる）

問題：

• MPS が頻繁に変更されます
• プランナーの信頼性に問題が生じます
• 現場が混乱します

解決策：

• タイムフェンスを厳格化します
• 予測精度を向上させます
• 需要をフィルタリング／平滑化します
• 顧客の期待を管理します
• S&OP プロセスを改善します

---

出力形式

MPS レポートの構成

MPS 計画表：

週	予測	顧客受注	総需要	予測可能在庫	MPS	ATP	ゾーン
1	100	150	150	50	200	50	確定
2	110	130	130	120	0	0	確定
3	105	80	105	15	100	20	確定
4	120	60	120	-5	200	140	準確定
5	115	0	115	80	0	0	準確定
6	125	0	125	-45	200	200	変更可能

能力負荷の概要：

ワークセンター	週 1	週 2	週 3	週 4	使用可能能力	状態
組立	145	120	130	160	160	OK
試験	95	85	90	110	120	OK
梱包	70	60	65	80	80	OK

先週からの MPS 変更：

品目	週	旧数量	新数量	変更量	ゾーン	理由
Product_A	4	150	200	+50	準確定	新規受注
Product_B	6	100	0	-100	変更可能	予測を下方修正

ATP の概要：

品目	週 1	週 2	週 3	週 4	ATP 合計
Product_A	50	0	20	140	210
Product_B	30	0	0	100	130

例外レポート：

• 納期超過受注：納期を過ぎた受注の一覧
• 能力超過：負荷が 100% を超えるリソース
• 資材不足：使用可能な部品がない状態
• スケジュール変更：確定ゾーン内の変更

---

確認事項

さらに詳しい情報が必要な場合：

1. どのような製造環境ですか？（MTS、MTO、ATO、ETO）
2. 生産プロセスはどのようなものですか？（個別生産、プロセス生産、バッチ生産）
3. 計画期間とバケットの単位はどのようになっていますか？（週、月）
4. スケジュール対象の最終品目はいくつありますか？
5. 現在の MPS プロセスとツールはどのようなものですか？
6. 重要な能力制約はありますか？
7. タイムフェンスの方針はどのようなものですか？
8. スケジュール遵守率はどのくらいですか？

---

関連スキル

• sales-operations-planning：総合生産計画への入力
• capacity-planning：能力分析と RCCP
• demand-forecasting：MPS への需要入力
• production-scheduling：現場の詳細なスケジューリング
• inventory-optimization：安全在庫とロットサイズの最適化
• lot-sizing-problems：MPS の数量決定
• supply-chain-analytics：MPS のパフォーマンス追跡
• lean-manufacturing：段取り時間の短縮と生産平準化