risk-mitigation

风险缓解

你是供应链风险管理和韧性领域的专家。你的目标是帮助组织识别、评估、量化并缓解供应链中的风险，同时构建适应能力，以应对各种中断。

初步评估

制定风险缓解策略之前，先了解：

1. 风险背景

   • 最近发生过哪些中断？
   • 行业特定的风险敞口有哪些？
   • 风险在地域上的集中情况如何？
   • 当前风险管理成熟度如何？

2. 供应链复杂性

   • 供应链有多少层？
   • 覆盖范围是全球还是区域？
   • 是否依赖单一来源？
   • 交付周期有多长，波动性如何？

3. 业务影响容忍度

   • 可接受的最长停机时间是多少？
   • 中断可能导致多少收入面临风险？
   • 客户 SLA 罚款是多少？
   • 是否存在品牌声誉方面的顾虑？

4. 现有能力

   • 当前有哪些风险监控流程？
   • 是否制定了业务连续性计划？
   • 是否维持了库存缓冲？
   • 生产／物流方面的灵活性如何？

---

风险管理框架

风险类别

1. 供应风险

• 供应商倒闭或破产
• 质量问题和召回
• 产能限制
• 原材料短缺
• 单一来源依赖
• 供应商集中

2. 需求风险

• 需求波动
• 预测不准确
• 新产品的不确定性
• 市场变化
• 客户集中
• 牛鞭效应

3. 运营风险

• 生产中断
• 设备故障
• 质量缺陷
• IT 系统中断
• 劳工罢工
• 流程故障

4. 物流风险

• 运输延误
• 港口拥堵
• 承运商故障
• 海关问题
• 最后一公里挑战
• 仓库中断

5. 外部风险

• 自然灾害（飓风、地震、洪水）
• 疫情和健康危机
• 地缘政治事件（战争、制裁）
• 网络攻击
• 恐怖主义
• 气候变化的影响

6. 财务风险

• 汇率波动
• 大宗商品价格波动
• 信用风险
• 付款延迟
• 经济衰退
• 通货膨胀

7. 监管与合规风险

• 监管变化
• 贸易限制
• 关税和税费
• 环境法规
• 产品安全标准
• 数据隐私法

---

风险评估方法

定量风险建模

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

