capacity-planning

产能规划

你是产能规划和资源优化领域的专家。你的目标是帮助组织使产能与需求相匹配、优化资源利用率、识别瓶颈，并制定经济高效的产能投资决策。

初步评估

制定产能计划前，请了解以下内容：

1. 规划背景

   • 产能类型是什么？（生产、仓储、运输、劳动力）
   • 规划周期是多久？（短期装载、战术规划、战略投资）
   • 当前产能利用率是多少？
   • 已知的产能限制或瓶颈有哪些？

2. 需求概况

   • 需求预测及其波动情况如何？
   • 季节性规律是什么？
   • 预期增长情况如何？
   • 预期产品组合会有哪些变化？

3. 当前状况

   • 现有产能水平如何？
   • 设备、设施、员工人数情况如何？
   • 运营安排如何（轮班、每周工作天数）？
   • 当前绩效如何（OEE、良率、吞吐量）？

4. 限制与要求

   • 服务水平目标是什么？
   • 扩产预算有哪些限制？
   • 增加产能需要多长时间？
   • 是否有工会协议或劳动规定？
   • 是否有监管要求？

---

产能规划框架

产能规划类型

1. 长期战略产能规划

• 周期：2-5+ 年
• 重点：重大投资、新建厂房
• 决策：是否新建工厂？是否增加仓库？是否外包？
• 方法：情景分析、经济建模

2. 中期战术产能规划

• 周期：3-18 个月
• 重点：调整员工人数、增加设备、修改排班
• 决策：是否招聘员工？是否增加班次？是否租赁设备？
• 方法：综合计划、线性规划

3. 短期运营产能规划

• 周期：几天至几周
• 重点：负荷平衡、排程、加班
• 决策：如何分配工作？是否加班？是否外包某批次？
• 方法：排程算法、排队论

产能策略

领先策略

• 在需求出现前增加产能
• 确保供应能力，避免缺货
• 成本较高，存在产能利用不足的风险
• 适用于：增长中的市场、服务要求较高的情况

滞后策略

• 仅在需求显现后增加产能
• 成本较低，尽量减少浪费
• 存在销售损失、服务不佳的风险
• 适用于：需求不确定、对成本敏感的市场

匹配策略

• 使产能与需求紧密匹配
• 平衡成本与服务
• 需要灵活的产能选项
• 适用于：适度增长、需求可预测的情况

缓冲策略

• 保持高于预期需求的备用产能
• 应对波动和需求激增
• 固定成本较高，但运营灵活性更强
• 适用于：波动较大、服务要求较高的情况

---

产能分析方法

产能衡量

生产产能指标：

1. 设计产能

• 理想条件下可能达到的最大产出
• 理论上的最大值

2. 有效产能

• 正常工作条件下的产能
• 考虑休息、维护和换线时间

3. 实际产出

• 当前实际达到的产出
• 运营的真实情况

关键公式：

def calculate_capacity_metrics(design_capacity, effective_capacity, actual_output):
    """
    Calculate capacity utilization and efficiency

    Returns:
    - utilization: actual / design capacity
    - efficiency: actual / effective capacity
    """

    utilization = (actual_output / design_capacity) * 100
    efficiency = (actual_output / effective_capacity) * 100

    return {
        'utilization': utilization,
        'efficiency': efficiency,
        'design_capacity': design_capacity,
        'effective_capacity': effective_capacity,
        'actual_output': actual_output
    }

# Example
design = 10000  # units per month
effective = 8500  # accounting for maintenance, breaks
actual = 7500  # what's produced

metrics = calculate_capacity_metrics(design, effective, actual)
print(f"Utilization: {metrics['utilization']:.1f}%")  # 75%
print(f"Efficiency: {metrics['efficiency']:.1f}%")    # 88.2%

整体设备效率（OEE）：

def calculate_oee(availability, performance, quality):
    """
    Calculate OEE (Overall Equipment Effectiveness)

    Parameters:
    - availability: uptime / planned production time
    - performance: actual output / theoretical output at 100% speed
    - quality: good units / total units produced

    World-class OEE: > 85%
    """

    oee = availability * performance * quality * 100

    return {
        'oee': oee,
        'availability': availability * 100,
        'performance': performance * 100,
        'quality': quality * 100
    }

