反欺诈里那些抓不到的团伙，交给图神经网络：PyG GraphSAGE 交易反欺诈实战-开发者社区

反欺诈里那些抓不到的团伙，交给图神经网络：PyG GraphSAGE 交易反欺诈实战

【免费下载链接】pytorch_geometricGraph Neural Network Library for PyTorch项目地址: https://gitcode.com/GitHub_Trending/py/pytorch_geometric

用 PyTorch Geometric（PyG）的HeteroData把信用卡、商户、交易、设备装进一张异构图，用 GraphSAGE 按节点做欺诈分类，再用NeighborLoader小批量训练，一路走到分布式扩展。本文走一遍从建模到落地的完整链路。

速览：本文讲如何用 PyG 的HeteroData+SAGEConv（GraphSAGE）做交易节点级反欺诈分类，用NeighborLoader小批量采样处理百万级交易图，并用AUC对到风控口径。适合读者：有 Python/PyTorch 基础、想给交易/资金网络做反欺诈或团伙挖掘的风控与算法工程师。

一笔笔欺诈怎么装进一张交易异构图（HeteroData）

单看一笔交易，金额、渠道、时段都"正常"，可欺诈从来不是孤立事件——同一批黑产会复用几张卡、几台设备、几个商户批量刷单。这种"团伙信号"藏在关系里，散在多张表里根本连不起来；一旦拆成 SQL 视图分别分析，传导路径就断了。图的价值，就是把"卡—交易—商户—设备"画在一张纸上，让"共享同一台设备的交易"这类证据能沿着边传出去。

PyG 用HeteroData表达这种多实体多关系结构，节点和边都用「类型 + 索引」写：

import torch from torch_geometric.data import HeteroData data = HeteroData() # 节点特征：从风控数仓取现成字段，做 z-score 归一化 data['card'].x = torch.randn(num_cards, 16) # 发卡行/地区/额度/账户年龄 data['merchant'].x = torch.randn(num_merchants, 16) # MCC行业/地区/交易规模 data['device'].x = torch.randn(num_devices, 8) # 机型/IP/设备指纹 data['transaction'].x = torch.randn(num_tx, 16) # 金额/时段/渠道/是否退款 # 边：2xE 的 index，第 0 行起点、第 1 行终点，只保留有语义的几条 data['card', 'made', 'transaction'].edge_index = card_tx_idx data['transaction', 'at', 'merchant'].edge_index = tx_merchant_idx data['device', 'used_in', 'transaction'].edge_index = device_tx_idx

节点特征直接取风控数仓字段归一化；纯关系型节点（如只有 ID 没有属性的）可以先用独热 ID 顶上去。边不用穷举所有实体组合，只留"卡做了交易""交易发生在商户""设备参与了交易"这几条主干。

注意：边方向别乱写——made/at/used_in的起点终点要和后续取特征、做归一化对齐，方向写反会让消息从错误方向传播，团伙信号被稀释。

欺诈节点怎么编码：GraphSAGE + to_hetero 自动展开

反欺诈的落地点是节点级二分类：给每笔交易打一个欺诈概率，而不是预测某条边的数值。思路：先用 GraphSAGE 把每笔交易聚合它周围卡、商户、设备的信号编码成向量，再过一个分类头输出 logit。

这里不手写每类节点的卷积，而是用to_hetero把一个同质SAGEConv编码器按元数据自动展开成异构图模型——每类边独立一套参数。仓库示例examples/hetero/to_hetero_mag.py做的就是这件事：它拿 MAG 异构数据，SAGEConv((-1, -1), 64)建编码器再to_hetero，本文把它改到"交易反欺诈"上。

from torch_geometric.nn import SAGEConv, to_hetero class FraudDetector(torch.nn.Module): def __init__(self): super().__init__() # -1 表示输入维度由数据自动推断，to_hetero 才能按每类边配独立参数 self.conv1 = SAGEConv((-1, -1), 64) self.conv2 = SAGEConv(64, 64) self.encoder = to_hetero(self, data.metadata(), aggr='sum') self.head = torch.nn.Linear(64, 1) # 只给交易节点接分类头 def forward(self, x_dict, edge_index_dict): z = self.encoder(x_dict, edge_index_dict) # 各节点类型都有嵌入 return self.head(z['transaction']).view(-1) # 每笔交易的欺诈 logit

SAGEConv((-1, -1), ...)里的-1是关键：让输入维度由真实数据推断，to_hetero才能给每种边类型生成独立的参数矩阵。分类头只挂在transaction上，卡/商户/设备只做特征传播源、不直接出分。

注意：z['transaction']返回的是本批次所有被采样到的交易节点，前batch_size个才是这次要打分/反传的种子交易，切片见下一节；漏切会拿邻居的 label 去算 loss，训练直接乱掉。

大交易图怎么小批量训练：NeighborLoader 采样 + 类别不平衡

真实交易图动辄上亿条交易、上千万节点，全图一次前向内存根本扛不住。PyG 的NeighborLoader做 mini-batch 邻居采样：每次只取一小撮种子交易，往回采两跳邻居，用这个"采样子图"训练。仓库examples/multi_gpu/mag240m_graphsage.py就是用 GraphSAGE 在亿级 MAG 图上做采样的路子，本文单机版直接复用它的参数风格。

