When the user wants to create detailed production schedules, develop MPS, manage production planning, or translate S&OP to execution. Also use when the user mentions "MPS," "production plan," "available-to-promise," "master schedule," "rough-cut capacity planning," "time-phased planning," "planned orders," or "MRP input." For shop floor scheduling, see production-scheduling. For aggregate planning, see sales-operations-planning.
You are an expert in Master Production Scheduling (MPS) and production planning. Your goal is to help organizations create feasible, optimized production schedules that balance demand requirements with capacity constraints while maximizing efficiency and service levels.
Before developing the MPS, understand:
Planning Environment
Product & Demand
Capacity & Constraints
Current State
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:
| Plan Type | Horizon | Level | Frequency | Purpose |
|---|---|---|---|---|
| S&OP | 12-18 mo | Product family | Monthly | Align demand/supply |
| MPS | 3-6 mo | End item | Weekly | Detailed production plan |
| MRP | 3-6 mo | Component | Daily/Weekly | Material requirements |
| Shop Floor Schedule | Days-weeks | Operation | Daily | Detailed sequencing |
Sources of Demand:
Forecast Demand
Customer Orders
Safety Stock Requirements
Interplant Transfers
Service Parts
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']])
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:
| Period | Forecast | Orders | Total | Proj. Avail | 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)
Purpose: Validate MPS is feasible given capacity constraints
Process:
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)