CDA — 隐私保护因果渠道归因：无用户数据的多触点归因

1. 解决的问题

传统多触点归因（Multi-Touch Attribution, MTA）依赖用户级点击路径数据：追踪每个用户从广告曝光→点击→转化的完整旅程，才能判断各渠道贡献。

2. 核心算法逻辑

传统 MTA 在隐私时代的失效

3. 业务应用场景

业务背景： 母婴 DTC 品牌在欧洲市场同时投放 Google、Meta、TikTok 三渠道。GDPR 合规要求下，无法使用用户级 cookie 追踪，传统 MTA 完全失效。

产出： - Google 直接归因权重：42%（ROAS 计算基准） - Meta 归因权重：35%（含 TikTok 引流的间接效应） - TikTok 归因权重：23%（较 Last-Click 低估已修正）

业务背景： 618 大促期间 Google/Meta/TikTok 三渠道同时加大投入，传统 Last-Click 把所有转化归给最后点击渠道（通常是品牌词搜索），严重低估 TikTok 的种草效应。

4. 输入数据要求

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

5. 输出结果

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

6. 业务价值 / ROI

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

7. 代码模板

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

"""
CDA — 因果驱动归因（隐私保护多渠道归因）
论文：Causal-driven attribution (CDA): Estimating channel influence without user-level data
arXiv：2512.21211 | 2024年12月
"""

from __future__ import annotations
from dataclasses import dataclass, field
from typing import Optional
import math
import statistics
from collections import defaultdict


# ──────────────────────────────────────────────
# 数据类
# ──────────────────────────────────────────────

@dataclass
class ChannelTimeSeries:
    """单渠道时序数据"""
    channel_name: str
    daily_impressions: list[float]  # 每日曝光量
    daily_conversions: list[float]  # 每日转化量（汇总，非用户级）

    def __post_init__(self):
        assert len(self.daily_impressions) == len(self.daily_conversions), \
            "曝光量和转化量时序长度必须一致"

    @property
    def n_days(self) -> int:
        return len(self.daily_impressions)

    def impression_rate(self) -> list[float]:
        """归一化曝光率"""
        max_imp = max(self.daily_impressions) or 1.0
        return [x / max_imp for x in self.daily_impressions]


@dataclass
class CausalDAG:
    """因果有向无环图"""
    channels: list[str]
    edges: list[tuple[str, str, int]]  # (from_channel, to_channel, lag_days)
    edge_weights: dict[tuple[str, str], float] = field(default_factory=dict)

    def get_parents(self, channel: str) -> list[tuple[str, int]]:
        """获取某渠道的所有父节点（直接因果来源）"""
        return [(src, lag) for src, dst, lag in self.edges if dst == channel]

    def get_children(self, channel: str) -> list[tuple[str, int]]:
        """获取某渠道的所有子节点（直接因果影响）"""
        return [(dst, lag) for src, dst, lag in self.edges if src == channel]


# ──────────────────────────────────────────────
# 简化版 PCMCI 因果发现
# ──────────────────────────────────────────────

class PCMCICausalDiscovery:

8. 论文来源

• 2512.21211