📚 课程目录第4章图谱数据加载

4. 图谱数据加载：使用 py2neo 连接 Neo4j、Cypher 查询基础、将图谱数据读入 DataFrame

在反欺诈图谱应用中，告警阈值调优的第一步是高效、稳定地加载图谱数据。本章将系统讲解如何通过 py2neo 连接 Neo4j 图数据库，掌握核心 Cypher 查询语法，并将查询结果无缝转换为 Pandas DataFrame，为后续的阈值统计分析奠定数据基础。

4.1 py2neo 连接 Neo4j：建立数据通道

py2neo 是 Python 生态中最成熟的 Neo4j 客户端库之一，支持 Bolt 协议和 HTTP 协议。连接时需指定数据库 URI、用户名和密码。推荐使用 Graph 对象作为主入口。

  连接配置清单
  协议：bolt://（推荐，二进制高性能）或 http://
主机与端口：默认 bolt://localhost:7687
认证：用户名/密码（默认 neo4j/neo4j）
数据库：可指定目标数据库名称（Enterprise 版）

from py2neo import Graph

# 建立连接（Bolt 协议）
graph = Graph(
    uri="bolt://localhost:7687",
    user="neo4j",
    password="your_password",
    name="neo4j"  # 可选，指定数据库
)

# 验证连接
result = graph.run("RETURN 1 AS num").data()
print("连接成功，返回结果:", result)
# 输出: 连接成功，返回结果: [{'num': 1}]

连接最佳实践：在生产环境中，建议将连接参数配置在环境变量或配置文件中，避免硬编码密码。同时，使用 with 语句管理连接会话，确保资源释放。

4.2 Cypher 查询基础：反欺诈图谱的核心语言

Cypher 是 Neo4j 的声明式图查询语言，其语法直观，擅长表达节点、关系和路径模式。在反欺诈场景中，最常用的模式是匹配实体节点及其关联关系。

4.2.1 核心语法速览

语法元素	示例	说明
MATCH	`MATCH (n:Account)`	匹配所有 Account 标签的节点
RETURN	`RETURN n.id, n.risk_score`	返回节点的属性
WHERE	`WHERE n.risk_score > 0.8`	过滤条件
关系模式	`(a:Account)-[r:TRANSFER_TO]->(b:Account)`	匹配转账关系
WITH	`WITH n, count(*) AS cnt`	管道传递中间结果
LIMIT	`LIMIT 100`	限制返回行数

4.2.2 反欺诈典型查询模式

以下查询提取高风险账户及其最近 30 天内的转账目标，用于构建阈值分析所需的子图。

# 查询高风险账户及其转账关系
cypher_query = """
MATCH (a:Account)-[r:TRANSFER_TO]->(b:Account)
WHERE a.risk_score > 0.7
  AND r.timestamp >= datetime() - duration('P30D')
RETURN a.account_id AS source_id,
       a.risk_score AS source_risk,
       b.account_id AS target_id,
       b.risk_score AS target_risk,
       r.amount AS transfer_amount,
       r.timestamp AS transfer_time
LIMIT 5000
"""

# 执行查询并获取原始结果
raw_results = graph.run(cypher_query).data()
print(f"查询返回 {len(raw_results)} 条关系记录")

性能注意：在大型图谱上执行未加限制的 MATCH 查询可能导致内存溢出。始终优先使用 LIMIT、WHERE 过滤和索引（如对 risk_score、timestamp 创建索引）来约束扫描范围。

4.3 将图谱数据读入 DataFrame：结构化分析的基础

py2neo 的 .data() 方法返回字典列表，可直接传入 Pandas DataFrame 构造函数。这是从图数据过渡到表格分析的关键桥梁。

4.3.1 标准转换流程

import pandas as pd

# 将查询结果直接转为 DataFrame
df_relations = pd.DataFrame(raw_results)

# 查看数据概览
print(df_relations.head())
print(df_relations.info())

# 数据类型优化（反欺诈场景常用）
df_relations['transfer_amount'] = pd.to_numeric(df_relations['transfer_amount'], errors='coerce')
df_relations['transfer_time'] = pd.to_datetime(df_relations['transfer_time'])
df_relations['source_risk'] = df_relations['source_risk'].astype(float)

