За рамками REST и ZeroMQ: Создание моста Protocol Buffers между MetaTrader 5 и Python с выходом на gRPC для инференса в реальном времени

Введение

При подключении работающего советника MetaTrader 5 к серверу модели на Python обычно проходят одни и те же этапы: сначала опрос файлов (состояния гонки при блокировке файлов в Windows), затем WebRequest/REST (накладные расходы HTTP, разбор JSON, отсутствие схемы), затем ZeroMQ (исправляет проблему транспортного уровня, но разработчики по-прежнему вручную формируют JSON-строки, поэтому проблема отсутствия схемы никуда не исчезает). Ни REST, ни ZeroMQ не решают главную проблему: ничто не обеспечивает соответствие вектора признаков, отправляемого клиентом MQL5, тому, что ожидает модель на Python. Если несоответствие полей остается незамеченным, советник торгует на мусорных данных вместо того, чтобы остановиться с явной ошибкой.

В этой статье мы создадим недостающий контракт — схему Protocol Buffers для обмена данными между MetaTrader 5 и Python. Этот транспортный слой ведет себя подобно gRPC, но язык MQL5 не поддерживает gRPC нативно (нет клиента HTTP/2, мультиплексирования потоков и генерации кода с помощью protoc). Поэтому вместо громкого заявления о "gRPC на MQL5" мы реализуем передачу Protobuf по TCP с префиксом длины для фреймирования сообщений. Небольшая прослойка на Python передает эти сообщения полноценному сервису grpc.aio. На стороне Python используется настоящий gRPC: потоковая передача, health checks и сам протокол без имитации. На стороне клиента MQL5 используется строго типизированный формат сообщений с контролем соответствия схеме — без неоднозначности JSON, собранного вручную из строк.

К статье прилагаются семь основных файлов (а также скрипты для тестирования производительности и создания кэша). Ниже мы разберем код каждого файла: что он реализует, как устроен ключевой механизм и почему выбран именно этот подход, а не альтернативы.

О границах решения: это не "gRPC на языке MQL5" в строгом смысле — MQL5 не умеет устанавливать HTTP/2-соединение с поддержкой потоков. В результате вы получаете TCP-канал, по которому передаются сообщения Protobuf с префиксом длины. Прослойка на Python связывает этот канал с полноценным gRPC-бэкендом. К этому различию мы вернемся лишь один раз — в разделе "Архитектура и схема", — а затем повторять его не будем.

Архитектура и схема

Система состоит из трех слоев:

• клиент MQL5 (CGrpcInferenceClient, построенный на основе CProtoWriter/CProtoReader), который устанавливает TCP-соединение с прослойкой, отправляет сообщение Protobuf с префиксом длины и ожидает ответный фрейм в блокирующем режиме с таймаутом и повторными попытками;
• прослойка на Python (grpc_shim_server.py) — TCP-сервер на основе asyncio, который десериализует эти фреймы в объекты Protobuf и передает их серверу модели по полноценному каналу grpc.aio;
• сервер модели на Python (model_server.py) — полноценный сервер grpc.aio.server(), реализующий приведенную ниже схему и поддерживающий унарные и потоковые RPC.

Почему бы не обойтись без прослойки и не обратиться к серверу модели напрямую из клиента MQL5? Для этого пришлось бы самостоятельно реализовать на языке MQL5 фреймирование HTTP/2 и HPACK — масштабная задача с серьезным риском ошибок в реализации. Либо назвать "gRPC" то, что им не является. Прослойка — небольшой фрагмент кода на Python, который легко проверить и который позволяет честно обозначить, что именно реализовано в остальной системе.

Мост между MetaTrader 5 и Python для инференса: трехслойная архитектура

Рис. 1. Трехслойный мост: клиент MQL5 с вручную реализованным кодировщиком Protobuf, прослойка на Python для преобразования TCP в gRPC и полноценный сервер модели на основе grpc.aio. Полноценный gRPC поверх HTTP/2 используется только на участке между прослойкой и сервером модели.

Схема inference.proto — единственный источник истины. Все номера полей из нее вручную продублированы в клиенте MQL5, поскольку плагина protoc для этого языка нет:

syntax = "proto3";

package mt5inference;

// FeatureVector: the input payload built by GrpcInferenceEA.mq5's
// BuildFeatureVector(). "values" is a packed repeated float, encoded
// on the wire as one length-delimited blob of concatenated 4-byte
// little-endian floats rather than one tag per value (see
// CProtoWriter::WritePackedFloatArray in ProtobufWire.mqh).
message FeatureVector {
  string symbol = 1;
  string timeframe = 2;
  int64 timestamp = 3;
  repeated float values = 4;
}

// InferenceRequest: what CGrpcInferenceClient::Predict() sends per bar.
// "features" is an embedded FeatureVector message (nested/length-
// delimited), and "model_version" lets the server reject a request
// built against a stale contract before it ever reaches the model.
message InferenceRequest {
  string request_id = 1;
  FeatureVector features = 2;
  string model_version = 3;
}

// InferenceResponse: decoded by CProtoReader inside
// CGrpcInferenceClient::ParseResponse(). A non-empty "error" is this
// project's success/failure flag -- see SInferenceResult.valid in
// GrpcInferenceClient.mqh.
message InferenceResponse {
  string request_id = 1;
  float probability = 2;
  int32 class_label = 3;
  float latency_ms = 4;
  string model_version = 5;
  string error = 6;
}

// InferenceService: implemented for real by model_server.py using
// grpc.aio. grpc_shim_server.py is a client of this service on MT5's
// behalf -- MQL5 itself never calls these RPCs directly.
service InferenceService {
  rpc Predict(InferenceRequest) returns (InferenceResponse);
  rpc PredictStream(stream InferenceRequest) returns (stream InferenceResponse);
}

FeatureVector.values — повторяющееся поле типа float в упакованном формате: proto3 объединяет его элементы в один блок с префиксом длины, а не добавляет тег к каждому элементу. Это важно для вектора из восьми значений типа float, отправляемого на каждом баре. InferenceRequest содержит вложенное сообщение FeatureVector и версию модели, которую сервер может отклонить, если она устарела. Поле error в InferenceResponse также служит признаком успешного или неуспешного выполнения запроса. InferenceService объявляет унарный RPC Predict и потоковый RPC PredictStream. Пока клиент MQL5 использует только унарный вызов, но потоковый RPC уже работает на стороне Python.

Реализация Protobuf на MQL5 — ProtobufWire.mqh

Для языка MQL5 нет плагина protoc, поэтому в ProtobufWire.mqh формат передачи данных реализован вручную: сначала идет тег (номер поля, сдвинутый влево на 3 бита и объединенный операцией побитового ИЛИ с 3-битным типом представления на проводе), затем — полезная нагрузка. Целые числа кодируются как varint с основанием 128: в каждом байте 7 бит отведены под значение, а старший бит служит флагом продолжения. Поэтому небольшие номера полей и значения длины занимают всего один байт. CProtoWriter формирует исходящие сообщения, а CProtoReader разбирает входящие. Номера полей важнее, чем кажется: при передаче данных вместо имен используются именно они. Поэтому каждая константа клиента, воспроизводящая номер поля из inference.proto, нужна для согласованности написанного вручную кодировщика со схемой.

Запись — буфер и базовая операция. Внутренний байтовый буфер растет геометрически (размер удваивается с дополнительным запасом), а не увеличивается ровно до нужного размера. Поэтому амортизированная сложность добавления одного байта остается близкой к O(1): полное копирование не требуется при каждом добавлении. WriteVarint — базовая операция, к которой обращаются все остальные методы. WriteTag лишь объединяет номер поля и тип кодирования в одно значение и передает его в WriteVarint.

Буфер и базовая операция добавления данных:

//+------------------------------------------------------------------+
//|                                             ProtobufWire.mqh     |
//|                                       https://www.mql5.com       |
//+------------------------------------------------------------------+
#property strict

//+------------------------------------------------------------------+
//| Protobuf wire-type constants                                     |
//+------------------------------------------------------------------+
#define PB_WIRETYPE_VARINT   0     // int32/int64/bool/enum on the wire
#define PB_WIRETYPE_FIXED64  1     // double, fixed64, sfixed64
#define PB_WIRETYPE_LENDELIM 2     // string, bytes, embedded message, packed repeated
#define PB_WIRETYPE_FIXED32  5     // float, fixed32, sfixed32

