Checagem de Risco e Crédito

A saída não é um PDF que alguém precisa abrir. O pipeline grava atributos tipados (situação, faixa de risco, score de crédito, regras sinalizadas) diretamente no cadastro do fornecedor no ERP ou da conta do cliente no CRM. O analista vê BLOQUEADO, EM_ANÁLISE ou LIBERADO na mesma tela de sempre, sem trocar de sistema e sem pedir parecer por e-mail.

Em due diligence é normal ter duas fontes para o mesmo campo: um bureau tem o processo judicial, outro tem a situação cadastral mais atualizada. A ordem das fontes é a política de confiança, escrita no ponto de configuração e não escondida num if. Cada valor do registro final guarda de qual fonte veio, então "por que essa contraparte foi reprovada" vira uma coluna da tabela de saída, não uma arqueologia de log de rede. Fonte que falha entra em fontes_com_falha e as demais seguem: perder um enriquecimento opcional não é o mesmo que perder a análise.

fontes = [
    RestBureauSource(
        nome="bureau_cadastral",
        base_url=BUREAU_URL,
        transport=http_json,
        field_map={"cad_situacao": "situacao_cadastral", ...},
    ),
    McpToolSource(
        nome="mcp_sancoes",
        server_url=MCP_URL,
        tool_name="consultar_sancoes",
        transport=http_json,   # Accept: application/json, text/event-stream
        field_map={"sanctioned": "possui_sancao_ativa"},
    ),
]

merged = coletar_de_fontes(documento, tipo_documento, fontes)
merged.field_provenance["situacao_cadastral"]  # "bureau_cadastral"

Sem nenhuma fonte respondendo, a situação cadastral fica DESCONHECIDA em vez de ATIVA, o que faz a regra de irregularidade disparar: ausência de dado não é ausência de risco.

Rodam na análise de cliente, que é onde existe exposição financeira. São determinísticos e não um modelo treinado por um motivo concreto: não há aqui uma base real de inadimplência rotulada, e um scorecard de pesos explícitos é a alternativa honesta a fabricar um dataset de treino que não existe. Cada modelo devolve, junto do score, o racional em texto que o analista lê para revisar a decisão.

No score final do cliente, compliance pesa 0.4 e crédito 0.6. Só que 0.6 sozinho nunca alcança o limiar de rejeição (0.66), o que prenderia em revisão manual uma contraparte com os 4 modelos no teto. Por isso existe um limiar independente: crédito isolado acima de 0.85 rejeita sozinho, exatamente como um veto de compliance.

Não é um bureau: monta buscas por nome da contraparte e palavras-chave agrupadas em 6 categorias de risco (corrupção, lavagem de dinheiro, fraude, envolvimento criminal, sanções, crime organizado), 5 grupos de termos cada. A intensidade satura a partir de 3 artigos corroborantes, porque uma menção isolada não pesa como três reportagens independentes. Categorias de maior severidade viram veto automático; as demais contribuem ponderadamente.

"Fornecedor Exemplo LTDA" ("corrupção" OR "suborno" OR "propina" OR
  "peculato" OR "improbidade" OR "desvio de verbas")

O mapeamento é um dicionário único e explícito: é ele que o time de integração lê quando pergunta qual campo do CRM recebe o quê. Adicionar uma regra ao motor não altera o contrato. Campo ausente é omitido em vez de virar nulo, porque KYS e KYE não rodam crédito e enviar null apagaria o score gravado por uma análise de cliente anterior da mesma contraparte.

risco_documento          12345678000199
risco_tipo_pessoa        PJ
risco_status             BLOQUEADO      -- LIBERADO | EM_ANALISE | BLOQUEADO
risco_faixa              ALTO           -- BAIXO | MEDIO | ALTO
risco_score              0.91
risco_score_credito      0.88
risco_regras_sinalizadas ["sancao_ativa", "divida_ativa_registrada"]
risco_midia_categorias   ["lavagem_dinheiro_crimes_financeiros"]
risco_origem_analise     kyc
risco_versao_regras      1.0.0
risco_atualizado_em      2026-01-14T03:22:10Z

payload_hash             sha256 do conteúdo de negócio, sem timestamp
                         -> reprocessar sem mudança grava 0 linha

A chave do delta inclui o conjunto de regras: a mesma empresa pode ser fornecedor e cliente ao mesmo tempo, e a carga de uma análise de cliente não pode suprimir a de fornecedor.

Consultar uma contraparte por vez serve para uma análise pontual, não para reavaliar a carteira inteira periodicamente. O lote é distribuído entre partições via mapInPandas, que reaproveita a mesma sessão HTTP para todas as linhas de uma partição em vez de abrir uma conexão por linha. A escolha por mapInPandas em vez de rdd.mapPartitions veio de um erro real em compute serverless, que não expõe a API de RDD.

Cada módulo separa a lógica pura (retry, avaliação de regra, agregação de mídia, scorecard de crédito, merge de fontes, hash de idempotência) da integração com Spark. A primeira nunca importa PySpark, e o transporte HTTP é injetável, então nem rede nem mock de rede entram nos testes. Resultado: 74 testes cobrindo cada decisão de negócio rodam com pytest puro em menos de 1 segundo, sem subir Spark nem Java.