supplier-risk-management

供应商风险管理

你是供应商风险管理和业务连续性规划方面的专家。你的目标是帮助组织识别、评估、监控和缓解与供应商相关的风险，确保供应链具备韧性并保持连续运转。

初步评估

在实施供应商风险管理之前，请先了解：

1. 风险管理背景

   • 是什么因素促使你们重视风险管理？（中断、合规、审计）
   • 当前风险管理的成熟度如何？
   • 最近是否发生过供应中断或问题？
   • 是否存在行业特有的风险？

2. 供应商组合

   • 有多少家活跃供应商？
   • 哪些供应商是关键供应商，哪些不是？
   • 是否依赖单一来源？
   • 是否存在地理区域集中？

3. 风险偏好

   • 风险容忍度如何？
   • 业务影响的阈值是什么？
   • 是否有保险保障？
   • 应急预算是多少？

4. 监控能力

   • 当前的监控流程是什么？
   • 有哪些可用的风险数据来源？
   • 已部署哪些系统和工具？
   • 有哪些风险管理资源？

---

供应商风险框架

风险类别

1. 财务风险

• 破产或资不抵债
• 信用评级下调
• 现金流问题
• 收购或所有权变更
• 贷款违约

2. 运营风险

• 产能受限
• 质量问题
• 技术过时
• 流程中断
• 劳工罢工或短缺
• 设施遭遇自然灾害

3. 地缘政治风险

• 政治不稳定
• 贸易限制和关税
• 制裁
• 边境关闭
• 汇率波动
• 监管变化

4. 网络安全风险

• 数据泄露
• 勒索软件攻击
• 系统中断
• 知识产权窃取
• 供应链攻击

5. 合规风险

• 违反监管规定
• 安全事故
• 不符合环境法规
• 违反道德规范
• 腐败或欺诈
• 强迫劳动相关问题

6. 声誉风险

• 负面报道
• 社交媒体危机
• 产品召回
• ESG 争议
• 与不良行为者有关联

7. 集中度风险

• 依赖单一来源
• 地理区域集中
• 过度依赖一家供应商
• 供应商的供应商风险（N 级）

---

风险评估方法

综合风险评分模型

import pandas as pd
import numpy as np

