Como Usar a API da Binance com Python

Introdução ao Binance API

O Binance API fornece aos desenvolvedores acesso programático à plataforma de negociação da Binance, permitindo que eles criem bots de negociação, aplicativos e ferramentas que interajam com a exchange. Este tutorial irá guiá-lo pelo processo de uso do Binance API, desde a configuração da autenticação até a realização de vários tipos de solicitações.

Embora muitos desenvolvedores tradicionalmente tenham confiado no Postman para testes de API, o Apidog oferece uma experiência mais simplificada e especializada para trabalhar com APIs de criptomoedas como a Binance. Com sua interface intuitiva, manuseio aprimorado de autenticação e recursos específicos de criptomoedas, Apidog reduz significativamente o tempo de configuração necessário para começar a testar e implementar os endpoints do Binance API.

As ferramentas de colaboração em tempo real da plataforma e as capacidades abrangentes de geração de documentação a tornam particularmente valiosa para equipes que trabalham em aplicativos de negociação de criptomoedas.

Se você está sério sobre a integração eficaz do Binance API, mudar para o Apidog melhorará notavelmente seu fluxo de trabalho de desenvolvimento e reduzirá o tempo de implementação para seus projetos de criptomoeda.

Configurando a Autenticação do Binance API

Antes de usar o Binance API, você precisa criar chaves de API a partir da sua conta da Binance. Esta seção abrange como configurar e proteger suas chaves de API.

Criando Chaves de API da Binance

1. Faça login na sua conta Binance
2. Navegue até "Gerenciamento de API" nas configurações da sua conta
3. Crie uma nova chave de API
4. Anote sua chave de API e chave secreta
5. Configure as restrições (whitelisting de IP, permissões de negociação, etc.)

Tipos de Chaves de API Suportadas pelo Binance API

A Binance suporta três tipos de métodos de autenticação de chaves de API:

1. Chaves HMAC - O método de autenticação mais comum
2. Chaves RSA - Oferece recursos de segurança aprimorados
3. Chaves Ed25519 - Proporciona o melhor desempenho e segurança

Exemplo de Autenticação HMAC

Veja como autenticar com chaves HMAC:

import hmac
import hashlib
import time
import requests
from urllib.parse import urlencode

# Credenciais da API
api_key = 'sua_chave_api'
api_secret = 'sua_chave_secreta'

# URL base para o Binance API
base_url = 'https://api.binance.com'

# Função para criar uma assinatura para pontos de extremidade autenticados
def get_signature(query_string):
    return hmac.new(
        api_secret.encode('utf-8'),
        query_string.encode('utf-8'),
        hashlib.sha256
    ).hexdigest()

# Solicitação de exemplo para obter informações da conta
def get_account():
    endpoint = '/api/v3/account'
    timestamp = int(time.time() * 1000)

    params = {
        'timestamp': timestamp,
        'recvWindow': 5000  # Parâmetro opcional
    }

    query_string = urlencode(params)
    signature = get_signature(query_string)

    url = f"{base_url}{endpoint}?{query_string}&signature={signature}"

    headers = {
        'X-MBX-APIKEY': api_key
    }

    response = requests.get(url, headers=headers)
    return response.json()

# Chama a função
account_info = get_account()
print(account_info)

Compreendendo os Endpoints do Binance API

O Binance API é organizado em várias categorias de endpoints com base em sua funcionalidade.

Endpoints Gerais do Binance API

Esses endpoints fornecem informações gerais sobre a exchange:

# Testar conectividade
def test_connectivity():
    endpoint = '/api/v3/ping'
    url = f"{base_url}{endpoint}"
    response = requests.get(url)
    return response.json()

# Obter hora do servidor
def get_server_time():
    endpoint = '/api/v3/time'
    url = f"{base_url}{endpoint}"
    response = requests.get(url)
    return response.json()

# Obter informações da exchange
def get_exchange_info():
    endpoint = '/api/v3/exchangeInfo'
    url = f"{base_url}{endpoint}"
    response = requests.get(url)
    return response.json()

Endpoints de Dados de Mercado do Binance API

Esses endpoints fornecem acesso a dados de mercado:

