供应链What-If情景分析引擎 — 因果ML参数化多情景对比与韧性量化评估

import numpy as np
from typing import List, Dict, Optional, Any
from dataclasses import dataclass, field
import itertools

@dataclass
class ScenarioParameter:
    """情景参数定义"""
    name: str
    values: List[Any]           # 待比较的参数值
    unit: str = ""
    is_causal_treatment: bool = True  # 是否作为因果干预变量

@dataclass
class ScenarioResult:
    """单个情景的分析结果"""
    scenario_id: str
    params: Dict[str, Any]
    gmv_estimate: float
    gmv_p10: float
    gmv_p90: float
    profit_estimate: float
    risk_score: float           # 0-100, 越高越危险
    is_pareto_optimal: bool = False
    recommendation: str = ""

class WhatIfScenarioEngine:
    """
    供应链 What-If 情景分析引擎
    
    对标 Palantir Workshop Layer:
    - 参数化多情景定义
    - 因果效应估算（双重稳健）
    - Monte Carlo 不确定性量化
    - Pareto 最优情景识别
    """
    
    def __init__(self, causal_model=None, n_simulations: int = 500):
        """
        Args:
            causal_model: 因果模型（可传入 SCCausalDAG 实例）
            n_simulations: Monte Carlo 次数
        """
        self.causal_model = causal_model
        self.n_simulations = n_simulations
    
    def _estimate_causal_effect(self, treatment_var: str, treatment_value: float,
                                 control_value: float, baseline_metric: float,
                                 elasticity: float, std_frac: float = 0.15) -> tuple:
        """
        因果效应估算（简化的双重稳健估计）
        
        Returns:
            (effect_estimate, p10, p90)
        """
        # 点估计：线性因果效应
        delta_treatment = treatment_value - control_value
        effect = baseline_metric * (1 + elasticity * delta_treatment / max(abs(control_value), 1))
        
        # Monte Carlo 不确定性