DEV Community: Henrique Vital

Análise de Rompimentos no Ativo US30: Um Script Python para Traders Usando MetaTrader 5

Henrique Vital — Fri, 04 Oct 2024 01:48:32 +0000

import MetaTrader5 as mt5
import pandas as pd
from datetime import datetime, timedelta
import os
import sys

# Parâmetros de login
login = 1520378657  # Substitua pelo seu número de login
senha = "AA6=J?j67mU"  # Substitua pela sua senha
servidor = "FTMO-Demo2"  # Substitua pelo nome do servidor
mt5_path = r"C:\Program Files\FTMO MetaTrader 5\terminal64.exe"  # Caminho do MetaTrader 5

# Inicializa a conexão com o MetaTrader 5
if not mt5.initialize(path=mt5_path):
    print("Falha ao inicializar o MetaTrader 5")
    mt5.shutdown()
    exit()

# Tenta fazer login na conta específica
if not mt5.login(login, password=senha, server=servidor):
    print("Falha ao fazer login no MetaTrader 5")
    print("Erro:", mt5.last_error())
    mt5.shutdown()
    exit()

# Verifica as informações da conta
conta_info = mt5.account_info()
if conta_info is None:
    print("Falha ao obter informações da conta")
    mt5.shutdown()
    exit()

print("Conectado à conta:", conta_info.login)
print("Nome do servidor:", conta_info.server)

# Define o ativo e o período de análise
ativo = "US30.cash"  # Nome do ativo (Dow Jones)
periodo = mt5.TIMEFRAME_M1  # Timeframe de 1 minuto
data_inicio = datetime(2024, 7, 15)  # Data inicial
data_fim = datetime.now()  # Data final (hoje)

# Pega os dados históricos de candles entre as datas especificadas
dados_candles = mt5.copy_rates_range(ativo, periodo, data_inicio, data_fim)

# Converte os dados para um DataFrame para melhor visualização
df = pd.DataFrame(dados_candles)

# Verifica se o DataFrame está vazio
if df.empty:
    print("Nenhum dado foi retornado. Verifique a conexão com o MetaTrader 5 e a disponibilidade do ativo.")
    mt5.shutdown()
    exit()

# Exibe informações sobre o DataFrame
print("Formato do DataFrame:", df.shape)
print("Colunas do DataFrame:", df.columns)
print("Tipos de dados das colunas:")
print(df.dtypes)

# Salva o DataFrame como CSV para inspeção
df.to_csv('dados_us30_1629_1630.csv', index=False)
print("Dados salvos em 'dados_us30_1629_1630.csv'")

# Verifica se a coluna 'time' existe
if 'time' not in df.columns:
    print("A coluna 'time' não foi encontrada. Verifique o nome correto da coluna de tempo.")
    # Tenta encontrar uma coluna que possa ser a de tempo
    time_columns = df.select_dtypes(include=['int64', 'float64']).columns
    if len(time_columns) > 0:
        time_column = time_columns[0]
        print(f"Usando a coluna '{time_column}' como coluna de tempo.")
        df['time'] = pd.to_datetime(df[time_column], unit='s')
    else:
        print("Nenhuma coluna numérica encontrada para usar como tempo.")
        mt5.shutdown()
        exit()
else:
    df['time'] = pd.to_datetime(df['time'], unit='s')

# Filtra os candles das 16:29 e 16:30 (horário MT5)
df_1629 = df[df['time'].dt.time == datetime.strptime("16:29", "%H:%M").time()]
df_1630 = df[df['time'].dt.time == datetime.strptime("16:30", "%H:%M").time()]

# Exibe os resultados filtrados
print("\nCandle das 16:29:")
print(df_1629)

print("\nCandle das 16:30:")
print(df_1630)

# Inicializa contadores para as estatísticas
stats = {
    '0': 0,
    '0-500': 0,
    '500-1000': 0,
    '1000-1500': 0,
    '1500-2000': 0,
    '2000-2500': 0,
    '2500-3000': 0,
    '3000-3500': 0,
    '3500-4000': 0,
    '4000-4500': 0,
    '4500-5000': 0,
    '5000-5500': 0,
    '5500-6000': 0,
    '6000-6500': 0,
    '6500-7000': 0,
    '7000-7500': 0,
    '7500-8000': 0,
    '8000-8500': 0,
    '8500-9000': 0,
    '9000-9500': 0,
    '9500-10000': 0,
    '10000+': 0
}

# Inicializa contadores para as novas estatísticas
rompimentos_um_lado = 0
rompimentos_dois_lados = 0
sem_rompimentos = 0

# Inicializa uma lista para armazenar os detalhes dos rompimentos
detalhes_rompimentos = []

# Calcula os rompimentos do candle das 16:30 em relação ao das 16:29
for i in range(min(len(df_1629), len(df_1630))):
    data = df_1629.iloc[i]['time'].date()
    max_1629 = df_1629.iloc[i]['high']
    min_1629 = df_1629.iloc[i]['low']
    max_1630 = df_1630.iloc[i]['high']
    min_1630 = df_1630.iloc[i]['low']
    volume_1629 = df_1629.iloc[i]['tick_volume']
    volume_1630 = df_1630.iloc[i]['tick_volume']
    tamanho_1629 = (max_1629 - min_1629) * 100  # Multiplicado por 100
    tamanho_1630 = (max_1630 - min_1630) * 100  # Multiplicado por 100

    rompimento_baixo = (min_1629 - min_1630) * 100 if min_1630 < min_1629 else 0
    rompimento_cima = (max_1630 - max_1629) * 100 if max_1630 > max_1629 else 0

    # Adiciona os detalhes à lista
    detalhes_rompimentos.append({
        'DT': data,
        'RBaixo': rompimento_baixo,
        'RCima': rompimento_cima,
        'V29': volume_1629,
        'V30': volume_1630,
        'T29': tamanho_1629,
        'T30': tamanho_1630
    })

    # Atualiza as estatísticas para rompimento baixo e cima
    for rompimento in [rompimento_baixo, rompimento_cima]:
        if rompimento == 0:
            stats['0'] += 1
        elif 0 < rompimento <= 500:
            stats['0-500'] += 1
        elif 500 < rompimento <= 1000:
            stats['500-1000'] += 1
        elif 1000 < rompimento <= 1500:
            stats['1000-1500'] += 1
        elif 1500 < rompimento <= 2000:
            stats['1500-2000'] += 1
        elif 2000 < rompimento <= 2500:
            stats['2000-2500'] += 1
        elif 2500 < rompimento <= 3000:
            stats['2500-3000'] += 1
        elif 3000 < rompimento <= 3500:
            stats['3000-3500'] += 1
        elif 3500 < rompimento <= 4000:
            stats['3500-4000'] += 1
        # Continue a adicionar as condições conforme necessário

# Aqui você pode adicionar mais lógica para análise ou visualização dos resultados

Guia Completo: Construindo um Script Seguro para Análise de Dados no MetaTrader5 com Python

Henrique Vital — Fri, 04 Oct 2024 01:12:26 +0000

Bem-vindo! Este guia é destinado a iniciantes em Python que desejam criar um script para conectar-se ao MetaTrader5 (MT5), extrair dados de candles, analisar rompimentos e gerar estatísticas detalhadas. Vamos abordar cada parte do código, explicar sua lógica e implementar práticas de segurança para proteger suas credenciais sensíveis.

Índice

Introdução
Pré-requisitos
Configurando o Ambiente
- 1. Criar um Diretório para o Projeto
- 2. Configurar um Ambiente Virtual (Opcional, mas Recomendado)
- 3. Instalar Dependências Necessárias
Criando o Arquivo .env
Entendendo o Script Python
- Imports e Variáveis de Ambiente
- Configuração de Logging
- Configuração das Contas MT5
- Conexão com o MetaTrader5
- Extração de Dados de Candles
- Processamento e Análise de Dados
- Geração de Estatísticas
- Salvando Resultados
- Encerramento da Conexão
Executando o Script
Boas Práticas e Dicas
Conclusão

Introdução

Neste projeto, você criará um script em Python que realiza as seguintes tarefas:

Conectar-se ao MetaTrader5 (MT5): Utiliza as credenciais fornecidas para acessar uma conta específica no MT5.
Extrair Dados de Candles: Obtém dados históricos de candles para um ativo específico dentro de um período definido.
Processar e Analisar Dados: Converte os dados em um DataFrame do pandas, filtra candles em horários específicos e calcula rompimentos.
Gerar Relatórios: Salva os dados extraídos e as estatísticas em arquivos CSV e TXT para inspeção posterior.
Manter Práticas de Segurança: Gerencia credenciais sensíveis usando variáveis de ambiente para evitar exposição no código.

Ao final deste guia, você terá um script funcional e seguro que automatiza a extração e análise de dados do MT5.

Pré-requisitos

Antes de começar, certifique-se de ter o seguinte:

Conhecimento Básico de Python: Familiaridade com a sintaxe do Python e conceitos como funções, loops e módulos.
Python Instalado: Verifique se você tem o Python 3.7 ou posterior instalado em seu computador. Você pode baixá-lo do site oficial do Python.
Conta no MetaTrader5 (MT5): Acesse uma conta MT5, incluindo credenciais de login, senha e informações do servidor.
Entendimento de Ambientes Virtuais: Embora opcional, usar ambientes virtuais é recomendado para gerenciar dependências.

Configurando o Ambiente

1. Criar um Diretório para o Projeto

Escolha um local em seu computador e crie um novo diretório para o seu projeto. Por exemplo:

mkdir mt5_data_analysis
cd mt5_data_analysis

2. Configurar um Ambiente Virtual (Opcional, mas Recomendado)

Um ambiente virtual isola as dependências do seu projeto das de outros projetos Python em seu sistema.

Criar um Ambiente Virtual:
```
python -m venv venv
```
Este comando cria um ambiente virtual chamado venv dentro do diretório do seu projeto.
Ativar o Ambiente Virtual:
- No Windows:
```
venv\Scripts\activate
```
- No macOS e Linux:
```
source venv/bin/activate
```
Após a ativação, o nome do ambiente virtual aparecerá no prompt do terminal, indicando que as dependências instaladas agora serão isoladas para este projeto.

3. Instalar Dependências Necessárias

Com o ambiente virtual ativado, instale as bibliotecas necessárias usando o pip:

pip install MetaTrader5 pandas python-dotenv

MetaTrader5: Biblioteca para interagir com o MetaTrader5.
pandas: Biblioteca poderosa para análise e manipulação de dados.
python-dotenv: Biblioteca para carregar variáveis de ambiente a partir de um arquivo .env.

Além disso, para facilitar o gerenciamento de pacotes, é recomendável criar um arquivo requirements.txt:

pip freeze > requirements.txt

Este arquivo registra todas as dependências do projeto, permitindo que outras pessoas (ou você mesmo em outra máquina) instalem facilmente as mesmas versões com:

pip install -r requirements.txt

Criando o Arquivo `.env`

Para gerenciar informações sensíveis de forma segura, usaremos um arquivo .env que armazenará variáveis de ambiente. Este arquivo não deve ser versionado ou compartilhado.

1. Criar o Arquivo `.env`

No diretório raiz do seu projeto (mt5_data_analysis), crie um arquivo chamado .env e adicione as seguintes linhas, substituindo os valores pelos seus próprios:

# Credenciais do MetaTrader5 (MT5)
MT5_LOGIN=1520378657
MT5_PASSWORD=AA6=J?j67mU
MT5_SERVER=FTMO-Demo2
MT5_PATH=C:\Program Files\FTMO MetaTrader 5\terminal64.exe

# Configuração de Logging
LOG_FILE=app.log

Notas Importantes:

MT5_LOGIN: Número de login da sua conta MT5.
MT5_PASSWORD: Senha da sua conta MT5.
MT5_SERVER: Nome do servidor MT5.
MT5_PATH: Caminho para o executável do MetaTrader5 no seu sistema. Certifique-se de ajustar este caminho conforme a instalação do MT5 em seu computador.
LOG_FILE: Nome do arquivo onde os logs serão salvos.

2. Atualizar o Arquivo `.gitignore`

Para garantir que o arquivo .env não seja versionado (especialmente se você estiver usando o Git), crie ou atualize o arquivo .gitignore no diretório raiz com o seguinte conteúdo:

# Arquivo de variáveis de ambiente
.env

# Diretório do ambiente virtual
venv/

Isso impede que informações sensíveis sejam acidentalmente compartilhadas em repositórios públicos ou privados.

