AML Network Analysis with $graphLookup

Solution Overview

Financial criminals operate in networks. Money laundering, fraud rings, and sanctions evasion rely on webs of shell companies, proxy directors, and layered transaction paths. Compliance teams need to map and analyze the relationship network around a suspect entity.

Traditional approaches run into two bottlenecks:

• Expensive queries: Relational databases use recursive Common Table Expressions (CTEs). These queries become expensive beyond two or three hops.

• Constant network round-trips: Client-side graph processing pulls data into the application layer. The client runs a new query for each hop.

In this solution, you:

• Build a network analysis engine that uses MongoDB operators to avoid these bottlenecks.

• Run multi-hop graph traversal, shortest-path detection, centrality scoring, community detection, and risk propagation directly inside MongoDB’s aggregation framework.

This solution focuses on the network analysis engine and the MongoDB operators that power it. To implement the overall architecture with MongoDB Atlas, FastAPI, and Next.js, also follow the previous Solutions Library, Financial Crime Mitigation.

Reference Architectures

The network analysis engine sits between the FastAPI service layer and MongoDB Atlas and processes graph operations through the aggregation pipeline.

click to enlarge

Figure 1. Agentic Investigation Pipeline

Dual $graphLookup Pattern

Money laundering networks involve bidirectional flows. An entity can source funds in one relationship and receive them in another.

A single $graphLookup traverses edges in one direction only. This solution executes two parallel lookups — forward and reverse — and merges the results into a unified network graph.

pipeline = [
    {"$match": {"entityId": center_entity_id}},
    # Forward traversal: follow source → target edges
    {"$graphLookup": {
        "from": "relationships",
        "startWith": "$entityId",
        "connectFromField": "target.entityId",
        "connectToField": "source.entityId",
        "as": "forward_relationships",
        "maxDepth": max_depth - 1,
        "restrictSearchWithMatch": {
            "active": True,
            "confidence": {"$gte": min_confidence}
        }
    }},
    # Reverse traversal: follow target → source edges
    {"$graphLookup": {
        "from": "relationships",
        "startWith": "$entityId",
        "connectFromField": "source.entityId",
        "connectToField": "target.entityId",
        "as": "reverse_relationships",
        "maxDepth": max_depth - 1,
        "restrictSearchWithMatch": {
            "active": True,
            "confidence": {"$gte": min_confidence}
        }
    }},
    # Merge both directions
    {"$project": {
        "entityId": 1,
        "all_relationships": {
            "$concatArrays": [
                "$forward_relationships",
                "$reverse_relationships"
            ]
        }
    }},
    {"$unwind": "$all_relationships"},
    {"$replaceRoot": {"newRoot": "$all_relationships"}},
    {"$limit": max_relationships}
]

click to enlarge

Figure 2. Dual $graphLookup — Bidirectional Network Discovery

The application receives the complete network graph in a single round-trip. Both traversals run in one aggregation pipeline, which removes the N+1 query problem of issuing a separate follow-up query for each document returned in the initial query.

How $graphLookup Traverses the Graph

$graphLookup performs a breadth-first search (BFS) in discrete waves:

• Seed: Evaluate startWith against the input document. Array values seed the frontier simultaneously (multi-root BFS).

• Query: Construct { connectToField: { $in: [frontier_values] }}, merged with any restrictSearchWithMatch filter. Execute as a standard indexed query against the from collection.

• Expand: For each matched document not in the visited set, add it to results, extract connectFromField values, and push them into the next frontier.

• Repeat: Increment depth. Return to step 2 until the frontier empties or maxDepth is reached.

• Assemble: Place all accumulated documents into an array under the as field.

Cycle detection is automatic. An internal visited set prevents infinite loops in cyclic graphs (A → B → C → A), which are common in money laundering structures. Each document appears in results exactly once.

The restrictSearchWithMatch Advantage

restrictSearchWithMatch pushes filter criteria into the traversal itself, not as a post-filter. MongoDB prunes dead branches during traversal rather than discovering the full graph and filtering afterward. For large networks, this can reduce the working set by an order of magnitude.

click to enlarge

Figure 3. Filtering During Traversal vs. Post-Filtering

Data Model Approach

The solution uses two collections: entities and relationships. They follow an adjacency list pattern. Edge metadata (confidence, evidence, verification status) lives as first-class fields on the relationship document.

Relationship Schema

{
    "relationshipId": "REL_8910",
    "source": {
        "entityId": "ENT_123",
        "entityType": "individual"
    },
    "target": {
        "entityId": "ENT_456",
        "entityType": "organization"
    },
    "type": "beneficial_owner_of",
    "direction": "directed",
    "strength": 0.85,
    "confidence": 0.95,
    "active": true,
    "verified": true,
    "evidence": [
        {
            "evidence_type": "corporate_registry",
            "confidence": 0.95,
            "source": "Companies House UK"
        }
    ],
    "datasource": "KYC_onboarding"
}