# Obter livro de ordens para um símbolo
def get_order_book(symbol, limit=100):
    endpoint = '/api/v3/depth'
    params = {
        'symbol': symbol,
        'limit': limit
    }
    url = f"{base_url}{endpoint}?{urlencode(params)}"
    response = requests.get(url)
    return response.json()

# Obter negociações recentes
def get_recent_trades(symbol, limit=500):
    endpoint = '/api/v3/trades'
    params = {
        'symbol': symbol,
        'limit': limit
    }
    url = f"{base_url}{endpoint}?{urlencode(params)}"
    response = requests.get(url)
    return response.json()

# Obter dados de candles
def get_klines(symbol, interval, start_time=None, end_time=None, limit=500):
    endpoint = '/api/v3/klines'
    params = {
        'symbol': symbol,
        'interval': interval,
        'limit': limit
    }

    if start_time:
        params['startTime'] = start_time
    if end_time:
        params['endTime'] = end_time

    url = f"{base_url}{endpoint}?{urlencode(params)}"
    response = requests.get(url)

    # Formatar a resposta
    klines = response.json()
    formatted_klines = []
    for k in klines:
        formatted_klines.append({
            'open_time': k[0],
            'open': float(k[1]),
            'high': float(k[2]),
            'low': float(k[3]),
            'close': float(k[4]),
            'volume': float(k[5]),
            'close_time': k[6],
            'quote_volume': float(k[7]),
            'trades': k[8],
            'taker_buy_base_volume': float(k[9]),
            'taker_buy_quote_volume': float(k[10])
        })

    return formatted_klines

Negociação com Binance API

Os endpoints de negociação permitem que você faça e gerencie ordens.

Fazendo Ordens com Binance API

Veja como fazer diferentes tipos de ordens:

# Fazer uma ordem LIMIT
def place_limit_order(symbol, side, quantity, price):
    endpoint = '/api/v3/order'
    timestamp = int(time.time() * 1000)

    params = {
        'symbol': symbol,
        'side': side,  # 'BUY' ou 'SELL'
        'type': 'LIMIT',
        'timeInForce': 'GTC',  # Boa até ser cancelada
        'quantity': quantity,
        'price': price,
        'timestamp': timestamp,
        'recvWindow': 5000
    }

    query_string = urlencode(params)
    signature = get_signature(query_string)

    url = f"{base_url}{endpoint}?{query_string}&signature={signature}"

    headers = {
        'X-MBX-APIKEY': api_key
    }

    response = requests.post(url, headers=headers)
    return response.json()

# Fazer uma ordem MARKET
def place_market_order(symbol, side, quantity):
    endpoint = '/api/v3/order'
    timestamp = int(time.time() * 1000)

    params = {
        'symbol': symbol,
        'side': side,  # 'BUY' ou 'SELL'
        'type': 'MARKET',
        'quantity': quantity,
        'timestamp': timestamp,
        'recvWindow': 5000
    }

    query_string = urlencode(params)
    signature = get_signature(query_string)

    url = f"{base_url}{endpoint}?{query_string}&signature={signature}"

    headers = {
        'X-MBX-APIKEY': api_key
    }

    response = requests.post(url, headers=headers)
    return response.json()

# Fazer uma ordem STOP_LOSS
def place_stop_loss_order(symbol, side, quantity, stop_price):
    endpoint = '/api/v3/order'
    timestamp = int(time.time() * 1000)

    params = {
        'symbol': symbol,
        'side': side,
        'type': 'STOP_LOSS',
        'quantity': quantity,
        'stopPrice': stop_price,
        'timestamp': timestamp,
        'recvWindow': 5000
    }

    query_string = urlencode(params)
    signature = get_signature(query_string)

    url = f"{base_url}{endpoint}?{query_string}&signature={signature}"

    headers = {
        'X-MBX-APIKEY': api_key
    }

    response = requests.post(url, headers=headers)
    return response.json()

Gerenciando Ordens com Binance API

Veja como consultar, cancelar e rastrear ordens:

