如何利用知识图谱来检测金融欺诈行为：一份实用指南

在企业支付工作中，大多数反欺诈架构都是孤立地评估每笔交易的。

当一张发票到达系统时，自动规则引擎会依据静态阈值来检查它：确认金额是否在正常范围内，供应商是否在批准的供应商名单上，以及发送请求的IP地址是否可疑。如果这些单独的属性看起来都符合要求，支付流程就会继续进行。

但这种方法的本质缺陷在于其结构设计。像资金转移链条、循环开票机制或虚假供应商这类复杂的欺诈手段，很少会生成那些能触发单一阈值警报的交易。例如，一家空壳公司开具4,800美元的发票，从而低于5,000美元的双重审批门槛，单独来看这个交易似乎完全合法。但当25家看似独立的供应商都将资金汇入同一个银行账户时，这种协同作案的模式就会暴露出来。

你不会在单笔交易的属性中发现这些欺诈迹象，因为它们其实隐藏在支付网络的整体结构之中。

在关系型数据库中，要追踪这些涉及多个实体的交易关系，就需要在供应商表、发票表、支付记录表和银行账户表之间进行递归查询。然而，当表格中的数据行数达到数百万时，这种深度查询会严重降低查询效率。而图数据库通过将实体之间的关系存储为无索引的邻接指针，使得多跳查询能够以常数时间完成。

在本教程中，你将了解基于Neo4j构建的端到端关系型欺诈检测系统的架构。你会学习如何将金融实体建模为图结构，如何预先计算相关数据以优化查询速度，以及如何使用参数化的Cypher查询来识别五种常见的采购欺诈手段。

完整的参考实现代码可以在fraud-detection-knowledge-graphs仓库中找到。

我们将会涵盖的内容：

• 为什么表格规则会漏掉关系型欺诈行为
• 先决条件
• 如何建模金融图结构及数据导入方案
• 如何使用Cypher查询识别五种欺诈模式
• 如何运行端到端检测流程并验证结果
• 生产环境架构：如何扩展图结构处理能力
• 关键架构设计要点与总结
• 结论

为什么表格规则会漏掉关系型欺诈行为

在应付账款的处理流程中，风险检测通常是对孤立的记录进行核查的：

# 从交易层面来看，没有可疑之处
transactions = [
    {"vendor": "Alpha Corp", "amount": 5200.00, "date": "2025-08-01"},
    {"vendor": "Beta LLC", "amount": 4800.00, "date": "2025-08-02"},
    {"vendor": "Gamma Inc", "amount": 5100.00, "date": "2025-08-03"},
]
# 所有金额都在正常范围内（1000美元到15000美元之间）。所有供应商都是合法的。日期也是连续的，没有任何条目触发了任何规则。

但当将这些记录映射到一个相互关联的图中时，那些隐藏的结构就会立刻显现出来：

# 从图的角度来看，这三个供应商都是将款项支付到了同一个账户中
connections = {
    "ACC-MULE-001": ["Alpha Corp", "Beta LLC", "Gamma Inc"],
    "ACC-0042":     ["Delta Corp", "Epsilon LLC"],
    "ACC-0099":     ["Zeta GmbH"],
}
# ACC-MULE-001接收来自25家以上的供应商的款项。这显然是一个资金中转环节。这种资金的集中流动本身就是一个危险信号。

底层数据并没有发生变化，但当你在实体图中分析这些交易时，那些协同操作的模式就会变得清晰可见。

在关系型数据库中，要连接这些数据点，就需要对多个不同的表格进行联接操作；而在知识图谱中，这些关联关系是直接以节点之间的边的形式存储的，因此你可以一次性完成对这些数据的查询。

知识图谱并不能取代规则引擎或机器学习模型，但它确实弥补了这些工具在处理某些问题时所存在的不足——它能够揭示事物之间的关联关系。

先决条件

要跟随本教程进行学习，你需要满足以下要求：

• 你的电脑上已经安装了Python 3.10或更高版本。
• 你已经安装了Docker和Docker Compose，以便能够运行本地的Neo4j实例。
• 你对Python以及数据库的基本概念有一定的了解。

如何构建财务图谱及数据导入模型