Entendendo o Script Python

Agora, vamos criar o script Python que conectará ao MT5, extrairá dados de candles, processará esses dados e gerará estatísticas detalhadas. Este script também implementará práticas de segurança para proteger suas credenciais.

1. Imports e Variáveis de Ambiente

Primeiro, importamos todas as bibliotecas necessárias e carregamos as variáveis de ambiente.

import os
import sys
import MetaTrader5 as mt5
import pandas as pd
from datetime import datetime, timedelta
import logging
from dotenv import load_dotenv

Explicação dos Imports:

os: Interage com o sistema operacional, acessando variáveis de ambiente e caminhos de arquivos.
sys: Fornece acesso a variáveis e funções que interagem com o interpretador Python.
MetaTrader5 (mt5): Biblioteca para interagir com o MetaTrader5.
pandas: Biblioteca poderosa para análise e manipulação de dados.
datetime e timedelta: Trabalha com datas e horas.
logging: Gerencia o registro de mensagens de log para monitoramento e depuração.
dotenv: Carrega variáveis de ambiente de um arquivo .env.

Carregando Variáveis de Ambiente:

# Carrega variáveis de ambiente do arquivo .env
load_dotenv()

Esta linha lê o arquivo .env e carrega as variáveis de ambiente nele definidas, permitindo que sejam acessadas via os.getenv().

2. Configuração de Logging

O logging é essencial para monitorar o funcionamento do seu script e depurar problemas. Vamos configurar o logging para registrar mensagens tanto no console quanto em um arquivo.

# Acessa as variáveis de ambiente para configuração
LOG_FILE = os.getenv('LOG_FILE', 'app.log')

# Configura o logging
logging.basicConfig(
    level=logging.INFO,  # Nível mínimo de severidade para registrar
    format='%(asctime)s - %(levelname)s - %(message)s',  # Formato das mensagens de log
    handlers=[
        logging.FileHandler(LOG_FILE),  # Registra logs em um arquivo
        logging.StreamHandler(sys.stdout)  # Exibe logs no console
    ]
)
logger = logging.getLogger(__name__)

Detalhes da Configuração:

level: Define o nível mínimo de mensagens que serão registradas. INFO inclui INFO, WARNING, ERROR e CRITICAL.
format: Define como cada mensagem de log será formatada, incluindo o timestamp, o nível de severidade e a mensagem.
handlers: Define onde as mensagens de log serão enviadas. Aqui, estão sendo enviadas para um arquivo (app.log) e para o console.

3. Configuração das Contas MT5

Definimos as contas MT5 que o script utilizará para conectar e extrair dados.

# Configurações das contas MT5
CONTAS_MT5 = [
    {
        "login": os.getenv('MT5_LOGIN'),
        "senha": os.getenv('MT5_PASSWORD'),
        "servidor": os.getenv('MT5_SERVER'),
        "ativo": "US30.cash",  # Nome do ativo (Dow Jones)
        "periodo": mt5.TIMEFRAME_M1,  # Timeframe de 1 minuto
        "lote": 0.01  # Volume padrão das ordens (se aplicável)
    }
]

contas_ativas = []  # Lista para armazenar contas ativas

Explicação dos Campos:

login: Número de login da conta MT5.
senha: Senha da conta MT5.
servidor: Nome do servidor MT5.
ativo: Símbolo do ativo a ser analisado (por exemplo, "US30.cash").
periodo: Timeframe dos dados de candles (por exemplo, mt5.TIMEFRAME_M1 para 1 minuto).
lote: Volume padrão das ordens, caso o script seja expandido para enviar ordens (não utilizado neste script específico, mas mantido para futuras funcionalidades).

4. Conexão com o MetaTrader5

Estabelecemos a conexão com o MT5 utilizando as credenciais fornecidas.

# Inicializa a conexão com o MetaTrader 5
if not mt5.initialize(path=os.getenv('MT5_PATH')):
    logger.error("Falha ao inicializar o MetaTrader 5")
    mt5.shutdown()
    sys.exit()

# Itera sobre todas as contas configuradas para realizar login
for conta in CONTAS_MT5:
    if not mt5.login(int(conta['login']), password=conta['senha'], server=conta['servidor']):
        logger.error(f"Falha ao fazer login no MetaTrader 5 para a conta {conta['login']}")
        logger.error(f"Erro: {mt5.last_error()}")
        mt5.shutdown()
        sys.exit()
    else:
        contas_ativas.append(conta)
        logger.info(f"Conectado à conta: {conta['login']} no servidor: {conta['servidor']}")

Detalhes do Processo:

Inicialização do MT5:
- Utiliza o caminho para o executável do MT5 (MT5_PATH) para inicializar a conexão.
- Se a inicialização falhar, registra um erro, encerra a conexão e finaliza o script.
Login nas Contas MT5:
- Itera sobre cada conta definida em CONTAS_MT5.
- Tenta fazer login usando o login, senha e servidor fornecidos.
- Se o login falhar, registra um erro, encerra a conexão e finaliza o script.
- Se o login for bem-sucedido, adiciona a conta à lista contas_ativas e registra uma mensagem de sucesso.

5. Extração de Dados de Candles

Extraímos os dados históricos de candles para o ativo e período definidos.

# Define o ativo e o período de análise
ativo = contas_ativas[0]['ativo']  # Nome do ativo (por exemplo, "US30.cash")
periodo = contas_ativas[0]['periodo']  # Timeframe (por exemplo, mt5.TIMEFRAME_M1)
data_inicio = datetime(2024, 7, 15)  # Data inicial
data_fim = datetime.now()  # Data final (hoje)

logger.info(f"Extraindo dados de candles para {ativo} de {data_inicio} até {data_fim}")

# Pega os dados históricos de candles entre as datas especificadas
dados_candles = mt5.copy_rates_range(ativo, periodo, data_inicio, data_fim)

# Verifica se os dados foram retornados
if dados_candles is None:
    logger.error("Falha ao copiar dados de candles. Verifique a conexão e os parâmetros.")
    mt5.shutdown()
    sys.exit()

# Converte os dados para um DataFrame para melhor visualização
df = pd.DataFrame(dados_candles)

# Verifica se o DataFrame está vazio
if df.empty:
    logger.error("Nenhum dado foi retornado. Verifique a conexão com o MetaTrader 5 e a disponibilidade do ativo.")
    mt5.shutdown()
    sys.exit()

logger.info(f"Dados de candles extraídos: {df.shape[0]} registros")

Detalhes do Processo:

Definição dos Parâmetros:
- ativo: Símbolo do ativo a ser analisado.
- periodo: Timeframe dos candles (por exemplo, 1 minuto).
- data_inicio: Data inicial para a extração dos dados.
- data_fim: Data final para a extração dos dados (geralmente a data atual).
Extração dos Dados:
- Utiliza a função mt5.copy_rates_range para obter os dados de candles dentro do intervalo de datas especificado.
- Converte os dados retornados em um DataFrame do pandas para facilitar a manipulação e análise.
- Verifica se os dados foram extraídos com sucesso. Se não, encerra o script com uma mensagem de erro.

6. Processamento e Análise de Dados

Processamos os dados extraídos, filtrando candles em horários específicos e calculando rompimentos.

# Salva o DataFrame como CSV para inspeção
df.to_csv('dados_us30.csv', index=False)
logger.info("Dados salvos em 'dados_us30.csv'")

# Verifica se a coluna 'time' existe
if 'time' not in df.columns:
    logger.warning("A coluna 'time' não foi encontrada. Verifique o nome correto da coluna de tempo.")
    # Tenta encontrar uma coluna que possa ser a de tempo
    time_columns = df.select_dtypes(include=['int64', 'float64']).columns
    if len(time_columns) > 0:
        time_column = time_columns[0]
        logger.info(f"Usando a coluna '{time_column}' como coluna de tempo.")
        df['time'] = pd.to_datetime(df[time_column], unit='s')
    else:
        logger.error("Nenhuma coluna numérica encontrada para usar como tempo.")
        mt5.shutdown()
        sys.exit()
else:
    df['time'] = pd.to_datetime(df['time'], unit='s')

# Filtra os candles das 16:29 e 16:30 (horário MT5)
hora1 = datetime.strptime("16:29", "%H:%M").time()
hora2 = datetime.strptime("16:30", "%H:%M").time()

df_1629 = df[df['time'].dt.time == hora1]
df_1630 = df[df['time'].dt.time == hora2]

# Exibe os resultados filtrados
logger.info(f"Candles das {hora1} encontradas: {df_1629.shape[0]}")
logger.info(f"Candles das {hora2} encontradas: {df_1630.shape[0]}")

# Verifica se há dados suficientes para análise
if df_1629.empty or df_1630.empty:
    logger.error("Não há candles suficientes para realizar a análise.")
    mt5.shutdown()
    sys.exit()

Detalhes do Processo:

Salvando Dados para Inspeção:
- Salva o DataFrame extraído em um arquivo CSV (dados_us30.csv) para que você possa inspecionar os dados manualmente, se necessário.
Verificação e Conversão da Coluna de Tempo:
- Verifica se a coluna 'time' existe no DataFrame.
- Se não existir, tenta identificar uma coluna numérica que possa representar o tempo e converte para um formato datetime.
- Se não encontrar nenhuma coluna adequada, encerra o script com uma mensagem de erro.
Filtrando Candles em Horários Específicos:
- Define os horários de interesse: 16:29 e 16:30.
- Filtra os candles que correspondem a esses horários.
- Verifica se há candles suficientes para realizar a análise. Se não houver, encerra o script.

7. Geração de Estatísticas

Calculamos os rompimentos entre os candles das 16:29 e 16:30 e categorizamos esses rompimentos em diferentes faixas.

# Inicializa contadores para as estatísticas
stats = {
    '0': 0,
    '0-500': 0,
    '500-1000': 0,
    '1000-1500': 0,
    '1500-2000': 0,
    '2000-2500': 0,
    '2500-3000': 0,
    '3000-3500': 0,
    '3500+': 0
}

# Inicializa contadores para as novas estatísticas
rompimentos_um_lado = 0
rompimentos_dois_lados = 0
sem_rompimentos = 0

# Inicializa uma lista para armazenar os detalhes dos rompimentos
detalhes_rompimentos = []

# Calcula os rompimentos do candle das 16:30 em relação ao das 16:29
for i in range(min(len(df_1629), len(df_1630))):
    data = df_1629.iloc[i]['time'].date()
    max_1629 = df_1629.iloc[i]['high']
    min_1629 = df_1629.iloc[i]['low']
    max_1630 = df_1630.iloc[i]['high']
    min_1630 = df_1630.iloc[i]['low']

    rompimento_baixo = (min_1629 - min_1630) * 100 if min_1630 < min_1629 else 0
    rompimento_cima = (max_1630 - max_1629) * 100 if max_1630 > max_1629 else 0

    logger.info(f"Dia {data} -> Rompimento Baixo: {rompimento_baixo:.2f}, Rompimento Cima: {rompimento_cima:.2f}")

    # Adiciona os detalhes à lista
    detalhes_rompimentos.append({
        'data': data,
        'rompimento_baixo': rompimento_baixo,
        'rompimento_cima': rompimento_cima
    })

    # Atualiza as estatísticas para rompimento baixo e cima
    for rompimento in [rompimento_baixo, rompimento_cima]:
        if rompimento == 0:
            stats['0'] += 1
        elif 0 < rompimento <= 500:
            stats['0-500'] += 1
        elif 500 < rompimento <= 1000:
            stats['500-1000'] += 1
        elif 1000 < rompimento <= 1500:
            stats['1000-1500'] += 1
        elif 1500 < rompimento <= 2000:
            stats['1500-2000'] += 1
        elif 2000 < rompimento <= 2500:
            stats['2000-2500'] += 1
        elif 2500 < rompimento <= 3000:
            stats['2500-3000'] += 1
        elif 3000 < rompimento <= 3500:
            stats['3000-3500'] += 1
        else:
            stats['3500+'] += 1

    # Atualiza as novas estatísticas
    if rompimento_baixo > 0 and rompimento_cima > 0:
        rompimentos_dois_lados += 1
    elif rompimento_baixo > 0 or rompimento_cima > 0:
        rompimentos_um_lado += 1
    else:
        sem_rompimentos += 1

Detalhes do Processo:

