Bitcoin CLI com Python ■
Sistemas RPC & ZMQ
Guia técnico completo para integrar Bitcoin Core com Python via chamadas RPC e notificações em tempo
real com ZeroMQ
Tecnologias bitcoin-cli • python-bitcoinrpc • pyzmq
Nível Intermediário / Avançado
Aplicação Wallets, Exchanges, Monitoramento de Blocos
Bitcoin CLI com Python — RPC & ZMQ | SkillTech Página 1
1. Introdução
O Bitcoin Core expõe duas interfaces principais para comunicação com sistemas externos: JSON-RPC
(chamadas síncronas para consulta e controle do nó) e ZeroMQ (ZMQ) (notificações assíncronas de
novos blocos e transações). Juntas, essas interfaces permitem construir sistemas completos de
monitoramento, automação e integração financeira.
Arquitetura Geral
Camada Tecnologia Descrição
Nó Bitcoin Bitcoin Core Daemon bitcoind rodando em mainnet/testnet
RPC JSON-RPC / HTTP Consultas síncronas — saldo, bloco, tx
ZMQ ZeroMQ (PUB/SUB) Notificações assíncronas em tempo real
Cliente Python 3.x Scripts de integração e automação
Bitcoin CLI com Python — RPC & ZMQ | SkillTech Página 2
2. Configuração do Bitcoin Core
Para ativar RPC e ZMQ, edite o arquivo [Link] localizado em ~/.bitcoin/[Link]
(Linux/macOS) ou %APPDATA%\Bitcoin\[Link] (Windows).
[Link] — Configuração completa
# ■■■ Rede ■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■
# Use 'testnet=1' para ambiente de testes
mainnet=1
# ■■■ JSON-RPC ■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■
server=1
rpcuser=seu_usuario
rpcpassword=senha_forte_aqui
rpcallowip=[Link]
rpcport=8332 # mainnet | testnet: 18332
# ■■■ ZeroMQ (ZMQ) ■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■
zmqpubrawblock=tcp://[Link]:28332
zmqpubrawtx=tcp://[Link]:28332
zmqpubhashtx=tcp://[Link]:28332
zmqpubhashblock=tcp://[Link]:28332
# ■■■ Índices ■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■
txindex=1 # necessário para GetRawTransaction
■ Reinicie o bitcoind após editar o [Link]
$ bitcoind -daemon
$ bitcoin-cli getblockchaininfo # testar conexão
Bitcoin CLI com Python — RPC & ZMQ | SkillTech Página 3
3. Sistema RPC com Python
A biblioteca python-bitcoinrpc facilita chamadas ao nó Bitcoin via HTTP/JSON-RPC. Abaixo estão os
principais padrões de uso.
3.1 Instalação das dependências
pip install python-bitcoinrpc requests
3.2 Conexão e chamadas básicas
from [Link] import AuthServiceProxy, JSONRPCException
RPC_USER = "seu_usuario"
RPC_PASS = "senha_forte_aqui"
RPC_HOST = "[Link]"
RPC_PORT = 8332
def get_rpc():
url = f"[Link]
return AuthServiceProxy(url, timeout=120)
rpc = get_rpc()
# Informações do blockchain
info = [Link]()
print(f"Blocos sincronizados : {info['blocks']}")
print(f"Rede : {info['chain']}")
print(f"Progresso : {info['verificationprogress']:.2%}")
# Saldo total da wallet
balance = [Link]()
print(f"Saldo : {balance} BTC")
# Melhor bloco atual
best_hash = [Link]()
best_block = [Link](best_hash)
print(f"Último bloco : {best_block['height']}")
print(f"Transações no bloco : {len(best_block['tx'])}")
3.3 Operações de Transação
# Decodificar uma transação raw
raw_tx = [Link]("SEU_TXID_AQUI")
decoded = [Link](raw_tx)
print(f"TxID : {decoded['txid']}")
print(f"vsize : {decoded['vsize']} vbytes")
print(f"Inputs : {len(decoded['vin'])}")
print(f"Outputs: {len(decoded['vout'])}")
for out in decoded['vout']:
valor = out['value']
addrs = out['scriptPubKey'].get('address', 'N/A')
print(f" {valor:.8f} BTC -> {addrs}")
# Estimar fee para confirmação em N blocos
fee_rate = [Link](6) # 6 blocos (~60 min)
Bitcoin CLI com Python — RPC & ZMQ | SkillTech Página 4
print(f"Fee rate: {fee_rate['feerate']} BTC/kB")
3.4 Sistema de monitoramento de carteira
import time
from [Link] import AuthServiceProxy
def monitor_wallet(intervalo=30):
rpc = get_rpc()
print("[ Monitoramento de Carteira iniciado ]")
saldo_anterior = None
while True:
try:
saldo = [Link]()
txs = [Link]("*", 5) # últimas 5 tx
blocos = [Link]()
if saldo != saldo_anterior:
print(f"\n■ Saldo atualizado: {saldo} BTC (bloco #{blocos})")
saldo_anterior = saldo
for tx in txs:
if [Link]('confirmations', 0) == 0:
print(f"■ Tx pendente: {tx['txid'][:16]}... "
f"| {tx['amount']} BTC")
except Exception as e:
print(f"Erro RPC: {e}")
[Link](intervalo)
monitor_wallet()
Bitcoin CLI com Python — RPC & ZMQ | SkillTech Página 5
4. Sistema ZMQ com Python
O ZeroMQ permite receber notificações em tempo real do Bitcoin Core sem polling. O nó publica
mensagens nos tópicos configurados e o cliente Python faz subscribe para recebê-las.
4.1 Instalação
pip install pyzmq
4.2 Tópicos ZMQ disponíveis
Tópico Dado recebido Uso típico
rawblock Bloco completo em bytes Processar bloco inteiro
rawtx Transação em bytes Analisar tx ao vivo
hashblock Hash do bloco (32 bytes) Detectar novo bloco
hashtx Hash da tx (32 bytes) Detectar nova tx na mempool
4.3 Listener básico de novos blocos
import zmq
import struct
import hashlib
ZMQ_ADDR = "tcp://[Link]:28332"
ctx = [Link]()
socket = [Link]([Link])
[Link](ZMQ_ADDR)
# Subscribes
socket.setsockopt_string([Link], "hashblock")
socket.setsockopt_string([Link], "hashtx")
print("[ ZMQ Listener iniciado — aguardando eventos ]")
while True:
topic, body, seq = socket.recv_multipart()
topic = [Link]()
seq_num = [Link]("<I", seq)[0]
if topic == "hashblock":
block_hash = [Link]()
print(f"\n■ NOVO BLOCO | seq={seq_num}")
print(f" Hash: {block_hash}")
elif topic == "hashtx":
tx_hash = [Link]()
print(f" ■ Nova TX | {tx_hash[:32]}...")
4.4 Listener avançado — RPC + ZMQ integrados
import zmq
import struct
Bitcoin CLI com Python — RPC & ZMQ | SkillTech Página 6
from [Link] import AuthServiceProxy
ZMQ_ADDR = "tcp://[Link]:28332"
RPC_URL = "[Link]
def get_rpc():
return AuthServiceProxy(RPC_URL, timeout=60)
def process_block(block_hash: str):
rpc = get_rpc()
block = [Link](block_hash, 2) # verbosity=2 inclui txs
total_btc = sum(
out['value']
for tx in block['tx']
for out in tx['vout']
)
print(f" Altura : {block['height']}")
print(f" TXs : {len(block['tx'])}")
print(f" Volume : {total_btc:.4f} BTC")
print(f" Tamanho : {block['size']} bytes")
print(f" Dif : {block['difficulty']:.2f}")
ctx = [Link]()
socket = [Link]([Link])
[Link](ZMQ_ADDR)
socket.setsockopt_string([Link], "hashblock")
print("[ Sistema RPC+ZMQ ativo ]")
while True:
topic, body, seq = socket.recv_multipart()
if [Link]() == "hashblock":
hash_hex = [Link]()
print(f"\n■ Bloco detectado: {hash_hex[:24]}...")
try:
process_block(hash_hex)
except Exception as e:
print(f" Erro ao processar: {e}")
Bitcoin CLI com Python — RPC & ZMQ | SkillTech Página 7
5. Sistema Completo: Monitor + Alertas
Exemplo de sistema de produção que combina ZMQ para detecção de eventos e RPC para
enriquecimento de dados, com filas assíncronas para processamento paralelo.
import zmq
import struct
import threading
import queue
import time
from [Link] import AuthServiceProxy
ZMQ_ADDR = "tcp://[Link]:28332"
RPC_URL = "[Link]
TX_QUEUE = [Link](maxsize=1000)
# ■■■ Worker RPC (thread separada) ■■■■■■■■■■■■■■■■■■■■■■■■■
def rpc_worker():
rpc = AuthServiceProxy(RPC_URL, timeout=60)
while True:
tx_hash = TX_QUEUE.get()
try:
raw = [Link](tx_hash)
dec = [Link](raw)
outputs = [Link]('vout', [])
total = sum(o['value'] for o in outputs)
if total >= 1.0: # alerta para tx >= 1 BTC
print(f" ■ TX GRANDE: {tx_hash[:20]}... "
f"| {total:.4f} BTC")
except Exception:
pass
finally:
TX_QUEUE.task_done()
# Inicia 4 workers
for _ in range(4):
t = [Link](target=rpc_worker, daemon=True)
[Link]()
# ■■■ ZMQ Listener (thread principal) ■■■■■■■■■■■■■■■■■■■■■■
ctx = [Link]()
socket = [Link]([Link])
[Link](ZMQ_ADDR)
socket.setsockopt_string([Link], "hashblock")
socket.setsockopt_string([Link], "hashtx")
stats = {"blocos": 0, "txs": 0, "inicio": [Link]()}
print("[ Sistema completo ativo — Ctrl+C para parar ]")
try:
while True:
topic, body, seq = socket.recv_multipart()
topic = [Link]()
if topic == "hashblock":
stats["blocos"] += 1
uptime = ([Link]() - stats["inicio"]) / 60
Bitcoin CLI com Python — RPC & ZMQ | SkillTech Página 8
print(f"\n■ Bloco #{stats['blocos']} | "
f"TXs processadas: {stats['txs']} | "
f"Uptime: {uptime:.1f}min")
elif topic == "hashtx":
stats["txs"] += 1
if not TX_QUEUE.full():
TX_QUEUE.put([Link]())
except KeyboardInterrupt:
print("\nSistema encerrado.")
Bitcoin CLI com Python — RPC & ZMQ | SkillTech Página 9
6. Referência de Comandos RPC
Principais comandos do bitcoin-cli e equivalentes Python:
Comando CLI Equivalente Python RPC Descrição
getblockchaininfo [Link]() Info geral do blockchain
getblockcount [Link]() Altura atual
getbestblockhash [Link]() Hash do bloco mais recente
getblock <hash> [Link](hash, 2) Dados do bloco
getbalance [Link]() Saldo da wallet
getnewaddress [Link]() Gerar novo endereço
sendtoaddress [Link](addr, amt) Enviar BTC
getrawtransaction [Link](txid) TX em hex
decoderawtransaction [Link](hex) Decodificar TX
estimatesmartfee [Link](n) Estimar fee
listunspent [Link]() UTXOs disponíveis
getmempoolinfo [Link]() Info da mempool
getpeerinfo [Link]() Peers conectados
Bitcoin CLI com Python — RPC & ZMQ | SkillTech Página 10
7. Boas Práticas e Segurança
Segurança
• Nunca exponha o RPC na internet — use somente [Link]
• Use senhas longas e aleatórias no rpcpassword (mínimo 32 chars)
• Configure firewall para bloquear as portas 8332 e 28332 externamente
• Em produção, use autenticação via cookie (.cookie) ao invés de user/pass
• Rode o bitcoind com usuário sem privilégios (non-root)
Performance
• Use connection pooling — crie a conexão RPC uma vez e reutilize
• Para múltiplas chamadas seguidas, use batch RPC para reduzir latência
• Configure txindex=1 no [Link] para busca completa de transações
• Prefira ZMQ a polling RPC para detecção de novos eventos
• Use filas ([Link]) para processar TXs assincronamente
Batch RPC — exemplo
from [Link] import AuthServiceProxy
rpc = AuthServiceProxy("[Link]
# Batch: múltiplas chamadas em uma só requisição HTTP
batch = [
[Link],
[Link],
[Link],
]
# Chama tudo de uma vez
results = rpc.batch_([(fn.__name__,) for fn in batch])
blocos, saldo, mempool = results
print(f"Blocos : {blocos}")
print(f"Saldo : {saldo} BTC")
print(f"Mempool: {mempool['size']} transações pendentes")
Documento gerado pela SkillTech • Bitcoin Core RPC & ZMQ com Python
Bitcoin CLI com Python — RPC & ZMQ | SkillTech Página 11