深入 ‘Compliance-by-Design’：如何将金融行业（如 KYC/AML）的硬性规定直接编码进图的边缘逻辑？

各位同仁、技术爱好者们，

欢迎来到今天的讲座。我们今天要深入探讨一个在金融科技领域日益重要的概念：Compliance-by-Design (CbD)，即“合规即设计”。更具体地说，我们将聚焦于如何将金融行业的硬性合规规定，特别是像KYC（了解您的客户）和AML（反洗钱）这类复杂且动态的规则，直接编码进图数据库的边缘逻辑中，从而实现更高效、更智能、更具前瞻性的合规管理。

在传统的金融机构中，合规往往是一个事后审查的过程，它更像是一个成本中心，而非业务创新的驱动力。面对瞬息万变的监管环境、海量的交易数据以及日益复杂的洗钱和欺诈模式，传统的人工审查和基于关系型数据库的规则引擎显得力不从心。滞后性、高昂的人力成本、碎片化的数据视图以及难以捕捉的隐秘关联，是摆在所有金融机构面前的严峻挑战。

KYC和AML的复杂性尤为突出。它不仅仅是简单地核对黑名单，更需要对客户身份、资金来源、交易行为、关联网络进行多维度、深层次的洞察。这其中蕴含着海量的数据点和错综复杂的关系，而这些关系往往是识别风险的关键。

Compliance-by-Design 的核心思想，正是要颠覆这种传统模式。它倡导在系统和流程设计之初，就将合规要求内建其中，使其成为系统架构的固有组成部分，而非后期打补丁。其目标是让合规成为业务流程的自然延伸，而非阻碍。当我们将这一理念与图数据库的强大关系建模能力相结合时，我们便打开了一扇通往全新合规范式的大门。图数据库天然地适合表示实体之间的复杂关系，而这些关系恰恰是KYC/AML合规的核心。

我们将以编程专家的视角，深入剖析如何用代码实现这一愿景，让合规规则不再是冰冷的文档，而是活生生地融入我们的数据结构和业务逻辑。

---

第一章：图数据库：合规引擎的基石

在我们深入探讨规则编码之前，我们首先需要理解为什么图数据库是构建CbD合规引擎的理想选择。

1.1 图的基本概念回顾

图（Graph）是一种由节点（Nodes/Vertices）和边（Edges/Relationships）组成的数据结构。

• 节点： 代表实体，如客户、账户、交易、IP地址、设备等。节点可以拥有属性（Properties），用于存储实体的详细信息，例如客户的姓名、ID、风险等级，交易的金额、时间等。
• 边： 代表实体之间的关系，如“拥有”、“执行”、“发送给”、“居住在”等。边也可以拥有属性，例如关系建立的时间、关系的强度或权重。边通常是有方向的。

1.2 为什么图数据库特别适合KYC/AML？

特性	传统关系型数据库	图数据库	KYC/AML应用场景
关系建模	通过外键关联多个表，查询复杂关系需要大量JOIN操作。	原生支持关系，关系是数据模型的一等公民，查询效率高。	轻松表示客户与账户、账户与交易、客户与客户（亲属、商业伙伴）、地址、设备等之间的复杂多跳关联。
路径分析	难以高效发现多跳路径，通常需要递归查询，性能差。	内置路径查找算法，如最短路径、所有路径、深度广度优先搜索。	追踪资金流向（多跳交易路径），发现洗钱链条；识别多层股权结构中的最终受益人；发现“人头账户”网络。
模式匹配	难以表达和查询复杂的拓扑结构，需要复杂的SQL和业务逻辑。	支持基于图模式的查询语言（如Cypher），直观表达复杂模式。	识别可疑交易模式（如环形交易、扇入扇出模式）；发现隐藏的关联方网络；检测欺诈团伙。
灵活性	模式固定，修改表结构代价高昂，难以适应快速变化的监管规则。	模式灵活，可动态添加节点类型、边类型及属性，无需停机。	快速响应新的监管要求，灵活调整合规规则和数据模型，支持A/B测试新规则。
性能	随着关系深度增加，JOIN操作性能急剧下降。	查询性能与关系的数量和深度无关，只与遍历的节点和边数量有关。	处理海量交易和客户数据，能够实时或近实时地进行复杂关系查询和风险评估。
可解释性	查询结果通常是扁平化的表格，难以直观理解关系。	结果以图的形式呈现，直观易懂，便于分析师理解风险链条。	帮助合规分析师快速理解风险来源、传播路径，并向监管机构解释风险发现过程。

1.3 常见图数据库简介

市面上有多种成熟的图数据库，例如：

• Neo4j: 最流行的原生图数据库，支持Cypher查询语言，拥有强大的生态系统。
• Amazon Neptune: 亚马逊云服务提供的图数据库，支持Gremlin和openCypher。
• ArangoDB: 多模型数据库，支持图、文档和键值存储。
• JanusGraph: 分布式图数据库，构建在HBase、Cassandra等后端存储之上。

在本次讲座中，我们不会绑定到特定的图数据库产品，而是通过Python代码来模拟图结构和操作，以便更好地理解核心逻辑。

---

第二章：Compliance-by-Design的核心：将规则编码进图的边缘逻辑

现在，我们来深入探讨CbD的核心：如何将合规规则直接嵌入到图的边缘逻辑中。边缘逻辑不仅仅是边的存在与否，更是边的类型、方向、属性，以及这些元素共同承载的业务含义和合规约束。

2.1 传统规则引擎与图的融合

传统的规则引擎擅长处理基于事实的“如果-那么”规则，但它们在处理复杂、多跳关系时往往力不从心。图数据库则擅长表示和查询关系。CbD的精髓在于将两者融合：图数据库作为承载实体和关系的底层数据模型，规则引擎（或图查询本身）则用于在这些关系上执行合规逻辑。

