DEV Community

Cover image for Telegram bot para replicar sinais no mt5
Henrique Vital
Henrique Vital

Posted on

Telegram bot para replicar sinais no mt5

#python #telegram

Vamos analisar e explicar o código detalhadamente passo a passo. Este script utiliza a biblioteca Telethon para se conectar ao Telegram e recebe sinais de um grupo, os quais são usados para realizar ordens no MetaTrader 5 (MT5). A execução do código envolve a conexão com o MT5, a escuta de mensagens no Telegram e a execução de ordens de compra e venda com base nessas mensagens.

1. Importação de Bibliotecas 

from telethon import TelegramClient, events
import MetaTrader5 as mt5
import asyncio
import logging
from datetime import datetime
import signal
import os
import sys
import pkg_resources
Enter fullscreen mode Exit fullscreen mode
  • Telethon: Biblioteca para interação com o Telegram (enviar e receber mensagens).
  • MetaTrader5: Biblioteca que permite a interação com a plataforma MetaTrader 5, usada para automação de trading.
  • asyncio: Para trabalhar com operações assíncronas, como aguardar mensagens no Telegram sem bloquear o programa.
  • logging: Para registrar mensagens de log que ajudam no acompanhamento e depuração do código.
  • datetime: Para manipulação de datas e horários.
  • signal: Usada para capturar sinais do sistema, como o sinal de interrupção (Ctrl+C).
  • os, sys, pkg_resources: Para manipulação de arquivos, diretórios, e recursos do sistema.

2. Exibição de Informações do Ambiente 

print("Python executando de:", sys.executable)
print("Ambiente virtual:", sys.prefix)
print("Versão do Python:", sys.version)
print("VIRTUAL_ENV:", os.environ.get('VIRTUAL_ENV'))
Enter fullscreen mode Exit fullscreen mode

Aqui, o código imprime informações sobre o ambiente onde o Python está sendo executado, como a versão do Python, o caminho do ambiente virtual, e o local de execução do Python.

3. Listando Pacotes Instalados 

installed_packages = [d for d in pkg_resources.working_set]
for package in installed_packages:
    print(package)
Enter fullscreen mode Exit fullscreen mode

O código exibe todos os pacotes Python instalados no ambiente atual, usando a biblioteca pkg_resources.

4. Configuração de Logging 

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
Enter fullscreen mode Exit fullscreen mode
  • Configura o logging para registrar mensagens de nível INFO ou superior.
  • O format define o formato das mensagens de log, que inclui a data, o nível de severidade e a mensagem.

5. Configurações do Telegram 

API_ID = '78787878'
API_HASH = '12e957773a9a554cb6e32997122706f6'
PHONE_NUMBER = '+5512991111111'
GROUP_USERNAME = '@Nas100freepip'
Enter fullscreen mode Exit fullscreen mode
  • API_ID e API_HASH: Credenciais da API do Telegram, que são necessárias para usar o Telethon.
  • PHONE_NUMBER: O número de telefone do bot.
  • GROUP_USERNAME: O nome do grupo do Telegram de onde o bot vai ler as mensagens.

6. Configurações das Contas MT5 

CONTAS_MT5 = [
    {"login": 1690062, "senha": '5jsXlBg3~T', "servidor": 'ACGMarkets-Live', "us30": "US30.raw", "nas100": "NAS100.raw", "lote": 4.00}
]
Enter fullscreen mode Exit fullscreen mode

Define uma lista de contas MT5 que o bot pode usar para executar ordens. Cada conta contém:

  • login: O número de login da conta no MetaTrader 5.
  • senha: A senha da conta.
  • servidor: O servidor da corretora.
  • us30 e nas100: Os símbolos dos ativos a serem negociados.
  • lote: O tamanho do lote para as ordens.

7. Função para Reconectar ao MT5 

async def reconectar_mt5(conta, max_tentativas=3):
    for tentativa in range(max_tentativas):
        try:
            if mt5.initialize(path=MT5_PATH, login=conta['login'], server=conta['servidor'], password=conta['senha']):
                logger.info(f"Reconexão bem-sucedida para conta {conta['login']}")
                return True
            else:
                logger.warning(f"Tentativa {tentativa + 1} de reconexão falhou para conta {conta['login']}: {mt5.last_error()}")
        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)
    logger.error(f"Falha ao reconectar à conta {conta['login']} após {max_tentativas} tentativas")
    return False