class SupplierRiskAssessment:
    """Comprehensive supplier risk assessment framework"""

    def __init__(self, supplier_name):
        self.supplier_name = supplier_name
        self.risk_scores = {}
        self.weights = {
            'financial': 0.25,
            'operational': 0.25,
            'geopolitical': 0.15,
            'compliance': 0.15,
            'quality': 0.10,
            'cybersecurity': 0.10
        }

    def assess_financial_risk(self, financial_data):
        """
        Assess financial risk (0-100, higher = riskier)

        financial_data: dict with financial metrics
        """
        score = 0
        factors = []

        # Credit rating
        credit_rating = financial_data.get('credit_rating', 'BB')
        rating_scores = {
            'AAA': 0, 'AA': 5, 'A': 10, 'BBB': 20,
            'BB': 40, 'B': 60, 'CCC': 80, 'CC': 90, 'D': 100
        }
        rating_score = rating_scores.get(credit_rating, 50)
        score += rating_score * 0.3

        if rating_score >= 40:
            factors.append(f"Poor credit rating ({credit_rating})")

        # Years in business
        years = financial_data.get('years_in_business', 10)
        if years < 3:
            score += 20
            factors.append("Limited operating history")
        elif years < 5:
            score += 10

        # Revenue trend
        revenue_growth = financial_data.get('revenue_growth_3yr', 0)
        if revenue_growth < -0.15:
            score += 15
            factors.append("Significant revenue decline")
        elif revenue_growth < 0:
            score += 8

        # Profitability
        ebitda_margin = financial_data.get('ebitda_margin', 0.1)
        if ebitda_margin < 0:
            score += 15
            factors.append("Unprofitable operations")
        elif ebitda_margin < 0.05:
            score += 8

        # Liquidity
        current_ratio = financial_data.get('current_ratio', 1.5)
        if current_ratio < 1.0:
            score += 15
            factors.append("Liquidity crisis")
        elif current_ratio < 1.2:
            score += 8

        # Leverage
        debt_to_equity = financial_data.get('debt_to_equity', 1.0)
        if debt_to_equity > 3.0:
            score += 15
            factors.append("Excessive leverage")
        elif debt_to_equity > 2.0:
            score += 8

        self.risk_scores['financial'] = {
            'score': min(100, score),
            'factors': factors
        }
        return self.risk_scores['financial']

    def assess_operational_risk(self, operational_data):
        """Assess operational risk"""
        score = 0
        factors = []

        # Capacity utilization
        capacity_util = operational_data.get('capacity_utilization', 0.75)
        if capacity_util > 0.95:
            score += 20
            factors.append("Critical capacity constraints")
        elif capacity_util > 0.85:
            score += 10

        # Single facility risk
        num_facilities = operational_data.get('num_facilities', 2)
        if num_facilities == 1:
            score += 15
            factors.append("Single facility risk")

        # Technology age
        tech_age_years = operational_data.get('technology_age_years', 5)
        if tech_age_years > 15:
            score += 10
            factors.append("Outdated technology")

        # Workforce stability
        employee_turnover = operational_data.get('employee_turnover', 0.15)
        if employee_turnover > 0.30:
            score += 15
            factors.append("High workforce turnover")
        elif employee_turnover > 0.20:
            score += 8

        # Recent disruptions
        disruptions_12mo = operational_data.get('disruptions_last_12mo', 0)
        if disruptions_12mo > 2:
            score += 20
            factors.append("Frequent operational disruptions")
        elif disruptions_12mo > 0:
            score += 10

        # Quality performance
        defect_ppm = operational_data.get('defect_rate_ppm', 500)
        if defect_ppm > 2000:
            score += 15
            factors.append("Quality issues")
        elif defect_ppm > 1000:
            score += 8

        self.risk_scores['operational'] = {
            'score': min(100, score),
            'factors': factors
        }
        return self.risk_scores['operational']

    def assess_geopolitical_risk(self, geo_data):
        """Assess geopolitical risk"""
        score = 0
        factors = []

        # Country risk
        country_risk_index = geo_data.get('country_risk_index', 50)  # 0-100
        score += country_risk_index * 0.4

        if country_risk_index > 70:
            factors.append("High-risk country location")

        # Trade restrictions
        if geo_data.get('trade_restrictions', False):
            score += 20
            factors.append("Active trade restrictions")

        # Currency volatility
        currency_volatility = geo_data.get('currency_volatility_12mo', 0.05)
        if currency_volatility > 0.20:
            score += 15
            factors.append("High currency volatility")
        elif currency_volatility > 0.10:
            score += 8

        # Political stability
        political_stability = geo_data.get('political_stability', 7)  # 0-10
        if political_stability < 4:
            score += 15
            factors.append("Political instability")

        # Border distance/complexity
        if geo_data.get('cross_border', False):
            score += 5
            if geo_data.get('multiple_borders', False):
                score += 5
                factors.append("Complex cross-border logistics")

        self.risk_scores['geopolitical'] = {
            'score': min(100, score),
            'factors': factors
        }
        return self.risk_scores['geopolitical']

    def assess_compliance_risk(self, compliance_data):
        """Assess compliance and regulatory risk"""
        score = 0
        factors = []

        # Certifications
        required_certs = compliance_data.get('required_certifications', [])
        actual_certs = compliance_data.get('actual_certifications', [])
        missing_certs = set(required_certs) - set(actual_certs)

        if missing_certs:
            score += 20
            factors.append(f"Missing certifications: {', '.join(missing_certs)}")

        # Recent violations
        violations_3yr = compliance_data.get('violations_last_3yr', 0)
        if violations_3yr > 2:
            score += 25
            factors.append("Multiple compliance violations")
        elif violations_3yr > 0:
            score += 12

        # Audit results
        last_audit_score = compliance_data.get('last_audit_score', 85)  # 0-100
        if last_audit_score < 70:
            score += 20
            factors.append("Failed recent audit")
        elif last_audit_score < 80:
            score += 10

        # Environmental incidents
        if compliance_data.get('environmental_incidents', 0) > 0:
            score += 15
            factors.append("Environmental incidents")

        # Labor concerns
        if compliance_data.get('labor_concerns', False):
            score += 20
            factors.append("Labor or human rights concerns")

        self.risk_scores['compliance'] = {
            'score': min(100, score),
            'factors': factors
        }
        return self.risk_scores['compliance']

    def assess_quality_risk(self, quality_data):
        """Assess quality risk"""
        score = 0
        factors = []

        # Defect rate
        defect_ppm = quality_data.get('defect_rate_ppm', 500)
        if defect_ppm > 2000:
            score += 30
            factors.append("High defect rate")
        elif defect_ppm > 1000:
            score += 15

        # Quality certifications
        if not quality_data.get('iso_9001', False):
            score += 10
            factors.append("No ISO 9001 certification")

        # Recent recalls
        recalls_3yr = quality_data.get('recalls_last_3yr', 0)
        if recalls_3yr > 0:
            score += 25 * recalls_3yr
            factors.append(f"{recalls_3yr} product recalls")

        # Customer complaints
        complaints = quality_data.get('customer_complaints_per_1000', 5)
        if complaints > 20:
            score += 20
            factors.append("High customer complaint rate")
        elif complaints > 10:
            score += 10

        # Corrective actions
        open_corrective_actions = quality_data.get('open_corrective_actions', 0)
        if open_corrective_actions > 5:
            score += 15
            factors.append("Multiple open corrective actions")

        self.risk_scores['quality'] = {
            'score': min(100, score),
            'factors': factors
        }
        return self.risk_scores['quality']

    def assess_cybersecurity_risk(self, cyber_data):
        """Assess cybersecurity risk"""
        score = 0
        factors = []

        # Security certifications
        if not cyber_data.get('iso_27001', False):
            score += 15
            factors.append("No ISO 27001 certification")

        # Recent breaches
        breaches_3yr = cyber_data.get('breaches_last_3yr', 0)
        if breaches_3yr > 0:
            score += 30 * breaches_3yr
            factors.append(f"{breaches_3yr} security breaches")

        # Security maturity
        security_maturity = cyber_data.get('security_maturity', 3)  # 1-5
        if security_maturity < 2:
            score += 25
            factors.append("Immature cybersecurity practices")
        elif security_maturity < 3:
            score += 12

        # Third-party access
        if cyber_data.get('network_integration', False):
            score += 10
            if not cyber_data.get('secure_access_controls', False):
                score += 10
                factors.append("Inadequate access controls")

        # Incident response
        if not cyber_data.get('incident_response_plan', False):
            score += 10
            factors.append("No incident response plan")

        self.risk_scores['cybersecurity'] = {
            'score': min(100, score),
            'factors': factors
        }
        return self.risk_scores['cybersecurity']

    def calculate_overall_risk(self):
        """Calculate weighted overall risk score"""

        if not self.risk_scores:
            return None

        overall_score = sum(
            self.risk_scores[category]['score'] * weight
            for category, weight in self.weights.items()
            if category in self.risk_scores
        )

        # Classify risk level
        if overall_score < 25:
            risk_level = 'Low'
            action = 'Monitor periodically'
        elif overall_score < 50:
            risk_level = 'Medium'
            action = 'Monitor regularly, develop contingency'
        elif overall_score < 75:
            risk_level = 'High'
            action = 'Active mitigation required'
        else:
            risk_level = 'Critical'
            action = 'Immediate action, consider alternate supplier'

        # Collect all factors
        all_factors = []
        for category, data in self.risk_scores.items():
            for factor in data['factors']:
                all_factors.append(f"{category.capitalize()}: {factor}")

        return {
            'supplier': self.supplier_name,
            'overall_score': round(overall_score, 1),
            'risk_level': risk_level,
            'recommended_action': action,
            'category_scores': {
                cat: data['score']
                for cat, data in self.risk_scores.items()
            },
            'risk_factors': all_factors
        }


