[Python] Why is my MT4 client EA not receiving data from Python ZMQ server?

I want to build a copy trade system for forex trade activities.

My approach:

MT4 client EA uses zmq library for mql4 to receive message from python ZMQ server. MT4 Server EA uses zmq library for mql4 to send executed trade message to python ZMQ server.

Python side

I built python web server and python zmq server in one python file. when the mt4 client EA is attached to chart, it will use webrequest to validate the user by sending the user details to the python webserver, after validation the remaining communication will be through ZMQ. The MT4 server EA uses only ZMQ communication to python ZMQ server. I also plan to build another MT4 server EA that will use only Webrequest to communicate with Python webserver and the Python webserver will communicate with Python ZMQ server.

Additonally

The Python webserver and python zmq server has been deployed to hostinger vps

MT4 Client EA --> Python web server (for validation)

MT4 Client EA --> Python ZMQ server (for receiving trade details)

MT4 Server EA --> Python ZMQ server --> MT4 Client EA (for sending trade details to client EA)

MT4 Server EA --> Python web server --> Python ZMQ server --> MT4 Client EA (for sending trade details to client EA)

I have been able to make MT4 server EA send message to Python ZMQ server, Python ZMQ server is receiving this messages well. The problem is MT4 client EA could not receive any message which it receiving before I decided to test the MT4 server EA.

Let me explain it clearly, I have been testing client EA for many hours and it was working, working means it was receiving messages from Python ZMQ server in real time. I was happy then I said let me try the MT4 server EA , I noticed the MT4 server EA was not working at first, not working means it could not send message to Python ZMQ server after many attempts and changing somethings which I dont know, MT4 server EA started working by sending messages to Python ZMQ server I was happy again then I said let me now test both. I now discovered the MT4 Client EA stopped receiving trade/message from Python ZMQ server.

Python Code

import json
from dotenv import load_dotenv
import os
import zmq
import threading
from flask import Flask, render_template, request, jsonify
import signal
import sys
import time
from repository import (
get_all_v2,
validate_user,
)

# Load environment variables from .env file
load_dotenv()

Live_Envionment = os.environ["LIVE_ENVIRONMENT"]

# Configuration
ZMQ_PULL_PORT = 5557
ZMQ_PUB_PORT = 5558
SERVER = Live_Envionment
# Global Topic List
topicsInDb = []

# Flask Web Server
app = Flask(__name__)


# ZeroMQ Context for Flask
flask_context = zmq.Context()
zmq_socket = flask_context.socket(zmq.PUSH)
zmq_socket.connect(f"tcp://{SERVER}:{ZMQ_PULL_PORT}") # Connect to PULL socket


@app.route("/validate_user", methods=["POST"])
def validate():
data = request.json
broker_name = data.get("broker_name")
account_number = data.get("mt4_account_number")
account_name = data.get("account_name")

result = validate_user(acct_num=account_number)
if result["status"] == "active":
print("User has active subscriptions:", result["valid_topics"])
return jsonify(result), 200 # 200 OK for active subscription

elif result["status"] == "expired":
print(result["message"], "Expired topics:", result["expired_topics"])
return jsonify(result), 403 # 403 Forbidden for expired subscription

else:
# 400 Bad Request for no subscriptions or invalid user
print(result["message"])
return jsonify(result), 400

@app.route("/push_trade", methods=["POST"])
def push_trade():
data = request.json
try:
zmq_socket.send_json(data)
return jsonify({"status": "success"}), 200
except Exception as e:
return jsonify({"status": "error", "message": str(e)}), 500


# Helper to bind ZMQ socket with dynamic port handling
def bind_socket(socket, base_port):
while True:
try:
socket.bind(f"tcp://*:{base_port}")
print(f"Successfully bound to port {base_port}", flush=True)
return base_port
except zmq.ZMQError as e:
if e.errno == zmq.EADDRINUSE:
print(f"Port {base_port} in use. Trying next port...")
base_port += 1 # Try next port
else:
raise e


