File size: 3,531 Bytes
3c7592d |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 |
import hashlib
import networkx as nx
import torch
# Global Graph Storage
transaction_graphs = {}
# Generate Unique Hash for Transaction Groups
def generate_graph_hash(transactions):
hash_string = "-".join(sorted(transactions)) # Sort for consistency
return hashlib.sha256(hash_string.encode()).hexdigest()
# Hash Function for Keys
def hash_key(value):
return hashlib.sha256(value.encode()).hexdigest()
# Add Transaction to Graph
def add_transaction(txn):
sender_hash = hash_key(txn["SenderAccount"])
receiver_hash = hash_key(txn["ReceiverAccount"])
# Check if sender or receiver is already in a known graph
related_graphs = [h for h, g in transaction_graphs.items() if sender_hash in g or receiver_hash in g]
if related_graphs:
# Merge related graphs into one
new_graph_hash = generate_graph_hash(related_graphs)
merged_graph = nx.compose_all([transaction_graphs[h] for h in related_graphs])
merged_graph.add_edge(sender_hash, receiver_hash, **txn)
# Remove old graphs and add the merged one
for h in related_graphs:
del transaction_graphs[h]
transaction_graphs[new_graph_hash] = merged_graph
else:
# Create a new graph if no related transactions exist
new_graph = nx.DiGraph()
new_graph.add_edge(sender_hash, receiver_hash, **txn)
transaction_graphs[generate_graph_hash([sender_hash, receiver_hash])] = new_graph
# Detect Laundering Patterns
def detect_pattern(graph):
"""Detect laundering patterns in the transaction graph."""
# If input is a Torch Geometric graph
if isinstance(graph, torch.Tensor) or hasattr(graph, "edge_index"):
# Extract unique node indices
nodes = torch.unique(graph.edge_index).tolist()
successors = {node: [] for node in nodes}
predecessors = {node: [] for node in nodes}
for i in range(graph.edge_index.shape[1]): # Process edges
sender, receiver = graph.edge_index[:, i].tolist()
successors[sender].append(receiver)
predecessors[receiver].append(sender)
# If input is a NetworkX graph
elif hasattr(graph, "nodes"):
nodes = list(graph.nodes)
successors = {node: list(graph.successors(node)) for node in nodes}
predecessors = {node: list(graph.predecessors(node)) for node in nodes}
else:
raise ValueError("Unsupported graph type")
# Pattern detection logic
for node in nodes:
outgoing = successors[node]
incoming = predecessors[node]
if len(outgoing) > 5:
return "Fan-Out" # One sender, many receivers
elif len(incoming) > 5:
return "Fan-In" # Many senders, one receiver
elif node in incoming:
return "Cycle" # Circular laundering
elif len(outgoing) > 2 and len(incoming) > 2:
return "Scatter Gather" # Money moves across multiple accounts
return "Normal"
# Store Suspicious AML Clusters
aml_clusters = {}
def flag_suspicious_graph(graph_hash):
"""Mark a graph as an AML cluster if laundering is detected"""
if graph_hash in transaction_graphs:
pattern = detect_pattern(transaction_graphs[graph_hash])
if pattern != "Normal":
aml_clusters[graph_hash] = transaction_graphs[graph_hash]
print(f"🚨 AML Detected: {pattern} | Cluster ID: {graph_hash}")
|