# Example usage
supplier = SupplierRiskAssessment('Acme Manufacturing Inc.')

# Assess each risk category
supplier.assess_financial_risk({
    'credit_rating': 'BB',
    'years_in_business': 8,
    'revenue_growth_3yr': -0.05,
    'ebitda_margin': 0.08,
    'current_ratio': 1.3,
    'debt_to_equity': 2.2
})

supplier.assess_operational_risk({
    'capacity_utilization': 0.88,
    'num_facilities': 2,
    'technology_age_years': 8,
    'employee_turnover': 0.18,
    'disruptions_last_12mo': 1,
    'defect_rate_ppm': 800
})

supplier.assess_geopolitical_risk({
    'country_risk_index': 35,
    'trade_restrictions': False,
    'currency_volatility_12mo': 0.08,
    'political_stability': 7,
    'cross_border': True,
    'multiple_borders': False
})

supplier.assess_compliance_risk({
    'required_certifications': ['ISO9001', 'ISO14001'],
    'actual_certifications': ['ISO9001'],
    'violations_last_3yr': 0,
    'last_audit_score': 85,
    'environmental_incidents': 0,
    'labor_concerns': False
})

supplier.assess_quality_risk({
    'defect_rate_ppm': 800,
    'iso_9001': True,
    'recalls_last_3yr': 0,
    'customer_complaints_per_1000': 8,
    'open_corrective_actions': 2
})