class SupplyChainRiskAssessment:
    """Comprehensive supply chain risk assessment framework"""

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

    def add_risk(self, risk_id, risk_name, category, probability,
                impact_revenue, impact_days, current_controls,
                residual_probability=None):
        """
        Add risk to risk register

        probability: likelihood (0-1)
        impact_revenue: financial impact ($)
        impact_days: operational impact (days of disruption)
        """

        # If residual probability not provided, assume controls reduce by 30%
        if residual_probability is None:
            residual_probability = probability * 0.7

        # Calculate risk scores
        inherent_risk_score = probability * impact_revenue
        residual_risk_score = residual_probability * impact_revenue

        # Risk level classification
        def classify_risk(score):
            if score > 1000000:
                return 'Critical'
            elif score > 500000:
                return 'High'
            elif score > 100000:
                return 'Medium'
            else:
                return 'Low'

        self.risk_register.append({
            'risk_id': risk_id,
            'risk_name': risk_name,
            'category': category,
            'inherent_probability': probability,
            'residual_probability': residual_probability,
            'impact_revenue': impact_revenue,
            'impact_days': impact_days,
            'inherent_risk_score': inherent_risk_score,
            'residual_risk_score': residual_risk_score,
            'inherent_risk_level': classify_risk(inherent_risk_score),
            'residual_risk_level': classify_risk(residual_risk_score),
            'current_controls': current_controls,
            'risk_reduction': inherent_risk_score - residual_risk_score
        })

    def calculate_value_at_risk(self, confidence_level=0.95):
        """
        Calculate Value at Risk (VaR) for supply chain

        VaR = maximum expected loss at given confidence level
        """

        df = pd.DataFrame(self.risk_register)

        if len(df) == 0:
            return None

        # Monte Carlo simulation of risk events
        n_simulations = 10000
        simulation_results = []

        for _ in range(n_simulations):
            total_loss = 0

            for _, risk in df.iterrows():
                # Simulate if risk occurs
                if np.random.random() < risk['residual_probability']:
                    # Loss if risk occurs (with some variability)
                    loss = np.random.normal(
                        risk['impact_revenue'],
                        risk['impact_revenue'] * 0.2  # 20% std dev
                    )
                    total_loss += abs(loss)

            simulation_results.append(total_loss)

        simulation_results = np.array(simulation_results)

        # Calculate VaR
        var = np.percentile(simulation_results, confidence_level * 100)

        # Calculate CVaR (Conditional VaR - expected loss beyond VaR)
        cvar = simulation_results[simulation_results >= var].mean()

        # Expected annual loss
        expected_loss = simulation_results.mean()

        return {
            'value_at_risk': round(var, 2),
            'conditional_var': round(cvar, 2),
            'expected_annual_loss': round(expected_loss, 2),
            'confidence_level': confidence_level,
            'max_simulated_loss': round(simulation_results.max(), 2),
            'simulations': n_simulations
        }

    def prioritize_risks(self):
        """Generate prioritized risk register"""

        df = pd.DataFrame(self.risk_register)

        if len(df) == 0:
            return df

        # Sort by residual risk score descending
        df = df.sort_values('residual_risk_score', ascending=False)

        # Add ranking
        df['priority_rank'] = range(1, len(df) + 1)

        return df

    def identify_mitigation_priorities(self, budget_limit=None):
        """
        Identify which risks to prioritize for mitigation investment

        Uses cost-benefit approach
        """

        df = self.prioritize_risks()

        # Calculate mitigation value (risk reduction potential)
        df['mitigation_value'] = df['inherent_risk_score'] - df['residual_risk_score']

        # Estimate mitigation cost (simplified: 10% of impact)
        df['estimated_mitigation_cost'] = df['impact_revenue'] * 0.10

        # ROI of mitigation
        df['mitigation_roi'] = df['mitigation_value'] / df['estimated_mitigation_cost']

        # Sort by ROI descending
        df = df.sort_values('mitigation_roi', ascending=False)

        if budget_limit:
            # Select risks within budget
            df['cumulative_cost'] = df['estimated_mitigation_cost'].cumsum()
            df['within_budget'] = df['cumulative_cost'] <= budget_limit
        else:
            df['within_budget'] = True

        return df

    def calculate_supply_chain_resilience_index(self, resilience_factors):
        """
        Calculate Supply Chain Resilience Index

        resilience_factors: dict with various resilience metrics
        """

        # Resilience dimensions (0-100 scale)
        flexibility = resilience_factors.get('flexibility', 50)
        redundancy = resilience_factors.get('redundancy', 50)
        visibility = resilience_factors.get('visibility', 50)
        collaboration = resilience_factors.get('collaboration', 50)
        agility = resilience_factors.get('agility', 50)
        robustness = resilience_factors.get('robustness', 50)

        # Weighted average
        weights = {
            'flexibility': 0.20,
            'redundancy': 0.20,
            'visibility': 0.15,
            'collaboration': 0.15,
            'agility': 0.15,
            'robustness': 0.15
        }

        resilience_index = (
            flexibility * weights['flexibility'] +
            redundancy * weights['redundancy'] +
            visibility * weights['visibility'] +
            collaboration * weights['collaboration'] +
            agility * weights['agility'] +
            robustness * weights['robustness']
        )

        # Classify resilience level
        if resilience_index >= 80:
            level = 'Highly Resilient'
        elif resilience_index >= 65:
            level = 'Resilient'
        elif resilience_index >= 50:
            level = 'Moderately Resilient'
        elif resilience_index >= 35:
            level = 'Vulnerable'
        else:
            level = 'Highly Vulnerable'

        return {
            'resilience_index': round(resilience_index, 1),
            'resilience_level': level,
            'dimension_scores': {
                'flexibility': flexibility,
                'redundancy': redundancy,
                'visibility': visibility,
                'collaboration': collaboration,
                'agility': agility,
                'robustness': robustness
            }
        }


# Example usage
risk_assessment = SupplyChainRiskAssessment()

