scenario-planning

场景规划

你是供应链场景规划和风险分析方面的专家。你的目标是帮助组织评估不同的未来情景、评估风险、制定应急计划，并在不确定性下做出稳健决策。

初步评估

在制定场景计划之前，请先了解：

1. 规划背景

   • 需要做出哪些决策？
   • 哪些不确定性或风险最令人担忧？
   • 规划期限是多久？（短期战术规划还是长期战略规划）
   • 当前的规划流程和工具是什么？

2. 业务环境

   • 主要的市场不确定性有哪些？（需求、供应、定价）
   • 外部风险有哪些？（地缘政治、自然灾害、经济）
   • 内部脆弱点有哪些？（单一来源供应商、产能限制）
   • 过去经历过哪些中断？

3. 范围与目标

   • 供应链的哪些部分？（端到端、特定环节）
   • 进行定量分析还是定性场景分析？
   • 是一次性分析还是持续开展的流程？
   • 决策背景是什么？（投资、战略、运营）

4. 数据与资源

   • 是否有可用的历史波动数据？
   • 是否有成本和绩效数据？
   • 是否有风险数据库或事件日志？
   • 是否具备建模工具和能力？

---

场景规划框架

场景分析类型

1. 敏感性分析

• 测试单个变量变化的影响
• “如果需求增长 10%，会怎样？”
• 简单快捷地获取见解
• 深入分析的基础

2. 假设情景

• 基于特定事件的场景
• “如果供应商 X 失败，会怎样？”
• 明确定义的离散事件
• 测试准备情况

3. 蒙特卡洛模拟

• 通过随机抽样进行概率分析
• 数千种场景
• 生成概率分布
• 量化风险

4. 战略场景

• 长期的不同未来情景
• 多个变量共同变化
• 定性 + 定量
• 3-5 年期

5. 压力测试

• 极端事件场景
• 测试系统韧性
• 找出临界点
• 满足监管或内部要求

---

场景制定流程

阶段 1：明确范围与目标

关键问题：

• 要支持哪项决策？
• 我们想要了解什么？
• 需要关注哪个时间范围？
• 谁需要使用这项分析？

**输出：**明确的问题陈述和成功标准

阶段 2：识别关键不确定性

外部不确定性：

• 需求波动：市场变化、客户行为
• 供应中断：供应商故障、自然灾害
• 成本波动：原材料、运输、劳动力
• 监管变化：贸易政策、环境法规
• 技术颠覆：新竞争对手、技术淘汰
• 经济状况：经济衰退、通货膨胀、汇率

内部不确定性：

• 生产良率和质量
• 设备可靠性
• 劳动力可用性
• IT 系统性能
• 新产品上市的成功与否

优先级矩阵：

import pandas as pd
import matplotlib.pyplot as plt

def prioritize_uncertainties(uncertainties_data):
    """
    Prioritize uncertainties by impact and likelihood

    Parameters:
    - uncertainties_data: list of dicts with 'name', 'impact', 'likelihood'
    """
    df = pd.DataFrame(uncertainties_data)

    # Calculate priority score
    df['priority_score'] = df['impact'] * df['likelihood']
    df = df.sort_values('priority_score', ascending=False)

    # Plot impact-likelihood matrix
    plt.figure(figsize=(10, 8))
    plt.scatter(df['likelihood'], df['impact'], s=200, alpha=0.6)

    for idx, row in df.iterrows():
        plt.annotate(row['name'],
                    (row['likelihood'], row['impact']),
                    fontsize=8)

    plt.xlabel('Likelihood (1-10)')
    plt.ylabel('Business Impact (1-10)')
    plt.title('Uncertainty Prioritization Matrix')
    plt.grid(True, alpha=0.3)

    # Add quadrant lines
    plt.axhline(y=5, color='r', linestyle='--', alpha=0.3)
    plt.axvline(x=5, color='r', linestyle='--', alpha=0.3)

    return df

# Example usage
uncertainties = [
    {'name': 'Supplier Bankruptcy', 'impact': 9, 'likelihood': 3},
    {'name': 'Demand Spike', 'impact': 7, 'likelihood': 6},
    {'name': 'Port Strike', 'impact': 8, 'likelihood': 4},
    {'name': 'Fuel Price Increase', 'impact': 6, 'likelihood': 7},
    {'name': 'Quality Issue', 'impact': 7, 'likelihood': 5}
]

priority_df = prioritize_uncertainties(uncertainties)

阶段 3：定义场景

方法：

A. 基于驱动因素的场景

• 选择 2-3 个关键不确定性驱动因素
• 为每个因素定义高位/低位状态
• 创建 2x2 或 2x2x2 场景矩阵