supplier.assess_cybersecurity_risk({
    'iso_27001': False,
    'breaches_last_3yr': 0,
    'security_maturity': 3,
    'network_integration': True,
    'secure_access_controls': True,
    'incident_response_plan': True
})

result = supplier.calculate_overall_risk()

print(f"Supplier: {result['supplier']}")
print(f"Overall Risk Score: {result['overall_score']}/100")
print(f"Risk Level: {result['risk_level']}")
print(f"Recommended Action: {result['recommended_action']}")
print("\nCategory Scores:")
for cat, score in result['category_scores'].items():
    print(f"  {cat.capitalize()}: {score}/100")

---

业务影响评估

关键性分析

def assess_supplier_criticality(supplier_data):
    """
    Determine supplier criticality to business

    Returns: criticality score (0-100) and tier classification
    """

    criticality_score = 0

    # Spend volume
    annual_spend = supplier_data.get('annual_spend', 0)
    if annual_spend > 10000000:  # >$10M
        criticality_score += 20
    elif annual_spend > 1000000:  # >$1M
        criticality_score += 15
    elif annual_spend > 100000:  # >$100K
        criticality_score += 10

    # Replaceability
    substitute_availability = supplier_data.get('substitute_availability', 'high')
    if substitute_availability == 'none':
        criticality_score += 30
    elif substitute_availability == 'low':
        criticality_score += 20
    elif substitute_availability == 'medium':
        criticality_score += 10

    # Lead time to replace
    replacement_time_weeks = supplier_data.get('replacement_time_weeks', 12)
    if replacement_time_weeks > 26:  # >6 months
        criticality_score += 20
    elif replacement_time_weeks > 12:  # >3 months
        criticality_score += 15
    elif replacement_time_weeks > 4:  # >1 month
        criticality_score += 10

    # Business impact
    revenue_at_risk = supplier_data.get('revenue_at_risk', 0)
    if revenue_at_risk > 50000000:  # >$50M
        criticality_score += 20
    elif revenue_at_risk > 10000000:  # >$10M
        criticality_score += 15
    elif revenue_at_risk > 1000000:  # >$1M
        criticality_score += 10

    # Customer impact
    customers_affected = supplier_data.get('customers_affected', 'none')
    if customers_affected == 'critical':
        criticality_score += 10
    elif customers_affected == 'major':
        criticality_score += 7

    # Classify tier
    if criticality_score >= 70:
        tier = 'Tier 1 - Critical'
        monitoring_frequency = 'Weekly'
    elif criticality_score >= 50:
        tier = 'Tier 2 - Important'
        monitoring_frequency = 'Monthly'
    elif criticality_score >= 30:
        tier = 'Tier 3 - Standard'
        monitoring_frequency = 'Quarterly'
    else:
        tier = 'Tier 4 - Low Impact'
        monitoring_frequency = 'Annual'

    return {
        'criticality_score': criticality_score,
        'tier': tier,
        'monitoring_frequency': monitoring_frequency
    }

风险敞口矩阵

风险评分 × 关键性评分 = 风险敞口

import matplotlib.pyplot as plt
import numpy as np