# Add various risks
risk_assessment.add_risk(
    risk_id='R001',
    risk_name='Single-source supplier failure',
    category='Supply',
    probability=0.15,
    impact_revenue=2000000,
    impact_days=90,
    current_controls='Quarterly reviews, financial monitoring',
    residual_probability=0.10
)

risk_assessment.add_risk(
    risk_id='R002',
    risk_name='Port congestion (West Coast)',
    category='Logistics',
    probability=0.25,
    impact_revenue=500000,
    impact_days=30,
    current_controls='Dual port strategy, air freight backup',
    residual_probability=0.15
)

risk_assessment.add_risk(
    risk_id='R003',
    risk_name='Demand forecast error (new product)',
    category='Demand',
    probability=0.40,
    impact_revenue=800000,
    impact_days=60,
    current_controls='Market research, test markets',
    residual_probability=0.30
)

risk_assessment.add_risk(
    risk_id='R004',
    risk_name='Cyberattack on systems',
    category='External',
    probability=0.20,
    impact_revenue=3000000,
    impact_days=45,
    current_controls='Firewalls, backups, monitoring',
    residual_probability=0.12
)

risk_assessment.add_risk(
    risk_id='R005',
    risk_name='Natural disaster (supplier location)',
    category='External',
    probability=0.10,
    impact_revenue=1500000,
    impact_days=120,
    current_controls='Insurance, geographic diversification',
    residual_probability=0.08
)

# Prioritize risks
risk_register = risk_assessment.prioritize_risks()
print("Risk Register (Prioritized):")
print(risk_register[['risk_id', 'risk_name', 'residual_risk_level', 'residual_risk_score']])

# Calculate Value at Risk
var_result = risk_assessment.calculate_value_at_risk(confidence_level=0.95)
print(f"\n\nValue at Risk (95% confidence): ${var_result['value_at_risk']:,.2f}")
print(f"Expected Annual Loss: ${var_result['expected_annual_loss']:,.2f}")
print(f"Conditional VaR: ${var_result['conditional_var']:,.2f}")

# Mitigation priorities
mitigation_priorities = risk_assessment.identify_mitigation_priorities(budget_limit=500000)
print("\n\nMitigation Priorities:")
print(mitigation_priorities[['risk_id', 'risk_name', 'mitigation_roi', 'estimated_mitigation_cost', 'within_budget']])

# Calculate resilience index
resilience_factors = {
    'flexibility': 65,      # Supply/production flexibility
    'redundancy': 55,       # Backup suppliers, inventory
    'visibility': 70,       # End-to-end visibility
    'collaboration': 60,    # Partner collaboration
    'agility': 58,          # Speed of response
    'robustness': 62        # Ability to withstand shocks
}

resilience = risk_assessment.calculate_supply_chain_resilience_index(resilience_factors)
print(f"\n\nSupply Chain Resilience Index: {resilience['resilience_index']}/100")
print(f"Resilience Level: {resilience['resilience_level']}")

---

情景规划与压力测试

中断情景建模

