P paper2skillsPlaybook
AI 路线图 →

PromoGuardian — 促销欺诈多关系图神经网络检测

Skill-PromoGuardian-Promotion-Fraud-GNN · 19-风控反欺诈

causalexperimentknowledge_graphfraud_detectionpricing知识图谱与RAG定价与利润风控与合规WF-D 选品扫描WF-F 动态定价WF-K 全域风险防御
年化 ROI$144,000-$672,000
实现难度⭐⭐⭐☆☆
业务视角
适用角色运营负责人 / 合规负责人 · 品牌负责人 · CEO
适用平台Amazon 刷评检测与举报 · TikTok Shop 刷单识别 · 竞品 Listing 攻击溯源
什么情况下用竞品刷单刷好评,自己的 BSR 和评分被打压;账号/ASIN 被恶意投诉删除;店铺有异常订单不确定是真实买家
成功是什么样的识别过滤刷评/恶意竞争行为,账号风险提前预警,维权有数据证据,降低封号风险
业务痛点
竞品刷评打压我们我们的好评被恶意举报删除不知道差评是真实的还是恶意的如何证明竞品恶意行为

1. 解决的问题

促销运营负责人面临大促期优惠券被批量套利日损数万元——多关系图神经网络同时捕获设备/地址/时序三维欺诈信号,促销套利损失减少60-70%,年化节省100-500万元

2. 核心算法逻辑

传统欺诈检测把用户孤立看待,但促销欺诈通常是群体协作行为:一个主账号带动多个小号组成"刷单环",利用促销规则批量套利。PromoGuardian 的核心洞察是:欺诈行为在关系图上留下的结构信号远比单节点特征更强。

3. 业务应用场景

业务问题:大促期间针对新用户的 $10 优惠券被批量注册账号套取。黑产团伙注册50个小号,同一设备/IP 轮流使用优惠券购买低价商品后立即退货,净套取优惠金额。估算损失:$10/单 × 200单/天 × 30天 = $60,000/月。

PromoGuardian 处理: - 构建大促期间全用户-设备-地址-商品 4 类关系图 - GNN 在2轮消息传递后识别出「设备共享度 > 3 且地址重叠 > 2」的强连通集团 - 对集团内节点提升欺诈概率阈值,触发人工复查

数据要求: - 用户注册信息(设备 fingerprint、注册 IP、邮箱域名) - 订单明细(购买商品 ASIN、金额、时间戳) - 退货记录(退货率 > 80% 为强信号) - 优惠券使用记录(同批次优惠券在同设备被使用 ≥ 2次)

4. 输入数据要求

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

5. 输出结果

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

6. 业务价值 / ROI

  • ROI 预估
  • 促销欺诈典型损失:$20,000-$80,000/月(大促期间)
  • 检测后减少 60-70%:$12,000-$56,000/月
  • 年化:$144,000-$672,000(人民币约 100-500 万元)
  • 实施难度:⭐⭐⭐☆☆
  • 核心挑战:构建用户-设备-地址关系图(需要数据脱敏处理)

7. 代码模板

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

"""
PromoGuardian - 促销欺诈多关系图神经网络检测
基于多关系异构图 + GNN 的欺诈用户识别

依赖: numpy, torch (可选替换为纯numpy简化版)
"""

import numpy as np
from collections import defaultdict
from typing import Dict, List, Tuple, Set

# ─────────────────────────────────────────────
# 多关系图构建
# ─────────────────────────────────────────────

class MultiRelationGraph:
    """构建促销欺诈检测所需的多关系用户图"""
    
    RELATION_TYPES = ['device', 'address', 'product', 'timing']
    
    def __init__(self, time_window_hours: int = 24):
        self.time_window = time_window_hours * 3600
        self.edges: Dict[str, List[Tuple]] = {r: [] for r in self.RELATION_TYPES}
        self.node_features: Dict[str, np.ndarray] = {}
    
    def add_orders(self, orders: List[dict]):
        """
        订单列表格式:
        [{'user_id': str, 'device_id': str, 'ip': str, 
          'address_hash': str, 'product_id': str, 
          'timestamp': int, 'coupon_used': bool, 'return_rate': float}]
        """
        # 建立各关系的倒排索引
        device_map = defaultdict(list)
        address_map = defaultdict(list)
        product_map = defaultdict(list)
        
        for o in orders:
            uid = o['user_id']
            device_map[o['device_id']].append(uid)
            device_map[o['ip']].append(uid)
            address_map[o['address_hash']].append(uid)
            product_map[o['product_id']].append(uid)
            
            # 构建用户特征向量
            self.node_features[uid] = np.array([
                o.get('return_rate', 0),
                1.0 if o.get('coupon_used') else 0.0,
                min(o.get('order_count', 1) / 10.0, 1.0),
                min(o.get('account_age_days', 0) / 365.0, 1.0),
            ])
        
        # 建立边:共享设备/IP
        for device, users in device_map.items():
            users = list(set(users))
            for i in range(len(users)):
                for j in range(i+1, len(users)):
                    self.edges['device'].append((users[i], users[j]))
        
        # 建立边:共享收货地址

8. 论文来源

  • 2510.12652