Inicialização de Contadores:
- stats: Dicionário que categoriza os rompimentos em diferentes faixas.
- rompimentos_um_lado, rompimentos_dois_lados, sem_rompimentos: Contadores adicionais para categorizar os rompimentos de formas específicas.
- detalhes_rompimentos: Lista para armazenar detalhes de cada rompimento.
Cálculo dos Rompimentos:
- Itera sobre cada par de candles das 16:29 e 16:30.
- Calcula o rompimento abaixo (rompimento_baixo) e acima (rompimento_cima).
- Multiplica por 100 para converter em uma escala mais legível.
- Adiciona os detalhes do rompimento à lista detalhes_rompimentos.
- Atualiza os contadores no dicionário stats com base no valor do rompimento.
- Atualiza os contadores adicionais para categorizar os rompimentos de formas específicas.

8. Salvando Resultados

Geramos um relatório com todas as estatísticas calculadas e salvamos em um arquivo de texto.

# Cria uma string com todas as estatísticas
estatisticas = f"""
Total de dias analisados: {len(df_1629)}

Detalhes dos Rompimentos:
"""

for detalhe in detalhes_rompimentos:
    estatisticas += f"Data: {detalhe['data']}, Rompimento Baixo: {detalhe['rompimento_baixo']:.2f}, Rompimento Cima: {detalhe['rompimento_cima']:.2f}\n"

estatisticas += f"""
Estatisticas de Rompimentos 16:29 a 16:30 :
Rompimentos = 0: {stats['0']}
Rompimentos > 0 e <= 500: {stats['0-500']}
Rompimentos > 500 e <= 1000: {stats['500-1000']}
Rompimentos > 1000 e <= 1500: {stats['1000-1500']}
Rompimentos > 1500 e <= 2000: {stats['1500-2000']}
Rompimentos > 2000 e <= 2500: {stats['2000-2500']}
Rompimentos > 2500 e <= 3000: {stats['2500-3000']}
Rompimentos > 3000 e <= 3500: {stats['3000-3500']}
Rompimentos > 3500: {stats['3500+']}

Estatisticas adicionais:
Candles rompidos somente de um lado: {rompimentos_um_lado}
Candles rompidos dos dois lados: {rompimentos_dois_lados}
Candles não rompidos de nenhum lado: {sem_rompimentos}
"""

# Imprime as estatísticas no console
print(estatisticas)
logger.info("Estatísticas calculadas e impressas no console.")

# Obtenha o nome do arquivo atual e mude a extensão para .txt
nome_arquivo_atual = os.path.basename(sys.argv[0])
nome_arquivo_base = os.path.splitext(nome_arquivo_atual)[0]
nome_arquivo_txt = f"{nome_arquivo_base}.txt"

# Salve as estatísticas em um arquivo de texto
with open(nome_arquivo_txt, 'w') as arquivo:
    arquivo.write(estatisticas)

logger.info(f"Estatísticas salvas no arquivo: {os.path.abspath(nome_arquivo_txt)}")

Detalhes do Processo:

Criação da String de Estatísticas:
- Compila todas as estatísticas em uma única string formatada para fácil leitura.
Impressão no Console:
- Exibe as estatísticas no console para visualização imediata.
Salvando em Arquivo de Texto:
- Determina o nome do arquivo atual (nome_arquivo_atual) e altera a extensão para .txt.
- Salva as estatísticas no arquivo de texto, permitindo que você consulte os resultados posteriormente.

9. Encerramento da Conexão

Finalizamos a conexão com o MT5 para liberar recursos.

# Desconecta o MetaTrader 5
mt5.shutdown()
logger.info("Conexão com o MetaTrader 5 encerrada.")

Detalhes do Processo:

mt5.shutdown(): Fecha a conexão com o MT5 de forma ordenada.
Log: Registra uma mensagem indicando que a conexão foi encerrada.

Script Python Completo com Práticas de Segurança

A seguir, apresentamos o script completo incorporando todas as práticas discutidas, especialmente a gestão segura de credenciais usando variáveis de ambiente.

import os
import sys
import MetaTrader5 as mt5
import pandas as pd
from datetime import datetime, timedelta
import logging
from dotenv import load_dotenv

# Carrega variáveis de ambiente do arquivo .env
load_dotenv()

# Acessa as variáveis de ambiente para configuração
MT5_LOGIN = os.getenv('MT5_LOGIN')
MT5_PASSWORD = os.getenv('MT5_PASSWORD')
MT5_SERVER = os.getenv('MT5_SERVER')
MT5_PATH = os.getenv('MT5_PATH')
LOG_FILE = os.getenv('LOG_FILE', 'app.log')

# Configura logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler(LOG_FILE),
        logging.StreamHandler(sys.stdout)
    ]
)
logger = logging.getLogger(__name__)

# Configurações das contas MT5
CONTAS_MT5 = [
    {
        "login": MT5_LOGIN,
        "senha": MT5_PASSWORD,
        "servidor": MT5_SERVER,
        "ativo": "US30.cash",  # Nome do ativo (Dow Jones)
        "periodo": mt5.TIMEFRAME_M1,  # Timeframe de 1 minuto
        "lote": 0.01  # Volume padrão das ordens (se aplicável)
    }
]

contas_ativas = []  # Lista para armazenar contas ativas

# Inicializa a conexão com o MetaTrader 5
if not mt5.initialize(path=MT5_PATH):
    logger.error("Falha ao inicializar o MetaTrader 5")
    mt5.shutdown()
    sys.exit()

# Itera sobre todas as contas configuradas para realizar login
for conta in CONTAS_MT5:
    if not mt5.login(int(conta['login']), password=conta['senha'], server=conta['servidor']):
        logger.error(f"Falha ao fazer login no MetaTrader 5 para a conta {conta['login']}")
        logger.error(f"Erro: {mt5.last_error()}")
        mt5.shutdown()
        sys.exit()
    else:
        contas_ativas.append(conta)
        logger.info(f"Conectado à conta: {conta['login']} no servidor: {conta['servidor']}")

# Define o ativo e o período de análise
ativo = contas_ativas[0]['ativo']  # Nome do ativo (por exemplo, "US30.cash")
periodo = contas_ativas[0]['periodo']  # Timeframe (por exemplo, mt5.TIMEFRAME_M1)
data_inicio = datetime(2024, 7, 15)  # Data inicial
data_fim = datetime.now()  # Data final (hoje)

logger.info(f"Extraindo dados de candles para {ativo} de {data_inicio} até {data_fim}")

# Pega os dados históricos de candles entre as datas especificadas
dados_candles = mt5.copy_rates_range(ativo, periodo, data_inicio, data_fim)

# Verifica se os dados foram retornados
if dados_candles is None:
    logger.error("Falha ao copiar dados de candles. Verifique a conexão e os parâmetros.")
    mt5.shutdown()
    sys.exit()

# Converte os dados para um DataFrame para melhor visualização
df = pd.DataFrame(dados_candles)

# Verifica se o DataFrame está vazio
if df.empty:
    logger.error("Nenhum dado foi retornado. Verifique a conexão com o MetaTrader 5 e a disponibilidade do ativo.")
    mt5.shutdown()
    sys.exit()

logger.info(f"Dados de candles extraídos: {df.shape[0]} registros")

# Salva o DataFrame como CSV para inspeção
df.to_csv('dados_us30.csv', index=False)
logger.info("Dados salvos em 'dados_us30.csv'")

# Verifica se a coluna 'time' existe
if 'time' not in df.columns:
    logger.warning("A coluna 'time' não foi encontrada. Verifique o nome correto da coluna de tempo.")
    # Tenta encontrar uma coluna que possa ser a de tempo
    time_columns = df.select_dtypes(include=['int64', 'float64']).columns
    if len(time_columns) > 0:
        time_column = time_columns[0]
        logger.info(f"Usando a coluna '{time_column}' como coluna de tempo.")
        df['time'] = pd.to_datetime(df[time_column], unit='s')
    else:
        logger.error("Nenhuma coluna numérica encontrada para usar como tempo.")
        mt5.shutdown()
        sys.exit()
else:
    df['time'] = pd.to_datetime(df['time'], unit='s')

# Filtra os candles das 16:29 e 16:30 (horário MT5)
hora1 = datetime.strptime("16:29", "%H:%M").time()
hora2 = datetime.strptime("16:30", "%H:%M").time()

df_1629 = df[df['time'].dt.time == hora1]
df_1630 = df[df['time'].dt.time == hora2]

# Exibe os resultados filtrados
logger.info(f"Candles das {hora1} encontradas: {df_1629.shape[0]}")
logger.info(f"Candles das {hora2} encontradas: {df_1630.shape[0]}")

# Verifica se há dados suficientes para análise
if df_1629.empty or df_1630.empty:
    logger.error("Não há candles suficientes para realizar a análise.")
    mt5.shutdown()
    sys.exit()

# Inicializa contadores para as estatísticas
stats = {
    '0': 0,
    '0-500': 0,
    '500-1000': 0,
    '1000-1500': 0,
    '1500-2000': 0,
    '2000-2500': 0,
    '2500-3000': 0,
    '3000-3500': 0,
    '3500+': 0
}

# Inicializa contadores para as novas estatísticas
rompimentos_um_lado = 0
rompimentos_dois_lados = 0
sem_rompimentos = 0

# Inicializa uma lista para armazenar os detalhes dos rompimentos
detalhes_rompimentos = []

# Calcula os rompimentos do candle das 16:30 em relação ao das 16:29
for i in range(min(len(df_1629), len(df_1630))):
    data = df_1629.iloc[i]['time'].date()
    max_1629 = df_1629.iloc[i]['high']
    min_1629 = df_1629.iloc[i]['low']
    max_1630 = df_1630.iloc[i]['high']
    min_1630 = df_1630.iloc[i]['low']

    rompimento_baixo = (min_1629 - min_1630) * 100 if min_1630 < min_1629 else 0
    rompimento_cima = (max_1630 - max_1629) * 100 if max_1630 > max_1629 else 0

    logger.info(f"Dia {data} -> Rompimento Baixo: {rompimento_baixo:.2f}, Rompimento Cima: {rompimento_cima:.2f}")

    # Adiciona os detalhes à lista
    detalhes_rompimentos.append({
        'data': data,
        'rompimento_baixo': rompimento_baixo,
        'rompimento_cima': rompimento_cima
    })

    # Atualiza as estatísticas para rompimento baixo e cima
    for rompimento in [rompimento_baixo, rompimento_cima]:
        if rompimento == 0:
            stats['0'] += 1
        elif 0 < rompimento <= 500:
            stats['0-500'] += 1
        elif 500 < rompimento <= 1000:
            stats['500-1000'] += 1
        elif 1000 < rompimento <= 1500:
            stats['1000-1500'] += 1
        elif 1500 < rompimento <= 2000:
            stats['1500-2000'] += 1
        elif 2000 < rompimento <= 2500:
            stats['2000-2500'] += 1
        elif 2500 < rompimento <= 3000:
            stats['2500-3000'] += 1
        elif 3000 < rompimento <= 3500:
            stats['3000-3500'] += 1
        else:
            stats['3500+'] += 1

    # Atualiza as novas estatísticas
    if rompimento_baixo > 0 and rompimento_cima > 0:
        rompimentos_dois_lados += 1
    elif rompimento_baixo > 0 or rompimento_cima > 0:
        rompimentos_um_lado += 1
    else:
        sem_rompimentos += 1

# Cria uma string com todas as estatísticas
estatisticas = f"""
Total de dias analisados: {len(df_1629)}

Detalhes dos Rompimentos:
"""

for detalhe in detalhes_rompimentos:
    estatisticas += f"Data: {detalhe['data']}, Rompimento Baixo: {detalhe['rompimento_baixo']:.2f}, Rompimento Cima: {detalhe['rompimento_cima']:.2f}\n"

estatisticas += f"""
Estatisticas de Rompimentos 16:29 a 16:30 :
Rompimentos = 0: {stats['0']}
Rompimentos > 0 e <= 500: {stats['0-500']}
Rompimentos > 500 e <= 1000: {stats['500-1000']}
Rompimentos > 1000 e <= 1500: {stats['1000-1500']}
Rompimentos > 1500 e <= 2000: {stats['1500-2000']}
Rompimentos > 2000 e <= 2500: {stats['2000-2500']}
Rompimentos > 2500 e <= 3000: {stats['2500-3000']}
Rompimentos > 3000 e <= 3500: {stats['3000-3500']}
Rompimentos > 3500: {stats['3500+']}

Estatisticas adicionais:
Candles rompidos somente de um lado: {rompimentos_um_lado}
Candles rompidos dos dois lados: {rompimentos_dois_lados}
Candles não rompidos de nenhum lado: {sem_rompimentos}
"""