# Consultar status da ordem
def query_order(symbol, order_id=None, orig_client_order_id=None):
    endpoint = '/api/v3/order'
    timestamp = int(time.time() * 1000)

    params = {
        'symbol': symbol,
        'timestamp': timestamp,
        'recvWindow': 5000
    }

    if order_id:
        params['orderId'] = order_id
    elif orig_client_order_id:
        params['origClientOrderId'] = orig_client_order_id
    else:
        raise ValueError("order_id ou orig_client_order_id devem ser fornecidos")

    query_string = urlencode(params)
    signature = get_signature(query_string)

    url = f"{base_url}{endpoint}?{query_string}&signature={signature}"

    headers = {
        'X-MBX-APIKEY': api_key
    }

    response = requests.get(url, headers=headers)
    return response.json()

# Cancelar uma ordem
def cancel_order(symbol, order_id=None, orig_client_order_id=None):
    endpoint = '/api/v3/order'
    timestamp = int(time.time() * 1000)

    params = {
        'symbol': symbol,
        'timestamp': timestamp,
        'recvWindow': 5000
    }

    if order_id:
        params['orderId'] = order_id
    elif orig_client_order_id:
        params['origClientOrderId'] = orig_client_order_id
    else:
        raise ValueError("order_id ou orig_client_order_id devem ser fornecidos")

    query_string = urlencode(params)
    signature = get_signature(query_string)

    url = f"{base_url}{endpoint}?{query_string}&signature={signature}"

    headers = {
        'X-MBX-APIKEY': api_key
    }

    response = requests.delete(url, headers=headers)
    return response.json()

# Obter todas as ordens abertas
def get_open_orders(symbol=None):
    endpoint = '/api/v3/openOrders'
    timestamp = int(time.time() * 1000)

    params = {
        'timestamp': timestamp,
        'recvWindow': 5000
    }

    if symbol:
        params['symbol'] = symbol

    query_string = urlencode(params)
    signature = get_signature(query_string)

    url = f"{base_url}{endpoint}?{query_string}&signature={signature}"

    headers = {
        'X-MBX-APIKEY': api_key
    }

    response = requests.get(url, headers=headers)
    return response.json()

Dados da Conta e do Usuário com Binance API

Esses endpoints permitem que você acesse informações da conta e transmita dados do usuário.

Informações da Conta com Binance API

# Obter informações da conta
def get_account_information():
    endpoint = '/api/v3/account'
    timestamp = int(time.time() * 1000)

    params = {
        'timestamp': timestamp,
        'recvWindow': 5000
    }

    query_string = urlencode(params)
    signature = get_signature(query_string)

    url = f"{base_url}{endpoint}?{query_string}&signature={signature}"

    headers = {
        'X-MBX-APIKEY': api_key
    }

    response = requests.get(url, headers=headers)
    return response.json()

# Obter lista de negociações da conta
def get_account_trades(symbol, start_time=None, end_time=None, limit=500):
    endpoint = '/api/v3/myTrades'
    timestamp = int(time.time() * 1000)

    params = {
        'symbol': symbol,
        'timestamp': timestamp,
        'limit': limit,
        'recvWindow': 5000
    }

    if start_time:
        params['startTime'] = start_time
    if end_time:
        params['endTime'] = end_time

    query_string = urlencode(params)
    signature = get_signature(query_string)

    url = f"{base_url}{endpoint}?{query_string}&signature={signature}"

    headers = {
        'X-MBX-APIKEY': api_key
    }

    response = requests.get(url, headers=headers)
    return response.json()

Fluxos de Dados do Usuário com Binance API

# Iniciar um fluxo de dados do usuário
def start_user_data_stream():
    endpoint = '/api/v3/userDataStream'

    headers = {
        'X-MBX-APIKEY': api_key
    }

    response = requests.post(f"{base_url}{endpoint}", headers=headers)
    return response.json()

# Manter ativo um fluxo de dados do usuário
def keep_alive_user_data_stream(listen_key):
    endpoint = '/api/v3/userDataStream'

    params = {
        'listenKey': listen_key
    }

    headers = {
        'X-MBX-APIKEY': api_key
    }

    response = requests.put(f"{base_url}{endpoint}?{urlencode(params)}", headers=headers)
    return response.json()

