Sistema Snapshot Blockchain Bitcoin com Python

---

@CanalQb no YouTube

---

Como Usar a Testnet Chainphon e se Preparar para o Lançamento Oficial

---

Sempre crie uma frase de segurança única para jogos, testnets ou airdrops e evite usar sua carteira principal.

---

Como Criar um Sistema Snapshot Blockchain Bitcoin com Python

Neste tutorial prático, você aprenderá como construir um sistema completo em Python para capturar um snapshot da blockchain do Bitcoin contendo os saldos de todos os endereços. O processo envolve o download de um arquivo .tsv.gz com milhões de registros, sua extração, e a importação dos dados para um banco de dados SQLite local.

Este tipo de ferramenta é útil para desenvolvedores que desejam realizar análises estatísticas, auditorias de endereços ou integrar sistemas com informações históricas da blockchain.

Pré-requisitos

• Python 3.7 ou superior
• Bibliotecas: requests, tqdm, pandas, sqlite3, gzip, gc, datetime
• Espaço em disco (o arquivo .tsv descompactado pode ter vários GB)

1. Download do Arquivo de Endereços Bitcoin

O script inicia baixando um arquivo GZ contendo dados atualizados de endereços e saldos. O arquivo é hospedado no projeto Loyce.club.

Exemplo de URL utilizada:

http://addresses.loyce.club/blockchair_bitcoin_addresses_and_balance_LATEST.tsv.gz

2. Extração do Arquivo GZ

O arquivo GZ é descompactado utilizando a biblioteca gzip. A extração é feita com controle de progresso utilizando a biblioteca tqdm, oferecendo uma experiência visual mais amigável ao usuário.

3. Importação para SQLite com Pandas

Após a extração, o script importa os dados para um banco de dados SQLite local, chamado relatorio_btc.db. Utilizamos leitura em chunks com pandas.read_csv() para evitar estouro de memória.

O banco de dados criado contém uma única tabela:

CREATE TABLE IF NOT EXISTS relatorio_btc (
    address TEXT PRIMARY KEY,
    balance REAL
)

4. Progresso em Tempo Real

Durante a execução, o sistema mostra o progresso da importação em tempo real e realiza o commit em blocos. O sistema também trata interrupções com KeyboardInterrupt e erros durante a transação com rollback.

5. Código Completo do Script

import requests
from tqdm import tqdm
from datetime import datetime
import gzip
import os
import sqlite3
import pandas as pd
import gc
import ctypes
from time import time

try:
    ctypes.windll.kernel32.SetConsoleTitleW("Sistema SNAPSHOT Blockchain Bitcoin - V 1.0")
except:
    pass

def baixar_arquivo(url, destino):
    response = requests.get(url, stream=True)
    total_size = int(response.headers.get('content-length', 0))
    with open(destino, 'wb') as file, tqdm(
        desc=destino, total=total_size, unit='B', unit_scale=True
    ) as bar:
        for data in response.iter_content(chunk_size=1024):
            file.write(data)
            bar.update(len(data))

def extrair_arquivo(gz_path, tsv_path):
    with gzip.open(gz_path, 'rb') as f_in:
        total_size = os.path.getsize(gz_path)
        with open(tsv_path, 'wb') as f_out, tqdm(
            total=total_size, desc="Extraindo", unit='B', unit_scale=True
        ) as bar:
            buffer_size = 1024 * 1024
            while True:
                buf = f_in.read(buffer_size)
                if not buf:
                    break
                f_out.write(buf)
                bar.update(len(buf))

def importar_para_sqlite(tsv_path, db_path):
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS relatorio_btc (
            address TEXT PRIMARY KEY,
            balance REAL
        )
    ''')
    cursor.execute('''
        CREATE INDEX IF NOT EXISTS idx_address ON relatorio_btc(address)
    ''')

    chunk_size = 100_000
    df_iterator = pd.read_csv(tsv_path, sep='\t', chunksize=chunk_size)

    with open(tsv_path) as f:
        total_lines = sum(1 for _ in f)

    processed_lines = 0
    print(f"\nIniciando a importação dos dados...\nTotal de linhas: {total_lines}")

    try:
        conn.isolation_level = None
        cursor.execute('BEGIN TRANSACTION;')

        for df in df_iterator:
            if 'address' not in df.columns or 'balance' not in df.columns:
                print("\n❌ O arquivo não contém as colunas 'address' e 'balance'.")
                conn.close()
                return

            cursor.executemany('''
                INSERT OR IGNORE INTO relatorio_btc (address, balance)
                VALUES (?, ?)
            ''', [(row['address'], row['balance']) for _, row in df.iterrows()])

            conn.commit()
            processed_lines += len(df)
            progresso = (processed_lines / total_lines) * 100
            print(f"\r{progresso:.2f}% já importado.", end='', flush=True)

            del df
            gc.collect()

        print("\n✅ Importação concluída com sucesso.")

    except KeyboardInterrupt:
        print("\n⏹ Interrompido manualmente. Fazendo rollback...")
        conn.rollback()
    except Exception as e:
        print(f"\n❌ Erro durante a importação: {e}")
        conn.rollback()
    finally:
        conn.close()
        print("🔒 Conexão com o banco encerrada.")

if __name__ == "__main__":
    try:
        start_time = time()

        url = "http://addresses.loyce.club/blockchair_bitcoin_addresses_and_balance_LATEST.tsv.gz"
        datadehoje = datetime.today().strftime('%d%m%Y')
        filename_gz = f"blockchair_bitcoin_addresses_and_balance_{datadehoje}.tsv.gz"
        filename_tsv = f"blockchair_bitcoin_addresses_and_balance_{datadehoje}.tsv"
        db_path = "relatorio_btc.db"

        print("🔽 Iniciando download do arquivo...")
        baixar_arquivo(url, filename_gz)

        print("📂 Extraindo arquivo .tsv...")
        extrair_arquivo(filename_gz, filename_tsv)

        if os.path.exists(filename_gz):
            os.remove(filename_gz)
            print(f"🗑️ Arquivo {filename_gz} excluído.")

        importar_para_sqlite(filename_tsv, db_path)

        print(f"\n⏱️ Tempo total de execução: {round(time() - start_time, 2)} segundos.")

    except KeyboardInterrupt:
        print("\n⏹ Execução interrompida pelo usuário.")

Conclusão

Este sistema é ideal para entusiastas e desenvolvedores de blockchain que desejam explorar os dados de forma offline. Ele permite criar consultas SQL para encontrar endereços específicos, analisar grandes saldos e muito mais.

Importante: Este conteúdo tem caráter educacional. Antes de investir ou utilizar dados obtidos da blockchain, é essencial fazer uma análise crítica individual e compreender os riscos envolvidos.

Links Oficiais e Créditos

• Loyce.club - Arquivos Bitcoin
• Pandas - Site Oficial
• SQLite - Site Oficial