① 问题 ② 算法 ③ 场景 ④ 代码 ⑤ 关联 ⑥ 价值

KG Supply Chain Cost Attribution — 图神经网络 + 因果推断的供应链成本归因

Skill-KG-Supply-Chain-Cost-Attribution · 08-知识图谱

causalexperimentforecastingknowledge_graph广告与投放供应链与补货客服与VOC知识图谱与RAGWF-A 智能补货WF-B 广告优化WF-C 客服分诊WF-D 选品扫描WF-E Review监控

收录于数据治理基础手册竞品情报→产品迭代加速器

年化 ROI3000-5000 万元

实现难度⭐⭐⭐⭐☆

业务优先级⭐⭐⭐☆☆

业务视角

适用角色选品负责人 / 运营负责人 · 数据分析师 · 供应链负责人

适用平台Amazon 品类体系 · 竞品 ASIN 网络分析

什么情况下用品类很多，不清楚品类间的关联，没法做系统性类目扩张规划；竞品矩阵太复杂，品牌/SKU/渠道理不清

成功是什么样的建立品类知识图谱，清晰看到哪些是入口品/引流品/利润品，指导下一步选品扩张方向

业务痛点

品类太多不知道先做哪个竞品关系理不清楚不知道用户买了奶瓶还会买什么类目扩张没有逻辑

1. 解决的问题

跨境母婴品牌供应链成本季度性激增，头程/原材料/仓储三因素难以区分——时序生产图 TPG + PC 算法因果归因将成本驱动识别准确率从 40% 提升至 85%，年化节省错误决策成本 50-150 万元

2. 核心算法逻辑

时序生产图（Temporal Production Graph, TPG）将供应链建模为有向无环图：节点 = 供应链环节（工厂、头程、仓库、SKU），边 = 成本流量。Stanford AAAI 2025 工作用异构 GNN 从买家供应商历史交易记录自动推断隐式 BOM（物料清单）结构和生产函数——即使没有显式 BOM 文档，也能从数据中学习"谁依赖谁、成本如何传导"。

3. 业务应用场景

- 业务问题：某母婴品牌（M5 吸奶器）Q4 总成本较 Q3 上涨 18%，财务团队不知道该压缩哪个环节——海运谈判、原材料采购还是换仓提效？ - 数据要求：6个月以上的 SKU 级成本流水（原材料采购价、头程运费发票、FBA 月结账单、制造工单） - 执行步骤： 1. 构建供应链 TPG（工厂→头程→FBA仓→SKU），每条边填入 Q3/Q4 单位成本 2. 路径分解得出各链路总成本变化（海运链路 +11.8%，空运链路 +14.8%） 3. PC 算法 + SCM 识别：头程运价方差贡献 73.5%（主因），制造成本贡献 38.2%（次因） 4. 反事实模拟：若海运谈判将头程降价 20 元

- **业务问题**：同时运营消毒器、吸奶器、婴儿车三类产品，CFO 想知道哪个品类供应链最脆弱（成本波动最难归因） - **数据要求**：按 SKU 分组的月度成本分项数据（至少 12 个月） - **执行方式**：对每个品类分别构建 TPG + 运行 SCM 归因，对比各品类"头程运价方差贡献"作为脆弱性指标 - **业务价值**：优先对高脆弱性品类建立运价对冲策略，年化稳定收益 20-50 万元

4. 输入数据要求

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

5. 输出结果

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

6. 业务价值 / ROI

ROI 估算：中型跨境品牌（年营收 3000-5000 万元）每个季度因成本归因不清导致错误决策（优先压错环节）的隐性损失约 15-40 万元/次；图谱归因将准确率从 40% 提升至 85%，每年可避免 2-3 次错误决策，年化节省 50-150 万元
实施难度：⭐⭐⭐⭐☆（需要整理历史成本分项流水；PC 算法需要 ≥90 天数据样本；无需外部 API 依赖）
优先级：⭐⭐⭐☆☆（建议在 SKU 级 P&L 体系建立后再引入，否则输入数据质量不足）
线性 SCM 在成本非线性传导（如量价联动）时精度下降，可替换为 XGBoost 代理模型
观测数据中存在隐藏混淆变量（如汇率同时影响原材料和头程）时需引入工具变量
PC 算法在变量数 > 10 时计算量指数增长，推荐使用 FCI 变体

7. 代码模板

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

import numpy as np
import networkx as nx
import warnings
warnings.filterwarnings("ignore")


class SupplyChainKG:

    def __init__(self):
        self.G = nx.DiGraph()
        self._build_demo_graph()

    def _build_demo_graph(self):
        nodes = [
            ("factory_CN",   {"type": "factory",   "name": "中国工厂"}),
            ("freight_sea",  {"type": "freight",   "name": "海运头程"}),
            ("freight_air",  {"type": "freight",   "name": "空运头程"}),
            ("warehouse_US", {"type": "warehouse", "name": "FBA美国仓"}),
            ("sku_M5",       {"type": "sku",       "name": "M5吸奶器"}),
        ]
        for nid, attr in nodes:
            self.G.add_node(nid, **attr)

        edges = [
            ("factory_CN",   "freight_sea",  {"cost_q3": 180, "cost_q4": 185, "label": "原材料+制造"}),
            ("factory_CN",   "freight_air",  {"cost_q3": 180, "cost_q4": 220, "label": "原材料+制造"}),
            ("freight_sea",  "warehouse_US", {"cost_q3": 45,  "cost_q4": 67,  "label": "海运费"}),
            ("freight_air",  "warehouse_US", {"cost_q3": 180, "cost_q4": 195, "label": "空运费"}),
            ("warehouse_US", "sku_M5",       {"cost_q3": 38,  "cost_q4": 42,  "label": "FBA仓储费"}),
        ]
        for src, dst, attr in edges:
            self.G.add_edge(src, dst, **attr)

    def decompose_cost_paths(self, period_a="cost_q3", period_b="cost_q4"):
        results = []
        sources = [n for n, d in self.G.nodes(data=True) if d["type"] == "factory"]
        sinks   = [n for n, d in self.G.nodes(data=True) if d["type"] == "sku"]
        for src in sources:
            for dst in sinks:
                for path in nx.all_simple_paths(self.G, src, dst):
                    cost_a = sum(self.G[u][v].get(period_a, 0) for u, v in zip(path[:-1], path[1:]))
                    cost_b = sum(self.G[u][v].get(period_b, 0) for u, v in zip(path[:-1], path[1:]))
                    delta  = cost_b - cost_a
                    pct    = delta / cost_a * 100 if cost_a else 0
                    path_label = " → ".join(self.G.nodes[n]["name"] for n in path)
                    results.append({
                        "path": path_label,
                        "cost_q3": cost_a, "cost_q4": cost_b,
                        "delta": delta, "pct": pct,
                    })
        return results

    def edge_contribution(self, period_a="cost_q3", period_b="cost_q4"):
        contributions = []
        for u, v, data in self.G.edges(data=True):
            ca = data.get(period_a, 0)
            cb = data.get(period_b, 0)
            delta = cb - ca
            contributions.append({
                "edge": f"{self.G.nodes[u]['name']} → {self.G.nodes[v]['name']}",

8. 论文来源

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