因果时间序列预测 - GCF 反事实需求建模

Skill-Causal-Time-Series-Forecasting-GCF · 03-时间序列

causalexperimentforecastingknowledge_graphmulti_agent广告与投放供应链与补货知识图谱与RAGMAS与智能体工程WF-A 智能补货WF-B 广告优化WF-G Listing内容优化WF-H 复购增长

年化 ROI50-100 万元

实现难度⭐⭐⭐⭐☆

业务优先级⭐⭐⭐⭐☆

业务视角

适用角色供应链负责人 / 采购负责人 · 运营负责人 · 财务负责人

适用平台Amazon FBA · 海外仓 · 多市场多仓

什么情况下用大促前备货总是不是多了就是少了；新品上线第一个月断货，再补又积压；年底预算不知道各月目标怎么定

成功是什么样的提前 4-8 周准确预判各 SKU 需求峰值，库存积压减少 30%，断货率降低 50%

业务痛点

备货总是压货或断货旺季淡季波动太大预测不准补货周期 30 天但预测只看 7 天

1. 解决的问题

平台在大促期对核心母婴 SKU(纸尿裤、婴儿车)做搜索权重提升 + 首页 Banner 曝光,需要回答"如果没做促销,需求应该是多少"——避免把自然增长功劳归到促销 - 数据要求:全品类销量历史 + 促销标记 + 商品图谱(同类竞品关系) - GCF 配置:节点=SKU,边=同品类竞品,干预=促销曝光,合成控制=未受促销的同类 - 业务价值:促销 ROI 计算精度提升 30-50%,避免

2. 核心算法逻辑

电商场景中,商品因配送延迟、缺货、Banner 压制等干预导致需求被"压制",真实需求 $Y(0)$(无干预反事实)永远不可观测。GCF 用 RGCN + Dilated CNN 同时建模商品间空间关系(同类竞品图)和时序长程依赖,自动选取未受干预的相似商品作为合成控制组,估计反事实需求。

3. 业务应用场景

- 业务问题:平台在大促期对核心母婴 SKU(纸尿裤、婴儿车)做搜索权重提升 + 首页 Banner 曝光,需要回答"如果没做促销,需求应该是多少"——避免把自然增长功劳归到促销 - 数据要求:全品类销量历史 + 促销标记 + 商品图谱(同类竞品关系) - GCF 配置:节点=SKU,边=同品类竞品,干预=促销曝光,合成控制=未受促销的同类 - 业务价值:促销 ROI 计算精度提升 30-50%,避免无效促销重复投放,以单次大促 500 万元预算计,节省浪费支出 50-100 万元/次

- 业务问题:海运延误/缺货时商品 listing 自动下架(干预 $T=1$),恢复后需要估计"缺货期间真实需求是多少"以确定补货量,避免二次缺货或过量 - 数据要求:订单日志 + 配送 SLA 状态 + 同品类同价位段竞品销量 - GCF 配置:干预=配送 SLA 突变,控制组=未受影响的同类竞品图邻居,反事实=正常供货下需求曲线 - 业务价值:补货精度提升 40-60%,避免二次缺货导致的客户流失;按缺货事件平均损失 20 万元/单仓/次计,年化避免损失 200-400 万元

4. 输入数据要求

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

5. 输出结果

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

6. 业务价值 / ROI

难处:无开源代码,需自行实现 RGCN + Dilated CNN(参考 PyTorch Geometric)
难处:需要构建商品关系图(可与 Hierarchical-Product-KG 配合)
GPU 需求中-高(多个 RGCN 层 + 长序列)

7. 代码模板

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

"""
GCF Causal Forecasting 最小骨架
论文 AAAI 2025 (Amazon), DOI: 10.1609/aaai.v39i28.35148
无公开代码,以下骨架按论文 §3-4 还原。
"""
from __future__ import annotations
import torch
import torch.nn as nn
import torch.nn.functional as F


class RGCNEncoder(nn.Module):
    def __init__(self, in_dim: int = 32, hid_dim: int = 64, n_relations: int = 3):
        super().__init__()
        self.W = nn.Parameter(torch.randn(n_relations, in_dim, hid_dim) * 0.1)
        self.W_self = nn.Linear(in_dim, hid_dim)
        self.n_relations = n_relations

    def forward(self, x: torch.Tensor, edge_index: torch.Tensor, edge_type: torch.Tensor) -> torch.Tensor:
        h = self.W_self(x)
        for r in range(self.n_relations):
            mask = edge_type == r
            if mask.sum() == 0:
                continue
            src = edge_index[0][mask]
            dst = edge_index[1][mask]
            messages = x[src] @ self.W[r]
            h = h.index_add(0, dst, messages)
        return F.relu(h)


class DilatedTCN(nn.Module):
    def __init__(self, hid_dim: int = 64):
        super().__init__()
        self.conv1 = nn.Conv1d(hid_dim, hid_dim, kernel_size=3, dilation=1, padding=1)
        self.conv2 = nn.Conv1d(hid_dim, hid_dim, kernel_size=3, dilation=2, padding=2)
        self.pool = nn.AdaptiveAvgPool1d(1)
        self.out = nn.Linear(hid_dim, 1)

    def forward(self, x: torch.Tensor) -> torch.Tensor:
        h = F.relu(self.conv1(x))
        h = F.relu(self.conv2(h))
        h = self.pool(h).squeeze(-1)
        return self.out(h)


class GCF(nn.Module):
    def __init__(self, in_dim: int = 32, hid_dim: int = 64):
        super().__init__()
        self.encoder = RGCNEncoder(in_dim, hid_dim)
        self.decoder = DilatedTCN(hid_dim)

    def forward(self, x_seq: torch.Tensor, edge_index: torch.Tensor, edge_type: torch.Tensor) -> torch.Tensor:
        T = x_seq.shape[1]
        h_list = [self.encoder(x_seq[:, t, :], edge_index, edge_type) for t in range(T)]
        h_seq = torch.stack(h_list, dim=2)
        return self.decoder(h_seq)


def estimate_ate(y_factual: torch.Tensor, y_counterfactual: torch.Tensor, treated_mask: torch.Tensor) -> float:

8. 论文来源

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