# Imprime as estatísticas no console
print(estatisticas)
logger.info("Estatísticas calculadas e impressas no console.")

# Obtenha o nome do arquivo atual e mude a extensão para .txt
nome_arquivo_atual = os.path.basename(sys.argv[0])
nome_arquivo_base = os.path.splitext(nome_arquivo_atual)[0]
nome_arquivo_txt = f"{nome_arquivo_base}.txt"

# Salve as estatísticas em um arquivo de texto
with open(nome_arquivo_txt, 'w') as arquivo:
    arquivo.write(estatisticas)

logger.info(f"Estatísticas salvas no arquivo: {os.path.abspath(nome_arquivo_txt)}")

# Desconecta o MetaTrader 5
mt5.shutdown()
logger.info("Conexão com o MetaTrader 5 encerrada.")

Executando o Script

Após configurar o ambiente e criar o arquivo analyze_mt5.py, siga os passos abaixo para executar o script.

1. Ativar o Ambiente Virtual

Se você configurou um ambiente virtual, ative-o:

No Windows:
```
venv\Scripts\activate
```
No macOS e Linux:
```
source venv/bin/activate
```

2. Executar o Script

No diretório do seu projeto, execute o seguinte comando:

python analyze_mt5.py

O que esperar:

Logs no Console: Você verá mensagens informando sobre o status da conexão com o MT5, extração de dados e cálculo de estatísticas.
Arquivo de Log: Todas as mensagens de log também serão gravadas no arquivo app.log (ou outro definido em LOG_FILE).
Arquivo CSV: Os dados de candles serão salvos em dados_us30.csv para inspeção manual.
Arquivo TXT: As estatísticas calculadas serão salvas em um arquivo de texto com o mesmo nome do script (por exemplo, analyze_mt5.txt).

3. Encerrando o Script

O script encerra automaticamente após concluir todas as operações. Se desejar interromper o script manualmente antes que ele termine, pressione Ctrl+C no terminal onde o script está sendo executado. No entanto, como o script executa todas as operações de forma síncrona e termina após a conclusão, isso normalmente não será necessário.

Boas Práticas e Dicas

Manutenção da Segurança:
- Nunca compartilhe seu arquivo .env ou exponha suas credenciais em repositórios públicos.
- Use Ambientes Virtuais para isolar as dependências do projeto.
Monitoramento e Logs:
- Revise regularmente os logs (app.log) para identificar e resolver possíveis problemas.
- Implementar Rotação de Logs: Para evitar que o arquivo de log cresça indefinidamente, considere implementar rotação usando bibliotecas como logging.handlers.
Testes em Ambiente de Demonstração:
- Antes de operar em contas reais, teste seu script em ambientes de demonstração para garantir que tudo funcione conforme o esperado.
- Verifique os Dados: Assegure-se de que os dados extraídos estão corretos e que os cálculos de estatísticas estão precisos.
Gerenciamento de Erros:
- Tratamento Robusto de Exceções: Certifique-se de que todas as possíveis exceções sejam tratadas para evitar que o script pare inesperadamente.
- Alertas: Considere implementar alertas (por exemplo, enviando mensagens de log críticas para seu email) para notificá-lo sobre falhas graves.
Atualizações de Dependências:
- Mantenha suas bibliotecas atualizadas para aproveitar melhorias de segurança e correções de bugs.
- Verifique as Documentações: Consulte as documentações oficiais das bibliotecas para entender mudanças ou atualizações.
Escalabilidade:
- Adicionar Mais Ativos: Atualize o script para suportar mais ativos além de "US30.cash" conforme sua necessidade.
- Automatizar Tarefas: Se desejar realizar análises regularmente, considere agendar a execução do script usando ferramentas como cron no Linux ou Agendador de Tarefas no Windows.
Documentação e Comentários:
- Comente Seu Código: Mantenha comentários atualizados e relevantes para facilitar futuras manutenções.
- Documente Configurações: Mantenha um documento com detalhes sobre como configurar e executar o script.

Conclusão

Parabéns! Você criou um script em Python que conecta ao MetaTrader5, extrai dados de candles, processa e analisa esses dados, e gera estatísticas detalhadas. Além disso, implementou práticas de segurança essenciais para proteger suas credenciais e garantir a integridade do seu projeto.

Próximos Passos:

Expandir Funcionalidades: Adicione mais funcionalidades, como diferentes tipos de análises, visualizações gráficas ou integração com outras plataformas.
Interface de Usuário: Considere criar uma interface gráfica ou um dashboard para monitorar os resultados das análises de forma mais intuitiva.
Automatizar Deployments: Se desejar que o script funcione continuamente, explore opções de deployment em servidores ou serviços de nuvem.

Recursos Adicionais:

Documentação do MetaTrader5 Python: Para aprofundar-se nas funcionalidades da biblioteca MetaTrader5.
Pandas Documentation: Para aprender mais sobre manipulação e análise de dados com pandas.
Python Logging Best Practices: Para aprimorar suas habilidades de logging.
Python Dotenv Documentation: Para entender melhor como gerenciar variáveis de ambiente.
12-Factor App - Config: Para aprender mais sobre como gerenciar configurações e segredos de forma segura.

🔒 Fique Seguro e Boas Análises! 🔒

A segurança das suas credenciais é fundamental para proteger suas contas e dados. Sempre siga as melhores práticas para garantir a integridade e confidencialidade das suas informações.

Guia Completo: Construindo um Bot Seguro de Integração entre Telegram e MetaTrader5 em Python

Henrique Vital — Fri, 04 Oct 2024 01:02:29 +0000

Bem-vindo! Este guia é projetado desenvolvedores intusiastas em Python que desejam criar um bot que monitora mensagens no Telegram e interage com o MetaTrader5 (MT5) para executar ordens de negociação com base nos sinais recebidos. Vamos guiá-lo por todo o processo, desde a configuração do ambiente até a compreensão detalhada de cada parte do código.

Índice

Introdução
Pré-requisitos
Configurando o Ambiente
- 1. Criar um Diretório para o Projeto
- 2. Configurar um Ambiente Virtual (Opcional, mas Recomendado)
- 3. Instalar Dependências Necessárias
Criando o Arquivo .env
Entendendo o Script Python
- Imports e Variáveis de Ambiente
- Configuração de Logging
- Configuração das Contas MT5
- Função de Reconexão com MT5
- Função para Enviar Ordens para MT5
- Processando Sinais do Telegram
- Monitoramento de Conexões
- Manipulação de Sinais para Encerramento Suave
- Função Principal (main)
Executando o Script
Boas Práticas e Dicas
Conclusão

Introdução

Neste projeto, você criará um bot em Python que realiza as seguintes tarefas:

Monitora um Grupo no Telegram: O bot monitora mensagens em um grupo específico do Telegram em busca de sinais de negociação (por exemplo, "buy US30").
Processa Sinais: Quando um sinal é detectado, o bot interpreta para determinar qual ativo negociar e se deve comprar ou vender.
Interage com o MetaTrader5 (MT5): Com base no sinal, o bot envia ordens de negociação para sua conta MT5.
Mantém Práticas de Segurança: Informações sensíveis, como chaves de API e senhas, são gerenciadas de forma segura usando variáveis de ambiente.

Ao final deste guia, você terá um bot funcional e seguro que automatiza ações de negociação com base em mensagens do Telegram.

Pré-requisitos

Antes de começar, certifique-se de ter o seguinte:

Conhecimento Básico de Python: Familiaridade com a sintaxe do Python e conceitos como funções, loops e módulos.
Python Instalado: Verifique se você tem o Python 3.7 ou posterior instalado em seu computador. Você pode baixá-lo do site oficial do Python.
Conta no Telegram: Você precisará acessar a API do Telegram. Registre sua aplicação aqui para obter o API_ID e o API_HASH.
Conta no MetaTrader5: Acesse uma conta MT5, incluindo credenciais de login e informações do servidor.
Entendimento de Ambientes Virtuais: Embora opcional, usar ambientes virtuais é recomendado para gerenciar dependências.

Configurando o Ambiente

1. Criar um Diretório para o Projeto

Escolha um local em seu computador e crie um novo diretório para o seu projeto. Por exemplo:

mkdir telegram_mt5_bot
cd telegram_mt5_bot

2. Configurar um Ambiente Virtual (Opcional, mas Recomendado)

Um ambiente virtual isola as dependências do seu projeto das de outros projetos Python em seu sistema.

Criar um Ambiente Virtual:
```
python -m venv venv
```
Este comando cria um ambiente virtual chamado venv dentro do diretório do seu projeto.
Ativar o Ambiente Virtual:
- No Windows:
```
venv\Scripts\activate
```
- No macOS e Linux:
```
source venv/bin/activate
```
Após a ativação, o nome do ambiente virtual aparecerá no prompt do terminal, indicando que as dependências instaladas agora serão isoladas para este projeto.

3. Instalar Dependências Necessárias

Com o ambiente virtual ativado, instale as bibliotecas necessárias usando o pip:

pip install telethon MetaTrader5 python-dotenv

telethon: Biblioteca para interagir com a API do Telegram.
MetaTrader5: Biblioteca para interagir com o MetaTrader5.
python-dotenv: Biblioteca para carregar variáveis de ambiente a partir de um arquivo .env.

Além disso, para facilitar o gerenciamento de pacotes, é recomendável criar um arquivo requirements.txt:

pip freeze > requirements.txt

Este arquivo registra todas as dependências do projeto, permitindo que outras pessoas (ou você mesmo em outra máquina) instalem facilmente as mesmas versões com:

pip install -r requirements.txt

Criando o Arquivo `.env`

Para gerenciar informações sensíveis de forma segura, usaremos um arquivo .env que armazenará variáveis de ambiente. Este arquivo não deve ser versionado ou compartilhado.

1. Criar o Arquivo `.env`

No diretório raiz do seu projeto (telegram_mt5_bot), crie um arquivo chamado .env e adicione as seguintes linhas, substituindo os valores pelos seus próprios:

# Credenciais do Telegram
TELEGRAM_API_ID=seu_api_id
TELEGRAM_API_HASH=seu_api_hash
TELEGRAM_PHONE_NUMBER=seu_numero_de_telefone
TELEGRAM_GROUP_USERNAME=@superus30

# Credenciais do MetaTrader5 (MT5)
MT5_LOGIN=seu_login_mt5
MT5_PASSWORD=sua_senha_mt5
MT5_SERVER=seu_servidor_mt5
MT5_PATH=C:\Caminho\Para\Seu\MetaTrader5\terminal64.exe

# Configuração de Logging
LOG_FILE=app.log

Notas Importantes:

TELEGRAM_API_ID e TELEGRAM_API_HASH: Obtidos ao registrar sua aplicação no Telegram.
TELEGRAM_PHONE_NUMBER: Número de telefone associado à sua conta do Telegram.
TELEGRAM_GROUP_USERNAME: Nome de usuário do grupo Telegram que o bot irá monitorar.
MT5_LOGIN, MT5_PASSWORD, MT5_SERVER: Informações da sua conta MT5.
MT5_PATH: Caminho para o executável do MetaTrader5 no seu sistema.

2. Atualizar o Arquivo `.gitignore`

Para garantir que o arquivo .env não seja versionado (especialmente se você estiver usando o Git), crie ou atualize o arquivo .gitignore no diretório raiz com o seguinte conteúdo:

# Arquivo de variáveis de ambiente
.env

# Diretório do ambiente virtual
venv/

Isso impede que informações sensíveis sejam acidentalmente compartilhadas em repositórios públicos ou privados.

Entendendo o Script Python

Agora, vamos criar o script Python que integrará o Telegram com o MetaTrader5. Este script realizará as seguintes funções:

Carregar Variáveis de Ambiente: Utiliza o python-dotenv para carregar credenciais e configurações do arquivo .env.
Configurar Logging: Define como e onde os logs serão armazenados, facilitando o monitoramento e a depuração.
Configurar Contas MT5: Define as contas MT5 que serão utilizadas pelo bot.
Funções de Reconexão e Envio de Ordens: Gerencia a conexão com o MT5 e envia ordens de negociação com base nos sinais recebidos.
Processar Sinais do Telegram: Interpreta mensagens do Telegram para determinar as ações de negociação.
Monitorar Conexões: Verifica periodicamente se as conexões com o MT5 estão ativas e tenta reconectar se necessário.
Manipular Sinais de Encerramento: Garante que o bot encerre suas operações de forma suave ao receber sinais de interrupção.
Função Principal (main): Orquestra todas as operações acima, iniciando o cliente Telegram e gerenciando as tarefas assíncronas.