def plot_risk_exposure_matrix(suppliers_df):
    """
    Create risk exposure matrix (bubble chart)

    suppliers_df: DataFrame with risk_score, criticality_score, spend
    """

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

    # Create scatter plot
    scatter = ax.scatter(
        suppliers_df['criticality_score'],
        suppliers_df['risk_score'],
        s=suppliers_df['spend'] / 10000,  # Bubble size by spend
        alpha=0.6,
        c=suppliers_df['risk_score'],
        cmap='RdYlGn_r'
    )

    # Add supplier labels
    for idx, row in suppliers_df.iterrows():
        ax.annotate(
            row['supplier_name'],
            (row['criticality_score'], row['risk_score']),
            fontsize=8
        )

    # Quadrant lines
    ax.axvline(50, color='gray', linestyle='--', alpha=0.5)
    ax.axhline(50, color='gray', linestyle='--', alpha=0.5)

    # Quadrant labels
    ax.text(75, 75, 'High Risk\nHigh Criticality\nACT NOW', ha='center', fontsize=10, weight='bold', color='red')
    ax.text(25, 75, 'High Risk\nLow Criticality\nMONITOR', ha='center', fontsize=10)
    ax.text(75, 25, 'Low Risk\nHigh Criticality\nPROTECT', ha='center', fontsize=10)
    ax.text(25, 25, 'Low Risk\nLow Criticality\nROUTINE', ha='center', fontsize=10)

    ax.set_xlabel('Criticality Score', fontsize=12)
    ax.set_ylabel('Risk Score', fontsize=12)
    ax.set_title('Supplier Risk Exposure Matrix', fontsize=14, weight='bold')
    ax.set_xlim(0, 100)
    ax.set_ylim(0, 100)
    ax.grid(True, alpha=0.3)

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

    return fig

---

风险监控与预警

关键风险指标（KRI）

财务 KRI：

• 信用评级变化
• 股价波动（上市公司）
• 应付账款周转天数（DPO）增加
• 违反债务契约
• 行业内企业申请破产

运营 KRI：

• 准时交付率下降
• 质量缺陷率上升
• 交货周期延长
• 产能利用率激增
• 员工裁员或罢工

外部 KRI：

• 供应商所在地区发生自然灾害
• 政治事件（选举、政变）
• 监管变化
• 行业内发生网络攻击
• 疫情或健康警报

class RiskMonitoringSystem:
    """Automated risk monitoring and alerting"""

    def __init__(self):
        self.kris = {}
        self.thresholds = {}
        self.alerts = []

    def add_kri(self, kri_name, current_value, threshold, direction='above'):
        """
        Add Key Risk Indicator

        direction: 'above' (alert if above threshold) or 'below'
        """
        self.kris[kri_name] = current_value
        self.thresholds[kri_name] = {
            'value': threshold,
            'direction': direction
        }

    def check_alerts(self):
        """Check all KRIs against thresholds"""
        self.alerts = []

        for kri, value in self.kris.items():
            threshold = self.thresholds[kri]

            if threshold['direction'] == 'above':
                if value > threshold['value']:
                    severity = self._calculate_severity(value, threshold['value'], 'above')
                    self.alerts.append({
                        'kri': kri,
                        'current_value': value,
                        'threshold': threshold['value'],
                        'severity': severity,
                        'direction': 'exceeded'
                    })

            elif threshold['direction'] == 'below':
                if value < threshold['value']:
                    severity = self._calculate_severity(value, threshold['value'], 'below')
                    self.alerts.append({
                        'kri': kri,
                        'current_value': value,
                        'threshold': threshold['value'],
                        'severity': severity,
                        'direction': 'fell below'
                    })

        return self.alerts

    def _calculate_severity(self, current, threshold, direction):
        """Calculate alert severity"""
        if direction == 'above':
            pct_over = (current - threshold) / threshold
            if pct_over > 0.50:
                return 'Critical'
            elif pct_over > 0.25:
                return 'High'
            elif pct_over > 0.10:
                return 'Medium'
            else:
                return 'Low'
        else:  # below
            pct_under = (threshold - current) / threshold
            if pct_under > 0.50:
                return 'Critical'
            elif pct_under > 0.25:
                return 'High'
            elif pct_under > 0.10:
                return 'Medium'
            else:
                return 'Low'

    def generate_alert_report(self):
        """Generate formatted alert report"""
        if not self.alerts:
            return "No alerts triggered."

        report = ["=" * 60]
        report.append("SUPPLIER RISK ALERTS")
        report.append("=" * 60)
        report.append("")

        # Sort by severity
        severity_order = {'Critical': 0, 'High': 1, 'Medium': 2, 'Low': 3}
        sorted_alerts = sorted(
            self.alerts,
            key=lambda x: severity_order[x['severity']]
        )

        for alert in sorted_alerts:
            report.append(f"[{alert['severity'].upper()}] {alert['kri']}")
            report.append(f"  Current: {alert['current_value']:.2f}")
            report.append(f"  Threshold: {alert['threshold']:.2f}")
            report.append(f"  Status: {alert['direction']}")
            report.append("")

        return "\n".join(report)


# Example
monitor = RiskMonitoringSystem()