Build the Solution

Clone the repository and follow the setup instructions in the GitHub README:

git clone https://github.com/mongodb-industry-solutions/fsi-aml-fraud-detection.git
cd fsi-aml-fraud-detection/aml-backend
poetry install
poetry run uvicorn main:app --host 0.0.0.0 --port 8001 --reload

The following subsections walk through the six core graph operations implemented in the NetworkRepository.

db.relationships.createIndex({
    "source.entityId": 1,
    "active": 1,
    "confidence": -1
});
db.relationships.createIndex({
    "target.entityId": 1,
    "active": 1,
    "confidence": -1
});

pipeline = [
    {"$match": {"entityId": source_entity_id}},
    {"$graphLookup": {
        "from": "relationships",
        "startWith": "$entityId",
        "connectFromField": "source.entityId",
        "connectToField": "target.entityId",
        "as": "forward_paths",
        "maxDepth": max_depth - 1,
        "depthField": "depth"
    }},
    {"$graphLookup": {
        "from": "relationships",
        "startWith": "$entityId",
        "connectFromField": "target.entityId",
        "connectToField": "source.entityId",
        "as": "reverse_paths",
        "maxDepth": max_depth - 1,
        "depthField": "depth"
    }},
    {"$project": {
        "all_paths": {"$concatArrays": ["$forward_paths", "$reverse_paths"]}
    }},
    {"$unwind": "$all_paths"},
    {"$match": {"$or": [
        {"all_paths.source.entityId": target_entity_id},
        {"all_paths.target.entityId": target_entity_id}
    ]}},
    {"$sort": {"all_paths.depth": 1}},
    {"$limit": 1}
]

stats_pipeline = [
    {"$match": {"entityId": {"$in": network_entity_ids}}},
    {"$addFields": {
        "connection_count": {"$size": {"$ifNull": ["$connected_entities", []]}}
    }},
    {"$facet": {
        "basic_stats": [{"$group": {
            "_id": None,
            "total_nodes": {"$sum": 1},
            "avg_risk_score": {"$avg": "$riskAssessment.overall.score"},
            "max_risk_score": {"$max": "$riskAssessment.overall.score"}
        }}],
        "risk_distribution": [
            {"$group": {"_id": "$riskAssessment.overall.level", "count": {"$sum": 1}}},
            {"$sort": {"_id": 1}}
        ],
        "hub_entities": [
            {"$match": {"connection_count": {"$gte": 2}}},
            {"$sort": {"connection_count": -1}},
            {"$limit": 5},
            {"$project": {"entityId": 1, "name": 1, "connection_count": 1}}
        ]
    }}
]

"outgoing_risk_weighted": {
    "$sum": {"$multiply": [
        "$confidence",
        {"$switch": {
            "branches": [
                {"case": {"$in": ["$type", [
                    "confirmed_same_entity",
                    "business_associate_suspected"
                ]]}, "then": 0.9},
                {"case": {"$in": ["$type", [
                    "director_of", "ubo_of",
                    "parent_of_subsidiary"
                ]]}, "then": 0.7},
                {"case": {"$in": ["$type", [
                    "household_member",
                    "professional_colleague_public"
                ]]}, "then": 0.3}
            ],
            "default": 0.5
        }}
    ]}
}

adjacency_pipeline = [
    {"$match": {
        "$or": [
            {"source.entityId": {"$in": entity_ids}},
            {"target.entityId": {"$in": entity_ids}}
        ],
        "active": True,
        "confidence": {"$gte": 0.7}
    }},
    {"$group": {
        "_id": "$source.entityId",
        "connections": {"$addToSet": "$target.entityId"}
    }}
]

propagated_risk = (
    parent_entity_risk
    * propagation_factor        # decay per hop (default: 0.5)
    * relationship_confidence   # trust level for this edge
    * type_risk_weight          # domain-specific weight
)

Key Learnings

• Use dual $graphLookup for bidirectional network discovery: Run two parallel lookups, forward and reverse, and merge the results with $concatArrays to capture the full network in one aggregation.

• Push filters into the traversal with restrictSearchWithMatch: Prune dead branches during BFS instead of filtering the full graph afterward to reduce the working set for large networks.

• Create compound indexes on connectToField and your traversal filters: $graphLookup issues a $match/$in query at every BFS wave, so indexing connectToField together with active and confidence prevents collection scans at each hop.

• Compute network analytics server‑side with $facet and $switch: Run risk distribution, hub detection, and centrality scoring in parallel sub‑pipelines, and use $switch to assign domain‑specific risk weights to relationship types inside the aggregation.

• Apply exponential decay for risk propagation: Combine $graphLookup-based network discovery with a per‑hop decay formula that includes relationship confidence and type so the model automatically distinguishes high‑risk structural connections from low‑risk social links. This matches the guideline to keep key learnings in a consistent imperative style.

Authors

• Luis Pazmiño

• Mehar Grewal

• Andrea Alaman Calderon