Lesson 7.1.3 · 75 min

Patterns: events, message queues and hexagonal architecture

The breakout strategy from the previous lesson, turned into a core that does not know where its bars come from or where its orders go. This lesson surrounds it with ports and adapters, checks the dependency rule, connects it to an event bus and then to queues between processes (Redis Pub/Sub and Streams, ZeroMQ, Kafka), and measures what each one loses or duplicates when the consumer crashes or falls behind.

Fig. — What happens to messages when the consumer crashes or stops reading: ZeroMQ drops what does not fit in the queue, Pub/Sub loses what is published while nobody listens, Streams loses nothing but duplicates if the consumer is not idempotent

03/03

Module: 7.1 · Software engineering for trading
Duration: 75 min
Prerequisites: 2.4.2 · Event-driven backtest from scratch: loop, portfolio, orders and fills, 4.5.1 · Live system architecture: from data to reconciliation, 4.5.5 · Fault tolerance: idempotency, persistent state and reconciliation at startup, 7.1.1 · Project structure, configuration and quality: from notebook to repository, 7.1.2 · Performance: measure, profile and compile only what is needed
Author: Jesús Cuesta
Updated: 05 Nov 2026

Level 7 lessons

One core, three worlds

Lesson 4.5.1 built a live system with an event bus and a logical clock, and checked that the same components serve for backtest, paper and live trading. This lesson stops at the patterns that make it possible and at what happens when the pieces no longer live in the same process: the data reader on one machine, the strategy on another, the order manager on a third, connected by message queues.