//+------------------------------------------------------------------+
//| CProtoWriter                                                     |
//| Minimal hand-rolled Protobuf writer. MQL5 has no protoc plugin,  |
//| so every method below encodes exactly what protoc would generate |
//| for the matching .proto field type -- documented per method.     |
//+------------------------------------------------------------------+
class CProtoWriter
  {
private:
   //+------------------------------------------------------------------+
   //| Internal growable byte buffer                                    |
   //+------------------------------------------------------------------+
   uchar             m_buf[];         // raw output bytes accumulated so far
   int               m_len;           // number of bytes actually used in m_buf

   //+------------------------------------------------------------------+
   //| EnsureCapacity                                                   |
   //+------------------------------------------------------------------+
   void              EnsureCapacity(const int extra)
     {
      //--- grow geometrically (double + slack) instead of by exact need,
      //--- so repeated single-byte appends don't cause O(n^2) resizing
      if(m_len+extra>ArraySize(m_buf))
         ArrayResize(m_buf,MathMax(m_len+extra,ArraySize(m_buf)*2+64));
     }

   //+------------------------------------------------------------------+
   //| AppendByte                                                       |
   //+------------------------------------------------------------------+
   void              AppendByte(const uchar b)
     {
      EnsureCapacity(1);
      m_buf[m_len++]=b;
     }

Создание объекта и извлечение байтов:

public:
   //+------------------------------------------------------------------+
   //| CProtoWriter (constructor)                                       |
   //+------------------------------------------------------------------+
                     CProtoWriter(void) : m_len(0) { ArrayResize(m_buf,64); }

   //+------------------------------------------------------------------+
   //| Reset                                                            |
   //+------------------------------------------------------------------+
   void              Reset(void) { m_len=0; } // reuse the buffer across calls, no realloc

   //+------------------------------------------------------------------+
   //| Size                                                             |
   //+------------------------------------------------------------------+
   int               Size(void) const { return m_len; }

   //+------------------------------------------------------------------+
   //| GetBytes                                                         |
   //| Copies exactly m_len bytes out -- m_buf itself may be larger     |
   //| than m_len because of the geometric growth in EnsureCapacity.    |
   //+------------------------------------------------------------------+
   void              GetBytes(uchar &out[])
     {
      ArrayResize(out,m_len);
      for(int i=0;i<m_len;i++)
         out[i]=m_buf[i];
     }

Методы кодирования varint и тегов, к которым обращается весь остальной код:

   //+------------------------------------------------------------------+
   //| WriteVarint                                                      |
   //| Unsigned LEB128 varint: 7 payload bits per byte, MSB=1 means     |
   //| "more bytes follow". Every other Write*Field method below funnels|
   //| through this for its length or numeric payload.                  |
   //+------------------------------------------------------------------+
   void              WriteVarint(ulong value)
     {
      while(value>=0x80)
        {
         //--- emit low 7 bits with continuation bit set, then shift
         AppendByte((uchar)((value & 0x7F) | 0x80));
         value>>=7;
        }
      AppendByte((uchar)value); // final byte, continuation bit is 0
     }

   //+------------------------------------------------------------------+
   //| WriteTag                                                         |
   //| Builds the (field_number<<3 | wire_type) tag varint that         |
   //| precedes every field on the wire, per the Protobuf spec.         |
   //+------------------------------------------------------------------+
   void              WriteTag(const int field_number,const int wire_type)
     {
      WriteVarint((ulong)((field_number<<3) | wire_type));
     }

Запись — скалярные и составные поля. Типы int32 и int64 кодируются как varint: в бинарном представлении Protobuf эти типы кодируются одинаково. Различие проявляется лишь в типе, который присваивается декодированному значению. Для чисел с плавающей запятой используется небольшой тип объединения (UFloatBytes), реализованный вручную и объявленный ближе к началу файла. В отличие от C/C++, язык MQL5 не поддерживает объявление анонимных объединений непосредственно в теле функции, поэтому требуется именованный тип объединения — как и в случае со структурой. Значения fixed32 записываются в порядке от младшего байта к старшему (little-endian). При этом предполагается, что система, на которой выполняется код, использует тот же порядок байтов (x86/x64). При переносе кода на другую платформу проверьте это предположение еще раз. Строки кодируются в UTF-8 и снабжаются префиксом длины. Обратите внимание на вычитание единицы (-1): оно исключает из длины завершающий нулевой байт, добавляемый StringToCharArray, который не должен учитываться в префиксе длины Protobuf. Если пропустить это вычитание, каждое строковое поле окажется на один байт длиннее, причем ошибка останется незамеченной.

Сам вектор признаков записывается с помощью WritePackedFloatArray с использованием упакованного кодирования, принятого в proto3 по умолчанию для повторяющихся скалярных полей: один тег и один блок с префиксом длины, содержащий последовательно записанные четырехбайтовые значения float, на весь массив из восьми элементов вместо отдельного тега для каждого элемента. С вложенными сообщениями (InferenceRequest.features) все проще, чем кажется: сначала вложенное сообщение полностью сериализуется отдельным экземпляром CProtoWriter, а затем полученные байты записываются как еще одно поле с префиксом длины. Именно так функция Predict() далее в статье формирует внешний запрос, включающий вектор признаков.

Поля типов int64 и int32 (оба типа используют одно и то же кодирование varint):

   //+------------------------------------------------------------------+
   //| WriteInt64Field                                                  |
   //+------------------------------------------------------------------+
   void              WriteInt64Field(const int field_number,const long value)
     {
      WriteTag(field_number,PB_WIRETYPE_VARINT);
      WriteVarint((ulong)value);
     }

   //+------------------------------------------------------------------+
   //| WriteInt32Field                                                  |
   //| int32 has no distinct wire representation -- it is promoted to   |
   //| int64 and encoded the same way, matching protoc's own behavior.  |
   //+------------------------------------------------------------------+
   void              WriteInt32Field(const int field_number,const int value)
     {
      WriteInt64Field(field_number,(long)value);
     }

Именованный тип объединения, используемый приведенными ниже методами работы с float:

//+------------------------------------------------------------------+
//| UFloatBytes                                                      |
//+------------------------------------------------------------------+
union UFloatBytes
  {
   float             f;      // the value as a 32-bit IEEE-754 float
   uchar             b[4];   // the same 4 bytes, addressable individually
  };

Кодирование float с использованием этого объединения:

   //+------------------------------------------------------------------+
   //| FloatToBytesLE                                                   |
   //| Protobuf fixes 32-bit numeric fields as little-endian on the     |
   //| wire. The union re-interprets the float's 4 raw bytes without any|
   //| numeric conversion -- this assumes a little-endian host CPU,     |
   //| true for the x86/x64 terminals this EA targets.                  |
   //+------------------------------------------------------------------+
   static void       FloatToBytesLE(const float value,uchar &out[])
     {
      ArrayResize(out,4);
      UFloatBytes u;
      u.f=value;
      out[0]=u.b[0]; out[1]=u.b[1]; out[2]=u.b[2]; out[3]=u.b[3];
     }

   //+------------------------------------------------------------------+
   //| WriteFloatField                                                  |
   //+------------------------------------------------------------------+
   void              WriteFloatField(const int field_number,const float value)
     {
      WriteTag(field_number,PB_WIRETYPE_FIXED32);
      uchar b[4];
      FloatToBytesLE(value,b);
      for(int i=0;i<4;i++)
         AppendByte(b[i]);
     }

Строковые поля (UTF-8 с префиксом длины):

   //+------------------------------------------------------------------+
   //| WriteStringField                                                 |
   //| Strings are length-delimited: tag, then a varint byte count,     |
   //| then the raw UTF-8 bytes. StringToCharArray appends a trailing   |
   //| null terminator that Protobuf does not want on the wire, so the  |
   //| count is taken one short of the returned array size.             |
   //+------------------------------------------------------------------+
   void              WriteStringField(const int field_number,const string value)
     {
      uchar sb[];
      int n=StringToCharArray(value,sb,0,-1,CP_UTF8)-1; // drop trailing '\0'
      if(n<0)
         n=0;
      WriteTag(field_number,PB_WIRETYPE_LENDELIM);
      WriteVarint((ulong)n);
      for(int i=0;i<n;i++)
         AppendByte(sb[i]);
     }

Упакованный массив float, используемый для вектора признаков:

   //+------------------------------------------------------------------+
   //| WritePackedFloatArray                                            |
   //| proto3 packs repeated scalar-numeric fields by default: ONE      |
   //| length-delimited blob of concatenated fixed32 floats, rather     |
   //| than a separate tag per array element. This is the encoding for  |
   //| FeatureVector.values in inference.proto.                         |
   //+------------------------------------------------------------------+
   void              WritePackedFloatArray(const int field_number,const float &values[])
     {
      int n=ArraySize(values);
      WriteTag(field_number,PB_WIRETYPE_LENDELIM);
      WriteVarint((ulong)(n*4)); // byte length of the packed blob, not element count
      for(int i=0;i<n;i++)
        {
         uchar b[4];
         FloatToBytesLE(values[i],b);
         for(int j=0;j<4;j++)
            AppendByte(b[j]);
        }
     }

Вложение одного сообщения в другое:

   //+------------------------------------------------------------------+
   //| WriteEmbeddedMessage                                             |
   //| Nested messages (e.g. InferenceRequest.features) are encoded by  |
   //| serializing the submessage first with its own writer instance,   |
   //| then treating the resulting bytes as one length-delimited blob.  |
   //| The caller is responsible for serializing submsg beforehand.     |
   //+------------------------------------------------------------------+
   void              WriteEmbeddedMessage(const int field_number,const uchar &submsg[])
     {
      int n=ArraySize(submsg);
      WriteTag(field_number,PB_WIRETYPE_LENDELIM);
      WriteVarint((ulong)n);
      for(int i=0;i<n;i++)
         AppendByte(submsg[i]);
     }
  };

Чтение — механизм совместимости с будущими версиями. CProtoReader реализует операции, обратные операциям CProtoWriter: Attach копирует массив байтов и сбрасывает позицию чтения (сохраняется копия, а не ссылка, поэтому объект чтения остается работоспособным, даже если размер исходного массива изменится прямо во время разбора). ReadVarint и ReadTag выполняют обратное преобразование той же битовой упаковки. Защита от переполнения ограничивает сдвиг 63 битами, чтобы при поврежденном или обрезанном фрейме разбор не зацикливался в ожидании сброса бита продолжения. Особо стоит выделить метод SkipField: зная только тип кодирования, он пропускает полезную нагрузку поля независимо от его смысла. Именно это обеспечивает совместимость с будущими версиями: если в model_server.py когда-нибудь добавят седьмое поле ответа, советник, скомпилированный с использованием кода из этой статьи, продолжит работать. Он просто проигнорирует новое поле, а не завершит разбор всего сообщения с ошибкой. Именно этот механизм используется в ветке default метода ParseResponse, которую мы рассмотрим в следующем разделе.

Конструктор CProtoReader и состояние курсора чтения:

