Асинхронный Telegram-бот в MQL5 через Socket API

Суть: Стандартная функция WebRequest() в MQL5 является синхронной и блокирует поток выполнения советника, что недопустимо в высокочастотном и реальном трейдинге. Решение — использование неблокирующих сокетов (Socket API) в связке с легковесным локальным HTTP-to-HTTPS шлюзом. Это позволяет отправлять уведомления и получать торговые команды из Telegram с нулевым влиянием на скорость обработки тиков в OnTick.

Содержание

Почему WebRequest — это яд для торгового робота

В реальном трейдинге задержка (latency) решает все. Когда ваш советник вызывает WebRequest() для отправки сообщения в Telegram, поток выполнения приостанавливается в ожидании ответа от серверов Telegram. Это ожидание может длиться от 100 мс до нескольких секунд. В этот момент робот «слепнет»: он не может обрабатывать новые тики, модифицировать ордера или реагировать на изменение рынка.

Для решения этой проблемы мы разработаем асинхронную архитектуру. Так как встроенный в MQL5 Socket API не поддерживает SSL/TLS «из коробки» без сложных внешних DLL, лучшим архитектурным решением является использование локального ультрабыстрого Node.js или Python шлюза, который принимает обычные HTTP-запросы от MQL5 по TCP-сокету и мгновенно перенаправляет их в Telegram по защищенному HTTPS-каналу. Советник общается со шлюзом асинхронно, не блокируя поток котировок.

При расчете объемов сделок на основе сигналов из Telegram важно учитывать спецификации контрактов, о чем подробно рассказано в статье Расчет лота для фьючерсов MOEX в MQL5 с учетом шага цены. Для фильтрации сигналов и определения внутридневных уровней можно интегрировать индикаторы, как описано в руководстве Реализация сессионного VWAP в MQL5: Пошаговое руководство. Если вы также работаете с предыдущим поколением терминала, вам пригодится материал про Расчет лота от свободной маржи в MQL4: Скрипт с автоокруглением.

Исходный код

Ниже представлен код асинхронного класса CAsyncTelegram для MQL5, который работает в неблокирующем режиме внутри системного таймера, а также код микро-шлюза на Python.

//+------------------------------------------------------------------+
//|                                                AsyncTelegram.mqh |
//|                                      Copyright 2024, Quant Dev   |
//+------------------------------------------------------------------+
#property copyright "Quant Dev"
#property version   "1.00"

#define SOCKET_TIMEOUT 10 // Таймаут чтения/записи в миллисекундах

enum ENUM_SOCKET_STATE
{
   STATE_DISCONNECTED,
   STATE_CONNECTING,
   STATE_CONNECTED,
   STATE_SENDING,
   STATE_RECEIVING,
   STATE_COMPLETED
};

class CAsyncTelegram
{
private:
   int               m_socket;
   string            m_host;
   int               m_port;
   ENUM_SOCKET_STATE m_state;
   string            m_send_buffer;
   string            m_receive_buffer;
   uint              m_last_action_time;

   void              Log(string msg) { Print("[AsyncTelegram] ", msg); }

public:
                     CAsyncTelegram(string host = "127.0.0.1", int port = 8080);
                    ~CAsyncTelegram();
   bool              SendMessageAsync(string chat_id, string text);
   void              OnTimer();
   ENUM_SOCKET_STATE GetState() { return m_state; }
};

CAsyncTelegram::CAsyncTelegram(string host, int port) 
   : m_socket(INVALID_HANDLE), m_host(host), m_port(port), m_state(STATE_DISCONNECTED)
{
}

CAsyncTelegram::~CAsyncTelegram()
{
   if(m_socket != INVALID_HANDLE)
   {
      SocketClose(m_socket);
   }
}

bool CAsyncTelegram::SendMessageAsync(string chat_id, string text)
{
   if(m_state != STATE_DISCONNECTED)
   {
      Log("Ошибка: Сокет занят предыдущим запросом. Состояние: " + EnumToString(m_state));
      return false;
   }

   m_socket = SocketCreate();
   if(m_socket == INVALID_HANDLE)
   {
      Log("Не удалось создать сокет. Ошибка: " + (string)GetLastError());
      return false;
   }

   // Формируем HTTP POST запрос для нашего локального шлюза
   string payload = "{\"chat_id\":\"" + chat_id + "\",\"text\":\"" + text + "\"}";
   int content_len = StringLen(payload);
   
   m_send_buffer = "POST /send HTTP/1.1\r\n" +
                   "Host: " + m_host + "\r\n" +
                   "Content-Type: application/json\r\n" +
                   "Content-Length: " + (string)content_len + "\r\n" +
                   "Connection: close\r\n\r\n" +
                   payload;

   m_state = STATE_CONNECTING;
   m_last_action_time = GetTickCount();
   
   // Запускаем неблокирующее подключение
   if(!SocketConnect(m_socket, m_host, m_port, 1))
   {
      int err = GetLastError();
      if(err != ERR_NET_SOCKET_CONNECT_PENDING && err != 0)
      {
         Log("Ошибка подключения: " + (string)err);
         SocketClose(m_socket);
         m_socket = INVALID_HANDLE;
         m_state = STATE_DISCONNECTED;
         return false;
      }
   }
   
   return true;
}

