Ir para o conteúdo

Cotações Históricas B3

Guia completo para usar a API HistoricalQuotesB3 e extrair cotações históricas da B3 (Brasil, Bolsa, Balcão) a partir de arquivos COTAHIST.


Visão Geral

A classe HistoricalQuotesB3 fornece uma interface poderosa para processar arquivos COTAHIST da B3, extraindo cotações históricas de diferentes classes de ativos e convertendo-as para o formato Parquet otimizado para análise.

Características

  • ✅ Extração de múltiplas classes de ativos
  • ✅ Processamento de alto desempenho (modo fast/slow)
  • ✅ Conversão automática para formato Parquet
  • ✅ Suporte a dados desde 1986
  • ✅ Filtragem inteligente por tipo de ativo
  • ✅ Progress tracking detalhado

Classes de Ativos Disponíveis

A B3 disponibiliza cotações históricas para as seguintes classes de ativos:

Código Descrição Mercados Incluídos
ações Ações Mercado à vista (010) e fracionário (012)
etf ETFs Exchange Traded Funds
opções Opções Calls (070) e Puts (080)
termo Mercado a Termo Contratos a termo
exercicio_opcoes Exercício de Opções Exercício de opções
forward Mercado Forward Contratos forward
leilao Leilão Mercado de leilão

Dados Históricos

Cotações históricas da B3 estão disponíveis desde 1986 até o ano atual.


Uso Básico

Importação

from globaldatafinance import HistoricalQuotesB3

Criar Instância

b3 = HistoricalQuotesB3()

Extração Simples

# Extrair cotações de ações do ano atual
result = b3.extract(
    path_of_docs="/home/usuario/cotahist_zips",
    assets_list=["ações"],
    initial_year=2023
)

print(f"✓ Extraídos {result['total_records']:,} registros")

Métodos Principais

extract()

Extrai cotações históricas de arquivos COTAHIST ZIP para formato Parquet.

Assinatura

def extract(
    self,
    path_of_docs: str,
    assets_list: List[str],
    initial_year: Optional[int] = None,
    last_year: Optional[int] = None,
    destination_path: Optional[str] = None,
    output_filename: str = "cotahist_extracted",
    processing_mode: str = "fast",
) -> Dict[str, Any]

Parâmetros

Parâmetro Tipo Obrigatório Descrição
path_of_docs str ✅ Sim Diretório contendo arquivos COTAHIST ZIP
assets_list List[str] ✅ Sim Lista de classes de ativos a extrair
initial_year int ❌ Não Ano inicial (padrão: 1986)
last_year int ❌ Não Ano final (padrão: ano atual)
destination_path str ❌ Não Diretório de saída (padrão: mesmo que path_of_docs)
output_filename str ❌ Não Nome do arquivo de saída sem extensão
processing_mode str ❌ Não Modo de processamento: "fast" ou "slow"

Retorno

Dicionário com as seguintes chaves:

Chave Tipo Descrição
success bool True se extração foi bem-sucedida
message str Mensagem resumida do resultado
total_files int Total de arquivos ZIP processados
success_count int Arquivos processados com sucesso
error_count int Arquivos com erro
total_records int Total de registros extraídos
output_file str Caminho completo do arquivo Parquet gerado
errors List[str] Lista de erros (se houver)

Exemplos

Exemplo 1: Extração básica de ações

b3 = HistoricalQuotesB3()

result = b3.extract(
    path_of_docs="/data/cotahist",
    assets_list=["ações"],
    initial_year=2022,
    last_year=2023
)

if result['success']:
    print(f"✓ Arquivo gerado: {result['output_file']}")
    print(f"✓ Total de registros: {result['total_records']:,}")

Exemplo 2: Múltiplas classes de ativos

result = b3.extract(
    path_of_docs="/data/cotahist",
    assets_list=["ações", "etf", "opções"],
    initial_year=2020,
    last_year=2023,
    output_filename="multi_ativos_2020_2023"
)

Exemplo 3: Modo de baixa performance (economia de recursos)

result = b3.extract(
    path_of_docs="/data/cotahist",
    assets_list=["ações"],
    initial_year=2023,
    processing_mode="slow"  # Usa menos CPU/RAM
)

Exemplo 4: Destino personalizado

result = b3.extract(
    path_of_docs="/data/cotahist_zips",
    destination_path="/data/cotacoes_extraidas",
    assets_list=["ações", "etf"],
    initial_year=2023,
    output_filename="acoes_etf_2023"
)
# Arquivo salvo em: /data/cotacoes_extraidas/acoes_etf_2023.parquet