//+------------------------------------------------------------------+
//| CProtoReader                                                     |
//| Minimal hand-rolled Protobuf reader used to parse the            |
//| InferenceResponse bytes that come back from the Python shim.     |
//+------------------------------------------------------------------+
class CProtoReader
  {
private:
   uchar             m_buf[];        // local copy of the bytes being parsed
   int               m_len;          // total length of m_buf in use
   int               m_pos;          // current read cursor

public:
   //+------------------------------------------------------------------+
   //| Attach                                                           |
   //| Copies the caller's byte array in and resets the read cursor.    |
   //| A copy (not a reference) is kept deliberately so the reader      |
   //| stays valid even if the caller's array is resized afterward.     |
   //+------------------------------------------------------------------+
   void              Attach(const uchar &data[],const int len)
     {
      ArrayResize(m_buf,len);
      for(int i=0;i<len;i++)
         m_buf[i]=data[i];
      m_len=len;
      m_pos=0;
     }

Проверка достижения конца буфера и декодирование varint:

   //+------------------------------------------------------------------+
   //| Eof                                                              |
   //+------------------------------------------------------------------+
   bool              Eof(void) const { return m_pos>=m_len; }

   //+------------------------------------------------------------------+
   //| ReadVarint                                                       |
   //| Mirror of CProtoWriter::WriteVarint. The shift>63 guard exists   |
   //| because a corrupted or truncated frame could otherwise never set |
   //| the continuation bit to 0, spinning through the whole buffer.    |
   //+------------------------------------------------------------------+
   bool              ReadVarint(ulong &value)
     {
      value=0;
      int shift=0;
      while(m_pos<m_len)
        {
         uchar b=m_buf[m_pos++];
         value|=((ulong)(b & 0x7F))<<shift;
         if((b & 0x80)==0)
            return true;      // continuation bit clear -- varint complete
         shift+=7;
         if(shift>63)
            return false;     // malformed-varint guard, see Edge Cases section
        }
      return false;           // ran out of bytes mid-varint
     }

Разбор тегов и пропуск неизвестных полей:

   //+------------------------------------------------------------------+
   //| ReadTag                                                          |
   //+------------------------------------------------------------------+
   bool              ReadTag(int &field_number,int &wire_type)
     {
      ulong tag;
      if(!ReadVarint(tag))
         return false;
      field_number=(int)(tag>>3);
      wire_type=(int)(tag & 0x07);
      return true;
     }

   //+------------------------------------------------------------------+
   //| SkipField                                                        |
   //| Skips a field's payload based purely on its wire type, without   |
   //| knowing what the field means. This is what makes the parser      |
   //| forward-compatible: an unknown field number from a newer server  |
   //| is skipped cleanly instead of breaking the whole parse.          |
   //+------------------------------------------------------------------+
   bool              SkipField(const int wire_type)
     {
      ulong tmp;
      switch(wire_type)
        {
         case PB_WIRETYPE_VARINT:
            return ReadVarint(tmp);
         case PB_WIRETYPE_FIXED64:
            m_pos+=8;
            return(m_pos<=m_len);
         case PB_WIRETYPE_FIXED32:
            m_pos+=4;
            return(m_pos<=m_len);
         case PB_WIRETYPE_LENDELIM:
           {
            ulong n;
            if(!ReadVarint(n))
               return false;
            m_pos+=(int)n;
            return(m_pos<=m_len);
           }
         default:
            return false;     // unrecognized wire type -- cannot safely skip
        }
     }

Чтение значений float фиксированного размера и данных с префиксом длины:

   //+------------------------------------------------------------------+
   //| ReadFloat                                                        |
   //| Inverse of CProtoWriter::FloatToBytesLE -- reinterprets 4 raw    |
   //| little-endian bytes back into a float via the same union trick.  |
   //+------------------------------------------------------------------+
   bool              ReadFloat(float &value)
     {
      if(m_pos+4>m_len)
         return false;
      UFloatBytes u;
      u.b[0]=m_buf[m_pos]; u.b[1]=m_buf[m_pos+1]; u.b[2]=m_buf[m_pos+2]; u.b[3]=m_buf[m_pos+3];
      m_pos+=4;
      value=u.f;
      return true;
     }

   //+------------------------------------------------------------------+
   //| ReadLengthDelimited                                              |
   //+------------------------------------------------------------------+
   bool              ReadLengthDelimited(uchar &out[])
     {
      ulong n;
      if(!ReadVarint(n))
         return false;
      if(m_pos+(int)n>m_len)
         return false;        // declared length would overrun the buffer
      ArrayResize(out,(int)n);
      for(int i=0;i<(int)n;i++)
         out[i]=m_buf[m_pos+i];
      m_pos+=(int)n;
      return true;
     }

Вспомогательные обертки для string и int32:

   //+------------------------------------------------------------------+
   //| ReadString                                                       |
   //+------------------------------------------------------------------+
   bool              ReadString(string &value)
     {
      uchar raw[];
      if(!ReadLengthDelimited(raw))
         return false;
      value=CharArrayToString(raw,0,ArraySize(raw),CP_UTF8);
      return true;
     }

   //+------------------------------------------------------------------+
   //| ReadInt32                                                        |
   //+------------------------------------------------------------------+
   bool              ReadInt32(int &value)
     {
      ulong v;
      if(!ReadVarint(v))
         return false;
      value=(int)v;
      return true;
     }

Клиент MQL5 и его интеграция с советником

Ниже показано расположение байтов в данных, которые отправляет клиент. Эту схему полезно держать перед глазами при чтении раздела:

Фреймирование сообщений Protobuf с префиксом длины

Рис. 2. Расположение байтов в данных, отправляемых MetaTrader 5: четырехбайтовый префикс длины, за которым следует сериализованное сообщение с тегами полей. На увеличенном фрагменте показано устройство повторяющегося поля float с упакованным представлением внутри этого сообщения.

В GrpcInferenceClient.mqh кодировщик используется для выполнения сетевых вызовов. Реализация состоит из четырех частей. Константы и структура результата: константы с номерами полей, заданные директивами #define, должны точно соответствовать inference.proto. Иначе при передаче данных поле незаметно получит неверный тег.

//+------------------------------------------------------------------+
//|                                      GrpcInferenceClient.mqh     |
//|                                       https://www.mql5.com       |
//+------------------------------------------------------------------+
#property strict
#include "ProtobufWire.mqh"

//+------------------------------------------------------------------+
//| InferenceRequest field numbers -- MUST mirror inference.proto    |
//+------------------------------------------------------------------+
#define GRPC_FIELD_REQ_ID        1     // request_id       (string)
#define GRPC_FIELD_FEATURES      2     // features         (FeatureVector, embedded)
#define GRPC_FIELD_MODEL_VERSION 3     // model_version    (string)

//+------------------------------------------------------------------+
//| FeatureVector field numbers -- MUST mirror inference.proto       |
//+------------------------------------------------------------------+
#define GRPC_FV_FIELD_SYMBOL     1     // symbol           (string)
#define GRPC_FV_FIELD_TIMEFRAME  2     // timeframe        (string)
#define GRPC_FV_FIELD_TIMESTAMP  3     // timestamp        (int64)
#define GRPC_FV_FIELD_VALUES     4     // values           (repeated float, packed)

//+------------------------------------------------------------------+
//| InferenceResponse field numbers -- MUST mirror inference.proto   |
//+------------------------------------------------------------------+
#define GRPC_RESP_FIELD_REQ_ID       1     // request_id     (string)
#define GRPC_RESP_FIELD_PROBABILITY  2     // probability    (float)
#define GRPC_RESP_FIELD_CLASS_LABEL  3     // class_label    (int32)
#define GRPC_RESP_FIELD_LATENCY_MS   4     // latency_ms     (float)
#define GRPC_RESP_FIELD_MODEL_VER    5     // model_version  (string)
#define GRPC_RESP_FIELD_ERROR        6     // error          (string)

//+------------------------------------------------------------------+
//| SInferenceResult                                                 |
//| Decoded, ready-to-use view of an InferenceResponse. valid is set |
//| by ParseResponse() based on whether error came back empty.       |
//+------------------------------------------------------------------+
struct SInferenceResult
  {
   string            request_id;      // echoed back from the request
   float             probability;     // model output in [0,1]
   int               class_label;     // 1 = long signal, 0 = short signal, -1 = unset
   float             latency_ms;      // server-side inference time, for diagnostics
   string            model_version;   // model version the server actually used
   string            error;           // non-empty means the call failed logically
   bool              valid;           // convenience flag mirroring (error=="")
  };

Фреймирование сообщений по префиксу длины: в обоих направлениях перед каждой полезной нагрузкой передается четырехбайтовый заголовок с ее длиной в порядке от старшего байта к младшему (big-endian). SendFramed отправляет сначала заголовок, затем полезную нагрузку и проверяет, что в обоих случаях переданы все байты. RecvExact выполняет чтение в цикле, поскольку за один вызов SocketRead может вернуть меньше байтов, чем запрошено. RecvFramed сначала считывает заголовок и, прежде чем использовать указанную в нем длину для чтения полезной нагрузки, проверяет, что она не превышает 16 МБ. Это ограничение защищает от неконтролируемого выделения памяти или бесконечного ожидания из-за поврежденного поля длины.

private:
   //+------------------------------------------------------------------+
   //| Connection state and configuration                               |
   //+------------------------------------------------------------------+
   int               m_socket;        // MQL5 socket handle, or INVALID_HANDLE
   string            m_host;          // shim host, e.g. "127.0.0.1"
   int               m_port;          // shim port, e.g. 50151
   int               m_timeout_ms;    // per-call receive deadline
   int               m_max_retries;   // attempts beyond the first before giving up

   //+------------------------------------------------------------------+
   //| SendFramed                                                       |
   //| Writes the 4-byte big-endian length prefix, then the payload     |
   //| itself. Both SocketSend calls are checked against the byte count |
   //| they were asked to send -- a short write means the transport is  |
   //| in a bad state and the caller should reconnect, not keep going.  |
   //+------------------------------------------------------------------+
   bool              SendFramed(const uchar &payload[])
     {
      int n=ArraySize(payload);
      uchar header[4];
      header[0]=(uchar)((n>>24)&0xFF);
      header[1]=(uchar)((n>>16)&0xFF);
      header[2]=(uchar)((n>>8)&0xFF);
      header[3]=(uchar)(n&0xFF);

      if(SocketSend(m_socket,header,4)!=4)
         return false;
      if(n>0 && SocketSend(m_socket,payload,n)!=n)
         return false;
      return true;
     }

   //+------------------------------------------------------------------+
   //| RecvExact                                                        |
   //| Blocks until exactly n bytes have arrived or deadline_ms has     |
   //| elapsed. SocketRead can return fewer bytes than requested on a   |
   //| single call, so this loop keeps accumulating into out[] rather   |
   //| than assuming one call returns the whole frame. Unlike a         |
   //| self-growing stream API, SocketRead expects its destination      |
   //| buffer pre-sized to the amount being requested -- chunk[] is     |
   //| resized to exactly how many bytes are both available and still   |
   //| needed before each call.                                         |
   //+------------------------------------------------------------------+
   bool              RecvExact(uchar &out[],const int n,const int deadline_ms)
     {
      ArrayResize(out,n);
      int received=0;
      uint start=GetTickCount();
      while(received<n)
        {
         if((int)(GetTickCount()-start)>deadline_ms)
            return false;                       // deadline exceeded
         uint avail=SocketIsReadable(m_socket);
         if(avail==0)
           {
            Sleep(1);
            continue;                            // nothing to read yet
           }
         uint want=(uint)(n-received);
         if(avail<want)
            want=avail;                          // don't ask for more than is buffered
         uchar chunk[];
         ArrayResize(chunk,(int)want);
         int got=SocketRead(m_socket,chunk,want,10);
         if(got<=0)
           {
            Sleep(1);
            continue;
           }
         for(int i=0;i<got;i++)
            out[received+i]=chunk[i];
         received+=got;
        }
      return true;
     }

   //+------------------------------------------------------------------+
   //| RecvFramed                                                       |
   //| Reads the 4-byte length header first, sanity-checks it against   |
   //| a hard upper bound (see Edge Cases section), then reads exactly  |
   //| that many payload bytes.                                         |
   //+------------------------------------------------------------------+
   bool              RecvFramed(uchar &payload[],const int deadline_ms)
     {
      uchar header[];
      if(!RecvExact(header,4,deadline_ms))
         return false;
      int n=(header[0]<<24)|(header[1]<<16)|(header[2]<<8)|header[3];
      if(n<0 || n>16*1024*1024)
         return false;                           // corrupt-frame sanity guard
      return RecvExact(payload,n,deadline_ms);
     }

Разбор ответа с пропуском неизвестных полей: ParseResponse перебирает поля ответа с помощью switch: для известных полей вызывает типизированные методы установки значений, а остальные пропускает с помощью SkipField. Так реализуется описанная в предыдущем разделе совместимость с будущими версиями. Значение result.valid определяется единственным условием — пусто ли поле error; именно оно служит критерием успешного выполнения запроса.

   //+------------------------------------------------------------------+
   //| ParseResponse                                                    |
   //| Walks the InferenceResponse bytes field by field. The default    |
   //| case in the switch calls SkipField() rather than failing, so a   |
   //| response with fields this client doesn't recognize still parses  |
   //| the fields it does know about.                                   |
   //+------------------------------------------------------------------+
   bool              ParseResponse(const uchar &data[],SInferenceResult &result)
     {
      CProtoReader r;
      r.Attach(data,ArraySize(data));

      result.request_id="";
      result.probability=0.0f;
      result.class_label=-1;
      result.latency_ms=0.0f;
      result.model_version="";
      result.error="";

      while(!r.Eof())
        {
         int fn,wt;
         if(!r.ReadTag(fn,wt))
            break;

         switch(fn)
           {
            case GRPC_RESP_FIELD_REQ_ID:      r.ReadString(result.request_id);    break;
            case GRPC_RESP_FIELD_PROBABILITY: r.ReadFloat(result.probability);    break;
            case GRPC_RESP_FIELD_CLASS_LABEL: r.ReadInt32(result.class_label);    break;
            case GRPC_RESP_FIELD_LATENCY_MS:  r.ReadFloat(result.latency_ms);     break;
            case GRPC_RESP_FIELD_MODEL_VER:   r.ReadString(result.model_version); break;
            case GRPC_RESP_FIELD_ERROR:       r.ReadString(result.error);         break;
            default:                          r.SkipField(wt);                   break;
           }
        }

      result.valid=(StringLen(result.error)==0);
      return result.valid;
     }

Порядок работы Predict — повторное подключение, повторные попытки и задержки между ними: Predict формирует вложенное сообщение FeatureVector, включает его во внешний запрос InferenceRequest, а затем выполняет цикл отправки и приема с экспоненциальным увеличением задержки между повторными попытками (50 мс × 2^attempt: так при кратковременном перезапуске прослойка не перегружается частыми запросами, а после единичного сбоя обмен по-прежнему быстро восстанавливается). При любом сбое — подключения, отправки или приема — перед повторной попыткой вызывается Disconnect(), поскольку после неудачного обмена состояние сокета неизвестно. Если лимит повторных попыток исчерпан, транспортная ошибка записывается в result.error — то же поле, которое используется при логической ошибке (несоответствии признаков). Поэтому OnTick() одинаково обрабатывает оба вида ошибок, без специальных проверок для каждого из них.

   //+------------------------------------------------------------------+
   //| Predict                                                          |
   //| Full request/response cycle: serialize the embedded FeatureVector|
   //| first, wrap it into the outer InferenceRequest, then send/receive|
   //| with retry. Each retry uses exponential backoff (50ms * 2^n) so a|
   //| transient shim restart doesn't get hammered by a tight loop.     |
   //+------------------------------------------------------------------+
   bool              Predict(const string request_id,
                              const string symbol,
                              const string timeframe,
                              const long timestamp,
                              const float &features[],
                              const string model_version,
                              SInferenceResult &result)
     {
      result.valid=false;

      //--- build FeatureVector submessage first, then embed it below
      CProtoWriter fvw;
      fvw.WriteStringField(GRPC_FV_FIELD_SYMBOL,symbol);
      fvw.WriteStringField(GRPC_FV_FIELD_TIMEFRAME,timeframe);
      fvw.WriteInt64Field(GRPC_FV_FIELD_TIMESTAMP,timestamp);
      fvw.WritePackedFloatArray(GRPC_FV_FIELD_VALUES,features);
      uchar fv_bytes[];
      fvw.GetBytes(fv_bytes);

      //--- build the outer InferenceRequest around the embedded bytes
      CProtoWriter reqw;
      reqw.WriteStringField(GRPC_FIELD_REQ_ID,request_id);
      reqw.WriteEmbeddedMessage(GRPC_FIELD_FEATURES,fv_bytes);
      reqw.WriteStringField(GRPC_FIELD_MODEL_VERSION,model_version);
      uchar req_bytes[];
      reqw.GetBytes(req_bytes);

      int attempt=0;
      while(attempt<=m_max_retries)
        {
         if(!IsConnected() && !Connect())
           {
            attempt++;
            Sleep(50*(1<<attempt));           // exponential backoff before retry
            continue;
           }

         if(!SendFramed(req_bytes))
           {
            Disconnect();                     // socket is in an unknown state, drop it
            attempt++;
            Sleep(50*(1<<attempt));
            continue;
           }

         uchar resp_bytes[];
         if(!RecvFramed(resp_bytes,m_timeout_ms))
           {
            Disconnect();
            attempt++;
            Sleep(50*(1<<attempt));
            continue;
           }

         return ParseResponse(resp_bytes,result);
        }

      result.error="transport failure after retries";
      return false;
     }

Всем этим управляет советник GrpcInferenceEA.mq5. В самом начале OnInit() проверяет, совпадает ли фактическое число элементов вектора, сформированного BuildFeatureVector(), со значением константы NUM_FEATURES. При несоответствии работа прерывается с явным сообщением об ошибке, чтобы не отправить некорректный вектор. Кроме того, OnInit() однократно определяет, запущен ли советник в тестере стратегий (MQLInfoInteger(MQL_TESTER)). От этого зависит, установит ли советник соединение через сокет или загрузит заранее записанный кэш, — подробнее об этом в разделе "Согласованность и тестирование".

OnInit():

//+------------------------------------------------------------------+
//| OnInit                                                           |
//| Runs the feature-contract cross-check before doing anything else:|
//| BuildFeatureVector()'s actual output width must equal the        |
//| NUM_FEATURES constant, or initialization aborts loudly rather    |
//| than silently sending a mismatched vector to the model server.   |
//+------------------------------------------------------------------+
int OnInit()
  {
   float probe[];
   int built_width=BuildFeatureVector(probe);
   if(built_width!=NUM_FEATURES)
     {
      PrintFormat("[FEATURE_CONTRACT_MISMATCH] NUM_FEATURES=%d but BuildFeatureVector() produced %d columns. Aborting init.",
                  NUM_FEATURES,built_width);
      return(INIT_PARAMETERS_INCORRECT);
     }

   ExtAtrHandle=iATR(_Symbol,_Period,InpAtrPeriod);
   if(ExtAtrHandle==INVALID_HANDLE)
     {
      Print("[INIT_FAIL] Could not create ATR handle.");
      return(INIT_FAILED);
     }

   ExtClient.Configure(InpServerHost,InpServerPort,InpSocketTimeoutMs,InpMaxRetries);
   if(!ExtClient.Connect())
      //--- non-fatal: the retry/backoff logic inside Predict() will
      //--- attempt to reconnect on the first real tick regardless
      Print("[GRPC_WARN] Initial connect to inference shim failed; will retry on first tick.");

   ExtTrade.SetExpertMagicNumber(990211);
   return(INIT_SUCCEEDED);
  }

OnDeinit():

//+------------------------------------------------------------------+
//| OnDeinit                                                         |
//+------------------------------------------------------------------+
void OnDeinit(const int reason)
  {
   ExtClient.Disconnect();
   if(ExtAtrHandle!=INVALID_HANDLE)
      IndicatorRelease(ExtAtrHandle);
  }

BuildFeatureVector() вычисляет на каждом баре восемь значений: три однобарные доходности с лагом, относительные размеры диапазона, тела и теней свечи, а также нормализованный ATR. Каждое отношение рассчитывается с делением на текущую цену закрытия, чтобы признаки оставались сопоставимыми независимо от абсолютного уровня цены инструмента (диапазон в пять пунктов имеет совершенно разное значение для акции стоимостью 50 долларов и для XAUUSD с ценой около 4 300 долларов). Одна и та же функция вызывается и из OnInit(), и из OnTick(). Это важно: при запуске проверяется число элементов вектора, сформированного тем же кодом, который используется в рабочем режиме, а не отдельно поддерживаемой копией, которая может незаметно перестать ему соответствовать.

//+------------------------------------------------------------------+
//| BuildFeatureVector                                               |
//| Computes the 8-column feature row sent to the model: three       |
//| lagged returns, candle range/body/wick ratios, and normalized    |
//| ATR. All ratios are normalized by the current close so the model |
//| sees scale-invariant inputs regardless of instrument price level.|
//| Also called from OnInit() with an empty probe array purely to    |
//| report the column count for the feature-contract check -- in that|
//| case there may not be enough warmed-up bars yet, so the guards   |
//| below return a correctly-sized zero vector rather than failing.  |
//+------------------------------------------------------------------+
int BuildFeatureVector(float &out[])
  {
   ArrayResize(out,NUM_FEATURES);

   if(Bars(_Symbol,_Period)<InpAtrPeriod+5)
     {
      ArrayInitialize(out,0.0f);
      return NUM_FEATURES;              // not enough history yet, e.g. at OnInit
     }

   double atr[];
   ArraySetAsSeries(atr,true);
   if(ExtAtrHandle==INVALID_HANDLE || CopyBuffer(ExtAtrHandle,0,0,1,atr)<=0)
     {
      ArrayInitialize(out,0.0f);
      return NUM_FEATURES;
     }

   MqlRates rates[];
   ArraySetAsSeries(rates,true);
   if(CopyRates(_Symbol,_Period,0,6,rates)<6)
     {
      ArrayInitialize(out,0.0f);
      return NUM_FEATURES;
     }

   //--- three lagged one-bar returns, most recent first
   double ret1 = (rates[0].close-rates[1].close)/rates[1].close;
   double ret2 = (rates[1].close-rates[2].close)/rates[2].close;
   double ret3 = (rates[2].close-rates[3].close)/rates[3].close;

   //--- candle-shape ratios, all normalized by the current close
   double range = (rates[0].high-rates[0].low)/rates[0].close;
   double body  = MathAbs(rates[0].close-rates[0].open)/rates[0].close;
   double atr_norm    = atr[0]/rates[0].close;
   double upper_wick  = (rates[0].high-MathMax(rates[0].open,rates[0].close))/rates[0].close;
   double lower_wick  = (MathMin(rates[0].open,rates[0].close)-rates[0].low)/rates[0].close;

   out[0]=(float)ret1;
   out[1]=(float)ret2;
   out[2]=(float)ret3;
   out[3]=(float)range;
   out[4]=(float)body;
   out[5]=(float)atr_norm;
   out[6]=(float)upper_wick;
   out[7]=(float)lower_wick;

   return NUM_FEATURES;
  }

В OnTick() обработка ограничена одним разом на каждый новый бар: каждый вызов, прошедший эту проверку, приводит к полному циклу сетевого обмена. Обращение к серверу на каждом тике (на активно торгуемой паре тики могут поступать десятки раз в секунду) расходовало бы допустимое время на задержки, о котором говорится в разделе "Результаты измерения задержки", и под нагрузкой гораздо чаще задействовало бы описанный выше механизм повторных попыток. OnTick() формирует вектор признаков, для подстраховки повторно проверяет число его элементов и либо вызывает Predict(), либо выполняет поиск в кэше — в зависимости от значения ExtUsingCache, однократно определенного в OnInit(). ManageSignal() разрешает сделку при условии probability >= InpProbThreshold и рассчитывает расстояния до стоп-лосса и тейк-профита по текущему ATR, а не задает их фиксированным числом пипсов. Волатильность XAUUSD слишком сильно различается между сессиями: стоп-лосс, подходящий для спокойной азиатской сессии, может оказаться слишком близким при новостном всплеске во время американской сессии.

OnTick():

//+------------------------------------------------------------------+
//| OnTick                                                           |
//| Rate-limited to once per new M5 bar -- every call below is a real|
//| network round trip, so calling on every tick would both waste the|
//| latency budget and make the client's retry/backoff far more      |
//| likely to trigger under load. Runtime feature width is checked   |
//| again here (not just at OnInit) in case history conditions change|
//| the vector shape mid-run.                                        |
//+------------------------------------------------------------------+
void OnTick()
  {
   datetime cur_bar_time=iTime(_Symbol,_Period,0);
   if(cur_bar_time==ExtLastBarTime)
      return;                             // still the same bar, nothing to do
   ExtLastBarTime=cur_bar_time;

   float features[];
   int width=BuildFeatureVector(features);
   if(width!=NUM_FEATURES)
     {
      PrintFormat("[FEATURE_CONTRACT_MISMATCH] Runtime width %d != NUM_FEATURES %d. Skipping tick.",width,NUM_FEATURES);
      return;
     }

   string req_id=StringFormat("%s-%I64d",_Symbol,(long)TimeCurrent());
   SInferenceResult result;

   bool ok=ExtClient.Predict(req_id,_Symbol,EnumToString(_Period),(long)TimeCurrent(),
                              features,InpModelVersion,result);

   if(!ok || !result.valid)
     {
      PrintFormat("[GRPC_ERROR] Inference call failed: %s",result.error);
      return;
     }

   if(result.model_version!=InpModelVersion)
      //--- non-fatal by design: the server is authoritative on which
      //--- model actually ran, this is a visibility log line only
      PrintFormat("[MODEL_VERSION_DRIFT] Server responded with model_version=%s, expected %s.",
                  result.model_version,InpModelVersion);

   ManageSignal(result);
  }

ManageSignal():

//+------------------------------------------------------------------+
//| ManageSignal                                                     |
//| Converts a validated InferenceResult into a threshold-gated,     |
//| ATR-sized market order. Single-position gating keeps the example |
//| easy to follow -- production use would likely add per-symbol     |
//| exposure limits and a cooldown after a losing trade.             |
//+------------------------------------------------------------------+
void ManageSignal(const SInferenceResult &result)
  {
   if(PositionSelect(_Symbol))
      return;                             // already in a position, skip this signal

   double atr[];
   ArraySetAsSeries(atr,true);
   if(CopyBuffer(ExtAtrHandle,0,0,1,atr)<=0)
      return;
   double atr_val=atr[0];
   if(atr_val<=0)
      return;                             // guard against a zero/invalid ATR read

   double price_ask=SymbolInfoDouble(_Symbol,SYMBOL_ASK);
   double price_bid=SymbolInfoDouble(_Symbol,SYMBOL_BID);

   if(result.class_label==1 && result.probability>=InpProbThreshold)
     {
      double sl=price_ask-InpAtrMultiplier*atr_val;
      double tp=price_ask+InpAtrMultiplier*atr_val;
      ExtTrade.Buy(InpLotSize,_Symbol,price_ask,sl,tp,"grpc-inference-long");
     }
   else if(result.class_label==0 && result.probability>=InpProbThreshold)
     {
      double sl=price_bid+InpAtrMultiplier*atr_val;
      double tp=price_bid-InpAtrMultiplier*atr_val;
      ExtTrade.Sell(InpLotSize,_Symbol,price_bid,sl,tp,"grpc-inference-short");
     }
  }

Ретранслятор на Python и сервер модели

Именно благодаря grpc_shim_server.py решение можно честно назвать "мостом gRPC", а не просто сокетным ретранслятором под новым названием. Прослойка считывает фреймы MetaTrader 5 с префиксом длины с помощью read_frame. Эта функция воспроизводит на Python формат фреймов клиента MQL5: struct.unpack(">I", header) задает порядок байтов big-endian, соответствующий побайтовым сдвигам в коде MQL5. Затем прослойка десериализует фреймы в объекты inference_pb2.InferenceRequest и передает их серверу модели по полноценному каналу grpc.aio. Это единственный участок во всей системе, где безоговорочно используется полноценный gRPC: формирование и обработку фреймов HTTP/2 полностью берет на себя библиотека grpc, а не написанный вручную код.

Исключение grpc.aio.AioRpcError (сервер модели недоступен, превышено время ожидания или возникла ошибка) перехватывается и преобразуется в обычный ответ с заполненным полем error. Поэтому клиенту MQL5 достаточно одной проверки result.error независимо от того, в каком слое произошел сбой. При запуске создается один канал, который затем используется для всех подключений MetaTrader 5, принимаемых прослойкой. Канал grpc.aio.insecure_channel рассчитан на безопасное одновременное использование, а его создание требует значительных ресурсов. Это также позволяет нескольким агентам тестера стратегий совместно использовать прослойку при многоагентной оптимизации, без дополнительного механизма пула соединений.

"""
grpc_shim_server.py

Bridges MT5's length-prefixed Protobuf-over-TCP framing (MQL5 has no HTTP/2
stack, so it cannot speak real gRPC) to a genuine grpc.aio channel talking to
model_server.py. This file is intentionally small (~60 lines of logic) and
does exactly one job: translate framing, nothing more -- it never inspects
or modifies message contents, only relays the raw serialized bytes.

Wire framing expected from MT5 (see CGrpcInferenceClient.mqh):
    [4 bytes big-endian length][serialized InferenceRequest protobuf bytes]
Response is framed identically with a serialized InferenceResponse.
"""
import asyncio
import logging
import struct

import grpc

import inference_pb2
import inference_pb2_grpc

SHIM_HOST = "127.0.0.1"
SHIM_PORT = 50151            # matches InpServerPort default in GrpcInferenceEA.mq5
MODEL_SERVER_TARGET = "127.0.0.1:50152"

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
log = logging.getLogger("grpc_shim")


async def read_frame(reader: asyncio.StreamReader) -> bytes:
    """Reads exactly one framed message: a 4-byte big-endian length prefix
    followed by that many payload bytes. readexactly() blocks until the
    full amount is available or raises IncompleteReadError on disconnect --
    this is the Python-side mirror of CGrpcInferenceClient::RecvExact() on
    the MQL5 side, and both sides agree on big-endian for this length field
    specifically (independent of the little-endian floats inside the
    Protobuf payload itself)."""
    header = await reader.readexactly(4)
    (length,) = struct.unpack(">I", header)
    if length > 16 * 1024 * 1024:
        # same 16 MB sanity bound enforced on the MQL5 side in RecvFramed();
        # keeping both bounds identical avoids one side accepting a frame
        # the other would have rejected.
        raise ValueError(f"refusing oversized frame: {length} bytes")
    return await reader.readexactly(length)


def write_frame(writer: asyncio.StreamWriter, payload: bytes) -> None:
    """Writes the mirror-image framing: big-endian length, then payload,
    in a single writer.write() call so the two pieces can never be
    interleaved with another coroutine's write on the same connection."""
    writer.write(struct.pack(">I", len(payload)) + payload)


async def handle_client(reader: asyncio.StreamReader, writer: asyncio.StreamWriter,
                         stub: inference_pb2_grpc.InferenceServiceStub) -> None:
    """One coroutine per connected MT5 client -- asyncio.start_server spins
    this up automatically per accepted socket, which is what lets multiple
    Strategy Tester agents connect to this shim concurrently during a
    multi-agent optimization run without any extra pooling logic here."""
    peer = writer.get_extra_info("peername")
    log.info("MT5 client connected: %s", peer)
    try:
        while True:
            try:
                raw = await read_frame(reader)
            except asyncio.IncompleteReadError:
                break  # client closed the socket

            # deserialize into a REAL Protobuf message object (not a dict,
            # not JSON) using the generated inference_pb2 classes -- this
            # is the same InferenceRequest type any other gRPC client
            # (Python, Go, Java) would construct.
            request = inference_pb2.InferenceRequest()
            request.ParseFromString(raw)

            try:
                # forwarded over a genuine grpc.aio channel -- this call
                # is real gRPC/HTTP2 end to end, the shim is just acting
                # as a client on MT5's behalf.
                response = await stub.Predict(request, timeout=2.0)
            except grpc.aio.AioRpcError as exc:
                # transport-level gRPC failure (model server down, timeout,
                # etc) gets turned into a normal InferenceResponse with the
                # error field set, so MQL5 only ever has to check one thing
                # (result.error) regardless of which layer actually failed.
                response = inference_pb2.InferenceResponse(
                    request_id=request.request_id, error=f"upstream gRPC error: {exc.code()}"
                )

            write_frame(writer, response.SerializeToString())
            await writer.drain()
    except Exception as exc:  # noqa: BLE001 -- log and drop this connection, keep shim alive
        log.exception("shim connection error: %s", exc)
    finally:
        writer.close()
        log.info("MT5 client disconnected: %s", peer)


async def main() -> None:
    # one shared grpc.aio channel/stub reused across all MT5 connections --
    # channels are safe for concurrent use and expensive to open, so this
    # is created once at startup rather than per client.
    channel = grpc.aio.insecure_channel(MODEL_SERVER_TARGET)
    stub = inference_pb2_grpc.InferenceServiceStub(channel)

    server = await asyncio.start_server(
        lambda r, w: handle_client(r, w, stub), SHIM_HOST, SHIM_PORT
    )
    log.info("grpc_shim_server listening on %s:%d (TCP framing -> gRPC relay)", SHIM_HOST, SHIM_PORT)
    async with server:
        await server.serve_forever()


if __name__ == "__main__":
    asyncio.run(main())

В model_server.py библиотека grpc.aio используется стандартным образом: серверу не требуется знать, что на другом конце находится MetaTrader 5. Он работал бы точно так же при обращении любого полноценного gRPC-клиента на любом языке. ToyLogisticModel намеренно проста: она вычисляет sigmoid(w·x+b) с вектором случайных весов, полученным при фиксированном начальном значении генератора. Это сделано специально, чтобы приведенные далее измерения задержки отражали накладные расходы передачи данных и сериализации — именно то, что важно при оценке моста, — а не определялись главным образом временем работы реальной модели, которую впоследствии подключат. Predict проверяет число элементов вектора признаков при каждом запросе, а не ограничивается однократной проверкой. Сервер не может заранее знать, собран ли подключившийся клиент с совместимой версией контракта: устаревшая сборка клиента на языке MQL5 может подключиться в любой момент.

В latency_ms записывается время, измеренное с помощью time.perf_counter() от начала функции. Это только время обработки на сервере, без передачи по сети. Поэтому для оценки самого моста важны приведенные далее отдельные измерения задержки полного цикла обмена. PredictStream намеренно вызывает Predict, а не дублирует его логику, поэтому проверка соответствия вектора признаков контракту выполняется одинаково и при потоковых вызовах. Потоковый RPC уже полностью работает на стороне gRPC, хотя клиент MQL5 пока использует только унарный вызов. Значит, для последующей отправки данных нескольких баров в одном потоковом соединении достаточно будет изменить только клиент.

"""
model_server.py

Real grpc.aio InferenceService implementation -- streaming, interceptors, and
health checks all operate against a genuine HTTP/2 gRPC server here. MT5 never
talks to this process directly; it talks to grpc_shim_server.py, which relays
into this server over a local gRPC channel. Everything in this file is
standard grpc.aio usage with zero awareness that MT5 exists on the other end
of the bridge -- it would work identically when called from any real gRPC client.

Generate inference_pb2.py / inference_pb2_grpc.py first with:
    python -m grpc_tools.protoc -I. --python_out=. --grpc_python_out=. inference.proto
"""
import asyncio
import logging
import time

import grpc
import numpy as np

import inference_pb2
import inference_pb2_grpc

# MUST mirror GrpcInferenceEA.mq5's NUM_FEATURES #define and the width
# ToyLogisticModel below is initialized with. A mismatch here is caught
# per-request in Predict() rather than trusted blindly.
NUM_FEATURES = 8
MODEL_VERSION = "v1"

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
log = logging.getLogger("model_server")


class ToyLogisticModel:
    """Placeholder model -- swap for an ONNX Runtime session or a joblib-
    loaded scikit-learn model in production. The weight vector's shape is
    tied to NUM_FEATURES on construction specifically so the mismatch guard
    in InferenceServicer.Predict() below has a real, dimensioned array to
    fail against rather than an implicit assumption."""

    def __init__(self, num_features: int):
        rng = np.random.default_rng(seed=42)  # fixed seed: reproducible demo
        self.weights = rng.normal(0, 0.5, size=num_features)
        self.bias = 0.0

    def predict_proba(self, x: np.ndarray) -> float:
        # plain logistic regression: sigmoid(w·x+b). This is the entire
        # "model" -- deliberately trivial so the article's numbers measure
        # transport/serialization overhead, not model complexity.
        z = float(np.dot(self.weights, x) + self.bias)
        return 1.0 / (1.0 + np.exp(-z))


class InferenceServicer(inference_pb2_grpc.InferenceServiceServicer):
    """Implements both RPCs declared in inference.proto. Predict() is the
    unary call GrpcInferenceEA.mq5 exercises through the shim on every bar;
    PredictStream() exists to demonstrate the streaming path is genuinely
    live on the gRPC side, even though the current MQL5 client only calls
    the unary RPC (see the article's conclusion for the natural extension
    to a real bidirectional streaming client)."""

    def __init__(self):
        self.model = ToyLogisticModel(NUM_FEATURES)

    async def Predict(self, request: inference_pb2.InferenceRequest, context) -> inference_pb2.InferenceResponse:
        t0 = time.perf_counter()
        resp = inference_pb2.InferenceResponse(request_id=request.request_id, model_version=MODEL_VERSION)

        # feature contract check: mirrors the compile-time NUM_FEATURES
        # check in GrpcInferenceEA.mq5's OnInit(), but here it runs on
        # EVERY request rather than once at startup, because this process
        # cannot assume the client that connected was built against the
        # same contract version.
        values = list(request.features.values)
        if len(values) != NUM_FEATURES:
            resp.error = f"feature width mismatch: got {len(values)}, expected {NUM_FEATURES}"
            log.error("[FEATURE_CONTRACT_MISMATCH] req=%s %s", request.request_id, resp.error)
            return resp

        x = np.asarray(values, dtype=np.float32)
        proba = self.model.predict_proba(x)

        resp.probability = float(proba)
        resp.class_label = 1 if proba >= 0.5 else 0
        resp.latency_ms = (time.perf_counter() - t0) * 1000.0  # server-side only
        return resp

    async def PredictStream(self, request_iterator, context):
        # bidirectional streaming RPC: consumes an async iterator of
        # requests and yields a response per request, reusing Predict()
        # so both paths share identical feature-contract and inference
        # logic rather than duplicating it.
        async for request in request_iterator:
            yield await self.Predict(request, context)


async def serve(host: str = "127.0.0.1", port: int = 50152):
    server = grpc.aio.server()
    inference_pb2_grpc.add_InferenceServiceServicer_to_server(InferenceServicer(), server)
    server.add_insecure_port(f"{host}:{port}")
    log.info("model_server listening on %s:%d (real gRPC/HTTP2)", host, port)
    await server.start()
    await server.wait_for_termination()


if __name__ == "__main__":
    asyncio.run(serve())

Согласованность и тестирование

verify_consistency.py выполняет предварительную проверку, в том числе в CI: с помощью регулярных выражений извлекает NUM_FEATURES из кода советника на языке MQL5 и из model_server.py и при несовпадении завершается с ненулевым кодом. Скрипт выявляет те же рассогласования, что и проверки во время выполнения, но уже через несколько секунд на подготовительном этапе, а не через три часа после начала оптимизации в тестере стратегий.

"""
verify_consistency.py

Guards against silent feature-contract drift: NUM_FEATURES in the MQL5 EA and
model_server.py must agree. Run before every backtest/deployment -- a static
regex check rather than an import, so it works without a compiled MetaEditor
toolchain and can run in a plain CI job alongside the Python test suite.
"""
import re
import sys
from pathlib import Path

EA_PATH = Path("MQL5/Experts/GrpcMT5Bridge/GrpcInferenceEA.mq5")
MODEL_SERVER_PATH = Path("model_server.py")


def extract_mql5_num_features(path: Path) -> int:
    text = path.read_text()
    m = re.search(r"#define\s+NUM_FEATURES\s+(\d+)", text)
    if not m:
        raise ValueError(f"NUM_FEATURES #define not found in {path}")
    return int(m.group(1))


def extract_python_num_features(path: Path) -> int:
    text = path.read_text()
    m = re.search(r"NUM_FEATURES\s*=\s*(\d+)", text)
    if not m:
        raise ValueError(f"NUM_FEATURES assignment not found in {path}")
    return int(m.group(1))


def main() -> int:
    mql5_n = extract_mql5_num_features(EA_PATH)
    python_n = extract_python_num_features(MODEL_SERVER_PATH)

    if mql5_n != python_n:
        print(f"[MISMATCH] EA NUM_FEATURES={mql5_n} != model_server NUM_FEATURES={python_n}")
        return 1

    print(f"[OK] Feature contract consistent: NUM_FEATURES={mql5_n} across MQL5 and Python.")
    return 0


if __name__ == "__main__":
    sys.exit(main())

Встроенные сокеты языка MQL5 не работают в тестере стратегий: вызовы SocketCreate()/SocketConnect() возвращают ошибку 4014 (ERR_FUNCTION_NOT_ALLOWED) независимо от разрешений терминала, даже если тот же советник с теми же настройками без проблем подключается при работе на графике реального или демо-счета. Это документированное ограничение платформы MetaTrader, о котором сообщают как минимум с 2019 года и которое сохраняется в текущих сборках. Его нельзя устранить изменением клиентского кода; дело не в трудноуловимых временных рассогласованиях.

InferenceCache.mqh — запасной вариант, позволяющий провести полноценное тестирование на исторических данных, а не подменять его вымышленными результатами. В тестере он воспроизводит заранее записанные результаты работы модели, сопоставляя их со временем баров. При торговле на реальном или демо-счете по-прежнему используется описанный выше клиент с сокетным соединением. Кэш содержит реальные, а не синтетические данные: build_inference_cache.py получает исторические бары через официальный пакет MetaTrader 5 для Python и заново рассчитывает те же восемь признаков, которые BuildFeatureVector() формирует в коде MQL5 (включая самостоятельно реализованный ATR Уайлдера со сглаживанием, в точности соответствующим iATR()). Затем скрипт обращается к model_server.py один раз на каждый бар — используются тот же сервер, те же веса и тот же код, которые описаны выше.

Load() считывает полученный CSV из общей папки с использованием FILE_COMMON: у каждого агента тестера стратегий своя изолированная папка Files, тогда как общая папка от этого ограничения свободна. Lookup() выполняет двоичный поиск среди загруженных записей по времени бара, начиная с последнего найденного совпадения, поскольку при проходе тестера вызовы OnTick() поступают в хронологическом порядке. С помощью MQLInfoInteger(MQL_TESTER) OnInit() однократно определяет, будет ли советник устанавливать соединение через сокет или загружать кэш. Логика исполнения сделок в ManageSignal() в обоих случаях остается прежней — меняется только источник результата.

Load():

   //+------------------------------------------------------------------+
   //| Load                                                             |
   //| Reads the cache CSV from the shared FILE_COMMON location, since  |
   //| the Strategy Tester runs each agent with its own isolated Files  |
   //| folder -- FILE_COMMON points to the terminal-wide shared folder, |
   //| reachable from every Tester agent regardless of which one runs   |
   //| this pass. Returns false (and logs why) on any failure so an     |
   //| OnInit() caller can fail loudly rather than silently trade blind.|
   //+------------------------------------------------------------------+
   bool              Load(const string relative_path)
     {
      ResetLastError();
      int handle=FileOpen(relative_path,FILE_READ|FILE_TXT|FILE_ANSI|FILE_COMMON);
      if(handle==INVALID_HANDLE)
        {
         PrintFormat("[CACHE_LOAD_FAILED] Could not open '%s' in Common\\Files, GetLastError=%d. Run build_inference_cache.py first.",
                     relative_path,GetLastError());
         return false;
        }

      ArrayResize(m_entries,0);
      m_count=0;
      bool header_skipped=false;

      while(!FileIsEnding(handle))
        {
         string line=FileReadString(handle);
         if(StringLen(line)==0)
            continue;
         if(!header_skipped)
           {
            header_skipped=true;   // first line is the column header, not data
            continue;
           }
         SCacheEntry entry;
         if(!ParseLine(line,entry))
            continue;              // skip any malformed row rather than aborting the whole load
         ArrayResize(m_entries,m_count+1);
         m_entries[m_count]=entry;
         m_count++;
        }
      FileClose(handle);

      if(m_count==0)
        {
         PrintFormat("[CACHE_LOAD_FAILED] '%s' loaded but contained zero valid rows.",relative_path);
         return false;
        }

      PrintFormat("[CACHE_LOADED] %d entries from '%s', range %s to %s.",
                  m_count,relative_path,
                  TimeToString(m_entries[0].bar_time,TIME_DATE|TIME_MINUTES),
                  TimeToString(m_entries[m_count-1].bar_time,TIME_DATE|TIME_MINUTES));
      return true;
     }

Lookup():

   //+------------------------------------------------------------------+
   //| Lookup                                                           |
   //| Binary search for the exact bar_time. m_last_hit_idx is used as a|
   //| starting hint since OnTick calls arrive in chronological order |
   //| during a Tester pass, making the common case near-O(1) instead of|
   //| a full O(log n) search on every single call.                     |
   //+------------------------------------------------------------------+
   bool              Lookup(const datetime bar_time,SInferenceResult &result)
     {
      int lo=0,hi=m_count-1;
      if(m_last_hit_idx>=0 && m_last_hit_idx<m_count && m_entries[m_last_hit_idx].bar_time<=bar_time)
         lo=m_last_hit_idx;        // resume near the last position instead of from zero

      while(lo<=hi)
        {
         int mid=(lo+hi)/2;
         if(m_entries[mid].bar_time==bar_time)
           {
            m_last_hit_idx=mid;
            result.request_id="cache";
            result.probability=m_entries[mid].probability;
            result.class_label=m_entries[mid].class_label;
            result.latency_ms=0.0f;      // no real call made; not meaningful for a cache hit
            result.model_version=m_entries[mid].model_version;
            result.error="";
            result.valid=true;
            return true;
           }
         if(m_entries[mid].bar_time<bar_time)
            lo=mid+1;
         else
            hi=mid-1;
        }

      result.valid=false;
      result.error=StringFormat("no cache entry for bar_time=%s",TimeToString(bar_time,TIME_DATE|TIME_MINUTES));
      return false;
     }
  };