Vamos explorar cada parte do código com detalhes.

1. Imports e Variáveis de Ambiente

Primeiro, importamos todas as bibliotecas necessárias e carregamos as variáveis de ambiente.

import os
import sys
import asyncio
import logging
import signal
import pkg_resources
from datetime import datetime
from dotenv import load_dotenv
from telethon import TelegramClient, events
import MetaTrader5 as mt5

Explicação dos Imports:

os: Interage com o sistema operacional, acessando variáveis de ambiente e caminhos de arquivos.
sys: Fornece acesso a variáveis e funções que interagem com o interpretador Python.
asyncio: Facilita a escrita de código assíncrono, permitindo que múltiplas tarefas sejam executadas concorrentemente.
logging: Gerencia o registro de mensagens de log para monitoramento e depuração.
signal: Permite a captura de sinais do sistema (como interrupções) para realizar ações específicas.
pkg_resources: Gerencia recursos de pacotes Python, aqui utilizado para listar pacotes instalados.
datetime: Trabalha com datas e horas.
dotenv: Carrega variáveis de ambiente de um arquivo .env.
Telethon: Biblioteca para interagir com a API do Telegram de forma assíncrona.
MetaTrader5: Biblioteca para interagir com o MetaTrader5.

Carregando Variáveis de Ambiente:

# Carrega variáveis de ambiente do arquivo .env
load_dotenv()

Esta linha lê o arquivo .env e carrega as variáveis de ambiente nele definidas, permitindo que sejam acessadas via os.getenv().

2. Configuração de Logging

O logging é essencial para monitorar o funcionamento do seu bot e depurar problemas. Vamos configurar o logging para registrar mensagens tanto no console quanto em um arquivo.

# Acessa as variáveis de ambiente para configuração
LOG_FILE = os.getenv('LOG_FILE', 'app.log')

# Configura o logging
logging.basicConfig(
    level=logging.INFO,  # Nível mínimo de severidade para registrar
    format='%(asctime)s - %(levelname)s - %(message)s',  # Formato das mensagens de log
    handlers=[
        logging.FileHandler(LOG_FILE),  # Registra logs em um arquivo
        logging.StreamHandler(sys.stdout)  # Exibe logs no console
    ]
)
logger = logging.getLogger(__name__)

Detalhes da Configuração:

level: Define o nível mínimo de mensagens que serão registradas. INFO inclui INFO, WARNING, ERROR e CRITICAL.
format: Define como cada mensagem de log será formatada, incluindo o timestamp, o nível de severidade e a mensagem.
handlers: Define onde as mensagens de log serão enviadas. Aqui, estão sendo enviadas para um arquivo (app.log) e para o console.

3. Configuração das Contas MT5

Definimos as contas MT5 que o bot utilizará para enviar ordens de negociação.

# Configurações das contas MT5
CONTAS_MT5 = [
    {
        "login": os.getenv('MT5_LOGIN'),
        "senha": os.getenv('MT5_PASSWORD'),
        "servidor": os.getenv('MT5_SERVER'),
        "us30": "US30.cash",
        "nas100": "US100.cash",
        "lote": 0.01
    }
]

contas_ativas = []  # Lista para armazenar contas ativas
shutdown_event = asyncio.Event()  # Evento para sinalizar o encerramento do programa

Explicação dos Campos:

login: Número de login da conta MT5.
senha: Senha da conta MT5.
servidor: Nome do servidor MT5.
us30 e nas100: Símbolos dos ativos que serão negociados.
lote: Volume padrão das ordens a serem enviadas.

4. Função de Reconexão com MT5

Esta função tenta reconectar à conta MT5 caso a conexão seja perdida.

async def reconectar_mt5(conta, max_tentativas=3):
    """
    Tenta reconectar à conta MT5 com um número máximo de tentativas.

    Args:
        conta (dict): Informações da conta MT5.
        max_tentativas (int): Número máximo de tentativas de reconexão.

    Returns:
        bool: True se a reconexão for bem-sucedida, False caso contrário.
    """
    for tentativa in range(max_tentativas):
        try:
            # Tenta inicializar a conexão com MT5
            if mt5.initialize(path=os.getenv('MT5_PATH'), login=int(conta['login']), server=conta['servidor'], password=conta['senha']):
                logger.info(f"Reconexão bem-sucedida para conta {conta['login']}")
                return True
            else:
                erro = mt5.last_error()
                logger.warning(f"Tentativa {tentativa + 1} de reconexão falhou para conta {conta['login']}: {erro}")
        except Exception as e:
            logger.error(f"Erro durante a tentativa {tentativa + 1} de reconexão para conta {conta['login']}: {e}")
        await asyncio.sleep(5)  # Espera 5 segundos antes de tentar novamente
    logger.error(f"Falha ao reconectar à conta {conta['login']} após {max_tentativas} tentativas")
    return False

Detalhes da Função:

Objetivo: Garantir que o bot mantenha a conexão com o MT5, tentando reconectar em caso de falha.
Parâmetros:
- conta: Dicionário contendo as informações da conta MT5.
- max_tentativas: Número máximo de tentativas de reconexão antes de desistir.
Processo:
- Tenta inicializar a conexão com MT5 usando as credenciais fornecidas.
- Se falhar, registra um aviso e aguarda 5 segundos antes de tentar novamente.
- Após exceder o número máximo de tentativas, registra um erro e retorna False.

5. Função para Enviar Ordens para MT5

Esta função envia ordens de negociação para o MT5 com base nos sinais processados.

async def enviar_ordem(conta, simbolo, acao, lote):
    """
    Envia uma ordem de negociação para o MT5.

    Args:
        conta (dict): Informações da conta MT5.
        simbolo (str): Símbolo do ativo a ser negociado.
        acao (int): Tipo de ação (compra ou venda).
        lote (float): Volume da ordem.

    Returns:
        bool: True se a ordem for enviada com sucesso, False caso contrário.
    """
    # Tenta reconectar à conta MT5 antes de enviar a ordem
    if not await reconectar_mt5(conta):
        logger.error(f"Não foi possível enviar ordem para {conta['login']} devido a falha na reconexão")
        return False

    # Obtém informações do símbolo
    symbol_info = mt5.symbol_info(simbolo)
    if symbol_info is None:
        logger.error(f"Símbolo {simbolo} não encontrado")
        return False

    # Verifica se o símbolo está disponível para trading
    if not symbol_info.visible:
        logger.warning(f"Símbolo {simbolo} não está visível, tentando habilitá-lo")
        if not mt5.symbol_select(simbolo, True):
            logger.error(f"Falha ao selecionar o símbolo {simbolo}")
            return False

    # Obtém o tick atual do símbolo
    tick = mt5.symbol_info_tick(simbolo)
    if tick is None:
        logger.error(f"Não foi possível obter o tick para o símbolo {simbolo}")
        return False

    # Define o preço baseado na ação (compra ou venda)
    price = tick.ask if acao == mt5.ORDER_TYPE_BUY else tick.bid

    # Prepara a estrutura do pedido
    pedido = {
        "action": mt5.TRADE_ACTION_DEAL,  # Tipo de ação de negociação
        "symbol": simbolo,                # Símbolo do ativo
        "volume": float(lote),            # Volume da ordem
        "type": acao,                     # Tipo de ordem (compra ou venda)
        "price": price,                   # Preço da ordem
        "deviation": 20,                  # Desvio permitido no preço
        "magic": 234000,                   # Identificador único para a ordem
        "comment": "python script order",  # Comentário para a ordem
        "type_time": mt5.ORDER_TIME_GTC,   # Tipo de tempo da ordem (Good Till Cancelled)
        "type_filling": mt5.ORDER_FILLING_IOC,  # Tipo de preenchimento (Immediate or Cancel)
    }

    # Envia a ordem para o MT5
    resultado = mt5.order_send(pedido)
    if resultado.retcode != mt5.TRADE_RETCODE_DONE:
        logger.error(f"Erro ao enviar ordem para {conta['login']}: {resultado.comment}")
        logger.debug(f"Detalhes do pedido: {pedido}")
        logger.debug(f"Código de retorno: {resultado.retcode}")
        return False
    else:
        logger.info(f"Ordem enviada com sucesso para {conta['login']} com lote {lote}")
        logger.debug(f"Detalhes da ordem: {resultado}")
        return True

Detalhes da Função:

Objetivo: Enviar uma ordem de compra ou venda para o MT5 com base nos sinais recebidos.
Parâmetros:
- conta: Informações da conta MT5.
- simbolo: Símbolo do ativo a ser negociado (por exemplo, "US30.cash").
- acao: Tipo de ação (mt5.ORDER_TYPE_BUY para compra ou mt5.ORDER_TYPE_SELL para venda).
- lote: Volume da ordem (por exemplo, 0.01).
Processo:
- Reconecta à conta MT5 se necessário.
- Verifica se o símbolo está disponível e visível para negociação.
- Obtém o preço atual do símbolo (ask para compra, bid para venda).
- Prepara a estrutura do pedido com todos os parâmetros necessários.
- Envia a ordem para o MT5 e verifica se foi bem-sucedida.

6. Processando Sinais do Telegram

Esta função interpreta mensagens recebidas no Telegram para determinar quais ações de negociação devem ser executadas.

async def processar_sinal(mensagem):
    """
    Processa uma mensagem recebida do Telegram para determinar a ação de negociação.

    Args:
        mensagem (str): Texto da mensagem recebida.

    """
    logger.info(f"Mensagem recebida do Telegram: {mensagem}")

    palavras = mensagem.lower().split()

    ativo = None
    acao = None

    # Verifica o ativo na mensagem
    if 'us30' in palavras:
        ativo = 'us30'
    elif 'nas100' in palavras:
        ativo = 'nas100'

    # Verifica a ação na mensagem
    if 'buy' in palavras:
        acao = mt5.ORDER_TYPE_BUY
    elif 'sell' in palavras:
        acao = mt5.ORDER_TYPE_SELL

    # Se nenhum ativo for reconhecido, encerra a função
    if not ativo:
        logger.info("Nenhum ativo reconhecido na mensagem.")
        return

    # Se nenhuma ação for reconhecida, encerra a função
    if acao is None:
        logger.info("Nenhuma ação (compra/venda) reconhecida na mensagem.")
        return

    acao_str = "COMPRA" se acao == mt5.ORDER_TYPE_BUY else "VENDA"
    logger.info(f"Interpretação: Ativo: {ativo.upper()}, Ação: {acao_str}")

    # Itera sobre todas as contas ativas para enviar a ordem
    for conta in contas_ativas[:]:  # Cria uma cópia da lista para iteração segura
        try:
            simbolo = conta.get(ativo)
            if not simbolo:
                logger.warning(f"Ativo {ativo} não configurado para a conta {conta['login']}. Pulando.")
                continue
            sucesso = await enviar_ordem(conta, simbolo, acao, conta['lote'])
            if not sucesso:
                logger.warning(f"Falha ao processar sinal para conta {conta['login']}. Removendo da lista de contas ativas.")
                contas_ativas.remove(conta)
        except Exception as e:
            logger.error(f"Erro ao processar sinal para conta {conta['login']}: {e}")
            logger.warning(f"Removendo conta {conta['login']} da lista de contas ativas devido a erro")
            contas_ativas.remove(conta)

Detalhes da Função:

Objetivo: Interpretar a mensagem recebida do Telegram para determinar qual ativo negociar e se deve comprar ou vender.
Parâmetros:
- mensagem: Texto da mensagem recebida no Telegram.
Processo:
- Converte a mensagem para letras minúsculas e divide em palavras.
- Identifica o ativo mencionado (us30 ou nas100).
- Identifica a ação (buy para compra ou sell para venda).
- Itera sobre todas as contas ativas e envia a ordem correspondente.
- Se a ordem falhar, remove a conta da lista de contas ativas para evitar tentativas futuras.

7. Monitoramento de Conexões

Esta função verifica periodicamente se as conexões com as contas MT5 estão ativas e tenta reconectar se necessário.

