How to Detect Financial Fraud Patterns with Knowledge Graphs: A Production Guide

# 交易级视图 —— 没有可疑之处
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多个供应商的付款。这是一个洗钱团伙。
# 收敛是红色警报。

(Employee)-[:APPROVED]->(Payment)-[:LINKED_TO]->(Invoice)-[:ISSUED_BY]->(Vendor)
                                   |
                                   v
                            (Payment)-[:PAID_TO]->(BankAccount)

from dataset.generate import FraudDatasetGenerator

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

供应商: 100
发票: 220 (200基础 + 20欺诈)
支付: 331 (300基础 + 31欺诈)
银行账户: 51 (50 + 1中间人账户)
员工: 30
组织: 5
欺诈模式: 5种类型

docker compose up -d

from graph.connection import FraudGraph

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

class FraudGraph:
    def __init__(self, uri=None, user=None, password=None):
        self.uri = uri or os.environ.get('NEO4J_URI', 'bolt://localhost:7687')
        self.user = user or os.environ.get('NEO4J_USER', 'neo4j')
        self.password = password or os.environ.get('NEO4J_PASSWORD', 'password')
        self.driver = None

    def connect(self):
        self.driver = GraphDatabase.driver(
            self.uri, auth=(self.user, self.password),
            max_connection_pool_size=50,
            connection_acquisition_timeout=30,
        )
        with self.driver.session() as session:
            session.run('RETURN 1')
        return self

    def __enter__(self):
        self.connect()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.close()

    def create_constraints(self):
        with self.driver.session() as session:
            session.run("CREATE CONSTRAINT vendor_tax_id IF NOT EXISTS FOR (v:Vendor) REQUIRE v.tax_id IS UNIQUE")
            session.run("CREATE CONSTRAINT invoice_id IF NOT EXISTS FOR (i:Invoice) REQUIRE i.invoice_id IS UNIQUE")
            session.run("CREATE CONSTRAINT account_id IF NOT EXISTS FOR (a:BankAccount) REQUIRE a.account_id IS UNIQUE")
            session.run("CREATE CONSTRAINT employee_id IF NOT EXISTS FOR (e:Employee) REQUIRE e.employee_id IS UNIQUE")

from graph.loader import FraudGraphLoader

with FraudGraph() as graph:
    graph.create_constraints()
    loader = FraudGraphLoader(graph)
    loader.load_all(data)

UNWIND $data AS v
MERGE (vendor:Vendor {tax_id: v.tax_id})
SET vendor.name = v.name,
    vendor.country = v.country,
    vendor.city = v.city
WITH vendor, v
MATCH (org:Organization {org_id: v.organization})
MERGE (vendor)-[:BELONGS_TO]->(org)

(Vendor)<-[:ISSUED_BY]-(Invoice)<-[:LINKED_TO]-(Payment)-[:PAID_TO]->(BankAccount)<-[:PAID_TO]-(Payment)-[:LINKED_TO]->(Invoice)-[:ISSUED_BY]->(Vendor)

MATCH (v1:Vendor)<-[:ISSUED_BY]-(i1:Invoice)<-[:LINKED_TO]-(p1:Payment)-[:PAID_TO]->(a:BankAccount)
MATCH (v2:Vendor)<-[:ISSUED_BY]-(i2:Invoice)<-[:LINKED_TO]-(p2:Payment)-[:PAID_TO]->(a)
WHERE id(v1) < id(v2)
MERGE (v1)-[:SHARED_ACCOUNT_WITH]->(v2)

-- 同一城市和国家
MATCH (v1:Vendor), (v2:Vendor)
WHERE id(v1) < id(v2)
  AND v1.city = v2.city
  AND v1.country = v2.country
MERGE (v1)-[:SHARED_LOCATION]->(v2)

-- 具有相同路由前缀（前3位数字）的银行账户
MATCH (v1:Vendor)<-[:ISSUED_BY]-(i1:Invoice)<-[:LINKED_TO]-(p1:Payment)-[:PAID_TO]->(a1:BankAccount)
MATCH (v2:Vendor)<-[:ISSUED_BY]-(i2:Invoice)<-[:LINKED_TO]-(p2:Payment)-[:PAID_TO]->(a2:BankAccount)
WHERE id(v1) < id(v2)
  AND a1.account_id <> a2.account_id
  AND substring(a1.routing_number, 0, 3) = substring(a2.routing_number, 0, 3)
MERGE (v1)-[:SHARED_ROUTING_PREFIX]->(v2)

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

