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. 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. 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 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, } 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 ) 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 ) 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. 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} 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 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 ( {data.nodes.map(node => ( ))} {data.edges.map((edge, i) => ( ))} ); } 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. 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.