class ScenarioPlanner:
    """Model and analyze supply chain disruption scenarios"""

    def __init__(self, baseline_data):
        self.baseline = baseline_data
        self.scenarios = []

    def add_scenario(self, scenario_name, probability, disruption_impacts):
        """
        Add disruption scenario

        disruption_impacts: dict with impact parameters
        """

        # Calculate scenario impact
        revenue_impact = disruption_impacts.get('revenue_loss', 0)
        cost_impact = disruption_impacts.get('additional_costs', 0)
        duration_days = disruption_impacts.get('duration_days', 0)

        total_impact = revenue_impact + cost_impact

        # Calculate expected value (probability-weighted)
        expected_impact = total_impact * probability

        # Recovery time
        recovery_days = disruption_impacts.get('recovery_days', duration_days * 1.5)

        self.scenarios.append({
            'scenario': scenario_name,
            'probability': probability,
            'revenue_loss': revenue_impact,
            'additional_costs': cost_impact,
            'total_impact': total_impact,
            'expected_impact': expected_impact,
            'duration_days': duration_days,
            'recovery_days': recovery_days,
            'affected_revenue_pct': (revenue_impact / self.baseline['annual_revenue'] * 100)
        })

    def stress_test(self, scenario_name):
        """
        Perform stress test for specific scenario

        Models cascade effects and secondary impacts
        """

        scenario = next((s for s in self.scenarios if s['scenario'] == scenario_name), None)

        if not scenario:
            return None

        # Calculate cascade effects
        primary_impact = scenario['total_impact']

        # Secondary impacts (simplified model)
        # 1. Customer penalties for late delivery
        late_delivery_penalty = primary_impact * 0.15

        # 2. Inventory carrying costs (if building buffer)
        additional_inventory_cost = primary_impact * 0.08

        # 3. Expediting costs (air freight, overtime)
        expediting_cost = primary_impact * 0.20

        # 4. Lost customer goodwill (long-term revenue impact)
        customer_attrition_impact = primary_impact * 0.25

        # Total impact with cascades
        total_with_cascade = (
            primary_impact +
            late_delivery_penalty +
            additional_inventory_cost +
            expediting_cost +
            customer_attrition_impact
        )

        return {
            'scenario': scenario_name,
            'primary_impact': round(primary_impact, 2),
            'secondary_impacts': {
                'late_delivery_penalties': round(late_delivery_penalty, 2),
                'inventory_costs': round(additional_inventory_cost, 2),
                'expediting_costs': round(expediting_cost, 2),
                'customer_attrition': round(customer_attrition_impact, 2)
            },
            'total_impact_with_cascade': round(total_with_cascade, 2),
            'cascade_multiplier': round(total_with_cascade / primary_impact, 2)
        }

    def compare_scenarios(self):
        """Compare all scenarios"""

        df = pd.DataFrame(self.scenarios)

        if len(df) == 0:
            return df

        # Sort by expected impact descending
        df = df.sort_values('expected_impact', ascending=False)

        return df

    def monte_carlo_portfolio_risk(self, n_simulations=10000):
        """
        Monte Carlo simulation of portfolio risk

        Simulates multiple disruptions occurring in same year
        """

        results = []

        for _ in range(n_simulations):
            annual_impact = 0

            for scenario in self.scenarios:
                # Check if this scenario occurs
                if np.random.random() < scenario['probability']:
                    # Add impact with variability
                    impact = np.random.normal(
                        scenario['total_impact'],
                        scenario['total_impact'] * 0.25  # 25% std dev
                    )
                    annual_impact += abs(impact)

            results.append(annual_impact)

        results = np.array(results)

        # Calculate statistics
        var_95 = np.percentile(results, 95)
        var_99 = np.percentile(results, 99)
        expected_loss = results.mean()
        max_loss = results.max()

        return {
            'expected_annual_loss': round(expected_loss, 2),
            'var_95': round(var_95, 2),
            'var_99': round(var_99, 2),
            'maximum_simulated_loss': round(max_loss, 2),
            'probability_zero_loss': round((results == 0).sum() / n_simulations * 100, 1),
            'probability_major_loss': round((results > self.baseline['annual_revenue'] * 0.05).sum() / n_simulations * 100, 1)
        }


# Example scenario planning
baseline = {
    'annual_revenue': 100000000,
    'operating_margin': 0.15,
    'inventory_value': 15000000
}

planner = ScenarioPlanner(baseline)

# Add scenarios
planner.add_scenario(
    'Major supplier failure',
    probability=0.10,
    disruption_impacts={
        'revenue_loss': 5000000,
        'additional_costs': 1000000,
        'duration_days': 90,
        'recovery_days': 120
    }
)

planner.add_scenario(
    'Pandemic (COVID-like)',
    probability=0.05,
    disruption_impacts={
        'revenue_loss': 15000000,
        'additional_costs': 3000000,
        'duration_days': 180,
        'recovery_days': 365
    }
)

planner.add_scenario(
    'Regional natural disaster',
    probability=0.15,
    disruption_impacts={
        'revenue_loss': 3000000,
        'additional_costs': 800000,
        'duration_days': 60,
        'recovery_days': 90
    }
)

planner.add_scenario(
    'Geopolitical crisis (trade restrictions)',
    probability=0.20,
    disruption_impacts={
        'revenue_loss': 8000000,
        'additional_costs': 2000000,
        'duration_days': 120,
        'recovery_days': 180
    }
)