Тестер стратегий: фактическая кривая баланса советника с инференсом через мост gRPC (XAUUSD M5, 2026.01-2026.08)

Рис. 3. Фактическая кривая баланса по закрытым сделкам, полученная в описанном ниже проходе тестера стратегий: 2 176 сделок с последовательным воспроизведением записанных результатов модели на каждом баре примерно за восемь месяцев истории XAUUSD на таймфрейме M5.

Настройка	Значение
Символ / таймфрейм	XAUUSD / M5
Период тестирования	2026.01.01-2026.08.25 (реальные исторические тики)
Модель исполнения	Каждый тик на основе реальных тиков
Порог вероятности	0,50
Множитель ATR / период	2,0 / 14
Размер лота	0,1 (фиксированный)
Источник результатов инференса	InferenceCache.mqh, воспроизведение результатов работы build_inference_cache.py

Фактические показатели: 2 176 сделок, все на покупку; доля прибыльных сделок — 49,03%, итоговая чистая прибыль — −1 315,20 долл. США при начальном депозите 10 000 долл. США (профит фактор — 0,99, коэффициент Шарпа — −0,26, максимальная просадка баланса — 74,28%). Этот результат ожидаем. ToyLogisticModel инициализирована случайным образом и не обучена, поэтому на реальных ценовых данных закономерны результаты, близкие к случайному угадыванию, и профит фактор около единицы, то есть на уровне безубыточности. Модель не дает полезного прогнозного сигнала, и тестирование на исторических данных корректно это показывает, а не скрывает. То, что все сделки оказались покупками, объясняется конкретным случайным вектором весов: на этих данных рассчитанная вероятность чаще оказывается по одну сторону порога 0,5. Это не свидетельство преобладания какого-либо направления движения XAUUSD — другое начальное значение генератора случайных чисел вполне могло бы дать иное соотношение сделок. Следующий естественный шаг — подключить должным образом обученную модель, оставив остальные компоненты моста без изменений, и тем самым подготовить решение к практическому использованию.