在开始进行欺诈检测之前，你需要将数据组织成一个结构清晰的知识图谱。参考数据集模拟了金融系统中常见的六种实体类型：

这些基础关系构成了一个二分支付网络：

(员工)-[:批准]->(付款记录)-[:关联到]->(发票)-[:由...开具]->(供应商)
                                   |
                                   v
                            (付款记录)-[:支付给]->(银行账户)

步骤1：如何使用已知的欺诈模式生成数据集

由于隐私保护和监管限制，现实世界中的金融数据集很难获取。使用合成数据可以人为添加已知的欺诈模式，从而系统地验证检测查询是否能够有效发现这些模式。

该数据集生成工具会创建大约500个实体，并在其中嵌入五种已知的欺诈模式：

• 12对重复的发票记录
• 1个循环式的发票交易链（涉及3家供应商)
• 1个资金洗钱团伙（1个账户关联25家供应商)
• 1个供应商串通作案集团（由4家供应商组成)
• 3例异常的付款记录

from dataset.generate import FraudDatasetGenerator

gen = FraudDatasetGenerator(
    num_vendors=100,
    num_employees=30,
    num_organizations=5,
    seed=42,
)
output_dir = gen.save("dataset/output")

通过设置固定的seed=42，可以确保数据集的可重复性。每次运行该生成工具都会产生完全相同的欺诈模式，这样就可以为测试查询提供可靠的基准数据：

供应商数量：100
发票数量：220张（其中200张是正常记录，20张是虚假发票）
付款记录数量：331条（其中300条是正常记录，31条是异常记录）
银行账户数量：51个（其中50个是正常账户，1个是用于资金洗钱的账户）
员工数量：30人
组织数量：5个
欺诈模式类型：5种

该生成工具还会输出一个包含所有虚假数据的清单文件ground_truth.json，这个文件可以作为自动验证的工具。

步骤2：如何连接Neo4j并创建约束条件

你可以使用Docker Compose来启动一个启用了APOC程序的本地Neo4j 5.20容器：

docker compose up -d

接下来，需要初始化数据库驱动程序并创建唯一性约束条件：

from graph.connection import FraudGraph

with FraudGraph() as graph:
    graph.create_constraints()

这些约束条件对查询性能至关重要。在Neo4j中，唯一属性约束会自动生成相应的B树索引。如果没有这些约束，任何MERGE操作都会迫使系统对数据库中的所有节点进行全量扫描：

步骤3：如何将实体加载到图中

FraudGraphLoader利用Cypher的UNWIND语法，在参数化事务中批量插入实体：

实体的加载顺序遵循外键依赖关系：首先加载组织信息，然后是供应商、银行账户和员工信息，最后加载发票和付款记录。

所有的加载操作都遵循相同的模式：

MERGE用于创建或匹配节点（利用约束条件可实现O(1)级别的查找速度），SET用于为节点添加属性，而MERGE则用于建立节点之间的关系。

步骤4：如何为快速查询生成派生关系

在图数据库中，一个关键的架构决策就是决定是在查询时计算这些关系，还是在数据导入阶段就预先生成它们。

要判断两个供应商是否共享同一个银行账户，就需要遍历一条包含四个节点的路径：

(供应商)<-[:发行方]-(发票)<-[:关联到]-(付款)-[:支付给]->(银行账户)<-[:支付给]-(付款)-[:关联到]->(发票)-[:发行方]->(供应商)

在每次进行实时欺诈检测时都遍历这条路径会带来巨大的计算负担。因此，可以在数据进入系统时直接生成这些关联关系。这样在批量导入数据时确实会略微增加写入延迟，但在进行运行时检测时，多跳的查询过程就可以简化为直接的单跳查询了。

第一个被生成的关联关系是SHARED_ACCOUNT_WITH，它用于连接那些共享同一银行账户的供应商：

MATCH (v1:供应商)<-[:发行方]-(i1:发票)<-[:关联到]-(p1:付款)-[:支付给]->(a:银行账户)
MATCH (v2:供应商)<-[:发行方]-(i2:发票)<-[:关联到]-(p2:付款)-[:支付给]->(a)
WHERE id(v1) < id(v2)
MERGE (v1)-[:SHARED_ACCOUNT_WITH]->(v2)

