Bens Declarados

https://cdn.tse.jus.br/estatistica/sead/odsele/bem_candidato/bem_candidato_2022.zip
https://cdn.tse.jus.br/estatistica/sead/odsele/bem_candidato/bem_candidato_2020.zip
https://cdn.tse.jus.br/estatistica/sead/odsele/bem_candidato/bem_candidato_2018.zip

import requests
import zipfile
import pandas as pd
from io import BytesIO
from pathlib import Path
 
 
def baixar_bens(ano: int, destino: Path = Path("./dados_tse")) -> pd.DataFrame:
    """
    Baixa e carrega dados de bens declarados por candidatos.
 
    Args:
        ano: Ano da eleição (ex: 2022, 2020, 2018)
        destino: Diretório para salvar os arquivos
 
    Returns:
        DataFrame com bens declarados de todos os candidatos do ano
    """
    url = (
        f"https://cdn.tse.jus.br/estatistica/sead/odsele/"
        f"bem_candidato/bem_candidato_{ano}.zip"
    )
    print(f"Baixando bens declarados {ano}...")
 
    response = requests.get(url)
    response.raise_for_status()
 
    destino.mkdir(parents=True, exist_ok=True)
 
    with zipfile.ZipFile(BytesIO(response.content)) as zf:
        arquivo_brasil = [
            f for f in zf.namelist()
            if "BRASIL" in f.upper() and f.endswith(".csv")
        ]
        if arquivo_brasil:
            zf.extract(arquivo_brasil[0], destino)
            caminho_csv = destino / arquivo_brasil[0]
        else:
            zf.extractall(destino)
            csvs = [f for f in zf.namelist() if f.endswith(".csv")]
            caminho_csv = destino / csvs[0]
 
    df = pd.read_csv(
        caminho_csv,
        sep=";",
        encoding="latin-1",
        dtype=str,
    )
 
    print(f"Registros de bens carregados: {len(df):,}")
    return df
 
 
# Baixar bens declarados nas eleições de 2022
df_bens = baixar_bens(2022)
print(df_bens.head())

def patrimonio_por_candidato(
    df: pd.DataFrame,
    cargo: str = None,
    top_n: int = 20,
) -> pd.DataFrame:
    """
    Calcula o patrimônio total declarado por candidato.
 
    Args:
        df: DataFrame com bens declarados
        cargo: Filtrar por cargo (opcional)
        top_n: Quantidade de candidatos a retornar
 
    Returns:
        DataFrame com patrimônio total, ordenado decrescente
    """
    df["VR_BEM_CANDIDATO"] = pd.to_numeric(
        df["VR_BEM_CANDIDATO"].str.replace(",", "."), errors="coerce"
    )
 
    resultado = df.copy()
    if cargo:
        # Juntar com base de candidaturas para filtrar por cargo
        resultado = resultado[resultado.get("DS_CARGO", pd.Series()).str.upper() == cargo.upper()]
 
    patrimonio = (
        resultado.groupby(["SQ_CANDIDATO", "NM_CANDIDATO"])
        .agg(
            patrimonio_total=("VR_BEM_CANDIDATO", "sum"),
            qtd_bens=("VR_BEM_CANDIDATO", "count"),
        )
        .reset_index()
        .sort_values("patrimonio_total", ascending=False)
    )
 
    return patrimonio.head(top_n)
 
 
top_patrimonio = patrimonio_por_candidato(df_bens)
print("Top 20 candidatos por patrimônio declarado — 2022:")
for _, row in top_patrimonio.iterrows():
    print(f"  {row['NM_CANDIDATO']}: R$ {row['patrimonio_total']:,.2f} ({int(row['qtd_bens'])} bens)")

def evolucao_patrimonial(
    nome_candidato: str,
    anos: list[int] = [2014, 2018, 2022],
) -> pd.DataFrame:
    """
    Acompanha a evolução patrimonial de um candidato ao longo de eleições.
 
    Args:
        nome_candidato: Nome do candidato (busca parcial, case-insensitive)
        anos: Lista de anos eleitorais para comparação
 
    Returns:
        DataFrame com patrimônio total por ano
    """
    evolucao = []
 
    for ano in anos:
        df = baixar_bens(ano)
        df["VR_BEM_CANDIDATO"] = pd.to_numeric(
            df["VR_BEM_CANDIDATO"].str.replace(",", "."), errors="coerce"
        )
 
        filtro = df["NM_CANDIDATO"].str.upper().str.contains(
            nome_candidato.upper(), na=False
        )
        bens_candidato = df[filtro]
 
        if not bens_candidato.empty:
            patrimonio = bens_candidato["VR_BEM_CANDIDATO"].sum()
            qtd_bens = len(bens_candidato)
            evolucao.append({
                "ano": ano,
                "nome": bens_candidato.iloc[0]["NM_CANDIDATO"],
                "patrimonio_total": patrimonio,
                "qtd_bens": qtd_bens,
            })
 
    df_evolucao = pd.DataFrame(evolucao)
    if len(df_evolucao) > 1:
        df_evolucao["variacao_pct"] = df_evolucao["patrimonio_total"].pct_change() * 100
    return df_evolucao
 
 
# Exemplo: evolução patrimonial de um candidato
resultado = evolucao_patrimonial("NOME DO CANDIDATO", [2014, 2018, 2022])
print(resultado)

import pandas as pd
 
# 1. Carregar bens declarados
df_bens = pd.read_csv(
    "dados_tse/bem_candidato_2022_BRASIL.csv",
    sep=";",
    encoding="latin-1",
    dtype=str,
)
 
# 2. Filtrar bens do tipo "participação societária" ou "quotas de capital"
participacoes = df_bens[
    df_bens["DS_TIPO_BEM_CANDIDATO"].str.upper().str.contains(
        "QUOTA|AÇÃO|AÇÕES|PARTICIPAÇÃO|SOCIETÁRI", na=False
    )
].copy()
 
# 3. Carregar QSA da Receita Federal
df_qsa = pd.read_csv(
    "dados_rfb/Socios0.csv",
    sep=";",
    header=None,
    names=[
        "cnpj_basico", "identificador_socio", "nome_socio",
        "cpf_cnpj_socio", "qualificacao_socio", "data_entrada",
        "pais", "representante_legal", "nome_representante",
        "qualificacao_representante", "faixa_etaria",
    ],
    dtype=str,
    encoding="latin-1",
)
 
# 4. Buscar CPFs de candidatos que declararam participações
cpfs_com_participacao = participacoes["NR_CPF_CANDIDATO"].dropna().unique()
 
# 5. Cruzar com QSA para verificar se constam como sócios
socios_candidatos = df_qsa[
    df_qsa["cpf_cnpj_socio"].isin(cpfs_com_participacao)
]
 
# 6. Identificar candidatos que NÃO declararam todas as participações
cpfs_no_qsa = socios_candidatos["cpf_cnpj_socio"].unique()
cpfs_nao_declararam = set(cpfs_no_qsa) - set(cpfs_com_participacao)
 
print(f"Candidatos com participação declarada: {len(cpfs_com_participacao)}")
print(f"Candidatos encontrados no QSA: {len(cpfs_no_qsa)}")
print(f"Possíveis omissões (no QSA mas sem declaração): {len(cpfs_nao_declararam)}")