Результаты измерения задержки

Методика: одна и та же логика ToyLogisticModel была доступна тремя способами — через REST-эндпоинт на Flask, возвращающий JSON, напрямую через сокет ZeroMQ REQ/REP с обменом вручную сформированными JSON-строками и через описанный в статье мост Protobuf-over-TCP. Измерения выполнялись клиентом на Python, использующим тот же формат сообщений с префиксом длины, что и CGrpcInferenceClient.mqh. Поэтому измеренная задержка отражает условия работы советника MQL5, а не наилучший случай прямого обращения к модели. Для каждого варианта выполнили по 1 000 последовательных запросов через петлевой интерфейс (loopback). Это позволяет оценить накладные расходы передачи данных и сериализации без влияния колебаний задержки в реальной сети. Такое сравнение корректно, поскольку именно эти расходы и рассматриваются в статье.

Задержка полного цикла обмена: REST, ZeroMQ и мост gRPC (измерения по 1 000 запросам)

Рис. 4. Измеренные средняя задержка полного цикла обмена и ее 95-й процентиль (P95) по 1 000 запросам для каждого варианта через петлевой интерфейс: REST — 16,12 мс / 31,20 мс; ZeroMQ — 0,98 мс / 1,15 мс; мост gRPC — 3,20 мс / 3,24 мс.

