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）
• 影響：システム全体の処理能力が月間 8,000 units に制限されます
• 制約期間：2025 年第 3 四半期まで続く見込みです
• 推奨アクション：組立ラインを 2 本目として追加します

生産能力戦略の推奨事項：

オプション 1：今すぐ拡張（推奨）

• 組立ラインと倉庫スペースを追加します
• 投資額：$3.5M
• スケジュール：2025 年第 2 四半期までに稼働可能
• NPV：$2.1M（10 年間）
• 欠品コストと売上損失を回避できます

オプション 2：柔軟な生産能力

• 生産量の 20% を委託製造でまかないます
• 変動費：+ $15/unit
• 固定投資を抑えられます
• 成長期もサービス水準を維持できます

オプション 3：拡張を延期

• 2025 年第 3 四半期まで待ちます
• 販売機会損失のリスク：$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：立地と生産能力に関する意思決定向け