Supply Chain Causal SCM — 数据驱动供应链根因归因：DAG + DoWhy GCM

1. 解决的问题

为什么传统归因不够：瀑布式逻辑（"缺货→往上查库存→往上查采购"）本质上是相关性分析，无法区分"A 导致 B"与"C 同时导致 A 和 B"。实际供应链中，多因素常常通过间接路径（中介变量）影响结果——比如"需求预测误差→战术产能调整→Capped Out Hours（COH）"，传统归因会错误地把间接效应归给直接可见的变量。

2. 核心算法逻辑

为什么传统归因不够：瀑布式逻辑（"缺货→往上查库存→往上查采购"）本质上是相关性分析，无法区分"A 导致 B"与"C 同时导致 A 和 B"。实际供应链中，多因素常常通过间接路径（中介变量）影响结果——比如"需求预测误差→战术产能调整→Capped Out Hours（COH）"，传统归因会错误地把间接效应归给直接可见的变量。

3. 业务应用场景

- 业务问题：母婴奶粉 SKU 出现缺货（库存降至 0），损失 GMV 约 8 万元/周。管理层需要快速定位：是需求预测误差（预测比实际低 40%）、交货期延误（供应商迟发货 3 天），还是安全库存设置过低？三个因素同时出现，传统瀑布分析无法量化各自责任； - 数据要求：30 天历史数据，每日变量：需求预测值、实际需求、采购触发量、供应商交货期、实际到货量、库存水位；共 6 个节点 - SCM 做法：PC 算法学习 DAG（需求预测误差→补货量→库存水位，交货期延误→到货量→库存水位，安全库存参数→补货触发点→补货量），拟合结构方程后对缺货时间点执行根因归因 - 预期产出：输出各因素归因比例，

场景二：WF-A 补货决策干预分析（"如果预测精度提升 10% 会怎样？"）

- 业务问题：WF-A 仓库补货计划师希望量化：如果把需求预测 MAPE 从 25% 降低到 15%，补货量应该如何调整？库存降低多少？这是一个反事实/干预问题，不是回归问题 - 数据要求：历史 60 天补货决策链路数据（预测误差、补货量、库存日均水位、缺货率），需已有拟合好的 SCM - SCM 做法：执行干预分析 `do(预测误差 = 原误差 × 0.6)` （模拟精度提升40%），通过 SCM 前向传播估计补货量和库存水位的变化；再用反事实推理量化已发生时间段的"假设场景" - 预期产出：预测精度提升 10% → 补货量减少约 8%（减少超订），库存日均水位降低 12%，缺货率从 3.2

4. 输入数据要求

请查看原始代码模板获取输入规格。

5. 输出结果

请查看原始代码模板获取输出规格。

6. 业务价值 / ROI

• ROI：供应链异常定位时间从 2-3 天→2-3 小时，量化干预效果避免无效行动；母婴旺季缺货损失每次约 5-10 万元 GMV，快速归因可减少 50%+ 的重复缺货事件
• 实施难度：⭐⭐⭐⭐☆（需要因果发现 + SCM 建模知识；但供应链 DAG 结构相对清晰，可由领域专家直接构建）
• 优先级：⭐⭐⭐⭐⭐（缺货/积压归因是供应链最高频的运营痛点，所有 SKU 类目均适用）
• 评估依据：Amazon Science 案例验证 SCM 比传统瀑布分析提供更深入可行洞察；母婴品类季节性强、供应链链路短，SCM 节点数少（5-8个），实施复杂度可控

7. 代码模板

代码块数量：1 · 路径：未检测到

"""
Supply Chain Causal SCM Attribution
Amazon Science 2025 | Pole, Rakshit et al.
端到端因果归因：DAG → SCM → 根因/干预/反事实
纯标准库 + numpy/scipy 实现（不依赖 DoWhy，原理实现）
"""
from __future__ import annotations

import math
import numpy as np
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Tuple, Any
from scipy import stats


# ─── 数据结构 ─────────────────────────────────────────────────────────────────

@dataclass
class SCMVariable:
    """供应链 SCM 变量节点"""
    name: str
    description: str
    var_type: str   # "exogenous"（外生）或 "endogenous"（内生）
    values: Optional[np.ndarray] = None  # 历史观测值


@dataclass
class AnomalyResult:
    """根因归因结果"""
    total_anomaly: float
    attributions: Dict[str, float]   # 各节点归因值
    attribution_ratio: Dict[str, float]  # 各节点归因比例
    root_causes: List[str]           # 主要根因（归因比例 > 20%）

    def __str__(self) -> str:
        lines = [f"异常总量: {self.total_anomaly:.4f}", "归因分解:"]
        for node, ratio in sorted(
            self.attribution_ratio.items(), key=lambda x: -x[1]
        ):
            mark = " ← 主要根因" if node in self.root_causes else ""
            lines.append(f"  {node}: {ratio:.1%}{mark}")
        return "\n".join(lines)


# ─── 因果 DAG ─────────────────────────────────────────────────────────────────

class CausalDAG:
    """
    有向无环图（DAG）：供应链因果结构

    支持：
    - 手动构建（领域专家知识）
    - 简单数据驱动验证（条件独立性检验）
    """

    def __init__(self):
        self._edges: List[Tuple[str, str]] = []
        self._nodes: set = set()

    def add_edge(self, from_node: str, to_node: str) -> None:

8. 论文来源

未自动抽取；请查看原始 Skill 卡片。