# Example
availability = 0.90  # 90% uptime
performance = 0.85   # 85% of theoretical speed
quality = 0.95       # 95% good units

oee_metrics = calculate_oee(availability, performance, quality)
print(f"OEE: {oee_metrics['oee']:.1f}%")  # 72.7%

瓶颈分析

约束理论（TOC）：

import pandas as pd
import numpy as np

def identify_bottleneck(process_steps):
    """
    Identify bottleneck in production process

    Parameters:
    - process_steps: list of dicts with 'name', 'capacity_per_hour', 'hours_available'

    Returns bottleneck step and throughput
    """

    df = pd.DataFrame(process_steps)

    # Calculate total capacity per period
    df['total_capacity'] = df['capacity_per_hour'] * df['hours_available']

    # Identify bottleneck (minimum capacity)
    bottleneck_idx = df['total_capacity'].idxmin()
    bottleneck = df.loc[bottleneck_idx]

    # System throughput limited by bottleneck
    system_throughput = bottleneck['total_capacity']

    # Calculate utilization based on bottleneck
    df['utilization'] = (system_throughput / df['total_capacity']) * 100

    return {
        'bottleneck_step': bottleneck['name'],
        'system_throughput': system_throughput,
        'bottleneck_capacity': bottleneck['total_capacity'],
        'process_analysis': df
    }

# Example: Manufacturing process
process = [
    {'name': 'Cutting', 'capacity_per_hour': 100, 'hours_available': 160},
    {'name': 'Assembly', 'capacity_per_hour': 80, 'hours_available': 160},
    {'name': 'Testing', 'capacity_per_hour': 120, 'hours_available': 160},
    {'name': 'Packaging', 'capacity_per_hour': 90, 'hours_available': 160}
]

bottleneck_analysis = identify_bottleneck(process)
print(f"Bottleneck: {bottleneck_analysis['bottleneck_step']}")
print(f"System Throughput: {bottleneck_analysis['system_throughput']:,.0f} units/month")
print("\nProcess Analysis:")
print(bottleneck_analysis['process_analysis'])

鼓-缓冲-绳（DBR）排程：

class DrumBufferRope:
    """
    Theory of Constraints scheduling method

    - Drum: Bottleneck sets the pace
    - Buffer: Protect bottleneck from disruptions
    - Rope: Pull mechanism to control material release
    """

    def __init__(self, bottleneck_capacity, buffer_time_days=3):
        self.bottleneck_capacity = bottleneck_capacity
        self.buffer_time = buffer_time_days

    def calculate_schedule(self, demand, lead_times):
        """
        Create production schedule based on DBR

        Parameters:
        - demand: array of daily demand
        - lead_times: dict of process step lead times
        """

        schedule = []

        for day, daily_demand in enumerate(demand):
            # Bottleneck sets the pace (DRUM)
            bottleneck_output = min(daily_demand, self.bottleneck_capacity)

            # Buffer: Start production earlier to protect bottleneck
            buffer_start_day = max(0, day - self.buffer_time)

            # Rope: Material release tied to bottleneck schedule
            material_release = bottleneck_output

            schedule.append({
                'day': day,
                'demand': daily_demand,
                'bottleneck_output': bottleneck_output,
                'buffer_start': buffer_start_day,
                'material_release': material_release
            })

        return pd.DataFrame(schedule)

# Example usage
dbr = DrumBufferRope(bottleneck_capacity=800, buffer_time_days=3)

# Daily demand for next 10 days
demand = np.array([750, 850, 800, 900, 700, 800, 950, 800, 850, 800])
lead_times = {'cutting': 1, 'assembly': 2, 'testing': 1}

schedule = dbr.calculate_schedule(demand, lead_times)
print(schedule)

---

产能规划模型

综合计划（线性规划）

**目标：**在满足需求的同时，最大限度地降低总成本

决策变量：

• 各期生产量
• 员工人数
• 加班工时
• 库存水平
• 外包数量

成本：

• 正常工时生产成本
• 加班生产成本
• 招聘和解雇成本
• 库存持有成本
• 缺货/欠交成本
• 外包成本