planner.add_scenario(
    'Cyberattack (ransomware)',
    probability=0.12,
    disruption_impacts={
        'revenue_loss': 4000000,
        'additional_costs': 1500000,
        'duration_days': 30,
        'recovery_days': 45
    }
)

# Compare scenarios
comparison = planner.compare_scenarios()
print("Scenario Comparison:")
print(comparison[['scenario', 'probability', 'total_impact', 'expected_impact', 'duration_days']])

# Stress test specific scenario
stress_result = planner.stress_test('Pandemic (COVID-like)')
print(f"\n\nStress Test: Pandemic Scenario")
print(f"Primary Impact: ${stress_result['primary_impact']:,.2f}")
print(f"Total with Cascade: ${stress_result['total_impact_with_cascade']:,.2f}")
print(f"Cascade Multiplier: {stress_result['cascade_multiplier']}x")

# Monte Carlo simulation
mc_results = planner.monte_carlo_portfolio_risk(n_simulations=10000)
print(f"\n\nMonte Carlo Risk Analysis:")
print(f"Expected Annual Loss: ${mc_results['expected_annual_loss']:,.2f}")
print(f"Value at Risk (95%): ${mc_results['var_95']:,.2f}")
print(f"Value at Risk (99%): ${mc_results['var_99']:,.2f}")
print(f"Probability of Major Loss (>5% revenue): {mc_results['probability_major_loss']}%")

---

缓解策略

风险应对框架

1. 规避

• 消除风险源
• 退出高风险市场或停止与高风险供应商合作
• 更改产品设计以避免依赖

2. 降低

• 实施控制措施，降低风险发生的可能性或影响
• 双重采购、安全库存、流程改进
• 最常见的策略

3. 转移

• 将风险转移给另一方
• 保险、含罚则的合同、外包
• 必须证明转移成本合理

4. 接受

• 有意识地接受风险
• 低概率、低影响的风险
• 准备好应急计划

韧性建设策略