void CAsyncTelegram::OnTimer()
{
   if(m_state == STATE_DISCONNECTED) return;

   // Защита от зависания сокета (таймаут 5 секунд)
   if(GetTickCount() - m_last_action_time > 5000)
   {
      Log("Таймаут операции. Сброс сокета.");
      SocketClose(m_socket);
      m_socket = INVALID_HANDLE;
      m_state = STATE_DISCONNECTED;
      return;
   }

   switch(m_state)
   {
      case STATE_CONNECTING:
      {
         if(SocketIsConnected(m_socket))
         {
            m_state = STATE_SENDING;
            m_last_action_time = GetTickCount();
         }
         break;
      }

      case STATE_SENDING:
      {
         if(SocketIsWritable(m_socket))
         {
            uchar data[];
            StringToCharArray(m_send_buffer, data, 0, StringLen(m_send_buffer), CP_UTF8);
            int sent = SocketSend(m_socket, data, ArraySize(data));
            if(sent > 0)
            {
               m_state = STATE_RECEIVING;
               m_receive_buffer = "";
               m_last_action_time = GetTickCount();
            }
            else
            {
               Log("Ошибка отправки данных. Код: " + (string)GetLastError());
               m_state = STATE_DISCONNECTED;
               SocketClose(m_socket);
            }
         }
         break;
      }

      case STATE_RECEIVING:
      {
         if(SocketIsReadable(m_socket))
         {
            uchar data[256];
            ArrayInitialize(data, 0);
            int read = SocketRead(m_socket, data, 256, SOCKET_TIMEOUT);
            if(read > 0)
            {
               m_receive_buffer += CharArrayToString(data, 0, read, CP_UTF8);
               m_last_action_time = GetTickCount();
            }
            else if(read == 0 || GetLastError() == ERR_NET_SOCKET_CLOSED)
            {
               // Чтение завершено, сервер закрыл соединение
               m_state = STATE_COMPLETED;
            }
         }
         break;
      }

      case STATE_COMPLETED:
      {
         // Проверяем HTTP статус ответа
         if(StringFind(m_receive_buffer, "HTTP/1.1 200 OK") != -1)
         {
            Log("Сообщение успешно отправлено асинхронно!");
         }
         else
         {
            Log("Шлюз вернул ошибку: " + m_receive_buffer);
         }
         SocketClose(m_socket);
         m_socket = INVALID_HANDLE;
         m_state = STATE_DISCONNECTED;
         break;
      }
   }
}

А теперь напишем легковесный асинхронный шлюз на Python, который будет принимать локальные запросы от MQL5 по HTTP и пересылать их в Telegram API через библиотеку aiohttp.

# gateway.py
import asyncio
from aiohttp import web, ClientSession

BOT_TOKEN = "ВАШ_ТЕЛЕГРАМ_ТОКЕН"
PORT = 8080

async def handle_send(request):
    try:
        data = await request.json()
        chat_id = data.get("chat_id")
        text = data.get("text")
        
        if not chat_id or not text:
            return web.Response(text="Missing parameters", status=400)
            
        telegram_url = f"https://api.telegram.org/bot{BOT_TOKEN}/sendMessage"
        payload = {"chat_id": chat_id, "text": text}
        
        async with ClientSession() as session:
            async with session.post(telegram_url, json=payload) as resp:
                resp_text = await resp.text()
                return web.Response(text=resp_text, status=resp.status, content_type="application/json")
    except Exception as e:
        return web.Response(text=str(e), status=500)

app = web.Application()
app.router.add_post("/send", handle_send)

if __name__ == "__main__":
    print(f"Запуск MQL5-Telegram шлюза на порту {PORT}...")
    web.run_app(app, host="127.0.0.1", port=PORT)

Разбор параметров

m_socket: Дескриптор системного сокета, создаваемый функцией SocketCreate().
SOCKET_TIMEOUT: Микро-таймаут в миллисекундах для неблокирующего чтения данных из сокета, предотвращающий зависание потока.
STATE_CONNECTING: Состояние ожидания установки TCP-соединения с локальным шлюзом без блокировки основного потока.
SocketIsWritable: Системная функция MQL5, проверяющая готовность сокета к отправке данных. Предотвращает блокировку при вызове SocketSend.
SocketIsReadable: Функция, сигнализирующая о наличии входящих данных в буфере сокета для безопасного вызова SocketRead.

Как запустить

Для развертывания асинхронной системы выполните следующие шаги:

1. Установите Python версии 3.8 или выше. Установите библиотеку aiohttp через терминал: pip install aiohttp.

2. Вставьте ваш токен бота в переменную BOT_TOKEN в файле gateway.py и запустите скрипт. Он начнет слушать порт 8080.

3. Поместите файл AsyncTelegram.mqh в папку MQL5\Include\ вашего терминала.

4. Создайте советник (EA) и подключите класс. В функции OnInit() инициализируйте таймер: EventSetMillisecondTimer(50); (опрос каждые 50 мс). В функции OnTimer() вызывайте метод OnTimer() нашего класса.

5. Для отправки сообщения вызовите SendMessageAsync("ID_ЧАТА", "Ваш текст"). Советник мгновенно продолжит работу, а отправка и получение ответа произойдут в фоновом режиме через сокеты.