Как и ожидалось, REST оказался самым медленным: полный цикл HTTP-запроса и ответа дополняется кодированием и декодированием JSON на обеих сторонах. Неожиданным оказалось то, что прямой обмен через ZeroMQ значительно опередил мост gRPC: средняя задержка составила 0,98 мс против 3,20 мс. Это не означает, что кодирование Protobuf медленнее кодирования JSON: причина в архитектуре. В этих тестах производительности ZeroMQ и REST используется одно звено связи: клиент на Python обращается напрямую к процессу, в котором работает модель. При измерении задержки моста задействованы два звена связи — так устроена его архитектура. Запрос клиента в TCP-фрейме поступает в grpc_shim_server.py, который выполняет отдельный вызов grpc.aio к model_server.py, дожидается ответа и лишь затем отвечает клиенту. Именно эта роль прослойки описана в разделе "Архитектура и схема".

3,20 мс — фактические постоянные затраты на второе звено связи. Взамен мы получаем поддержку потоковой передачи, полноценный сервис grpc.aio с возможностью проверки работоспособности и сам протокол, а не просто его название. При этом задержка по-прежнему на порядок ниже, чем у REST, и с большим запасом укладывается во время пятиминутного бара советника на M5. Эти измерения выполнены через петлевой интерфейс. При размещении сервера инференса на удаленном узле основную долю задержки составляло бы время прохождения запроса и ответа по сети (RTT), поэтому относительные различия между вариантами сократились бы. Именно поэтому эта архитектура наиболее оправданна в условиях, для которых проводились измерения: при размещении сервера инференса на том же компьютере или в локальной сети — реалистичном варианте развертывания для чувствительного к задержкам советника на M5.

