Master Production Scheduling (MPS)

MPS Framework

MPS Definition

Master Production Schedule (MPS) is a time-phased plan that specifies how many of each end item will be produced and when. It disaggregates the aggregate production plan (from S&OP) into a specific build schedule for individual products.

Key Characteristics:

• Time-phased: Specific quantities by time period
• Item-level: End items or planning items
• Feasible: Respects capacity and material constraints
• Firm zone: Near-term frozen, far-term flexible
• Drives MRP: Input to Material Requirements Planning

MPS vs. Related Plans

MPS Planning Process

Step 1: Review Demand Requirements

Sources of Demand:

1. Forecast Demand

   • Statistical forecast from demand planning
   • Marketing/sales input
   • Seasonality and trends

2. Customer Orders

   • Booked orders
   • Firm commitments
   • Contracts

3. Safety Stock Requirements

   • Buffer inventory targets
   • Service level policies

4. Interplant Transfers

   • Distribution requirements
   • Network demand

5. Service Parts

   • Aftermarket demand
   • Warranty requirements

Demand Aggregation:

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']])

Step 2: Develop Preliminary MPS

MPS Logic:

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 Planning Table:

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)

Step 3: Rough-Cut Capacity Planning (RCCP)

Purpose: Validate MPS is feasible given capacity constraints

Process:

1. Calculate resource requirements from MPS
2. Compare to available capacity
3. Identify overloads
4. Adjust MPS or capacity

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)