示例：需求增长与供应稳定性

	低需求增长	高需求增长
供应稳定	优化	扩大产能
供应不稳定	整合	高风险/高回报

B. 基于事件的场景

• 特定的颠覆性事件
• 示例：飓风、网络攻击、供应商破产
• 详细的影响建模

C. 蒙特卡洛场景

• 为不确定变量设定统计分布
• 随机抽样
• 大量场景（1000+）

阶段 4：建立场景模型

建立基础模型：

import numpy as np
import pandas as pd
from dataclasses import dataclass
from typing import Dict, List

@dataclass
class SupplyChainState:
    """Represents supply chain configuration"""
    facilities: List[str]
    suppliers: Dict[str, float]  # supplier: reliability
    inventory_levels: Dict[str, float]  # location: units
    capacity: Dict[str, float]  # facility: units/month
    costs: Dict[str, float]

class ScenarioModel:
    """Supply chain scenario modeling"""

    def __init__(self, base_state: SupplyChainState):
        self.base_state = base_state
        self.scenarios = {}

    def add_scenario(self, name: str, changes: Dict):
        """Define a scenario with changes from base"""
        self.scenarios[name] = changes

    def evaluate_scenario(self, scenario_name: str,
                         demand: np.ndarray,
                         time_periods: int = 12) -> Dict:
        """
        Simulate supply chain performance under scenario

        Returns metrics: cost, service level, inventory
        """

        scenario_changes = self.scenarios[scenario_name]

        # Apply scenario changes to base state
        state = self._apply_changes(self.base_state, scenario_changes)

        # Simulate over time periods
        results = {
            'total_cost': 0,
            'service_level': [],
            'inventory': [],
            'stockouts': 0,
            'periods': []
        }

        inventory = state.inventory_levels.copy()

        for t in range(time_periods):
            period_demand = demand[t]

            # Check if can meet demand
            total_inventory = sum(inventory.values())

            if total_inventory >= period_demand:
                # Meet demand
                service = 1.0
                stockout = 0
                units_sold = period_demand
            else:
                # Stockout
                service = total_inventory / period_demand
                stockout = period_demand - total_inventory
                units_sold = total_inventory

            # Calculate costs
            holding_cost = sum(inventory.values()) * state.costs.get('holding', 1.0)
            stockout_cost = stockout * state.costs.get('stockout', 100.0)
            period_cost = holding_cost + stockout_cost

            # Production/replenishment
            production = min(
                period_demand * 1.2,  # Target 120% of demand
                sum(state.capacity.values())
            )

            # Apply supply reliability
            actual_production = production * scenario_changes.get('supply_reliability', 1.0)

            # Update inventory
            for loc in inventory:
                inventory[loc] = max(0, inventory[loc] - units_sold / len(inventory))
                inventory[loc] += actual_production / len(inventory)

            # Record results
            results['total_cost'] += period_cost
            results['service_level'].append(service)
            results['inventory'].append(sum(inventory.values()))
            results['stockouts'] += stockout
            results['periods'].append(t)

        # Calculate summary metrics
        results['avg_service_level'] = np.mean(results['service_level'])
        results['avg_inventory'] = np.mean(results['inventory'])
        results['total_stockouts'] = results['stockouts']

        return results

    def _apply_changes(self, base_state, changes):
        """Apply scenario changes to base state"""
        import copy
        state = copy.deepcopy(base_state)

        # Apply modifications based on scenario
        if 'capacity_reduction' in changes:
            for facility in state.capacity:
                state.capacity[facility] *= (1 - changes['capacity_reduction'])

        if 'cost_increase' in changes:
            for cost_type in state.costs:
                state.costs[cost_type] *= (1 + changes['cost_increase'])

        return state

    def compare_scenarios(self, demand: np.ndarray) -> pd.DataFrame:
        """Compare all scenarios"""

        results_list = []

        for scenario_name in self.scenarios:
            result = self.evaluate_scenario(scenario_name, demand)
            results_list.append({
                'Scenario': scenario_name,
                'Total_Cost': result['total_cost'],
                'Avg_Service_Level': result['avg_service_level'] * 100,
                'Avg_Inventory': result['avg_inventory'],
                'Total_Stockouts': result['total_stockouts']
            })

        return pd.DataFrame(results_list)

# Example usage
base = SupplyChainState(
    facilities=['DC_East', 'DC_West'],
    suppliers={'Supplier_A': 0.95, 'Supplier_B': 0.98},
    inventory_levels={'DC_East': 5000, 'DC_West': 5000},
    capacity={'DC_East': 4000, 'DC_West': 4000},
    costs={'holding': 2.0, 'stockout': 150.0}
)