过滤条件WHERE id(v1) < id(v2)确保了这些关联关系仅以有向边的形式存储在系统中，从而避免了重复关系的出现。

你还可以生成表示供应商共享的地理位置信息以及银行账户的路由前缀，以此来检测那些可能存在串通行为的团伙：

-- 同一城市和国家
MATCH (v1:供应商), (v2:供应商)
WHERE id(v1) < id(v2)
  AND v1.city = v2.city
  AND v1.country = v2.country
MERGE (v1)-[:共享位置]->(v2)

-- 具有相同路由前缀的银行账户
MATCH (v1:供应商)<-[:发行方]-(i1:发票)<-[:关联到]-(p1:付款)-[:支付给]->(a1:银行账户)
MATCH (v2:供应商)<-[:发行方]-(i2:发票)<-[:关联到]-(p2:付款)-[:支付给]->(a2:银行账户)
WHERE id(v1) < id(v2)
  AND a1.account_id != a2.account_id
  AND substring(a1.routing_number, 0, 3) = substring(a2routing_number, 0, 3)
MERGE (v1)-[:共享路由前缀]->(v2)

数据导入完成后，你可以检查图结构，确认所有基础关系及衍生关系都已被正确加载：

info = graph.get_info()
for label, count in sorted(info['nodes'].items(), key=lambda x: -x[1]):
    print(f"  {label}: {count} 个节点")
for rel, count in sorted(info['relationships'].items(), key=lambda x: -x[1]):
    print(f"  {rel}: {count} 条关系")

输出结果会显示所有的原始实体以及生成的衍生关系：

  发票: 220个节点
  付款: 331个节点
  供应商: 100个节点
  银行账户: 51个节点
  员工: 30个节点

  SHARED_ACCOUNT_WITH: 412条关系
  SHARED_LOCATION: 387条关系
  LINKED_TO: 331条关系
  PAID_TO: 331条关系
  ISSUED_BY: 220条关系
  APPROVED: 331条关系
  SHARED_ROUTING_PREFIX: 52条关系

这些派生关系是专门为高速检测而设计的。在关系数据库中，要查找共享账户，需要通过对多个表进行自连接操作；而在图中，这种关系就表现为预先计算好的边，从而将复杂的遍历过程转化为直接的查询操作。

如何利用Cypher查询来检测五种欺诈模式

一旦你的数据架构及派生关系建立完成，就可以构建参数化的Cypher查询语句，从而识别出五种常见的采购欺诈行为。

模式1：如何查找重复的发票

欺诈者通常会间隔几天或几周提交重复的发票，因为他们认为应付账款部门会把这些发票当作常规的重复性支出进行处理。

基于规则的检测系统往往能够捕捉到完全匹配的情况，但那些日期略有差异的重复发票可能会被忽略。在图中，你可以查询同一供应商在指定时间窗口内发出的所有发票：

// graph/queries.py — 检测重复发票
MATCH (v:供应商)<-[:发行方]-(i1:发票),
      (v)<-[:发行方]-(i2:发票)
WHERE i1invoice_id < i2invoice_id
  AND i1金额 = i2金额
  AND i1日期 < i2日期
  AND duration.between(date(i1.date), date(i2.date)).days <= $max_days
RETURN i1invoice_id AS invoice_a,
       i2invoice_id AS invoice_b,
       v.name AS 供应商,
       i1金额 AS 金额,
       i1日期 AS date_a,
       i2日期 AS date_b,
       duration.between(date(i1.date), date(i2.date)).days AS 天数间隔
ORDER BY 天数间隔

关键点如下：

• 条件 i1invoice_id < i2invoice_id 可以避免重复返回相同的发票对，例如 (A, B) 和 (B, A)。
• i1金额 = i2金额 是判断欺诈行为的依据：如果两张发票的金额相同，那么很可能就是同一张发票被提交了两次。
• 由于查询范围仅限于同一供应商发出的发票，因此系统会首先利用索引快速查找相关信息，然后再使用内置的 duration.between() 函数来计算日期间隔。