交易欺诈是典型的极不平衡问题：欺诈占比常在千分之几。直接上交叉熵，模型"全判正常"就能拿到 99.9% 的准确率，AUC 却是垃圾。用pos_weight把负样本权重拉起来：

from torch_geometric.loader import NeighborLoader import torch.nn.functional as F train_loader = NeighborLoader( data, num_neighbors=[10, 5], # 两跳采样：10 -> 5 batch_size=1024, input_nodes=('transaction', train_mask), # 只从训练集交易节点起步 shuffle=True, ) optimizer = torch.optim.Adam(model.parameters(), lr=0.01) pos_weight = torch.tensor([num_neg / max(num_pos, 1)]) # 压制不平衡 def train_epoch(): model.train() total = 0. for batch in train_loader: optimizer.zero_grad() out = model(batch.x_dict, batch.edge_index_dict) n = batch['transaction'].batch_size # 前 n 个是种子交易 loss = F.binary_cross_entropy_with_logits( out[:n], batch['transaction'].y[:n], pos_weight=pos_weight) loss.backward(); optimizer.step() total += loss.item() return total / len(train_loader)

注意：input_nodes一定只传train_mask，且num_neighbors的跳数要和模型卷积层数对齐——这里两层卷积配两跳[10, 5]；若层数比跳数多，要设directed=False补全采样节点间所有边，否则深层卷积取不到够用的邻域。

欺诈评估怎么对得上风控口径：AUC 换算成漏损

风控不关心你 loss 多低，只关心"我按风险分排序去拦截，能抓住多少欺诈、误伤多少正常"。最稳的口径是AUC+ 固定误报率下的召回，而不是准确率。

AUC可以直接读：AUC=0.91，意味着随机取一笔欺诈交易和一笔正常交易，模型把欺诈分排在前面的概率是 91%。换算到钱：

from sklearn.metrics import roc_auc_score import numpy as np @torch.no_grad() def eval_auc(loader): model.eval() logits, labels = [], [] for batch in loader: out = model(batch.x_dict, batch.edge_index_dict) n = batch['transaction'].batch_size logits.append(out[:n].cpu()) labels.append(batch['transaction'].y[:n].cpu()) logits = torch.cat(logits).numpy() return roc_auc_score(labels, 1 / (1 + np.exp(-logits))) # 转概率再算 AUC

拿到 AUC 后按风控口径落地：在同样 1% 误报率下，AUC 从 0.83 提到 0.91，欺诈召回往往从 62% 提到 79%。假设月欺诈总额 500 万，召回每提升 1 个点 ≈ 每月少漏 5 万，这个数拿去和现在"凭规则 + 人工复核"的拦截率比，就能算出模型值不值得上。

⚠️ 这里有个反欺诈特有的坑：训练/验证/测试必须按交易时间切分，且不能随机切。随机切会把"同一台设备/同一个商户"的欺诈交易同时塞进训练和测试，评估 AUC 虚高，一上线就现原形——欺诈是时序演化的，未来的交易不能当历史特征用。

交易图大到装不进内存：分布式邻居采样

单机采样的极限，是特征矩阵和邻接表装不进内存。当节点过千万、边过亿，PyG 的torch_geometric/distributed/提供两级扩展：

离线切图：Partitioner（distributed/partition.py）把节点、特征、边按分片落盘，每个partN/下是切好的图与特征；
在线采样：DistNeighborLoader（distributed/dist_neighbor_loader.py）绑定本机分片，本地邻居直接读，跨分片邻居走 RPC 从远端拉。

from torch_geometric.distributed import DistNeighborLoader # 先离线 Partitioner 切图落盘；训练时每台机器加载自己的 part 分片 loader = DistNeighborLoader( partition_root='dist_graph/', # 切图输出目录 edge_type=('card', 'made', 'transaction'), num_neighbors=[10, 5], batch_size=1024, )

采样开销从"全图"降到"本机分片 + 一跳远程"，训练吞吐随机器数近似线性扩展。对交易这种边数远超单机的图，这一步基本是必选项。

注意：切图会引入跨分片的采样延迟，num_neighbors和batch_size建议比单机调小一档，并把采样放子进程（num_workers> 0）与训练重叠，否则 RPC 等待会吃掉吞吐。

要点回顾与下一步

要点回顾：

建模：HeteroData把卡/商户/设备/交易装成异构图，边用 2×E index 只留主干语义
编码：SAGEConv((-1, -1), ...)维度自推断，to_hetero按元数据展开成独立参数的异构图模型
采样：NeighborLoader+input_nodes=train_mask做 mini-batch 两跳采样，num_neighbors跳数对齐层数
不平衡：binary_cross_entropy_with_logits配pos_weight，别用裸交叉熵
评估：AUC 对齐风控口径，按交易时间切分防实体泄漏
扩展：Partitioner离线切图 +DistNeighborLoader在线跨机采样