100天跟着CP学PostgreSQL+AI,第23天 :反欺诈系统：数据库 + AI 如何识别交易异常？

Transaction = (payer, payee, {amount, time, location, device_info})

CREATE TABLE transactions (
    tx_id BIGINT PRIMARY KEY,
    payer_id TEXT,
    payee_id TEXT,
    amount NUMERIC(16,2),
    tx_time TIMESTAMP,
    features JSONB,  -- 设备指纹、IP地址等
    adj_list JSONB  -- 邻接节点列表，格式：{"nodes":["A","B","C"],"edges":[["A","B"],["B","C"]]}
);
安装 pg_pathman 扩展：
CREATE EXTENSION pg_pathman;
构建子图索引（以 3 跳范围内子图为例）：
SELECT create_pathman_index(
    'transactions', 
    'subgraph_index', 
    'adj_list', 
    'tx_id', 
    '3'  -- 最大路径长度
);
（二）动态子图提取算法
import psycopg2
from pg_pathman import PathmanClient
def extract_subgraph(start_node, depth=3):
    conn = psycopg2.connect("dbname=fraud_db user=postgres")
    client = PathmanClient(conn)
    
    query = f"""
    SELECT collect_subgraph(tx_id) 
    FROM transactions 
    WHERE find_paths(payer_id, '{start_node}', {depth})
    """
    subgraph_data = client.execute(query)
    return subgraph_data  # 返回格式：(node_list, edge_list, feature_dict)

from torch_geometric.data import HeteroData
data = HeteroData()
data['account'].x = account_features  # 账户基础特征
data['device'].x = device_features    # 设备指纹特征
data['location'].x = location_features# 地理位置编码
# 添加异质边
data['account', 'uses', 'device'].edge_index = account_device_edges
data['account', 'located_at', 'location'].edge_index = account_location_edges
data['device', 'used_at', 'location'].edge_index = device_location_edges
图数据增强（随机子图扰动）：
def augment_graph(data, drop_rate=0.2):
    num_nodes = data.num_nodes
    drop_mask = torch.rand(num_nodes) > drop_rate
    data.x = data.x[drop_mask]
    # 同步更新边索引（保留两端节点均未被删除的边）
    edge_mask = (drop_mask[data.edge_index[0]] & drop_mask[data.edge_index[1]])
    data.edge_index = data.edge_index[:, edge_mask]
    return data
（二）GNN 模型架构（以 GAT+GraphSAGE 为例）
import torch
import torch.nn.functional as F
from torch_geometric.nn import GATConv, GraphSAGEConv
class FraudGNN(torch.nn.Module):
    def __init__(self, hidden_dim=128):
        super().__init__()
        self.gat_conv = GATConv(-1, hidden_dim, heads=4)  # 自动适配异质图节点特征
        self.sage_conv = GraphSAGEConv((-1, -1), hidden_dim)
        self.lstm = torch.nn.LSTM(hidden_dim, hidden_dim//2, bidirectional=True)
        self.classifier = torch.nn.Linear(hidden_dim, 2)  # 0正常交易，1欺诈交易
    def forward(self, data):
        x, edge_index = data.x, data.edge_index
        x = self.gat_conv(x, edge_index)
        x = F.relu(x)
        x = self.sage_conv((x, x), edge_index)  # 双向传播
        x, _ = self.lstm(x.unsqueeze(0))  # 处理时间序列特征
        return self.classifier(x.squeeze(0))

class FocalLoss(torch.nn.Module):
    def __init__(self, alpha=0.8, gamma=2):
        super().__init__()
        self.alpha = alpha
        self.gamma = gamma
    
    def forward(self, logits, labels):
        ce_loss = F.cross_entropy(logits, labels)
        pt = torch.exp(-ce_loss)
        focal_loss = self.alpha * (1-pt)**self.gamma * ce_loss
        return focal_loss

from torch_geometric.loader import DataLoader
from torch_geometric.distributed import ClusterData, ClusterLoader
# 划分16个子图分区
cluster_data = ClusterData(data, num_parts=16, recursive=False)
loader = ClusterLoader(cluster_data, batch_size=32, shuffle=True)
# 使用混合精度训练
scaler = torch.cuda.amp.GradScaler()
for batch in loader:
    with torch.cuda.amp.autocast():
        out = model(batch.to(device))
        loss = focal_loss(out, batch.y)
    scaler.scale(loss).backward()
    scaler.step(optimizer)
    scaler.update()