Simples Nacional / MEI

https://arquivos.receitafederal.gov.br/dados/cnpj/dados_abertos_cnpj/Simples.zip

import requests
import zipfile
import pandas as pd
from io import BytesIO
from pathlib import Path
 
# Colunas do arquivo Simples conforme layout da Receita Federal
COLUNAS_SIMPLES = [
    "cnpj_basico",
    "opcao_simples",
    "data_opcao_simples",
    "data_exclusao_simples",
    "opcao_mei",
    "data_opcao_mei",
    "data_exclusao_mei",
]
 
def baixar_e_extrair(url: str, destino: Path) -> Path:
    """Baixa um arquivo ZIP e extrai o CSV contido nele."""
    print(f"Baixando {url}...")
    response = requests.get(url, stream=True)
    response.raise_for_status()
 
    with zipfile.ZipFile(BytesIO(response.content)) as zf:
        nome_csv = zf.namelist()[0]
        zf.extract(nome_csv, destino)
        print(f"Extraído: {destino / nome_csv}")
        return destino / nome_csv
 
 
# Baixar arquivo do Simples Nacional
url = "https://arquivos.receitafederal.gov.br/dados/cnpj/dados_abertos_cnpj/Simples.zip"
caminho_csv = baixar_e_extrair(url, Path("./dados_rfb"))
 
# Ler o CSV (sem cabeçalho, separador ";", encoding Latin-1)
df_simples = pd.read_csv(
    caminho_csv,
    sep=";",
    header=None,
    names=COLUNAS_SIMPLES,
    dtype=str,
    encoding="latin-1",
)
 
print(f"Registros carregados: {len(df_simples):,}")
print(df_simples.head())

def filtrar_optantes_simples(df: pd.DataFrame) -> pd.DataFrame:
    """
    Retorna apenas as empresas atualmente optantes pelo Simples Nacional.
    Critério: opcao_simples == 'S' e sem data de exclusão.
    """
    return df[
        (df["opcao_simples"] == "S") &
        (df["data_exclusao_simples"].isna() | (df["data_exclusao_simples"] == ""))
    ]
 
 
optantes_simples = filtrar_optantes_simples(df_simples)
print(f"Empresas optantes pelo Simples Nacional: {len(optantes_simples):,}")

def filtrar_mei(df: pd.DataFrame) -> pd.DataFrame:
    """
    Retorna apenas os MEIs atualmente ativos.
    Critério: opcao_mei == 'S' e sem data de exclusão.
    """
    return df[
        (df["opcao_mei"] == "S") &
        (df["data_exclusao_mei"].isna() | (df["data_exclusao_mei"] == ""))
    ]
 
 
meis_ativos = filtrar_mei(df_simples)
print(f"MEIs ativos: {len(meis_ativos):,}")

# Contagem por situação
print("=== Simples Nacional ===")
print(df_simples["opcao_simples"].value_counts())
print(f"\n=== MEI ===")
print(df_simples["opcao_mei"].value_counts())
 
# Evolução temporal: agrupar por ano de opção
df_simples["ano_opcao_simples"] = df_simples["data_opcao_simples"].str[:4]
evolucao = (
    df_simples[df_simples["opcao_simples"] == "S"]
    .groupby("ano_opcao_simples")
    .size()
    .reset_index(name="total")
)
print("\nNovas opções pelo Simples Nacional por ano:")
print(evolucao.tail(10))

def verificar_regime(df: pd.DataFrame, cnpj_basico: str) -> dict:
    """
    Verifica a situação de uma empresa no Simples Nacional e MEI.
 
    Args:
        df: DataFrame com dados do Simples
        cnpj_basico: 8 primeiros dígitos do CNPJ
 
    Returns:
        Dicionário com a situação do regime tributário
    """
    registro = df[df["cnpj_basico"] == cnpj_basico.zfill(8)]
    if registro.empty:
        return {"encontrado": False, "simples": None, "mei": None}
 
    r = registro.iloc[0]
    return {
        "encontrado": True,
        "simples": r["opcao_simples"] == "S",
        "data_opcao_simples": r["data_opcao_simples"],
        "data_exclusao_simples": r["data_exclusao_simples"],
        "mei": r["opcao_mei"] == "S",
        "data_opcao_mei": r["data_opcao_mei"],
        "data_exclusao_mei": r["data_exclusao_mei"],
    }
 
 
# Exemplo
resultado = verificar_regime(df_simples, "12345678")
if resultado["encontrado"]:
    print(f"Optante Simples: {'Sim' if resultado['simples'] else 'Não'}")
    print(f"Optante MEI: {'Sim' if resultado['mei'] else 'Não'}")
else:
    print("Empresa não encontrada na base do Simples Nacional.")

import pandas as pd
 
# Carregar dados do Simples (previamente baixados)
df_simples = pd.read_csv(
    "dados_rfb/Simples.csv",
    sep=";", header=None,
    names=["cnpj_basico", "opcao_simples", "data_opcao_simples",
           "data_exclusao_simples", "opcao_mei", "data_opcao_mei",
           "data_exclusao_mei"],
    dtype=str, encoding="latin-1",
)
 
# Carregar dados de empresas (previamente baixados)
df_empresas = pd.read_csv(
    "dados_rfb/Empresas0.csv",
    sep=";", header=None,
    names=["cnpj_basico", "razao_social", "natureza_juridica",
           "qualificacao_responsavel", "capital_social",
           "porte_empresa", "ente_federativo_responsavel"],
    dtype=str, encoding="latin-1",
)
 
# Cruzar: MEIs com dados cadastrais
meis = df_simples[df_simples["opcao_mei"] == "S"]
meis_com_dados = meis.merge(
    df_empresas[["cnpj_basico", "razao_social", "capital_social", "porte_empresa"]],
    on="cnpj_basico",
    how="inner",
)
 
print(f"MEIs com dados cadastrais encontrados: {len(meis_com_dados):,}")
print(meis_com_dados[["cnpj_basico", "razao_social", "data_opcao_mei"]].head(10))
 
# Distribuição de porte entre optantes do Simples
optantes = df_simples[df_simples["opcao_simples"] == "S"]
optantes_com_porte = optantes.merge(
    df_empresas[["cnpj_basico", "porte_empresa"]],
    on="cnpj_basico",
    how="inner",
)
portes = {
    "00": "Não informado",
    "01": "Micro empresa",
    "03": "Empresa de pequeno porte",
    "05": "Demais",
}
optantes_com_porte["porte_desc"] = optantes_com_porte["porte_empresa"].map(portes)
print("\nDistribuição por porte:")
print(optantes_com_porte["porte_desc"].value_counts())