2.2 边缘逻辑的深度理解

在图模型中，边不仅仅是连接两个节点的线。它代表着一种特定的交互、关联或依赖。将合规规则编码进边缘逻辑，意味着：

• 边类型即规则条件： 不同的边类型可以代表不同的业务操作或合规状态。例如，[:OWNS] 表示所有权，[:PERFORMS] 表示执行操作，而 [:TRANSACTION_ALERT] 则可能表示一个由规则触发的警报。
• 边属性即规则参数： 边的属性可以存储规则的参数，如交易金额、时间戳、交易类型、风险等级等。这些属性是规则判断的依据。
• 边的存在/缺失即合规/不合规： 某些规则可能关注特定边的存在（如是否存在与制裁名单的关联），或特定边的缺失（如客户身份信息是否完整）。
• 边序列即行为模式： 一系列有方向的边构成了行为序列，这对于识别洗钱模式至关重要。

2.3 规则分类与表示策略

为了更好地将规则编码，我们可以将合规规则大致分为几类，并针对性地采用不同的编码策略。

规则分类	描述	图编码策略
结构性规则	定义网络结构，禁止某些连接模式或要求某些连接必须存在。	边类型和属性： 定义特定的关系类型来表示合规状态（如 [:IS_SANCTIONED]）。模式匹配： 使用图查询语言查找禁止的连接模式（如 (Person)-[:TRANSACTS_WITH]->(SanctionedEntity)）。节点属性： 节点的 status 或 is_verified 属性。
行为性规则	基于一系列事件、交易或活动序列触发。	边序列和属性： 关注特定类型的边序列（如 (Account)-[:SENDER_OF]->(Transaction)-[:RECIPIENT_OF]->(Account)）。时间戳属性： 边的 timestamp 属性用于判断事件发生的顺序和时间窗口。聚合属性： 在路径上进行聚合计算（如总金额、交易频率）。
阈值规则	基于数值限制（如交易金额、账户余额、交易频率）。	边属性： 将阈值相关的数值存储为边或节点的属性（如 Transaction.amount, Account.balance）。查询过滤： 在图查询中使用 WHERE 子句过滤满足阈值条件的边或节点。
时间性规则	基于时间窗口、频率、持续时间。	边属性： 边的 timestamp 属性。时间函数： 图查询语言通常支持时间函数来计算时间差、聚合特定时间窗口内的事件。事件流处理： 结合实时事件流处理系统（如Kafka）和图数据库，实现滑动时间窗口内的规则检测。
组合规则	结合多种规则类型，例如“在特定时间窗口内，与高风险实体进行大额交易”。	复杂模式匹配： 结合多种边类型、节点属性、时间戳和数值条件，构建复杂的图模式查询。子图分析： 识别符合特定条件的子图，然后对子图进行进一步分析。图算法： 结合路径查找、社区发现等图算法来识别潜在风险。

---

第三章：案例分析：KYC/AML规则的图编码实践

现在，让我们通过具体的KYC/AML规则示例，演示如何将它们编码进图的边缘逻辑。我们将使用Python来模拟图的构建和规则的检测。

3.1 基础图模型构建

首先，我们定义一个简单的Python类来表示图中的节点和边。

import uuid
from datetime import datetime, timedelta

class Node:
    def __init__(self, node_id, labels=None, properties=None):
        self.id = node_id
        self.labels = set(labels) if labels else set()
        self.properties = properties if properties else {}

    def add_label(self, label):
        self.labels.add(label)

    def set_property(self, key, value):
        self.properties[key] = value

    def __repr__(self):
        return f"Node(ID='{self.id}', Labels={self.labels}, Props={self.properties})"

class Edge:
    def __init__(self, from_node_id, to_node_id, relationship_type, properties=None):
        self.from_node_id = from_node_id
        self.to_node_id = to_node_id
        self.type = relationship_type
        self.properties = properties if properties else {}

    def set_property(self, key, value):
        self.properties[key] = value

    def __repr__(self):
        return f"Edge({self.from_node_id}-[:{self.type}{{{self.properties}}}]->{self.to_node_id})"

