DS-DGA-GCN — 动态图假评论群组检测：冷启动新品防刷评

1. 解决的问题

核心思想：在"产品 → 评论 → 评论者"三方动态异构图上检测刷评团伙群组。不看单条评论文本质量，而是看评论者之间的网络行为模式——真实用户构成稀疏随机网络，刷评团伙则共现密集、行为高度同步。

2. 核心算法逻辑

核心思想：在"产品 → 评论 → 评论者"三方动态异构图上检测刷评团伙群组。不看单条评论文本质量，而是看评论者之间的网络行为模式——真实用户构成稀疏随机网络，刷评团伙则共现密集、行为高度同步。

3. 业务应用场景

业务问题：新款奶瓶（ASIN B0XXXX）上架后 6 小时内突现 38 条 5 星评论，评论者账号注册时间集中在近 7 天、互相之间都购买过同一批产品。肉眼难辨，但网络结构异常明显。

数据要求： | 类型 | 字段 | 来源 | |------|------|------| | 评论 | reviewer_id, product_id, rating, timestamp | Amazon API | | 评论者 | account_age, review_count, verified | 爬取 | | 产品 | asin, category, launch_date | 内部数据 |

预期产出： - 评论者群组分类：正常 / 可疑 / 虚假团伙 - 可疑群组图谱（哪些账号构成一个团伙） - NFS 得分热力图（团伙成员的网络指纹）

4. 输入数据要求

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

5. 输出结果

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

6. 业务价值 / ROI

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

7. 代码模板

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

"""
DS-DGA-GCN: 动态图假评论群组检测
论文: Detecting Fake Reviewer Groups in Dynamic Networks (arXiv 2603.08332)
场景: 母婴新品上架刷评检测 + 选品评论质量过滤
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Dict, List, Tuple
import numpy as np
from collections import defaultdict


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

@dataclass
class Review:
    reviewer_id: str
    product_id: str
    rating: float
    timestamp: float       # Unix 时间戳
    verified: bool = False
    review_text: str = ""


@dataclass
class ReviewerNode:
    reviewer_id: str
    account_age_days: int
    total_review_count: int
    reviews: List[Review] = field(default_factory=list)


# ─── 三方动态图 ──────────────────────────────────────────────────────────────

class ReviewerNetworkGraph:
    """产品-评论-评论者 三方动态图（支持时序边添加）"""

    def __init__(self):
        self.reviewer_nodes: Dict[str, ReviewerNode] = {}
        self.product_ids: set = set()
        # 边: reviewer_id -> set of product_ids（co-review 边）
        self.coview_edges: Dict[str, set] = defaultdict(set)
        # 时序边: (reviewer_id, product_id) -> timestamp
        self.temporal_edges: Dict[Tuple[str, str], List[float]] = defaultdict(list)

    def add_review(self, review: Review, reviewer_node: ReviewerNode) -> None:
        """动态添加一条评论（时序边扩展）"""
        rid = review.reviewer_id
        pid = review.product_id

        self.reviewer_nodes[rid] = reviewer_node
        self.product_ids.add(pid)
        self.coview_edges[rid].add(pid)
        self.temporal_edges[(rid, pid)].append(review.timestamp)
        reviewer_node.reviews.append(review)

    def get_co_reviewers(self, reviewer_id: str) -> List[str]:
        """找出与该评论者评论过同一产品的所有评论者（共现关系）"""
        target_products = self.coview_edges.get(reviewer_id, set())
        co_reviewers = []

8. 论文来源

• 2603.08332