Граничные случаи, подводные камни и заключение

Предположение о порядке байтов little-endian для значений float. Переинтерпретация байтов в UFloatBytes корректна только на платформах x86/x64, где порядок байтов соответствует требованию формата передачи данных: little-endian для полей fixed32. На платформах с порядком байтов big-endian эта реализация не проверялась, поэтому при переносе кода ее следует проверить заново, а не предполагать, что она будет работать правильно.

Сокеты не работают в тестере стратегий (ошибка 4014). Это документированное ограничение платформы MetaTrader, а не проблема с настройками. Практическое решение описано выше в разделе об InferenceCache.mqh.

Защита от некорректного varint. ReadVarint ограничивает величину сдвига 63 битами, чтобы поврежденный или усеченный фрейм не приводил к бесконечному циклу разбора в ожидании сброса бита продолжения.

Ограничение размера фрейма — 16 МБ. Это ограничение одинаково применяется и в клиенте MQL5, и в прослойке, чтобы поврежденный префикс длины не привел на любой из сторон к попытке выделить чрезмерный объем памяти или к зависанию в ожидании байтов, которые никогда не поступят.

От двух настроек терминала зависит работа не только WebRequest, но и сокетов. Настройки "Allow algorithmic trading" и "Allow WebRequest for listed URL" (с добавленным в список адресом узла прослойки) распространяются и на прямые вызовы SocketCreate() / SocketConnect(), а не только на WebRequest(). Это легко упустить из виду: ни в одном из названий флажков сокеты не упоминаются.