async def verificar_conexoes():
    """
    Verifica periodicamente se as contas MT5 estão conectadas e tenta reconectar se necessário.
    """
    while not shutdown_event.is_set():
        for conta in contas_ativas[:]:
            if shutdown_event.is_set():
                break
            if not await reconectar_mt5(conta):
                logger.warning(f"Conta {conta['login']} removida da lista de contas ativas devido a falha na conexão")
                contas_ativas.remove(conta)
        await asyncio.sleep(60)  # Verifica a cada 60 segundos

Detalhes da Função:

Objetivo: Garantir que todas as contas MT5 ativas mantenham uma conexão estável.
Processo:
- Enquanto o evento de encerramento não for sinalizado, itera sobre todas as contas ativas.
- Para cada conta, tenta reconectar.
- Se a reconexão falhar após as tentativas definidas, remove a conta da lista de contas ativas.
- Aguarda 60 segundos antes da próxima verificação.

8. Manipulação de Sinais para Encerramento Suave

Esta função lida com sinais de interrupção (como Ctrl+C) para encerrar o programa de forma limpa.

def signal_handler(signum, frame):
    """
    Manipula sinais de interrupção para encerrar o programa de forma suave.

    Args:
        signum: Número do sinal.
        frame: Frame atual.
    """
    logger.info("Sinal de interrupção recebido. Encerrando o programa...")
    asyncio.get_event_loop().call_soon_threadsafe(shutdown_event.set)

Detalhes da Função:

Objetivo: Capturar sinais de interrupção e iniciar o processo de encerramento do bot de forma ordenada.
Processo:
- Quando um sinal de interrupção é recebido (como SIGINT ou SIGTERM), registra uma mensagem de log.
- Sinaliza o evento de encerramento para que todas as tarefas assíncronas possam finalizar adequadamente.

9. Função Principal (`main`)

Esta é a função central que coordena todas as operações do bot.

async def main():
    """
    Função principal que inicializa o bot e gerencia suas operações.
    """
    # Configurar o manipulador de sinais
    signal.signal(signal.SIGINT, signal_handler)
    signal.signal(signal.SIGTERM, signal_handler)

    # Inicializar contas MT5
    for conta in CONTAS_MT5:
        if await reconectar_mt5(conta):
            contas_ativas.append(conta)

    if not contas_ativas:
        logger.error("Nenhuma conta pôde ser inicializada. Encerrando o programa.")
        return

    logger.info(f"Programa continuando com {len(contas_ativas)} conta(s) ativa(s)")

    # Inicializar o cliente Telegram
    client = TelegramClient('session', os.getenv('TELEGRAM_API_ID'), os.getenv('TELEGRAM_API_HASH'))

    @client.on(events.NewMessage(chats=os.getenv('TELEGRAM_GROUP_USERNAME')))
    async def handler(event):
        """
        Manipulador de eventos para novas mensagens no Telegram.
        """
        if not shutdown_event.is_set():
            try:
                logger.info(f"Nova mensagem recebida do grupo {os.getenv('TELEGRAM_GROUP_USERNAME')}")
                await processar_sinal(event.message.text)
            except Exception as e:
                logger.error(f"Erro ao processar mensagem do Telegram: {e}")

    # Iniciar a tarefa de verificação de conexões
    verificar_task = asyncio.create_task(verificar_conexoes())

    try:
        await client.start(phone=os.getenv('TELEGRAM_PHONE_NUMBER'))
        logger.info("Bot Telegram iniciado. Aguardando mensagens...")
        await shutdown_event.wait()  # Aguarda até que o evento de encerramento seja sinalizado
    except Exception as e:
        logger.error(f"Erro no cliente Telegram: {e}")
    finally:
        await client.disconnect()
        verificar_task.cancel()
        for conta in contas_ativas:
            if mt5.shutdown():
                logger.info(f"MT5 desligado para a conta {conta['login']}")
            else:
                logger.warning(f"Falha ao desligar MT5 para a conta {conta['login']}")
        logger.info("Programa encerrado.")

Detalhes da Função:

Configuração de Sinais:
- Associa os sinais de interrupção (SIGINT e SIGTERM) à função signal_handler para garantir um encerramento suave.
Inicialização das Contas MT5:
- Itera sobre todas as contas definidas em CONTAS_MT5.
- Tenta reconectar a cada conta e adiciona à lista de contas_ativas se bem-sucedido.
- Se nenhuma conta puder ser inicializada, registra um erro e encerra o programa.
Inicialização do Cliente Telegram:
- Cria uma instância do TelegramClient usando as credenciais do Telegram.
- Define um manipulador de eventos (handler) que será chamado sempre que uma nova mensagem for recebida no grupo especificado.
- O manipulador chama a função processar_sinal para interpretar e agir sobre a mensagem.
Iniciar Tarefas Assíncronas:
- Cria uma tarefa assíncrona para verificar periodicamente as conexões MT5 (verificar_conexoes).
- Inicia o cliente Telegram e aguarda até que o evento de encerramento seja sinalizado.
Encerramento:
- Ao receber um sinal de encerramento, desconecta o cliente Telegram.
- Cancela a tarefa de verificação de conexões.
- Desliga o MT5 para todas as contas ativas.
- Registra que o programa foi encerrado.

10. Executando o Script

Para executar o script, crie um arquivo chamado bot.py no diretório do seu projeto e cole o seguinte código completo, que integra todas as partes que discutimos:

import os
import sys
import asyncio
import logging
import signal
import pkg_resources
from datetime import datetime
from dotenv import load_dotenv
from telethon import TelegramClient, events
import MetaTrader5 as mt5

# Carrega variáveis de ambiente do arquivo .env
load_dotenv()

# Acessa as variáveis de ambiente para configuração
API_ID = os.getenv('TELEGRAM_API_ID')
API_HASH = os.getenv('TELEGRAM_API_HASH')
PHONE_NUMBER = os.getenv('TELEGRAM_PHONE_NUMBER')
GROUP_USERNAME = os.getenv('TELEGRAM_GROUP_USERNAME')

MT5_LOGIN = os.getenv('MT5_LOGIN')
MT5_PASSWORD = os.getenv('MT5_PASSWORD')
MT5_SERVER = os.getenv('MT5_SERVER')
MT5_PATH = os.getenv('MT5_PATH')

LOG_FILE = os.getenv('LOG_FILE', 'app.log')

# Configura logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler(LOG_FILE),
        logging.StreamHandler(sys.stdout)
    ]
)
logger = logging.getLogger(__name__)

# Configurações das contas MT5
CONTAS_MT5 = [
    {
        "login": MT5_LOGIN,
        "senha": MT5_PASSWORD,
        "servidor": MT5_SERVER,
        "us30": "US30.cash",
        "nas100": "US100.cash",
        "lote": 0.01
    }
]

contas_ativas = []
shutdown_event = asyncio.Event()

async def reconectar_mt5(conta, max_tentativas=3):
    """
    Tenta reconectar à conta MT5 com um número máximo de tentativas.

    Args:
        conta (dict): Informações da conta MT5.
        max_tentativas (int): Número máximo de tentativas de reconexão.

    Returns:
        bool: True se a reconexão for bem-sucedida, False caso contrário.
    """
    for tentativa in range(max_tentativas):
        try:
            # Tenta inicializar a conexão com MT5
            if mt5.initialize(path=MT5_PATH, login=int(conta['login']), server=conta['servidor'], password=conta['senha']):
                logger.info(f"Reconexão bem-sucedida para conta {conta['login']}")
                return True
            else:
                erro = mt5.last_error()
                logger.warning(f"Tentativa {tentativa + 1} de reconexão falhou para conta {conta['login']}: {erro}")
        except Exception as e:
            logger.error(f"Erro durante a tentativa {tentativa + 1} de reconexão para conta {conta['login']}: {e}")
        await asyncio.sleep(5)  # Espera 5 segundos antes de tentar novamente
    logger.error(f"Falha ao reconectar à conta {conta['login']} após {max_tentativas} tentativas")
    return False

async def enviar_ordem(conta, simbolo, acao, lote):
    """
    Envia uma ordem de negociação para o MT5.

    Args:
        conta (dict): Informações da conta MT5.
        simbolo (str): Símbolo do ativo a ser negociado.
        acao (int): Tipo de ação (compra ou venda).
        lote (float): Volume da ordem.

    Returns:
        bool: True se a ordem for enviada com sucesso, False caso contrário.
    """
    # Tenta reconectar à conta MT5 antes de enviar a ordem
    if not await reconectar_mt5(conta):
        logger.error(f"Não foi possível enviar ordem para {conta['login']} devido a falha na reconexão")
        return False

    # Obtém informações do símbolo
    symbol_info = mt5.symbol_info(simbolo)
    if symbol_info is None:
        logger.error(f"Símbolo {simbolo} não encontrado")
        return False

    # Verifica se o símbolo está disponível para trading
    if not symbol_info.visible:
        logger.warning(f"Símbolo {simbolo} não está visível, tentando habilitá-lo")
        if not mt5.symbol_select(simbolo, True):
            logger.error(f"Falha ao selecionar o símbolo {simbolo}")
            return False

    # Obtém o tick atual do símbolo
    tick = mt5.symbol_info_tick(simbolo)
    if tick is None:
        logger.error(f"Não foi possível obter o tick para o símbolo {simbolo}")
        return False

    # Define o preço baseado na ação (compra ou venda)
    price = tick.ask if acao == mt5.ORDER_TYPE_BUY else tick.bid

    # Prepara a estrutura do pedido
    pedido = {
        "action": mt5.TRADE_ACTION_DEAL,  # Tipo de ação de negociação
        "symbol": simbolo,                # Símbolo do ativo
        "volume": float(lote),            # Volume da ordem
        "type": acao,                     # Tipo de ordem (compra ou venda)
        "price": price,                   # Preço da ordem
        "deviation": 20,                  # Desvio permitido no preço
        "magic": 234000,                   # Identificador único para a ordem
        "comment": "python script order",  # Comentário para a ordem
        "type_time": mt5.ORDER_TIME_GTC,   # Tipo de tempo da ordem (Good Till Cancelled)
        "type_filling": mt5.ORDER_FILLING_IOC,  # Tipo de preenchimento (Immediate or Cancel)
    }

    # Envia a ordem para o MT5
    resultado = mt5.order_send(pedido)
    if resultado.retcode != mt5.TRADE_RETCODE_DONE:
        logger.error(f"Erro ao enviar ordem para {conta['login']}: {resultado.comment}")
        logger.debug(f"Detalhes do pedido: {pedido}")
        logger.debug(f"Código de retorno: {resultado.retcode}")
        return False
    else:
        logger.info(f"Ordem enviada com sucesso para {conta['login']} com lote {lote}")
        logger.debug(f"Detalhes da ordem: {resultado}")
        return True

async def processar_sinal(mensagem):
    """
    Processa uma mensagem recebida do Telegram para determinar a ação de negociação.

    Args:
        mensagem (str): Texto da mensagem recebida.
    """
    logger.info(f"Mensagem recebida do Telegram: {mensagem}")

    palavras = mensagem.lower().split()

    ativo = None
    acao = None

    # Verifica o ativo na mensagem
    if 'us30' in palavras:
        ativo = 'us30'
    elif 'nas100' in palavras:
        ativo = 'nas100'

    # Verifica a ação na mensagem
    if 'buy' in palavras:
        acao = mt5.ORDER_TYPE_BUY
    elif 'sell' in palavras:
        acao = mt5.ORDER_TYPE_SELL

    # Se nenhum ativo for reconhecido, encerra a função
    if not ativo:
        logger.info("Nenhum ativo reconhecido na mensagem.")
        return

    # Se nenhuma ação for reconhecida, encerra a função
    if acao is None:
        logger.info("Nenhuma ação (compra/venda) reconhecida na mensagem.")
        return

    acao_str = "COMPRA" se acao == mt5.ORDER_TYPE_BUY else "VENDA"
    logger.info(f"Interpretação: Ativo: {ativo.upper()}, Ação: {acao_str}")

    # Itera sobre todas as contas ativas para enviar a ordem
    for conta in contas_ativas[:]:  # Cria uma cópia da lista para iteração segura
        try:
            simbolo = conta.get(ativo)
            if not simbolo:
                logger.warning(f"Ativo {ativo} não configurado para a conta {conta['login']}. Pulando.")
                continue
            sucesso = await enviar_ordem(conta, simbolo, acao, conta['lote'])
            if not sucesso:
                logger.warning(f"Falha ao processar sinal para conta {conta['login']}. Removendo da lista de contas ativas.")
                contas_ativas.remove(conta)
        except Exception as e:
            logger.error(f"Erro ao processar sinal para conta {conta['login']}: {e}")
            logger.warning(f"Removendo conta {conta['login']} da lista de contas ativas devido a erro")
            contas_ativas.remove(conta)