model = ScenarioModel(base)

# Define scenarios
model.add_scenario('Base_Case', {
    'supply_reliability': 1.0,
    'capacity_reduction': 0.0,
    'cost_increase': 0.0
})

model.add_scenario('Supplier_Disruption', {
    'supply_reliability': 0.6,
    'capacity_reduction': 0.0,
    'cost_increase': 0.0
})

model.add_scenario('Capacity_Constraint', {
    'supply_reliability': 1.0,
    'capacity_reduction': 0.3,
    'cost_increase': 0.0
})

model.add_scenario('Cost_Inflation', {
    'supply_reliability': 1.0,
    'capacity_reduction': 0.0,
    'cost_increase': 0.25
})

model.add_scenario('Perfect_Storm', {
    'supply_reliability': 0.7,
    'capacity_reduction': 0.2,
    'cost_increase': 0.3
})

# Generate demand scenarios
np.random.seed(42)
demand = np.random.normal(8000, 1000, 12)  # 12 months

# Compare scenarios
comparison = model.compare_scenarios(demand)
print(comparison)

---

蒙特卡洛模拟

概率情景分析

适用场景：

• 存在多个不确定变量
• 需要概率分布，而非点估计
• 需要量化风险
• 需要做投资组合或投资决策

流程：

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from scipy import stats

class MonteCarloScenarioAnalysis:
    """Monte Carlo simulation for supply chain scenarios"""

    def __init__(self, n_simulations=1000):
        self.n_simulations = n_simulations
        self.results = None

    def simulate_supply_chain(self,
                             demand_params: Dict,
                             cost_params: Dict,
                             supply_params: Dict,
                             periods: int = 12):
        """
        Run Monte Carlo simulation

        Parameters:
        - demand_params: {'mean': x, 'std': y, 'distribution': 'normal'}
        - cost_params: {'transport': {...}, 'holding': {...}}
        - supply_params: {'reliability': {...}}
        """

        results = []

        for sim in range(self.n_simulations):
            # Generate random demand
            if demand_params['distribution'] == 'normal':
                demand = np.random.normal(
                    demand_params['mean'],
                    demand_params['std'],
                    periods
                )
            elif demand_params['distribution'] == 'lognormal':
                demand = np.random.lognormal(
                    np.log(demand_params['mean']),
                    demand_params['std'],
                    periods
                )

            # Generate random costs
            transport_cost = np.random.uniform(
                cost_params['transport']['min'],
                cost_params['transport']['max']
            )

            holding_cost = np.random.normal(
                cost_params['holding']['mean'],
                cost_params['holding']['std']
            )

            # Generate supply reliability
            supply_reliability = np.random.beta(
                supply_params['reliability']['alpha'],
                supply_params['reliability']['beta']
            )

            # Simulate supply chain performance
            total_demand = demand.sum()
            actual_supply = total_demand * supply_reliability
            shortage = max(0, total_demand - actual_supply)

            # Calculate total cost
            transport_cost_total = actual_supply * transport_cost
            holding_cost_total = actual_supply * holding_cost * 0.1
            shortage_cost_total = shortage * 200  # Penalty

            total_cost = (transport_cost_total +
                         holding_cost_total +
                         shortage_cost_total)

            service_level = (actual_supply / total_demand) * 100

            results.append({
                'simulation': sim,
                'total_demand': total_demand,
                'actual_supply': actual_supply,
                'shortage': shortage,
                'transport_cost': transport_cost_total,
                'holding_cost': holding_cost_total,
                'shortage_cost': shortage_cost_total,
                'total_cost': total_cost,
                'service_level': service_level
            })

        self.results = pd.DataFrame(results)
        return self.results

    def analyze_results(self):
        """Statistical analysis of simulation results"""

        if self.results is None:
            raise ValueError("Run simulation first")

        analysis = {
            'total_cost': {
                'mean': self.results['total_cost'].mean(),
                'std': self.results['total_cost'].std(),
                'p10': self.results['total_cost'].quantile(0.10),
                'p50': self.results['total_cost'].quantile(0.50),
                'p90': self.results['total_cost'].quantile(0.90),
                'p95': self.results['total_cost'].quantile(0.95),
                'min': self.results['total_cost'].min(),
                'max': self.results['total_cost'].max()
            },
            'service_level': {
                'mean': self.results['service_level'].mean(),
                'std': self.results['service_level'].std(),
                'p10': self.results['service_level'].quantile(0.10),
                'p50': self.results['service_level'].quantile(0.50),
                'p90': self.results['service_level'].quantile(0.90)
            }
        }

        return analysis

    def value_at_risk(self, confidence_level=0.95):
        """Calculate Value at Risk (VaR) for costs"""

        if self.results is None:
            raise ValueError("Run simulation first")

        var = self.results['total_cost'].quantile(confidence_level)

        # Conditional VaR (CVaR) - expected loss beyond VaR
        cvar = self.results[
            self.results['total_cost'] >= var
        ]['total_cost'].mean()

        return {'VaR': var, 'CVaR': cvar}

    def plot_distributions(self):
        """Visualize simulation results"""

        fig, axes = plt.subplots(2, 2, figsize=(14, 10))

        # Total cost distribution
        axes[0, 0].hist(self.results['total_cost'], bins=50,
                       edgecolor='black', alpha=0.7)
        axes[0, 0].set_title('Total Cost Distribution')
        axes[0, 0].set_xlabel('Total Cost ($)')
        axes[0, 0].set_ylabel('Frequency')
        axes[0, 0].axvline(self.results['total_cost'].mean(),
                          color='r', linestyle='--', label='Mean')
        axes[0, 0].legend()

        # Service level distribution
        axes[0, 1].hist(self.results['service_level'], bins=50,
                       edgecolor='black', alpha=0.7, color='green')
        axes[0, 1].set_title('Service Level Distribution')
        axes[0, 1].set_xlabel('Service Level (%)')
        axes[0, 1].set_ylabel('Frequency')

        # Shortage distribution
        axes[1, 0].hist(self.results['shortage'], bins=50,
                       edgecolor='black', alpha=0.7, color='orange')
        axes[1, 0].set_title('Shortage Distribution')
        axes[1, 0].set_xlabel('Shortage Units')
        axes[1, 0].set_ylabel('Frequency')

        # Cost components
        cost_components = self.results[[
            'transport_cost', 'holding_cost', 'shortage_cost'
        ]].mean()
        axes[1, 1].bar(cost_components.index, cost_components.values)
        axes[1, 1].set_title('Average Cost Components')
        axes[1, 1].set_ylabel('Cost ($)')
        axes[1, 1].tick_params(axis='x', rotation=45)

        plt.tight_layout()
        return fig