Invoice: 220 nodes
  Payment: 331 nodes
  Vendor: 100 nodes
  BankAccount: 51 nodes
  Employee: 30 nodes

  SHARED_ACCOUNT_WITH: 412 relationships
  SHARED_LOCATION: 387 relationships
  LINKED_TO: 331 relationships
  PAID_TO: 331 relationships
  ISSUED_BY: 220 relationships
  APPROVED: 331 relationships
  SHARED_ROUTING_PREFIX: 52 relationships

// graph/queries.py — detect_duplicates()
MATCH (v:Vendor)<-[:ISSUED_BY]-(i1:Invoice),
      (v)<-[:ISSUED_BY]-(i2:Invoice)
WHERE i1.invoice_id < i2.invoice_id
  AND i1.amount = i2.amount
  AND i1.date < i2.date
  AND duration.between(date(i1.date), date(i2.date)).days <= $max_days
RETURN i1.invoice_id AS invoice_a,
       i2.invoice_id AS invoice_b,
       v.name AS vendor,
       i1.amount AS amount,
       i1.date AS date_a,
       i2.date AS date_b,
       duration.between(date(i1.date), date(i2.date)).days AS days_apart
ORDER BY days_apart

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['vendor']} | ${dup['amount']} | "
          f"{dup['days_apart']} 天间隔")

// graph/queries.py — detect_circular_schemes()
MATCH path = (v1:Vendor)-[:SHARED_ACCOUNT_WITH]-(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

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)}")

您可以使用以下风险指南，根据入账连接评估账户：

供应商数量

风险等级

1–3

正常

4–9

调查

10+

可能为洗钱中介

在生产环境中，您可以添加总接收金额的阈值以减少误报：

在Python中运行检测查询：

该查询成功识别了注入的洗钱中介环（ACC-MULE-001），该环通过25个空壳供应商聚合付款。

### 模式4：如何识别供应商共谋集群

共谋的空壳公司通常通过使用不同的公司名称来隐藏其关联。但它们经常泄露元数据，例如共享的物理办公室套房或相同的银行分行路由前缀。

要识别这些集群，您可以同时评估多个派生关系，并计算复合结构风险评分：

- 在此处使用OPTIONAL MATCH至关重要。严格的MATCH会丢弃那些表现出共享路由和位置但尚未共享银行账户的供应商。

- 您可以根据业务环境调整风险评分。在大型商业枢纽中，共享城市是常见的，但同时共享路由前缀和办公室套房会将复合评分提升至调查阈值以上。

多属性评分查询成功标记了四供应商共谋集群。

### 模式5：如何标记超出司法管辖区的付款批准