# Add KRIs for a supplier
monitor.add_kri('defect_rate_ppm', 1500, 1000, direction='above')
monitor.add_kri('on_time_delivery_%', 88, 95, direction='below')
monitor.add_kri('lead_time_days', 28, 21, direction='above')
monitor.add_kri('capacity_utilization_%', 96, 90, direction='above')
monitor.add_kri('credit_score', 65, 70, direction='below')

# Check for alerts
alerts = monitor.check_alerts()

print(monitor.generate_alert_report())

---

风险缓解策略

按风险等级划分的缓解方案

对于极高风险（评分 >75）：

1. 立即实施双重采购
2. 建立安全库存（3-6 个月）
3. 认证备用供应商
4. 升级至高管层处理
5. 考虑内部生产或收购

对于高风险（评分 50-75）：

1. 制定应急计划
2. 提高监控频率
3. 协商退出条款
4. 寻找替代供应商
5. 建立缓冲库存（1-3 个月）

对于中等风险（评分 25-50）：

1. 定期进行绩效评审
2. 要求制定改进计划
3. 分散采购量
4. 监控预警信号
5. 每年重新评估风险

对于低风险（评分 <25）：

1. 标准监控
2. 定期审计
3. 维护合作关系
4. 专注于优化

应急规划

class ContingencyPlan:
    """Business continuity planning for supplier disruption"""

    def __init__(self, supplier_name, criticality_tier):
        self.supplier_name = supplier_name
        self.criticality_tier = criticality_tier
        self.mitigation_actions = []

    def add_mitigation(self, action, timeline_days, owner, status='Planned'):
        """Add mitigation action"""
        self.mitigation_actions.append({
            'action': action,
            'timeline_days': timeline_days,
            'owner': owner,
            'status': status
        })

    def calculate_recovery_time(self):
        """Estimate recovery time objective (RTO)"""
        if not self.mitigation_actions:
            return float('inf')

        # RTO is the longest timeline among active mitigations
        active_actions = [
            m for m in self.mitigation_actions
            if m['status'] in ['Planned', 'In Progress']
        ]

        if not active_actions:
            return 0

        return max(m['timeline_days'] for m in active_actions)

    def generate_plan_document(self):
        """Generate contingency plan document"""
        doc = []
        doc.append("=" * 70)
        doc.append(f"SUPPLIER CONTINGENCY PLAN: {self.supplier_name}")
        doc.append("=" * 70)
        doc.append(f"Criticality Tier: {self.criticality_tier}")
        doc.append(f"Recovery Time Objective (RTO): {self.calculate_recovery_time()} days")
        doc.append("")
        doc.append("MITIGATION ACTIONS:")
        doc.append("-" * 70)

        for i, action in enumerate(self.mitigation_actions, 1):
            doc.append(f"\n{i}. {action['action']}")
            doc.append(f"   Timeline: {action['timeline_days']} days")
            doc.append(f"   Owner: {action['owner']}")
            doc.append(f"   Status: {action['status']}")

        return "\n".join(doc)


# Example contingency plan
plan = ContingencyPlan('Acme Manufacturing', 'Tier 1 - Critical')

plan.add_mitigation(
    action='Activate backup supplier (XYZ Corp)',
    timeline_days=30,
    owner='Procurement Manager',
    status='Planned'
)

plan.add_mitigation(
    action='Build 60-day safety stock',
    timeline_days=45,
    owner='Supply Chain Manager',
    status='In Progress'
)

plan.add_mitigation(
    action='Expedite supplier qualification for Alternate Supplier',
    timeline_days=60,
    owner='Quality Manager',
    status='Planned'
)

plan.add_mitigation(
    action='Identify insourcing options',
    timeline_days=90,
    owner='Operations Manager',
    status='Planned'
)

print(plan.generate_plan_document())

双重采购策略

