From Polling to Events: Building a Reactive MQL5 Trading System with Python and Redis Streams
122
0
Introduction
An Expert Advisor can poll an external service for signals, and at low volume nothing feels wrong. A timer, an HTTP request, and a list of pending actions are enough, and the code is easy to follow.
The trouble starts when the system has to survive a crash, a retry, or a broker reply that is neither a clear yes nor a clear no. Was this signal already handled? Did that order reach the broker before the restart? Should this message ever be read again?
This article follows that problem from a simple polling loop to a small prototype built on Redis Streams. MQL5 publishes market events, Python turns them into trading ideas, and the EA decides what happens next. The EA still polls on a timer. What changes is not the mechanism, but the semantics: every message gets a stable identity, an explicit acknowledgment, and a way out when it cannot be used.
One rule runs through the design. Python is allowed to suggest a trade. It is not allowed to place one. The final decision always stays inside MQL5.
When polling stops being simple
In its simplest form, OnTimer() calls WebRequest() on a Python endpoint every few seconds, and the EA tries an order for every signal returned. With one producer, one consumer, and at most one signal per bar, this works. The real problem appears one crash later.
Messages can be processed twice: if the EA crashes after sending an order but before recording it, the signal comes back looking brand new. Without an acknowledgment, "finished" and "never seen" look the same. Invalid records can get stuck: a message the EA cannot parse returns on every poll. And broker outcomes are rarely binary: a loop that only knows "filled" retries a partial fill that already traded, and retries a margin rejection forever.
None of this is caused by polling itself. Polling works when the protocol behind it is explicit: stable IDs, acknowledgments, a rejection path, and processing that gives the same result when it runs twice. Part of that comes ready-made from Redis Streams. The rest has to be built in the EA.
Redis Streams help, but only with delivery
A Redis stream is an append-only log. Producers add entries with XADD, and every entry gets a transport ID such as 1792622276177-0. Consumers read through a consumer group, which keeps a pending entries list (PEL): an entry handed to a consumer stays pending until the consumer confirms it with XACK, and after a restart the consumer reads its own pending entries again with XREADGROUP and ID 0, and new ones with ID >.
That sounds minor, but it changes the whole failure model. A crash no longer loses a message, and a message can no longer disappear quietly: every entry is either acknowledged, still pending, or explicitly moved somewhere else.
The wrapper uses only the commands it needs. Creating the group comes first:
# ensure_group - create a consumer group if it does not exist
def ensure_group(self, stream_name: str, group_name: str) -> None:
"""Create a consumer group if it does not already exist.
A new group starts at '$': it receives only messages published after
it was created, so a newly started EA does not replay the whole stream
history (those signals would be stale anyway).
Only BUSYGROUP (the group already exists) is ignored. Any other Redis
error, such as WRONGTYPE when the key is not a stream, is logged and
raised to the caller.
"""
try:
self.redis.xgroup_create(stream_name, # target stream
group_name, # group name
id='$', # only messages after creation
mkstream=True) # create stream if absent
except redis.exceptions.ResponseError as e:
if str(e).startswith('BUSYGROUP'): # group already exists?
return # expected: nothing to do
logging.error(f'XGROUP CREATE {group_name} on {stream_name} failed: {e}') # unexpected Redis error
raise # let the caller decide XGROUP CREATE fails with BUSYGROUP when the group already exists, which is normal on every restart, so only that error is ignored. Anything else, such as WRONGTYPE, is logged and raised; hiding it would leave an EA that silently receives nothing. A new group starts at $, so an EA does not replay old signals, and the strategy must start before the EA, or its group misses the first bars.
Reading new entries has a trap of its own:
# consume_new_events - fetch never-delivered messages
def consume_new_events(self, stream_name: str, # source stream
group_name: str, # consumer group
consumer_name: str, # consumer instance
count: int = 10, # max messages
block_ms: Optional[int] = None # None = never block
) -> List[Tuple[str, Dict]]:
"""Return messages that have never been delivered to this consumer group.
block_ms=None -> no BLOCK argument: the call returns immediately.
block_ms=N > 0 -> wait at most N milliseconds for new messages.
BLOCK 0 is deliberately never sent: in Redis it means "wait forever".
"""
if block_ms is not None and block_ms <= 0: # refuse an unbounded wait
raise ValueError('block_ms must be None or a positive number of milliseconds') # BLOCK 0 = wait forever
self.ensure_group(stream_name, group_name) # group must exist
response = self.redis.xreadgroup(group_name, # consumer group
consumer_name, # consumer name
{stream_name: '>'}, # never-delivered only
count=count, # max messages
block=block_ms) # None = do not block
return self._flatten(response) # return the list In Redis, BLOCK 0 does not mean "do not wait". It means "wait forever". The gateway is single-threaded, so one blocked read would stall every request, including the EA's publication of the very bar that would produce the awaited signal. The wrapper therefore never sends it.
So far, so good. But delivery is only half of the story.
XACK tells Redis that an entry no longer needs to stay pending. It does not tell us whether an order was sent, accepted, or filled before a restart. Treat an ACK as proof of execution, and the first crash in the wrong place produces a duplicate trade. To close that gap, the EA needs three things of its own: a durable journal that recognizes a redelivered signal even after a restart, a link from every order and deal back to its signal, and a decision for every broker reply: accepted, worth retrying, or final.
Why MQL5 keeps the final say
Business correctness starts with a question of authority. A signal carries a direction and a reference to its bar, but no volume, no price, and no stop levels, and nothing in it can switch a check off. Every check that decides whether an order is valid, safe, and sendable runs inside the EA, in ExecutionGuard.
Why be this strict? Because the boundary has to hold even when Python is wrong. A strategy bug, a misconfigured deployment, or a local process writing to Redis can do no more than a correct signal could. The worst a bad signal can do is what the guard would allow a good one to do. Note what this promises: every order conforms to the EA's rules, not that the trade is a good idea.
A signal that fails any check in Table 1 is rejected, journaled, and acknowledged.
| Check | Why |
|---|---|
| The event is a SIGNAL_GENERATED for the chart symbol, with direction BUY or SELL | Anything else is a malformed signal, not an opinion |
| Age on the trade-server clock between zero and MaxSignalAgeSeconds | A decision based on old market data is stale |
| Terminal connected; trading allowed in the terminal, for the program, for the account, and for Expert Advisors on the account | Trading can be switched off at each level |
| Symbol not disabled, and its trade mode allows the direction | Long-only, short-only, and close-only symbols exist |
| No EA-owned position (hedging) or no position at all (netting) on the symbol | On a netting account, a new order would modify a position the EA may not own |
| No EA-owned order still active on the symbol | An accepted but unfilled order has no position yet |
| Spread at most MaxSpreadPoints | Avoids rollover and news spikes |
| Configured volume valid after normalization | Python never supplies a volume |
Table 1. What the guard checks, and why.
Normalization clamps the configured volume into the broker's range and rounds it down to whole volume steps, taking the decimals from the step itself, so 0.125 keeps three. A final check in NormalizeVolume() rejects anything still invalid. A stricter risk policy could refuse an out-of-range volume instead of clamping it. After the guard, OrderCheck() validates margin and the full request.
Python keeps only the soft decision. Here an up bar means BUY and anything else SELL. The confidence of 0.8 in the signal is decorative: nothing reads it.
The age check needs one clock. TimeCurrent() is server time, Python's time.time() is local time, and the two differ by the broker's time-zone offset. So the EA stamps BAR_CLOSED with TimeCurrent(), the strategy copies it unchanged into source_event_time, and the guard measures on that same clock:
//--- 4. confirm freshness on ONE clock: the trade server clock. //--- source_event_time is the TimeCurrent() value the EA stamped on the //--- BAR_CLOSED event; Python copies it into the signal payload unchanged. //--- The Python-side event_time (local system clock) is audit data only. long source_time = GetPayloadLong(signal.payload, "source_event_time"); // server-clock origin time if(source_time <= 0) // field missing or invalid? { reason = "Signal carries no source_event_time"; // describe the anomaly return(false); // reject } long age = (long)TimeCurrent() - source_time; // age on the same clock (seconds) if(age < 0) // future-dated signal? { reason = "Signal has a future timestamp"; // describe the anomaly return(false); // reject } if(age > m_max_age_seconds) // exceeds maximum age? { reason = "Signal is stale (age = " + IntegerToString(age) + " s)"; // describe the staleness return(false); // reject }
This measures the age of the market data, so any delay in Python, Redis, or the gateway counts against the signal. That is deliberate: a decision based on old data is stale, however recently it was computed. Python's event_time and Redis stream IDs play no part in freshness.
The event model
The guard can only check what an event carries, and acknowledgments need a stable envelope:
//+------------------------------------------------------------------+ //| Enum: ENUM_EVENT_TYPE - all event categories in the system | //+------------------------------------------------------------------+ enum ENUM_EVENT_TYPE { EVENT_BAR_CLOSED = 0, // a chart bar has closed EVENT_SIGNAL_GENERATED = 1, // Python has produced a trading signal EVENT_ORDER_ATTEMPTED = 2, // the EA is about to send an order EVENT_DEAL_EXECUTED = 3, // the broker has filled a deal EVENT_POSITION_UPDATED = 4 // an open position has changed }; //+------------------------------------------------------------------+ //| Struct: EventEnvelope - compact container for one event | //+------------------------------------------------------------------+ struct EventEnvelope { string event_id; // application-level unique identifier string stream_id; // Redis transport ID (empty when publishing) ENUM_EVENT_TYPE event_type; // category of this event string source; // module that produced this event datetime event_time; // creation timestamp (Unix seconds) string symbol; // trading symbol related to this event string payload; // JSON string carrying extra fields };
The event_id is the logical identity: what the event is about. The stream_id is where Redis stored this copy. They sit next to each other on purpose, because mixing them up is the classic mistake. A redelivery keeps the stream_id; a republished copy gets a new one but keeps its event_id. Deduplication uses the event_id, acknowledgment the stream_id, never the other way around.
| Field | Gateway check |
|---|---|
| event_id, source, symbol, event_type | Required text; event_type must be known; a missing event_id is never replaced by the Redis ID |
| event_time | Positive integer (Unix seconds); for signals, audit data only |
| payload | A flat JSON object; values are finite numbers or text without quotes, backslashes, braces, or control characters; booleans, null, arrays, and nested objects are refused |
Table 2. Envelope fields and gateway checks.
The payload rules look fussy, and they exist for one reason. The EA parses JSON with a small, constrained parser, so the gateway refuses anything it cannot read instead of repairing it. A boolean is valid JSON, but the parser does not handle it, so it is refused before it is stored. Inside the EA the payload is a JSON string that the serializer maps to and from the object on the wire.
Every event_id is built from durable facts, never from a counter (Table 3), so the same bar announced twice gives the same signal ID and the journal recognizes it. Redis does not deduplicate; the EA does, per signal.
| Event | Producer and stream | event_id | Payload fields |
|---|---|---|---|
| BAR_CLOSED | MQL5, market.events | BAR_CLOSED:{symbol}:{timeframe}:{bar_open_time} | bar_time, open, high, low, close, tick_volume |
| SIGNAL_GENERATED | Python, signal.events | SIGNAL:{source_event_id}:{strategy_version} | direction, confidence, strategy_version, source_event_id, source_event_time |
| ORDER_ATTEMPTED | MQL5, trade.events | ORDER_ATTEMPTED:{signal_event_id}:{attempt} | signal_event_id, attempt, direction, volume, magic, order_tag |
| DEAL_EXECUTED | MQL5, trade.events | DEAL_EXECUTED:{deal_ticket} | deal, order, signal_event_id, direction, entry, volume, price |
| POSITION_UPDATED | MQL5, trade.events | Reserved, not yet published | None |
Table 3. Event types, identities, and payloads.
ORDER_ATTEMPTED records what the EA decided to submit; DEAL_EXECUTED records what the broker executed. One attempt can produce no deal, or several partial ones.
The strategy enforces the BAR_CLOSED payload in validate_bar_event() instead of trusting it:
# validate_bar_event - check required fields and the complete OHLC bar
def validate_bar_event(event: dict) -> bool:
"""Check that the event is a valid BAR_CLOSED with a complete, consistent OHLC bar.
open, high, low, and close are all required, must be finite and positive,
and the high-low range must contain both the open and the close.
"""
if event.get('event_type') != 'BAR_CLOSED': # must be the right type
return False # reject wrong events
if not event.get('event_id'): # must have an ID
return False # reject missing ID
if not event.get('symbol'): # must have a symbol
return False # reject missing symbol
try: # must have an integer timestamp
if int(event.get('event_time')) <= 0: # positive Unix seconds (server clock)
return False # reject non-positive time
except (TypeError, ValueError): # missing or non-numeric
return False # reject missing time
# decode the payload and require all four prices
try:
payload = json.loads(event.get('payload', '{}')) # decode payload JSON
bar = {k: float(payload[k]) for k in ('open', 'high', 'low', 'close')} # all four required
except (json.JSONDecodeError, KeyError, TypeError, ValueError): # malformed or missing price
return False # reject incomplete payload
# reject NaN, Infinity, zero, and negative prices
if not all(math.isfinite(p) and p > 0 for p in bar.values()): # float('nan') does not raise
return False # reject invalid numbers
# OHLC consistency: the high-low range must contain the open and the close
if bar['high'] < max(bar['open'], bar['close']): # high below the body?
return False # reject inconsistent bar
if bar['low'] > min(bar['open'], bar['close']): # low above the body?
return False # reject inconsistent bar
return True # event is valid All four prices must be finite, positive, and consistent. The isfinite() check matters because float("nan") does not raise. An invalid bar is acknowledged and discarded; a failed publication of the signal leaves the bar pending for a retry.
From market event to trade attempt
The EA publishes BAR_CLOSED through the gateway. The strategy, which talks only to Redis, publishes SIGNAL_GENERATED. On each timer tick the EA fetches pending and new signals, filters duplicates through its journal, runs the guard and OrderCheck(), publishes ORDER_ATTEMPTED, and calls OrderSend(). Deals arrive later as DEAL_EXECUTED (Figure 1).
Figure 1. Reactive event flow from market event to confirmed trade.
The MQL5 side is timer-driven, and that is not a stylistic choice: MQL5 has neither a Redis client nor an asynchronous HTTP client. Figure 2 shows how each event crosses the stream boundaries.
Figure 2. Event lifecycle across stream boundaries.
One bar, end to end
This cycle comes from the demo test described later. At 17:03:00 an M1 bar of EURUSD# closed and the EA published BAR_CLOSED. The strategy published a SELL signal with the bar's server time and acknowledged the bar; the EA never acknowledges BAR_CLOSED, since it only publishes it. About a second later the EA received the signal, found no journal entry, passed the guard and OrderCheck(), published ORDER_ATTEMPTED, wrote ATTEMPTING, and sent the order. The broker answered 10009 (DONE); the EA journaled ACCEPTED with the ticket and acknowledged the signal in its own group, mql5-EURUSD_-123456, where "#" became "_" because group names travel in a URL.
17:03:00.305 Published BAR_CLOSED event: BAR_CLOSED:EURUSD#:M1:1792602120 17:03:01.281 OrderCheck passed for SIGNAL:BAR_CLOSED:EURUSD#:M1:1792602120:simple-1: retcode=0 (Done) 17:03:01.997 Order accepted: EURUSD#, volume=0.1, type=SELL, order=999271786, retcode=10009 17:03:02.069 Published DEAL_EXECUTED: DEAL_EXECUTED:1187726132 (signal SIGNAL:BAR_CLOSED:EURUSD#:M1:1792602120:simple-1)
Figure 3. Event identity across the round trip.
The link from deal to signal rests on ordering. The EA writes result.order into the journal before OnTimer() returns, and trade transactions arrive only after that, so OnTradeTransaction() finds the signal through DEAL_ORDER. A partial fill produces several DEAL_EXECUTED events with the same IDs. The order comment carries a second link, an 11-character tag: RS- plus the FNV-1a hash of the signal ID, which lets the EA find its order after a restart even before any deal. The signal is acknowledged on the OrderSend() result, not on the first deal, because the ACK answers "did this become an accepted order?", not "what was executed?".
Reading and acknowledging signals
The gateway is the only component that speaks both HTTP and Redis. It exposes /publish, /consume, /ack, and /reject, and contains no trading logic. This is how it answers a poll:
# handle_consume - return pending and new events for a consumer
def handle_consume(self):
"""Return up to `count` events: pending entries first, then new ones.
New entries are only requested for the remaining capacity, so the
gateway never claims messages it does not return. Neither read
blocks. Records that violate the contract are dead-lettered and
acknowledged here instead of being returned.
"""
# parse query parameters from the URL
params = parse_qs(urlparse(self.path).query) # parse query string
stream = params.get('stream', ['signal.events'])[0] # stream name
group = params.get('group', [CONSUMER_GROUP])[0] # group name
consumer = params.get('consumer', [CONSUMER_NAME])[0] # consumer name
try:
count = max(1, min(int(params.get('count', ['10'])[0]), MAX_COUNT)) # bounded batch size
except ValueError: # non-numeric count?
self.send_json(400, {'error': 'count must be an integer'}) # bad request
return # abort handling
try:
# recover messages already assigned to this consumer
pending = redis_stream.consume_pending_events(stream, group, consumer, count=count) # XREADGROUP 0
# fetch never-delivered messages only for the remaining capacity
remaining = count - len(pending) # free slots in this response
new = redis_stream.consume_new_events(stream, group, consumer,
count=remaining) if remaining > 0 else [] # XREADGROUP >
# validate each record; dead-letter the ones that break the contract
result = [] # response accumulator
for msg_id, record in pending + new: # pending first, then new
envelope, error = normalize_envelope(record) # validate and normalize
if error: # poison message?
reject_record(stream, group, msg_id, record, error) # dead-letter and ACK
continue # do not return it
result.append({'stream_id': msg_id, **envelope}) # transport ID emitted first
# return the compact JSON array to the caller
self.send_json(200, result) # success
except Exception as e: # catch any error
logging.error(f'Consume error: {e}') # log server-side
self.send_json(500, {'error': str(e)}) # server error Pending entries come first, and new ones fill only the remaining capacity, so the gateway never claims an entry it does not return. The same normalize_envelope() check runs on publication, where a violation gets HTTP 400, and here, where a bad record is copied to signal.events.dlq with its reason and acknowledged. The value check is short:
# parser_safe - True if a value can be read by the MQL5 parser def parser_safe(value): """Strings must not contain quotes, backslashes, braces, or control characters; numbers must be finite. Booleans, nested objects, arrays, and null are not allowed.""" if isinstance(value, bool): # true / false? return False # not in the contract; checked first because a bool is also an int if isinstance(value, (int, float)): # numeric value return math.isfinite(value) # NaN and Infinity are not JSON if isinstance(value, str): # text value return not any(c in UNSAFE_CHARS or ord(c) < 32 for c in value) # no breaking characters return False # dict, list, None
The bool test comes first because in Python a bool is also an int. If a record still fails to parse in the EA, its stream_id goes to /reject, and the gateway dead-letters and acknowledges it. That second line of defense was never needed in the tests.
Acknowledging is where an earlier version trusted too much. The gateway returns HTTP 200 whenever it handled the request, but XACK reports how many entries it removed, and zero means the entry was not pending: already acknowledged, or a wrong group or ID. Redis does not say which. So the EA counts an ACK only when the reply says "acknowledged":true:
//+------------------------------------------------------------------+ //| Acknowledged - did XACK actually remove the entry? | //| HTTP 200 only means that the gateway answered. XACK returns 0 | //| both for an entry that was already acknowledged and for a wrong | //| group or ID, so "acknowledged":false never counts as success. | //+------------------------------------------------------------------+ bool EventConsumer::Acknowledged(const string reply, // gateway reply body string &error_msg) const // error message if any { if(StringFind(reply, "\"acknowledged\":true") >= 0) // XACK removed one entry? { error_msg = ""; // no error to report return(true); // acknowledgment confirmed } error_msg = "entry was not pending (already acknowledged, or wrong group or ID)"; // describe the outcome return(false); // not confirmed } //+------------------------------------------------------------------+ //| AckEvent - acknowledge a processed event via Redis stream ID | //+------------------------------------------------------------------+ bool EventConsumer::AckEvent(const string stream_id, // Redis stream entry ID string &error_msg) // error message if any { //--- acknowledge in the SAME group that consumed the message string json = "{\"stream_id\":\"" + EscapeJSON(stream_id) + // JSON body: stream_id "\",\"stream\":\"" + STREAM_SIGNAL + // JSON body: stream name "\",\"group\":\"" + EscapeJSON(m_consumer_group) + "\"}"; // JSON body: consuming group string reply; // gateway reply body if(!PostJSON("/ack", json, reply, error_msg)) // POST /ack return(false); // transport or HTTP failure return(Acknowledged(reply, error_msg)); // XACK must have removed the entry }
An unconfirmed ACK leaves the entry pending; it returns on the next poll and the journal recognizes it. It costs a round trip, never a second order.
One more line came from testing:
# send_json - write a JSON response with a status code
def send_json(self, status, body):
"""Serialize body as compact JSON and send it with the given status."""
data = json.dumps(body, separators=COMPACT).encode('utf-8') # compact UTF-8 body
self.send_response(status) # status line
self.send_header('Content-Type', 'application/json') # JSON content
self.send_header('Content-Length', str(len(data))) # body length
self.send_header('Connection', 'close') # new connection per request; MT5 under Wine fails on reuse
self.end_headers() # finish headers
self.wfile.write(data) # write body Python's HTTPServer closes the connection after every reply. MetaTrader 5 under Wine appeared to reuse it, so every second request failed with error 1003 before reaching the gateway. Connection: close fixes that.
Surviving restarts: the journal and reconciliation
Redis remembers what is pending, not what the EA already did. That memory is SignalJournal, an append-only file per symbol and magic number, where every transition is flushed before the EA acts on it:
//+------------------------------------------------------------------+ //| Append - write one line and flush it to disk | //+------------------------------------------------------------------+ bool SignalJournal::Append(const JournalEntry &entry) // entry to persist { //--- open the file for appending (created if absent) int h = FileOpen(m_file, FILE_READ | FILE_WRITE | FILE_TXT | FILE_ANSI | FILE_SHARE_READ); // file handle if(h == INVALID_HANDLE) // open failed? { PrintFormat("SignalJournal: cannot open %s, error=%d", m_file, GetLastError()); // log failure return(false); // report failure } //--- write the line at the end and make it durable FileSeek(h, 0, SEEK_END); // move to end of file bool ok = (FileWriteString(h, Format(entry) + "\r\n") > 0); // write one line FileFlush(h); // durable before the caller acts FileClose(h); // release the handle return(ok); // report the write result }
MarkAttempting(), MarkAccepted(), and MarkRejected() all go through Transition(), which updates memory only if Append() succeeded and returns that result. Each line reads event_id|state|updated|attempts|order. On start-up, entries older than a day are dropped; if the journal cannot be loaded, OnInit() fails and the EA refuses to trade.
| State | Written when | On redelivery |
|---|---|---|
| ATTEMPTING (1) | After ORDER_ATTEMPTED, before OrderSend() | Reconcile with the broker first |
| ACCEPTED (2) | The broker accepted; the ticket is stored | Acknowledge only; not proof of a fill |
| REJECTED (3) | Guard or broker refused for good | Acknowledge only |
Table 4. Journal states.
The final state is always written before the ACK:
//+------------------------------------------------------------------+ //| AckSignal - acknowledge in the same group the signal came from | //+------------------------------------------------------------------+ void AckSignal(const string stream_id) // Redis stream entry ID { string error; // error message holder if(!consumer.AckEvent(stream_id, error)) // ACK not confirmed? PrintFormat("ACK not confirmed for %s: %s (a pending entry is redelivered and deduplicated)", stream_id, error); // log; the journal makes redelivery harmless } //+------------------------------------------------------------------+ //| FinishSignal - acknowledge only after the final state is durable | //| The journal flushes every transition to disk before it returns. | //| If that write failed, the signal stays pending, and its next | //| delivery is resolved from the journal and the broker's orders. | //+------------------------------------------------------------------+ void FinishSignal(const EventEnvelope &signal, // signal being finished const bool written) // result of the journal write { if(written) // final state on disk? AckSignal(signal.stream_id); // now it is safe to acknowledge else PrintFormat("Journal write failed for %s; left pending without ACK", signal.event_id); // retried on redelivery }
If the write fails, there is no ACK and the signal stays pending. Acknowledging first would let a crash erase the only record of a real trade. ProcessSignal() puts it all together:
//+------------------------------------------------------------------+ //| ProcessSignal - one signal through dedup, guard, and execution | //+------------------------------------------------------------------+ void ProcessSignal(const EventEnvelope &signal) // signal to process { string id = signal.event_id; // application identity //--- 0. a signal for another symbol belongs to another EA, which receives its //--- own copy in its own consumer group: acknowledge it here and skip it if(signal.symbol != _Symbol) // not for this EA? { AckSignal(signal.stream_id); // ACK in this EA's group only return; // no journal entry needed } ENUM_SIGNAL_STATE state = journal.State(id); // durable state of this signal //--- 1. final states: this is a duplicate delivery, only the ACK is missing if(state == SIGNAL_STATE_ACCEPTED || state == SIGNAL_STATE_REJECTED) // already final? { PrintFormat("Duplicate signal ignored: %s", id); // log duplicate AckSignal(signal.stream_id); // complete the missing ACK return; // nothing else to do } //--- 2. an attempt was started earlier but its outcome was never recorded //--- (crash, restart, or an ambiguous broker reply): reconcile first if(state == SIGNAL_STATE_ATTEMPTING) // outcome unknown? { ulong found_order = 0; // order found at the broker ENUM_RECONCILE found = ReconcileAttempt(signal.symbol, id, journal.LastUpdate(id), found_order); // search orders and history if(found == RECONCILE_UNKNOWN) // history not trustworthy now? { PrintFormat("Cannot reconcile %s yet; left pending", id); // log and wait return; // no ACK: retried next tick } if(found == RECONCILE_FOUND) // order already exists? { PrintFormat("Signal %s already reached the broker (order %I64u); not resending", id, found_order); // log FinishSignal(signal, journal.MarkAccepted(id, found_order)); // final state first, then ACK return; // never resend } PrintFormat("Previous attempt for %s left no order; re-validating", id); // log and continue } //--- 3. hard constraints (volume is normalized inside the guard) double volume = 0.0; // output volume from guard string reason; // textual rejection reason if(!guard.ValidateSignal(signal, volume, reason)) // run all safety checks { PrintFormat("Signal rejected: %s | Reason: %s", id, reason); // log FinishSignal(signal, journal.MarkRejected(id)); // final state first, then ACK return; // move to next signal } //--- 4. execution with explicit outcome classification ulong order = 0; // broker order ticket ENUM_EXEC_OUTCOME outcome = ExecuteOrder(signal, volume, order, reason); // attempt submission if(outcome == EXEC_ACCEPTED) // broker accepted the request? { //--- TEST ONLY: simulate a crash between OrderSend() and the final journal write if(TestStopAfterOrderSend) // test switch on? { PrintFormat("TEST: stopping after OrderSend() for %s (order %I64u); journal stays ATTEMPTING, no ACK", id, order); // log the simulated crash test_stopping = true; // skip the rest of this batch ExpertRemove(); // unload the EA, as a crash would return; // no journal write, no ACK } FinishSignal(signal, journal.MarkAccepted(id, order)); // persist the order -> signal link, then ACK } else if(outcome == EXEC_PERMANENT_FAILURE) // can never succeed? { PrintFormat("Order permanently rejected for %s: %s", id, reason); // log FinishSignal(signal, journal.MarkRejected(id)); // final state first, then ACK } else // transient failure PrintFormat("Transient failure for %s: %s (left pending; bounded by MaxSignalAgeSeconds)", id, reason); // leave in the PEL for retry }
A signal for another symbol is acknowledged only in this EA's group, because every EA reads every signal in its own group; two EAs on one symbol therefore need different magic numbers. The TestStopAfterOrderSend block is a test switch, off by default, that stops the EA right after an accepted OrderSend() so a crash can be reproduced.
The window between OrderSend() and the journal
ATTEMPTING marks an attempt whose result was never recorded: the EA stopped just before or just after OrderSend(), or the reply was ambiguous, such as TIMEOUT or CONNECTION. A deal is no proof either way, since a PLACED order may not have filled. On redelivery, ReconcileAttempt() searches active orders, then the order history, for this signal's tag, magic number, and symbol. Some brokers replace order comments, so an order without any tag is matched by its setup time; an order with another signal's tag never matches. There are three answers:
- RECONCILE_FOUND: the order exists and was not rejected, so the signal becomes ACCEPTED and is never resent;
- RECONCILE_NONE: the history is trustworthy and empty, so the signal goes through the guard again, freshness included;
- RECONCILE_UNKNOWN: the terminal is disconnected or HistorySelect() failed, so the signal waits; guessing "no order" would cause a duplicate.
This is still a heuristic, and the last section says where it can fail.
Sending the order
ExecuteOrder() runs in a fixed order, so that recovery always finds a journal entry and a tag:
//--- 3. let the terminal check the complete request before anything is recorded; //--- the call must succeed AND report a success code (0, or DONE) MqlTradeCheckResult check = {}; // zero-initialize check result bool checked = OrderCheck(request, check); // margin, volume, stops, trade mode if(!checked || (check.retcode != 0 && check.retcode != TRADE_RETCODE_DONE)) // call or check failed? { reason = StringFormat("OrderCheck %s, retcode=%u (%s)", checked ? "returned true" : "failed", check.retcode, check.comment); // describe the outcome //--- same rules as OrderSend(): a transient code stays pending, the rest is final return(ClassifyRetcode(check.retcode) == EXEC_TRANSIENT_FAILURE ? EXEC_TRANSIENT_FAILURE : EXEC_PERMANENT_FAILURE); // never "accepted" here } PrintFormat("OrderCheck passed for %s: retcode=%u (%s)", signal.event_id, check.retcode, check.comment); // record the success code in the log //--- 4. audit: no ORDER_ATTEMPTED record, no order (fail closed) EventEnvelope ev; // build the attempt event ev.event_id = "ORDER_ATTEMPTED:" + signal.event_id + ":" + IntegerToString(attempt); // deterministic ID ev.stream_id = ""; // assigned by Redis, never sent ev.event_type = EVENT_ORDER_ATTEMPTED; // tag as attempt event ev.source = "MQL5_EA"; // identify the producer ev.event_time = TimeCurrent(); // trade-server clock ev.symbol = signal.symbol; // carry over symbol string payload = "{}"; // start with empty JSON SetPayloadField(payload, "signal_event_id", signal.event_id); // causal link to the signal SetPayloadField(payload, "attempt", IntegerToString(attempt)); // attempt number SetPayloadField(payload, "direction", direction); // intended direction SetPayloadField(payload, "volume", LotsToString(volume)); // intended size SetPayloadField(payload, "magic", IntegerToString((long)MagicNumber)); // ownership tag SetPayloadField(payload, "order_tag", tag); // correlation tag in the order comment ev.payload = payload; // attach payload to event string pub_error; // error holder for publish if(!publisher.PublishEvent(ev, pub_error)) // audit record not confirmed? { reason = "ORDER_ATTEMPTED could not be published: " + pub_error; // describe the block return(EXEC_TRANSIENT_FAILURE); // no order without an audit record } //--- 5. write-ahead: the attempt is durable BEFORE the order leaves the terminal if(!journal.MarkAttempting(signal.event_id, attempt)) // journal write failed? { reason = "journal write failed before OrderSend"; // describe the block return(EXEC_TRANSIENT_FAILURE); // no order without durable state } //--- 6. refresh the price: publishing may have taken up to two seconds if(SymbolInfoTick(signal.symbol, tick)) // fresh tick available? request.price = (order_type == ORDER_TYPE_BUY) ? tick.ask : tick.bid; // keep the checked price otherwise //--- 7. send the order to the broker ResetLastError(); // clear any stale error bool sent = OrderSend(request, result); // submit the trade order = result.order; // broker order ticket (0 if none) //--- 8. classify by retcode, whatever OrderSend() returned ENUM_EXEC_OUTCOME outcome = ClassifyRetcode(result.retcode); // accepted, transient or permanent if(outcome != EXEC_ACCEPTED) // not accepted? { reason = StringFormat("OrderSend %s, retcode=%u (%s), last_error=%d", sent ? "returned true" : "failed", result.retcode, result.comment, GetLastError()); // describe the outcome return(outcome); // report the classification }
OrderCheck() must both succeed and report a success code. The documentation does not name that code, so the EA accepts 0 or DONE and logs it; the demo run returned 0. A failed check is classified like an OrderSend() retcode but never counts as accepted, and it happens before anything is recorded.
ORDER_ATTEMPTED is fail-closed: if the gateway does not confirm it, no order is sent. An audit trail with gaps could not be trusted to show what the EA sent. Then ATTEMPTING is written, the price refreshed, and the order sent. ClassifyRetcode() sorts the reply:
//+------------------------------------------------------------------+ //| ClassifyRetcode - transient vs permanent broker outcomes | //+------------------------------------------------------------------+ ENUM_EXEC_OUTCOME ClassifyRetcode(const uint retcode) // trade server return code { switch(retcode) // dispatch on the return code { case TRADE_RETCODE_DONE: // full fill case TRADE_RETCODE_DONE_PARTIAL: // partial fill case TRADE_RETCODE_PLACED: // order accepted, fill pending return(EXEC_ACCEPTED); // final: ACK case TRADE_RETCODE_REQUOTE: // price moved case TRADE_RETCODE_PRICE_CHANGED: // price moved case TRADE_RETCODE_PRICE_OFF: // no quotes right now case TRADE_RETCODE_TOO_MANY_REQUESTS: // throttled case TRADE_RETCODE_LOCKED: // request locked for processing case TRADE_RETCODE_FROZEN: // order or position frozen case TRADE_RETCODE_TIMEOUT: // outcome unknown: reconciled on the next pass case TRADE_RETCODE_CONNECTION: // outcome unknown: reconciled on the next pass case TRADE_RETCODE_MARKET_CLOSED: // session closed: retried until the signal is stale return(EXEC_TRANSIENT_FAILURE); // leave pending default: // NO_MONEY, INVALID_VOLUME, INVALID_PRICE, return(EXEC_PERMANENT_FAILURE); // TRADE_DISABLED, INVALID_STOPS, REJECT, 0, ... } }
DONE, DONE_PARTIAL, and PLACED are accepted: accepted by the broker, not necessarily filled, and while a PLACED order is active the guard blocks new entries. Price and load errors are retried; TIMEOUT and CONNECTION stay pending and are reconciled before any resend.
MARKET_CLOSED is transient, because a closed session is a temporary condition: the signal is retried until it is older than MaxSignalAgeSeconds, then rejected as stale, so a signal from a closed session is dropped after a minute, not executed at the reopen. Everything else, such as NO_MONEY, is final. The same age limit bounds every retry, which is why one day of journal history is plenty.
Setting up the environment
By now the design is complete on paper. Running it is short work, but the order in which the components start matters more than any version. Redis must be 5.0 or later, where consumer groups appeared; redis-cli ping should answer PONG. Python needs one package, installed with pip install redis. In MetaTrader 5, add http://127.0.0.1:8000 under Tools > Options > Expert Advisors > Allow WebRequest for listed URL, and switch Algo Trading on.
Then start python gateway.py, which prints "Gateway running on port 8000", and python strategy_consumer.py, before attaching the EA, because the strategy's group starts at $ and would miss earlier bars. Finally, redis-cli XINFO GROUPS signal.events should list mql5-{symbol}-{magic}, and XINFO GROUPS market.events the group strategy.
My own first run still failed, for a configuration reason: plain EURUSD was not tradeable on the demo account, and the guard said so in the log until the EA moved to EURUSD#.
The defaults deserve a closer look, because several of them are design decisions in disguise. Every setting is an EA input or a code constant. Each default is a deliberate constraint, a localhost or demo example, a broker-specific value, or a safe placeholder (Table 5).
| Input | Default | Kind | Why |
|---|---|---|---|
| GatewayURL | http://127.0.0.1:8000 | Localhost example | Loopback only; any port works if the WebRequest list matches |
| TimerIntervalSeconds | 2 | Deliberate constraint | Signals arrive within about two seconds of the bar; faster polling only adds load |
| MaxSignalAgeSeconds | 60 | Deliberate constraint | One M1 bar; also bounds every retry; raise it on higher timeframes |
| MaxSpreadPoints | 50 | Broker-specific value | Five pips on a five-digit major; tune per instrument |
| DefaultVolume | 0.10 | Demo example | Normalized to the broker's step anyway |
| MagicNumber | 123456 | Safe placeholder | Must be unique per EA: ownership, group, and journal depend on it |
Table 5. EA inputs, the kind of each default, and why.
The code constants follow the same logic. The 2000 ms WebRequest() timeout assumes a local gateway, where a slower reply means a failure. The batch of 10 signals per tick bounds the work per tick. The 10 points of slippage are broker-specific, one pip on a five-digit symbol. The one-day JOURNAL_RETENTION far exceeds any signal's useful life.
On the Python side, port 8000 and localhost:6379 (REDIS_HOST, REDIS_PORT) are local defaults, NEW_READ_BLOCK_MS of 1000 keeps the strategy loop responsive without BLOCK 0, and STRATEGY_VERSION, simple-1, is part of every signal ID so a changed strategy never reuses old IDs. TestStopAfterOrderSend stays off outside testing. The journal itself is state, not configuration: deleting it only removes protection for signals that are still fresh.
Testing on a demo account
With that in place, the prototype was run for real. The tests ran on a demo hedging account, on M1, with MetaTrader 5 under Wine on Linux Mint and Redis and Python on the same machine. The first run was supposed to show the normal path. Instead, the log repeated two lines every two seconds:
16:46:37.850 Duplicate signal ignored: SIGNAL:BAR_CLOSED:EURUSD:M1:1792593480:simple-1 16:46:37.850 ACK not confirmed for 1792582741224-0: HTTP error code: 1003 (a pending entry is redelivered and deduplicated) 16:46:39.879 Duplicate signal ignored: SIGNAL:BAR_CLOSED:EURUSD:M1:1792593480:simple-1 16:46:39.880 ACK not confirmed for 1792582741224-0: HTTP error code: 1003 (a pending entry is redelivered and deduplicated)
A signal handled in an earlier run was in the journal, but its ACK never got through, so Redis returned it more than 60 times in just over two minutes. The journal caught it every time, nothing was processed twice, and no failed ACK was counted as a success. The duplicate protection held under a transport fault the design had not anticipated.
The cause turned out to be more interesting than the symptom. Comparing the Experts log with the gateway's request log second by second showed that none of the failing requests reached the gateway, and that failures alternated exactly with successes. That pattern points to connection reuse: the terminal kept a connection the gateway had already closed, the first request on it failed, and the next one succeeded on a fresh connection. After Connection: close, no 1003 error appeared again.
The next run, on EURUSD#, was the clean 17:03 cycle shown earlier. The gateway's request log, switched on for that run and condensed here, confirms the order of events:
17:03:00 POST /publish BAR_CLOSED 17:03:01 GET /consume the signal is returned 17:03:01 POST /publish ORDER_ATTEMPTED 17:03:02 POST /ack the signal is acknowledged 17:03:02 POST /publish DEAL_EXECUTED
ORDER_ATTEMPTED left before the order did, and the ACK came only after the broker's accepted reply.
The following night, a position from an earlier session was still open when the EA started, so the guard refused every signal. It recognized that position purely by its magic number:
01:21:00.928 Published BAR_CLOSED event: BAR_CLOSED:EURUSD#:M1:1792632000 01:21:01.046 Signal rejected: SIGNAL:BAR_CLOSED:EURUSD#:M1:1792632000:simple-1 | Reason: EA-owned position already exists on EURUSD#
Each refusal was journaled as state 3 before the ACK, and XINFO GROUPS showed pending 0 throughout:
SIGNAL:BAR_CLOSED:EURUSD#:M1:1792632000:simple-1|3|1792632060|0|0 SIGNAL:BAR_CLOSED:EURUSD#:M1:1792632060:simple-1|3|1792632120|0|0 SIGNAL:BAR_CLOSED:EURUSD#:M1:1792632120:simple-1|3|1792632186|0|0
Finally, I added two broken records with redis-cli, one without an event_id and one with braces in it:
XADD signal.events * event_type SIGNAL_GENERATED source manual_test event_time 1 symbol EURUSD# payload {}
XADD signal.events * event_id BAD{ID} event_type SIGNAL_GENERATED source manual_test event_time 1 symbol EURUSD# payload {} The gateway turned both away on the EA's next poll:
WARNING Rejected signal.events/1792622276177-0: missing or empty event_id WARNING Rejected signal.events/1792622309263-0: event_id contains characters the MQL5 parser cannot read
Both went to the dead-letter stream with their reason and original content, as XRANGE signal.events.dlq - + shows:
1) 1) "1792622277040-0"
2) 1) "original_stream"
2) "signal.events"
3) "original_stream_id"
4) "1792622276177-0"
5) "reason"
6) "missing or empty event_id"
7) "record"
8) "{\"event_type\":\"SIGNAL_GENERATED\",\"source\":\"manual_test\",\"event_time\":\"1\",\"symbol\":\"EURUSD#\",\"payload\":\"{}\"}" The EA's group had nothing pending afterwards, and the Experts log has no line for either record:
1) 1) "name" 2) "mql5-EURUSD_-123456" 3) "consumers" 4) (integer) 1 5) "pending" 6) (integer) 0
Crash recovery and the paths not seen in testing
The crash between OrderSend() and the journal write was not staged live. It was checked with unit tests that run the EA's own ProcessSignal(), ExecuteOrder(), and ReconcileAttempt(), code unchanged, outside the terminal, with OrderCheck(), OrderSend(), and the order history replaced by a simulation. They cover the crash (ATTEMPTING, no ACK), the restart that finds the order (ACCEPTED, then the ACK, no second order), an unreadable history (UNKNOWN, not "no order"), and a failed write after an accepted order (no ACK).
To see it live, set TestStopAfterOrderSend to true with no EA position open and wait for a signal. The EA sends the order, logs "TEST: stopping after OrderSend()", and removes itself, leaving the journal at ATTEMPTING and the signal pending. Attach it again with the switch off, and the log should report that the signal "already reached the broker" with the same order ticket, and exactly one position should be open.
Partial fills, OrderCheck() refusals, and temporary broker refusals such as MARKET_CLOSED follow from the code above but did not occur in these runs; each leaves its own line in the Experts log. Behind every path is one rule: a failure is either final, and then it is written down and acknowledged, or it is not, and the signal stays pending until it succeeds, goes stale, or is reconciled. Nothing is acknowledged on hope.
Guarantees and limitations of the prototype
The tests show what the prototype does. The remaining question is what it promises, and as a research prototype, some of its guarantees are intentionally limited.
Under four assumptions (a magic number no other program uses, an available order history, an intact journal, and at most one EA-owned position or active order per symbol), the prototype:
- never acts again on a signal whose final state is journaled, and reconciles an interrupted attempt with the broker before any resend;
- acknowledges only after the final state is on disk, and only counts an ACK that Redis confirms;
- dead-letters every record that breaks the contract, with its reason;
- publishes ORDER_ATTEMPTED before every order, and sends no order without it;
- keeps volume, checks, and ownership in MQL5.
The limitations matter just as much:
- Delivery is at-least-once; the EA, not Redis, filters repeats.
- Reconciliation is a heuristic: an order the server accepts late, or the terminal shows late after a reconnect, can be missed and resent. The EA does not reconcile while disconnected, and the active-order check blocks a resend whenever the first order is already visible, but neither closes the gap completely. The tag-less time fallback can mistake another EA-owned order for the attempt, which is safe because nothing is resent.
- ACCEPTED is not a fill: a cancelled PLACED order is not retried, and only the missing DEAL_EXECUTED shows it.
- Without the journal, a restart loses protection for signals that are still fresh.
- Fail-closed auditing costs availability: no ORDER_ATTEMPTED, no order.
- Nothing times out while the EA or gateway is stopped, and pending entries are resolved when next read; there is no XAUTOCLAIM or XCLAIM, and no DLQ tooling beyond XRANGE.
- XACK does not say why it removed nothing, so the EA treats every such case as unconfirmed.
- WebRequest() is synchronous, so a slow gateway blocks OnTimer() for up to two seconds per request, and the single-threaded gateway suits a few EAs on one machine.
- On netting accounts the EA cannot share its symbol; each extra EA reads every signal again, and a new group sees only later signals.
- There is no authentication. A local process can forge signals; each is treated like a real one and can open at most one EA-owned position or order of the configured size at a time, and another once that position is closed or that order is cancelled. Binding to 127.0.0.1 keeps out the network, not local processes. Production needs Redis ACLs, gateway authentication, and signed signals.
- The live tests ran only under Wine on Linux, not on a Windows terminal.
These limits are deliberate scope choices for a research prototype, not unnoticed defects.
Conclusion
The prototype does not replace polling. What it replaces is ambiguity. The EA still polls every two seconds, but every answer now has a defined meaning: a signal has an identity that survives restarts, an ACK counts only when Redis confirms it and the EA has written down what it did, unusable messages leave through the dead-letter stream, and every broker reply is accepted, retried, or final.
Redis Streams gives the system an explicit delivery model, while MQL5 remains the only component allowed to decide whether a trade is valid, safe, and actually sendable. That boundary is the point of the design. The next steps are cross-consumer recovery with XAUTOCLAIM, signed and authenticated signals, an asynchronous transport, and a complete JSON layer.
Project Files:
| # | Name | Type | Description |
|---|---|---|---|
| 1 | EventTypes.mqh | Include | Event types and the EventEnvelope struct |
| 2 | EventSerializer.mqh | Include | Constrained JSON serializer and payload helpers |
| 3 | EventPublisher.mqh | Include | Posts events to the gateway through WebRequest() |
| 4 | EventConsumer.mqh | Include | Consumes, acknowledges, and rejects signals in the EA's own group; counts an ACK only on "acknowledged":true |
| 5 | ExecutionGuard.mqh | Include | Hard checks before execution, including freshness, ownership, spread, and volume normalization |
| 6 | SignalJournal.mqh | Include | Append-only, flushed journal of signal states and the order-to-signal link |
| 7 | ReactiveEA.mq5 | Expert | Main EA: deduplication, ACK after the journal write, reconciliation, OrderCheck() before the fail-closed ORDER_ATTEMPTED, retcode classification, DEAL_EXECUTED, and the test switch |
| 8 | gateway.py | Python | HTTP gateway for /publish, /consume, /ack, and /reject; enforces the contract, dead-letters violations, reports XACK results, sends Connection: close |
| 9 | redis_stream.py | Python | Redis Streams wrapper; ignores only BUSYGROUP, never sends BLOCK 0 |
| 10 | strategy_consumer.py | Python | Validates complete OHLC bars and publishes SIGNAL_GENERATED with the bar's server time |
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.
Testing Random Walk vs Mean Reversion in MQL5: Implementing the Lo-MacKinlay Variance Ratio Test
From Evidence to Execution (Part 1): How to Log, Measure, and Decode a Trading Strategy Inside MetaTrader 5 AI
From Basic to Intermediate: Subwindows (II)
Archimedes Optimization Algorithm (AOA)
- Free trading apps
- Over 8,000 signals for copying
- Economic news for exploring financial markets
You agree to website policy and terms of use