get_available_assets()

Retorna lista de todas as classes de ativos disponíveis.

Assinatura

def get_available_assets(self) -> List[str]

Retorno

Lista de strings com códigos das classes de ativos.

Exemplo

b3 = HistoricalQuotesB3()
assets = b3.get_available_assets()

print("Classes de ativos disponíveis:")
for asset in assets:
    print(f"  • {asset}")

Saída:

Classes de ativos disponíveis:
  • ações
  • etf
  • opções
  • termo
  • exercicio_opcoes
  • forward
  • leilao

get_available_years()

Retorna informações sobre o intervalo de anos disponível.

Assinatura

def get_available_years(self) -> Dict[str, int]

Retorno

Dicionário com:

Chave Descrição
"minimal_year" Ano mínimo disponível (1986)
"current_year" Ano atual

Exemplo

b3 = HistoricalQuotesB3()
years = b3.get_available_years()

print(f"Dados disponíveis de {years['minimal_year']} até {years['current_year']}")

Saída:

Dados disponíveis de 1986 até ano atual

Modos de Processamento

A extração suporta dois modos de processamento:

Modo Fast (Padrão) ⚡

  • Performance: Alto desempenho
  • CPU: Uso intensivo (multi-core)
  • RAM: Maior consumo de memória
  • Recomendado para: Máquinas com bons recursos, processamento de grandes volumes
result = b3.extract(
    path_of_docs="/data/cotahist",
    assets_list=["ações"],
    processing_mode="fast"  # Padrão
)

Modo Slow 🐢

  • Performance: Moderada
  • CPU: Uso reduzido (single-core ou poucos cores)
  • RAM: Menor consumo de memória
  • Recomendado para: Máquinas com recursos limitados, processamento em background
result = b3.extract(
    path_of_docs="/data/cotahist",
    assets_list=["ações"],
    processing_mode="slow"
)

Comparação de Performance

Modo Tempo (100k registros) CPU RAM Recomendado
fast ~5s Alto ~2GB ✅ Sim (padrão)
slow ~15s Baixo ~500MB Recursos limitados

Exemplos Avançados

Extração de Todos os Ativos

from globaldatafinance import HistoricalQuotesB3

b3 = HistoricalQuotesB3()

# Obter todas as classes de ativos
all_assets = b3.get_available_assets()

# Extrair tudo
result = b3.extract(
    path_of_docs="/data/cotahist",
    assets_list=all_assets,  # Todas as classes
    initial_year=2023,
    output_filename="todos_ativos_2023"
)

print(f"✓ Extraídos {result['total_records']:,} registros de {len(all_assets)} classes")

Extração Incremental por Ano

from globaldatafinance import HistoricalQuotesB3
import os

b3 = HistoricalQuotesB3()
base_path = "/data/cotahist"
output_path = "/data/cotacoes_extraidas"

# Extrair cada ano separadamente
for year in range(2020, 2024):
    output_file = f"acoes_{year}"

    result = b3.extract(
        path_of_docs=base_path,
        destination_path=output_path,
        assets_list=["ações"],
        initial_year=year,
        last_year=year,
        output_filename=output_file
    )

    if result['success']:
        print(f"✓ {year}: {result['total_records']:,} registros")
    else:
        print(f"✗ {year}: Erro na extração")

Validação Antes da Extração

from globaldatafinance import HistoricalQuotesB3
import os

b3 = HistoricalQuotesB3()
path_docs = "/data/cotahist"

# 1. Verificar se diretório existe
if not os.path.exists(path_docs):
    print(f"✗ Diretório não encontrado: {path_docs}")
    exit(1)

# 2. Verificar se há arquivos COTAHIST
zip_files = [f for f in os.listdir(path_docs) if f.startswith("COTAHIST") and f.endswith(".ZIP")]

if not zip_files:
    print(f"✗ Nenhum arquivo COTAHIST encontrado em {path_docs}")
    exit(1)

print(f"✓ Encontrados {len(zip_files)} arquivos COTAHIST")

# 3. Validar classes de ativos
requested_assets = ["ações", "etf"]
available_assets = b3.get_available_assets()

invalid_assets = [a for a in requested_assets if a not in available_assets]
if invalid_assets:
    print(f"✗ Ativos inválidos: {invalid_assets}")
    print(f"Ativos disponíveis: {available_assets}")
    exit(1)

# 4. Prosseguir com extração
result = b3.extract(
    path_of_docs=path_docs,
    assets_list=requested_assets,
    initial_year=2023
)