# Fechar um fluxo de dados do usuário
def close_user_data_stream(listen_key):
    endpoint = '/api/v3/userDataStream'

    params = {
        'listenKey': listen_key
    }

    headers = {
        'X-MBX-APIKEY': api_key
    }

    response = requests.delete(f"{base_url}{endpoint}?{urlencode(params)}", headers=headers)
    return response.json()

Recursos Avançados do Binance API

Esta seção aborda alguns dos recursos mais avançados do Binance API.

Trabalhando com Listas de Ordens (OCO) no Binance API

Ordens One-Cancels-the-Other (OCO) permitem que você faça um par de ordens onde, se uma for acionada, a outra é automaticamente cancelada:

# Fazer uma ordem OCO
def place_oco_order(symbol, side, quantity, price, stop_price):
    endpoint = '/api/v3/orderList/oco'
    timestamp = int(time.time() * 1000)

    params = {
        'symbol': symbol,
        'side': side,
        'quantity': quantity,
        'price': price,
        'stopPrice': stop_price,
        'timestamp': timestamp,
        'recvWindow': 5000
    }

    # Definir os tipos da ordem OCO com base no lado
    if side == 'SELL':
        params['aboveType'] = 'STOP_LOSS'
        params['belowType'] = 'LIMIT_MAKER'
    else:
        params['aboveType'] = 'LIMIT_MAKER'
        params['belowType'] = 'STOP_LOSS'

    params['aboveTimeInForce'] = 'GTC'
    params['belowTimeInForce'] = 'GTC'

    query_string = urlencode(params)
    signature = get_signature(query_string)

    url = f"{base_url}{endpoint}?{query_string}&signature={signature}"

    headers = {
        'X-MBX-APIKEY': api_key
    }

    response = requests.post(url, headers=headers)
    return response.json()

Lidando com Limites de Taxas no Binance API

A Binance aplica limites de taxa às solicitações de API. Veja como lidar com eles:

import time
from functools import wraps

# Decorador para lidar com limites de taxa
def handle_rate_limit(max_retries=3, retry_delay=30):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            retries = 0
            while retries <= max_retries:
                try:
                    response = func(*args, **kwargs)
                    if 'code' in response and response['code'] == -1429:  # Limite de taxa excedido
                        retries += 1
                        if retries > max_retries:
                            raise Exception(f"Limite de taxa excedido após {max_retries} tentativas")
                        print(f"Limite de taxa excedido. Aguardando {retry_delay} segundos antes de tentar novamente...")
                        time.sleep(retry_delay)
                    else:
                        return response
                except Exception as e:
                    if "429" in str(e) or "418" in str(e):
                        retries += 1
                        if retries > max_retries:
                            raise
                        print(f"Limite de taxa excedido. Aguardando {retry_delay} segundos antes de tentar novamente...")
                        time.sleep(retry_delay)
                    else:
                        raise
        return wrapper
    return decorator

# Exemplo de uso
@handle_rate_limit()
def get_exchange_info_with_retry():
    endpoint = '/api/v3/exchangeInfo'
    url = f"{base_url}{endpoint}"
    response = requests.get(url)
    return response.json()

Construindo um Bot de Negociação Completo do Binance API

Agora, vamos juntar tudo para construir um bot de negociação simples:

import hmac
import hashlib
import time
import requests
import json
import websocket
import threading
from urllib.parse import urlencode

