Account Association Risk Detection — 电商多账户关联风险检测
Skill-Account-Association-Risk-Detection · 19-风控反欺诈
experimentdata_collectionfraud_detection广告与投放客服与VOC数据采集与治理风控与合规WF-B 广告优化WF-C 客服分诊WF-D 选品扫描WF-E Review监控WF-G Listing内容优化WF-I 智能体工程WF-K 全域风险防御
年化 ROI50-500 万元
实现难度⭐⭐☆☆☆
业务优先级⭐⭐⭐⭐⭐
业务视角
适用角色运营负责人 / 合规负责人 · 品牌负责人 · CEO
适用平台Amazon 刷评检测与举报 · TikTok Shop 刷单识别 · 竞品 Listing 攻击溯源
什么情况下用竞品刷单刷好评，自己的 BSR 和评分被打压；账号/ASIN 被恶意投诉删除；店铺有异常订单不确定是真实买家
成功是什么样的识别过滤刷评/恶意竞争行为，账号风险提前预警，维权有数据证据，降低封号风险
业务痛点
1. 解决的问题
新员工设备、共享 IP、旧供应商账号形成隐性关联，被 Amazon 认定即全部封禁损失 50-500 万元——多维信号图分析（银行卡/设备/IP/行为）提前 30 天识别关联风险，建议隔离操作防止连坐
2. 核心算法逻辑
核心思想：Amazon 严禁同一主体运营多个卖家账号，一旦被认定关联即可全部封禁（连坐）。但很多卖家因为历史原因（员工离职带走账号、IP 共享、银行卡关联）不知道自己的账号已经"被关联"。账号关联风险检测通过多维信号图分析，提前识别关联风险并指导隔离操作。
3. 业务应用场景
- 业务问题：某母婴品牌运营团队扩张，新员工用自己的设备登录管理账号，同时发现前供应商也在 Amazon 上开了店。一次 IP 关联就可能触发 Amazon 的关联审查，导致主账号封禁损失 GMV 数百万元。 - 数据要求：账号登录记录（IP/设备/时间）、关联邮箱/银行信息、已知关联方账号列表。 - 预期产出： - 账号关联图（可视化每个账号的连接链路） - 各关联路径的风险等级（红/橙/黄/绿） - 具体隔离建议（"建议更换登录设备"/"需要独立银行账户"） - 防御操作： - 高风险：立即分离资金链路 + 更换登录设备 - 中风险：使用 VPN 隔离 IP + 监控登录行为 - 定期检查
4. 输入数据要求
请查看原始代码模板获取输入规格。
5. 输出结果
请查看原始代码模板获取输出规格。
6. 业务价值 / ROI
- ROI 预估：账号封禁损失 50-500 万元，提前预防成本极低（主要是数据整理），ROI 极高
- 实施难度：⭐⭐☆☆☆（低，主要是数据整理 + 图算法）
- 优先级：⭐⭐⭐⭐⭐（账号是跨境卖家最核心资产，关联封禁是毁灭性风险）
- 评估依据：Amazon 关联封禁真实案例普遍存在，多维信号图分析是业界标准方法
7. 代码模板
代码块数量：2 · 路径：未检测到
from dataclasses import dataclass, field
from typing import List, Dict, Set, Optional
@dataclass
class AccountNode:
account_id: str
email: str
bank_card_last4: str
ip_ranges: List[str]
device_ids: List[str]
address_hash: Optional[str] = None
category: Optional[str] = None
def extract_shared_signals(accounts: List[AccountNode]) -> Dict[str, List]:
shared = {"email": {}, "bank": {}, "ip": {}, "device": {}, "address": {}}
for acc in accounts:
shared["email"].setdefault(acc.email, []).append(acc.account_id)
shared["bank"].setdefault(acc.bank_card_last4, []).append(acc.account_id)
if acc.address_hash:
shared["address"].setdefault(acc.address_hash, []).append(acc.account_id)
for ip in acc.ip_ranges:
shared["ip"].setdefault(ip, []).append(acc.account_id)
for dev in acc.device_ids:
shared["device"].setdefault(dev, []).append(acc.account_id)
return {k: {v: ids for v, ids in d.items() if len(ids) > 1} for k, d in shared.items()}
def compute_association_risk(account_id: str, all_accounts: List[AccountNode]) -> Dict:
target = next((a for a in all_accounts if a.account_id == account_id), None)
if not target:
return {"account_id": account_id, "error": "not found"}
shared = extract_shared_signals(all_accounts)
WEIGHTS = {"email": 0.35, "bank": 0.30, "device": 0.20, "ip": 0.10, "address": 0.05}
risk_score = 0.0
associations = []
for signal_type, weight in WEIGHTS.items():
for signal_val, account_ids in shared.get(signal_type, {}).items():
if account_id in account_ids:
other_ids = [aid for aid in account_ids if aid != account_id]
risk_score += weight
associations.append({"type": signal_type, "shared_value": signal_val[:8] + "***",
"linked_accounts": other_ids, "weight": weight})
risk_score = min(1.0, risk_score)
if risk_score >= 0.7:
level = "🔴 高风险"
action = "立即分离资金链路和登录设备，检查供应商关系"
elif risk_score >= 0.4:
level = "🟡 中风险"
action = "使用独立网络登录，监控关联账号行为"
else:
level = "🟢 低风险"
action = "定期例行检查即可"
return {"account_id": account_id, "risk_score": round(risk_score, 3),
"risk_level": level, "action": action, "associations": associations}
accounts = [
AccountNode("BRAND_MAIN", "main@brand.com", "1234", ["192.168.1.x", "10.0.0.x"],
["DEVICE_A", "DEVICE_B"], "ADDR_001", "baby"),
AccountNode("EMP_OLD", "old_emp@gmail.com", "5678", ["192.168.1.x"],
["DEVICE_C"], None, "baby"),
AccountNode("SUPPLIER_STORE", "supplier@supplier.com", "9012", ["172.16.x.x"],8. 论文来源
未自动抽取；请查看原始 Skill 卡片。