你可以通过Python封装接口来执行这个查询：

from graph.queries import FraudQueries

queries = FraudQueries(graph)
duplicates = queries.detect_duplicates(max_days_apart=14)
print(f"找到了 {len(duplicates)} 对重复发票")
for dup in duplicates[:5]:
    print(f"  {dup['invoice_a']} ↔ {dup['invoice_b']} | "
          f"{dup['供应商']} | ${dup['金额']} | "
          f"{dup['天数间隔']} 天")

该查询成功检测出了合成数据集中所有12对重复发票，准确率达到了100%。

模式2：如何揭露循环式发票交易行为

在循环式发票交易方案中，三个或更多的供应商会形成封闭的循环关系，互相开具发票。每笔交易单独来看似乎都是合法的，但实际上并没有任何货物或服务真正被交付。

由于你已经预先计算出了SHARED_ACCOUNT_WITH这个值，因此检测循环结构的过程就被简化为在供应商子图中寻找由三个节点组成的三角形：

// graph/queries.py — detect_circular_schemes()
MATCH path = (v1:Vendor)-[:SHARED ACCOUNTWITH]-(v2:Vendor)
              -[:SHARED ACCOUNT_WITH]-(v3:Vendor)
              -[:SHARED ACCOUNT_WITH]-(v1)
WHERE id(v1) < id(v2) AND id(v2) < id(v3)
RETURN path,
       [n IN nodes(path) | {name: n.name, tax_id: n.tax_id}] AS vendors,
       length(path) AS hops

这里的关键点如下：

• 约束条件id(v1) < id(v2)和id(v2) < id(v3)可以排除循环排列的情况。如果没有这些约束，那么一个由三个节点组成的环结构(A, B, C)就会被多次检测到。
• 如果直接在未经优化的、包含四跳交易的原始路径上执行这个查询，那么要找到一个三环结构就需要遍历十二次，这样会在处理大型数据集时导致较高的CPU使用率。

circles = queries.detect_circular_schemes(min_cycle_length=3)
print(f"找到了{len(circles)}个循环性发票交易环")
for ring in circles:
    vendor_names = [v['name'] for v in ring['vendors']]
    print(f"  环结构({ring['hops']}跳): {' ↔ '.join(vendor_names)}")

这个查询能够以100%的准确率检测出循环性发票交易环结构。

模式3：如何检测资金骡子网络

在资金骡子网络的架构中，多个虚假实体会将资金汇集到一个由犯罪集团控制的账户中。虽然每次转账的金额都很小，但目标账户的接收节点数量却异常多。

你可以通过对目标银行账户进行度分析来发现这种模式：

// graph/queries.py — detect_money_mule_rings()
MATCH (a:BankAccount)<->[p_paid:PAID_TO]-(p:Payment)-[:LINKED_TO]->-(i:Invoice)
                  -[:ISSUED_BY]->-(v:Vendor)
WITH a, collect(DISTINCT v) AS vendors,
     sum(p.amount) AS total_received,
     count(DISTINCT v) AS unique_vendors,
     count(p) AS payment_count
WHERE unique_vendors >= $min_vendors
RETURN a.account_id,
       a.routing_number,
       a.bank_name,
       unique_vendors AS vendor_count,
       payment_count,
       total_received AS total_amount,
       [vendor IN vendors | vendor.name] AS vendor_names
ORDER BY unique_vendors DESC

你可以根据账户接收到的连接数量来使用以下风险评估标准：

在实际应用中，你可以设置一个最低接收金额阈值，以减少误报的情况：

WHERE unique_vendors >= 10 AND total_received > 50000

mules = queries.detect_money_mule_rings(min_vendor_count=5)
for acct in mules:
    print(f"  {acct['a.account_id']} ({acct['a.bank_name']}) | "
          f"{acct['vendor_count']}家供应商 | ${acct['total_amount']:.0f}")

模式4：如何识别供应商串通团伙