The underlying pattern is hexagonal architecture, or ports and adapters, which Alistair Cockburn described in 20051Note 1 · Alistair Cockburn, “Hexagonal architecture” (https://alistair.cockburn.us/hexagonal-architecture/). Accessed on October 23, 2026.: the application has a core that depends on nothing external and talks to the outside through ports (interfaces defined by the core, according to what it needs); each concrete technology is plugged in with an adapter. NautilusTrader (lesson 2.4.5) is built this way: the same strategy runs against a backtest engine or against an Interactive Brokers adapter.

Mermaid source

flowchart TB
  P[Parquet] --> M[Mercado port]
  R[Redis Streams] --> M
  M --> N[Ruptura core]
  N --> E[Ejecutor port]
  E --> S[simulated broker]
  E --> B[live broker]

The lesson's core is the breakout strategy from lesson 7.1.2 rewritten to receive one bar at a time (an event) and return intents: “I want to hold 1 contract”, “I want to hold 0”. It does not decide how that is executed, nor does it know whether the bar comes from a file or from a queue. Events are immutable dataclasses with a sequence number, which is what allows ordering, deduplicating and replaying them:

IN [01]python

from collections import deque
from dataclasses import dataclass

import pandas as pd

@dataclass(frozen=True)
class Barra:
    seq: int
    instrumento: str
    maximo: float
    minimo: float
    cierre: float
    nueva_sesion: bool

@dataclass(frozen=True)
class Objetivo:
    seq: int                    # the bar that triggers it
    instrumento: str
    posicion: int
    precio_ref: float
    motivo: str

@dataclass(frozen=True)
class Ejecucion:
    seq: int
    instrumento: str
    cantidad: int
    precio: float

class Ruptura:
    """n-bar breakout with a k-ATR stop (lesson 7.1.2), one bar
    at a time. No I/O: receives events, returns targets."""

    def __init__(self, n=30, k=4.0, alfa=2 / 15):
        self.n, self.k, self.alfa = n, k, alfa
        self.maximos = deque(maxlen=n)
        self.atr, self.pos, self.pico = None, 0, 0.0

    def al_recibir(self, b):
        rango = b.maximo - b.minimo
        self.atr = (rango if self.atr is None
                    else self.alfa * rango + (1 - self.alfa) * self.atr)
        antes, motivo = self.pos, ""
        if len(self.maximos) < self.n or b.nueva_sesion:
            self.pos, motivo = 0, "session"
        elif self.pos == 0:
            if b.cierre > max(self.maximos):
                self.pos, self.pico, motivo = 1, b.cierre, "breakout"
        else:
            self.pico = max(self.pico, b.cierre)
            if b.cierre < self.pico - self.k * self.atr:
                self.pos, motivo = 0, "stop"
        self.maximos.append(b.maximo)
        if self.pos == antes:
            return []
        return [Objetivo(b.seq, b.instrumento, self.pos, b.cierre,
                         motivo)]

The deque of 30 highs replaces the iloc[i - n:i] of the pandas version: the core keeps what it needs from its past and nothing more. The ports are what the rest of the program offers it: a source of bars and an executor of targets. They are written as Protocol (lesson 0.4.2): any class with those methods will do, without inheriting from anything. With an adapter that reads the Parquet file and a simulated executor, the core runs a backtest:

IN [02]python

from collections.abc import Iterable
from typing import Protocol

class Mercado(Protocol):
    def barras(self) -> Iterable[Barra]: ...

class Ejecutor(Protocol):
    def enviar(self, o: Objetivo) -> list[Ejecucion]: ...

class MercadoParquet:
    """Adapter: bars from a Parquet file, with their sequence number."""

    def __init__(self, ruta, instrumento, filas):
        df = pd.read_parquet(ruta).iloc[:filas]
        dia = df["ts"].dt.floor("D")
        self.df = df.assign(nueva=dia != dia.shift())
        self.instrumento = instrumento

    def barras(self):
        filas = self.df[["maximo", "minimo", "cierre", "nueva"]]
        for seq, (mx, mn, c, nueva) in enumerate(
                filas.itertuples(index=False), 1):
            yield Barra(seq, self.instrumento, mx, mn, c, bool(nueva))

class EjecutorSimulado:
    """Adapter: moves the position to the target at the ref price."""

    def __init__(self):
        self.posiciones = {}

    def enviar(self, o):
        q = o.posicion - self.posiciones.get(o.instrumento, 0)
        self.posiciones[o.instrumento] = o.posicion
        if q == 0:
            return []
        return [Ejecucion(o.seq, o.instrumento, q, o.precio_ref)]

def ejecutar(mercado, estrategia, ejecutor, anotar=lambda ev: None):
    for b in mercado.barras():
        anotar(b)
        for o in estrategia.al_recibir(b):
            anotar(o)
            for e in ejecutor.enviar(o):
                anotar(e)

def resumen(eventos, n_barras):
    objetivos = [e for e in eventos if isinstance(e, Objetivo)]
    entradas = [o.seq for o in objetivos if o.posicion == 1]
    salidas = [o.seq for o in objetivos if o.posicion == 0]
    dentro = sum(s - e for e, s in zip(entradas, salidas))
    if len(entradas) > len(salidas):
        dentro += n_barras - entradas[-1] + 1
    return len(entradas), dentro / n_barras

diario = []
ejecutar(MercadoParquet("hf5_barras_1min.parquet", "FUT", 20_000),
         Ruptura(), EjecutorSimulado(), diario.append)
entradas, fraccion = resumen(diario, 20_000)
print(f"{entradas} entries, in the market {fraccion:.1%} "
      "of the bars")
print({t.__name__: sum(isinstance(e, t) for e in diario)
       for t in (Barra, Objetivo, Ejecucion)})

OUT [02]Ports, adapters and a backtest

226 entries, in the market 40.7% of the bars
{'Barra': 20000, 'Objetivo': 451, 'Ejecucion': 451}

The same 226 entries and the same 40.7% in the market as the backtest of lesson 7.1.2 over the same 20,000 bars. This is the check that matters when reorganizing code: the behavior does not change. The journal gathers the 20,000 bars and the targets and fills they triggered, in the order in which they happened.

The dependency rule, checked

That the core does not depend on the infrastructure is an easy rule to break: all it takes is someone importing the Redis client “just for one thing” inside the strategy. From then on, the strategy can no longer be tested without Redis. The rule is checked like any other, with a tool. The cell writes a toy package with the shape of a real project and reads its imports with the ast module, without executing anything:

IN [03]python

import ast
from pathlib import Path

PAQUETE = {
    "bot/nucleo/eventos.py": "from dataclasses import dataclass\n",
    "bot/nucleo/ruptura.py": "from collections import deque\n"
                             "from bot.nucleo.eventos import Barra\n",
    "bot/adaptadores/redis_mercado.py":
        "import redis\nfrom bot.nucleo.eventos import Barra\n",
    "bot/app.py": "from bot.adaptadores import redis_mercado\n"
                  "from bot.nucleo import ruptura\n",
}
for ruta, texto in PAQUETE.items():
    Path(ruta).parent.mkdir(parents=True, exist_ok=True)
    Path(ruta).write_text(texto)

PERMITIDAS = ("bot.nucleo", "collections", "dataclasses", "math",
              "typing", "enum", "decimal")

def importaciones(fichero):
    for nodo in ast.walk(ast.parse(Path(fichero).read_text())):
        if isinstance(nodo, ast.Import):
            yield from (a.name for a in nodo.names)
        elif isinstance(nodo, ast.ImportFrom) and nodo.module:
            yield nodo.module

def violaciones(raiz="bot/nucleo"):
    return [(f.name, m) for f in sorted(Path(raiz).rglob("*.py"))
            for m in importaciones(f)
            if not any(m == p or m.startswith(p + ".")
                       for p in PERMITIDAS)]

print("clean core:", violaciones())
with open("bot/nucleo/ruptura.py", "a") as f:
    f.write("import redis\n"
            "from bot.adaptadores.redis_mercado import leer\n")
print("with a shortcut:", violaciones())

OUT [03]The core's imports, read with ast

clean core: []
with a shortcut: [('ruptura.py', 'redis'), ('ruptura.py', 'bot.adaptadores.redis_mercado')]

The list of allowed imports is a design decision written in code: the core may use the parts of the standard library that do no I/O and its own package. In a real project, the usual tool is import-linter, which reads contracts (forbidden, layers, independence…) from pyproject.toml and runs in pre-commit and in continuous integration like the checks of lesson 7.1.12Note 2 · import-linter (https://import-linter.readthedocs.io/): the forbidden, protected, layers, independence and acyclic siblings contract types. Accessed on October 23, 2026..

A queued event bus

Inside a process, the pieces communicate through a bus: whoever produces an event publishes it and whoever needs it subscribes by type. Lesson 4.5.1 used one. There is a design detail that decides whether the system is correct: what happens when a subscriber publishes in turn. A recursive bus delivers the new event on the spot, inside the delivery of the previous one; a queued bus puts it at the end and delivers it when the current one finishes. The cell builds both versions, with three subscribers: the strategy (bars), the executor (targets) and a risk check (bars), subscribed after the strategy, which records the position it sees on each bar:

IN [04]python

from collections import defaultdict

class Bus:
    def __init__(self, con_cola=True):
        self.subs = defaultdict(list)
        self.con_cola, self.cola = con_cola, deque()
        self.ocupado = False
        self.diario, self.fallidos = [], []

    def suscribir(self, tipo, f, critico=False):
        self.subs[tipo].append((f, critico))

    def publicar(self, ev):
        if not self.con_cola:
            return self._entregar(ev)
        self.cola.append(ev)
        if self.ocupado:
            return                       # delivered in its turn
        self.ocupado = True
        try:
            while self.cola:
                self._entregar(self.cola.popleft())
        finally:
            self.ocupado = False

    def _entregar(self, ev):
        self.diario.append(ev)
        for f, critico in self.subs[type(ev)]:
            try:
                f(ev)
            except Exception as e:
                if critico:
                    raise
                self.fallidos.append((type(ev).__name__, ev.seq,
                                      f.__name__, repr(e)))

def montar(bus, traza):
    est, ejec = Ruptura(), EjecutorSimulado()

    def estrategia(b):
        traza.append(f"strategy sees bar {b.seq}")
        for o in est.al_recibir(b):
            bus.publicar(o)

    def ejecutor(o):
        traza.append(f"target {o.posicion} from bar {o.seq}")
        for e in ejec.enviar(o):
            bus.publicar(e)

    def riesgo(b):
        traza.append(f"risk sees bar {b.seq} with position "
                     f"{ejec.posiciones.get('FUT', 0)}")

    bus.suscribir(Barra, estrategia, critico=True)
    bus.suscribir(Objetivo, ejecutor, critico=True)
    bus.suscribir(Barra, riesgo, critico=True)
    return est, ejec

barras = list(MercadoParquet("hf5_barras_1min.parquet", "FUT",
                             20_000).barras())
primera = next(o.seq for o in diario if isinstance(o, Objetivo)
               and o.posicion == 1)
for con_cola in (False, True):
    bus, traza = Bus(con_cola), []
    montar(bus, traza)
    for b in barras[:primera]:
        traza.clear()
        bus.publicar(b)
    print("queued:" if con_cola else "recursive:", " | ".join(traza))

OUT [04]Recursion versus queue

recursive: strategy sees bar 170 | target 1 from bar 170 | risk sees bar 170 with position 1
queued: strategy sees bar 170 | risk sees bar 170 with position 0 | target 1 from bar 170

With the recursive bus, the risk check sees the entry bar with the position already at 1: the strategy published the target, the executor executed it and all of that happened before the bar reached the next subscriber. For the risk check, the fill triggered by the bar happened before the bar. In this example it is a wrong label; with an exposure limit evaluated on every bar, it is an order rejected or accepted by mistake, and the result depends on the order in which the handlers subscribed. With the queue, everyone sees the bar before whatever it triggers exists, and the order of the journal is the causal order.

The bus has two other responsibilities. The first is deciding what to do with a subscriber that fails. A metrics calculation with a bug (it looks up the session's reference before having stored it) fails on the first bar of every session; it must not stop the system, but it must not vanish silently either. A critical subscriber (strategy, executor, risk) that fails, on the other hand, must stop it. The second is keeping the journal: with it, the state is rebuilt and the session replayed.

IN [05]python

bus, traza = Bus(), []
est, ejec = montar(bus, traza)
metricas, referencia = [], {}

def metrica_sesion(b):
    """Return since the session's first close. Bug: on the
    first bar of each session there is no reference yet."""
    if b.nueva_sesion:
        referencia.pop(b.instrumento, None)
    else:
        referencia.setdefault(b.instrumento, b.cierre)
    metricas.append(b.cierre / referencia[b.instrumento] - 1)

bus.suscribir(Barra, metrica_sesion)         # non-critical
for b in barras:
    bus.publicar(b)
print("isolated failures:", len(bus.fallidos), "|", bus.fallidos[0])

ejecs = [e for e in bus.diario if isinstance(e, Ejecucion)]
pos = sum(e.cantidad for e in ejecs)
print("position from the journal:", pos,
      "| same as the executor's:", pos == ejec.posiciones["FUT"])

repetida = Ruptura()
objetivos = [o for b in bus.diario if isinstance(b, Barra)
             for o in repetida.al_recibir(b)]
originales = [o for o in bus.diario if isinstance(o, Objetivo)]
print("identical replay:", objetivos == originales,
      f"({len(objetivos)} targets)")

OUT [05]Isolated failures and replay from the journal

isolated failures: 52 | ('Barra', 1, 'metrica_sesion', "KeyError('FUT')")
position from the journal: 1 | same as the executor's: True
identical replay: True (451 targets)

The first bar of every session fails in the metrics subscriber, each failure is recorded with its sequence number and the error, and the system carries on; the final position and the targets are the same as without that subscriber. Replaying the journal's bars with a new strategy gives exactly the same targets: the core is deterministic, and the journal is enough to rebuild the session. It is the idea of event sourcing3Note 3 · Martin Fowler, “Event Sourcing” (https://martinfowler.com/eaaDev/EventSourcing.html). Accessed on October 23, 2026.: the state is a function of the sequence of events, and the sequence is what gets stored.

Queues between processes: Redis Pub/Sub and Streams

When the data reader and the strategy are separate processes (because one restarts without the other, because several strategies read the same data, because they live on different machines), the bus becomes a messaging service. Redis, which many systems already use as a cache, offers two with very different guarantees. The lesson uses fakeredis, which implements Redis commands in memory with the same redis-py client; with a real server only the line that creates the connection changes.

Pub/Sub is broadcasting: the message reaches whoever is subscribed at that moment and Redis does not store it. The documentation calls it “at-most-once” semantics4Note 4 · Redis, “Redis Pub/Sub” (https://redis.io/docs/latest/develop/pubsub/), section “Delivery semantics”: “Redis' Pub/Sub exhibits at-most-once message delivery semantics”. Accessed on October 23, 2026.. The experiment: 100 bars published, and the subscriber goes down and comes back between bar 41 and bar 60:

IN [06]python

import json

import fakeredis
import redis

r = fakeredis.FakeRedis(decode_responses=True)

def leer_pubsub(ps):
    vistos = []
    while (m := ps.get_message(timeout=0)) is not None:
        if m["type"] == "message":
            vistos.append(json.loads(m["data"])["seq"])
    return vistos

ps = r.pubsub()
ps.subscribe("barras")
recibidos, receptores = [], []
for b in barras[:100]:
    if b.seq == 41:
        ps.unsubscribe("barras")              # the process dies
        recibidos += leer_pubsub(ps)
    if b.seq == 61:
        ps.subscribe("barras")                # and comes back
    receptores.append(r.publish("barras", json.dumps({"seq": b.seq})))
recibidos += leer_pubsub(ps)
perdidos_pubsub = sorted(set(range(1, 101)) - set(recibidos))
print(len(recibidos), "received; lost:", perdidos_pubsub[0], "to",
      perdidos_pubsub[-1], "| receivers per message:",
      sorted(set(receptores)))

OUT [06]Pub/Sub with the subscriber down for a while

80 received; lost: 41 to 60 | receivers per message: [0, 1]

Twenty bars lost, with no error: publish returns the number of receivers, 0 while the subscriber is away, and that is the only clue. Pub/Sub is suitable for what can be lost (a dashboard that shows the latest price, a notification) and not for what has to be processed in full.

Streams is a log: each message is appended with an increasing identifier and stays there. A consumer group distributes the messages among the group's consumers and remembers two things: how far it has delivered and which delivered messages have not been acknowledged yet (the pending entries list, PEL). A message is acknowledged with XACK once its effect is done5Note 5 · Redis, “Redis Streams” (https://redis.io/docs/latest/develop/data-types/streams/): consumer groups, the pending entries list (PEL), XPENDING, XCLAIM, XAUTOCLAIM and the delivery counter. Accessed on October 23, 2026.:

Mermaid source

stateDiagram-v2
  [*] --> InStream: XADD
  InStream --> Pending: XREADGROUP
  Pending --> Acked: XACK
  Pending --> Pending: crash and reread
  Pending --> Claimed: XCLAIM
  Claimed --> Acked: XACK

The equivalent experiment: the consumer processes the bars one at a time and crashes right after processing bar 41, before acknowledging it. On startup, it first reads its own pending entries (XREADGROUP with the identifier 0) and then the new ones (>). Two consumers: one that does not remember what it has done and another that stores the effect and the last processed sequence number in the same Redis transaction (MULTI/EXEC):

IN [07]python

class Caida(BaseException):
    """The process dies: nothing that follows is executed."""

def consumir(r, nombre, idempotente, caer_en=None):
    procesados = []

    def procesar(id_, campos):
        seq = int(campos["seq"])
        if idempotente and seq <= int(r.get("ultimo_seq") or 0):
            return                               # already done
        with r.pipeline(transaction=True) as t:  # effect + marker
            t.rpush("efectos", seq)
            t.set("ultimo_seq", seq)
            t.execute()
        procesados.append(seq)
        if seq == caer_en:
            raise Caida(f"dies after processing {seq}, without XACK")

    for desde in ("0", ">"):                     # pending, new
        while True:
            resp = r.xreadgroup("estrategia", nombre,
                                {"barras": desde}, count=1)
            if not resp or not resp[0][1]:
                break
            [(id_, campos)] = resp[0][1]
            procesar(id_, campos)
            r.xack("barras", "estrategia", id_)
    return procesados

resultados = {}
for idempotente in (False, True):
    r = fakeredis.FakeRedis(decode_responses=True)
    for b in barras[:100]:
        r.xadd("barras", {"seq": b.seq}, id=f"{b.seq}-0")
    r.xgroup_create("barras", "estrategia", id="0")
    try:
        consumir(r, "c1", idempotente, caer_en=41)
    except Caida as c:
        pend = r.xpending_range("barras", "estrategia", "-", "+", 10)
        print(c, "| pending:", [(p["message_id"],
              p["times_delivered"]) for p in pend])
    consumir(r, "c1", idempotente)               # restart
    efectos = [int(x) for x in r.lrange("efectos", 0, -1)]
    dobles = sorted({s for s in efectos if efectos.count(s) > 1})
    resultados[idempotente] = (len(set(efectos)), dobles)
    print("idempotent:" if idempotente else "naive:",
          len(set(efectos)), "bars with an effect; duplicated:", dobles,
          "| pending at the end:",
          r.xpending("barras", "estrategia")["pending"])

OUT [07]Streams: crash before XACK, with and without idempotency

dies after processing 41, without XACK | pending: [('41-0', 1)]
naive: 100 bars with an effect; duplicated: [41] | pending at the end: 0
dies after processing 41, without XACK | pending: [('41-0', 1)]
idempotent: 100 bars with an effect; duplicated: [] | pending at the end: 0

No bar is lost: on crashing, bar 41 remains pending for consumer c1 with one delivery, and on startup it is delivered again. The naive consumer processes it again: its effect is there twice. The idempotent one sees that bar 41 is already in ultimo_seq (written in the same transaction as the effect, so either both are there or neither is) and only acknowledges it. It is the rule of lesson 4.5.5 carried over to messaging: the network and the queues give “at least once”; the idempotent consumer turns that into the effect of “exactly once”. If c1 did not come back, another consumer in the group could take over its pending entries with XCLAIM or XAUTOCLAIM, stating how long they must have gone unacknowledged.

ZeroMQ: messaging without a broker

ZeroMQ is not a server but a socket library with patterns: processes connect directly, with no intermediary that stores anything. That makes it fast and simple to deploy, and shifts whatever persistence is needed to the application. The cell uses the inproc:// transport (between threads of the same process; with tcp:// the code is the same) and three patterns: PUSH/PULL to distribute work, PUB/SUB to broadcast prices with prefix filtering, and the queue limit (high water mark, HWM) when the subscriber does not read:

IN [08]python

import zmq

ctx = zmq.Context()

def drenar(sock):
    vistos = []
    while sock.poll(0):
        vistos.append(sock.recv_string())
    return vistos

push = ctx.socket(zmq.PUSH)
push.bind("inproc://trabajo")
obreros = [ctx.socket(zmq.PULL) for _ in range(2)]
for o in obreros:
    o.connect("inproc://trabajo")
for i in range(6):
    push.send_string(f"backtest {i}")
print("PUSH/PULL:", [drenar(o) for o in obreros])

pub = ctx.socket(zmq.XPUB)
pub.bind("inproc://precios")
sub = ctx.socket(zmq.SUB)
sub.connect("inproc://precios")
sub.setsockopt_string(zmq.SUBSCRIBE, "ORSC ")
print("subscription received by the XPUB:", pub.recv())
for txt in ("ORSC 234.84", "ORUS 716.82", "ORSC 234.90"):
    pub.send_string(txt)
print("PUB/SUB with filter:", drenar(sub))

def lento(hwm, conflate=False):
    """The SUB does not read while 5,000 prices are published."""
    p, s = ctx.socket(zmq.XPUB), ctx.socket(zmq.SUB)
    p.setsockopt(zmq.SNDHWM, hwm)
    if conflate:
        s.setsockopt(zmq.CONFLATE, 1)
    else:
        s.setsockopt(zmq.RCVHWM, hwm)
    p.bind(f"inproc://hwm{hwm}{conflate}")
    s.connect(f"inproc://hwm{hwm}{conflate}")
    s.setsockopt_string(zmq.SUBSCRIBE, "")
    p.recv()                                  # the subscription
    for i in range(5_000):
        p.send_string(str(i))
    vistos = drenar(s)
    p.close(0)
    s.close(0)
    return vistos

for hwm in (1, 10, 100, 1000):
    v = lento(hwm)
    print(f"HWM {hwm:>4}: {len(v):>4} received, from {v[0]} to {v[-1]}")
print("CONFLATE:", lento(10, conflate=True))
for s in (push, *obreros, pub, sub):
    s.close(0)

OUT [08]ZeroMQ: distribution, broadcasting and queue limit

PUSH/PULL: [['backtest 0', 'backtest 2', 'backtest 4'], ['backtest 1', 'backtest 3', 'backtest 5']]
subscription received by the XPUB: b'\x01ORSC '
PUB/SUB with filter: ['ORSC 234.84', 'ORSC 234.90']
HWM    1:    2 received, from 0 to 1
HWM   10:   20 received, from 0 to 19
HWM  100:  200 received, from 0 to 199
HWM 1000: 2000 received, from 0 to 1999
CONFLATE: ['4999']

Fig. — A ZeroMQ subscriber that does not read while 5,000 prices are published: with a queue limit it receives exactly the sum of both limits, and they are the oldest prices; with CONFLATE, only the latest

PUSH distributes in turns (0, 2, 4 to one worker; 1, 3, 5 to the other): it is useful for launching a sweep of backtests across several processes. The SUB only receives what starts with ORSC ; the filter is applied on the publisher's side, which receives the subscription as a message (\x01 plus the prefix). Using XPUB and waiting for that message before publishing avoids the slow joiner problem: with a plain PUB, whatever is published before the subscription arrives is lost.

The part that matters most in trading is the last one. With a subscriber that does not read, the PUB fills the queue up to the HWM on each side and then drops whatever is new, with no error, as the ZeroMQ documentation describes for PUB sockets in the mute state7Note 7 · ZeroMQ, “zmq_socket” (https://libzmq.readthedocs.io/en/latest/zmq_socket.html): a ZMQ_PUB socket that reaches a subscriber's limit drops the messages for it for as long as that state lasts; and “zmq_setsockopt” (https://libzmq.readthedocs.io/en/latest/zmq_setsockopt.html) for ZMQ_SNDHWM, ZMQ_RCVHWM and ZMQ_CONFLATE. Accessed on October 23, 2026.. In the figure, the number received is exactly the sum of the two limits, and what is received are the oldest prices: a slow consumer that reads again trades on stale prices and does not notice. For prices, the sensible choice is usually the opposite: ZMQ_CONFLATE keeps only the last message, which is what the SUB receives (4999). For orders or fills, neither option works: none can be lost, and a persistent log is needed.

Kafka: the distributed log

Kafka takes the idea of Redis Streams to a distributed system: a topic is a log partitioned and replicated across several servers, messages are kept for the configured retention time (days, months or indefinitely) and each consumer group keeps its position (offset) in each partition. Messages with the same key go to the same partition, and Kafka guarantees that a consumer reads each partition in the order in which it was written8Note 8 · Apache Kafka, “Introduction” (https://kafka.apache.org/intro/): messages with the same key go to the same partition, which is read in the same order in which it was written; and “Design” (https://kafka.apache.org/43/design/design/) for the idempotent producer. Accessed on October 23, 2026.: with the instrument as the key, all of ORSC's bars arrive in order even if the topic has many partitions. The idempotent producer prevents a retry from writing a message twice: in the Java client it is enabled by default, whereas in librdkafka, the C library that confluent-kafka uses in Python, it has to be requested (enable.idempotence, default false)9Note 9 · Apache Kafka 4.3, “Producer Configs” (https://kafka.apache.org/43/configuration/producer-configs/): enable.idempotence, default true if no other option prevents it; librdkafka, “CONFIGURATION.md” (https://github.com/confluentinc/librdkafka/blob/master/CONFIGURATION.md): enable.idempotence, default false. Accessed on October 23, 2026.. It requires a cluster, so the code is for reading only:

IN [09]python

from confluent_kafka import Consumer, Producer

prod = Producer({"bootstrap.servers": "kafka:9092",
                 "enable.idempotence": True, "acks": "all"})
for b in barras:
    prod.produce("barras", key=b.instrumento,
                 value=json.dumps({"seq": b.seq, "cierre": b.cierre}))
prod.flush()

cons = Consumer({"bootstrap.servers": "kafka:9092",
                 "group.id": "estrategia",
                 "enable.auto.commit": False,
                 "auto.offset.reset": "earliest"})
cons.subscribe(["barras"])
while (m := cons.poll(1.0)) is not None:
    procesar(json.loads(m.value()))      # idempotent, as before
    cons.commit(message=m)               # after the effect

The discipline is the same as with Streams: commit (commit) after the effect and not before, and an idempotent consumer. Kafka's own advantage is replay: a new consumer can read the topic from the beginning, and a backtest can be fed exactly the messages the live system saw. The price is operating a cluster (or paying for a managed one). For an individual trader with a bot on a server, it is almost always overkill.

	Persistence	Delivery	Ordering	Replay	Operations
In-memory bus	none (own journal)	synchronous	total	from the journal	nothing
Redis Pub/Sub	none	at most once	per channel	no	one Redis
Redis Streams	in Redis	at least once (groups)	total per stream	yes, by id	one Redis
ZeroMQ	none	drops when full (PUB)	per connection	no	nothing (library)
Kafka	replicated log	at least once; exactly once with transactions	per partition	yes, by offset	a cluster

The core on Redis Streams

To close, the core from the beginning, without changing a line, connected to two new adapters: one that reads the bars from a stream with a consumer group and another that writes the targets to another stream with the sequence number as the identifier. Redis rejects an XADD whose identifier is not greater than the stream's last one, so writing the same target twice is impossible: it is the idempotency key of lesson 4.5.5, enforced by the queue. The process crashes halfway, at bar 2,500, and on startup it rebuilds the strategy's state by rereading the bars stream from the beginning:

IN [10]python

class MercadoStream:
    """Adapter: the group's bars; acks each one when control returns."""

    def __init__(self, r, consumidor):
        self.r, self.consumidor = r, consumidor

    def barras(self):
        for desde in ("0", ">"):
            while resp := self.r.xreadgroup(
                    "estrategia", self.consumidor, {"barras": desde},
                    count=100):
                if not resp[0][1]:
                    break
                for id_, c in resp[0][1]:
                    yield Barra(int(c["seq"]), c["ins"],
                                float(c["max"]), float(c["min"]),
                                float(c["cie"]), c["nueva"] == "1")
                    self.r.xack("barras", "estrategia", id_)

class EjecutorStream:
    """Adapter: one target per bar in the `objetivos` stream."""

    def __init__(self, r, caer_en=None):
        self.r, self.caer_en, self.rechazados = r, caer_en, 0

    def enviar(self, o):
        try:
            self.r.xadd("objetivos", {"pos": o.posicion},
                        id=f"{o.seq}-0")
        except redis.ResponseError:
            self.rechazados += 1           # sent before the crash
        return []

def caer(en):
    def anotar(ev):
        if isinstance(ev, Barra) and ev.seq == en:
            raise Caida(f"dies on receiving bar {en}")
    return anotar

N5 = 5_000
r = fakeredis.FakeRedis(decode_responses=True)
for b in barras[:N5]:
    r.xadd("barras", {"seq": b.seq, "ins": b.instrumento,
                      "max": repr(b.maximo), "min": repr(b.minimo),
                      "cie": repr(b.cierre),
                      "nueva": int(b.nueva_sesion)}, id=f"{b.seq}-0")
r.xgroup_create("barras", "estrategia", id="0")

try:
    ejecutar(MercadoStream(r, "c1"), Ruptura(), EjecutorStream(r),
             caer(2_500))
except Caida as c:
    pendientes = r.xpending("barras", "estrategia")["pending"]
    print(c, "| targets written:", r.xlen("objetivos"),
          "| pending:", pendientes)

# restart: state rebuilt from the bars stream
nueva, ejec2 = Ruptura(), EjecutorStream(r)
ultima = r.xinfo_groups("barras")[0]["last-delivered-id"]
for id_, c in r.xrange("barras", "-", ultima):
    b = Barra(int(c["seq"]), c["ins"], float(c["max"]),
              float(c["min"]), float(c["cie"]), c["nueva"] == "1")
    for o in nueva.al_recibir(b):
        ejec2.enviar(o)
pend = r.xpending_range("barras", "estrategia", "-", "+", 100)
for p in pend:
    r.xack("barras", "estrategia", p["message_id"])
ejecutar(MercadoStream(r, "c1"), nueva, ejec2)

en_redis = [(int(i.split("-")[0]), int(c["pos"]))
            for i, c in r.xrange("objetivos")]
diario5 = []
ejecutar(MercadoParquet("hf5_barras_1min.parquet", "FUT", N5),
         Ruptura(), EjecutorSimulado(), diario5.append)
en_proceso = [(o.seq, o.posicion) for o in diario5
              if isinstance(o, Objetivo)]
print("rejected for a repeated id:", ejec2.rechazados)
print("targets in Redis:", len(en_redis), "| identical to the single "
      "process:", en_redis == en_proceso)

OUT [10]The core with two Redis adapters and a crash

dies on receiving bar 2500 | targets written: 57 | pending: 1
rejected for a repeated id: 57
targets in Redis: 122 | identical to the single process: True

At the crash, the targets of the bars before 2,500 are already in the stream and bar 2,500 has been delivered but not acknowledged. On startup, the new strategy knows nothing (its deque, its ATR and its highest close since the entry were lost with the process), so it rereads the bars stream up to the last delivered one; the targets it produces again collide with those already there and Redis rejects them. Then it acknowledges the pending one and continues with the new ones. The result matches the single-process backtest one by one. Two simplifications compared with a real system: rereading the whole stream does not scale to months of bars, and it is replaced by a periodic snapshot of the strategy's state plus the subsequent events; and here the state lives in the process's memory, whereas in live trading the broker knows the position, and the startup reconciles with it (lesson 4.5.5).

The header figure summarizes the four delivery experiments: with the consumer down or slow, Pub/Sub and ZeroMQ lose messages with no error; Streams loses none but duplicates if the consumer is not idempotent; with idempotency, each bar has its effect exactly once.

Exercises

Add to the lesson's bus a critical subscriber that fails (for example, a risk check that raises an exception when the position is 1 and the close drops below a threshold) and check that the queued bus stops and that the queue is left with the undelivered events. What should the system do next?

Solution

IN [11]python

bus, traza = Bus(), []
est, ejec = montar(bus, traza)

class LimiteDeRiesgo(Exception):
    pass

def riesgo_duro(b):
    if ejec.posiciones.get("FUT", 0) == 1 and b.cierre < 5_200:
        raise LimiteDeRiesgo(f"bar {b.seq}: close {b.cierre}")

bus.suscribir(Barra, riesgo_duro, critico=True)
try:
    for b in barras:
        bus.publicar(b)
except LimiteDeRiesgo as e:
    print("stopped:", e)
print("undelivered in the queue:",
      [(type(ev).__name__, ev.seq, ev.posicion) for ev in bus.cola])
print("isolated failures:", len(bus.fallidos))

OUT [11]A critical failure stops the bus

stopped: bar 2427: close 5195.5
undelivered in the queue: [('Objetivo', 2427, 0)]
isolated failures: 0

The failure of a critical subscriber propagates out of the bus and stops publishing; whatever was in the queue at that moment is left undelivered and in plain sight. Here it is the exit target (position 0) that bar 2,427 itself had triggered: the strategy wanted to close and the risk check stopped the system before the order went out. Stopping does not undo the position, so the system is left with one contract and a pending exit. It should move to a locked state (lesson 4.5.5) with an explicit procedure for the open position (close it, hedge it or leave it, decided in advance), alert a person (lesson 4.5.4) and not restart on its own. What it must not do is what a generic except Exception around everything would do: log the error and keep trading as if the risk check had said yes.

With fakeredis, reproduce the case of a consumer that goes down for good: c1 is left with bar 41 pending and a second consumer, c2, claims it with XAUTOCLAIM and processes it. How many times has it been delivered?

Solution

IN [12]python

r = fakeredis.FakeRedis(decode_responses=True)
for b in barras[:100]:
    r.xadd("barras", {"seq": b.seq}, id=f"{b.seq}-0")
r.xgroup_create("barras", "estrategia", id="0")
try:
    consumir(r, "c1", idempotente=True, caer_en=41)
except Caida:
    pass
cursor, reclamados, _ = r.xautoclaim("barras", "estrategia", "c2",
                                     min_idle_time=0, start_id="0-0")
print("c2 claims:", [i for i, _ in reclamados])
p = r.xpending_range("barras", "estrategia", "-", "+", 10)[0]
print(p["consumer"], "| deliveries:", p["times_delivered"])
consumir(r, "c2", idempotente=True)
print("pending:", r.xpending("barras", "estrategia")["pending"],
      "| effects:", r.llen("efectos"))

OUT [12]Another consumer claims the pending entries

c2 claims: ['41-0']
c2 | deliveries: 2
pending: 0 | effects: 100

XAUTOCLAIM hands over to c2 the pending entries that have gone unacknowledged for more than min_idle_time milliseconds (here 0, so that the example does not depend on the clock; in live trading, something longer than it takes to process a message, so as not to steal work from a consumer that is slow but alive). This is the second delivery. Since the consumer is idempotent and the ultimo_seq marker was stored with the effect, c2 acknowledges bar 41 without repeating it and carries on with the rest: 100 effects, one per bar.

A colleague proposes publishing the system's orders over Redis Pub/Sub “because it is faster than Streams”. What would you answer?

Solution

That speed is not the criterion for orders; the delivery guarantee is. Pub/Sub delivers at most once: if the order manager restarts, disconnects or lags, the orders published in that interval disappear with no error (in the lesson, 20 out of 100 bars). Orders need a persistent log with acknowledgment (Streams with a consumer group, Kafka or a table in a database), an idempotent consumer and an order key derived from the intent (lesson 4.5.5). Pub/Sub can still be useful for what can be lost, such as broadcasting the latest price to a dashboard.

Alistair Cockburn, “Hexagonal architecture” (https://alistair.cockburn.us/hexagonal-architecture/). Accessed on October 23, 2026. ↩
import-linter (https://import-linter.readthedocs.io/): the forbidden, protected, layers, independence and acyclic siblings contract types. Accessed on October 23, 2026. ↩
Martin Fowler, “Event Sourcing” (https://martinfowler.com/eaaDev/EventSourcing.html). Accessed on October 23, 2026. ↩
Redis, “Redis Pub/Sub” (https://redis.io/docs/latest/develop/pubsub/), section “Delivery semantics”: “Redis' Pub/Sub exhibits at-most-once message delivery semantics”. Accessed on October 23, 2026. ↩
Redis, “Redis Streams” (https://redis.io/docs/latest/develop/data-types/streams/): consumer groups, the pending entries list (PEL), XPENDING, XCLAIM, XAUTOCLAIM and the delivery counter. Accessed on October 23, 2026. ↩
Redis, “Redis persistence” (https://redis.io/docs/latest/operate/oss_and_stack/management/persistence/): RDB and AOF. Accessed on October 23, 2026. ↩
ZeroMQ, “zmq_socket” (https://libzmq.readthedocs.io/en/latest/zmq_socket.html): a ZMQ_PUB socket that reaches a subscriber's limit drops the messages for it for as long as that state lasts; and “zmq_setsockopt” (https://libzmq.readthedocs.io/en/latest/zmq_setsockopt.html) for ZMQ_SNDHWM, ZMQ_RCVHWM and ZMQ_CONFLATE. Accessed on October 23, 2026. ↩
Apache Kafka, “Introduction” (https://kafka.apache.org/intro/): messages with the same key go to the same partition, which is read in the same order in which it was written; and “Design” (https://kafka.apache.org/43/design/design/) for the idempotent producer. Accessed on October 23, 2026. ↩
Apache Kafka 4.3, “Producer Configs” (https://kafka.apache.org/43/configuration/producer-configs/): enable.idempotence, default true if no other option prevents it; librdkafka, “CONFIGURATION.md” (https://github.com/confluentinc/librdkafka/blob/master/CONFIGURATION.md): enable.idempotence, default false. Accessed on October 23, 2026. ↩

Each answer is folded under its question

Self-assessment

In the lesson's recursive event bus, the risk manager, subscribed to bars after the strategy, sees position 1 on the bar where the entry happens. Why?
- ABecause the risk manager computes the position incorrectly
- BBecause the recursive bus delivers the target and the fill triggered by the bar before it finishes delivering the bar itself to the remaining subscribers: the risk manager sees a state later than the bar it is evaluating
- CBecause the strategy runs twice
- DBecause the risk manager is subscribed to fills
Show answer
Answer B. Because the recursive bus delivers the target and the fill triggered by the bar before it finishes delivering the bar itself to the remaining subscribers: the risk manager sees a state later than the bar it is evaluating
With recursion, publishing inside a handler delivers the new event on the spot (depth first). With a queue, events are delivered in the order in which they were published: every subscriber sees the bar before the order it triggers exists.
A process subscribed to a Redis Pub/Sub channel restarts during 20 bars. Which of those 20 does it receive when it comes back?
- AAll 20, in order
- BOnly the last one
- CNone: Pub/Sub delivers at most once to whoever is subscribed at that moment and stores nothing
- DThose that fit in Redis's buffer
Show answer
Answer C. None: Pub/Sub delivers at most once to whoever is subscribed at that moment and stores nothing
The Redis documentation puts it this way: at-most-once semantics. Not losing messages while the consumer is down requires a persistent log: Redis Streams, Kafka or a database.
A consumer in a Redis Streams group processes bar 41 and crashes before XACK. What happens when it starts up?
- ABar 41 has been lost
- BRedis resends it to every consumer
- CThe group skips to 42
- DBar 41 is still in the consumer's pending list; when it reads its pending entries (or another consumer claims it) it is delivered again, with the delivery counter at 2: if the consumer is not idempotent, it processes it twice
Show answer
Answer D. Bar 41 is still in the consumer's pending list; when it reads its pending entries (or another consumer claims it) it is delivered again, with the delivery counter at 2: if the consumer is not idempotent, it processes it twice
Consumer groups give at-least-once delivery. The duplicate is avoided by the consumer: by storing the effect and the last processed sequence number in the same transaction, or by rejecting effects with a key already seen.
A ZeroMQ SUB stops reading while the PUB publishes 5,000 prices, with queue limits (HWM) of 10 on each side. What does it have in its queue when it reads again?
- AThe first 20: on reaching the limit, the PUB drops the new ones without warning, so the slow consumer is left with the oldest prices
- BThe last 20
- CAll 5,000
- DNone
Show answer
Answer A. The first 20: on reaching the limit, the PUB drops the new ones without warning, so the slow consumer is left with the oldest prices
A PUB in the mute state drops whatever no longer fits; what was in the queue stays there. For prices, what is useful is usually the opposite, keeping the latest: the ZMQ_CONFLATE option keeps only the most recent message.

Key points

The strategy core receives events and returns intents; it does not import pandas, Redis or the broker. The ports (Protocol) state what it needs and the adapters plug it in: Parquet in the backtest, a queue in live trading. A check with ast (or import-linter) guards the rule.
An event bus must deliver through a queue, not by recursion: otherwise a subscriber sees a state later than the event it is processing. Failures of a non-critical subscriber are isolated and logged; those of a critical one stop the system.
The event journal allows rebuilding the state and replaying the session exactly; in the lesson, the replayed strategy gives the same targets.
Redis Pub/Sub delivers at most once; Redis Streams with groups, at least once; ZeroMQ has no broker or persistence and drops what does not fit; Kafka keeps a partitioned log with per-key ordering and replay from any position.
At least once plus an idempotent consumer gives the effect of exactly once: in the lesson, after a crash halfway through, the consumer rebuilds its state from the stream and the targets match those of the single process one by one.

Resources

One core, three worlds

The dependency rule, checked

A queued event bus

Queues between processes: Redis Pub/Sub and Streams

ZeroMQ: messaging without a broker

Kafka: the distributed log

The core on Redis Streams

Exercises

Footnotes

Self-assessment.

Key points.

Resources.

Research

Documentation

Data

Self-assessment

Key points

Resources