Enter fullscreen mode Exit fullscreen mode

Essa função tenta reconectar-se ao MetaTrader 5 para uma conta específica até o número máximo de tentativas (max_tentativas). Se a reconexão falhar após o número de tentativas, ela retorna False.

8. Função para Enviar Ordens ao MT5 

async def enviar_ordem(conta, simbolo, acao, lote):
    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
    ...
    resultado = mt5.order_send(pedido)
    ...
    return True
Enter fullscreen mode Exit fullscreen mode

Esta função envia uma ordem de compra ou venda para o MetaTrader 5, dependendo do tipo de ação (compra ou venda). A função:

  • Verifica a disponibilidade do símbolo.
  • Prepara o pedido com as informações necessárias (preço, volume, tipo de ordem, etc.).
  • Envia a ordem para o MT5.

9. Processamento da Mensagem Recebida do Telegram 

async def processar_sinal(mensagem):
    logger.info(f"Mensagem recebida do Telegram: {mensagem}")
    palavras = mensagem.lower().split()
    ativo = None
    acao = None
    ...
    for conta in contas_ativas[:]:
        try:
            sucesso = await enviar_ordem(conta, conta[ativo], 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}")
            contas_ativas.remove(conta)
Enter fullscreen mode Exit fullscreen mode
  • Recebe a mensagem do Telegram e a processa para identificar o ativo (como us30 ou nas100) e a ação (compra ou venda).
  • Para cada conta ativa, tenta enviar a ordem usando a função enviar_ordem. Se uma ordem falhar, a conta é removida da lista de contas ativas.

10. Função de Verificação das Conexões com o MT5 

async def verificar_conexoes():
    while not shutdown_event.is_set():
        for conta in contas_ativas[:]:
            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)
Enter fullscreen mode Exit fullscreen mode

Essa função verifica periodicamente a conexão com o MetaTrader 5 para cada conta ativa. Se uma conta não conseguir se reconectar, ela é removida da lista.

11. Manipulação de Sinais de Interrupção 

def signal_handler(signum, frame):
    logger.info("Sinal de interrupção recebido. Encerrando o programa...")
    asyncio.get_event_loop().call_soon_threadsafe(shutdown_event.set)
Enter fullscreen mode Exit fullscreen mode

Captura sinais de interrupção (SIGINT ou SIGTERM), como o comando Ctrl+C ou a finalização do processo, e termina o programa de forma limpa.

12. Função Principal 

async def main():
    signal.signal(signal.SIGINT, signal_handler)
    signal.signal(signal.SIGTERM, signal_handler)
    ...
    await client.start(phone=PHONE_NUMBER)
    logger.info("Bot Telegram iniciado. Aguardando mensagens...")
    await shutdown_event.wait()
Enter fullscreen mode Exit fullscreen mode
  • Configura o manipulador de sinais.
  • Inicializa as conexões das contas MT5.
  • Cria o cliente Telegram, que fica escutando novas mensagens no grupo especificado. Quando uma mensagem chega, ela é processada e a ordem correspondente é enviada.
  • Aguardará até que um sinal de interrupção seja recebido para encerrar o programa.

13. Execução do Código 

if __name__ == "__main__":
    asyncio.run(main())
Enter fullscreen mode Exit fullscreen mode

A função main() é executada utilizando o asyncio.run() para gerenciar a execução assíncrona do código.

Conclusão:

Esse código é um bot de trading automatizado que usa o Telegram para receber sinais de compra e venda, processa esses sinais, e envia ordens ao MetaTrader 5 para operar no mercado de acordo com as instruções recebidas. O código usa funcionalidades assíncronas para lidar com múltiplas

Aqui está o código completo com os detalhes explicados anteriormente:

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

# Informações do ambiente Python
print("Python executando de:", sys.executable)
print("Ambiente virtual:", sys.prefix)
print("Versão do Python:", sys.version)
print("VIRTUAL_ENV:", os.environ.get('VIRTUAL_ENV'))

print("\nPacotes instalados:")
installed_packages = [d for d in pkg_resources.working_set]
for package in installed_packages:
    print(package)

# Configurar logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

# Configurações do Telegram
API_ID = '88870888'
API_HASH = '12e9588873a9d884cb4e32887182706f8'
PHONE_NUMBER = '+5512996088779'
GROUP_USERNAME = '@Nas100freepip'

