SIAFI — Sistema Integrado de Administração Financeira da União

import requests
import pandas as pd
 
BASE_URL = "https://apidatalake.tesouro.gov.br/ords/sadipem/tt"
 
 
def consultar_execucao_despesa(exercicio: int, codigo_orgao: str = None):
    """
    Consulta a execução orçamentária de despesas da União.
 
    Args:
        exercicio: Ano do exercício fiscal
        codigo_orgao: Código SIAFI do órgão (opcional)
 
    Returns:
        DataFrame com dados da execução de despesas
    """
    url = f"{BASE_URL}/execucao-despesa"
    params = {
        "an_exercicio": exercicio,
        "limit": 5000,
    }
    if codigo_orgao:
        params["co_orgao"] = codigo_orgao
 
    registros = []
    offset = 0
 
    while True:
        params["offset"] = offset
        response = requests.get(url, params=params)
        response.raise_for_status()
        dados = response.json()
 
        items = dados.get("items", [])
        registros.extend(items)
 
        if not dados.get("hasMore", False):
            break
        offset += len(items)
 
    return pd.DataFrame(registros)
 
 
# Exemplo: execução de despesas do Ministério da Saúde (36000) em 2024
df = consultar_execucao_despesa(2024, "36000")
print(f"Registros: {len(df)}")
if not df.empty:
    print(df[["no_funcao", "no_subfuncao", "vl_empenhado", "vl_liquidado", "vl_pago"]].head(10))

def consultar_receita(exercicio: int):
    """
    Consulta as receitas arrecadadas pela União.
 
    Args:
        exercicio: Ano do exercício fiscal
 
    Returns:
        DataFrame com dados de receitas
    """
    url = f"{BASE_URL}/execucao-receita"
    params = {
        "an_exercicio": exercicio,
        "limit": 5000,
    }
 
    response = requests.get(url, params=params)
    response.raise_for_status()
    dados = response.json()
    registros = dados.get("items", dados)
    return pd.DataFrame(registros)
 
 
# Exemplo: receitas da União em 2024
df_receita = consultar_receita(2024)
print(f"Registros de receita: {len(df_receita)}")
if not df_receita.empty:
    print(df_receita.head())

def consultar_resultado_primario(exercicio_inicio: int, exercicio_fim: int):
    """
    Consulta o resultado primário do Governo Central ao longo de vários exercícios.
 
    Args:
        exercicio_inicio: Ano inicial
        exercicio_fim: Ano final
 
    Returns:
        DataFrame consolidado com resultados primários
    """
    todos = []
    for ano in range(exercicio_inicio, exercicio_fim + 1):
        url = f"{BASE_URL}/resultado-primario"
        params = {"an_exercicio": ano, "limit": 5000}
 
        response = requests.get(url, params=params)
        if response.status_code == 200:
            dados = response.json()
            items = dados.get("items", [])
            for item in items:
                item["exercicio"] = ano
            todos.extend(items)
 
    return pd.DataFrame(todos)
 
 
# Exemplo: resultado primário de 2019 a 2024
df_primario = consultar_resultado_primario(2019, 2024)
if not df_primario.empty:
    print(df_primario.groupby("exercicio").sum(numeric_only=True).head())

import requests
import pandas as pd
 
BASE_URL = "https://apidatalake.tesouro.gov.br/ords/sadipem/tt"
 
 
def comparar_dotacao_execucao(exercicio: int):
    """
    Compara a dotação orçamentária inicial com os valores efetivamente executados
    por função orçamentária.
 
    Args:
        exercicio: Ano do exercício fiscal
 
    Returns:
        DataFrame com dotação vs. execução por função
    """
    url = f"{BASE_URL}/execucao-despesa"
    params = {"an_exercicio": exercicio, "limit": 5000}
 
    registros = []
    offset = 0
 
    while True:
        params["offset"] = offset
        response = requests.get(url, params=params)
        response.raise_for_status()
        dados = response.json()
        items = dados.get("items", [])
        registros.extend(items)
        if not dados.get("hasMore", False):
            break
        offset += len(items)
 
    df = pd.DataFrame(registros)
    if df.empty:
        return df
 
    # Agrupar por função orçamentária
    resumo = (
        df.groupby("no_funcao")
        .agg(
            dotacao_inicial=("vl_dotacao_inicial", "sum"),
            empenhado=("vl_empenhado", "sum"),
            pago=("vl_pago", "sum"),
        )
        .reset_index()
    )
    resumo["taxa_execucao_%"] = (resumo["pago"] / resumo["dotacao_inicial"] * 100).round(2)
    return resumo.sort_values("dotacao_inicial", ascending=False)
 
 
# Exemplo: comparação dotação x execução em 2024
resumo = comparar_dotacao_execucao(2024)
if not resumo.empty:
    print("Execução orçamentária por função (R$):")
    print(resumo.head(10).to_string(index=False))