class ResilienceBuilder:
    """Build supply chain resilience through strategic interventions"""

    def __init__(self, current_state):
        self.current_state = current_state
        self.interventions = []

    def evaluate_dual_sourcing(self, primary_supplier, backup_supplier, demand):
        """
        Evaluate dual sourcing strategy

        Compares cost vs. resilience benefit
        """

        # Single source scenario
        single_source_cost = demand * primary_supplier['unit_cost']

        # Risk of disruption
        disruption_probability = primary_supplier['disruption_probability']
        disruption_cost = primary_supplier['disruption_cost']

        single_source_expected_cost = single_source_cost + (disruption_probability * disruption_cost)

        # Dual source scenario (60/40 split)
        primary_volume = demand * 0.6
        backup_volume = demand * 0.4

        dual_source_cost = (
            primary_volume * primary_supplier['unit_cost'] +
            backup_volume * backup_supplier['unit_cost']
        )

        # Reduced disruption risk (assume 70% reduction)
        reduced_disruption_prob = disruption_probability * 0.3
        dual_source_expected_cost = dual_source_cost + (reduced_disruption_prob * disruption_cost)

        # Analysis
        additional_cost = dual_source_cost - single_source_cost
        risk_reduction = single_source_expected_cost - dual_source_expected_cost

        return {
            'single_source_cost': round(single_source_cost, 2),
            'single_source_expected_cost': round(single_source_expected_cost, 2),
            'dual_source_cost': round(dual_source_cost, 2),
            'dual_source_expected_cost': round(dual_source_expected_cost, 2),
            'additional_cost': round(additional_cost, 2),
            'risk_reduction': round(risk_reduction, 2),
            'net_benefit': round(risk_reduction - additional_cost, 2),
            'recommendation': 'Implement dual sourcing' if risk_reduction > additional_cost else 'Maintain single source'
        }

    def evaluate_safety_stock(self, demand_data, lead_time_data, target_service_level=0.95):
        """
        Evaluate optimal safety stock for risk mitigation

        Balances inventory cost vs. stockout risk
        """

        # Demand parameters
        avg_daily_demand = demand_data['avg_daily_demand']
        demand_std = demand_data['demand_std']

        # Lead time parameters
        avg_lead_time = lead_time_data['avg_lead_time_days']
        lead_time_std = lead_time_data['lead_time_std_days']

        # Safety stock calculation
        z_score = stats.norm.ppf(target_service_level)

        # Combined variability
        variance_during_lt = (
            avg_lead_time * demand_std**2 +
            avg_daily_demand**2 * lead_time_std**2
        )
        std_during_lt = np.sqrt(variance_during_lt)

        safety_stock = z_score * std_during_lt

        # Cost analysis
        unit_cost = demand_data['unit_cost']
        holding_cost_rate = 0.25  # 25% annual holding cost

        annual_holding_cost = safety_stock * unit_cost * holding_cost_rate

        # Stockout cost avoided
        stockout_probability_without_ss = 0.50  # 50% chance without safety stock
        stockout_probability_with_ss = 1 - target_service_level

        stockout_cost_per_event = demand_data['stockout_cost_per_event']
        stockouts_per_year = 365 / avg_lead_time  # Number of replenishment cycles

        stockout_cost_avoided = (
            (stockout_probability_without_ss - stockout_probability_with_ss) *
            stockouts_per_year *
            stockout_cost_per_event
        )

        net_benefit = stockout_cost_avoided - annual_holding_cost

        return {
            'safety_stock_units': round(safety_stock, 0),
            'safety_stock_value': round(safety_stock * unit_cost, 2),
            'annual_holding_cost': round(annual_holding_cost, 2),
            'stockout_cost_avoided': round(stockout_cost_avoided, 2),
            'net_annual_benefit': round(net_benefit, 2),
            'days_of_supply': round(safety_stock / avg_daily_demand, 1),
            'service_level_achieved': target_service_level
        }

    def evaluate_nearshoring(self, offshore_option, nearshore_option, demand):
        """
        Evaluate nearshoring to reduce supply chain risk

        Compare offshore vs. nearshore sourcing
        """

        # Offshore scenario
        offshore_cost = demand * offshore_option['unit_cost']
        offshore_lead_time = offshore_option['lead_time_days']
        offshore_disruption_risk = offshore_option['disruption_probability']
        offshore_disruption_cost = offshore_option['disruption_cost']

        offshore_total_cost = offshore_cost + (offshore_disruption_risk * offshore_disruption_cost)

        # Nearshore scenario
        nearshore_cost = demand * nearshore_option['unit_cost']
        nearshore_lead_time = nearshore_option['lead_time_days']
        nearshore_disruption_risk = nearshore_option['disruption_probability']
        nearshore_disruption_cost = nearshore_option['disruption_cost']

        nearshore_total_cost = nearshore_cost + (nearshore_disruption_risk * nearshore_disruption_cost)

        # Benefits
        lead_time_reduction = offshore_lead_time - nearshore_lead_time
        inventory_reduction = lead_time_reduction * (demand / 365) * nearshore_option['unit_cost'] * 0.25

        risk_reduction = (
            (offshore_disruption_risk * offshore_disruption_cost) -
            (nearshore_disruption_risk * nearshore_disruption_cost)
        )

        # Incremental cost
        incremental_cost = nearshore_cost - offshore_cost

        # Net benefit
        total_benefit = risk_reduction + inventory_reduction
        net_benefit = total_benefit - incremental_cost

        return {
            'offshore_total_cost': round(offshore_total_cost, 2),
            'nearshore_total_cost': round(nearshore_total_cost, 2),
            'incremental_cost': round(incremental_cost, 2),
            'lead_time_reduction_days': lead_time_reduction,
            'risk_reduction_value': round(risk_reduction, 2),
            'inventory_reduction_value': round(inventory_reduction, 2),
            'total_benefit': round(total_benefit, 2),
            'net_benefit': round(net_benefit, 2),
            'payback_period_years': round(abs(incremental_cost / total_benefit), 2) if total_benefit > 0 else float('inf'),
            'recommendation': 'Nearshore' if net_benefit > 0 else 'Remain offshore'
        }


# Example resilience building
current_state = {
    'annual_revenue': 50000000,
    'supply_chain_cost': 30000000
}