# ZeroMQ Server Setup (runs in a separate thread)
def zmq_server():
zmq_context = zmq.Context()
zmq_socket_pull = zmq_context.socket(zmq.PULL)
pull_port = bind_socket(zmq_socket_pull, ZMQ_PULL_PORT)

publisher = zmq_context.socket(zmq.PUB)
pub_port = bind_socket(publisher, ZMQ_PUB_PORT)
time.sleep(1) # Give time for the socket to bind properly

# updates topics in db
def announce_topics():
"""Broadcasts available topics to ensure clients subscribe early."""
global topicsInDb
for topic in topicsInDb:
dummy_message = {"info": f"Initialized topic {topic}"}
print(f" Announcing topic: {topic}", flush=True)
# Send an empty message
publisher.send_string(f"{topic} " + json.dumps(dummy_message))

def update_topics():
"""Background thread to update topics from the database without disconnecting clients."""
global topicsInDb
while True:
new_topics = get_all_v2("topics")
# Extract only topic names
topic_names = [t["topic_name"] for t in new_topics]
# print(f"dbTopics: {topic_names} topicsInDb: {topicsInDb}")
if set(topic_names) != set(topicsInDb): # Only update if topics change
print(f" Updating topics: {topic_names}", flush=True)
topicsInDb = topic_names # Update global topic list
announce_topics() # Broadcast new topics to clients
time.sleep(5) # Check for updates every 5 seconds

# Start the background thread
threading.Thread(target=update_topics, daemon=True).start()

# **Step 1: Announce Topics When Server Starts**
time.sleep(1) # Wait a moment before broadcasting
announce_topics()

def send_heartbeat():
"""Sends a heartbeat message every 5 seconds."""
while True:
try:
heartbeat = {"status": "alive"}
publisher.send_string(f"HEARTBEAT " + json.dumps(heartbeat))
time.sleep(5) # Send heartbeat every 5 seconds
except zmq.ZMQError as e:
print(f"Error sending heartbeat: {e}")
break

# Start the heartbeat in a separate thread
heartbeat_thread = threading.Thread(target=send_heartbeat, daemon=True)
heartbeat_thread.start()

print("ZMQ Server running...")
while True:
try:
trade_data = zmq_socket_pull.recv_json(flags=zmq.NOBLOCK)
topics = trade_data.get("trading_topics")
del trade_data["trading_topics"]
print(f"Received trade data: {trade_data}, topics: {topics}", flush=True)
# print(f"topicsInDb: {topicsInDb}")
# publisher.send_json(trade_data)
for db_topic in topicsInDb: # Iterate over each string in the list
if db_topic in topics:
print(f" Sending trade to {db_topic}", flush=True)
publisher.send_string(f"{db_topic} " + json.dumps(trade_data))
except zmq.Again:
time.sleep(0.01) # Prevent 100% CPU usage
except Exception as e:
print(f"Error processing trade data: {e}")


# Graceful Shutdown Handling
def graceful_shutdown(signal, frame):
print("Shutting down gracefully...")
zmq_socket.close()
flask_context.term()
sys.exit(0)


signal.signal(signal.SIGINT, graceful_shutdown)
signal.signal(signal.SIGTERM, graceful_shutdown)

# Start ZMQ server in a separate thread
if os.environ.get("WERKZEUG_RUN_MAIN") == "true":
zmq_thread = threading.Thread(target=zmq_server)
zmq_thread.daemon = True
zmq_thread.start()

# use_reloader=False
if __name__ == "__main__":
app.run(debug=True, host="0.0.0.0", port=5000)


MQL4 Client EA

#import "kernel32.dll"
bool GetVolumeInformationW (string lpRootPathName,string lpVolumeNameBuffer,int nVolumeNameSize,long &lpVolumeSerialNumber[],long &lpMaximumComponentLength[],long &lpFileSystemFlags[],string lpFileSystemNameBuffer,int nFileSystemNameSize);
#import