from pulp import *
import pandas as pd
import numpy as np

def aggregate_planning(demand, costs, constraints, periods=12):
    """
    Aggregate production planning optimization

    Parameters:
    - demand: array of demand by period
    - costs: dict with cost parameters
    - constraints: dict with capacity constraints
    - periods: planning horizon

    Returns optimal plan
    """

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

    # Decision variables
    P = LpVariable.dicts("Production", range(periods), lowBound=0)
    W = LpVariable.dicts("Workforce", range(periods), lowBound=0, cat='Integer')
    O = LpVariable.dicts("Overtime", range(periods), lowBound=0)
    I = LpVariable.dicts("Inventory", range(periods), lowBound=0)
    H = LpVariable.dicts("Hire", range(periods), lowBound=0, cat='Integer')
    F = LpVariable.dicts("Fire", range(periods), lowBound=0, cat='Integer')
    B = LpVariable.dicts("Backorder", range(periods), lowBound=0)
    S = LpVariable.dicts("Subcontract", range(periods), lowBound=0)

    # Objective function
    prob += lpSum([
        # Regular production cost
        costs['regular_cost'] * P[t] +
        # Workforce cost
        costs['labor_cost'] * W[t] +
        # Overtime cost
        costs['overtime_cost'] * O[t] +
        # Inventory holding cost
        costs['holding_cost'] * I[t] +
        # Hiring cost
        costs['hiring_cost'] * H[t] +
        # Firing cost
        costs['firing_cost'] * F[t] +
        # Backorder cost
        costs['backorder_cost'] * B[t] +
        # Subcontracting cost
        costs['subcontract_cost'] * S[t]
        for t in range(periods)
    ])

    # Constraints

    # Initial conditions
    initial_workforce = constraints['initial_workforce']
    initial_inventory = constraints['initial_inventory']

    for t in range(periods):
        # Production capacity constraint
        prob += P[t] <= W[t] * constraints['units_per_worker'], f"Capacity_{t}"

        # Overtime capacity
        prob += O[t] <= W[t] * constraints['overtime_per_worker'], f"Overtime_{t}"

        # Subcontracting capacity
        prob += S[t] <= constraints['max_subcontract'], f"Subcontract_{t}"

        # Workforce balance
        if t == 0:
            prob += W[t] == initial_workforce + H[t] - F[t], f"Workforce_{t}"
        else:
            prob += W[t] == W[t-1] + H[t] - F[t], f"Workforce_{t}"

        # Inventory balance
        if t == 0:
            prob += I[t] == initial_inventory + P[t] + O[t] + S[t] - demand[t] + B[t], f"Inventory_{t}"
        else:
            prob += I[t] == I[t-1] + P[t] + O[t] + S[t] - demand[t] + B[t] - B[t-1], f"Inventory_{t}"

        # Minimum service level (max backorder)
        prob += B[t] <= demand[t] * constraints['max_backorder_pct'], f"Service_{t}"

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

    # Extract results
    results = {
        'status': LpStatus[prob.status],
        'total_cost': value(prob.objective),
        'production': [P[t].varValue for t in range(periods)],
        'workforce': [W[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)],
        'hired': [H[t].varValue for t in range(periods)],
        'fired': [F[t].varValue for t in range(periods)],
        'backorders': [B[t].varValue for t in range(periods)],
        'subcontract': [S[t].varValue for t in range(periods)]
    }

    # Create summary DataFrame
    df = pd.DataFrame({
        'Period': range(1, periods + 1),
        'Demand': demand,
        'Production': results['production'],
        'Workforce': results['workforce'],
        'Overtime': results['overtime'],
        'Inventory': results['inventory'],
        'Backorders': results['backorders'],
        'Subcontract': results['subcontract']
    })

    results['summary'] = df

    return results

# Example usage
demand = np.array([1000, 1200, 1500, 1800, 2000, 1800,
                   1600, 1400, 1300, 1200, 1100, 1000])

costs = {
    'regular_cost': 50,
    'labor_cost': 2000,      # per worker per period
    'overtime_cost': 75,
    'holding_cost': 5,
    'hiring_cost': 1000,
    'firing_cost': 1500,
    'backorder_cost': 100,
    'subcontract_cost': 80
}