# Example usage
mc = MonteCarloScenarioAnalysis(n_simulations=10000)

demand_params = {
    'mean': 10000,
    'std': 2000,
    'distribution': 'normal'
}

cost_params = {
    'transport': {'min': 2.0, 'max': 4.0},
    'holding': {'mean': 1.5, 'std': 0.3}
}

supply_params = {
    'reliability': {'alpha': 9, 'beta': 1}  # Beta distribution
}

results = mc.simulate_supply_chain(
    demand_params, cost_params, supply_params, periods=12
)

# Analyze
analysis = mc.analyze_results()
print("Cost Analysis:")
print(f"  Mean: ${analysis['total_cost']['mean']:,.0f}")
print(f"  Std Dev: ${analysis['total_cost']['std']:,.0f}")
print(f"  P95: ${analysis['total_cost']['p95']:,.0f}")

var_metrics = mc.value_at_risk(0.95)
print(f"\nValue at Risk (95%): ${var_metrics['VaR']:,.0f}")
print(f"Conditional VaR: ${var_metrics['CVaR']:,.0f}")

# Plot
mc.plot_distributions()

---

敏感性分析

单变量敏感性分析

测试单个变量的影响：

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

def sensitivity_analysis(base_model, variable_name,
                        test_range, other_params):
    """
    Perform one-way sensitivity analysis

    Parameters:
    - base_model: function that returns metric given parameters
    - variable_name: name of variable to test
    - test_range: array of values to test
    - other_params: dict of fixed parameters
    """

    results = []

    for value in test_range:
        # Update parameter
        params = other_params.copy()
        params[variable_name] = value

        # Evaluate model
        metric = base_model(**params)

        results.append({
            variable_name: value,
            'metric': metric
        })

    return pd.DataFrame(results)

# Example: Test demand sensitivity
def calculate_cost(demand, unit_cost, holding_rate, capacity):
    """Simple cost model"""
    production = min(demand, capacity)
    shortage = max(0, demand - capacity)

    production_cost = production * unit_cost
    holding_cost = production * holding_rate * 0.5
    shortage_cost = shortage * unit_cost * 3  # 3x penalty

    return production_cost + holding_cost + shortage_cost

# Test demand scenarios
demand_range = np.linspace(5000, 15000, 20)

other_params = {
    'unit_cost': 10,
    'holding_rate': 0.2,
    'capacity': 10000
}

demand_sensitivity = sensitivity_analysis(
    calculate_cost,
    'demand',
    demand_range,
    other_params
)

