PVM 跨平台广告归因窗口统一化 - 母婴跨境多渠道 ROAS 去偏

1. 解决的问题

WF-B 跨渠道归因痛点:Amazon 14d-click、Meta 7d-click、TikTok 7d-click 归因窗口不一致,Last-Click Mechanism (LCM) 让平台策略性延迟上报点击时间抢归因信用,LCM 不满足 DSIC (Dominant Strategy Incentive Compatible),准确率最低可趋近于 0. PVM (Peer-Validated Mechanism) 让每个平台的归因信用仅依赖其他平台的报告而非自身,消除策略操控动机,理论最

2. 核心算法逻辑

WFB 跨渠道归因痛点:Amazon 14dclick、Meta 7dclick、TikTok 7dclick 归因窗口不一致,LastClick Mechanism (LCM) 让平台策略性延迟上报点击时间抢归因信用,LCM 不满足 DSIC (Dominant Strategy Incentive Compatible),准确率最低可趋近于 0. PVM (PeerValidated Mechanism) 让每个平台的归因信用仅依赖

3. 业务应用场景

- 业务问题:Momcozy 同投 Amazon SP + Meta DPA + TikTok Shop. 用户 6 天前在 TikTok 看吸奶器短视频, 2 天前看 Meta DPA 再营销, 5 分钟前在 Amazon 搜索点击购买. Amazon 14d-click 抢走 100% 归因信用, TikTok/Meta ROAS 系统性低估,导致削减二者预算 → 全渠道流量恶性循环 - 数据要求:三平台 click_time + 转化时间戳 + spend - PVM 配置: - 各平台触点统一映射到 $(-W, 0]$ 转化对齐时间轴 - PVM 按"同行报告时序概率"分配信用(Sof

- 业务问题:大促后 Amazon 声称带来 GMV 500 万,Meta 声称 200 万,TikTok 声称 150 万,三者之和远超实际 GMV 600 万(双重计数). 无法判断真实渠道贡献 → 下一轮大促预算分配错误率 40%+ - 数据要求:大促期间(6.1-6.18)所有订单 + 三平台触点记录 - PVM 配置: - 窗口截断审计:每笔订单提取三平台窗口内最后点击 - 重复转化识别:同 order_id 多平台归因 → 强制 $\sum_i x_i \leq 1$ - 输出"渠道矫正 ROAS"+ 重复计数率报告 - 业务价值: - 跨渠道 ROAS 可比性 → 下一轮大促渠道

4. 输入数据要求

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

5. 输出结果

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

6. 业务价值 / ROI

• 易处:PyMAL GitHub 开源(MAC 基准 + 多归因学习基线)
• 易处:PVM 数学框架明确,可纯 Python 实现核心
• 难处:PVM 主论文是理论论文,无官方代码
• 难处:三平台 click_time 精确对齐需要 ETL 工程(timezone / 时间戳格式)

7. 代码模板

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

"""
PVM 跨平台归因窗口统一化最小骨架
主论文 arXiv:2511.22918 (NeurIPS 2025)
辅: PyMAL https://github.com/alimama-tech/PyMAL
依赖: pip install numpy
"""
from __future__ import annotations
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from typing import Dict, List, Optional

import numpy as np


ATTRIBUTION_WINDOWS = {
    "amazon": timedelta(days=14),
    "meta": timedelta(days=7),
    "tiktok": timedelta(days=7),
}


@dataclass
class TouchPoint:
    platform: str
    click_time: datetime
    spend: float = 0.0


@dataclass
class Conversion:
    order_id: str
    convert_time: datetime
    revenue: float
    touchpoints: List[TouchPoint] = field(default_factory=list)


def normalize_to_conversion_timeline(tp: TouchPoint, t0: datetime, window: timedelta) -> Optional[float]:
    """Step 1: 转化对齐时间归一化 t_i = t_i^abs - t_0"""
    relative_seconds = (tp.click_time - t0).total_seconds()
    if relative_seconds > 0:
        return None
    if tp.click_time < t0 - window:
        return None
    return relative_seconds


def pvm_attribution(conversion: Conversion, windows: Dict[str, timedelta] = None) -> Dict[str, float]:
    """Peer-Validated Mechanism: 信用按时序加权,sum <= 1"""
    windows = windows or ATTRIBUTION_WINDOWS
    eligible: Dict[str, float] = {}

    for tp in conversion.touchpoints:
        window = windows.get(tp.platform, timedelta(days=7))
        rel_t = normalize_to_conversion_timeline(tp, conversion.convert_time, window)
        if rel_t is not None:
            if tp.platform not in eligible or rel_t > eligible[tp.platform]:
                eligible[tp.platform] = rel_t

    if not eligible:
        return {}

8. 论文来源

• 2511.22918
• 2603.02184