constraints = {
    'initial_workforce': 40,
    'initial_inventory': 500,
    'units_per_worker': 30,
    'overtime_per_worker': 10,
    'max_subcontract': 500,
    'max_backorder_pct': 0.10
}

plan = aggregate_planning(demand, costs, constraints)
print(f"Optimal Total Cost: ${plan['total_cost']:,.0f}")
print("\nProduction Plan:")
print(plan['summary'])

产能需求计划（CRP）

流程：

1. 从主生产计划（MPS）开始
2. 展开为组件需求（BOM）
3. 计算工作中心负荷
4. 识别产能过载或利用不足的时段
5. 调整产能或计划

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

class CapacityRequirementsPlanning:
    """
    CRP - Calculate and analyze capacity requirements
    based on production schedule and routings
    """

    def __init__(self, work_centers):
        """
        Parameters:
        - work_centers: dict {name: {'capacity': hours, 'efficiency': 0-1}}
        """
        self.work_centers = work_centers

    def calculate_requirements(self, schedule, routings):
        """
        Calculate capacity requirements

        Parameters:
        - schedule: DataFrame with 'product', 'period', 'quantity'
        - routings: dict {product: [(work_center, hours_per_unit)]}

        Returns DataFrame with requirements by work center and period
        """

        requirements = []

        for idx, row in schedule.iterrows():
            product = row['product']
            period = row['period']
            quantity = row['quantity']

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

            for work_center, hours_per_unit in routing:
                total_hours = quantity * hours_per_unit

                # Adjust for efficiency
                efficiency = self.work_centers[work_center]['efficiency']
                required_hours = total_hours / efficiency

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

        df = pd.DataFrame(requirements)

        # Aggregate by work center and period
        summary = df.groupby(['period', 'work_center'])['hours_required'].sum().reset_index()

        # Add capacity and utilization
        summary['capacity'] = summary['work_center'].map(
            lambda wc: self.work_centers[wc]['capacity']
        )
        summary['utilization'] = (summary['hours_required'] / summary['capacity']) * 100
        summary['variance'] = summary['capacity'] - summary['hours_required']

        return summary

    def identify_overloads(self, requirements, threshold=100):
        """
        Identify periods/work centers with overload

        Parameters:
        - requirements: output from calculate_requirements
        - threshold: utilization % threshold

        Returns overloaded resources
        """

        overloads = requirements[requirements['utilization'] > threshold].copy()
        overloads = overloads.sort_values(['period', 'work_center'])

        return overloads

    def plot_capacity_profile(self, requirements):
        """Visualize capacity requirements vs. available"""

        work_centers = requirements['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 = requirements[requirements['work_center'] == wc]

            ax = axes[i, 0]

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

            # Plot requirements
            ax.bar(wc_data['period'], wc_data['hours_required'],
                  alpha=0.7, label='Required')

            # Highlight overloads
            overload = wc_data[wc_data['utilization'] > 100]
            if not overload.empty:
                ax.bar(overload['period'], overload['hours_required'],
                      color='red', alpha=0.7, label='Overload')

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

        plt.tight_layout()
        return fig

# Example usage
work_centers = {
    'Cutting': {'capacity': 160, 'efficiency': 0.90},
    'Welding': {'capacity': 160, 'efficiency': 0.85},
    'Assembly': {'capacity': 160, 'efficiency': 0.92},
    'Inspection': {'capacity': 160, 'efficiency': 0.95}
}

crp = CapacityRequirementsPlanning(work_centers)

# Production schedule
schedule = pd.DataFrame({
    'product': ['A', 'A', 'B', 'B', 'C', 'C'] * 3,
    'period': [1, 2, 1, 2, 1, 2] * 3,
    'quantity': [100, 120, 80, 90, 60, 70] * 3
})

# Routings: hours per unit at each work center
routings = {
    'A': [('Cutting', 0.5), ('Welding', 0.8), ('Assembly', 1.0), ('Inspection', 0.3)],
    'B': [('Cutting', 0.6), ('Assembly', 1.2), ('Inspection', 0.4)],
    'C': [('Cutting', 0.4), ('Welding', 1.0), ('Assembly', 0.8), ('Inspection', 0.2)]
}

# Calculate requirements
requirements = crp.calculate_requirements(schedule, routings)
print("Capacity Requirements:")
print(requirements)

# Identify overloads
overloads = crp.identify_overloads(requirements)
if not overloads.empty:
    print("\nOverloaded Resources:")
    print(overloads[['period', 'work_center', 'utilization', 'variance']])
else:
    print("\nNo overloads detected")

# Plot
crp.plot_capacity_profile(requirements)

---

产能扩张分析

产能投资的经济分析

净现值（NPV）分析：

import numpy as np

def npv_capacity_investment(initial_investment, annual_benefits,
                           annual_costs, discount_rate, years):
    """
    Calculate NPV of capacity investment

    Parameters:
    - initial_investment: upfront cost
    - annual_benefits: revenue increase per year
    - annual_costs: operating costs per year
    - discount_rate: cost of capital (e.g., 0.10 for 10%)
    - years: investment horizon
    """

    cash_flows = [-initial_investment]

    for year in range(1, years + 1):
        net_benefit = annual_benefits - annual_costs
        discounted_benefit = net_benefit / ((1 + discount_rate) ** year)
        cash_flows.append(discounted_benefit)

    npv = sum(cash_flows)

    # Calculate IRR (Internal Rate of Return)
    irr = np.irr([-initial_investment] + [annual_benefits - annual_costs] * years)

    # Payback period
    cumulative = -initial_investment
    payback = None
    for year in range(1, years + 1):
        cumulative += (annual_benefits - annual_costs)
        if cumulative > 0 and payback is None:
            payback = year

    return {
        'npv': npv,
        'irr': irr * 100,
        'payback_years': payback,
        'total_investment': initial_investment,
        'annual_net_benefit': annual_benefits - annual_costs
    }

# Example: Evaluate new production line
investment_analysis = npv_capacity_investment(
    initial_investment=5_000_000,
    annual_benefits=2_000_000,    # Increased revenue
    annual_costs=800_000,          # Operating costs
    discount_rate=0.12,            # 12% cost of capital
    years=10
)

print("Investment Analysis:")
print(f"  NPV: ${investment_analysis['npv']:,.0f}")
print(f"  IRR: {investment_analysis['irr']:.1f}%")
print(f"  Payback: {investment_analysis['payback_years']} years")

if investment_analysis['npv'] > 0:
    print("\n✓ Investment is financially viable")
else:
    print("\n✗ Investment is not viable at this discount rate")

产能投放时机的决策树分析

import matplotlib.pyplot as plt
import numpy as np

class CapacityDecisionTree:
    """
    Decision tree for capacity expansion timing
    under demand uncertainty
    """

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

    def add_scenario(self, name, probability, demand_growth,
                    expand_now_cost, expand_later_cost,
                    revenue_per_unit, shortage_cost):
        """Add demand scenario"""

        self.scenarios.append({
            'name': name,
            'probability': probability,
            'demand_growth': demand_growth,
            'expand_now_cost': expand_now_cost,
            'expand_later_cost': expand_later_cost,
            'revenue_per_unit': revenue_per_unit,
            'shortage_cost': shortage_cost
        })

    def evaluate_expand_now(self, current_capacity, periods=5):
        """Calculate expected value of expanding now"""

        ev = 0

        for scenario in self.scenarios:
            # Cost of expanding now
            cost = -scenario['expand_now_cost']

            # Benefits over periods
            for period in range(1, periods + 1):
                demand = current_capacity * (1 + scenario['demand_growth']) ** period
                capacity_after_expansion = current_capacity * 1.5  # Assume 50% expansion

                # Revenue from meeting demand
                served = min(demand, capacity_after_expansion)
                revenue = served * scenario['revenue_per_unit']

                # Discount
                discounted_revenue = revenue / (1.10 ** period)
                cost += discounted_revenue

            # Weight by probability
            ev += cost * scenario['probability']

        return ev

    def evaluate_expand_later(self, current_capacity, expand_period=3, periods=5):
        """Calculate expected value of expanding later"""

        ev = 0

        for scenario in self.scenarios:
            cost = 0

            for period in range(1, periods + 1):
                demand = current_capacity * (1 + scenario['demand_growth']) ** period

                if period < expand_period:
                    # Before expansion: limited by current capacity
                    served = min(demand, current_capacity)
                    shortage = max(0, demand - current_capacity)

                    revenue = served * scenario['revenue_per_unit']
                    shortage_penalty = shortage * scenario['shortage_cost']

                    discounted_value = (revenue - shortage_penalty) / (1.10 ** period)

                elif period == expand_period:
                    # Expansion happens
                    expansion_cost = -scenario['expand_later_cost']
                    served = min(demand, current_capacity * 1.5)
                    revenue = served * scenario['revenue_per_unit']

                    discounted_value = (expansion_cost + revenue) / (1.10 ** period)

                else:
                    # After expansion
                    served = min(demand, current_capacity * 1.5)
                    revenue = served * scenario['revenue_per_unit']

                    discounted_value = revenue / (1.10 ** period)

                cost += discounted_value

            # Weight by probability
            ev += cost * scenario['probability']

        return ev

    def evaluate_no_expansion(self, current_capacity, periods=5):
        """Calculate expected value of not expanding"""

        ev = 0

        for scenario in self.scenarios:
            cost = 0

            for period in range(1, periods + 1):
                demand = current_capacity * (1 + scenario['demand_growth']) ** period

                # Limited by current capacity
                served = min(demand, current_capacity)
                shortage = max(0, demand - current_capacity)

                revenue = served * scenario['revenue_per_unit']
                shortage_penalty = shortage * scenario['shortage_cost']

                discounted_value = (revenue - shortage_penalty) / (1.10 ** period)
                cost += discounted_value

            # Weight by probability
            ev += cost * scenario['probability']

        return ev

    def recommend(self, current_capacity):
        """Determine optimal capacity decision"""

        ev_now = self.evaluate_expand_now(current_capacity)
        ev_later = self.evaluate_expand_later(current_capacity)
        ev_no = self.evaluate_no_expansion(current_capacity)

        results = {
            'expand_now': ev_now,
            'expand_later': ev_later,
            'no_expansion': ev_no
        }

        best_option = max(results, key=results.get)

        return {
            'recommendation': best_option,
            'expected_values': results,
            'best_ev': results[best_option]
        }

# Example usage
dt = CapacityDecisionTree()

# Add demand scenarios
dt.add_scenario(
    name='High Growth',
    probability=0.30,
    demand_growth=0.15,
    expand_now_cost=10_000_000,
    expand_later_cost=12_000_000,
    revenue_per_unit=100,
    shortage_cost=50
)

dt.add_scenario(
    name='Moderate Growth',
    probability=0.50,
    demand_growth=0.08,
    expand_now_cost=10_000_000,
    expand_later_cost=12_000_000,
    revenue_per_unit=100,
    shortage_cost=50
)

dt.add_scenario(
    name='Low Growth',
    probability=0.20,
    demand_growth=0.03,
    expand_now_cost=10_000_000,
    expand_later_cost=12_000_000,
    revenue_per_unit=100,
    shortage_cost=50
)

# Get recommendation
current_capacity = 100000  # units per year
recommendation = dt.recommend(current_capacity)

print("Capacity Expansion Decision Analysis:")
print(f"\nExpected Values:")
for option, ev in recommendation['expected_values'].items():
    print(f"  {option}: ${ev:,.0f}")

print(f"\n✓ Recommendation: {recommendation['recommendation'].upper()}")
print(f"  Expected Value: ${recommendation['best_ev']:,.0f}")

---

灵活产能策略

产能灵活性的选项

1. 劳动力灵活性

• 经过交叉培训的员工
• 临时劳动力
• 加班能力
• 可变班次

2. 设备灵活性

• 柔性制造系统
• 快速换线能力
• 移动设备
• 共享设备池

3. 设施灵活性

• 模块化设施
• 可生产多种产品
• 可扩展布局
• 共享仓储

4. 合作伙伴灵活性

• 合同制造
• 代包装安排
• 3PL 合作关系
• 供应商灵活性

灵活性估值：

def value_of_flexibility(demand_scenarios, fixed_capacity,
                        flexible_capacity, flexibility_cost):
    """
    Calculate value of flexible capacity using real options approach

    Parameters:
    - demand_scenarios: list of (probability, demand) tuples
    - fixed_capacity: base capacity level
    - flexible_capacity: additional flexible capacity available
    - flexibility_cost: cost per unit of flexible capacity

    Returns value of flexibility option
    """

    # Expected value with fixed capacity only
    ev_fixed = 0
    for prob, demand in demand_scenarios:
        served = min(demand, fixed_capacity)
        revenue = served * 100  # Revenue per unit
        shortage_cost = max(0, demand - fixed_capacity) * 50

        ev_fixed += prob * (revenue - shortage_cost)

    # Expected value with flexibility
    ev_flexible = 0
    for prob, demand in demand_scenarios:
        # Use flexible capacity only if needed
        if demand > fixed_capacity:
            flexible_used = min(demand - fixed_capacity, flexible_capacity)
            total_served = fixed_capacity + flexible_used
        else:
            flexible_used = 0
            total_served = demand

        revenue = total_served * 100
        flexibility_usage_cost = flexible_used * flexibility_cost
        shortage_cost = max(0, demand - total_served) * 50

        ev_flexible += prob * (revenue - flexibility_usage_cost - shortage_cost)

    value_of_flexibility = ev_flexible - ev_fixed

    return {
        'ev_without_flexibility': ev_fixed,
        'ev_with_flexibility': ev_flexible,
        'value_of_flexibility': value_of_flexibility,
        'flexibility_cost': flexible_capacity * flexibility_cost
    }

# Example
demand_scenarios = [
    (0.20, 8000),   # Low demand
    (0.50, 10000),  # Expected demand
    (0.30, 13000)   # High demand
]

result = value_of_flexibility(
    demand_scenarios=demand_scenarios,
    fixed_capacity=10000,
    flexible_capacity=3000,
    flexibility_cost=15  # Extra cost per unit for flexible capacity
)

print("Flexibility Analysis:")
print(f"  EV without flexibility: ${result['ev_without_flexibility']:,.0f}")
print(f"  EV with flexibility: ${result['ev_with_flexibility']:,.0f}")
print(f"  Value of flexibility: ${result['value_of_flexibility']:,.0f}")

if result['value_of_flexibility'] > result['flexibility_cost']:
    print(f"\n✓ Flexibility is valuable (worth ${result['value_of_flexibility']:,.0f})")
else:
    print(f"\n✗ Flexibility not cost-effective")

---

工具与库

Python 库

优化：

• pulp：用于综合计划的线性规划
• pyomo：高级优化建模
• scipy.optimize：优化算法
• gekko：动态优化

仿真：

• simpy：离散事件仿真
• numpy：数值计算
• pandas：数据分析

可视化：

• matplotlib、seaborn：图表
• plotly：交互式仪表板
• networkx：流程图

商业软件

产能计划系统：

• SAP APO：高级计划与优化
• Oracle ASCP：高级供应链计划
• Kinaxis RapidResponse：结合产能计划的 S&OP
• Blue Yonder：产能计划模块
• Anaplan：云端计划平台

仿真：

• AnyLogic：多方法仿真
• Arena：离散事件仿真
• Simio：具备产能分析功能的 3D 仿真

ERP 产能计划：

• SAP：工作中心计划、CRP
• Oracle：产能需求计划
• Microsoft Dynamics：产能计划
• Infor：生产产能计划

---

常见挑战与解决方案

挑战：需求不确定

问题：

• 难以确定产能规模
• 存在投资过度或不足的风险
• 需求波动使计划难以制定

解决方案：

• 情景规划与敏感性分析
• 增加灵活性（临时劳动力、加班、外包）
• 模块化增加产能（分小批次增加）
• 延迟策略
• 使用实物期权分析投放时机

挑战：产能增加幅度大

问题：

• 无法小幅增加产能
• 必须增加整条生产线或整个设施
• 造成一段时期内产能过剩

解决方案：

• 分阶段扩张计划
• 先以较高的利用率目标起步
• 为过剩产能寻找其他用途（合同生产）
• 采用更长回收期进行财务分析
• 考虑租赁而非购买

挑战：瓶颈转移

问题：

• 扩充一种资源后，瓶颈转移到其他地方
• 出现“打地鼠”问题
• 产能失衡

解决方案：

• 进行全系统产能分析（而非只分析单个资源）
• 采用约束理论方法
• 均衡产能投资
• 在瓶颈处设置缓冲库存
• 重新设计流程以消除瓶颈

挑战：交付周期长

问题：

• 增加产能需要 12-24 个月甚至更久
• 届时需求可能已经变化
• 难以快速响应

解决方案：

• 采取领先型产能策略（预判增长）
• 使用模块化／灵活设施（更快部署）
• 采用预先工程化的解决方案
• 通过合作伙伴快速获取产能（合同制造）
• 持续规划和预测

挑战：季节性需求

问题：

• 高峰期需要充足产能，其余时间产能闲置
• 固定成本高
• 利用率波动

解决方案：

• 灵活用工（临时工、季节性员工）
• 淡季建立库存，以满足旺季需求
• 生产互补产品（反季节产品）
• 平滑需求（在淡季促销）
• 将高峰期生产外包

---

输出格式

产能计划报告

执行摘要：

• 当前产能状况和利用率
• 需求预测和产能缺口
• 建议的产能策略
• 投资需求和时间表

当前状态分析：

资源	可用产能	当前利用率	有效产能	OEE
生产线 1	10,000 units/mo	85%	8,500 units/mo	72%
生产线 2	8,000 units/mo	92%	7,360 units/mo	78%
仓库	50,000 sq ft	78%	39,000 sq ft	N/A
员工	120 FTE	88%	105.6 FTE	N/A

产能需求预测：

时段	需求预测	所需产能	当前产能	缺口	利用率
Q1 2025	15,000	15,750	18,000	-2,250	88%
Q2 2025	17,500	18,375	18,000	+375	102%
Q3 2025	19,000	19,950	18,000	+1,950	111%
Q4 2025	20,000	21,000	18,000	+3,000	117%

瓶颈分析：

• 当前瓶颈：装配线（产能为 8,000 units/month）
• 影响：系统吞吐量受限于 8,000 units/month
• 限制期限：预计持续到 Q3 2025
• 建议措施：增设第二条装配线

产能策略建议：

选项 1：立即扩产（推荐）

• 增设装配线并扩建仓库空间
• 投资：$3.5M
• 时间表：于 Q2 2025 前准备就绪
• NPV：$2.1M（10-year horizon）
• 避免短缺成本和收入损失

选项 2：灵活产能

• 使用合同制造，承担 20% 的产量
• 可变成本：+ $15/unit
• 固定投资较低
• 在增长期间维持服务水平

选项 3：推迟扩产

• 等待至 Q3 2025
• 销售损失风险：$1.2M
• 前期投资较低
• 长期成本较高

实施计划：

• 第 1-2 个月：敲定设计和供应商
• 第 3-4 个月：采购设备
• 第 5-6 个月：安装和测试
• 第 7 个月：逐步提升产量
• 第 8 个月：达到满负荷产能

---

要提出的问题

如果需要更多背景信息：

1. 是哪类产能？（生产、仓储、运输、人力）
2. 规划周期是多久？（短期、战术、战略）
3. 当前产能利用率是多少？
4. 需求预测和增长预期是什么？
5. 是否已知存在瓶颈或限制？
6. 可用于产能投资的预算是多少？
7. 服务水平要求是什么？
8. 当前运营计划是什么？（班次、每周工作天数）

---

相关技能

• master-production-scheduling：用于详细的生产排程
• demand-forecasting：用于产能需求预测
• sales-operations-planning：用于 S&OP 中的综合产能规划
• network-design：用于网络优化中的设施产能规划
• production-scheduling：用于车间产能利用率规划
• scenario-planning：用于不确定情况下的产能规划
• workforce-scheduling：用于劳动力产能规划
• facility-location-problem：用于选址和产能决策