供应链决策结果闭环学习 — 从执行结果到模型改进的Palantir决策飞轮

"""
供应链决策结果闭环学习系统
功能：选择偏差纠正 / 双重鲁棒估计 / 在线模型更新 / 决策飞轮监控
输入：历史决策日志（含倾向得分）
输出：无偏因果效应估计 + 更新后模型 + 飞轮健康度报告
"""
import numpy as np
from dataclasses import dataclass, field
from collections import deque
import warnings
warnings.filterwarnings('ignore')


@dataclass
class DecisionRecord:
    """决策记录——从Palantir Audit Log中提取"""
    decision_id: str
    context: dict           # 特征向量（库存水平、需求预测等）
    action_taken: float     # 执行的决策值（如补货量）
    propensity_score: float # 该决策被选择的概率（e分数）
    outcome: float          # 实际结果（如服务水平）
    predicted_outcome: float # 模型预测的结果
    confidence_at_decision: float # 决策时的置信度


class DoublyRobustLearner:
    """
    双重鲁棒学习器
    纠正供应链历史数据中的选择偏差
    """
    
    def __init__(self, n_components: int = 5):
        self.n_components = n_components
        # 结果模型（预测不同action下的结果）
        self._outcome_params_treated = np.zeros(n_components)
        self._outcome_params_control = np.zeros(n_components)
        self._fitted = False
    
    def _extract_features(self, context: dict) -> np.ndarray:
        """从决策上下文中提取特征"""
        features = [
            context.get('inventory_level', 0) / 1000,
            context.get('predicted_demand', 0) / 1000,
            context.get('service_level_history', 0.9),
            context.get('season_index', 0.5),
            context.get('lead_time_days', 28) / 60,
        ]
        return np.array(features[:self.n_components])
    
    def fit(self, records: list):
        """用历史记录拟合结果模型"""
        if len(records) < 10:
            return self
        
        # 简化的线性结果模型
        treated = [r for r in records if r.action_taken > np.median([r.action_taken for r in records])]
        control = [r for r in records if r.action_taken <= np.median([r.action_taken for r in records])]
        
        if treated and control:
            X_treated = np.array([self._extract_features(r.context) for r in treated])