The ZeroMQ Message Transfer Protocol in MQL5: Implementing the REQ/REP pattern
135
0
Contents
- Introduction
- Protocol Foundations and ZMTP Mechanics
- The REQ Pattern Abstraction
- Practical Application Architecture: The Markov-switching GARCH Pipeline
- Conclusion and Future Extensions
Introduction
Algorithmic traders using MetaTrader 5 often need to execute complex quantitative calculations. MQL5 sometimes lacks native implementations for advanced computations that are readily available in Python and R. This motivates bridging solutions that let MetaTrader 5 work with these external programs. One example of such a solution is message queuing libraries like ZeroMQ. Historically, connecting MetaTrader 5 to external environments via ZeroMQ required using external Dynamic Link Libraries (DLLs). DLLs introduce platform dependencies and deployment complexities. While modern MQL5 includes raw network sockets, they are mostly suited for connecting clients to remote hosts, which should then afford MetaTrader 5 programs the means to communicate directly with ZeroMQ servers without third-party binaries.
This article explores a native MQL5 implementation of the ZeroMQ Message Transfer Protocol (ZMTP), built on top of raw MQL5 sockets. Readers will learn how to send requests and receive responses from a ZeroMQ server using a standard request-reply messaging sequence implemented as the CZmqReqSocket class. As a demonstration, we present a complete end-to-end quantitative pipeline where an MQL5 script streams financial return series directly to a hybrid Python/R server running MS-GARCH estimation and receives real-time market regime probabilities in return.
To discard DLL dependencies and achieve native integration, we must inspect the mechanics of the ZeroMQ protocol implementation before diving into the details of MQL5 ZeroMQ clients, the nature of the server, and how they all work together.
Protocol Foundations and ZMTP Mechanics
For MQL5 sockets to communicate with ZeroMQ servers, they must follow specific rules. These rules cover connection initiation, message structure, and other requirements. Together, they form the ZeroMQ Message Transfer Protocol (ZMTP). The core utilities in Zmtp.mqh map directly to the ZMTP specification. On top of these utilities, the library builds higher-level abstractions that handle patterns of communication between programs. In this text, we focus on the request-reply pattern. This pattern is easy to understand because it mimics a standard client-server relationship: the client sends a request, and the server returns a reply. Here, our abstractions focus on the client side of that interaction, implemented as the CZmqReqSocket class in Zmtp.mqh.
The code in Zmtp.mqh is organized into three main components.
- ZMTP protocol constants and frame structures.
- The underlying transport manager (CZmtpTransport).
- The high-level socket pattern abstraction (CZmqReqSocket).
ZMTP protocol constants and frame structures
The library establishes two main enumerations: ENUM_ZMQ_SOCKET_TYPE, which lists the supported ZeroMQ socket patterns (note that not all have been implemented); and ENUM_ZMTP_SECURITY, which specifies the available security mechanisms, including unauthenticated NULL, cleartext PLAIN, and an unimplemented CURVE option.
//+------------------------------------------------------------------+ //| Socket-Type property this library presents to the ZMTP peer | //+------------------------------------------------------------------+ enum ENUM_ZMQ_SOCKET_TYPE { ZMQ_TYPE_REQ, ZMQ_TYPE_REP, ZMQ_TYPE_DEALER, ZMQ_TYPE_ROUTER, ZMQ_TYPE_PUB, ZMQ_TYPE_SUB, ZMQ_TYPE_PUSH, ZMQ_TYPE_PULL, ZMQ_TYPE_PAIR }; enum ENUM_ZMTP_SECURITY { ZMTP_SEC_NULL, // no authentication, no encryption ZMTP_SEC_PLAIN, // username/password, sent in clear text ZMTP_SEC_CURVE // NOT IMPLEMENTED - see note at bottom of file };
The helper function ZmqSocketTypeName maps each socket-type enum value to its standard ZeroMQ string. The string is used during the handshake to identify the endpoint type to the peer.
string ZmqSocketTypeName(ENUM_ZMQ_SOCKET_TYPE t) { switch(t) { case ZMQ_TYPE_REQ: return "REQ"; case ZMQ_TYPE_REP: return "REP"; case ZMQ_TYPE_DEALER: return "DEALER"; case ZMQ_TYPE_ROUTER: return "ROUTER"; case ZMQ_TYPE_PUB: return "PUB"; case ZMQ_TYPE_SUB: return "SUB"; case ZMQ_TYPE_PUSH: return "PUSH"; case ZMQ_TYPE_PULL: return "PULL"; case ZMQ_TYPE_PAIR: return "PAIR"; } return ""; }
The ZmtpFrame structure represents an individual message frame transmitted over or received from a ZMTP network connection. The structure uses boolean flags to track protocol states: the more flag indicates whether additional frames follow in a multi-part message, while the command flag differentiates control frames (used for protocol-level operations like greetings or metadata exchange) from standard application payload data. Finally, the frame's raw content is stored as a dynamic byte array, allowing it to accommodate variable-length payloads.
//+------------------------------------------------------------------+ //| One ZMTP frame as read off / written to a node | //+------------------------------------------------------------------+ struct ZmtpFrame { bool more; // MORE flag - another frame follows in this message bool command; // COMMAND flag - control frame (handshake), not user data uchar data[]; };
The underlying transport manager (CZmtpTransport)
The CZmtpTransport class is the low-level transport manager for native ZMTP communication in MQL5. It sits on top of raw TCP sockets and handles connection lifecycle, receive buffering, session negotiation, and frame serialization. It acts as the foundational transport interface upon which higher-level messaging patterns are constructed.
//+------------------------------------------------------------------+ //| CZmtpTransport | //| Low-level ZMTP 3.0 connection: raw socket, receive buffering, | //| greeting exchange, security handshake, frame read/write. | //| Every socket-pattern class (REQ, and later DEALER/PUB/SUB/etc.) | //| is built on top of one of these. | //+------------------------------------------------------------------+ class CZmtpTransport { private: int m_socket; uchar m_rxbuf[]; int m_rxlen; // valid bytes currently in m_rxbuf int m_rxpos; // read cursor into m_rxbuf int m_timeout_ms; ENUM_ZMTP_SECURITY m_security; string m_username; string m_password; bool EnsureBytes(int n); void PopBytes(int n, uchar &out[]); bool RawSend(const uchar &buf[], int len); bool SendGreeting(); bool RecvGreeting(string &peer_mechanism); bool DoPlainHandshake(); bool SendReady(ENUM_ZMQ_SOCKET_TYPE stype, string identity); bool RecvReady(); public: CZmtpTransport(); ~CZmtpTransport(); bool Connect(string host, int port, ENUM_ZMTP_SECURITY sec, int timeout_ms=5000); void Disconnect(); bool IsConnected(); void SetPlainCredentials(string user, string pass) { m_username=user; m_password=pass; } // Full ZMTP handshake: greeting -> (security mechanism) -> READY/READY. // Call once, immediately after Connect(). bool Handshake(ENUM_ZMQ_SOCKET_TYPE stype, string identity=""); bool SendFrame(const uchar &data[], bool more, bool command=false); bool SendCommand(string name, const uchar &body[]); bool RecvFrame(ZmtpFrame &frame); };
A ZMTP session over a raw TCP stream starts with a 64-byte greeting sequence exchanged by both peers right after connecting. This sequence identifies the protocol version, security mechanism, and peer type. Bytes 0 to 9 contain a fixed signature starting with 0xFF, followed by eight zero bytes and ending with 0x7F. The bytes 10 to 11 specify the ZMTP major and minor versions. Thereafter, the bytes 12 to 31 hold a 20-byte security mechanism identifier (like "NULL" or "PLAIN"), padded with null bytes (0x00). Byte 32 sets the peer type (0 for client, 1 for server), and bytes 33 to 63 are reserved zero bytes.
This sequence gives both endpoints the details they need to establish the session before sending normal messages. The SendGreeting() and RecvGreeting() methods construct and validate this packet.
bool CZmtpTransport::SendGreeting() { uchar g[]; ArrayResize(g, 64); ArrayFill(g, 0, 64, 0); g[0] = 0xFF; // signature start g[9] = 0x7F; // signature end g[10] = 3; // version-major = 3 (ZMTP 3.x) g[11] = 0; // version-minor string mech = "NULL"; if(m_security == ZMTP_SEC_PLAIN) mech = "PLAIN"; else if(m_security == ZMTP_SEC_CURVE) mech = "CURVE"; uchar mech_bytes[]; StringToCharArray(mech, mech_bytes, 0, StringLen(mech)); for(int i=0; i<ArraySize(mech_bytes) && i<20; i++) g[12+i] = mech_bytes[i]; g[32] = 0; // as-server: this library only ever connects as a client // g[33..63] filler, already zero return RawSend(g, 64); } bool CZmtpTransport::RecvGreeting(string &peer_mechanism) { if(!EnsureBytes(64)) return false; uchar g[]; PopBytes(64, g); if(g[0] != 0xFF || g[9] != 0x7F) { Print(__FUNCTION__," : Zmtp: peer sent an invalid greeting signature - is this really a ZMTP 3.x endpoint?"); return false; } int major = g[10]; if(major < 3) { Print(__FUNCTION__," : Zmtp: peer speaks ZMTP ", major, ".x - this library only supports ZMTP 3.x"); return false; } uchar mech_bytes[20]; for(int i=0; i<20; i++) mech_bytes[i] = g[12+i]; int mlen = 0; while(mlen < 20 && mech_bytes[mlen] != 0) mlen++; peer_mechanism = CharArrayToString(mech_bytes, 0, mlen); return true; }
After validating greetings, peers finish security handshakes and exchange metadata using ZMTP control commands. A command is encoded as a frame with the command flag bit (0x04) set. The payload contains key-value properties, such as the type of socket and an optional identifier. ZMTP specifies three security modes.
- No Authentication: Peers trust each other implicitly.
- PLAIN: Usernames and passwords are sent in plain text across the network.
- Encrypted: Uses encryption to secure traffic over public networks.
Neither mode 1 nor mode 2 is safe for public networks. The demonstration later in this text uses no authentication, as encryption support is not yet implemented.
Standard TCP is stream-oriented, so data can arrive in arbitrary chunks. The CZmtpTransport class handles buffer management internally using two methods. The EnsureBytes() method reads data from the socket via SocketIsReadable() and appends incoming bytes to an internal dynamic array (m_rxbuf[]) until at least `n` bytes are ready.
bool CZmtpTransport::EnsureBytes(int n) { uint start = GetTickCount(); while((m_rxlen - m_rxpos) < n) { if(!SocketIsConnected(m_socket)) { Print(__FUNCTION__," : Zmtp: connection dropped mid-receive"); return false; } uint avail = SocketIsReadable(m_socket); if(avail > 0) { uchar tmp[]; ArrayResize(tmp, avail); int got = SocketRead(m_socket, tmp, avail, 100); if(got > 0) { int need = m_rxlen + got; if(need > ArraySize(m_rxbuf)) ArrayResize(m_rxbuf, need); for(int i=0; i<got; i++) m_rxbuf[m_rxlen+i] = tmp[i]; m_rxlen += got; start = GetTickCount(); // reset timeout on forward progress } } else Sleep(1); } return true; }
The PopBytes() method extracts `n` bytes from m_rxbuf[] for parsing and compacts the buffer periodically to save memory.
void CZmtpTransport::PopBytes(int n, uchar &out[]) { ArrayResize(out, n); for(int i=0; i<n; i++) out[i] = m_rxbuf[m_rxpos+i]; m_rxpos += n; // compact the buffer once we've drained a meaningful chunk of it if(m_rxpos > 4096) { int remain = m_rxlen - m_rxpos; for(int i=0; i<remain; i++) m_rxbuf[i] = m_rxbuf[m_rxpos+i]; m_rxlen = remain; m_rxpos = 0; } }
The REQ Pattern Abstraction
The CZmqReqSocket class provides a high-level API built on CZmtpTransport. The ZeroMQ API works based on messages. Thereby enforcing the use of messaging patterns. In ZeroMQ, a request (REQ) socket requires two things: strict alternation and envelope delimitation.
- Strict alternation is when the socket must alternate between send and receive. Sending two requests in a row causes a state error.
- Envelope delimitation describes how ZeroMQ routes messages using envelopes. On every request, a REQ socket automatically prepends an empty delimiter frame (len = 0, MORE = true) before the payload frame.
//+------------------------------------------------------------------+ //| CZmqReqSocket - ZMQ REQ pattern over ZMTP. | //| | //| Enforces the strict send/recv/send/recv alternation that REQ | //| requires, and handles the empty delimiter-frame envelope that | //| REQ/REP sockets use automatically in real ZeroMQ. | //+------------------------------------------------------------------+ class CZmqReqSocket { private: CZmtpTransport m_t; bool m_awaiting_reply; public: CZmqReqSocket() { m_awaiting_reply = false; } bool Connect(string host, int port, ENUM_ZMTP_SECURITY sec=ZMTP_SEC_NULL, int timeout_ms=5000, string identity="", string plain_user="", string plain_pass=""); void Disconnect() { m_t.Disconnect(); } bool IsConnected() { return m_t.IsConnected(); } // Single-part string request. bool Send(const string &request); // Single-part binary request (e.g. a packed struct of doubles). bool SendRaw(const uchar &request[]); // Blocks for the matching reply (up to the connect-time timeout). bool Recv(string &reply); bool RecvRaw(uchar &reply[]); };
The CZmqReqSocket class exposes request/reply semantics directly over ZMTP. An internal variable, m_awaiting_reply, tracks state to prevent sending two requests back-to-back or receiving without sending first. The Connect() method sets up the TCP connection and completes the ZMTP handshake. If using PLAIN security, SetPlainCredentials() passes credentials before connecting.
bool Connect(string host, int port, ENUM_ZMTP_SECURITY sec=ZMTP_SEC_NULL, int timeout_ms=5000, string identity="", string plain_user="", string plain_pass="") { if(sec == ZMTP_SEC_PLAIN) m_t.SetPlainCredentials(plain_user, plain_pass); if(!m_t.Connect(host, port, sec, timeout_ms)) return false; return m_t.Handshake(ZMQ_TYPE_REQ, identity); }
The library then calls the handshake routine to exchange socket metadata, setting ZMQ_TYPE_REQ so the peer understands the expected messaging pattern. The Disconnect() and IsConnected() methods simply expose the underlying transport status. The Send() method takes strings, converts them into byte arrays, and calls SendRaw().
bool Send(const string &request) { uchar body[]; StringToCharArray(request, body, 0, StringLen(request)); return SendRaw(body); }
In SendRaw(), the method builds the REQ envelope by sending two ZMTP frames.
bool SendRaw(const uchar &request[]) { if(m_awaiting_reply) { Print(__FUNCTION__," : Zmq REQ: Send() called out of order - a reply is still pending"); return false; } uchar empty[]; ArrayResize(empty, 0); // REQ envelope: empty delimiter frame, then the request body if(!m_t.SendFrame(empty, true)) return false; if(!m_t.SendFrame(request, false)) return false; m_awaiting_reply = true; return true; }
[empty delimiter frame, MORE=1] → [request body frame, MORE=0]
First, a zero-length frame, marked to show another frame is coming, is sent. Then, the request body with `MORE` turned off is sent to finish the message. This delimiter handles internal routing, so the main application code only deals with the actual request payload. The Recv() and RecvRaw() methods handle the reply phase. The RecvRaw() routine checks m_awaiting_reply to ensure a request was sent, then calls RecvFrame() for the first frame. If communicating through an intermediary, extra routing frames may arrive before the empty delimiter. The code discards these extra frames using a while loop until it hits the empty delimiter.
bool Recv(string &reply) { uchar raw[]; if(!RecvRaw(raw)) return false; reply = CharArrayToString(raw, 0, ArraySize(raw)); return true; } bool RecvRaw(uchar &reply[]) { if(!m_awaiting_reply) { Print(__FUNCTION__," : Zmq REQ: Recv() called before Send()"); return false; } ZmtpFrame f; if(!m_t.RecvFrame(f)) return false; // If a ROUTER sits between us and the REP, extra identity/envelope // frames may precede the empty delimiter - skip anything non-empty. while(ArraySize(f.data) > 0 && f.more) { if(!m_t.RecvFrame(f)) return false; } ArrayResize(reply, 0); while(f.more) { if(!m_t.RecvFrame(f)) return false; int old = ArraySize(reply); int add = ArraySize(f.data); ArrayResize(reply, old+add); ArrayCopy(reply, f.data, old, 0, add); } m_awaiting_reply = false; return true; }
Once past the delimiter, the socket gathers response frames. Because a reply can span multiple frames, the code reads incoming frames as long as `f.more` is set and appends their bytes to `reply`. Once a frame arrives with `MORE=0`, the message is complete. The Recv() method converts the byte array back to an MQL5 string, resets `m_awaiting_reply` to `false`, and prepares the socket for the next request. This workflow underlines the role of the protocol and the messaging pattern defined on top of it. ZMTP handles wire-level details—connections, handshakes, frames, and flags—while the REQ socket enforces the higher-level request/reply procedures. In the next section, we will look at a practical demonstration of the CZmqReqSocket class.
Practical Application Architecture: The Markov-switching GARCH Pipeline
Financial return series frequently display dynamic variance patterns across different market regimes (for example, low-volatility trending markets versus high-volatility stress periods). Standard GARCH models assume static parameters over time, whereas Markov-switching GARCH (MS-GARCH) models allow parameters to transition dynamically across hidden Markov states.
If we wanted to apply such a model, we would have to implement MS-GARCH models in MQL5. But why should we when robust implementations already exist on other platforms? The most prominent being the R package MSGARCH. We can leverage this library in our own workflow with the help of ZeroMQ. We therefore present an end-to-end quantitative pipeline. MetaTrader 5 streams return data to an external ZeroMQ server running in Python, which invokes R's `MSGARCH` package, fits the model, and returns regime probability estimates. The pipeline consists of a Python server and an MQL5 script. The routines that constitute the MQL5 script can also work within an EA.
The Python Server Architecture
The `msgarch_server.py` application is a Python service that bridges ZeroMQ network requests with R’s econometric `MSGARCH` library to calculate Markov-switching GARCH states on return series. Using Python's Ryp library, it initializes an R environment directly inside Python to run R-native estimation functions while serving predictions over a standard JSON API. The lesser-known `ryp` module was favored because the better-known `rpy2` package does not work well on Microsoft Windows operating systems.
The application runs a synchronous ZeroMQ event loop listening on a REP (Reply) socket bound to TCP port 5555 by default. The internal logic relies on the RMSGarchEstimator class, which constructs and manages dynamic model specifications. To optimize throughput, estimators are lazily instantiated and stored in an in-memory dictionary cache keyed by their exact configuration parameters. This strategy avoids the overhead of repeatedly rebuilding R specification objects for recurring request signatures.
class RMSGarchEstimator: """ Wraps an R MSGARCH::CreateSpec object with support for: - arbitrary number of regimes (n_states) - a different variance model / distribution per regime - mixture (do.mix=TRUE) vs Markov-switching (do.mix=FALSE) - ML or MCMC estimation, with MCMC controls exposed fit_and_predict() can return either just the latest-bar summary or the full in-sample regime path so it can be overlaid on the original return series. """ def __init__(self, n_states=2, variance_models="sGARCH", distributions="std", do_mix=False, estimation_method="ML", mcmc_nburn=1000, mcmc_nmesh=2500, mcmc_nthin=1): """ n_states: number of regimes (>= 2) variance_models: single model name (broadcast to all states) or a list of length n_states, e.g. ["sGARCH", "eGARCH", "gjrGARCH"] Any MSGARCH-supported variance model is valid: sGARCH, eGARCH, gjrGARCH, tGARCH, fGARCH, ... distributions: single distribution name (broadcast) or a list of length n_states, e.g. "norm", "std" (Student-t), "ged", "snorm", ... do_mix: True for an iid mixture, False (default) for a Markov-switching model estimation_method: "ML" or "MCMC" mcmc_nburn/mcmc_nmesh/mcmc_nthin: only used when estimation_method == "MCMC" """ if n_states < 2: raise ValueError("n_states must be >= 2") self.n_states = n_states self.variance_models = self._broadcast(variance_models, n_states) self.distributions = self._broadcast(distributions, n_states) self.do_mix = bool(do_mix) self.method = estimation_method self.mcmc_ctr = {"nburn": mcmc_nburn, "nmesh": mcmc_nmesh, "nthin": mcmc_nthin} self._spec_var = f"spec_{uuid.uuid4().hex[:8]}" self._init_model() @staticmethod def _broadcast(val, n): """Allow a single string to be broadcast to all states, or an explicit list.""" if isinstance(val, str): return [val] * n val = list(val) if len(val) != n: raise ValueError(f"Expected {n} entries, got {len(val)}: {val}") return val def _init_model(self): var_vec = "c(" + ", ".join(f'"{v}"' for v in self.variance_models) + ")" dist_vec = "c(" + ", ".join(f'"{d}"' for d in self.distributions) + ")" do_mix_r = "TRUE" if self.do_mix else "FALSE" r(f''' {self._spec_var} <- CreateSpec( variance.spec = list(model = {var_vec}), distribution.spec = list(distribution = {dist_vec}), switch.spec = list(do.mix = {do_mix_r}) ) ''') print(f"[R Engine] Initialized {self.n_states}-state MS-GARCH " f"(variance={self.variance_models}, dist={self.distributions}, " f"do.mix={self.do_mix}, method={self.method})") @staticmethod def _r_matrix_or_none(varname): """Fetch an R object as a numpy array, or None if it's NULL / missing. ryp converts R NULL -> Python None on its own, so this is just a thin, explicitly-typed wrapper around to_py().""" val = to_py(varname) return val @staticmethod def _regime_label(rank, n_states): if n_states == 2: return ["CALM", "CRISIS"][rank] if n_states == 3: return ["CALM", "TRANSITION", "CRISIS"][rank] return f"REGIME_{rank}" # 0 = lowest average volatility ... n_states-1 = highest def fit_and_predict(self, returns, timestamps=None, include_series=False): """ Fits the model on `returns` and returns: - a latest-bar summary (state probabilities, dominant regime, etc.) - if include_series=True, the full in-sample regime path aligned to `returns` / `timestamps`, suitable for overlaying regimes on the original series. Regimes are consistently ordered by ascending average volatility (rank 0 = calmest ... rank n_states-1 = most volatile) rather than by R's arbitrary internal state index, since MSGARCH state labels are not guaranteed to be stable across fits. """ to_r(np.array(returns, dtype=np.float64), 'returns_vec') try: if self.method == "ML": r(f''' fit <- FitML(spec = {self._spec_var}, data = returns_vec) st <- State(fit) predprob <- tryCatch(st$PredProb, error = function(e) NULL) filtprob <- tryCatch(st$FiltProb, error = function(e) NULL) viterbi <- tryCatch(st$Viterbi, error = function(e) NULL) vol <- tryCatch(Volatility(fit), error = function(e) NULL) ''') elif self.method == "MCMC": r(f''' fit <- FitMCMC(spec = {self._spec_var}, data = returns_vec, ctr = list(nburn = {self.mcmc_ctr["nburn"]}, nmesh = {self.mcmc_ctr["nmesh"]}, nthin = {self.mcmc_ctr["nthin"]})) st <- State(fit) predprob <- tryCatch(st$PredProb, error = function(e) NULL) filtprob <- tryCatch(st$FiltProb, error = function(e) NULL) viterbi <- tryCatch(st$Viterbi, error = function(e) NULL) vol <- tryCatch(Volatility(fit), error = function(e) NULL) ''') else: return {"status": "error", "message": f"Unknown estimation_method: {self.method}"} filt_probs = self._r_matrix_or_none('filtprob') pred_probs = self._r_matrix_or_none('predprob') viterbi = self._r_matrix_or_none('viterbi') vol = self._r_matrix_or_none('vol') # Prefer filtered/smoothed probabilities for in-sample regime labeling; # fall back to predicted probabilities if FiltProb isn't available. state_probs = filt_probs if filt_probs is not None else pred_probs if state_probs is None: return {"status": "error", "message": "MSGARCH did not return state probabilities."} n_states = self.n_states if self.method == "MCMC": state_probs = np.mean(state_probs,axis=-2,keepdims=True) pred_probs = np.mean(pred_probs,axis=-2,keepdims=True) viterbi = np.mean(viterbi,axis=-1) viterbi = np.round(viterbi,0) # Order regimes by ascending average volatility across the whole fit, # not just the last bar, so the calm/crisis labeling is stable. if vol is not None and vol.ndim == 2 and vol.shape[1] == n_states: avg_vol = vol.mean(axis=0) rank_order = np.argsort(avg_vol) # rank_order[0] = calmest raw state index else: avg_vol = None rank_order = np.arange(n_states) raw_to_rank = {int(raw): rank for rank, raw in enumerate(rank_order)} latest_probs_ranked = np.reshape(state_probs[-1],state_probs.shape[-1]) dominant_state_raw = int(np.argmax(latest_probs_ranked)) dominant_rank = raw_to_rank[dominant_state_raw] summary = { "status": "success", "method": self.method, "n_states": n_states, "variance_models": self.variance_models, "distributions": self.distributions, "do_mix": self.do_mix, "state_probabilities": latest_probs_ranked.tolist(), # ordered calm -> turbulent "dominant_regime": dominant_rank, "regime_label": self._regime_label(dominant_rank, n_states), "state_volatility": (avg_vol[rank_order].tolist() if avg_vol is not None else None), } # Backward-compatible fields for existing 2-state consumers (e.g. CSignalGarchTouch.mqh) if n_states == 2: summary["p_calm"] = float(latest_probs_ranked[0]) summary["p_crisis"] = float(latest_probs_ranked[1]) summary["current_state"] = "CRISIS" if latest_probs_ranked[1] > 0.5 else "CALM" summary["vol_calm"] = summary["state_volatility"][0] if avg_vol is not None else None summary["vol_crisis"] = summary["state_volatility"][1] if avg_vol is not None else None if include_series: T = state_probs.shape[0] ranked_probs_series = state_probs.reshape((state_probs.shape[0],state_probs.shape[-1])) dominant_per_bar_raw = np.argmax(state_probs, axis=2) dominant_rank_per_bar = np.array([raw_to_rank[int(r_[0])] for r_ in dominant_per_bar_raw]) vol_series_ranked = None if vol is not None and vol.ndim == 2 and vol.shape[1] == n_states: vol_series_ranked = vol[:, rank_order] viterbi_rank_series = None if viterbi is not None: # MSGARCH's Viterbi path is 1-indexed R state labels viterbi_raw = np.array(viterbi).astype(int).flatten() - 1 viterbi_rank_series = np.array([raw_to_rank[int(r_)] for r_ in viterbi_raw]) returns_arr = np.array(returns, dtype=np.float64) series = [] for i in range(T): entry = { "index": i, "timestamp": timestamps[i] if timestamps is not None and i < len(timestamps) else None, "return": float(returns_arr[i]) if i < len(returns_arr) else None, "state_probs": ranked_probs_series[i].tolist(), "regime": int(dominant_rank_per_bar[i]), "regime_label": self._regime_label(int(dominant_rank_per_bar[i]), n_states), } if vol_series_ranked is not None: entry["state_vol"] = vol_series_ranked[i].tolist() if viterbi_rank_series is not None and i < len(viterbi_rank_series): entry["viterbi_regime"] = int(viterbi_rank_series[i]) series.append(entry) summary["series"] = series return summary except Exception as e: return {"status": "error", "message": str(e)}
When a prediction request arrives, the server passes the return series into R via `ryp` and executes maximum likelihood (`FitML`) or MCMC (`FitMCMC`) estimation. Once fitted, it extracts state probabilities, Viterbi regime paths, and regime volatility matrices. The application exposes a single ZeroMQ REP endpoint that expects JSON-encoded request strings and returns structured JSON responses. The API supports two distinct actions.
- The `configure` specification modifies the server's global default configuration at runtime. It accepts parameters that correspond to the inputs of MSGARCH's CreateSpec(). It validates and updates defaults for subsequent prediction calls that do not specify their own configurations. Parameter names map directly to parameters used in R's `MSGARCH` library.
- The `predict` specification ingests a container of observations (minimum 50) alongside optional parameters like `timestamps`, an `include_series` boolean flag, and a single-request `config` override object. If `include_series` is set to `true`, the server returns a bar-by-bar regime breakdown overlay; otherwise, it returns a concise summary containing the latest bar's state probabilities, dominant regime rank, and regime labels.
The MQL5 Client Architecture
The MS_GARCH_Regimes.mq5 script acts as an MQL5 client that requests real-time regime-switching models from msgarch_server.py and overlays the resulting volatility regimes directly onto a MetaTrader 5 chart. It communicates via the high-level CMSGarchClient wrapper class, which internally builds upon the low-level native ZMTP implementation.
//+------------------------------------------------------------------+ //| Client wrapping a single ZMQ REQ socket to the MSGARCH server. | //+------------------------------------------------------------------+ class CMSGarchClient { private: CZmqReqSocket m_req; int m_port; string m_endpoint; int m_timeout_ms; bool m_connected; //--- send a request string and block (up to m_timeout_ms) for the reply bool SendReceive(const string request_json, string &response_json); //--- build a JSON array literal like [1.0,2.0,-3.5] from a double[] string DoubleArrayToJson(const double &arr[]); //--- build a JSON array literal of quoted strings like ["sGARCH","eGARCH"] string StringArrayToJson(const string &arr[]) ; //--- parse the fields common to every successful "predict" response void ParseSummary(const CJsonValue *root, MSGarchPredictResult &result); //--- parse the optional "series" array into an MSGarchRegimeBar[] void ParseSeries(const CJsonValue *root, MSGarchRegimeBar &series[]); public: CMSGarchClient(const string endpoint = "tcp://127.0.0.1", const int port = 5555, const int timeout_ms = 5000) { m_port = port; m_endpoint = endpoint; m_timeout_ms = timeout_ms; m_connected = false; } ~CMSGarchClient(void) { if(m_connected) m_req.Disconnect(); } bool Connect(void); //--- {"action":"configure", ...}. Pass an empty array to leave variance/dist //--- at their broadcast default ("sGARCH" / "std") server-side. bool Configure(const int n_states, const string &variance_models[], const string &distributions[], const bool do_mix, const string estimation_method, string &error_message); //--- {"action":"predict", "returns":[...], "include_series": bool} //--- `returns` should already be whatever return series your model expects //--- (e.g. log returns), computed by the caller. bool Predict(const double &returns[], MSGarchPredictResult &result, const bool include_series, MSGarchRegimeBar &series[]); };
The wrapper lets an MQL5 EA or script send return data to a Python process running a Markov-switching GARCH model. It returns market regimes, state probabilities, and state-specific volatility in a structured format. Communication payloads are encoded in JSON, while Zmtp.mqh provides ZeroMQ socket functionality and JsonValue.mqh handles JSON parsing.
The CMSGarchClient class encapsulates client-side communication. Connection management is maintained via a single ZeroMQ REQ socket, configured with the server endpoint, port, timeout, and connection state. Connect() establishes the connection to the Python server, while the destructor automatically disconnects the socket when the client object is destroyed.
bool Connect(void) { m_connected = m_req.Connect(m_endpoint,m_port); if(!m_connected) Print(__FUNCTION__," : [MSGarchClient] connect() failed for ", m_endpoint); return m_connected; }
The SendReceive() method provides the basic request/response mechanism. It sends a JSON request through the REQ socket and repeatedly attempts to receive the server's response, sleeping between attempts until a reply arrives or the retry limit is exhausted.
bool SendReceive(const string request_json, string &response_json) { response_json = ""; if(!m_req.Send(request_json)) { Print(__FUNCTION__," : [MSGarchClient] send() failed for endpoint ", m_endpoint); return false; } uint retries = 100; while(!m_req.Recv(response_json) && retries) { --retries; Sleep(m_timeout_ms); } if(StringLen(response_json)<1) { Print(__FUNCTION__," : [MSGarchClient] recv() timed out / failed after ", m_timeout_ms, " ms"); return false; } //response_json = reply.getData(); return true; }
The JSON helper DoubleArrayToJson() converts a container of double type into a JSON numerical array, while StringArrayToJson() converts a string array into a quoted JSON array.
string DoubleArrayToJson(const double &arr[]) const { string out = "["; int n = ArraySize(arr); for(int i = 0; i < n; i++) { out += DoubleToString(arr[i], 10); if(i < n - 1) out += ","; } out += "]"; return out; } //--- build a JSON array literal of quoted strings like ["sGARCH","eGARCH"] string StringArrayToJson(const string &arr[]) const { string out = "["; int n = ArraySize(arr); for(int i = 0; i < n; i++) { out += "\"" + arr[i] + "\""; if(i < n - 1) out += ","; } out += "]"; return out; }
The Configure() method sends an `action="configure"` request specifying the number of states, variance models, distributions, model mixing settings, and estimation method. Optional variance-model and distribution arrays are only included when supplied, allowing the Python server to fall back to its own defaults.
bool Configure(const int n_states, const string &variance_models[], const string &distributions[], const bool do_mix, const string estimation_method, string &error_message) { string req = "{\"action\":\"configure\""; req += ",\"n_states\":" + IntegerToString(n_states); if(ArraySize(variance_models) > 0) req += ",\"variance_models\":" + StringArrayToJson(variance_models); if(ArraySize(distributions) > 0) req += ",\"distributions\":" + StringArrayToJson(distributions); req += ",\"do_mix\":" + (do_mix ? "true" : "false"); req += ",\"estimation_method\":\"" + estimation_method + "\""; req += "}"; string response_json; if(!SendReceive(req, response_json)) { error_message = "transport error (no reply from server)"; return false; } CJsonParser parser; CJsonValue *root = parser.Parse(response_json); if(root == NULL) { error_message = "could not parse server response"; return false; } bool ok = (root.GetString("status", "error") == "success"); error_message = ok ? "" : root.GetString("message", "unknown error"); delete root; return ok; }
The Predict() method accepts an array of returns calculated by the MQL5 application and sends them using an `action="predict"` JSON request. After receiving the response, it parses the JSON, using ParseSummary() to populate the latest prediction and ParseSeries() to populate the historical MSGarchRegimeBar[] array.
bool Predict(const double &returns[], MSGarchPredictResult &result, const bool include_series, MSGarchRegimeBar &series[]) { ArrayResize(result.state_probabilities, 0); ArrayResize(result.state_volatility, 0); ArrayResize(series, 0); result.success = false; string req = "{\"action\":\"predict\",\"returns\":" + DoubleArrayToJson(returns) + ",\"include_series\":" + (include_series ? "true" : "false") + "}"; string response_json; if(!SendReceive(req, response_json)) { result.error_message = "transport error (no reply from server)"; return false; } CJsonParser parser; CJsonValue *root = parser.Parse(response_json); if(root == NULL) { result.error_message = "could not parse server response"; return false; } bool ok = (root.GetString("status", "error") == "success"); if(!ok) { result.error_message = root.GetString("message", "unknown error"); delete root; return false; } ParseSummary(root, result); if(include_series) ParseSeries(root, series); result.success = true; delete root; return true; } void ParseSummary(const CJsonValue *root, MSGarchPredictResult &result) const { result.method = root.GetString("method", ""); result.n_states = (int)root.GetDouble("n_states", 0); result.dominant_regime = (int)root.GetDouble("dominant_regime", -1); result.regime_label = root.GetString("regime_label", ""); CJsonValue *probs = root.Get("state_probabilities"); if(probs != NULL) probs.ToDoubleArray(result.state_probabilities); CJsonValue *vols = root.Get("state_volatility"); if(vols != NULL && !vols.IsNull()) vols.ToDoubleArray(result.state_volatility); result.p_calm = root.GetDouble("p_calm", -1.0); result.p_crisis = root.GetDouble("p_crisis", -1.0); result.current_state = root.GetString("current_state", ""); } void ParseSeries(const CJsonValue *root, MSGarchRegimeBar &series[]) const { CJsonValue *arr = root.Get("series"); if(arr == NULL) { ArrayResize(series, 0); return; } int n = arr.Size(); ArrayResize(series, n); for(int i = 0; i < n; i++) { CJsonValue *bar = arr.At(i); series[i].index = (int)bar.GetDouble("index", i); series[i].bar_time = (datetime)(long)bar.GetDouble("timestamp", 0); series[i].ret = bar.GetDouble("return", 0.0); series[i].regime = (int)bar.GetDouble("regime", 0); series[i].regime_label = bar.GetString("regime_label", ""); CJsonValue *sp = bar.Get("state_probs"); if(sp != NULL) sp.ToDoubleArray(series[i].state_probs); CJsonValue *sv = bar.Get("state_vol"); if(sv != NULL && !sv.IsNull()) sv.ToDoubleArray(series[i].state_vol); CJsonValue *vit = bar.Get("viterbi_regime"); if(vit != NULL && !vit.IsNull()) { series[i].has_viterbi = true; series[i].viterbi_regime = vit.AsInt(); } else { series[i].has_viterbi = false; series[i].viterbi_regime = -1; } } }
Two structures define the data returned by the server. The MSGarchPredictResult structure represents the latest-bar prediction summary. It contains request success status, fitted model method, number of states, posterior probability of each regime, dominant regime, textual label, and volatility associated with each state. For two-state models, convenience fields make it easy for an EA to directly check the probability of a calm or crisis regime.
struct MSGarchPredictResult { bool success; string error_message; string method; int n_states; double state_probabilities[]; // ordered calm(0) -> turbulent(n-1) int dominant_regime; // rank of the highest-probability state string regime_label; double state_volatility[]; // same ordering as state_probabilities //--- convenience fields, only populated when n_states == 2 double p_calm; double p_crisis; string current_state; };
The MSGarchRegimeBar structure represents one observation in the optional historical regime series returned when `include_series=true`. Each bar stores its index, timestamp, return, filtered state probabilities, inferred regime and label, state-specific volatility, and optionally the Viterbi-decoded regime.
struct MSGarchRegimeBar { int index; datetime bar_time; // 0 if the server wasn't given timestamps double ret; double state_probs[]; int regime; string regime_label; double state_vol[]; bool has_viterbi; int viterbi_regime; };
Script Execution and Operational Flow
The script runs as a one-shot MQL5 program through three distinct stages.
- Data Collection: The script fetches a sample of closed bars (excluding the currently forming bar) to prevent repainting. It converts rates to chronological order and calculates log returns alongside matching bar timestamps.
- Server Integration: A CMSGarchClient instance is initialized. Calling Connect() establishes a native raw MQL5 socket connection that performs the 64-byte ZMTP greeting and `READY` handshake over the wire. The script then transmits a `configure` payload requesting an N-state model with standard GARCH variance and Student-t error distributions estimated via maximum likelihood. Next, Predict() sends log returns with `include_series = true`. The client parses the JSON response into MSGarchPredictResult and an array of MSGarchRegimeBar structs.
- Chart Rendering: Invoking DrawRegimeBands() colors each regime band using a dynamic gradient generated by RegimeColor(). The visualization remains on screen for 30 seconds before clearing rectangle objects via ClearRegimeObjects() and terminating.
void OnStart(void) { int bars = InpLookbackBars + 1; // +1 because returns need one extra prior close if(Bars(_Symbol, _Period) < bars) { Print("Not enough history: need ", bars, " bars."); return; } MqlRates rates[]; ArraySetAsSeries(rates, true); int copied = CopyRates(_Symbol, _Period, 1, bars, rates); // exclude the still-forming bar if(copied < bars) { Print("CopyRates only returned ", copied, " bars."); return; } ArraySetAsSeries(rates, false); // switch back to chronological order for the loop below double returns[]; datetime bar_times[]; int n_returns = copied - 1; ArrayResize(returns, n_returns); ArrayResize(bar_times, n_returns); double average = 0.0; for(int i = 0; i < n_returns; i++) { returns[i] = MathLog(rates[i + 1].close / rates[i].close); bar_times[i] = rates[i + 1].time; average += returns[i]; } //--- Demean to remove ARCH effects average /= double(n_returns); for(int i = 0; i < n_returns; returns[i] -= average, ++i); CMSGarchClient client(InpEndpoint, InpPort, InpTimeoutMs); if(!client.Connect()) { Print("Could not connect to MSGARCH server at ", InpEndpoint); return; } //--- 3-state model: sGARCH in every regime, Student-t innovations, Markov-switching string variance_models[3] = {"sGARCH", "sGARCH", "sGARCH"}; string distributions[3] = {"std", "std", "std"}; string err; if(!client.Configure(InpNStates, variance_models, distributions, false, "ML", err)) { Print("Configure failed: ", err); return; } MSGarchPredictResult result; MSGarchRegimeBar series[]; if(!client.Predict(returns, result, true, series)) { Print("Predict failed: ", result.error_message); return; } PrintFormat("MSGARCH: %d-state fit, dominant regime = %s, latest probs: %s", result.n_states, result.regime_label, DoubleArrayToString(result.state_probabilities)); DrawRegimeBands(series, bar_times); PrintFormat("Painted %d regime bars onto the chart.", ArraySize(series)); Sleep(30000); ClearRegimeObjects(); ChartRedraw();
To run this pipeline, start the Python server first. Before invoking the script in MetaTrader 5, add the server's IP address and port to the list of allowed WebRequest/socket hosts in the MetaTrader 5 options menu. The graphic below captures the sequence of events as the pipeline is executed.
Conclusion and Future Extensions
The Zmtp.mqh library provides a simple way to connect MetaTrader 5 with external analytical tools using native MQL5 sockets. Implementation of ZMTP framing directly removes the need for DLL wrappers and external dependencies. This gives developers a cleaner way to build MQL5 applications that communicate with external systems. The current implementation focuses on the REQ/REP messaging pattern. This is sufficient for client-server workflows, as demonstrated in the text. This pattern works well as a type of remote procedure call as well as general task distribution.
Looking towards the future, the CZmtpTransport class provides the foundation for supporting additional ZeroMQ socket patterns. For example, PUSH/PULL could support unidirectional data streams such as continuous tick or trade telemetry. PUB/SUB could provide topic-based market data distribution to multiple services. DEALER/ROUTER could support more flexible asynchronous and non-blocking message routing.
These extensions would allow native MQL5 applications to move beyond simple request-response communication. They could become part of larger distributed systems in which MetaTrader 5 handles market data and execution while external services perform tasks as part of a broader pipeline. The result is a flexible foundation for building distributed quantitative trading architectures without requiring DLL-based networking. All code referenced in the article is listed below and available on AlgoForge.
Note that the Python server requires a standard R installation with the MSGARCH package.
| MQL5/Files/ZmqRequestSocket/requirements.txt: | These are the dependencies required for the Python application ms_garch_server.py. |
| MQL5/Files/ZmqRequestSocket/Python/ms_garch_server.py: | The ZeroMQ-based REP server fitting MS-GARCH regimes. |
| MQL5/Include/ZmqRequestSocket/JsonValue.mqh: | Minimal JSON parser used in MSGarchClient.mqh. |
| MQL5/Include/ZmqRequestSocket/Zmtp.mqh: | Header for ZeroMQ Message Transfer Protocol implementation. |
| MQL5/Include/ZmqRequestSocket/MSGarchClient.mqh: | This header contains utilities used by clients connecting to ms_garch_server.py, used in the MS_GARCH_Regimes script. |
| MQL5/Scripts/ZmqRequestSocket/MS_GARCH_Regimes.mq5: | An example script that connects to ms_garch_server.py to fit MS-GARCH models and paints the regimes on the chart. |
Warning: All rights to these materials are reserved by MetaQuotes Ltd. Copying or reprinting of these materials in whole or in part is prohibited.
This article was written by a user of the site and reflects their personal views. MetaQuotes Ltd is not responsible for the accuracy of the information presented, nor for any consequences resulting from the use of the solutions, strategies or recommendations described.
Neural Networks in Trading: The Temporal Query Model (Conclusion)
Native Isolation Forest for Execution-Quality Anomaly Detection in MQL5
Defining your Edge (Part 5): Using GARCH Variance and Volatility-Scaled LSTM in an Expert Advisor
The MQL5 Standard Library Explorer (Part 16): Building a Regime-Adaptive Expert Advisor
- Free trading apps
- Over 8,000 signals for copying
- Economic news for exploring financial markets
You agree to website policy and terms of use