async def verificar_conexoes():
    """
    Verifica periodicamente se as contas MT5 estão conectadas e tenta reconectar se necessário.
    """
    while not shutdown_event.is_set():
        for conta in contas_ativas[:]:
            if shutdown_event.is_set():
                break
            if not await reconectar_mt5(conta):
                logger.warning(f"Conta {conta['login']} removida da lista de contas ativas devido a falha na conexão")
                contas_ativas.remove(conta)
        await asyncio.sleep(60)  # Verifica a cada 60 segundos

def signal_handler(signum, frame):
    """
    Manipula sinais de interrupção para encerrar o programa de forma suave.

    Args:
        signum: Número do sinal.
        frame: Frame atual.
    """
    logger.info("Sinal de interrupção recebido. Encerrando o programa...")
    asyncio.get_event_loop().call_soon_threadsafe(shutdown_event.set)

async def main():
    """
    Função principal que inicializa o bot e gerencia suas operações.
    """
    # Configurar o manipulador de sinais
    signal.signal(signal.SIGINT, signal_handler)
    signal.signal(signal.SIGTERM, signal_handler)

    # Inicializar contas MT5
    for conta in CONTAS_MT5:
        if await reconectar_mt5(conta):
            contas_ativas.append(conta)

    if not contas_ativas:
        logger.error("Nenhuma conta pôde ser inicializada. Encerrando o programa.")
        return

    logger.info(f"Programa continuando com {len(contas_ativas)} conta(s) ativa(s)")

    # Inicializar o cliente Telegram
    client = TelegramClient('session', API_ID, API_HASH)

    @client.on(events.NewMessage(chats=GROUP_USERNAME))
    async def handler(event):
        """
        Manipulador de eventos para novas mensagens no Telegram.
        """
        if not shutdown_event.is_set():
            try:
                logger.info(f"Nova mensagem recebida do grupo {GROUP_USERNAME}")
                await processar_sinal(event.message.text)
            except Exception as e:
                logger.error(f"Erro ao processar mensagem do Telegram: {e}")

    # Iniciar a tarefa de verificação de conexões
    verificar_task = asyncio.create_task(verificar_conexoes())

    try:
        await client.start(phone=PHONE_NUMBER)
        logger.info("Bot Telegram iniciado. Aguardando mensagens...")
        await shutdown_event.wait()  # Aguarda até que o evento de encerramento seja sinalizado
    except Exception as e:
        logger.error(f"Erro no cliente Telegram: {e}")
    finally:
        await client.disconnect()
        verificar_task.cancel()
        for conta in contas_ativas:
            if mt5.shutdown():
                logger.info(f"MT5 desligado para a conta {conta['login']}")
            else:
                logger.warning(f"Falha ao desligar MT5 para a conta {conta['login']}")
        logger.info("Programa encerrado.")

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except Exception as e:
        logger.critical(f"Erro crítico no programa: {e}")
        sys.exit(1)

Explicação Completa do Código:

Carregamento das Variáveis de Ambiente:
- Usa load_dotenv() para carregar as variáveis definidas no arquivo .env.
- Acessa essas variáveis usando os.getenv() para configurar o Telegram e MT5.
Configuração de Logging:
- Configura o logging para registrar mensagens tanto em um arquivo (app.log) quanto no console.
- Define o nível mínimo de registro como INFO.
Configuração das Contas MT5:
- Define uma lista CONTAS_MT5 contendo dicionários com as informações necessárias para cada conta MT5.
- Inicializa uma lista vazia contas_ativas para armazenar contas que estão conectadas.
Função de Reconexão (reconectar_mt5):
- Tenta reconectar à conta MT5 até um número máximo de tentativas.
- Registra mensagens de sucesso ou falha no log.
Função para Enviar Ordens (enviar_ordem):
- Garante que a conexão com MT5 está ativa.
- Verifica se o símbolo está disponível e visível.
- Obtém o preço atual do símbolo (ask para compra, bid para venda).
- Prepara a ordem com os parâmetros necessários e a envia para o MT5.
- Registra o resultado da ordem no log.
Função para Processar Sinais (processar_sinal):
- Interpreta a mensagem recebida no Telegram para determinar qual ativo negociar e qual ação (compra ou venda) executar.
- Itera sobre todas as contas ativas e envia a ordem correspondente.
- Se uma ordem falhar, remove a conta da lista de contas ativas para evitar tentativas futuras.
Função de Monitoramento de Conexões (verificar_conexoes):
- Executa em um loop contínuo, verificando a conexão de cada conta MT5 ativa a cada 60 segundos.
- Tenta reconectar se a conexão estiver perdida e remove contas que não puderam ser reconectadas.
Manipulador de Sinais (signal_handler):
- Captura sinais de interrupção (como Ctrl+C) e sinaliza o encerramento do programa.
- Garante que todas as tarefas assíncronas sejam finalizadas corretamente.
Função Principal (main):
- Configura os manipuladores de sinais.
- Inicializa as contas MT5 e adiciona as contas bem-sucedidas à lista de contas ativas.
- Inicializa o cliente Telegram e define um manipulador de eventos para novas mensagens.
- Inicia a tarefa assíncrona de monitoramento de conexões.
- Aguarda até que um sinal de encerramento seja recebido.
- Ao encerrar, desconecta o cliente Telegram, cancela a tarefa de monitoramento e desliga o MT5 para todas as contas ativas.
Execução do Script:
- Verifica se o script está sendo executado diretamente.
- Executa a função principal usando asyncio.run().
- Captura quaisquer exceções críticas, registra no log e encerra o programa com um código de erro.

Executando o Script

Após configurar o ambiente e criar o arquivo bot.py, siga os passos abaixo para executar o bot.

1. Ativar o Ambiente Virtual

Se você configurou um ambiente virtual, ative-o:

No Windows:
```
venv\Scripts\activate
```
No macOS e Linux:
```
source venv/bin/activate
```

2. Executar o Script

No diretório do seu projeto, execute o seguinte comando:

python bot.py

O que esperar:

Logs no Console: Você verá mensagens informando sobre o status da conexão com o MT5 e do cliente Telegram.
Arquivo de Log: Todas as mensagens de log também serão gravadas no arquivo app.log (ou outro definido em LOG_FILE).

Interagindo com o Bot:

Envie mensagens no grupo do Telegram especificado (@superus30) com comandos como "buy US30" ou "sell NAS100".
O bot interpretará esses sinais e enviará ordens de negociação correspondentes para a conta MT5 configurada.

3. Encerrando o Bot

Para encerrar o bot de forma segura, pressione Ctrl+C no terminal onde o script está sendo executado. Isso acionará o manipulador de sinais, que garantirá que todas as conexões sejam fechadas corretamente antes de encerrar o programa.

Boas Práticas e Dicas

Manutenção da Segurança:
- Nunca compartilhe seu arquivo .env ou exponha suas credenciais em repositórios públicos.
- Use Ambientes Virtuais para isolar as dependências do projeto.
Monitoramento e Logs:
- Revise regularmente os logs (app.log) para identificar e resolver possíveis problemas.
- Implementar Rotação de Logs: Para evitar que o arquivo de log cresça indefinidamente, considere implementar rotação usando bibliotecas como logging.handlers.
Testes em Ambiente de Demonstração:
- Antes de operar em contas reais, teste seu bot em ambientes de demonstração para garantir que tudo funcione conforme o esperado.
- Verifique as Ordens: Assegure-se de que as ordens estão sendo enviadas corretamente e que o volume está adequado.
Gerenciamento de Erros:
- Tratamento Robusto de Exceções: Certifique-se de que todas as possíveis exceções sejam tratadas para evitar que o bot pare inesperadamente.
- Alertas: Considere implementar alertas (por exemplo, enviando mensagens de log críticas para seu email) para notificá-lo sobre falhas graves.
Atualizações de Dependências:
- Mantenha suas bibliotecas atualizadas para aproveitar melhorias de segurança e correções de bugs.
- Verifique as Documentações: Consulte as documentações oficiais das bibliotecas para entender mudanças ou atualizações.
Escalabilidade:
- Gerencie Múltiplas Contas: Se você pretende usar múltiplas contas MT5, expanda a configuração de CONTAS_MT5 conforme necessário.
- Adicionar Mais Ativos: Atualize o script para suportar mais ativos além de "US30" e "NAS100" conforme sua estratégia de negociação.
Documentação e Comentários:
- Comente Seu Código: Mantenha comentários atualizados e relevantes para facilitar futuras manutenções.
- Documente Configurações: Mantenha um documento com detalhes sobre como configurar e executar o bot.

Conclusão

Parabéns! Você criou um bot em Python que integra o Telegram com o MetaTrader5, capaz de automatizar ordens de negociação com base em sinais recebidos no Telegram. Ao seguir este guia, você não apenas implementou a funcionalidade desejada, mas também adotou práticas de segurança e organização de código que são fundamentais para projetos de software robustos.

Próximos Passos:

Expandir Funcionalidades: Adicione mais funcionalidades, como diferentes tipos de ordens, gerenciamento de riscos ou integração com outras plataformas.
Interface de Usuário: Considere criar uma interface gráfica ou um dashboard para monitorar o desempenho do bot.
Automatizar Deployments: Se desejar que o bot funcione continuamente, explore opções de deployment em servidores ou serviços de nuvem.

Recursos Adicionais:

Documentação do Telethon: Para aprofundar-se nas funcionalidades da biblioteca Telethon.
Documentação do MetaTrader5 Python: Para entender todas as possibilidades de interação com o MT5.
Python Logging Best Practices: Para aprimorar suas habilidades de logging.
Python Asyncio Documentation: Para dominar a programação assíncrona em Python.
Gerenciamento de Segredos em Python: Para aprender mais sobre como gerenciar configurações e segredos de forma segura.

🔒 Fique Seguro e Boas Operações! 🔒

A segurança das suas credenciais é fundamental para proteger suas contas e dados. Sempre siga as melhores práticas para garantir a integridade e confidencialidade das suas informações.

Automated Trading with MetaTrader5: Order Management and Market Data Collection

Henrique Vital — Thu, 03 Oct 2024 16:14:38 +0000

Your AsimovMT class provides a comprehensive interface for interacting with MetaTrader5 (MT5) using Python. However, there are several areas in your code that could benefit from improvements, corrections, and enhancements to ensure robustness and maintainability. Below is a detailed review with suggestions:

1. Initialization and Error Handling

Issue:

In the __init__ method, if mt5.initialize() fails, you print an error message and call mt5.shutdown(), but the program continues to execute. This can lead to unexpected behavior since subsequent operations depend on a successful initialization.

Recommendation:

After shutting down MT5, you should raise an exception or exit the program to prevent further execution.

Code Correction:

def __init__(self):
    if not mt5.initialize():
        print("initialize() failed, error code =", mt5.last_error())
        mt5.shutdown()
        raise ConnectionError("Failed to initialize MetaTrader5.")

    # Rest of your initialization code...

2. Typographical Errors

Issue:

There is a consistent typo in the attribute name self.positons. It should be self.positions.

Impact:

This typo will lead to AttributeError when accessing or modifying self.positions elsewhere in the class.

Code Correction:

# Corrected attribute name
self.positions = [i._asdict() for i in mt5.positions_get()]

Ensure that all instances of self.positons are corrected to self.positions throughout the class, including methods like check_positions_and_orders.

3. Use of `self` in the Main Block

Issue:

In the __main__ block, you use self to refer to the instance of AsimovMT. Typically, self is reserved for instance methods within a class.

Recommendation:

Use a different variable name (e.g., asimov_mt) to avoid confusion.

Code Correction:

if __name__ == "__main__":
    # Instance
    asimov_mt = AsimovMT()

    # Testing methods using asimov_mt instead of self
    start_date = datetime.today() - timedelta(days=1)
    df_data = asimov_mt.get_ohlc_range('PETR4', 'M1', start_date, datetime.today())
    df_data = asimov_mt.get_ohlc_pos('PETR4', 'M1', 0, 1000)

    # ... and so on

4. Market Order Price Adjustment

Issue:

In the send_market_order method, you adjust the price by adding or subtracting 5 * ticks for buy and sell orders, respectively. Market orders are typically executed at the current market price without such adjustments.

Recommendation:

Use the current ask price for buy orders and bid price for sell orders without manual adjustments. If you intend to place a limit order, ensure that it aligns with your strategy.

