4. 图谱数据构建：从原始交易数据构建图谱、节点特征提取、边权重计算

图谱构建是反欺诈图谱分析的基础环节。原始交易数据通常是扁平的表格结构，需要经过实体抽取、关系映射、特征工程三个步骤，才能转化为可供图算法使用的节点-边结构。本节将详细拆解这一过程，并给出可落地的 Python 实现方案。

4.1 实体抽取与节点定义

从原始交易数据中识别出具有反欺诈意义的实体，每个实体对应图中的一个节点。常见的反欺诈实体包括：

import pandas as pd
import hashlib

def extract_nodes(transactions: pd.DataFrame) -> dict:
    """
    从交易数据中提取所有实体节点，返回 {node_type: set(node_ids)} 字典。
    """
    nodes = {
        'account': set(),
        'device': set(),
        'ip': set(),
        'card': set(),
        'phone': set()
    }
    
    # 账户节点
    nodes['account'].update(transactions['payer_id'].dropna().apply(lambda x: f"account_{x}"))
    nodes['account'].update(transactions['payee_id'].dropna().apply(lambda x: f"account_{x}"))
    
    # 设备节点（对设备ID做哈希脱敏）
    for dev in transactions['device_id'].dropna().unique():
        dev_hash = hashlib.sha256(str(dev).encode()).hexdigest()[:12]
        nodes['device'].add(f"device_{dev_hash}")
    
    # IP节点
    nodes['ip'].update(transactions['client_ip'].dropna().apply(lambda x: f"ip_{x}"))
    
    # 银行卡节点（掩码处理）
    for card in transactions['card_no'].dropna().unique():
        masked = str(card)[:4] + '****' + str(card)[-4:]
        nodes['card'].add(f"card_{masked}")
    
    # 手机号节点
    nodes['phone'].update(transactions['phone'].dropna().apply(lambda x: f"phone_{x}"))
    
    return nodes

4.2 关系映射与边定义

实体之间的交互行为构成图的边。每条边需要明确：源节点、目标节点、关系类型、时间戳、交易金额等属性。反欺诈图谱中常见的边类型包括：

• 交易边：payer → payee，属性包含金额、时间、交易类型
• 设备共用边：account ↔ device，表示账户登录/使用某设备
• IP关联边：account ↔ ip，表示账户从某IP发起交易
• 资金循环边：account → account，表示短时间内资金回流
• 社交关系边：account ↔ phone，表示账户绑定了某手机号

def build_edges(transactions: pd.DataFrame) -> pd.DataFrame:
    """
    从交易数据构建边DataFrame，包含源节点、目标节点、边类型、权重基础字段。
    """
    edges = []
    
    # 交易边
    for _, row in transactions.iterrows():
        edges.append({
            'source': f"account_{row['payer_id']}",
            'target': f"account_{row['payee_id']}",
            'edge_type': 'transfer',
            'amount': row['amount'],
            'timestamp': row['timestamp'],
            'txn_id': row['txn_id']
        })
        
        # 设备关联边
        if pd.notna(row.get('device_id')):
            dev_hash = hashlib.sha256(str(row['device_id']).encode()).hexdigest()[:12]
            edges.append({
                'source': f"account_{row['payer_id']}",
                'target': f"device_{dev_hash}",
                'edge_type': 'uses_device',
                'amount': 0,
                'timestamp': row['timestamp'],
                'txn_id': row['txn_id']
            })
        
        # IP关联边
        if pd.notna(row.get('client_ip')):
            edges.append({
                'source': f"account_{row['payer_id']}",
                'target': f"ip_{row['client_ip']}",
                'edge_type': 'from_ip',
                'amount': 0,
                'timestamp': row['timestamp'],
                'txn_id': row['txn_id']
            })
    
    return pd.DataFrame(edges)

4.3 节点特征提取

节点特征分为静态属性特征和动态行为特征。静态特征来自注册信息、实名认证等级等；动态特征需要从交易历史中聚合计算。