class Graph:
    def __init__(self):
        self.nodes = {}  # {node_id: Node_object}
        self.edges = []  # List of Edge_object

    def add_node(self, node):
        if node.id in self.nodes:
            # Update existing node if new labels/properties are provided
            existing_node = self.nodes[node.id]
            existing_node.labels.update(node.labels)
            existing_node.properties.update(node.properties)
        else:
            self.nodes[node.id] = node
        return node

    def add_edge(self, edge):
        # Ensure nodes exist before adding an edge
        if edge.from_node_id not in self.nodes or edge.to_node_id not in self.nodes:
            raise ValueError(f"One or both nodes for edge {edge} do not exist in the graph.")
        self.edges.append(edge)
        return edge

    def get_node(self, node_id):
        return self.nodes.get(node_id)

    def get_out_edges(self, node_id):
        return [edge for edge in self.edges if edge.from_node_id == node_id]

    def get_in_edges(self, node_id):
        return [edge for edge in self.edges if edge.to_node_id == node_id]

    def find_paths(self, start_node_id, end_node_id, max_depth=3, current_path=None):
        """Simple DFS for path finding"""
        if current_path is None:
            current_path = [start_node_id]

        paths = []
        if start_node_id == end_node_id:
            return [[self.nodes[n_id] for n_id in current_path]]

        if max_depth <= 0:
            return []

        for edge in self.get_out_edges(start_node_id):
            if edge.to_node_id not in current_path: # Avoid cycles for simplicity
                new_path = current_path + [edge.to_node_id]
                found_paths = self.find_paths(edge.to_node_id, end_node_id, max_depth - 1, new_path)
                paths.extend(found_paths)
        return paths

    def find_all_paths_with_edges(self, start_node_id, end_node_id, max_depth=3, current_path_nodes=None, current_path_edges=None):
        """Finds all paths between two nodes, returning nodes and edges"""
        if current_path_nodes is None:
            current_path_nodes = [self.nodes[start_node_id]]
            current_path_edges = []

        all_paths = []

        if start_node_id == end_node_id:
            return [(current_path_nodes, current_path_edges)]

        if max_depth <= 0:
            return []

        for edge in self.get_out_edges(start_node_id):
            next_node_id = edge.to_node_id
            if next_node_id not in [n.id for n in current_path_nodes]: # Avoid simple cycles
                extended_node_path = current_path_nodes + [self.nodes[next_node_id]]
                extended_edge_path = current_path_edges + [edge]

                found_paths = self.find_all_paths_with_edges(
                    next_node_id, end_node_id, max_depth - 1, 
                    extended_node_path, extended_edge_path
                )
                all_paths.extend(found_paths)
        return all_paths

    def find_subgraph_by_pattern(self, pattern_nodes_labels, pattern_edges_types, max_depth=2):
        """
        A very simplified pattern matching (conceptual)
        Finds paths that match a sequence of node labels and edge types.
        Example: pattern_nodes_labels = ['Person', 'Account', 'Transaction']
                 pattern_edges_types = ['OWNS', 'PERFORMS']
        """
        matched_subgraphs = []

        if not pattern_nodes_labels or not pattern_edges_types:
            return matched_subgraphs

        # Start with nodes matching the first label
        start_nodes = [node for node in self.nodes.values() if pattern_nodes_labels[0] in node.labels]

        for s_node in start_nodes:
            # For each starting node, try to build a path matching the pattern
            # This is a conceptual simplification; real graph pattern matching is more complex

            # Simple 2-hop pattern matching for illustration
            if len(pattern_nodes_labels) >= 2 and len(pattern_edges_types) >= 1:
                for edge1 in self.get_out_edges(s_node.id):
                    if edge1.type == pattern_edges_types[0]:
                        node2 = self.get_node(edge1.to_node_id)
                        if node2 and pattern_nodes_labels[1] in node2.labels:
                            if len(pattern_nodes_labels) == 2: # Pattern is just 2 nodes, 1 edge
                                matched_subgraphs.append(([s_node, node2], [edge1]))
                            elif len(pattern_nodes_labels) >= 3 and len(pattern_edges_types) >= 2:
                                for edge2 in self.get_out_edges(node2.id):
                                    if edge2.type == pattern_edges_types[1]:
                                        node3 = self.get_node(edge2.to_node_id)
                                        if node3 and pattern_nodes_labels[2] in node3.labels:
                                            matched_subgraphs.append(([s_node, node2, node3], [edge1, edge2]))
                                            # Can extend for deeper patterns
        return matched_subgraphs

# Helper to generate unique IDs
def generate_id(prefix):
    return f"{prefix}-{str(uuid.uuid4())[:8]}"

# --- Graph Initialization Example ---
my_graph = Graph()

# Create some nodes
person_a = my_graph.add_node(Node(generate_id("P"), labels={"Person"}, properties={"name": "Alice", "risk_score": 30, "is_pep": False, "country": "USA"}))
person_b = my_graph.add_node(Node(generate_id("P"), labels={"Person"}, properties={"name": "Bob", "risk_score": 20, "is_pep": False, "country": "USA"}))
person_c = my_graph.add_node(Node(generate_id("P"), labels={"Person"}, properties={"name": "Charlie", "risk_score": 70, "is_pep": True, "country": "USA"})) # PEP
sanctioned_entity = my_graph.add_node(Node(generate_id("ORG"), labels={"Organization", "Sanctioned"}, properties={"name": "Evil Corp", "is_sanctioned": True, "country": "USA"}))

account_a1 = my_graph.add_node(Node(generate_id("ACC"), labels={"Account"}, properties={"account_number": "1001", "balance": 50000}))
account_b1 = my_graph.add_node(Node(generate_id("ACC"), labels={"Account"}, properties={"account_number": "2001", "balance": 20000}))
account_c1 = my_graph.add_node(Node(generate_id("ACC"), labels={"Account"}, properties={"account_number": "3001", "balance": 100000}))
account_se1 = my_graph.add_node(Node(generate_id("ACC"), labels={"Account"}, properties={"account_number": "4001", "balance": 10000}))

ip_home_a = my_graph.add_node(Node(generate_id("IP"), labels={"IPAddress"}, properties={"ip": "192.168.1.1", "geo": "USA"}))
ip_foreign = my_graph.add_node(Node(generate_id("IP"), labels={"IPAddress"}, properties={"ip": "203.0.113.42", "geo": "Nigeria"}))

# Create some edges
my_graph.add_edge(Edge(person_a.id, account_a1.id, "OWNS"))
my_graph.add_edge(Edge(person_b.id, account_b1.id, "OWNS"))
my_graph.add_edge(Edge(person_c.id, account_c1.id, "OWNS"))
my_graph.add_edge(Edge(sanctioned_entity.id, account_se1.id, "OWNS"))

my_graph.add_edge(Edge(person_a.id, person_b.id, "RELATED_TO", properties={"type": "Friend"}))
my_graph.add_edge(Edge(person_a.id, person_c.id, "RELATED_TO", properties={"type": "BusinessPartner"})) # Alice is partner with PEP Charlie

my_graph.add_edge(Edge(person_a.id, ip_home_a.id, "USES_IP"))
my_graph.add_edge(Edge(person_b.id, ip_home_a.id, "USES_IP")) # Bob and Alice share an IP

# --- Transaction Examples ---
now = datetime.now()

