Tools: Como eu resolvi o limite de 10 anos da API do Banco Central do Brasil com Python

O problema em detalhes ## A solução: paginação automática ## Funcionalidades principais ## 1. Aliases para séries populares ## 2. Múltiplas séries de uma vez ## 3. Últimos N valores ## 4. Cache inteligente ## 5. Metadados das séries ## Como funciona por dentro ## A paginação em detalhe ## Stack técnica ## API REST ## Dashboard interativo ## Como contribuir Em março de 2025, o Banco Central do Brasil implementou uma mudança silenciosa na API SGS (Sistema Gerenciador de Séries Temporais): consultas passaram a ser limitadas a no máximo 10 anos por requisição. Para quem trabalha com séries históricas curtas, nada mudou. Mas se você precisa da cotação do Dólar desde 1984, da Selic desde 1986 ou do IPCA desde 1980 — como é comum em análise econômica, modelagem de risco ou estudos acadêmicos — essa limitação se tornou um problema real. Criei o BacenData para resolver isso de forma transparente. A API SGS do Banco Central é pública e gratuita. Você faz uma requisição HTTP com o código da série, data inicial e data final, e recebe um JSON com os valores. Até 2025, era possível consultar qualquer período: Agora, se o intervalo for maior que 10 anos, a API retorna erro 400 ou uma resposta vazia. Para obter dados completos do Dólar (código 1), que existe desde 1984, você precisaria fazer pelo menos 5 requisições separadas, concatenar os resultados e remover duplicatas. É exatamente isso que o BacenData automatiza. Por baixo dos panos, o BacenData: O resultado é um pandas.DataFrame limpo, pronto para análise. Em vez de decorar códigos numéricos, use nomes: O BacenData vem com 14 séries pré-configuradas: O BacenData mantém um cache SQLite local com TTL automático por tipo de série: A arquitetura é simples e pragmática: O coração do BacenData é a função _buscar_com_paginacao: O Semaphore(5) é crucial: sem ele, uma consulta de 40 anos geraria 5 requisições simultâneas de uma vez, podendo sobrecarregar o servidor do BACEN. Com o semáforo, nunca passamos de 5 conexões paralelas. O try/except no fetch_com_semaforo resolve outro problema sutil: quando o usuário pede "todos os dados disponíveis", o BacenData gera intervalos desde 1960. Para séries que começaram em 1984, os intervalos de 1960-1970 e 1970-1980 retornam 404. Sem o try/except, o asyncio.gather falharia inteiramente. O BacenData também expõe uma API REST para integração com qualquer linguagem/sistema: Para quem não programa, o BacenData tem um dashboard Streamlit público: https://bacendata.streamlit.app O projeto é open source (MIT). Algumas ideias para contribuição: Se você trabalha com dados econômicos no Brasil — como economista, analista, desenvolvedor ou pesquisador — o BacenData pode te poupar horas de trabalho manual. Dá uma olhada e me conta o que achou. Templates let you quickly answer FAQs or store snippets for re-use. Are you sure you want to hide this comment? It will become hidden in your post, but will still be visible via the comment's permalink. Hide child comments as well For further actions, you may consider blocking this person and/or reporting abuse CODE_BLOCK: GET https://api.bcb.gov.br/dados/serie/bcdata.sgs.1/dados?formato=json&dataInicial=01/01/1984&dataFinal=31/12/2025 Enter fullscreen mode Exit fullscreen mode CODE_BLOCK: GET https://api.bcb.gov.br/dados/serie/bcdata.sgs.1/dados?formato=json&dataInicial=01/01/1984&dataFinal=31/12/2025 CODE_BLOCK: GET https://api.bcb.gov.br/dados/serie/bcdata.sgs.1/dados?formato=json&dataInicial=01/01/1984&dataFinal=31/12/2025 COMMAND_BLOCK: pip install bacendata Enter fullscreen mode Exit fullscreen mode COMMAND_BLOCK: pip install bacendata COMMAND_BLOCK: pip install bacendata COMMAND_BLOCK: from bacendata import sgs # Série completa do Dólar — 40+ anos em uma chamada dolar = sgs.get("dolar", start="1984-01-01") print(f"{len(dolar)} registros de {dolar.index[0]} a {dolar.index[-1]}") Enter fullscreen mode Exit fullscreen mode COMMAND_BLOCK: from bacendata import sgs # Série completa do Dólar — 40+ anos em uma chamada dolar = sgs.get("dolar", start="1984-01-01") print(f"{len(dolar)} registros de {dolar.index[0]} a {dolar.index[-1]}") COMMAND_BLOCK: from bacendata import sgs # Série completa do Dólar — 40+ anos em uma chamada dolar = sgs.get("dolar", start="1984-01-01") print(f"{len(dolar)} registros de {dolar.index[0]} a {dolar.index[-1]}") COMMAND_BLOCK: # Estes são equivalentes: sgs.get(1, start="2020-01-01") sgs.get("dolar", start="2020-01-01") Enter fullscreen mode Exit fullscreen mode COMMAND_BLOCK: # Estes são equivalentes: sgs.get(1, start="2020-01-01") sgs.get("dolar", start="2020-01-01") COMMAND_BLOCK: # Estes são equivalentes: sgs.get(1, start="2020-01-01") sgs.get("dolar", start="2020-01-01") CODE_BLOCK: df = sgs.get({"Selic": 11, "IPCA": 433, "Dólar": 1}, start="2015-01-01") print(df.head()) Enter fullscreen mode Exit fullscreen mode CODE_BLOCK: df = sgs.get({"Selic": 11, "IPCA": 433, "Dólar": 1}, start="2015-01-01") print(df.head()) CODE_BLOCK: df = sgs.get({"Selic": 11, "IPCA": 433, "Dólar": 1}, start="2015-01-01") print(df.head()) CODE_BLOCK: Selic IPCA Dólar data 2015-01-02 0.046189 NaN 2.694 2015-01-05 0.046189 NaN 2.702 ... Enter fullscreen mode Exit fullscreen mode CODE_BLOCK: Selic IPCA Dólar data 2015-01-02 0.046189 NaN 2.694 2015-01-05 0.046189 NaN 2.702 ... CODE_BLOCK: Selic IPCA Dólar data 2015-01-02 0.046189 NaN 2.694 2015-01-05 0.046189 NaN 2.702 ... COMMAND_BLOCK: # Últimos 12 valores do IPCA ipca = sgs.get("ipca", last=12) Enter fullscreen mode Exit fullscreen mode COMMAND_BLOCK: # Últimos 12 valores do IPCA ipca = sgs.get("ipca", last=12) COMMAND_BLOCK: # Últimos 12 valores do IPCA ipca = sgs.get("ipca", last=12) COMMAND_BLOCK: # Desativar cache se necessário from bacendata.wrapper.cache import desativar desativar() Enter fullscreen mode Exit fullscreen mode COMMAND_BLOCK: # Desativar cache se necessário from bacendata.wrapper.cache import desativar desativar() COMMAND_BLOCK: # Desativar cache se necessário from bacendata.wrapper.cache import desativar desativar() COMMAND_BLOCK: meta = sgs.metadata(1) print(meta) # {'codigo': 1, 'nome': 'Taxa de câmbio - Livre - Dólar americano (venda) - diário', # 'periodicidade': 'D', 'unidade': 'u.m.c./US$', ...} Enter fullscreen mode Exit fullscreen mode COMMAND_BLOCK: meta = sgs.metadata(1) print(meta) # {'codigo': 1, 'nome': 'Taxa de câmbio - Livre - Dólar americano (venda) - diário', # 'periodicidade': 'D', 'unidade': 'u.m.c./US$', ...} COMMAND_BLOCK: meta = sgs.metadata(1) print(meta) # {'codigo': 1, 'nome': 'Taxa de câmbio - Livre - Dólar americano (venda) - diário', # 'periodicidade': 'D', 'unidade': 'u.m.c./US$', ...} COMMAND_BLOCK: bacendata/ ├── wrapper/ │ ├── bacen_sgs.py # Core: get(), aget(), metadata() │ ├── catalogo.py # 14 séries pré-configuradas │ ├── cache.py # Cache SQLite com TTL │ └── exceptions.py # Exceções customizadas ├── api/ │ ├── app.py # FastAPI app factory │ └── routes/ # Endpoints REST └── schemas/ └── series.py # Pydantic models Enter fullscreen mode Exit fullscreen mode COMMAND_BLOCK: bacendata/ ├── wrapper/ │ ├── bacen_sgs.py # Core: get(), aget(), metadata() │ ├── catalogo.py # 14 séries pré-configuradas │ ├── cache.py # Cache SQLite com TTL │ └── exceptions.py # Exceções customizadas ├── api/ │ ├── app.py # FastAPI app factory │ └── routes/ # Endpoints REST └── schemas/ └── series.py # Pydantic models COMMAND_BLOCK: bacendata/ ├── wrapper/ │ ├── bacen_sgs.py # Core: get(), aget(), metadata() │ ├── catalogo.py # 14 séries pré-configuradas │ ├── cache.py # Cache SQLite com TTL │ └── exceptions.py # Exceções customizadas ├── api/ │ ├── app.py # FastAPI app factory │ └── routes/ # Endpoints REST └── schemas/ └── series.py # Pydantic models COMMAND_BLOCK: async def _buscar_com_paginacao(codigo, inicio, fim, client): intervalos = _gerar_intervalos(inicio, fim, max_anos=10) semaforo = asyncio.Semaphore(5) async def fetch_com_semaforo(ini, fi): async with semaforo: try: return await _buscar_serie_periodo(codigo, ini, fi, client) except (SerieNaoEncontrada, BacenAPIError): return [] # Intervalo sem dados, não aborta tasks = [fetch_com_semaforo(ini, fi) for ini, fi in intervalos] resultados = await asyncio.gather(*tasks) todos_dados = [item for sublista in resultados for item in sublista] if not todos_dados: raise SerieNaoEncontrada(codigo) return todos_dados Enter fullscreen mode Exit fullscreen mode COMMAND_BLOCK: async def _buscar_com_paginacao(codigo, inicio, fim, client): intervalos = _gerar_intervalos(inicio, fim, max_anos=10) semaforo = asyncio.Semaphore(5) async def fetch_com_semaforo(ini, fi): async with semaforo: try: return await _buscar_serie_periodo(codigo, ini, fi, client) except (SerieNaoEncontrada, BacenAPIError): return [] # Intervalo sem dados, não aborta tasks = [fetch_com_semaforo(ini, fi) for ini, fi in intervalos] resultados = await asyncio.gather(*tasks) todos_dados = [item for sublista in resultados for item in sublista] if not todos_dados: raise SerieNaoEncontrada(codigo) return todos_dados COMMAND_BLOCK: async def _buscar_com_paginacao(codigo, inicio, fim, client): intervalos = _gerar_intervalos(inicio, fim, max_anos=10) semaforo = asyncio.Semaphore(5) async def fetch_com_semaforo(ini, fi): async with semaforo: try: return await _buscar_serie_periodo(codigo, ini, fi, client) except (SerieNaoEncontrada, BacenAPIError): return [] # Intervalo sem dados, não aborta tasks = [fetch_com_semaforo(ini, fi) for ini, fi in intervalos] resultados = await asyncio.gather(*tasks) todos_dados = [item for sublista in resultados for item in sublista] if not todos_dados: raise SerieNaoEncontrada(codigo) return todos_dados COMMAND_BLOCK: # Instalar pip install bacendata[api] # Rodar uvicorn bacendata.api.app:create_app --factory --port 8000 Enter fullscreen mode Exit fullscreen mode COMMAND_BLOCK: # Instalar pip install bacendata[api] # Rodar uvicorn bacendata.api.app:create_app --factory --port 8000 COMMAND_BLOCK: # Instalar pip install bacendata[api] # Rodar uvicorn bacendata.api.app:create_app --factory --port 8000 CODE_BLOCK: GET /api/v1/series/{codigo}?start=2020-01-01&end=2025-01-01 GET /api/v1/series/{codigo}/metadata POST /api/v1/series/bulk (múltiplas séries) GET /api/v1/catalogo GET /api/v1/catalogo/search?q=selic GET /health Enter fullscreen mode Exit fullscreen mode CODE_BLOCK: GET /api/v1/series/{codigo}?start=2020-01-01&end=2025-01-01 GET /api/v1/series/{codigo}/metadata POST /api/v1/series/bulk (múltiplas séries) GET /api/v1/catalogo GET /api/v1/catalogo/search?q=selic GET /health CODE_BLOCK: GET /api/v1/series/{codigo}?start=2020-01-01&end=2025-01-01 GET /api/v1/series/{codigo}/metadata POST /api/v1/series/bulk (múltiplas séries) GET /api/v1/catalogo GET /api/v1/catalogo/search?q=selic GET /health COMMAND_BLOCK: curl "http://localhost:8000/api/v1/series/1?start=2025-01-01&last=5" Enter fullscreen mode Exit fullscreen mode COMMAND_BLOCK: curl "http://localhost:8000/api/v1/series/1?start=2025-01-01&last=5" COMMAND_BLOCK: curl "http://localhost:8000/api/v1/series/1?start=2025-01-01&last=5" CODE_BLOCK: { "codigo": 1, "nome": "Taxa de câmbio - Livre - Dólar americano (venda) - diário", "dados": [ {"data": "2025-01-02", "valor": 6.1797}, {"data": "2025-01-03", "valor": 6.1627}, ... ], "total_registros": 5 } Enter fullscreen mode Exit fullscreen mode CODE_BLOCK: { "codigo": 1, "nome": "Taxa de câmbio - Livre - Dólar americano (venda) - diário", "dados": [ {"data": "2025-01-02", "valor": 6.1797}, {"data": "2025-01-03", "valor": 6.1627}, ... ], "total_registros": 5 } CODE_BLOCK: { "codigo": 1, "nome": "Taxa de câmbio - Livre - Dólar americano (venda) - diário", "dados": [ {"data": "2025-01-02", "valor": 6.1797}, {"data": "2025-01-03", "valor": 6.1627}, ... ], "total_registros": 5 } - Calcula os intervalos: divide o período solicitado em chunks de no máximo 10 anos - Faz requisições paralelas: usa asyncio.gather com um Semaphore(5) para limitar a 5 requisições simultâneas (cortesia com o servidor do BACEN) - Combina os resultados: concatena todos os DataFrames, remove duplicatas de datas sobrepostas e ordena cronologicamente - Trata erros individuais: se um intervalo retornar 404 (dados não disponíveis naquele período), ele é ignorado sem abortar a consulta inteira - Séries diárias: cache de 6 horas - Séries mensais: cache de 24 horas - httpx (async): HTTP client — escolhido por suporte nativo a async, ao contrário do requests - asyncio: paralelismo de I/O para paginação - pandas: retorno em DataFrame para integração fácil com o ecossistema Python de dados - SQLite: cache local leve, sem dependências extras - FastAPI: API REST com docs automáticas (Swagger/ReDoc) - Streamlit + Plotly: dashboard interativo para não-programadores - Gráficos interativos com zoom e hover - Comparação lado a lado de até 3 séries - Média móvel configurável - Download CSV/Excel - Todos os 14 indicadores disponíveis - Adicionar novas séries ao catálogo - Integrar com a API de expectativas Focus (séries futuras) - Criar wrappers para R ou JavaScript - Melhorar a documentação - GitHub: github.com/fmaignacio/bacendata - PyPI: pypi.org/project/bacendata - Dashboard: bacendata.streamlit.app - Site: bacendata.com