Análise de rede AML com $graphLookup

Visão Geral da Solução

Os criminos financeiros operam em redes. Limpeza de capital, anéis de fraude e evasão de autorizações dependem de redes de empresas de shell, directores proxy e caminhos de transação em camadas. As equipes de compliance precisam mapear e analisar a rede de relacionamento em torno de uma entidade suspeita.

As abordagens tradicionais se deparam com dois gargalos:

• Queries dispendiosas: os bancos de dados relacionais usam Expressões de Tabela Comum (CTEs) recursivas. Essas queries se tornam caras além de dois ou três saltos.

• Ida e volta da rede constantes: o processamento de gráficos do lado do cliente extrai dados para a camada do aplicação . O cliente executa uma nova query para cada salto.

Nesta solução, você:

• Crie um mecanismo de análise de rede que use operadores MongoDB para evitar esses gargalos.

• Execute grafos multi-op, detecção de caminho mais curto, pontuação de centralidade, detecção de comunidade e propagação de risco diretamente dentro da estrutura de agregação do MongoDB.

Esta solução se concentra no mecanismo de análise de rede e nos operadores MongoDB que o alimentam. Para implementar a arquitetura geral com MongoDB Atlas, FastAPI e Next.js, siga também a Biblioteca deSoluções anterior, Mitigação de Crimes financeiros.

Arquiteturas de referência

O mecanismo de análise de rede fica entre a camada de serviço FastAPI e MongoDB Atlas e processa operações de gráficos por meio do pipeline de agregação .

clique para ampliar

figura 1. Pipeline de Investigação Agente

Dual $graphLookup Padrão

As redes de limpeza de ativos envolvem fluxos bidirecionais. Uma entidade pode obter recursos em um relacionamento e recebê-los em outro.

Um único $graphLookup atravessa as bordas em apenas uma direção. Essa solução executa duas pesquisas paralelas - direta e reversa - e mescla os resultados em um gráfico de rede unificado.

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}
]

clique para ampliar

figura 2. $graphLookup duplo - Descoberta de rede bidirecional

O aplicação recebe o gráfico de rede completo em uma única ida e volta. Ambos os traversais são executados em um pipeline de agregação , o que remove o problema de query N+1 de emitir uma query de acompanhamento separada para cada documento retornado na query inicial.

Como $graphLookup Percorre o gráfico

$graphLookup realiza uma pesquisa em largura (BFS) em ondes discretas:

• Semente: Avalie startWith em relação ao documento de entrada. Os valores de array semeiam a borda simultaneamente (BFS multiraiz).

• Query: Construa,{ connectToField: { $in: [frontier_values] }} mesclado com qualquer restrictSearchWithMatch filtro. Execute como uma query indexada padrão em relação à coleção.

• Expandir: para cada documento correspondente que não esteja no conjunto visitado, adicione-o aos resultados, extraia connectFromField valores e envie-os para a próxima borda.

• Repita: aumente a profundidade. Retorne à etapa 2 até que a borda esvazie ou maxDepth seja atingido.

• Montar: coloque todos os documentos acumulados em uma array sob o campo.

A detecção de ciclo é automática. Um conjunto visitado interno evita loops infinitos em gráficos cíclicos (A → B → C → A), que são comuns em estruturas de limpeza de capital. Cada documento aparece nos resultados exatamente uma vez.

O restrictSearchWithMatch Vantagem

restrictSearchWithMatch empurra critérios de filtro para a própria transversal, não como um pós-filtro. O MongoDB poda os ramos morto durante a travessia em vez de descobrir o gráfico completo e filtrar posteriormente. Para redes grandes, isso pode reduzir o conjunto de trabalho em uma ordem de magnitude.

clique para ampliar

figura 3. Filtragem durante a passagem versus pós-filtragem

Abordagem do modelo de dados

A solução utiliza duas collections: entities e relationships. Elas seguem um padrão de lista de adjacência. Os metadados de borda ( confiança , provas , status de verificação ) existem como campos de primeira classe no documento de relacionamento .

Esquema de relacionamento

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

Construir a solução

Clone o repositório e siga as instruções de configuração no 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

As subseções a seguir percorrem as seis operações principais de gráfico implementadas no 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
)

Principais Aprendizados

• Usar duplo $graphLookup para descoberta de rede bidirecional: execute duas pesquisas paralelas, direta e inversa, e mescle os resultados com $concatArrays para capturar a rede completa em uma agregação.

• Empurre os filtros para a travessia com restrictSearchWithMatch: podar ramificações mortas durante o BFS em vez de filtrar o grafo completo posteriormente para reduzir o conjunto de trabalho para redes grandes.

• Crie índices compostos em connectToField e seus filtros transversais: $graphLookup emite uma $match/$in query a cada onde de BFS, portanto, a indexação connectToField de junto com active e confidence evita varreduras de collection em cada salto.

• Faça a análise de rede do lado do servidor com $facet e a $switch: Execute a distribuição de risco, a detecção de hub e a pontuação de centralidade em subpipelines paralelos e use $switch para atribuir pesos de risco específicos do domínio a tipos de relacionamento dentro da agregação.

• Aplicar decaimento exponencial à propagação de riscos: Combine $graphLookupa descoberta de rede baseada em com uma fórmula de decaimento por salto que inclua confiança e tipo de relacionamento para que o modelo distinga automaticamente as conexões estruturas de alto risco dos links sociais de baixo risco. Isso corresponde à diretriz para manter os principais aprendizados em um estilo obrigatório consistente.

Autores

• Luis Pazmiño

• Mehar Crewal

• Andrea Alaman Calderon