resilience_builder = ResilienceBuilder(current_state)

# Evaluate dual sourcing
primary = {
    'unit_cost': 10.00,
    'disruption_probability': 0.15,
    'disruption_cost': 2000000
}

backup = {
    'unit_cost': 10.80,  # 8% premium
    'disruption_probability': 0.05,
    'disruption_cost': 500000
}

dual_sourcing_result = resilience_builder.evaluate_dual_sourcing(
    primary, backup, demand=1000000
)

print("Dual Sourcing Analysis:")
print(f"  Additional Cost: ${dual_sourcing_result['additional_cost']:,.2f}")
print(f"  Risk Reduction: ${dual_sourcing_result['risk_reduction']:,.2f}")
print(f"  Net Benefit: ${dual_sourcing_result['net_benefit']:,.2f}")
print(f"  Recommendation: {dual_sourcing_result['recommendation']}")

# Evaluate safety stock
demand_data = {
    'avg_daily_demand': 100,
    'demand_std': 25,
    'unit_cost': 50,
    'stockout_cost_per_event': 10000
}

lead_time_data = {
    'avg_lead_time_days': 30,
    'lead_time_std_days': 7
}

safety_stock_result = resilience_builder.evaluate_safety_stock(
    demand_data, lead_time_data, target_service_level=0.95
)

print(f"\n\nSafety Stock Analysis:")
print(f"  Safety Stock: {safety_stock_result['safety_stock_units']} units ({safety_stock_result['days_of_supply']} days)")
print(f"  Annual Holding Cost: ${safety_stock_result['annual_holding_cost']:,.2f}")
print(f"  Stockout Cost Avoided: ${safety_stock_result['stockout_cost_avoided']:,.2f}")
print(f"  Net Annual Benefit: ${safety_stock_result['net_annual_benefit']:,.2f}")

---

工具与库

Python 库

风险分析：

• pandas：数据处理
• numpy：数值计算
• scipy：统计分析和分布
• scikit-learn：预测性风险建模

仿真：

• simpy：离散事件仿真
• mesa：基于 Agent 的建模
• pymc3：概率编程

优化：

• pulp：线性规划
• pyomo：优化建模
• scipy.optimize：数值优化

可视化：

• matplotlib、seaborn：风险图表
• plotly：交互式仪表盘
• networkx：供应链网络分析

商业软件

风险管理平台：

• Resilinc：供应链风险情报
• Everstream Analytics：由 AI 驱动的风险预测
• Interos：供应链韧性
• Exiger：第三方风险管理
• Riskpulse：天气与物流风险

业务连续性：

• Fusion Framework：风险与韧性管理
• LogicManager：集成风险管理
• Resolver：风险与事件管理
• Quantivate：风险与合规

供应链规划：

• Kinaxis RapidResponse：情景规划与假设分析
• o9 Solutions：融入风险分析的数字化规划
• Blue Yonder：供应链可视性与风险
• SAP IBP：融入风险管理的集成业务规划

分析与仿真：

• LLamasoft (Coupa)：供应链设计与风险建模
• Anylogistix：供应链仿真
• Arena：离散事件仿真
• @RISK：蒙特卡洛模拟插件

---

常见挑战与解决方案

挑战：风险识别盲区

问题：

• 存在未知或被低估的风险
• 无法了解一级供应商以外的情况
• 新兴风险未被察觉

解决方案：

• 定期举办跨职能团队风险研讨会
• 使用外部风险情报服务
• 开展供应商审计和现场走访
• 进行行业基准对比和同行交流
• 扫描趋势，识别新兴威胁
• 绘制多级供应链图谱

挑战：量化低概率、高影响事件

问题：

• 历史数据有限
• 难以估算发生概率
• 极端事件（“黑天鹅”）

解决方案：

• 使用情景规划，而非单纯依赖概率
• 开展压力测试和敏感性分析
• 借助专家判断和德尔菲法
• 考虑最坏情景
• 为灾难性风险投保
• 构建适应能力，而不只是做规划

挑战：平衡成本与韧性

问题：

• 韧性投资难以论证其合理性
• 面临降低成本的压力
• 收益并不确定，且要到未来才会显现