// graph/queries.py — detectVendor_collusion()
MATCH (v:Vendor)
OPTIONAL MATCH (v)-[:SHARED_ACCOUNT_WITH]->(v2)
OPTIONAL MATCH (v)-[:SHARED_LOCATION]->(v3)
OPTIONAL MATCH (v)-[:SHARED_ROUTING_PREFIX]->(v4)
WITH v,
     collect(DISTINCT v2) AS shared_account_vendors,
     collect(DISTINCT v3) AS shared_location_vendors,
     collect(DISTINCT v4) AS sharedrouting_vendors
WITH v,
     size(shared_account_vendors) AS sc,
     size/shared_location_vendors) AS sl,
     sizeshared-routing_vendors) AS sr
WHERE (sc + sl + sr) >= $min_score AND (sc + sl + sr) > 0
RETURN v.name AS vendor,
       v.tax_id,
       (sc + sl + sr) AS risk_score,
       [n IN shared_account_vendors + shared_location_vendors + sharedrouting_vendors | n.name] AS related_vendors,
       [x IN range(1, sc) | 'shared_account'] +
       [x IN range(1, sl) | 'shared_location'] +
       [x IN range(1, sr) | 'shared-routing'] AS flags
ORDER BY risk_score DESC

• 在这里使用OPTIONAL MATCH是非常重要的。如果使用严格的MATCH规则，那些虽然共享了银行分行代码前缀和办公场所，但尚未共享银行账户的供应商就会被忽略。
• 你可以根据自己的业务环境来调整风险评分的标准。在大型商业中心，供应商们共享同一个城市地址是很常见的现象；但如果它们同时共享银行分行代码前缀和办公场所，那么它们的综合风险评分就会超过警戒值。

// graph/queries.py — 检测支付异常
MATCH (inv:Invoice) <-[:LINKED_TO]-(p:Payment),
      (e:Employee)-[:APPROVED]-> (p),
      (e)-[:WORKS_FOR]-> (org:Organization)
WITH inv, p, e, org,
     CASE
       WHEN inv.category = 'procurement' AND
            e.department IN ['finance', 'operations', 'procurement']
            THEN 'ok'
       WHEN inv.category = 'payroll' AND
            e.department IN ['hr', 'finance']
            THEN 'ok'
       WHEN inv.category = 'travel' AND
            e.department IN ['operations', 'finance', 'hr']
            THEN 'ok'
       WHEN inv.category = 'marketing' AND
            e.department IN ['marketing', 'finance']
            THEN 'ok'
       WHEN inv.category = 'consulting' AND
            e.department IN ['finance', 'operations', 'procurement']
            THEN 'ok'
       WHEN inv.category = 'services' AND
            e.department IN ['finance', 'operations', 'procurement']
            THEN 'ok'
       WHEN inv.category = 'supplies' AND
            e.department IN ['operations', 'procurement', 'finance']
            THEN 'ok'
       WHEN inv.category = 'maintenance' AND
            e.department IN ['operations', 'procurement', 'finance']
            THEN 'ok'
       WHEN inv.category = 'software' AND
            e.department IN ['engineering', 'operations', 'finance']
            THEN 'ok'
       ELSE 'ANOMALY'
     END AS approval_status
WHERE approval_status = 'ANOMALY'
RETURN invinvoice_id, inv.amount, inv.category,
       inv.vendor_name, inv.date,
       e.name AS approver, e.employee_id,
       e.department AS approver_department
ORDER BY inv.date DESC