Code Correction:

def send_market_order(self, symbol, side, volume, deviation=20, magic=1000, comment='test'):
    mt5.symbol_select(symbol, True)
    symbol_info = mt5.symbol_info(symbol)

    if side.lower() == 'buy':
        price = symbol_info.ask
        order_type = mt5.ORDER_TYPE_BUY
    elif side.lower() == 'sell':
        price = symbol_info.bid
        order_type = mt5.ORDER_TYPE_SELL
    else:
        raise ValueError("Invalid side. Use 'buy' or 'sell'.")

    if symbol_info.visible:
        request = {
            "action": mt5.TRADE_ACTION_DEAL,
            "symbol": symbol,
            "volume": volume,
            "type": order_type,
            "price": price,
            "deviation": deviation,
            "magic": magic,
            "comment": comment,
            "type_time": mt5.ORDER_TIME_GTC,
            "type_filling": mt5.ORDER_FILLING_RETURN,
        }
        result = mt5.order_send(request)
        if result.retcode != mt5.TRADE_RETCODE_DONE:
            print(f"Order failed, retcode={result.retcode}")
        return result
    else:
        print(f"{symbol} not visible. Unable to send market order.")

5. Consistent Error Handling and Logging

Issue:

The current implementation uses print statements for error messages and notifications. This approach can be limiting for larger applications where logging levels and outputs need to be managed more granularly.

Recommendation:

Use Python’s built-in logging module to provide flexible and configurable logging.

Code Enhancement:

import logging

# Configure logging at the beginning of your script
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

# Replace print statements with logging
logging.info("MetaTrader5 loaded. Ready to start.")
logging.error(f"initialize() failed, error code = {mt5.last_error()}")

6. Validating Input Parameters

Issue:

Methods like send_limit_order and update_limit_order assume that input parameters are valid without performing checks.

Recommendation:

Add validation for input parameters to ensure they meet expected formats and constraints.

Code Example:

def send_limit_order(self, symbol, side, price, volume, magic=1000, comment=""):
    if side.lower() not in ['buy', 'sell']:
        raise ValueError("Invalid side. Use 'buy' or 'sell'.")
    if price <= 0:
        raise ValueError("Price must be positive.")
    if volume <= 0:
        raise ValueError("Volume must be positive.")

    # Rest of your method...

7. Enhancing the `tf_dict`

Issue:

Your tf_dict maps timeframe strings to their MT5 constants and their equivalent in seconds. The comment suggests uncertainty about how timeframes represent days.

Recommendation:

Consider calculating the number of bars or days based on the timeframe and your specific use case. Additionally, you might want to include more descriptive information or helper methods to handle timeframe conversions.

Code Enhancement:

from enum import Enum

class TimeFrame(Enum):
    M1 = mt5.TIMEFRAME_M1
    M2 = mt5.TIMEFRAME_M2
    # ... other timeframes

self.tf_dict = {
    'M1': (TimeFrame.M1.value, 60),
    'M2': (TimeFrame.M2.value, 120),
    # ... other mappings
}

This approach makes the timeframes more manageable and less error-prone.

8. Handling `copy_rates_range` and `copy_rates_from_pos` Responses

Issue:

In methods get_ohlc_range and get_ohlc_pos, if data_raw is empty, the method returns None implicitly. This can lead to unexpected NoneType errors downstream.

Recommendation:

Explicitly handle empty responses by returning an empty DataFrame or raising an exception, depending on your use case.

Code Correction:

def get_ohlc_range(self, symbol, timeframe, start_date, end_date=datetime.now()):
    tf = self.tf_dict.get(timeframe, [None])[0]
    if tf is None:
        raise ValueError(f"Invalid timeframe: {timeframe}")

    data_raw = mt5.copy_rates_range(symbol, tf, start_date, end_date) 
    if data_raw is None:
        logging.warning("No data retrieved for the given range.")
        return pd.DataFrame()

    df_data = self._format_ohlc(data_raw)
    return df_data if df_data is not None else pd.DataFrame()

9. Improving the `update_limit_order` Method

Issue:

The update_limit_order method updates the order's price and volume but doesn't validate whether the order exists or whether the modification was successful.

Recommendation:

Check if the order exists before attempting to modify it and handle the response from order_send appropriately.

Code Enhancement:

def update_limit_order(self, order_id, price, volume):
    order = mt5.order_get(ticket=order_id)
    if order is None:
        logging.error(f"Order ID {order_id} not found.")
        return None

    request = {
        "action": mt5.TRADE_ACTION_MODIFY,
        "order": order_id,
        "price": price,
        "volume": volume,
    }
    result = mt5.order_send(request)
    if result.retcode != mt5.TRADE_RETCODE_DONE:
        logging.error(f"Failed to update order {order_id}, retcode={result.retcode}")
    return result._asdict()

10. Ensuring Thread Safety

Issue:

If you plan to use AsimovMT in a multi-threaded environment, concurrent access to shared resources like self.positions and self.orders could lead to race conditions.

Recommendation:

Implement thread synchronization mechanisms (e.g., threading locks) to protect shared resources.

Code Example:

import threading

class AsimovMT:
    def __init__(self):
        # Existing initialization code...
        self.lock = threading.Lock()

    def check_positions_and_orders(self):
        with self.lock:
            new_positions = [i._asdict() for i in mt5.positions_get()]
            new_orders = [i._asdict() for i in mt5.orders_get()]

            check = (self.positions != new_positions) or (self.orders != new_orders)
            self.positions = new_positions
            self.orders = new_orders
            return check

11. Finalizing MT5 Connection

Issue:

There’s no mechanism to properly shut down the MT5 connection when the program exits, which can lead to resource leaks.

Recommendation:

Implement a destructor method (__del__) or use context managers to ensure mt5.shutdown() is called appropriately.

Code Example Using Destructor:

def __del__(self):
    mt5.shutdown()
    logging.info("MetaTrader5 connection closed.")

Or Using Context Manager:

from contextlib import contextmanager

@contextmanager
def asimov_mt_context():
    asimov_mt = AsimovMT()
    try:
        yield asimov_mt
    finally:
        del asimov_mt

# Usage
if __name__ == "__main__":
    with asimov_mt_context() as asimov_mt:
        # Your testing code here

12. Enhancing Documentation and Readability

Issue:

The current code lacks docstrings and comments in English, which can hinder understanding and maintenance, especially for collaborators who may not speak Portuguese.

Recommendation:

Add docstrings to classes and methods, and ensure comments are in a consistent language (preferably English for broader accessibility).

Code Example:

class AsimovMT:
    """
    AsimovMT is a class that interfaces with MetaTrader5 to manage trading operations,
    retrieve market data, and monitor positions and orders.
    """

    def __init__(self):
        """
        Initializes the MetaTrader5 connection and retrieves current positions, orders,
        and historical data.
        """
        # Initialization code...

13. Sample Revised Main Block

Here’s how your __main__ block could look after applying the recommendations:

if __name__ == "__main__":
    try:
        asimov_mt = AsimovMT()

        # Testing market data methods
        start_date = datetime.today() - timedelta(days=1)
        df_range = asimov_mt.get_ohlc_range('PETR4', 'M1', start_date, datetime.today())
        print(df_range.head())

        df_pos = asimov_mt.get_ohlc_pos('PETR4', 'M1', 0, 1000)
        print(df_pos.head())

        # Testing balance methods
        positions_changed = asimov_mt.check_positions_and_orders()
        orders_changed = asimov_mt.check_positions_and_orders()
        print("Positions Changed:", positions_changed)
        print("Orders Changed:", orders_changed)

        h_positions_changed = asimov_mt.check_h_positions_and_orders(0)
        print("Historical Positions Changed:", h_positions_changed)
        print("Historical Deals:", asimov_mt.h_deals)
        print("Historical Orders:", asimov_mt.h_orders)

        # Testing order methods
        symbol = 'ITSA4'
        side = 'buy'
        volume = 100.0  # Should be float
        market_buy = asimov_mt.send_market_order(symbol, side, volume, comment='test')
        print("Market Buy Order:", market_buy)

        market_sell = asimov_mt.send_market_order(symbol, 'sell', volume, comment='test')
        print("Market Sell Order:", market_sell)

        limit_buy_price = 7.80
        limit_buy = asimov_mt.send_limit_order(symbol, 'buy', limit_buy_price, volume, comment='test')
        print("Limit Buy Order:", limit_buy)

        limit_sell_price = 8.00
        limit_sell = asimov_mt.send_limit_order(symbol, 'sell', limit_sell_price, volume, comment='test')
        print("Limit Sell Order:", limit_sell)

        # Update and cancel limit sell order
        if limit_sell:
            updated_order = asimov_mt.update_limit_order(limit_sell["order"], 8.01, volume=100.0)
            print("Updated Order:", updated_order)

            cancelled_order = asimov_mt.cancel_limit_order(limit_sell["order"])
            print("Cancelled Order:", cancelled_order)

    except Exception as e:
        logging.exception("An error occurred during execution.")

Conclusion

By addressing the issues and implementing the recommendations above, your AsimovMT class will become more robust, maintainable, and reliable. Proper error handling, input validation, and consistent naming conventions are crucial for developing scalable and professional-grade trading applications. Additionally, enhancing documentation and logging will aid in debugging and future development efforts.

Feel free to reach out if you have specific questions or need further assistance with particular aspects of your code!

DEV Community: Henrique Vital

Análise de Rompimentos no Ativo US30: Um Script Python para Traders Usando MetaTrader 5

Guia Completo: Construindo um Script Seguro para Análise de Dados no MetaTrader5 com Python

Índice

Introdução

Pré-requisitos

Configurando o Ambiente

1. Criar um Diretório para o Projeto

2. Configurar um Ambiente Virtual (Opcional, mas Recomendado)

3. Instalar Dependências Necessárias

Criando o Arquivo .env

1. Criar o Arquivo .env

2. Atualizar o Arquivo .gitignore

Entendendo o Script Python

1. Imports e Variáveis de Ambiente

2. Configuração de Logging

3. Configuração das Contas MT5

4. Conexão com o MetaTrader5

5. Extração de Dados de Candles

6. Processamento e Análise de Dados

7. Geração de Estatísticas

8. Salvando Resultados

9. Encerramento da Conexão

Script Python Completo com Práticas de Segurança

Executando o Script

1. Ativar o Ambiente Virtual

2. Executar o Script

3. Encerrando o Script

Boas Práticas e Dicas

Conclusão

Guia Completo: Construindo um Bot Seguro de Integração entre Telegram e MetaTrader5 em Python

Índice

Introdução

Pré-requisitos

Configurando o Ambiente

1. Criar um Diretório para o Projeto

2. Configurar um Ambiente Virtual (Opcional, mas Recomendado)

3. Instalar Dependências Necessárias

Criando o Arquivo .env

1. Criar o Arquivo .env

2. Atualizar o Arquivo .gitignore

Entendendo o Script Python

1. Imports e Variáveis de Ambiente

2. Configuração de Logging

3. Configuração das Contas MT5

4. Função de Reconexão com MT5

5. Função para Enviar Ordens para MT5

6. Processando Sinais do Telegram

7. Monitoramento de Conexões

8. Manipulação de Sinais para Encerramento Suave

9. Função Principal (main)

10. Executando o Script

Executando o Script

1. Ativar o Ambiente Virtual

2. Executar o Script

3. Encerrando o Bot

Boas Práticas e Dicas

Conclusão

Automated Trading with MetaTrader5: Order Management and Market Data Collection

1. Initialization and Error Handling

2. Typographical Errors

3. Use of self in the Main Block

4. Market Order Price Adjustment

5. Consistent Error Handling and Logging

6. Validating Input Parameters

7. Enhancing the tf_dict

8. Handling copy_rates_range and copy_rates_from_pos Responses

9. Improving the update_limit_order Method

10. Ensuring Thread Safety

11. Finalizing MT5 Connection

12. Enhancing Documentation and Readability

13. Sample Revised Main Block

Conclusion

Criando o Arquivo `.env`

1. Criar o Arquivo `.env`

2. Atualizar o Arquivo `.gitignore`

Criando o Arquivo `.env`

1. Criar o Arquivo `.env`

2. Atualizar o Arquivo `.gitignore`

9. Função Principal (`main`)

3. Use of `self` in the Main Block

7. Enhancing the `tf_dict`

8. Handling `copy_rates_range` and `copy_rates_from_pos` Responses

9. Improving the `update_limit_order` Method