# Txn 1: Alice to Bob (normal)
txn1_id = generate_id("TXN")
txn1 = my_graph.add_node(Node(txn1_id, labels={"Transaction"}, properties={"amount": 1000, "timestamp": now - timedelta(hours=2), "currency": "USD"}))
my_graph.add_edge(Edge(account_a1.id, txn1.id, "SENDER_OF"))
my_graph.add_edge(Edge(txn1.id, account_b1.id, "RECIPIENT_OF"))
my_graph.add_edge(Edge(txn1.id, ip_home_a.id, "FROM_IP"))

# Txn 2: Alice to Sanctioned Entity (direct violation)
txn2_id = generate_id("TXN")
txn2 = my_graph.add_node(Node(txn2_id, labels={"Transaction"}, properties={"amount": 5000, "timestamp": now - timedelta(hours=1), "currency": "USD"}))
my_graph.add_edge(Edge(account_a1.id, txn2.id, "SENDER_OF"))
my_graph.add_edge(Edge(txn2.id, account_se1.id, "RECIPIENT_OF"))
my_graph.add_edge(Edge(txn2.id, ip_home_a.id, "FROM_IP"))

# Txn 3: Charlie (PEP) to Bob (indirect risk)
txn3_id = generate_id("TXN")
txn3 = my_graph.add_node(Node(txn3_id, labels={"Transaction"}, properties={"amount": 15000, "timestamp": now - timedelta(minutes=30), "currency": "USD"}))
my_graph.add_edge(Edge(account_c1.id, txn3.id, "SENDER_OF"))
my_graph.add_edge(Edge(txn3.id, account_b1.id, "RECIPIENT_OF"))
my_graph.add_edge(Edge(txn3.id, ip_foreign.id, "FROM_IP")) # PEP transacting from foreign IP

# Txn 4, 5, 6: Alice to various accounts (smurfing pattern)
# Assume these are small amounts to different accounts
txn_smurf_ids = []
for i in range(3):
    txn_id = generate_id("TXN")
    txn_smurf_ids.append(txn_id)
    txn = my_graph.add_node(Node(txn_id, labels={"Transaction"}, properties={"amount": 9000, "timestamp": now - timedelta(minutes=5 * i), "currency": "USD"}))
    my_graph.add_edge(Edge(account_a1.id, txn.id, "SENDER_OF"))

    # Send to different dummy accounts
    dummy_acc = my_graph.add_node(Node(generate_id("ACC"), labels={"Account"}, properties={"account_number": f"DUMMY{i}"}))
    my_graph.add_edge(Edge(txn.id, dummy_acc.id, "RECIPIENT_OF"))
    my_graph.add_edge(Edge(txn.id, ip_home_a.id, "FROM_IP"))

3.2 规则编码示例

我们将基于上述图模型，演示如何编码和检测几类典型的KYC/AML规则。

---

规则 1: 禁止与制裁实体进行交易 (Sanctioned Entity Check)

规则描述： 任何个人或组织不得直接或间接与列入制裁名单的实体进行交易。

图编码策略：

• 节点属性： 将制裁实体标记为 Node 具有 is_sanctioned: True 属性，并添加 Sanctioned 标签。
• 边缘逻辑： 查找 (Account)-[:SENDER_OF|RECIPIENT_OF]->(Transaction)<-[:PERFORMS]-(OtherAccount) 路径中，OtherAccount 所属的 Person 或 Organization 是否为制裁实体。这里的关键在于 OWNS 边连接 Person/Organization 和 Account。

Python 实现 (模拟 Cypher 查询逻辑):

def check_sanctioned_entity_transaction(graph, transaction_node_id):
    transaction = graph.get_node(transaction_node_id)
    if not transaction or "Transaction" not in transaction.labels:
        return []

    alerts = []

    # Find all accounts involved in this transaction (sender and recipient)
    involved_accounts = set()
    for edge in graph.get_in_edges(transaction_node_id):
        if edge.type == "SENDER_OF":
            involved_accounts.add(edge.from_node_id)
    for edge in graph.get_out_edges(transaction_node_id):
        if edge.type == "RECIPIENT_OF":
            involved_accounts.add(edge.to_node_id)

    for account_id in involved_accounts:
        # Find who owns this account
        owners = []
        for owner_edge in graph.get_in_edges(account_id):
            if owner_edge.type == "OWNS":
                owners.append(graph.get_node(owner_edge.from_node_id))

        for owner in owners:
            if owner and "is_sanctioned" in owner.properties and owner.properties["is_sanctioned"]:
                alerts.append({
                    "rule": "Sanctioned Entity Transaction",
                    "transaction_id": transaction_node_id,
                    "account_id": account_id,
                    "sanctioned_entity_id": owner.id,
                    "reason": f"Transaction {transaction_node_id} involves account {account_id} owned by sanctioned entity {owner.properties.get('name', owner.id)}."
                })
    return alerts

# --- Test Rule 1 ---
print("n--- Rule 1: Sanctioned Entity Transaction Check ---")
alerts_txn1 = check_sanctioned_entity_transaction(my_graph, txn1_id)
print(f"Transaction {txn1_id} (Alice to Bob): Alerts={alerts_txn1}")

alerts_txn2 = check_sanctioned_entity_transaction(my_graph, txn2_id)
print(f"Transaction {txn2_id} (Alice to Sanctioned Entity): Alerts={alerts_txn2}")

# Expected output for txn2 should show an alert

Cypher 伪代码示例：

MATCH (p:Person|Organization)-[:OWNS]->(a:Account)-[s:SENDER_OF|RECIPIENT_OF]-(t:Transaction)
WHERE p.is_sanctioned = true
RETURN t.id AS TransactionID, p.name AS SanctionedEntity, s.type AS RelationshipType

---

规则 2: 短时间内大额、多笔、分散交易 (Smurfing/Structuring Detection)

