Organic Content Causal Attribution — 无用户数据的有机内容因果归因 (CDA)
Skill-Organic-Content-Causal-Attribution · 15-营销投放分析
causalexperimentforecastingoptimization广告与投放推荐与搜索WF-B 广告优化WF-D 选品扫描WF-G Listing内容优化WF-L 内容营销增长
年化 ROI15 万
实现难度⭐⭐⭐☆☆
业务优先级⭐⭐⭐⭐⭐
业务视角
适用角色CMO / 营销负责人 · 广告优化师 · CEO
适用平台Amazon + TikTok + Meta + KOL 四渠道 · Prime Day / Black Friday 预算前置
什么情况下用同时跑 Amazon 广告/TikTok/网红投放/邮件，不知道整体预算怎么分配最高效；网红投放花了大钱但不知道带来多少真实 GMV
成功是什么样的建立全渠道营销归因模型（MMM），每个渠道真实 ROI 可量化，大促前做预算优化模拟
业务痛点
1. 解决的问题
每月 15 万 KOL 种草投入但不知道有没有带来 Amazon 品牌词搜索增长——CDA 框架仅用聚合曝光数据推断各渠道因果贡献（无需用户 ID），砍掉无效内容投入，年化节省 54 万元以上
2. 核心算法逻辑
核心思想：传统营销归因（MTA/MMM）依赖用户级路径数据（cookie/IDFA），在 iOS14 隐私政策后几乎不可用。更大的问题是有机内容（KOL种草/博客/TikTok自然流量）根本没有用户路径——你不知道是哪篇小红书笔记带来了 Amazon 的品牌词搜索增长。
3. 业务应用场景
场景：KOL 种草 × Amazon 品牌词搜索量因果归因
- 业务问题：某母婴品牌每月在小红书/TikTok 上合作 20 个 KOL，总投入 15 万元，但不知道这些内容有没有带来 Amazon 的品牌词搜索增长（Momcozy 自然搜索），还是只是花了钱没效果。 - 数据要求： - KOL 发帖时间 + 曝光量（按渠道汇总，周维度） - Amazon 品牌词搜索量（Amazon Brand Analytics） - 自然流量点击量（Google Search Console） - GMV 时间序列（周维度，12 个月历史） - 预期产出： - 每个内容渠道的因果贡献度（"小红书 KOL → +12% 品牌搜索，滞后 14 天"） - 渠道因果 D
4. 输入数据要求
请查看原始代码模板获取输入规格。
5. 输出结果
请查看原始代码模板获取输出规格。
6. 业务价值 / ROI
- ROI 预估：优化内容渠道预算分配，砍掉 20-50% 无效投入，月投入 15 万 × 30% = 年化节省 54 万元
- 实施难度：⭐⭐⭐☆☆（中等，需要时序数据整合 + PCMCI 因果发现库）
- 优先级：⭐⭐⭐⭐⭐（随隐私政策收紧，有机归因是下一个核心竞争力）
- 评估依据：arXiv 2512.21211，开源代码 + 大规模模拟验证，RMSE 9.5%，隐私友好
7. 代码模板
代码块数量：2 · 路径：未检测到
import numpy as np
from dataclasses import dataclass
from typing import List, Dict, Tuple
@dataclass
class ChannelTimeSeries:
name: str
weekly_values: List[float]
def granger_causality_simple(cause: List[float], effect: List[float],
max_lag: int = 4) -> Dict[str, float]:
n = len(cause)
best_lag = 0
best_corr = 0.0
for lag in range(1, min(max_lag + 1, n // 3)):
x_lagged = cause[:-lag]
y_shifted = effect[lag:]
if len(x_lagged) < 4:
continue
x_arr = np.array(x_lagged)
y_arr = np.array(y_shifted)
if x_arr.std() < 1e-9 or y_arr.std() < 1e-9:
continue
corr = float(np.corrcoef(x_arr, y_arr)[0, 1])
if abs(corr) > abs(best_corr):
best_corr = corr
best_lag = lag
return {"optimal_lag_weeks": best_lag, "correlation": round(best_corr, 3),
"causal_strength": round(abs(best_corr), 3)}
def counterfactual_contribution(channel: ChannelTimeSeries, outcome: List[float],
reduction_pct: float = 1.0) -> Dict[str, float]:
gc = granger_causality_simple(channel.weekly_values, outcome)
avg_channel = np.mean(channel.weekly_values)
avg_outcome = np.mean(outcome)
contribution_pct = gc["causal_strength"] * reduction_pct * 100
counterfactual_loss = avg_outcome * (contribution_pct / 100)
return {"channel": channel.name, "causal_strength": gc["causal_strength"],
"optimal_lag_weeks": gc["optimal_lag_weeks"],
"estimated_contribution_pct": round(contribution_pct, 1),
"counterfactual_gmv_loss": round(counterfactual_loss, 0)}
def cda_attribution(channels: List[ChannelTimeSeries], gmv: List[float]) -> List[Dict]:
results = []
for ch in channels:
contrib = counterfactual_contribution(ch, gmv)
results.append(contrib)
total_strength = sum(r["causal_strength"] for r in results)
for r in results:
r["attribution_share"] = round(r["causal_strength"] / max(total_strength, 1e-9) * 100, 1)
return sorted(results, key=lambda x: -x["causal_strength"])
np.random.seed(42)
weeks = 52
t = np.linspace(0, 4 * np.pi, weeks)
kol_exposure = (500 + 200 * np.sin(t) + np.random.normal(0, 50, weeks)).clip(0)
paid_ads = (1000 + 300 * np.cos(t * 0.5) + np.random.normal(0, 80, weeks)).clip(0)
seo_organic = (300 + 100 * np.sin(t * 0.3) + np.random.normal(0, 30, weeks)).clip(0)
gmv = (50000 + 0.8 * np.roll(kol_exposure, 2) + 1.2 * paid_ads + 0.5 * seo_organic
+ np.random.normal(0, 2000, weeks)).clip(0)8. 论文来源
- 2512.21211