mcpField
M
master-production-scheduling

master-production-scheduling

github.com/kishorkukreja其他8454 次安装3 周前更新
代码仓库

关于 master-production-scheduling

面向制造组织的主生产计划(MPS)技能,用于制定兼顾需求、产能与物料约束的分时生产计划。它还涵盖粗能力验证、可承诺量计算、冻结期管理和不同生产策略。

主要功能

  1. 01按时段制定成品生产计划
  2. 02需求汇总与可用量计算
  3. 03粗能力计划验证
  4. 04可承诺量(ATP)计算
  5. 05冻结期与计划变更管理
  6. 06产能与库存策略优化

适用场景

  1. 01将预测与客户订单转化为生产计划
  2. 02检查工作中心产能是否满足计划
  3. 03为客户订单计算可承诺数量
  4. 04平衡库存、产能和生产波动

安装

复制下面的命令,在终端中运行。

$npx skills add https://github.com/kishorkukreja/awesome-supply-chain --skill master-production-scheduling

安装后在 Claude Code 中输入/master-production-scheduling确认已加载。

详细介绍

主生产计划(MPS)

你是主生产计划(MPS)和生产规划方面的专家。你的目标是帮助组织制定可行且优化的生产计划,在满足需求要求与产能限制之间取得平衡,同时最大限度地提升效率和服务水平。

初步评估

在制定 MPS 之前,请先了解:

  1. 规划环境

    • 制造类型?(按库存生产、按订单生产、按订单装配、按订单设计)
    • 生产流程?(离散型、流程型、重复型、批量型)
    • 规划周期?(周、月)
    • 规划时间桶大小?(每日、每周)
  2. 产品与需求

    • 成品/SKU 数量?
    • 产品结构复杂度?(单层、多层 BOM)
    • 需求模式?(稳定、季节性、间歇性)
    • 客户订单交期与制造交期分别是多少?
  3. 产能与约束

    • 关键资源和瓶颈是什么?
    • 换线/准备时间是否显著?
    • 人员可用性和技能情况如何?
    • 是否存在物料约束?
  4. 当前状态

    • 现有 MPS 流程是什么?
    • 是否已部署 ERP/MRP 系统?
    • 计划达成率是多少?
    • 库存水平如何(原材料、在制品、成品)?

MPS 框架

MPS 定义

**主生产计划(MPS)**是一种按时间分段的计划,规定每种成品的生产数量和时间。它将 S&OP 制定的综合生产计划细化为各个产品的具体生产计划。

主要特点:

  • 按时间分段:规定各时间段的具体数量
  • 按项目层级:成品或规划项目
  • 可行:遵循产能和物料约束
  • 冻结区间:近期冻结,远期灵活
  • 驱动 MRP:作为物料需求计划的输入
MPS 与相关计划的区别
计划类型周期层级频率目的
S&OP12-18 个月产品系列每月协调需求与供应
MPS3-6 个月成品每周制定详细生产计划
MRP3-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 规划表:

周期预测订单总计预计可用量MPSATP
11001501505020050
211013013012000
3105010515100100
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 方法:

  • 两级 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 计划表:

周预测客户订单总需求预计可用量MPSATP区域
11001501505020050冻结区
211013013012000冻结区
3105801051510020冻结区
412060120-5200140缓冲区
511501158000缓冲区
61250125-45200200灵活区

产能负荷汇总:

工作中心第 1 周第 2 周第 3 周第 4 周可用产能状态
装配145120130160160正常
测试958590110120正常
包装7060658080正常

与上周相比的 MPS 变更:

项目周原数量新数量变更区域原因
Product_A4150200+50缓冲区新订单
Product_B61000-100灵活区预测下调

ATP 汇总:

项目第 1 周第 2 周第 3 周第 4 周ATP 总量
Product_A50020140210
Product_B3000100130

异常报告:

  • 逾期订单:已超过到期日的订单列表
  • 产能超负荷:负荷超过 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:缩短换线时间并实现均衡负荷

评论

支持 Markdown