Marketing Data Pipeline — 营销归因多渠道数据采集管道

1. 解决的问题

营销归因的核心难题是数据孤岛：广告平台（Meta/Google/TikTok）、CRM（Salesforce/HubSpot）、电商平台（Amazon/Shopify）、社交媒体各持一方数据，无法直接关联。数据管道需要解决

2. 核心算法逻辑

营销归因的核心难题是数据孤岛：广告平台（Meta/Google/TikTok）、CRM（Salesforce/HubSpot）、电商平台（Amazon/Shopify）、社交媒体各持一方数据，无法直接关联。数据管道需要解决：

3. 业务应用场景

业务背景：某母婴 DTC 品牌月均广告支出 ¥85 万，分配在 Meta（40%）、TikTok（35%）、Amazon Sponsored（25%）。末次点击归因显示 Amazon 广告 ROAS=8.2（最高），团队准备大幅增加 Amazon 投入，削减 TikTok。

业务决策校正： - 维持 TikTok 预算（因果归因下增量最大） - 削减 Amazon Sponsored 20%（高 ROAS 主要来自已有意向用户） - 预计年度广告效率提升：节省 ¥12 万 + 增量 GMV +¥65 万

业务背景：品牌在小红书投放 KOL 种草内容（月均 ¥15 万），但小红书没有标准转化 API，无法直接归因到 DTC 站销售。

4. 输入数据要求

• Meta: `facebook-business` SDK → `AdInsights` API
• TikTok: TikTok for Business API → `/reports/integrated/get/`
• Amazon: Amazon Attribution API → `attributionReports`
• Shopify: Webhook → `orders/create` 事件

5. 输出结果

• Meta: `facebook-business` SDK → `AdInsights` API
• TikTok: TikTok for Business API → `/reports/integrated/get/`
• Amazon: Amazon Attribution API → `attributionReports`
• Shopify: Webhook → `orders/create` 事件

6. 业务价值 / ROI

12 万

7. 代码模板

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

"""
Marketing Attribution Data Pipeline
整合 UniMTA (身份拼接) + StreamAttrib (实时聚合) + CausalAttrib (因果归因)
使用 mock 数据，可直接运行
"""

import re
import hashlib
import random
import numpy as np
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Tuple, Set
from datetime import datetime, timedelta
from collections import defaultdict


# ── 数据结构 ────────────────────────────────────────────────────────────

@dataclass
class AdEvent:
    """广告事件（曝光/点击/转化）"""
    event_id: str
    channel: str          # meta / tiktok / amazon / xiaohongshu
    event_type: str       # impression / click / conversion
    user_id_raw: str      # 各平台原始 ID
    email_hash: str       # SHA256 邮箱（若有）
    ip_fingerprint: str   # IP + UA fingerprint
    timestamp: datetime
    value: float          # 广告花费 / 转化金额（视 event_type）
    creative_id: str      # 广告创意 ID


@dataclass
class UserJourney:
    """单用户完整触点旅程"""
    unified_user_id: str
    touchpoints: List[AdEvent]
    converted: bool
    conversion_value: float
    conversion_time: Optional[datetime]


@dataclass
class ChannelAttribution:
    """渠道归因结果"""
    channel: str
    last_click_credit: float    # 末次点击归因
    linear_credit: float        # 线性归因
    causal_cate: float          # 因果增量 CATE
    impression_count: int
    click_count: int
    attributed_gmv: float


# ── UniMTA：身份拼接 ──────────────────────────────────────────────────────

class IdentityGraph:
    """
    跨平台用户身份图谱
    三级匹配：邮箱 hash（确定性）→ IP fingerprint（概率）→ 行为模型

8. 论文来源

• 2407.14521
• 2411.16238
• 2501.09917