anomalies = queries.detect_payment_anomalies()
print(f"发现了 {len(anomalies)} 条异常审批记录")
for a in anomalies:
    print(f"  发票编号 {a['invinvoice_id']}: "
          f"{a['e.name']} ({a['e.department']}) 已批准支付，
          支付金额为 ${a['inv.amount']}")

python analysis/run_detections.py

生成数据集 → 加载到 Neo4j 中 → 建立关联关系 → 运行 5 条查询语句 → 保存结果 → 验证准确性

# analysis/run_detections.py
from dataset.generate import FraudDatasetGenerator
from graph.connection import FraudGraph
from graphloader import FraudGraphLoader
from graph.queries import FraudQueries

def run_pipeline(output_dir='analysis/output', force_regenerate=False):
    os.makedirs(output_dir, exist_ok=True)

    # 第一步：生成或加载数据集
    dataset_path = os.path.join(output_dir, 'data')
    if not os.path.exists(os.path.join(dataset_path, 'vendors.json')) or force_regenerate:
        gen = FraudDatasetGenerator(seed=42)
        gen.save(dataset_path)
    else:
        print(f'数据集已存在于：{dataset_path}')

    # 加载 JSON 文件
    data = {}
    for name in ['vendors', 'invoices', 'payments', 'bank_accounts',
                  'employees', 'organizations', 'ground_truth']:
        with open(os.path.join(dataset_path, f'{name}.json')) as f:
            data[name] = json.load(f)

    # 第二步：将数据加载到 Neo4j 中
    graph = FraudGraph()
    graph.connect()
    graph.create_constraints()
    loader = FraudGraphLoader(graph)
    loader.load_all(data)
    loader.create_shared_account_relationships()
    loader.create_attribute_based_relationships()

    info = graph.get_info()
    print('数据图已加载完成:')
    for label, count in sorted(info['nodes'].items(), key=lambda x: -x[1]):
        print(f'  {label}: {count} 个节点')

    # 第三步：运行所有检测查询
    queries = FraudQueries(graph)
    results = queries.detect_all()

    # 第四步：保存结果
    with open(os.path.join(output_dir, 'detection_results.json'), 'w') as f:
        json.dump(results, f, indent=2, default=str)

    for pattern, items in results.items():
        print(f'  {pattern}: 共检测到 {len(items)} 个异常情况')

# graph/queries.py
def detect_all(self):
    return {
        'duplicate_invoices': self.detect_duplicates(),
        'circular_schemes': self_detect_circular_schemes(),
        'money_mule_rings': selfdetect_money_mule_rings(),
        'vendor_collusion': self_detect/vendor_collusion(),
        'payment_anomalies': self.detect_payment_anomalies(),
    }

# analysis/validate.py
from collections import defaultdict

def validate(results_path, output_path=None):
    with open(results_path) as f:
        results = json.load(f)

    with open(os.path.join(os.path.dirname(results_path), 'data',
                           'ground_truth.json')) as f:
        groundtruth = json.load(f)

    gt_by_pattern = defaultdict(list)
    for pattern in groundTruth:
        gt_by_pattern[pattern['pattern]].append(pattern)

for pattern_name, gt_patterns in gt_by_pattern.items():
    found = results.get(pattern_name, [])
    gt_count = len(gtpatterns)

    if pattern_name == 'duplicate_invoices':
        print(f"重复发票：检测到 {len(found)} 对，共 {gt_count} 对——召回率为 {len(find)/max(1, gt_count):.0%")

    elif pattern_name == 'money_mule_rings':
        matched = sum(1 for gt in gt_patterns
                      for r in found
                      if r.get('a.account_id') == gt['shared_account'])
        print(f>资金转移环套：检测到 {matched} 个环套，共 {gt_count} 个环套——召回率为 {matched/max(1, gt_count):.0%}

重复发票：检测到 12 对，共 12 对——召回率为 100%
资金转移环套：检测到 1 个环套，共 1 个环套——召回率为 100%
供应商串通行为：共有 89 家高风险供应商被标记出来（采用多维度评分标准）
支付异常：检测到 3 例异常情况，共 3 例——召回率为 100%

# 1. 启动 Neo4j
docker compose up -d

# 2. 安装依赖项
pip install -r requirements.txt

# 3. 运行完整的处理流程
python analysis/run_detections.py

# 4. 验证结果
python analysis/validate.py

# 5. （可选）使用新鲜数据重新生成数据集并重新运行流程
python analysis/run_detections.py --regenerate

新交易
    ↓
阶段 1 — 确定性查询（快速、基于规则）
    ↓  （如果存在匹配的模式）
阶段 2 — 基于图结构特征进行机器学习评分
    ↓
风险等级 → 发出警报 / 批准 / 上报高级管理层

#根据误报率调整阈值

queries.detect_money_mule_rings(min_vendor_count=10) # 原值为5
queriesdetect_duplicates(max_days_apart=7) # 原值为14
queries_detect/vendor_collusion(min_risk_score=3) # 原值为2

python analysis/validate.py --input analysis/output/detection_results.json