供应链决策置信度校准与分层触发 — 从置信度到人机分工的动态决策框架

"""
供应链决策置信度校准与分层触发系统
功能：Temperature Scaling校准 / ECE评估 / 置信度分层 / Palantir Action路由
输入：原始模型置信度 + 历史校准数据
输出：校准后置信度 + 决策层级 + Palantir Action建议
"""
import numpy as np
from dataclasses import dataclass
from typing import Optional
from enum import Enum
import warnings
warnings.filterwarnings('ignore')


class ActionTier(Enum):
    AUTONOMOUS = ("autonomous", ">95%", "全自动执行，事后审计")
    AUTO_WATCH = ("auto_watch", "85-95%", "自动执行，Watch窗口24h")
    GUIDED = ("guided", "70-85%", "推荐+2h内确认")
    STAGED = ("staged", "50-70%", "人工审批，24h SLA")
    ESCALATE = ("escalate", "<50%", "升级专家，停止自动Action")
    
    def __init__(self, code, range_str, description):
        self.code = code
        self.range_str = range_str
        self.description = description


@dataclass
class CalibratedDecision:
    """校准后的决策对象——直接映射到Palantir Action Type"""
    original_confidence: float
    calibrated_confidence: float
    action_tier: ActionTier
    decision_id: str
    context: dict
    
    def to_palantir_action(self) -> dict:
        return {
            "action_tier": self.action_tier.code,
            "confidence": self.calibrated_confidence,
            "original_confidence": self.original_confidence,
            "calibration_applied": True,
            "approval_required": self.action_tier in [ActionTier.STAGED, ActionTier.ESCALATE],
            "auto_revert_window_hours": 24 if self.action_tier == ActionTier.AUTO_WATCH else None,
            "escalation_reason": (
                f"置信度{self.calibrated_confidence:.1%}低于阈值"
                if self.action_tier == ActionTier.ESCALATE else None
            ),
        }


class TemperatureScalingCalibrator:
    """Temperature Scaling置信度校准器"""
    
    def __init__(self):
        self.temperature = 1.0  # 初始无校准
        self._fitted = False
        self._calibration_history = []
    
    def fit(self, logits: np.ndarray, labels: np.ndarray,