# Plot
plt.figure(figsize=(10, 6))
plt.plot(demand_sensitivity['demand'],
         demand_sensitivity['metric'],
         marker='o', linewidth=2)
plt.xlabel('Demand (units)')
plt.ylabel('Total Cost ($)')
plt.title('Sensitivity to Demand Changes')
plt.grid(True, alpha=0.3)
plt.axvline(x=10000, color='r', linestyle='--', label='Capacity Limit')
plt.legend()
plt.show()

多变量敏感性分析（龙卷风图）

比较多个变量的影响：

def tornado_analysis(base_model, variables, ranges, base_params):
    """
    Create tornado diagram showing sensitivity to multiple variables

    Parameters:
    - base_model: function to evaluate
    - variables: list of variable names
    - ranges: dict {variable: (low, high)}
    - base_params: base case parameters
    """

    # Calculate base case
    base_result = base_model(**base_params)

    results = []

    for var in variables:
        # Test low value
        params_low = base_params.copy()
        params_low[var] = ranges[var][0]
        result_low = base_model(**params_low)

        # Test high value
        params_high = base_params.copy()
        params_high[var] = ranges[var][1]
        result_high = base_model(**params_high)

        # Calculate swing
        swing = abs(result_high - result_low)

        results.append({
            'variable': var,
            'low_value': ranges[var][0],
            'high_value': ranges[var][1],
            'result_low': result_low,
            'result_high': result_high,
            'swing': swing,
            'impact_pct': (swing / base_result) * 100
        })

    df = pd.DataFrame(results)
    df = df.sort_values('swing', ascending=True)

    # Plot tornado diagram
    fig, ax = plt.subplots(figsize=(10, 8))

    y_pos = np.arange(len(df))

    for i, row in df.iterrows():
        low = row['result_low']
        high = row['result_high']

        ax.barh(row['variable'], high - base_result,
                left=base_result, color='red', alpha=0.6)
        ax.barh(row['variable'], base_result - low,
                left=low, color='blue', alpha=0.6)

    ax.axvline(x=base_result, color='black', linestyle='--', linewidth=2)
    ax.set_xlabel('Total Cost ($)')
    ax.set_title('Tornado Diagram - Sensitivity Analysis')
    ax.grid(True, alpha=0.3, axis='x')

    return df, fig

# Example usage
variables = ['demand', 'unit_cost', 'holding_rate']

ranges = {
    'demand': (8000, 12000),
    'unit_cost': (8, 12),
    'holding_rate': (0.15, 0.25)
}

base_params = {
    'demand': 10000,
    'unit_cost': 10,
    'holding_rate': 0.2,
    'capacity': 10000
}

tornado_df, fig = tornado_analysis(
    calculate_cost, variables, ranges, base_params
)

---

风险评估与量化

风险概率-影响矩阵

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

class RiskAssessment:
    """Supply chain risk assessment framework"""

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

    def add_risk(self, name, category, probability, impact,
                detection_difficulty, mitigation_cost):
        """
        Add a risk to the register

        Parameters:
        - probability: 1-5 scale (1=very low, 5=very high)
        - impact: 1-5 scale (1=minimal, 5=catastrophic)
        - detection_difficulty: 1-5 (1=easy to detect, 5=hard)
        - mitigation_cost: 1-5 (1=low cost, 5=high cost)
        """

        risk = {
            'name': name,
            'category': category,
            'probability': probability,
            'impact': impact,
            'detection': detection_difficulty,
            'mitigation_cost': mitigation_cost,
            'risk_score': probability * impact,
            'risk_priority_number': probability * impact * detection_difficulty
        }

        self.risks.append(risk)

    def get_risk_register(self):
        """Return risk register as DataFrame"""
        df = pd.DataFrame(self.risks)
        df = df.sort_values('risk_score', ascending=False)
        return df

    def plot_risk_matrix(self):
        """Plot probability-impact matrix"""

        df = pd.DataFrame(self.risks)

        fig, ax = plt.subplots(figsize=(12, 10))

        # Color by risk score
        scatter = ax.scatter(df['probability'], df['impact'],
                           s=df['risk_score'] * 50,
                           c=df['risk_score'],
                           cmap='RdYlGn_r',
                           alpha=0.6,
                           edgecolors='black',
                           linewidth=1.5)

        # Annotate risks
        for idx, row in df.iterrows():
            ax.annotate(row['name'],
                       (row['probability'], row['impact']),
                       fontsize=8,
                       xytext=(5, 5),
                       textcoords='offset points')

        ax.set_xlabel('Probability (1-5)', fontsize=12)
        ax.set_ylabel('Impact (1-5)', fontsize=12)
        ax.set_title('Supply Chain Risk Matrix', fontsize=14, fontweight='bold')
        ax.grid(True, alpha=0.3)

        # Add risk zones
        ax.axhline(y=3, color='orange', linestyle='--', alpha=0.3, linewidth=2)
        ax.axvline(x=3, color='orange', linestyle='--', alpha=0.3, linewidth=2)

        # Add text labels for zones
        ax.text(1.2, 4.5, 'Low Prob\nHigh Impact', fontsize=9, alpha=0.5)
        ax.text(4.2, 4.5, 'HIGH RISK', fontsize=11, fontweight='bold',
               color='red', alpha=0.7)
        ax.text(4.2, 1.5, 'High Prob\nLow Impact', fontsize=9, alpha=0.5)
        ax.text(1.2, 1.5, 'LOW RISK', fontsize=10, alpha=0.5)

        plt.colorbar(scatter, label='Risk Score')

        return fig

    def prioritize_risks(self, min_score=10):
        """Return high-priority risks"""
        df = self.get_risk_register()
        return df[df['risk_score'] >= min_score]