string DriveLetter = "C";

#include <Zmq/Zmq.mqh>

//--- Input Parameters
string PROJECT_NAME = "DWX_ZeroMQ_MT4_Client";
string ZEROMQ_PROTOCOL = "tcp";
string HOSTNAME = "copier.archos.com"; // IP of the Python ZMQ Server
int SUB_PORT = 5558; // Subscription Port
string TOPIC = "TRADES"; // Topic to subscribe to
//"127.0.0.1"

// Prepare our context and socket
Context context(PROJECT_NAME);
Socket subSocket(context, ZMQ_SUB); // Corrected: Use SUB socket

string connection = ZEROMQ_PROTOCOL + "://" + HOSTNAME + ":" + IntegerToString(SUB_PORT);

uchar _data[];
ZmqMsg request;

string uniqueName = "Archintel_Copy_Trade_Solution";
string verifiedToday = "DreifachEloiMFM", Sopran = "SopranEloiMFM",Qìchē="QìchēEloiMFM";
string server = "https://copier.archos.com/";
string securedfilename = "FearGod.dat";
//+------------------------------------------------------------------+
//| Expert Initialization |
//+------------------------------------------------------------------+
int OnInit()
{
GoogleVerify("real");
string topics = LoadSecureData(securedfilename);
//Print("Topics: ",topics);
string trading_topics[];
StringSplit(topics, ',', trading_topics);

//Print("Connecting to ZeroMQ server to...");
bool isConnected = subSocket.connect(connection);
if (isConnected) {
Print("Connected to ZeroMQ server");
if(ArraySize(trading_topics) >= 0){
for (int i = 0; i < ArraySize(trading_topics); i++) {
string topic = trading_topics;
bool isSubscribed = subSocket.subscribe(topic); // Subscribe to the topic
if(isSubscribed){
Print("Client EA subscribed to ",topic," trading signal");
}
}
}else{
MessageBox("Account has no subscription","Information",MB_ICONINFORMATION);
ExpertRemove();
}
}

EventSetTimer(1); // Set timer to check for messages every second
return INIT_SUCCEEDED;
}

//+------------------------------------------------------------------+
//| Expert Deinitialization |
//+------------------------------------------------------------------+
void OnDeinit(const int reason)
{
string topics = LoadSecureData(securedfilename);
//Print("Topics: ",topics);
string trading_topics[];
StringSplit(topics, ',', trading_topics);

if(ArraySize(trading_topics) >= 0){
for (int i = 0; i < ArraySize(trading_topics); i++) {
string topic = trading_topics;
bool isUnsubscribed = subSocket.unsubscribe(topic); // Unsubscribe to the topic
//Print("isUnsubscribed: ",isUnsubscribed);
if(isUnsubscribed){
Print("Client EA unsubscribed from ",topic," trading signal");
}
}
bool isDisconnect = subSocket.disconnect(connection);
if(isDisconnect){
Print("Client EA has disconnected from Python ZMQ server.");
}
}else{
bool isDisconnect = subSocket.disconnect(connection);
if(isDisconnect){
Print("Client EA has disconnected from Python ZMQ server.");
}
MessageBox("Account has no subscription","Information",MB_ICONINFORMATION);
ExpertRemove();
}

if(reason == REASON_REMOVE){
context.shutdown();
context.destroy(0);
}
EventKillTimer();
Print("Client EA deinitialized.");
}

//+------------------------------------------------------------------+
//| Timer Event - Check for New Messages |
//+------------------------------------------------------------------+
void OnTimer()
{
//Sleep(500); // Add a slight delay to ensure messages are captured
if (subSocket.recv(request, true)) {
Print("Received message from Python ZMQ server");
if (request.size() > 0) {
MessageHandler(request);
}
}else{
//Print("⚠️ No messages received. Checking if ZMQ server is alive...");
// Try an artificial ping request
if (!subSocket.subscribe("PING")) {
Print("❌ Lost connection to ZMQ Server!");
ExpertRemove(); // Remove EA if connection is lost
}
}
}