class BinanceTrader:
    def __init__(self, api_key, api_secret, symbol='BTCUSDT'):
        self.api_key = api_key
        self.api_secret = api_secret
        self.base_url = 'https://api.binance.com'
        self.symbol = symbol
        self.ws = None
        self.listen_key = None
        self.keep_alive_thread = None

    def get_signature(self, query_string):
        return hmac.new(
            self.api_secret.encode('utf-8'),
            query_string.encode('utf-8'),
            hashlib.sha256
        ).hexdigest()

    def get_server_time(self):
        endpoint = '/api/v3/time'
        response = requests.get(f"{self.base_url}{endpoint}")
        return response.json()['serverTime']

    def get_account_info(self):
        endpoint = '/api/v3/account'
        timestamp = int(time.time() * 1000)

        params = {
            'timestamp': timestamp,
            'recvWindow': 5000
        }

        query_string = urlencode(params)
        signature = self.get_signature(query_string)

        url = f"{self.base_url}{endpoint}?{query_string}&signature={signature}"

        headers = {
            'X-MBX-APIKEY': self.api_key
        }

        response = requests.get(url, headers=headers)
        return response.json()

    def place_market_order(self, side, quantity):
        endpoint = '/api/v3/order'
        timestamp = int(time.time() * 1000)

        params = {
            'symbol': self.symbol,
            'side': side,
            'type': 'MARKET',
            'quantity': quantity,
            'timestamp': timestamp,
            'recvWindow': 5000
        }

        query_string = urlencode(params)
        signature = self.get_signature(query_string)

        url = f"{self.base_url}{endpoint}?{query_string}&signature={signature}"

        headers = {
            'X-MBX-APIKEY': self.api_key
        }

        response = requests.post(url, headers=headers)
        return response.json()

    def start_user_data_stream(self):
        endpoint = '/api/v3/userDataStream'

        headers = {
            'X-MBX-APIKEY': self.api_key
        }

        response = requests.post(f"{self.base_url}{endpoint}", headers=headers)
        self.listen_key = response.json()['listenKey']
        return self.listen_key

    def keep_alive_user_data_stream(self):
        while True:
            if self.listen_key:
                endpoint = '/api/v3/userDataStream'

                params = {
                    'listenKey': self.listen_key
                }

                headers = {
                    'X-MBX-APIKEY': self.api_key
                }

                requests.put(f"{self.base_url}{endpoint}?{urlencode(params)}", headers=headers)
            time.sleep(30 * 60)  # Manter ativo a cada 30 minutos

    def on_message(self, ws, message):
        data = json.loads(message)

        # Lidar com atualizações de conta
        if data.get('e') == 'outboundAccountPosition':
            print("Atualização da conta recebida:")
            print(f"Hora da atualização: {data['u']}")
            print("Saldos:")
            for balance in data['B']:
                print(f"Ativo: {balance['a']}, Livre: {balance['f']}, Bloqueado: {balance['l']}")

        # Lidar com atualizações de ordens
        elif data.get('e') == 'executionReport':
            print("Atualização de ordem recebida:")
            print(f"Símbolo: {data['s']}")
            print(f"Lado: {data['S']}")
            print(f"Tipo de ordem: {data['o']}")
            print(f"Status da ordem: {data['X']}")
            print(f"Preço: {data['p']}")
            print(f"Quantidade: {data['q']}")
            print(f"Quantidade executada: {data['z']}")

            # Implementar lógica de negociação aqui com base nas atualizações de ordens
            if data['X'] == 'FILLED' and data['S'] == 'BUY':
                # Exemplo: Fazer uma ordem de venda quando a compra for preenchida
                sell_price = float(data['p']) * 1.05  # Meta de lucro de 5%
                self.place_limit_order('SELL', data['q'], sell_price)

    def on_error(self, ws, error):
        print(f"Erro: {error}")

    def on_close(self, ws, close_status_code, close_msg):
        print("Conexão WebSocket fechada")

    def on_open(self, ws):
        print("Conexão WebSocket aberta")

    def start_websocket(self):
        self.start_user_data_stream()
        socket_url = f"wss://stream.binance.com:9443/ws/{self.listen_key}"

        self.ws = websocket.WebSocketApp(
            socket_url,
            on_message=self.on_message,
            on_error=self.on_error,
            on_close=self.on_close,
            on_open=self.on_open
        )

        # Iniciar a thread de manutenção
        self.keep_alive_thread = threading.Thread(target=self.keep_alive_user_data_stream)
        self.keep_alive_thread.daemon = True
        self.keep_alive_thread.start()

        # Executar o WebSocket em uma nova thread
        ws_thread = threading.Thread(target=self.ws.run_forever)
        ws_thread.daemon = True
        ws_thread.start()

        return ws_thread

