Uma das funcionalidades mais interessantes do CNPJ Aberto é a rede societária — dado um CNPJ, o sistema mapeia todos os sócios da empresa, encontra outras empresas desses mesmos sócios, e constrói um grafo de conexões.
Isso transforma dados tabulares (CSV da Receita Federal) em inteligência empresarial. Advogados usam para due diligence, jornalistas para investigação, e analistas de crédito para avaliação de risco.
Neste post, vou mostrar como construímos esse sistema.
O modelo de dados
A base da Receita Federal tem uma relação simples:
empresas (cnpj_basico) ←──1:N──→ socios (cnpj_basico, nome_socio)
Cada empresa tem N sócios. Cada sócio pode aparecer em múltiplas empresas (identificado pelo nome). E um sócio pode ser uma pessoa jurídica (outra empresa), criando conexões indiretas.
Empresa A ← sócio "João Silva" → Empresa B
Empresa B ← sócio PJ "Empresa C" → Empresa C
Empresa C ← sócio "Maria Santos" → Empresa D
Isso forma um grafo que pode revelar estruturas corporativas complexas.
Modelagem do grafo
from dataclasses import dataclass
@dataclass
class GrupoNode:
id: str
tipo: str # "empresa" | "pessoa"
label: str # razão social ou nome
cnpj: str | None
situacao_cadastral: str | None
capital_social: float | None
uf: str | None
is_target: bool # é a empresa que o usuário consultou?
@dataclass
class GrupoEdge:
source: str # node ID
target: str # node ID
label: str # qualificação do sócio
O grafo é uma lista de nodes (empresas e pessoas) e edges (relações societárias). Simples e serializável para JSON.
O traversal recursivo
O algoritmo começa em uma empresa e expande recursivamente:
MAX_DEPTH = 2
MAX_NETWORK_NODES = 150
async def get_grupo_empresarial(cnpj_basico: str, db):
nodes = {}
edges = []
await traverse_company(
cnpj_basico, db, nodes, edges, depth=0, is_target=True
)
return {
"nodes": list(nodes.values()),
"edges": edges,
}
traverse_company: o coração do algoritmo
async def traverse_company(cnpj_basico, db, nodes, edges, depth, is_target=False):
if depth > MAX_DEPTH:
return
if len(nodes) >= MAX_NETWORK_NODES:
return
company_id = f"emp:{cnpj_basico}"
if company_id in nodes:
return # Já visitado — evita ciclos
# 1. Buscar dados da empresa
empresa = db.query(Empresa).filter(
Empresa.cnpj_basico == cnpj_basico
).first()
if not empresa:
return
matriz = db.query(Estabelecimento).filter(
Estabelecimento.cnpj_basico == cnpj_basico,
Estabelecimento.identificador_matriz_filial == "1"
).first()
# Adicionar node da empresa
nodes[company_id] = GrupoNode(
id=company_id,
tipo="empresa",
label=empresa.razao_social,
cnpj=format_cnpj(cnpj_basico),
situacao_cadastral=matriz.situacao_cadastral if matriz else None,
capital_social=empresa.capital_social,
uf=matriz.uf if matriz else None,
is_target=is_target,
)
# 2. Buscar sócios desta empresa
socios = db.query(Socio).filter(
Socio.cnpj_basico == cnpj_basico
).all()
for socio in socios:
if len(nodes) >= MAX_NETWORK_NODES:
break
if socio.identificador_socio == "1" and socio.cpf_cnpj_socio:
# Sócio é PESSOA JURÍDICA → seguir como outra empresa
socio_cnpj = socio.cpf_cnpj_socio[:8]
socio_id = f"emp:{socio_cnpj}"
edges.append(GrupoEdge(
source=socio_id,
target=company_id,
label=socio.qualificacao or "Sócio PJ",
))
# Recursão: explorar a empresa sócia
await traverse_company(
socio_cnpj, db, nodes, edges, depth + 1
)
else:
# Sócio é PESSOA FÍSICA
person_id = f"pf:{sanitize(socio.nome_socio)}"
if person_id not in nodes:
nodes[person_id] = GrupoNode(
id=person_id,
tipo="pessoa",
label=socio.nome_socio,
cnpj=None,
situacao_cadastral=None,
capital_social=None,
uf=None,
is_target=False,
)
edges.append(GrupoEdge(
source=person_id,
target=company_id,
label=socio.qualificacao or "Sócio",
))
# 3. Buscar OUTRAS empresas desta pessoa
await expand_person(
socio.nome_socio, cnpj_basico, db,
nodes, edges, depth
)
expand_person: encontrando empresas conectadas
async def expand_person(nome_socio, exclude_cnpj, db, nodes, edges, depth):
# Buscar outros cnpj_basico onde esta pessoa é sócia
other_companies = db.query(Socio.cnpj_basico).filter(
Socio.nome_socio == nome_socio,
Socio.cnpj_basico != exclude_cnpj,
).distinct().limit(10).all()
person_id = f"pf:{sanitize(nome_socio)}"
for row in other_companies:
if len(nodes) >= MAX_NETWORK_NODES:
break
company_id = f"emp:{row.cnpj_basico}"
edges.append(GrupoEdge(
source=person_id,
target=company_id,
label="Sócio",
))
# Continuar traversal na empresa encontrada
await traverse_company(
row.cnpj_basico, db, nodes, edges, depth + 1
)
Proteções contra explosão combinatória
Sem limitações, o grafo pode explodir. Um sócio que aparece em 200 empresas, cada empresa com 5 sócios, cada sócio em 10 empresas... rapidamente vira milhões de nodes.
As proteções:
| Proteção | Valor | Por quê |
|---|---|---|
MAX_DEPTH |
2 | Limita a profundidade da recursão |
MAX_NETWORK_NODES |
150 | Cap total de nodes no grafo |
LIMIT 10 em expand_person
|
10 | Limita empresas por pessoa |
Checagem if company_id in nodes
|
— | Evita ciclos e re-processamento |
Depth 2 é suficiente? Na prática, sim. A maioria das estruturas societárias interessantes fica a 1-2 hops de distância. Ir mais fundo geralmente adiciona ruído sem valor.
Detecção de Red Flags
Com o grafo pronto, podemos detectar padrões suspeitos:
def detect_red_flags(cnpj_basico, db):
flags = []
# 1. Sócio em muitas empresas (possível laranja)
socios = db.query(
Socio.nome_socio, func.count(distinct(Socio.cnpj_basico))
).filter(
Socio.cnpj_basico == cnpj_basico
).group_by(Socio.nome_socio).all()
for nome, count in socios:
total = db.query(func.count(distinct(Socio.cnpj_basico))).filter(
Socio.nome_socio == nome
).scalar()
if total >= 5:
flags.append({
"tipo": "socio_multiplas_empresas",
"severidade": "media",
"titulo": f"Sócio em {total} empresas",
"descricao": f"{nome} é sócio em {total} empresas diferentes",
})
# 2. Muitas empresas no mesmo endereço
matriz = get_matriz(cnpj_basico, db)
if matriz and matriz.cep and matriz.logradouro:
same_address = db.query(func.count()).filter(
Estabelecimento.cep == matriz.cep,
Estabelecimento.logradouro == matriz.logradouro,
Estabelecimento.numero == matriz.numero,
Estabelecimento.cnpj_basico != cnpj_basico,
Estabelecimento.identificador_matriz_filial == "1",
).scalar()
if same_address >= 3:
flags.append({
"tipo": "concentracao_endereco",
"severidade": "baixa",
"titulo": f"{same_address} empresas no mesmo endereço",
"descricao": "Concentração incomum de empresas",
})
# 3. Contato compartilhado (email ou telefone)
if matriz and matriz.email:
shared = db.query(func.count(distinct(
Estabelecimento.cnpj_basico
))).filter(
Estabelecimento.email == matriz.email,
Estabelecimento.cnpj_basico != cnpj_basico,
).scalar()
if shared >= 1:
flags.append({
"tipo": "contato_compartilhado",
"severidade": "baixa",
"titulo": "Email usado por outra empresa",
"descricao": f"O email {matriz.email} aparece em {shared + 1} empresas",
})
# Calcular score de risco (0-100)
score = sum(
35 if f["severidade"] == "alta" else
20 if f["severidade"] == "media" else 10
for f in flags
)
return {"score": min(score, 100), "flags": flags}
Cache: essencial para grafos
A construção do grafo envolve múltiplas queries recursivas. Sem cache, cada visualização levaria 1-3 segundos. Com Redis:
async def get_grupo_cached(cnpj_basico, db):
cache_key = f"grupo:{cnpj_basico}"
cached = redis.get(cache_key)
if cached:
return json.loads(cached)
result = await get_grupo_empresarial(cnpj_basico, db)
# Cache por 24h — dados mudam mensalmente
redis.setex(cache_key, 86400, json.dumps(result))
return result
Frontend: visualização do grafo
O JSON {nodes, edges} é renderizado no frontend com uma biblioteca de grafos. O componente React recebe os dados e renderiza nodes como cards e edges como linhas de conexão:
// Simplificado
function CorporateGroup({ cnpj }) {
const [data, setData] = useState(null);
useEffect(() => {
fetch(`/api/intelligence/grupo/${cnpj}`)
.then(r => r.json())
.then(setData);
}, [cnpj]);
if (!data || data.nodes.length === 0) return null;
return (
<div className="relative">
{data.nodes.map(node => (
<NodeCard key={node.id} node={node} />
))}
{data.edges.map((edge, i) => (
<EdgeLine key={i} edge={edge} nodes={data.nodes} />
))}
</div>
);
}
Empresas ativas são destacadas em verde, inativas em vermelho. A empresa alvo da consulta fica em destaque. Sócios PF são mostrados com ícone de pessoa.
Casos de uso reais
Esse tipo de análise de rede revela coisas que dados tabulares escondem:
Due diligence: "O sócio da empresa que estou contratando também é sócio de uma empresa com situação 'Inapta'?"
Investigação: "Quem são as pessoas por trás de um grupo de empresas com o mesmo endereço?"
Análise de crédito: "Este solicitante de empréstimo tem sócios com histórico de empresas encerradas?"
Compliance: "O fornecedor que estamos avaliando tem conexões com empresas em situação irregular?"
Conclusão
Construir um explorador de rede societária envolveu:
- Traversal recursivo com proteções contra explosão (depth, max nodes, ciclos)
- Dual-type nodes (empresa vs pessoa) com expansão bidirecional
- Heurísticas de red flags baseadas em padrões estatísticos
- Cache agressivo — grafos são caros de construir e mudam raramente
- Limites pragmáticos — depth 2 e 150 nodes cobrem 95% dos casos úteis
A beleza é que tudo isso roda sobre dados públicos. Não há scraping, não há API paga, não há magia. São dados da Receita Federal, organizados e conectados de forma que se tornam inteligência de verdade.
Quer explorar a rede societária de qualquer empresa brasileira? Teste no CNPJ Aberto de forma gratuita.
Top comments (0)