规则描述： 客户在短时间内（例如24小时内）向多个不同账户进行多笔小额（低于报告阈值，如1万美元）交易，但总金额巨大。这通常是“拆分交易”或“洗钱”的常见模式。

图编码策略：

• 边属性： Transaction 边的 amount 和 timestamp 属性。
• 边缘逻辑： 关注 (Account)-[:SENDER_OF]->(Transaction)-[:RECIPIENT_OF]->(OtherAccount) 路径。
• 聚合与时间窗口： 遍历一个账户在特定时间窗口内的所有出账交易，并聚合其特征。

Python 实现：

def check_smurfing_pattern(graph, source_account_id, time_window_hours=24, min_transactions=3, max_single_amount=10000, min_total_amount=30000):
    account = graph.get_node(source_account_id)
    if not account or "Account" not in account.labels:
        return []

    alerts = []

    # Get all outgoing transactions from this account within the time window
    recent_transactions = []

    # In a real system, you'd query for transactions within a specific time range.
    # Here, we'll iterate all transactions and filter.
    for edge in graph.get_out_edges(source_account_id):
        if edge.type == "SENDER_OF":
            txn_node = graph.get_node(edge.to_node_id)
            if txn_node and "Transaction" in txn_node.labels:
                txn_time = txn_node.properties.get("timestamp")
                if txn_time and (datetime.now() - txn_time) < timedelta(hours=time_window_hours):
                    recent_transactions.append(txn_node)

    if len(recent_transactions) < min_transactions:
        return []

    total_amount = 0
    recipient_accounts = set()

    for txn in recent_transactions:
        amount = txn.properties.get("amount", 0)
        if amount > max_single_amount:
            # This transaction itself exceeds the small amount threshold, not smurfing
            return [] 
        total_amount += amount

        # Find recipient account for this transaction
        for recipient_edge in graph.get_out_edges(txn.id):
            if recipient_edge.type == "RECIPIENT_OF":
                recipient_accounts.add(recipient_edge.to_node_id)

    if total_amount >= min_total_amount and len(recipient_accounts) >= min_transactions:
        alerts.append({
            "rule": "Smurfing/Structuring Detection",
            "source_account_id": source_account_id,
            "total_amount": total_amount,
            "num_transactions": len(recent_transactions),
            "num_recipients": len(recipient_accounts),
            "reason": f"Account {source_account_id} performed {len(recent_transactions)} small transactions (total {total_amount}) to {len(recipient_accounts)} recipients within {time_window_hours} hours."
        })

    return alerts

# --- Test Rule 2 ---
print("n--- Rule 2: Smurfing/Structuring Detection ---")
# Alice's smurfing transactions (txn_smurf_ids)
# Note: For this to trigger, we need to ensure the time window aligns and total amount is met
alerts_smurf = check_smurfing_pattern(my_graph, account_a1.id, time_window_hours=1, min_transactions=3, max_single_amount=10000, min_total_amount=25000)
print(f"Account {account_a1.id} (Alice) smurfing check: Alerts={alerts_smurf}")

# Expected output for Alice's account should show an alert

Cypher 伪代码示例：

MATCH (a:Account)-[:SENDER_OF]->(t:Transaction)-[:RECIPIENT_OF]->(ra:Account)
WHERE t.timestamp > datetime() - duration({hours: 24})
  AND t.amount <= 10000
WITH a, COLLECT(t) AS transactions, COLLECT(DISTINCT ra) AS recipients, SUM(t.amount) AS totalAmount
WHERE SIZE(transactions) >= 3 AND SIZE(recipients) >= 3 AND totalAmount >= 30000
RETURN a.account_number AS SourceAccount, totalAmount, SIZE(transactions) AS NumTransactions, SIZE(recipients) AS NumRecipients

---

规则 3: 关联方交易网络分析 (Related Party Transaction Network)

规则描述： 识别高风险个人（如PEP）的关联方，并分析这些关联方之间的交易模式，是否存在资金转移或掩盖资金来源的迹象。

图编码策略：

• 边类型： RELATED_TO 边，带有 type 属性（如 Friend, BusinessPartner, Family）。
• 节点属性： Person 节点具有 is_pep: True 属性。
• 边缘逻辑： 查找从PEP开始，通过 RELATED_TO 边连接的关联网络，并分析这些网络成员的交易。

Python 实现：

