--- title: "7.1.3 Patterns: events, queues and hexagonal architecture" description: "A strategy core with ports and adapters, a queued event bus, Redis Streams, ZeroMQ and Kafka, and recovery after a crash without duplicates." url: "https://pythonparatrading.com/en/python-trading-course/level-7/event-driven-queues-hexagonal-architecture/" lang: "en" image: "https://pythonparatrading.com/og/en/python-trading-course/level-7/event-driven-queues-hexagonal-architecture.png" --- # 7.1.3 · 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. - Course: [Level 7: Production, MLOps, regulation, career and capstone](https://pythonparatrading.com/en/python-trading-course/level-7/) - Module: 7.1 · Software engineering for trading - Duration: 75 min - Prerequisites: [2.4.2 · Event-driven backtest from scratch: loop, portfolio, orders and fills](https://pythonparatrading.com/en/python-trading-course/level-2/event-driven-backtest-from-scratch/), [4.5.1 · Live system architecture: from data to reconciliation](https://pythonparatrading.com/en/python-trading-course/level-4/live-trading-system-architecture/), [4.5.5 · Fault tolerance: idempotency, persistent state and reconciliation at startup](https://pythonparatrading.com/en/python-trading-course/level-4/fault-tolerance-idempotency-and-reconciliation/), [7.1.1 · Project structure, configuration and quality: from notebook to repository](https://pythonparatrading.com/en/python-trading-course/level-7/python-project-structure-configuration-quality/), [7.1.2 · Performance: measure, profile and compile only what is needed](https://pythonparatrading.com/en/python-trading-course/level-7/python-performance-profiling-numba-cython-rust/) ## By the end of this lesson - Separate a pure domain core from its ports and adapters, and check automatically that the core does not import infrastructure. - Design a queued (non-recursive) event bus, with failure isolation and a journal that allows rebuilding the state. - Explain the delivery guarantees of Redis Pub/Sub (at most once) and of Redis Streams with consumer groups (at least once), and make the consumer idempotent. - Use ZeroMQ to distribute work and broadcast prices, and anticipate what it does with messages when the consumer falls behind. - Place Kafka and choose the transport according to persistence, ordering, replay and operating cost. ## 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 2005[^1]: 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 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: ```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: ```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)}) ``` ```output 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: ```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()) ``` ```output 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.1[^2]. ## 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: ```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)) ``` ```output 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. ```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)") ``` ```output 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 sourcing*[^3]: 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” semantics[^4]. The experiment: 100 bars published, and the subscriber goes down and comes back between bar 41 and bar 60: ```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))) ``` ```output 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 done[^5]: ```mermaid 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`): ```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"]) ``` ```output 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: ```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) ``` ```output 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.* Horizontal bars on a log scale of messages received. HWM 1: 2, 0 to 1. HWM 10: 20, 0 to 19. HWM 100: 200, 0 to 199. HWM 1000: 2,000, 0 to 1,999. CONFLATE, highlighted: 1, number 4,999, the last one published. 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 state[^6]. 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 written[^7]: 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`)[^8]. It requires a cluster, so the code is for reading only: ```python skip 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: ```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) ``` ```output 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 1. 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.** ```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)) ``` ```output 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. 2. 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.** ```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")) ``` ```output 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. 3. 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. [^1]: Alistair Cockburn, “Hexagonal architecture” (https://alistair.cockburn.us/hexagonal-architecture/). Accessed on October 23, 2026. [^2]: import-linter (https://import-linter.readthedocs.io/): the `forbidden`, `protected`, `layers`, `independence` and `acyclic siblings` contract types. Accessed on October 23, 2026. [^3]: Martin Fowler, “Event Sourcing” (https://martinfowler.com/eaaDev/EventSourcing.html). Accessed on October 23, 2026. [^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. [^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. [^6]: 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. [^7]: 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. [^8]: 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. [^9]: Redis, “Redis persistence” (https://redis.io/docs/latest/operate/oss_and_stack/management/persistence/): RDB and AOF. Accessed on October 23, 2026. ## Self-assessment 1. 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? - A. Because the risk manager computes the position incorrectly - 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 - C. Because the strategy runs twice - D. Because the risk manager is subscribed to fills *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. 2. A process subscribed to a Redis Pub/Sub channel restarts during 20 bars. Which of those 20 does it receive when it comes back? - A. All 20, in order - B. Only the last one - C. None: Pub/Sub delivers at most once to whoever is subscribed at that moment and stores nothing - D. Those that fit in Redis's buffer *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. 3. A consumer in a Redis Streams group processes bar 41 and crashes before `XACK`. What happens when it starts up? - A. Bar 41 has been lost - B. Redis resends it to every consumer - C. The group skips to 42 - 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 *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. 4. 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? - 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 - B. The last 20 - C. All 5,000 - D. None *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 ### Research - [Finite-State Machines for Quantitative Trading Execution in Python](https://pythonparatrading.com/en/finite-state-machines-trading-python/) · Research-0029 · 47 min ### Documentation - [Alistair Cockburn: Hexagonal architecture](https://alistair.cockburn.us/hexagonal-architecture/) · alistair.cockburn.us - [Redis: Streams (consumer groups, XPENDING, XCLAIM, XAUTOCLAIM)](https://redis.io/docs/latest/develop/data-types/streams/) · redis.io - [Redis: Pub/Sub (delivery semantics)](https://redis.io/docs/latest/develop/pubsub/) · redis.io - [ZeroMQ: zmq_setsockopt (ZMQ_SNDHWM, ZMQ_RCVHWM, ZMQ_CONFLATE)](https://libzmq.readthedocs.io/en/latest/zmq_setsockopt.html) · libzmq.readthedocs.io - [Apache Kafka: introduction (topics, partitions, per-key ordering)](https://kafka.apache.org/intro/) · kafka.apache.org ### Data - [Level 5 data: one-minute bars of a future (Parquet)](https://pythonparatrading.com/curso/datos/nivel-5/hf5_barras_1min.parquet) · Download · hf5_barras_1min.parquet Previous lesson: [7.1.2 · Performance: measure, profile and compile only what is needed](https://pythonparatrading.com/en/python-trading-course/level-7/python-performance-profiling-numba-cython-rust/) Next lesson: [7.2.1 · Docker and docker-compose: reproducible images with uv](https://pythonparatrading.com/en/python-trading-course/level-7/docker-compose-reproducible-images-uv/)