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

mkdir telegram_mt5_bot
cd telegram_mt5_bot

pip install telethon MetaTrader5 python-dotenv

pip freeze > requirements.txt

pip install -r requirements.txt

# 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

# Arquivo de variáveis de ambiente
.env

# Diretório do ambiente virtual
venv/

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
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__)

# 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

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

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', 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.")

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)

python bot.py