# Exemplo de uso
if __name__ == "__main__":
    api_key = "sua_chave_api"
    api_secret = "sua_chave_secreta"

    trader = BinanceTrader(api_key, api_secret, 'BTCUSDT')

    # Obter informações da conta
    account_info = trader.get_account_info()
    print("Informações da Conta:")
    print(f"Pode Negociar: {account_info['canTrade']}")
    print(f"Pode Retirar: {account_info['canWithdraw']}")
    print(f"Pode Depositar: {account_info['canDeposit']}")

    # Imprimir saldos
    print("\nSaldos:")
    for balance in account_info['balances']:
        if float(balance['free']) > 0 or float(balance['locked']) > 0:
            print(f"Ativo: {balance['asset']}, Livre: {balance['free']}, Bloqueado: {balance['locked']}")

    # Iniciar o WebSocket para atualizações em tempo real
    ws_thread = trader.start_websocket()

    try:
        # Manter a thread principal executando
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        print("Parando...")
        trader.ws.close()

Melhores Práticas para Usar o Binance API

Aqui estão algumas melhores práticas a serem seguidas ao trabalhar com o Binance API:

Práticas de Segurança para Binance API

1. Restringir Permissões da Chave de API: Ative apenas as permissões necessárias (por exemplo, somente leitura, negociação, saques)
2. Whitelist de Endereços IP: Restrinja suas chaves de API para funcionar apenas a partir de endereços IP específicos
3. Armazenamento Seguro das Chaves de API: Nunca insira chaves de API diretamente no seu código, use variáveis de ambiente ou cofres seguros
4. Rotação Regular de Chaves: Mude suas chaves de API periodicamente para mitigar riscos

import os

# Carregar chaves de API a partir de variáveis de ambiente
api_key = os.environ.get('BINANCE_API_KEY')
api_secret = os.environ.get('BINANCE_API_SECRET')

if not api_key or not api_secret:
    raise ValueError("Chaves de API não encontradas nas variáveis de ambiente")

Manipulação de Erros com Binance API

Implemente uma manipulação de erros robusta para lidar com erros de API e problemas de conexão:

def safe_request(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        try:
            response = func(*args, **kwargs)
            # Verificar códigos de erro da API
            if isinstance(response, dict) and 'code' in response:
                print(f"Erro da API: Código {response['code']}, Mensagem: {response.get('msg', 'Sem mensagem')}")
                # Lidar com códigos de erro específicos
                if response['code'] == -1021:  # INVALID_TIMESTAMP
                    print("O horário do servidor e o horário local não estão sincronizados. Ajustando timestamp...")
                    # Implementar lógica de ajuste de timestamp aqui
                elif response['code'] == -2010:  # INSUFFICIENT_BALANCE
                    print("Saldo insuficiente para a ordem.")
                # Adicione mais manipulação de erros específicos conforme necessário
            return response
        except requests.exceptions.RequestException as e:
            print(f"Erro de Solicitação: {e}")
            # Implemente lógica de nova tentativa ou fallback
            return None
        except json.JSONDecodeError:
            print("Falha ao decodificar a resposta JSON")
            return None
        except Exception as e:
            print(f"Erro inesperado: {e}")
            return None
    return wrapper

@safe_request
def get_account_info_safe():
    # Implementação de get_account_info com manipulação de erros apropriada
    # ...

Conclusão

O Binance API oferece uma maneira poderosa de interagir programaticamente com a exchange Binance. Este tutorial abordou os aspectos essenciais do uso da API, desde a autenticação até a realização de ordens e gerenciamento da sua conta. Seguindo as melhores práticas e implementando uma adequada manipulação de erros, você pode construir aplicativos robustos e bots de negociação que aproveitam o vasto conjunto de recursos da Binance.

Lembre-se de manter suas chaves de API seguras, respeitar os limites de taxa e testar minuciosamente seu código em um ambiente de teste antes de usá-lo com fundos reais. À medida que a Binance atualiza sua API, certifique-se de estar informado sobre quaisquer alterações nos endpoints ou parâmetros para garantir que seus aplicativos continuem funcionando corretamente.

Com o conhecimento adquirido neste tutorial, você deve estar bem preparado para começar a construir suas próprias ferramentas e aplicativos de negociação usando o Binance API.