4.3.2 批量加载节点属性

当需要加载大量节点属性用于阈值分布分析时，建议分批次查询并合并。

# 分批加载所有账户节点属性
batch_size = 10000
offset = 0
all_nodes = []

while True:
    batch_query = f"""
    MATCH (n:Account)
    RETURN n.account_id AS account_id,
           n.risk_score AS risk_score,
           n.create_time AS create_time,
           n.account_type AS account_type
    SKIP {offset} LIMIT {batch_size}
    """
    batch_data = graph.run(batch_query).data()
    if not batch_data:
        break
    all_nodes.extend(batch_data)
    offset += batch_size
    print(f"已加载 {offset} 条节点记录...")

df_nodes = pd.DataFrame(all_nodes)
print(f"节点总数: {len(df_nodes)}")

4.4 图谱数据加载架构

下图展示了从 Neo4j 到 DataFrame 的完整数据加载链路，以及各环节的关键操作。

4.5 实战：加载告警阈值调优所需的核心图谱

在反欺诈阈值调优场景中，通常需要加载以下三类数据：

账户节点属性：account_id、risk_score、account_type、create_time
交易关系：source_id、target_id、amount、timestamp、is_fraud（标签）
设备/IP 关联：device_id、ip_address、关联账户列表

# 完整加载示例：合并节点与关系数据
def load_fraud_graph_data(graph, days_back=30):
    """
    加载反欺诈图谱核心数据，返回节点和关系两个 DataFrame
    """
    # 1. 加载账户节点
    node_query = """
    MATCH (n:Account)
    RETURN n.account_id AS account_id,
           n.risk_score AS risk_score,
           n.account_type AS account_type,
           n.create_time AS create_time
    """
    df_nodes = pd.DataFrame(graph.run(node_query).data())

    # 2. 加载近期交易关系
    rel_query = f"""
    MATCH (a:Account)-[r:TRANSFER_TO]->(b:Account)
    WHERE r.timestamp >= datetime() - duration('P{days_back}D')
    RETURN a.account_id AS source_id,
           b.account_id AS target_id,
           r.amount AS amount,
           r.timestamp AS timestamp,
           r.is_fraud AS is_fraud
    """
    df_relations = pd.DataFrame(graph.run(rel_query).data())

    # 3. 类型优化
    df_nodes['risk_score'] = pd.to_numeric(df_nodes['risk_score'], errors='coerce')
    df_relations['amount'] = pd.to_numeric(df_relations['amount'], errors='coerce')
    df_relations['timestamp'] = pd.to_datetime(df_relations['timestamp'])
    df_relations['is_fraud'] = df_relations['is_fraud'].astype(int)

    return df_nodes, df_relations

# 执行加载
df_accounts, df_transfers = load_fraud_graph_data(graph, days_back=30)
print(f"账户数: {len(df_accounts)}, 交易关系数: {len(df_transfers)}")

数据质量检查：加载完成后，立即检查缺失值和异常值。例如，df_transfers['amount'].isna().sum() 统计空值数量；df_transfers['amount'].describe() 查看金额分布，为后续阈值设定提供初步参考。

4.6 常见问题与避坑指南

问题	原因	解决方案
连接超时	Neo4j 服务未启动或端口错误	检查 `neo4j status`，确认 bolt 端口（默认 7687）
查询返回空列表	标签名或属性名拼写错误	在 Neo4j Browser 中先用 `MATCH (n) RETURN labels(n) LIMIT 1` 验证标签
DataFrame 内存溢出	一次性加载全量数据	使用 SKIP/LIMIT 分批加载，或先聚合再导出
日期格式解析失败	Neo4j 返回 datetime 对象与 Pandas 不兼容	在 Cypher 中用 `toString(r.timestamp)` 转为字符串后再解析

通过本章的学习，你已经掌握了从 Neo4j 加载图谱数据到 DataFrame 的完整链路。下一阶段将基于这些结构化数据，进行告警阈值的统计分析与调优。