def extract_node_features(transactions: pd.DataFrame, nodes: dict) -> pd.DataFrame:
    """
    为每个节点提取特征，返回 node_id -> feature_vector 的映射表。
    """
    features = []
    
    # 按账户聚合交易行为
    account_txns = transactions.groupby('payer_id')
    
    for node_id in nodes['account']:
        account_id = node_id.replace('account_', '')
        if account_id in account_txns.groups:
            txns = account_txns.get_group(account_id)
            features.append({
                'node_id': node_id,
                'node_type': 'account',
                'txn_count_24h': (txns['timestamp'] > (pd.Timestamp.now() - pd.Timedelta(hours=24))).sum(),
                'avg_amount_7d': txns[txns['timestamp'] > (pd.Timestamp.now() - pd.Timedelta(days=7))]['amount'].mean(),
                'device_switch_24h': txns[txns['timestamp'] > (pd.Timestamp.now() - pd.Timedelta(hours=24))]['device_id'].nunique(),
                'ip_city_count': txns[txns['timestamp'] > (pd.Timestamp.now() - pd.Timedelta(hours=24))]['ip_city'].nunique()
            })
        else:
            # 只有收款记录的账户
            features.append({
                'node_id': node_id,
                'node_type': 'account',
                'txn_count_24h': 0,
                'avg_amount_7d': 0,
                'device_switch_24h': 0,
                'ip_city_count': 0
            })
    
    return pd.DataFrame(features)

4.4 边权重计算

边权重反映两个节点之间关联的强度或异常程度。在反欺诈场景中，权重越高通常意味着关系越可疑。常用的权重计算策略包括：

• 交易频次权重：w = log(1 + 交易次数)，高频交易获得更高权重
• 金额异常权重：w = (交易金额 - 均值) / 标准差，偏离正常范围越大权重越高
• 时间集中度权重：w = 1 / (时间间隔标准差 + 1)，短时间内密集交易权重更高
• 设备/IP共享权重：w = 1 + 共享设备数，多账户共用同一设备时权重增加
• 综合权重：w = α·频次 + β·金额异常 + γ·时间集中度

def compute_edge_weights(edges: pd.DataFrame, transactions: pd.DataFrame) -> pd.DataFrame:
    """
    为每条边计算综合权重，基于频次、金额异常度和时间集中度。
    """
    # 按源-目标-类型聚合
    edge_groups = edges.groupby(['source', 'target', 'edge_type'])
    
    weighted_edges = []
    for (src, tgt, etype), group in edge_groups:
        txn_ids = group['txn_id'].tolist()
        related_txns = transactions[transactions['txn_id'].isin(txn_ids)]
        
        # 频次权重
        freq_weight = np.log1p(len(group))
        
        # 金额异常权重（Z-score）
        amounts = related_txns['amount'].values
        if len(amounts) > 1:
            amount_z = np.abs((amounts.mean() - amounts.mean()) / (amounts.std() + 1e-8))
            amount_weight = np.mean(amount_z)
        else:
            amount_weight = 0.5  # 单笔交易默认中等异常度
        
        # 时间集中度（时间戳标准差越小越集中）
        timestamps = related_txns['timestamp'].values.astype(np.int64)
        if len(timestamps) > 1:
            time_std = timestamps.std()
            time_weight = 1.0 / (time_std / 1e9 + 1)  # 转换为秒
        else:
            time_weight = 0.3
        
        # 综合权重（α=0.4, β=0.4, γ=0.2）
        combined_weight = 0.4 * freq_weight + 0.4 * amount_weight + 0.2 * time_weight
        
        weighted_edges.append({
            'source': src,
            'target': tgt,
            'edge_type': etype,
            'weight': round(combined_weight, 4),
            'txn_count': len(group)
        })
    
    return pd.DataFrame(weighted_edges)

4.5 图谱构建流程总览

下图展示了从原始交易数据到最终加权图谱的完整构建流程：

4.6 图谱质量验证

构建完成后，需要对图谱进行质量检查，确保数据可用性：

def validate_graph(G, node_features: pd.DataFrame):
    """
    对构建好的图进行质量验证，输出关键指标。
    """
    import networkx as nx
    
    print(f"节点总数: {G.number_of_nodes()}")
    print(f"边总数: {G.number_of_edges()}")
    
    # 孤立节点
    isolated = list(nx.isolates(G))
    print(f"孤立节点数: {len(isolated)}, 占比: {len(isolated)/G.number_of_nodes():.2%}")
    
    # 图密度
    density = nx.density(G)
    print(f"图密度: {density:.4f}")
    
    # 边权重分布
    weights = [d['weight'] for _, _, d in G.edges(data=True) if 'weight' in d]
    if weights:
        print(f"边权重 - 均值: {np.mean(weights):.2f}, 标准差: {np.std(weights):.2f}")
        print(f"边权重 - 最小值: {min(weights):.2f}, 最大值: {max(weights):.2f}")
    
    # 特征缺失率
    if node_features is not None:
        missing_rate = node_features.isnull().sum() / len(node_features)
        print(f"\n特征缺失率:\n{missing_rate[missing_rate > 0]}")
    
    return {
        'n_nodes': G.number_of_nodes(),
        'n_edges': G.number_of_edges(),
        'isolated_ratio': len(isolated) / G.number_of_nodes(),
        'density': density
    }