//+------------------------------------------------------------------+
//| Process Incoming Trade Signal |
//+------------------------------------------------------------------+
void MessageHandler(ZmqMsg &_request)
{
ArrayResize(_data, _request.size());
_request.getData(_data);
string dataStr = CharArrayToString(_data);

// Remove topic prefix
if (StringFind(dataStr, "TRADES ") == 0) {
dataStr = StringSubstr(dataStr, StringLen("TRADES "));
}

Print("Received Data: ", dataStr);

// Further processing
string components[11];
StringReplace(dataStr, "{", "");
StringReplace(dataStr, "}", "");
StringReplace(dataStr, ":", ";");

ParseZmqMessage(dataStr, components);
InterpretZmqMessage(components);
}


// Interpret Zmq Message and perform actions
void InterpretZmqMessage(string &compArray[])
{
if (ArraySize(compArray) >= 2) {
string key = compArray[0];
string value = compArray[1];
Print("Key: ", key, " | Value: ", value);
}
}

// Parse Zmq Message
void ParseZmqMessage(string &message, string &retArray[])
{
ushort sep = StringGetCharacter(";", 0);
StringSplit(message, sep, retArray);
}


MQL4 Server EA

#include <Zmq/Zmq.mqh>

//--- Input Parameters
extern string PROJECT_NAME = "DWX_ZeroMQ_MT4_Master";
string ZEROMQ_PROTOCOL = "tcp";
string HOSTNAME = "copier.archos.com"; // IP of the Python ZMQ Server
int PUSH_PORT = 5557; // Port to push data to Python ZMQ Server

// Prepare our context and socket
Context context(PROJECT_NAME);
Socket pushSocket(context, ZMQ_PUSH); // Using PUSH socket

string connection = ZEROMQ_PROTOCOL + "://" + HOSTNAME + ":" + IntegerToString(PUSH_PORT);//147.93.86.50

uchar _data[];
ZmqMsg mssg;

//+------------------------------------------------------------------+
//| Expert Initialization |
//+------------------------------------------------------------------+
int OnInit()
{
Print("Connecting to ZeroMQ Python server...");
bool isConnected = pushSocket.connect(connection);
if (isConnected) {
Print("Server EA initialized and connected to Python ZMQ server.");
} else {
Print("Failed to connect to Python ZMQ server.");
}

EventSetTimer(5); // Set timer to send trades every 5 seconds (for testing)
return INIT_SUCCEEDED;
}

//+------------------------------------------------------------------+
//| Expert Deinitialization |
//+------------------------------------------------------------------+
void OnDeinit(const int reason)
{
pushSocket.disconnect(connection);
if(reason == REASON_REMOVE){
context.shutdown();
context.destroy(0);
}
EventKillTimer();
Print("Server EA deinitialized.");
}

//+------------------------------------------------------------------+
//| Timer Event - Send Trade Data |
//+------------------------------------------------------------------+
void OnTimer()
{
string symbol = Symbol();
double price = Ask;
int ticket = 12345; // Dummy ticket number for testing

// Create JSON-like message
string tradeData = StringFormat("{\"Symbol\":\"%s\",\"Price\":%s,\"Ticket\":%s,\"trading_topics\":%s}",
symbol, (string)price, (string)ticket, "[\"ELONTRADES\"]");
Print("Sending trade data to Python ZMQ server: ", tradeData);

// Convert to uchar array for ZeroMQ
//StringToCharArray(tradeData, _data);
//mssg.setData(_data);

//ZmqMsg msg(tradeData);


ZmqMsg msg;
msg.rebuild(tradeData);


// Send the message
if (pushSocket.send(msg)) {
Print("Trade data sent successfully.");
} else {
Print("Failed to send trade data.");
}
}

Continue reading...