# Example usage
risk_assessment = RiskAssessment()

# Add supply chain risks
risk_assessment.add_risk(
    name='Single Source Supplier',
    category='Supply',
    probability=3,
    impact=5,
    detection_difficulty=2,
    mitigation_cost=4
)

risk_assessment.add_risk(
    name='Port Congestion',
    category='Logistics',
    probability=4,
    impact=3,
    detection_difficulty=1,
    mitigation_cost=2
)

risk_assessment.add_risk(
    name='Demand Forecast Error',
    category='Demand',
    probability=5,
    impact=3,
    detection_difficulty=2,
    mitigation_cost=3
)

risk_assessment.add_risk(
    name='Natural Disaster',
    category='External',
    probability=2,
    impact=5,
    detection_difficulty=1,
    mitigation_cost=5
)

risk_assessment.add_risk(
    name='Cyber Attack',
    category='Technology',
    probability=3,
    impact=4,
    detection_difficulty=4,
    mitigation_cost=4
)

risk_assessment.add_risk(
    name='Quality Defect',
    category='Operations',
    probability=4,
    impact=4,
    detection_difficulty=3,
    mitigation_cost=3
)

# Get risk register
risk_register = risk_assessment.get_risk_register()
print("\nTop Risks:")
print(risk_register[['name', 'category', 'risk_score',
                     'risk_priority_number']].head(10))

# Plot
risk_assessment.plot_risk_matrix()

# High priority risks
high_risks = risk_assessment.prioritize_risks(min_score=12)
print(f"\nHigh Priority Risks (score >= 12): {len(high_risks)}")

---

战略情景制定

Shell 情景规划方法

适用于长期战略规划：

步骤 1：定义核心问题

• “未来 5 年，哪种供应链战略最具韧性？”

步骤 2：识别关键力量

• 既定因素（趋势、人口特征）
• 关键不确定因素（法规、技术采用）

步骤 3：选择情景维度

• 选择 2 个最重要／最不确定的驱动因素
• 创建 2x2 矩阵

步骤 4：编写情景叙述

• 为每个象限编写引人入胜的故事
• 用描述性名称为情景命名
• 详细说明其影响

步骤 5：识别影响与策略

• 哪些策略在所有情景中都有效？（稳健策略）
• 哪些策略取决于具体情景？（应变策略）
• 有哪些预警指标？

示例：全球贸易情景

维度：

• 经济一体化程度（高／低）
• 环境法规（严格／宽松）

四种情景：

1. “全球效率”（高一体化、宽松法规）

   • 优化全球网络
   • 最低成本生产
   • 长供应链
   • 区域化程度有限

2. “绿色全球化”（高一体化、严格法规）

   • 碳中和物流
   • 优先采用可持续采购
   • 投资绿色技术
   • 成本较高，合规性强

3. “区域韧性”（低一体化、严格法规）

   • 近岸外包／友岸外包
   • 区域供应链
   • 强调可靠性
   • 成本适中

4. “碎片化世界”（低一体化、宽松法规）

   • 贸易壁垒
   • 针对各国制定策略
   • 基础设施重复建设
   • 效率低下

---

应急规划

业务连续性计划

关键组成部分：

1. 风险识别

   • 可能出现哪些问题？
   • 发生可能性与影响

2. 影响分析

   • 对收入的影响
   • 对客户的影响
   • 恢复时间目标（RTO）
   • 恢复点目标（RPO）

3. 应对策略

   • 即时行动（0-24 小时）
   • 短期（1-7 天）
   • 中期（1-4 周）
   • 长期恢复

4. 资源需求

   • 人员、系统、设施
   • 备用供应商
   • 库存缓冲
   • 财务储备