def check_pep_related_transactions(graph, pep_person_id, max_relation_depth=2, min_transaction_amount=5000):
    pep_node = graph.get_node(pep_person_id)
    if not pep_node or "Person" not in pep_node.labels or not pep_node.properties.get("is_pep"):
        return []

    alerts = []

    # Find all related persons up to max_relation_depth
    related_persons_ids = {pep_person_id}
    queue = [(pep_person_id, 0)]
    visited = {pep_person_id}

    while queue:
        current_person_id, depth = queue.pop(0)
        if depth >= max_relation_depth:
            continue

        for edge in graph.get_out_edges(current_person_id):
            if edge.type == "RELATED_TO":
                related_person_id = edge.to_node_id
                if related_person_id not in visited:
                    visited.add(related_person_id)
                    related_persons_ids.add(related_person_id)
                    queue.append((related_person_id, depth + 1))
        for edge in graph.get_in_edges(current_person_id): # Also check incoming related_to
            if edge.type == "RELATED_TO":
                related_person_id = edge.from_node_id
                if related_person_id not in visited:
                    visited.add(related_person_id)
                    related_persons_ids.add(related_person_id)
                    queue.append((related_person_id, depth + 1))

    # Now, check transactions between any two related persons (including the PEP)
    # This can be very broad, so let's refine: check transactions involving any of these related people

    related_accounts_ids = set()
    for person_id in related_persons_ids:
        for edge in graph.get_out_edges(person_id):
            if edge.type == "OWNS":
                related_accounts_ids.add(edge.to_node_id)

    for txn_node in [n for n in graph.nodes.values() if "Transaction" in n.labels]:
        sender_account_id = None
        recipient_account_id = None

        for edge in graph.get_in_edges(txn_node.id):
            if edge.type == "SENDER_OF":
                sender_account_id = edge.from_node_id
        for edge in graph.get_out_edges(txn_node.id):
            if edge.type == "RECIPIENT_OF":
                recipient_account_id = edge.to_node_id

        if sender_account_id in related_accounts_ids or recipient_account_id in related_accounts_ids:
            # Check if this transaction is between two related parties, or involves PEP and a related party
            # Or just involves a related party and is over a certain amount

            amount = txn_node.properties.get("amount", 0)
            if amount >= min_transaction_amount:
                # Find owners of sender/recipient accounts to identify the people
                sender_owner_id = None
                recipient_owner_id = None

                for edge in graph.get_in_edges(sender_account_id):
                    if edge.type == "OWNS":
                        sender_owner_id = edge.from_node_id
                for edge in graph.get_in_edges(recipient_account_id):
                    if edge.type == "OWNS":
                        recipient_owner_id = edge.from_node_id

                # If transaction is between any two related parties
                if (sender_owner_id in related_persons_ids and recipient_owner_id in related_persons_ids):
                    alerts.append({
                        "rule": "PEP Related Party Transaction",
                        "transaction_id": txn_node.id,
                        "pep_id": pep_person_id,
                        "involved_persons": [sender_owner_id, recipient_owner_id],
                        "reason": f"Transaction {txn_node.id} ({amount}) between two parties related to PEP {pep_node.properties.get('name', pep_person_id)}."
                    })
                # Or if it involves a related party (could be PEP directly) and an external party, but it's large
                elif (sender_owner_id in related_persons_ids or recipient_owner_id in related_persons_ids) and 
                     (sender_owner_id not in related_persons_ids or recipient_owner_id not in related_persons_ids):
                    alerts.append({
                        "rule": "PEP Related Party Transaction (High Value)",
                        "transaction_id": txn_node.id,
                        "pep_id": pep_person_id,
                        "involved_persons": [p for p in [sender_owner_id, recipient_owner_id] if p in related_persons_ids],
                        "reason": f"Transaction {txn_node.id} ({amount}) involves a party related to PEP {pep_node.properties.get('name', pep_person_id)} and is of significant value."
                    })

    return alerts

# --- Test Rule 3 ---
print("n--- Rule 3: PEP Related Party Transaction Check ---")
alerts_pep_related = check_pep_related_transactions(my_graph, person_c.id, max_relation_depth=1, min_transaction_amount=10000)
print(f"PEP {person_c.properties.get('name')} related party transactions: Alerts={alerts_pep_related}")
# Expected: Charlie (PEP) to Bob (normal) transaction (txn3) should be flagged because Bob is related to Alice, who is business partner with Charlie.
# Also, if Alice (partner of PEP Charlie) transacts with Bob (friend of Alice), this might be flagged.

Cypher 伪代码示例：

MATCH (pep:Person {is_pep: true})-[*1..2]-(rp:Person) // Find PEP and its related parties up to 2 hops
WITH COLLECT(DISTINCT pep) + COLLECT(DISTINCT rp) AS high_risk_persons
UNWIND high_risk_persons AS p1
MATCH (p1)-[:OWNS]->(a1:Account)-[:SENDER_OF|RECIPIENT_OF]-(t:Transaction)-[:SENDER_OF|RECIPIENT_OF]-(a2:Account)<-[:OWNS]-(p2:Person)
WHERE p2 IN high_risk_persons // Transaction between two high-risk persons
  AND t.amount >= 10000
RETURN t.id AS TransactionID, p1.name AS Person1, p2.name AS Person2, t.amount AS Amount

---

规则 4: 异常地理位置/IP 地址交易 (Geographic/IP Anomaly)

规则描述： 客户的常用居住地、注册地址与交易发生时的IP地址地理位置显著不符。这可能预示着账户被盗用或试图掩盖真实交易位置。

图编码策略：

• 节点属性： Person 节点的 country 或 address 属性，IPAddress 节点的 geo 属性。
• 边缘逻辑： (Person)-[:LIVES_AT]->(Address)，(Transaction)-[:FROM_IP]->(IPAddress)。
• 比较逻辑： 在图查询中比较 Person.country 和 IPAddress.geo。

Python 实现：

def check_geographic_ip_anomaly(graph, transaction_node_id):
    transaction = graph.get_node(transaction_node_id)
    if not transaction or "Transaction" not in transaction.labels:
        return []

    alerts = []

    # Get IP address used for the transaction
    transaction_ip_node = None
    for edge in graph.get_out_edges(transaction_node_id):
        if edge.type == "FROM_IP":
            transaction_ip_node = graph.get_node(edge.to_node_id)
            break

    if not transaction_ip_node:
        return []

    ip_geo = transaction_ip_node.properties.get("geo")
    if not ip_geo:
        return []

    # Get the sender's account and owner
    sender_account_id = None
    for edge in graph.get_in_edges(transaction_node_id):
        if edge.type == "SENDER_OF":
            sender_account_id = edge.from_node_id
            break

    if not sender_account_id:
        return []

    sender_person = None
    for edge in graph.get_in_edges(sender_account_id):
        if edge.type == "OWNS":
            sender_person = graph.get_node(edge.from_node_id)
            break

    if not sender_person or "Person" not in sender_person.labels:
        return []

    person_country = sender_person.properties.get("country")

    if person_country and ip_geo and person_country != ip_geo:
        alerts.append({
            "rule": "Geographic IP Anomaly",
            "transaction_id": transaction_node_id,
            "person_id": sender_person.id,
            "person_country": person_country,
            "transaction_ip_geo": ip_geo,
            "reason": f"Transaction {transaction_node_id} from IP in {ip_geo} by {sender_person.properties.get('name', sender_person.id)} whose registered country is {person_country}."
        })
    return alerts

