图基跨设备追踪 - 无监督IP-Domain图谱用户拼接

1. 解决的问题

母婴用户在 TikTok 看到吸奶器短视频种草，切换到 Safari 搜索品牌名进独立站下单

2. 核心算法逻辑

跨设备追踪的本质是：在没有用户级关联ID的情况下，用行为相似性判断"两台设备属于同一个人"。GraphTrack 将这一问题建模为异构图上的节点相似度计算：把设备的 IP 访问记录构建成「IPDevice 二部图」，把域名访问记录构建成「DomainDevice 二部图」，然后用随机游走重启（RWwR / Personalized PageRank）在图上扩散相似度信号，最终输出设备对的匹配分数。

3. 业务应用场景

业务问题：母婴用户在 TikTok 看到吸奶器短视频种草，切换到 Safari 搜索品牌名进独立站下单。现有追踪完全断裂——TikTok 报表显示零转化，独立站显示"直接访问"来源。需要无监督匹配这两个设备的浏览记录，恢复 TikTok 的真实转化贡献。

| 字段 | 类型 | 示例 | |------|------|------| | `device_id` | str | "device_TikTok_iPhone_A001" / "device_Safari_Mac_B001" | | `ip_address` | str | "192.168.1.10"（家庭 WiFi，两设备共享） | | `domain` | str | "momcozy.com", "tiktok.com", "google.com" | | `visit_count` | int | 3（该设备访问该IP/域名的次数） | | `timestamp_range`

预期产出： - 设备对匹配分数矩阵：`match_score(TikTok设备, 独立站设备) > 0.6` → 判定为同一用户 - 跨端归因链路：TikTok 曝光设备 → 独立站转化设备 的用户级关联表

4. 输入数据要求

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

5. 输出结果

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

6. 业务价值 / ROI

600-1200 万元

7. 代码模板

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

"""
GraphTrack: 无监督图基跨设备追踪
论文: arXiv:2203.06833 (AsiaCCS 2022)
Binghui Wang et al.

核心算法:
1. 构建 IP-Device 二部图 + Domain-Device 二部图
2. 用 Personalized PageRank (Random Walk with Restart) 计算设备间相似度
3. 双图融合 + 对称匹配 → 输出设备对匹配分数
4. 无监督：无需任何标注设备对

依赖: pip install numpy scipy networkx pandas
"""
from __future__ import annotations

import numpy as np
import pandas as pd
import networkx as nx
from dataclasses import dataclass, field
from typing import Dict, List, Tuple, Optional
from scipy import sparse


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

@dataclass
class BrowsingRecord:
    """设备浏览记录（IP 或 Domain 维度均通用）"""
    device_id: str
    feature: str       # IP 地址 或 域名
    visit_count: int = 1


# ─────────────────────────────────────────────
# 图构建
# ─────────────────────────────────────────────

class DeviceGraph:
    """
    构建 Device-Feature 二部图（Feature = IP 或 Domain）
    节点分两类: device_* 和 feature_*
    边权重 = 访问次数（频率加权）
    """

    def __init__(self, records: List[BrowsingRecord]):
        self.G = nx.Graph()
        self.device_nodes: List[str] = []
        self.feature_nodes: List[str] = []
        self._build(records)

    def _build(self, records: List[BrowsingRecord]):
        device_set = set()
        feature_set = set()

        for r in records:
            d_node = f"device::{r.device_id}"
            f_node = f"feature::{r.feature}"
            device_set.add(d_node)

8. 论文来源

• 2203.06833