应急计划模板：

from dataclasses import dataclass
from typing import List, Dict
from datetime import datetime

@dataclass
class ContingencyAction:
    """Single action in contingency plan"""
    action_id: str
    description: str
    trigger: str
    responsible_party: str
    timeline: str  # "Immediate", "24 hours", "Week 1", etc.
    resources_needed: List[str]
    cost_estimate: float
    dependencies: List[str]

@dataclass
class ContingencyPlan:
    """Complete contingency plan for a risk scenario"""
    risk_name: str
    risk_category: str
    trigger_conditions: List[str]
    severity_level: str  # "Low", "Medium", "High", "Critical"
    actions: List[ContingencyAction]
    communication_plan: Dict
    escalation_procedure: List[str]

    def get_immediate_actions(self):
        """Return actions needed immediately"""
        return [a for a in self.actions if a.timeline == "Immediate"]

    def get_actions_by_timeline(self, timeline: str):
        """Get actions for specific timeline"""
        return [a for a in self.actions if a.timeline == timeline]

    def estimate_total_cost(self):
        """Calculate total cost of plan execution"""
        return sum(a.cost_estimate for a in self.actions)

# Example: Supplier Disruption Contingency Plan
supplier_disruption_plan = ContingencyPlan(
    risk_name="Primary Supplier Disruption",
    risk_category="Supply Risk",
    trigger_conditions=[
        "Supplier bankruptcy filing",
        "Natural disaster at supplier location",
        "Quality issues requiring supplier shutdown",
        "Supply < 50% of normal for > 3 days"
    ],
    severity_level="High",
    actions=[
        ContingencyAction(
            action_id="SD-001",
            description="Activate backup supplier contracts",
            trigger="Confirmed disruption > 24 hours",
            responsible_party="Procurement Director",
            timeline="Immediate",
            resources_needed=["Backup supplier contact list", "Expedited PO"],
            cost_estimate=50000,
            dependencies=[]
        ),
        ContingencyAction(
            action_id="SD-002",
            description="Increase safety stock from alternative sources",
            trigger="Expected disruption > 1 week",
            responsible_party="Supply Chain Manager",
            timeline="24 hours",
            resources_needed=["Emergency budget authorization", "Warehouse space"],
            cost_estimate=200000,
            dependencies=["SD-001"]
        ),
        ContingencyAction(
            action_id="SD-003",
            description="Implement product rationing to priority customers",
            trigger="Inventory below critical level",
            responsible_party="Customer Service VP",
            timeline="48 hours",
            resources_needed=["Customer priority list", "Communication templates"],
            cost_estimate=0,
            dependencies=["SD-002"]
        ),
        ContingencyAction(
            action_id="SD-004",
            description="Qualify and onboard new supplier",
            trigger="Expected disruption > 30 days",
            responsible_party="Procurement Director",
            timeline="Week 2-4",
            resources_needed=["Supplier qualification team", "Quality testing"],
            cost_estimate=150000,
            dependencies=["SD-001", "SD-002"]
        )
    ],
    communication_plan={
        "internal": ["Executive team notification within 2 hours",
                    "Daily status updates to stakeholders"],
        "external": ["Customer notification within 24 hours if impact expected",
                    "Supplier engagement for resolution timeline"],
        "escalation": ["CFO if cost > $500K", "CEO if duration > 4 weeks"]
    },
    escalation_procedure=[
        "Level 1: Supply Chain Manager (0-24 hours)",
        "Level 2: VP Operations (24-72 hours)",
        "Level 3: COO (> 72 hours or impact > $1M)",
        "Level 4: CEO (Critical customer impact or > $5M)"
    ]
)

# Display immediate actions
print("IMMEDIATE ACTIONS:")
for action in supplier_disruption_plan.get_immediate_actions():
    print(f"  [{action.action_id}] {action.description}")
    print(f"      Owner: {action.responsible_party}")
    print(f"      Cost: ${action.cost_estimate:,}")

print(f"\nTotal Plan Cost: ${supplier_disruption_plan.estimate_total_cost():,}")

---

工具与库

Python 库

仿真与建模：

• simpy：离散事件仿真
• scipy.stats：统计分布
• numpy：随机数生成、数组
• pandas：数据处理与分析

不确定性下的优化：

• pyomo：随机规划
• pulp：基于情景的线性规划
• mip：混合整数规划

可视化：

• matplotlib、seaborn：统计图表
• plotly：交互式仪表板
• networkx：网络图

风险分析：

• risktools：风险指标与 VaR
• copulas：依赖关系建模

商业软件

情景规划平台：