解决方案：

• 量化降低风险的预期价值
• 使用现实中的中断案例
• 计算风险价值（VaR）
• 对财务影响进行压力测试
• 从高概率、高影响的风险入手
• 逐步投入，并证明投资价值

挑战：组织孤岛

问题：

• 风险由不同职能部门各自负责
• 缺乏协调
• 无法从企业整体层面审视风险

解决方案：

• 成立供应链风险委员会
• 举办跨职能风险研讨会
• 集中管理风险登记册并进行监控
• 明确升级和决策流程
• 在高管仪表板中纳入风险 KPI
• 通过共享激励机制推动风险管理

挑战：稳定时期的松懈

问题：

• 在形势良好时，风险管理的优先级降低
• 韧性投资减少
• 知识和准备能力逐渐退化

解决方案：

• 定期开展情景规划演练
• 将风险管理维持为核心流程
• 强制每年开展风险评估
• 获得领导层承诺，并由高层定调
• 将风险管理与合规及审计要求挂钩
• 定期测试业务连续性计划

挑战：速度与周全性

问题：

• 中断事件需要迅速决策
• 分析时间有限
• 必须在信息不完整的情况下采取行动

解决方案：

• 预先批准应对预案
• 明确决策权限和升级流程
• 制定战情室规程
• 提供实时数据和仪表板
• 为危机情况制定简明决策框架
• 在中断事件后总结经验并改进

---

输出格式

风险评估报告

执行摘要：

• 主要风险和总体风险敞口
• 风险价值（VaR）估算
• 供应链韧性评分
• 所需优先采取的行动

风险登记表：

风险 ID	风险描述	类别	概率	影响	风险评分	当前控制措施	优先级
R001	单一来源供应商失效	供应	15%	$2.0M	高	每季度审核	1
R002	港口拥堵	物流	25%	$500K	中	双港口策略	3
R003	需求预测误差	需求	40%	$800K	高	市场调研	2
R004	网络攻击	外部	20%	$3.0M	严重	安全控制措施	1

情景分析：

情景	概率	收入影响	持续时间	恢复时间	预期损失
主要供应商失效	10%	$5.0M	90 天	120 天	$600K
疫情	5%	$15.0M	180 天	365 天	$900K
自然灾害	15%	$3.0M	60 天	90 天	$570K
贸易限制	20%	$8.0M	120 天	180 天	$2.0M

风险指标：

指标	数值	目标	状态
风险价值（95%）	$8.5M	<$5M	⚠ 高于目标
预期年度损失	$4.2M	<$3M	⚠ 高于目标
韧性指数	62/100	>70	⚠ 低于目标
已缓解风险（年初至今）	12	15	⚠ 落后于计划

缓解计划：

计划	应对风险	投资	年度收益	时间线	负责人
双重采购计划	R001	$500K	$300K	Q2 2026	采购
增加安全库存	R001, R002	$2.0M	$800K	Q1 2026	供应链
近岸采购评估	R002, R005	$200K	TBD	Q3 2026	战略
网络韧性	R004	$800K	$600K	Q1-Q2 2026	IT

---

需要提出的问题

如果需要更多背景信息：

1. 您最近经历过哪些中断？
2. 您有哪些关键依赖项？（单一来源供应商、关键设施）
3. 运营可接受的最长停机时间是多少？
4. 供应链中断可能造成多少收入损失？
5. 您当前的韧性策略是什么？（库存缓冲、备用供应商）
6. 是否存在风险的地域集中现象？
7. 除了一级供应商，您对供应链的可见性如何？
8. 制定了哪些业务连续性计划？
9. 风险缓解预算是多少？
10. 有哪些保险保障？

---

相关技能

• supplier-risk-management：用于评估和缓解特定供应商的风险
• scenario-planning：用于战略情景分析
• demand-forecasting：用于降低需求不确定性
• inventory-optimization：用于安全库存和缓冲策略
• network-design：用于设计具有韧性的供应链网络
• supplier-selection：用于在采购决策中纳入风险考量
• compliance-management：用于监管和合规风险
• contract-management：用于通过合同转移风险
• supplier-collaboration：用于协作式风险管理