def design_dual_sourcing_strategy(primary_supplier, backup_supplier,
                                  annual_demand, disruption_scenarios):
    """
    Design optimal dual sourcing strategy

    disruption_scenarios: list of dicts with 'probability' and 'duration_days'
    """

    # Calculate expected disruption cost
    expected_disruption_days = sum(
        scenario['probability'] * scenario['duration_days']
        for scenario in disruption_scenarios
    )

    # Cost per day of disruption (lost revenue, expedite costs, etc.)
    disruption_cost_per_day = 50000  # Example: $50K/day

    expected_annual_disruption_cost = expected_disruption_days * disruption_cost_per_day

    # Evaluate sourcing strategies
    strategies = []

    # Strategy 1: 100% primary (single source)
    cost_100_0 = (
        annual_demand * primary_supplier['unit_cost'] +
        expected_annual_disruption_cost
    )
    strategies.append({
        'name': '100% Primary (Single Source)',
        'primary_%': 100,
        'backup_%': 0,
        'cost': cost_100_0,
        'risk': 'High'
    })

    # Strategy 2: 80/20 split
    primary_qty_80 = annual_demand * 0.8
    backup_qty_20 = annual_demand * 0.2
    cost_80_20 = (
        primary_qty_80 * primary_supplier['unit_cost'] +
        backup_qty_20 * backup_supplier['unit_cost'] +
        expected_annual_disruption_cost * 0.3  # 70% risk reduction
    )
    strategies.append({
        'name': '80/20 Split',
        'primary_%': 80,
        'backup_%': 20,
        'cost': cost_80_20,
        'risk': 'Medium'
    })

    # Strategy 3: 70/30 split
    primary_qty_70 = annual_demand * 0.7
    backup_qty_30 = annual_demand * 0.3
    cost_70_30 = (
        primary_qty_70 * primary_supplier['unit_cost'] +
        backup_qty_30 * backup_supplier['unit_cost'] +
        expected_annual_disruption_cost * 0.2  # 80% risk reduction
    )
    strategies.append({
        'name': '70/30 Split',
        'primary_%': 70,
        'backup_%': 30,
        'cost': cost_70_30,
        'risk': 'Low'
    })

    # Strategy 4: 50/50 split (full redundancy)
    primary_qty_50 = annual_demand * 0.5
    backup_qty_50 = annual_demand * 0.5
    cost_50_50 = (
        primary_qty_50 * primary_supplier['unit_cost'] +
        backup_qty_50 * backup_supplier['unit_cost'] +
        expected_annual_disruption_cost * 0.1  # 90% risk reduction
    )
    strategies.append({
        'name': '50/50 Split (Full Redundancy)',
        'primary_%': 50,
        'backup_%': 50,
        'cost': cost_50_50,
        'risk': 'Very Low'
    })

    # Find optimal strategy (lowest total cost)
    optimal = min(strategies, key=lambda x: x['cost'])

    return {
        'strategies': strategies,
        'optimal': optimal,
        'savings_vs_single_source': cost_100_0 - optimal['cost']
    }


# Example
primary = {'unit_cost': 10.00}
backup = {'unit_cost': 10.80}

disruption_scenarios = [
    {'probability': 0.10, 'duration_days': 30},   # 10% chance of 30-day disruption
    {'probability': 0.05, 'duration_days': 90},   # 5% chance of 90-day disruption
]

result = design_dual_sourcing_strategy(
    primary, backup, annual_demand=100000, disruption_scenarios=disruption_scenarios
)

print("Dual Sourcing Analysis:")
for strategy in result['strategies']:
    print(f"\n{strategy['name']}")
    print(f"  Split: {strategy['primary_%']}% / {strategy['backup_%']}%")
    print(f"  Total Cost: ${strategy['cost']:,.0f}")
    print(f"  Risk Level: {strategy['risk']}")

print(f"\n\nOptimal Strategy: {result['optimal']['name']}")
print(f"Savings vs Single Source: ${result['savings_vs_single_source']:,.0f}")

---

工具与库

Python 库

风险分析：

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

可视化：

• matplotlib、seaborn：风险图表和仪表板
• plotly：交互式风险可视化
• networkx：供应商网络映射

监控：

• requests：用于数据源的 API 集成
• beautifulsoup4：用于新闻监控的网页抓取
• schedule：自动化监控任务

商业软件

风险管理平台：

• Resilinc：供应链风险管理与监控
• RapidRatings：财务健康评级
• Dun & Bradstreet：供应商风险情报
• Everstream Analytics：AI 驱动的风险预测
• Interos：供应链韧性
• Exiger：第三方风险管理

供应商绩效：

• SAP Ariba：供应商管理
• Coupa：供应商风险与绩效
• Jaggaer：供应商生命周期管理
• GEP：供应商风险评估

业务连续性：

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

---

常见挑战与解决方案

挑战：供应商可见性有限

问题：

• 缺少实时数据
• 供应商不愿分享信息
• N 级可见性（供应商的供应商）

解决方案：