• Kinaxis RapidResponse：情景分析与协作
• o9 Solutions：AI 驱动的情景规划
• LLamasoft：供应链情景建模
• Blue Yonder：假设分析

风险管理：

• Resilinc：供应链风险情报
• Everstream Analytics：预测性风险分析
• Riskmethods：风险监控与评估

仿真：

• AnyLogic：多方法仿真
• Arena：离散事件仿真
• Simio：3D 仿真建模
• FlexSim：3D 仿真

---

常见挑战与解决方案

挑战：情景过多

问题：

• 分析瘫痪
• 组合数量过多，令人不知所措
• 无法做出决策

解决方案：

• 聚焦于 3-5 个关键情景
• 使用情景缩减技术
• 将相似情景归组
• 按概率 × 影响进行优先级排序
• 从基准／最佳／最差情景入手

挑战：数据不足

问题：

• 缺少历史中断数据
• 难以估算概率
• 对影响存在不确定性

解决方案：

• 采用专家判断（德尔菲法）
• 参考行业基准和案例研究
• 从区间估算入手，而非单点估算
• 通过敏感性分析了解影响因素
• 随着事件发生不断学习和更新

挑战：情景偏差

问题：

• 锚定当前状态
• 忽视低概率、高影响事件
• 情景选择中的确认偏误

解决方案：

• 纳入多元视角
• 使用结构化方法（Shell 方法）
• 明确质疑各项假设
• 纳入「黑天鹅」情景
• 由独立人员主持

挑战：可操作性

问题：

• 情景无法引导决策
• 分析无法转化为战略
• 缺少明确的后续步骤

解决方案：

• 将情景与具体决策关联
• 找出「无悔」举措（适用于所有情景）
• 确定触发点和预警信号
• 为每种情景制定应急计划
• 建立定期审查和更新流程

挑战：复杂性与易用性之间的平衡

问题：

• 模型对利益相关者而言过于复杂
• 无法解释结果
• 仿真过程如同黑盒

解决方案：

• 从简单模型入手，并按需增加复杂性
• 使用可视化仪表板和摘要
• 清晰记录各项假设
• 通过敏感性分析展示关键影响因素
• 提供情景叙述，而不只是数字

---

输出格式

情景分析报告

执行摘要：

• 分析目的和范围
• 评估的关键情景
• 重要发现和建议
• 对决策的影响

情景定义：

情景名称	描述	关键假设	概率
基准情景	最可能出现的未来	当前趋势延续	50%
高增长	需求激增	经济繁荣，开拓新市场	20%
供应中断	主要供应商失效	单一来源风险成为现实	15%
完美风暴	多重问题	需求激增 + 供应短缺	10%
成本通胀	成本上涨	燃料、人工、材料上涨 30%	25%

情景结果：

情景	总成本	服务水平	库存	关键指标
基准情景	$25M	95%	8,000 件	基线
高增长	$32M	87%	12,000 件	产能受限
供应中断	$38M	78%	5,000 件	缺货成本高
完美风暴	$45M	65%	7,000 件	危机状态
成本通胀	$31M	93%	8,000 件	利润率承压

风险评估：

• 确定高优先级风险
• 风险缓解策略
• 针对主要情景的应急计划
• 预警指标

建议：

1. 立即行动（0-3 个月）

   • 为排名前 10 的组件认证备用供应商
   • 增加关键物料的安全库存
   • 实施需求感知以便及早预警

2. 短期举措（3-12 个月）

   • 实现供应商基础多元化
   • 增加灵活的产能选项
   • 提升情景规划能力

3. 战略投资（1-3 年）

   • 建设区域配送中心
   • 实施控制塔技术
   • 制定双重采购策略

监控与审查：

• 需要跟踪的关键绩效指标
• 情景审查频率（每季度）
• 计划启动触发点
• 治理与升级流程

---

要提出的问题

如果需要更多背景信息：

1. 此情景分析支持哪项决策？
2. 哪些不确定性或风险最令人担忧？
3. 规划周期是多长？（战术还是战略）
4. 有哪些关于历史波动和中断的数据？
5. 是否有需要建模的具体事件或情景？
6. 需要定量分析还是定性情景？
7. 谁会使用此分析，以及如何使用？
8. 是否已有风险管理或业务连续性计划？

---

相关技能

• demand-forecasting：用于需求不确定性分析
• sales-operations-planning：用于 S&OP 情景整合
• capacity-planning：用于产能情景和压力测试
• network-design：用于网络策略情景
• inventory-optimization：用于库存缓冲策略
• risk-mitigation：用于风险应对规划
• supply-chain-analytics：用于绩效监控
• digital-twin-modeling：用于实时情景模拟