# --- Test Rule 4 ---
print("n--- Rule 4: Geographic IP Anomaly Check ---")
alerts_geo_txn1 = check_geographic_ip_anomaly(my_graph, txn1_id) # Alice (USA) -> IP (USA)
print(f"Transaction {txn1_id}: Alerts={alerts_geo_txn1}")

alerts_geo_txn3 = check_geographic_ip_anomaly(my_graph, txn3_id) # Charlie (USA) -> IP (Nigeria)
print(f"Transaction {txn3_id}: Alerts={alerts_geo_txn3}")

# Expected output for txn3 should show an alert

Cypher 伪代码示例：

MATCH (p:Person)-[:OWNS]->(a:Account)-[:SENDER_OF]->(t:Transaction)-[:FROM_IP]->(ip:IPAddress)
WHERE p.country <> ip.geo
RETURN t.id AS TransactionID, p.name AS PersonName, p.country AS PersonCountry, ip.ip AS IPAddress, ip.geo AS IPGeoLocation

---

规则 5: PEP (Politically Exposed Person) 风险评估 (PEP Risk Assessment)

规则描述： 与政治敏感人物（PEP）相关的交易或账户，无论金额大小，都应被视为高风险，并进行增强型尽职调查。

图编码策略：

• 节点属性： Person 节点具有 is_pep: True 属性。
• 边缘逻辑： 任何涉及 is_pep: True 的 Person 的 OWNS, SENDER_OF, RECIPIENT_OF, RELATED_TO 边。
• 规则触发： 只要发现 PEP 直接或间接参与，即触发警报。

Python 实现：

def check_pep_involvement(graph, node_id):
    """
    Checks if a given node (Person, Account, or Transaction) is directly or indirectly
    involved with a PEP.
    """
    alerts = []

    start_node = graph.get_node(node_id)
    if not start_node:
        return []

    # If it's a Person, check if they are PEP
    if "Person" in start_node.labels and start_node.properties.get("is_pep"):
        alerts.append({
            "rule": "PEP Direct Involvement",
            "entity_id": node_id,
            "reason": f"Person {start_node.properties.get('name', node_id)} is a PEP."
        })
        return alerts # Direct involvement is enough

    # Check paths to PEPs (e.g., Account -> OWNS -> Person (PEP))
    # Or Transaction -> SENDER_OF/RECIPIENT_OF -> Account -> OWNS -> Person (PEP)

    # We can use a simplified path search for "PEP" nodes
    # Let's find all PEPs first
    pep_nodes = [node for node in graph.nodes.values() if "Person" in node.labels and node.properties.get("is_pep")]

    for pep_node in pep_nodes:
        # Check paths from the starting node to the PEP
        # Max_depth can be tuned based on how "indirect" we want to consider
        paths_to_pep = graph.find_all_paths_with_edges(node_id, pep_node.id, max_depth=3)
        if paths_to_pep:
            for path_nodes, path_edges in paths_to_pep:
                # Filter to relevant paths (e.g., through accounts, transactions, related_to)
                relevant_path_found = False
                for edge in path_edges:
                    if edge.type in ["OWNS", "SENDER_OF", "RECIPIENT_OF", "RELATED_TO"]:
                        relevant_path_found = True
                        break

                if relevant_path_found:
                    alerts.append({
                        "rule": "PEP Indirect Involvement",
                        "entity_id": node_id,
                        "pep_id": pep_node.id,
                        "reason": f"Entity {node_id} is indirectly involved with PEP {pep_node.properties.get('name', pep_node.id)} via path: {' -> '.join([n.labels.pop() for n in path_nodes])}",
                        "path_details": [(n.id, n.labels, e.type) for n, e in zip(path_nodes[:-1], path_edges)] + [(path_nodes[-1].id, path_nodes[-1].labels, None)]
                    })
                    # For simplicity, we might just return the first found path
                    return alerts
    return alerts

# --- Test Rule 5 ---
print("n--- Rule 5: PEP Involvement Check ---")
alerts_pep_person_c = check_pep_involvement(my_graph, person_c.id) # Charlie is PEP
print(f"Person {person_c.properties.get('name')}: Alerts={alerts_pep_person_c}")

alerts_pep_account_c1 = check_pep_involvement(my_graph, account_c1.id) # Account owned by Charlie
print(f"Account {account_c1.id}: Alerts={alerts_pep_account_c1}")

alerts_pep_txn3 = check_pep_involvement(my_graph, txn3_id) # Transaction from Account owned by Charlie
print(f"Transaction {txn3_id}: Alerts={alerts_pep_txn3}")

alerts_pep_person_a = check_pep_involvement(my_graph, person_a.id) # Alice is partner with Charlie
print(f"Person {person_a.properties.get('name')}: Alerts={alerts_pep_person_a}")

---

将合规状态编码进图

除了触发警报，我们还可以将合规审查的状态和结果直接编码回图。

• 新增 COMPLIANCE_STATUS 边： Transaction -> [:HAS_ALERT {status: 'PENDING', rule_triggered: 'R2'}] -> Rule。
• 节点属性： Transaction 节点可以有 compliance_status, triggered_rules 属性。
• 警报节点： 每次触发警报，可以创建一个 Alert 节点，并用边连接到相关的 Transaction、Account 或 Person 节点。