// graph/queries.py — detect_payment_anomalies()
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 inv.invoice_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"Found {len(anomalies)} anomalous approvals")
for a in anomalies:
    print(f"  Invoice {a['inv.invoice_id']}: "
          f"{a['e.name']} ({a['e.department']}) approved "
          f"{a['inv.category']} | ${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 graph.loader 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)} 个发现')

detect_all() 方法会运行所有五个查询并返回一个字典：

### 如何使用注入的真实数据验证召回率

验证脚本会将查询结果与真实数据文件进行对比以测量召回率：

验证逻辑会评估每种模式类型的检测准确性：

运行验证器对使用 seed=42 生成的数据集进行检测，得到以下结果：

### 如何执行完整工作流程

以下是设置环境、运行流水线并验证结果的完整终端命令序列：

## 生产架构：如何扩展图流水线

虽然参考原型在数百个节点上运行，但生产环境的欺诈检测引擎每天需要处理数百万笔交易。扩展此架构需要解决关键的操作瓶颈。

### 如何实现两阶段检测架构

在在线支付路径中，对每笔入账交易同步运行全局循环检测查询会导致数据库连接池耗尽。

应将检测分为两个层级：

- 阶段 1（同步拦截）：亚毫秒级 Cypher 查询，用于捕获已知模式。它们速度快是因为使用了索引属性、预构建关系和有限长度的模式。

- 阶段 2（异步图分析）：对于未触发确定性规则的交易，计算基于图的特征并进行评分：

### 如何调整阈值和管理误报

检测系统不可避免会产生误报。在生产环境中，应使用合规分析师的反馈持续校准查询阈值：

每次调整阈值后，可以重新运行验证以监控召回率影响：

### 如何缓解超节点扩展问题

图系统中常见的故障是“超节点”问题。合法的共享实体（例如大型企业房东、国家电力公司或内部工资账户）可能累积数万个入边。

当查询尝试在不进行过滤的情况下通过超节点时，遍历延迟会呈指数级增长。

可通过以下两种策略缓解：

- 度数限制：在遍历中添加度数剪枝：

MATCH (a:BankAccount)

WHERE size((a)<-[:PAID_TO]-()) < 1000

- 节点标签：为已知的高流量机构实体添加 :Whitelisted 或 :Utility 标签，并明确排除它们在合谋和洗钱检测中的扫描。

### 如何使用图特征进行机器学习

最有效的欺诈检测系统不会将图和机器学习视为相互竞争的范式。相反，图作为预测模型的高级特征存储：

- **度中心性**：衡量账户的连接性，以识别资金汇集点。
- **中介中心性**：识别在看似不相关供应商群体之间充当桥梁的实体。
- **PageRank 和社区检测**：量化网络中集群关联性及实体重要性。

仅依赖交易金额的机器学习分类器可能不会对 4900 美元的支付产生异常判断。但当该特征与图衍生特征结合——显示供应商具有高中介中心性且与已确认的欺诈团伙存在二级关联时，分类器能以高精度标记该交易。

## 关键架构教训与要点

构建和调优基于图的欺诈流水线突显了多个关键权衡：

首先，在关系型数据库中，模式设计优先考虑第三范式（3NF）以消除数据冗余。而在图数据库中，应围绕需要执行的遍历操作设计模式。如果查询需要反复遍历四跳以查找相关属性，应在数据摄入时将该关联直接建模为关系。

其次，唯一属性约束不仅是数据完整性保障，它们还提供了使节点查找达到 $O(1)$ 复杂度的底层哈希索引。省略约束会导致所有图合并操作退化为未索引的全表扫描。

第三，应精心管理规则库。从少量高信号模式（4-5 个）开始，每个模式都需要阈值校准和处置流程。同时添加数十个未经验证的启发式规则会为合规团队带来操作噪声和警报疲劳。

最后，建立分析师反馈闭环。检测查询的输出是假设而非最终结论。当合规分析师调查警报并确认欺诈时，应将该状态写回图中作为节点属性或标签（例如：:ConfirmedFraud）。这会丰富邻域拓扑结构，并为下游机器学习模型训练提供干净的真实数据。

## 结论

在本教程中，你学习了为何实时关系检查会遗漏复杂欺诈模式，如何将应付账款数据建模为互联图，以及如何通过衍生边的实体化加速遍历延迟。

你还实现了五个 Cypher 查询，可检测重复计费、环形团伙、资金中间人、供应商合谋和未经授权的付款批准。

本文配套代码可在 github.com/kamalct/fraud-detection-knowledge-graphs 获取。所有数据均为合成数据且可完全复现。

推荐课程（合作伙伴推荐，1/4）

（新标签页中打开）

**广告**

- Udemy《Dynamics 365 Finance & Operations：财务部分1/3 MB310 D365 Studio》4.5 星（1,724 人评分）4小时9分钟 初学者 证书 查看课程（新标签页中打开）
- Udemy《Neo4j：使用 Cypher 的图数据库基础》Louis Sayers 4.6 星（5,652 人评分）4小时46分钟 初学者 证书 查看课程（新标签页中打开）
- Coursera《数据科学：图分析》Madecraft 4周（每周1小时）初学者 证书 查看课程（新标签页中打开）

- 在 Udemy 上的 SAP Uplatz 训练《应付账款》4.4（满分5分）（1,468 人评价）10小时42分钟 初级 证书 查看课程（在新标签页中打开）

更多 Neo4j 课程在 Class Central

→

推荐课程（来自合作伙伴，侧边栏第2个）

- 在 Udemy 上的《Neo4j：Cypher、GDS、GraphQL、LLM、用于 RAG 的知识图谱》Cramlays, Nitin Gujral 4.5（满分5分）（246 人评价）3小时 初级 证书 查看课程（在新标签页中打开）

- 在 Coursera 上的《银行运营与欺诈预防》EDUCBA 2周，每周2小时 初级 证书 查看课程（在新标签页中打开）

- 在 Coursera 上的《Neo4j：Cypher、GDS、GraphQL、LLM、用于 RAG 的知识图谱》Packt - 课程讲师 9周，每周1小时 中级 证书 查看课程（在新标签页中打开）

- 在 edX 上的《使用 Neo4j 理解图数据库》2周，每周2-4小时 初级 证书 查看课程（在新标签页中打开）

推荐课程（来自合作伙伴，侧边栏第3个）

- 在 Corporate Finance Institute 上的《支付技术基础》Esteban Santana 5（满分5分）（6 人评价）34分钟 初级 证书 查看课程（在新标签页中打开）

- 在 DataCamp 上的《使用 LangChain 和 Neo4j 构建图 RAG》Adam Cowley 4.8（满分5分）（125 人评价）3小时 中级 证书 查看课程（在新标签页中打开）

推荐课程（来自合作伙伴，侧边栏第4个）

阅读更多文章。

如果这篇文章对你有帮助，请分享。

免费学习编程。freeCodeCamp 的开源课程已帮助超过40,000人成为开发者。立即开始

推荐课程（来自合作伙伴）