Первый вызов по новому каналу gRPC может не уложиться в слишком короткий таймаут клиента. При первом RPC по каналу между прослойкой и сервером модели требуется дополнительное время на установление соединения TCP+HTTP/2. Задайте InpSocketTimeoutMs с достаточным запасом (несколько секунд), иначе клиент может отключиться, не дождавшись первого ответа. Повторные попытки при этом могут привести к накоплению нескольких соединений, ожидающих готовности одного и того же канала.

Работа системы зависит и от прослойки. В однопроцессном решении на REST или ZeroMQ этого дополнительного компонента нет. Его введение — осознанный компромисс ради корректной работы и полноценной семантики gRPC на стороне Python, а не бесплатное преимущество. Это следует учитывать при планировании развертывания и мониторинга.

На самом деле предложенное в статье решение устраняет рассогласование схемы и неоднозначность сериализации при обмене между MetaTrader 5 и Python. Файл .proto задает обязательный контракт между программами на двух языках: без него у них нет общего определения даже для "вектора признаков". А двоичный формат фреймов позволяет существенно сократить накладные расходы на каждый вызов по сравнению с передачей JSON по HTTP.

По сути, эта архитектура представляет собой два моста, а не единую систему без промежуточных границ. Это стоит обозначить прямо, без преувеличений: В коде MQL5 вручную реализован формат передачи данных Protobuf и используется разработанный для этой статьи формат TCP-фреймов с префиксом длины. Прослойка на Python преобразует такой обмен в полноценные вызовы grpc.aio. Сам протокол gRPC используется только на участке между прослойкой и сервером модели, но не на стороне клиента MQL5. Столь же прямо нужно говорить и об ограничениях тестера стратегий: сокеты там не работают из-за ограничений платформы, а не ошибки в коде. Поэтому InferenceCache.mqh воспроизводит заранее записанные фактические результаты модели, а не создает видимость обращения к серверу там, где оно невозможно.

Измерения показывают следующий компромисс: мост на порядок быстрее REST, но, вопреки первоначальным ожиданиям, медленнее прямого обмена через ZeroMQ. Дополнительное звено связи, создаваемое прослойкой, вносит архитектурные накладные расходы; это не признак неэффективности самого кодирования Protobuf. Взамен мы получаем полноценный сервис gRPC на стороне Python: потоковую передачу, проверки работоспособности и формат сообщений с жестким соблюдением схемы, которых нет в самописном решении с JSON поверх ZeroMQ, — без имитации и подмены названий.

Логичные следующие шаги: использование TLS на участке между прослойкой и сервером модели при соединениях за пределами localhost, доработка клиента MQL5 для прямого использования PredictStream вместо повторного подключения на каждом баре и замена демонстрационной логистической модели должным образом обученной и проверенной моделью — остальные компоненты моста в любом случае остаются без изменений.

Файл	Тип	Описание
ProtobufWire.mqh	Подключаемый файл	Реализованный вручную модуль записи и чтения Protobuf: varint, теги, упакованные значения float, вложенные сообщения
GrpcInferenceClient.mqh	Подключаемый файл	TCP-клиент для обмена сообщениями с префиксом длины, с повторными попытками, задержками между ними и разбором ответов
GrpcInferenceEA.mq5	Советник	Основной советник: проверка соответствия признаков контракту, вызовы инференса на каждом баре, исполнение сделок при достижении порога вероятности
inference.proto	Схема	Контракт Protobuf, определяющий FeatureVector, InferenceRequest, InferenceResponse и сервис
grpc_shim_server.py	Python-скрипт	Ретранслятор между TCP-фреймами и вызовами grpc.aio, связывающий MetaTrader 5 с полноценным gRPC-сервером модели
model_server.py	Python-скрипт	Полноценная реализация InferenceService на базе grpc.aio с поддержкой унарных и потоковых RPC
verify_consistency.py	Python-скрипт	Статическая проверка совпадения значений NUM_FEATURES в советнике на языке MQL5 и сервере модели на Python
InferenceCache.mqh	Подключаемый файл	Загружает заранее записанный кэш из CSV-файла и выполняет в нем двоичный поиск; используется только в тестере стратегий, где сокеты не работают
build_inference_cache.py	Python-скрипт	Скрипт предварительного создания кэша: получает реальные исторические бары, заново рассчитывает признаки советника и обращается к действующему серверу модели один раз на каждый бар
bench_rest_server.py	Python-скрипт	Базовый вариант на REST с той же моделью для сравнения задержек, представленного на рис. 4
bench_zmq_server.py	Python-скрипт	Базовый вариант с прямым обменом через ZeroMQ в формате JSON и той же моделью для сравнения задержек, представленного на рис. 4
run_latency_benchmark.py	Python-скрипт	Измеряет фактическую задержку полного цикла обмена для всех трех вариантов передачи данных; с его помощью получены значения, приведенные на рис. 4