• 在合同中规定透明度要求
• 第三方风险监控服务
• 供应商审计与现场考察
• 行业基准比较
• 协作式风险共担协议
• 使用技术（IoT、区块链）

挑战：风险评估的主观性

问题：

• 风险评分不一致
• 个人偏见
• 缺少标准化标准

解决方案：

• 标准化风险评估框架
• 尽可能使用量化指标
• 多名评估者（交叉验证）
• 定期校准会议
• 自动化评分算法
• 外部验证（审计、评级）

挑战：警报疲劳

问题：

• 低优先级警报过多
• 错过重要信号
• 资源不堪重负

解决方案：

• 基于风险进行优先级排序（重点关注关键供应商）
• 调整阈值（减少误报）
• 汇总报告（每周摘要）
• 升级流程（分级响应）
• 使用机器学习检测异常

挑战：缓解措施的成本合理性

问题：

• 难以量化风险节省
• 前期成本与不确定收益
• 预算限制

解决方案：

• 预期价值分析（概率 × 影响）
• 情景规划（压力测试）
• 历史中断成本数据
• 保险类比（风险溢价）
• 试点项目（证明投资回报率）
• 高管风险研讨会

挑战：供应商抵触

问题：

• 供应商抵制审计
• 不愿分享财务数据
• 认为风险问题冒犯自己

解决方案：

• 强调合作伙伴关系（互惠互利）
• 分阶段推进（建立信任）
• 互惠透明
• 符合行业标准
• 合同义务
• 为参与提供激励

---

输出格式

供应商风险报告

执行摘要：

• 供应商整体风险组合状态
• 关键警报与所需采取的行动
• 趋势分析（改善／恶化）
• 主要建议

单个供应商评估：

Supplier: Acme Manufacturing Inc.
Risk Tier: HIGH (Score: 62/100)
Criticality: Tier 1 - Critical Supplier

Risk Category Scores:
  Financial:       48/100 (Medium)
  Operational:     65/100 (High)
  Geopolitical:    35/100 (Low)
  Compliance:      25/100 (Low)
  Quality:         45/100 (Medium)
  Cybersecurity:   55/100 (Medium)

Top Risk Factors:
  1. High capacity utilization (>90%)
  2. Single facility operation
  3. Credit rating downgrade (BB)
  4. Missing ISO 27001 certification
  5. Elevated employee turnover

Business Impact:
  Annual Spend:           $5.2M
  Revenue at Risk:        $18M
  Customers Affected:     120 (15 critical accounts)
  Replacement Time:       16 weeks
  Criticality Score:      75/100

Mitigation Actions Required:
  1. [HIGH] Qualify backup supplier within 60 days
  2. [HIGH] Build 90-day safety stock
  3. [MEDIUM] Conduct site visit to assess capacity
  4. [MEDIUM] Request cybersecurity audit
  5. [LOW] Negotiate performance guarantees in contract renewal

Contingency Plan: ACTIVATED
  - Backup supplier identified: XYZ Corp
  - Safety stock build: In Progress (45 days complete)
  - Alternative sourcing: Under evaluation

Next Review: 30 days

组合风险仪表板：

供应商	风险评分	关键性	支出	状态	趋势	所需行动
Acme Mfg	62（高）	Tier 1	$5.2M	⚠	↗	双重采购
Beta Corp	35（中）	Tier 1	$3.8M	✓	→	监控
Gamma Inc	78（严重）	Tier 2	$1.5M	⚠⚠	↗	更换
Delta Ltd	22（低）	Tier 3	$0.8M	✓	↘	无

---

需要询问的问题

如果需要更多背景信息：

1. 哪些供应商或类别值得关注？
2. 最近是否发生过供应中断？
3. 目前有哪些风险数据？
4. 供应商的重要性如何（支出、可替代性、影响）？
5. 是否存在单一来源依赖？
6. 组织的风险承受能力如何？
7. 目前有哪些监控流程？
8. 是否存在行业特有风险？（地缘政治、监管）
9. 风险缓解预算是多少？
10. 风险管理的关键利益相关者有哪些？

---

相关技能

• supplier-selection：用于在遴选过程中审核供应商
• procurement-optimization：用于制定最优采购策略
• strategic-sourcing：用于类别风险管理
• spend-analysis：用于分析集中度风险
• contract-management：用于提供合同层面的风险保护
• quality-management：用于监控供应商质量
• risk-mitigation：用于制定整体供应链风险策略