def add_compliance_alert_to_graph(graph, entity_id, rule_name, reason):
    alert_id = generate_id("ALERT")
    alert_node = graph.add_node(Node(alert_id, labels={"Alert"}, properties={"rule": rule_name, "reason": reason, "timestamp": datetime.now(), "status": "PENDING"}))
    graph.add_edge(Edge(entity_id, alert_node.id, "HAS_ALERT", properties={"rule_triggered": rule_name}))
    print(f"Added Alert: {alert_node}")
    print(f"Edge from {entity_id} to {alert_node.id}")
    return alert_node

# Example of adding an alert after detection
print("n--- Adding Alerts to Graph ---")
alerts_txn2 = check_sanctioned_entity_transaction(my_graph, txn2_id)
if alerts_txn2:
    add_compliance_alert_to_graph(my_graph, txn2_id, alerts_txn2[0]["rule"], alerts_txn2[0]["reason"])

alerts_geo_txn3 = check_geographic_ip_anomaly(my_graph, txn3_id)
if alerts_geo_txn3:
    add_compliance_alert_to_graph(my_graph, txn3_id, alerts_geo_txn3[0]["rule"], alerts_geo_txn3[0]["reason"])

# Verify alerts in the graph
print("n--- Verifying Alerts in Graph ---")
for node in my_graph.nodes.values():
    if "Alert" in node.labels:
        print(node)

---

第四章：动态规则管理与合规生命周期

在真实的金融环境中，合规规则并非一成不变。监管要求会更新，新的洗钱模式会涌现。因此，一个有效的CbD系统必须支持规则的动态管理。

4.1 规则的变更与版本控制

图数据库的灵活模式（schema-less或schema-optional）特性使其更容易适应规则变更。新的规则可能意味着：

• 新增边类型或属性： 例如，引入 [:HAS_RISK_FLAG] 边或 Node.kyc_status 属性。
• 修改现有规则逻辑： 直接修改图查询或图遍历算法的参数（如阈值、时间窗口）。
• 引入新的图模式匹配： 针对新的风险模式编写新的查询。

对于规则本身，可以采用版本控制系统进行管理，每次规则的修改都记录在案，确保可追溯性和审计性。

4.2 实时合规检测

将图数据库与事件流处理系统（如Apache Kafka）结合，可以实现近实时甚至实时的合规监控。

1. 事件捕获： 交易、客户信息更新等事件通过Kafka等系统实时流入。
2. 图更新： 事件处理器根据事件内容更新图中的节点和边（例如，添加新的 Transaction 节点和 SENDER_OF/RECIPIENT_OF 边）。
3. 规则触发： 在图更新的同时或更新后，立即触发预设的图合规规则查询。例如，每当有新的 Transaction 边添加到图中，就立即运行检查与制裁实体的交易规则。
4. 警报与响应： 规则触发警报后，系统可以自动创建警报节点，通知合规分析师，或直接采取自动化措施（如冻结交易）。

4.3 人工干预与反馈回路

自动化检测并非万能。总会有误报和漏报。

• 警报生成： 系统生成结构化警报，包含所有相关图路径和证据。
• 人工调查： 合规分析师利用图可视化工具，在图上直观地探索警报，进行深度调查，发现自动化规则未能捕捉的复杂关联。
• 反馈回路： 人工调查的结果（例如，确认某模式为真实风险，或标记为误报）应反馈回系统，用于优化规则、调整风险评分模型，甚至训练机器学习模型。这形成了持续改进的合规生命周期。

---

第五章：挑战与展望

5.1 数据集成

将来自不同异构数据源（核心银行系统、CRM、交易历史、外部黑名单、政府数据库）的数据整合到统一的图模型中，是一个巨大的挑战。这需要强大的ETL（抽取、转换、加载）管道和数据治理策略。

5.2 性能与扩展性

面对海量客户、账户和交易数据，图数据库的性能和扩展性至关重要。

• 大规模图数据： 数十亿节点和边的数据量，需要分布式图数据库和高效的图分区策略。
• 复杂查询优化： 复杂的图模式匹配和路径查询可能非常耗时，需要索引优化、查询优化器和缓存机制。

5.3 规则复杂性管理

随着规则数量的增加，如何有效地管理规则，避免冲突，确保规则集的一致性和完整性，是一个持续的挑战。这可能需要引入规则建模语言、规则生命周期管理工具。

5.4 解释性与可审计性

合规系统必须能够清晰地解释为什么某笔交易或某个客户被标记为高风险。图数据库的优势在于其直观的关系表示，可以帮助系统生成清晰的“风险路径”或“关联证据”，满足监管审计要求。

5.5 图算法的运用

除了基于规则的模式匹配，高级图算法可以在AML中发挥更深层次的作用：

• 社区发现： 识别潜在的洗钱团伙或欺诈网络。
• 中心性分析： 识别网络中的关键节点（如资金中转站、核心欺诈者）。
• 路径预测： 预测资金的潜在流向。

5.6 AI与机器学习的结合

图神经网络（GNN）是当前研究热点，它能直接在图结构上学习特征表示，非常适合异常检测和风险评分。将图数据库作为GNN的特征工程和推理环境，将是未来CbD的重要方向。通过GNN，系统可以学习识别比硬编码规则更微妙、更复杂的风险模式。

---

结束语

我们今天探讨的 Compliance-by-Design 并非仅仅是一种技术实现，它更是一种思维模式的转变。它将合规从一个被动的、高成本的负担，转变为一个主动的、智能的、嵌入业务流程的价值驱动者。图数据库以其对关系数据的原生支持，为这一转变提供了坚实的技术基石。通过将KYC/AML规则直接编码进图的边缘逻辑，我们不仅能够更有效地识别风险，更能实现合规的自动化、智能化和可解释性。这不仅能帮助金融机构更好地履行社会责任，也能在日益复杂的监管环境中获得竞争优势。未来，随着图技术与人工智能的进一步融合，我们的合规系统将变得更加智能、自动化、更具预见性，为金融行业的健康发展保驾护航。

谢谢大家！