# Configurações das contas MT5
CONTAS_MT5 = [
    {"login": 1690062, "senha": '5jsXlBg3~T', "servidor": 'ACGMarkets-Live', "us30": "US30.raw", "nas100": "NAS100.raw", "lote": 4.00}
]

contas_ativas = []
shutdown_event = asyncio.Event()

# No início do script, defina o caminho do terminal MT5 que você quer usar
MT5_PATH = r"C:\Program Files\ACG Markets MetaTrader 5 Terminal\terminal64.exe"  # Ajuste este caminho conforme necessário

async def reconectar_mt5(conta, max_tentativas=3):
    for tentativa in range(max_tentativas):
        try:
            if mt5.initialize(path=MT5_PATH, login=conta['login'], server=conta['servidor'], password=conta['senha']):
                logger.info(f"Reconexão bem-sucedida para conta {conta['login']}")
                return True
            else:
                logger.warning(f"Tentativa {tentativa + 1} de reconexão falhou para conta {conta['login']}: {mt5.last_error()}")
        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):
    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

    # Obter 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

    # Verificar 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

    # Preparar o pedido
    price = mt5.symbol_info_tick(simbolo).ask if acao == mt5.ORDER_TYPE_BUY else mt5.symbol_info_tick(simbolo).bid

    pedido = {
        "action": mt5.TRADE_ACTION_DEAL,
        "symbol": simbolo,
        "volume": float(lote),
        "type": acao,
        "price": price,
        "deviation": 20,
        "magic": 234000,
        "comment": "python script order",
        "type_time": mt5.ORDER_TIME_GTC,
        "type_filling": mt5.ORDER_FILLING_IOC,
    }

    # Enviar a ordem
    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.error(f"Detalhes do pedido: {pedido}")
        logger.error(f"Código de retorno: {resultado.retcode}")
        return False
    else:
        logger.info(f"Ordem enviada com sucesso para {conta['login']} com lote {lote}")
        logger.info(f"Detalhes da ordem: {resultado}")
        return True

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

    palavras = mensagem.lower().split()

    ativo = None
    acao = None

    # Verificar ativo
    if 'us30' in palavras:
        ativo = 'us30'
    elif 'nas100' in palavras:
        ativo = 'nas100'

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

    if not ativo:
        logger.info("Nenhum ativo reconhecido na mensagem.")
        return

    if acao is None:
        logger.info("Nenhuma ação (compra/venda) reconhecida na mensagem.")
        return

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

    for conta in contas_ativas[:]:  # Cria uma cópia da lista para iteração segura
        try:
            sucesso = await enviar_ordem(conta, conta[ativo], 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():
    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):
    logger.info("Sinal de interrupção recebido. Encerrando o programa...")
    asyncio.get_event_loop().call_soon_threadsafe(shutdown_event.set)

async def main():
    # 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)")

    client = TelegramClient('session', API_ID, API_HASH)

    @client.on(events.NewMessage(chats=GROUP_USERNAME))
    async def handler(event):
        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()
    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.initialize(login=conta['login'], server=conta['servidor'], password=conta['senha']):
                mt5.shutdown()
        logger.info("Programa encerrado.")

if __name__ == "__main__":
    asyncio.run(main())
Enter fullscreen mode Exit fullscreen mode

Resumo das funções:

  1. reconectar_mt5: Tenta reconectar ao MetaTrader 5 até um número máximo de tentativas.
  2. enviar_ordem: Envia ordens de compra ou venda para o MetaTrader 5 com base nos sinais recebidos.
  3. processar_sinal: Interpreta as mensagens recebidas do Telegram e executa ordens correspondentes no MetaTrader 5.
  4. verificar_conexoes: Verifica se a conexão com o MetaTrader 5 está ativa e tenta reconectar quando necessário.
  5. signal_handler: Trata sinais de interrupção para finalizar o programa

Top comments (0)

Read next

nabata profile image

Replacing Only the Background of an Image with AI Generation Using the Stable Diffusion Web API

nabata -

algoorgoal profile image

Error: externally-managed-environment on python package installation

Chan -

cleobnvntra profile image

Hacktoberfest 2024 - Part 1

Cleo Buenaventura -

kelsey_ellis12 profile image

Guide: How to build an AI Agent for SRE Teams

Kelsey Ellis -