Tratamento de Erros

Exceções Comuns

Exceção Quando ocorre Como tratar
EmptyAssetListError assets_list está vazio Fornecer pelo menos um ativo
InvalidAssetsName Ativo inválido em assets_list Verificar com get_available_assets()
InvalidFirstYear initial_year fora do intervalo Usar 1986 ≤ ano ≤ ano atual
InvalidLastYear last_year inválido Usar initial_year ≤ ano ≤ ano atual
EmptyDirectoryError Diretório sem arquivos COTAHIST Verificar caminho e arquivos
ExtractionError Erro ao processar ZIP Verificar integridade dos arquivos

Formato dos Arquivos COTAHIST

Nomenclatura

Os arquivos COTAHIST seguem o padrão:

COTAHIST_AXXXX.ZIP

Onde XXXX é o ano (ex: COTAHIST_A2023.ZIP).

Onde Obter

Os arquivos COTAHIST podem ser baixados do site oficial da B3:

🔗 https://www.b3.com.br/pt_br/market-data-e-indices/servicos-de-dados/market-data/historico/mercado-a-vista/cotacoes-historicas/

Estrutura Interna

Cada arquivo ZIP contém um arquivo TXT com layout de largura fixa:

COTAHIST_A2023.ZIP
└── COTAHIST_A2023.TXT  (arquivo de texto com largura fixa)

O Global-Data-Finance processa automaticamente este formato e converte para Parquet.


Estrutura do Arquivo Parquet Gerado

Colunas

O arquivo Parquet gerado contém as seguintes colunas:

Coluna Tipo Descrição
data_pregao date Data do pregão
codigo_bdi string Código BDI
ticker string Código de negociação (ex: PETR4)
tipo_mercado string Tipo de mercado
nome_resumido string Nome resumido da empresa
especificacao_papel string Especificação do papel (ex: ON, PN)
preco_abertura decimal Preço de abertura
preco_maximo decimal Preço máximo do dia
preco_minimo decimal Preço mínimo do dia
preco_medio decimal Preço médio do dia
preco_fechamento decimal Preço de fechamento
melhor_oferta_compra decimal Melhor oferta de compra
melhor_oferta_venda decimal Melhor oferta de venda
numero_negocios int Número de negócios efetuados
quantidade_total int Quantidade total de títulos negociados
volume_total decimal Volume total financeiro
data_vencimento date Data de vencimento (opções/termo)
fator_cotacao int Fator de cotação
codigo_isin string Código ISIN
numero_distribuicao int Número de distribuição

Leitura com Pandas

import pandas as pd

df = pd.read_parquet("/data/cotacoes_extraidas/cotahist_extracted.parquet")

print(df.head())
print(f"\nShape: {df.shape}")
print(f"Período: {df['data_pregao'].min()} a {df['data_pregao'].max()}")

Leitura com Polars (Mais Rápido)

import polars as pl

df = pl.read_parquet("/data/cotacoes_extraidas/cotahist_extracted.parquet")

print(df.head())
print(f"\nShape: {df.shape}")
print(f"Memória: {df.estimated_size('mb'):.2f} MB")

Boas Práticas

1. Use Modo Fast para Grandes Volumes

# ✅ Recomendado para grandes volumes
result = b3.extract(
    path_of_docs="/data/cotahist",
    assets_list=["ações"],
    initial_year=1986,  # 23+ anos
    processing_mode="fast"
)

2. Separe Extrações por Classe de Ativo

# ✅ Melhor: arquivos separados por classe
for asset in ["ações", "etf", "opções"]:
    result = b3.extract(
        path_of_docs="/data/cotahist",
        assets_list=[asset],
        initial_year=2023,
        output_filename=f"{asset}_2023"
    )

# ❌ Evite: tudo em um arquivo (pode ficar muito grande)
result = b3.extract(
    path_of_docs="/data/cotahist",
    assets_list=["ações", "etf", "opções", "termo", "forward"],
    initial_year=1986,  # 38+ anos!
    output_filename="tudo"
)

3. Verifique Espaço em Disco

import shutil

stats = shutil.disk_usage("/data")
free_gb = stats.free / (1024**3)

if free_gb < 5:
    print(f"⚠️  Pouco espaço: {free_gb:.2f} GB")
    # Use modo slow ou processe menos anos
else:
    # Prosseguir normalmente
    pass

Próximos Passos


Dica de Análise

Após extrair para Parquet, use Polars para análises de alto desempenho. É significativamente mais rápido que Pandas para grandes volumes de dados.