Lección 7.1.3 · 75 min

Patrones: eventos, colas de mensajes y arquitectura hexagonal

La estrategia de ruptura de la lección anterior, convertida en un núcleo que no sabe de dónde le llegan las barras ni adónde van sus órdenes. Esta lección lo rodea de puertos y adaptadores, comprueba la regla de dependencias, lo conecta a un bus de eventos y después a colas entre procesos (Redis Pub/Sub y Streams, ZeroMQ, Kafka) y mide qué se pierde o se duplica en cada una cuando el consumidor se cae o se queda atrás.

Fig. — Qué pasa con los mensajes cuando el consumidor se cae o no lee: ZeroMQ descarta lo que no cabe en la cola, Pub/Sub pierde lo publicado mientras no hay nadie, Streams no pierde nada pero duplica si el consumidor no es idempotente

03/03

Módulo: 7.1 · Ingeniería de software para trading
Duración: 75 min
Requisitos: 2.4.2 · Backtest por eventos desde cero: bucle, cartera, órdenes y fills, 4.5.1 · Arquitectura de un sistema en vivo: de los datos a la reconciliación, 4.5.5 · Tolerancia a fallos: idempotencia, estado persistente y reconciliación al arrancar, 7.1.1 · Estructura de proyecto, configuración y calidad: del notebook al repositorio, 7.1.2 · Rendimiento: medir, perfilar y compilar solo lo que hace falta

Lecciones del nivel 7

Un núcleo, tres mundos

La lección 4.5.1 montó un sistema en vivo con un bus de eventos y un reloj lógico, y comprobó que los mismos componentes sirven para backtest, paper y real. Esta lección se detiene en los patrones que lo hacen posible y en lo que pasa cuando las piezas dejan de vivir en el mismo proceso: el lector de datos en una máquina, la estrategia en otra, el gestor de órdenes en una tercera, unidos por colas de mensajes.

El patrón de fondo es la arquitectura hexagonal o de puertos y adaptadores, que Alistair Cockburn describió en 20051Nota 1 · Alistair Cockburn, «Hexagonal architecture» (https://alistair.cockburn.us/hexagonal-architecture/). Consultada el 23 de octubre de 2026.: la aplicación tiene un núcleo que no depende de nada externo y que habla con el exterior a través de puertos (interfaces definidas por el núcleo, según lo que necesita); cada tecnología concreta se conecta con un adaptador. NautilusTrader (lección 2.4.5) está construido así: la misma estrategia corre contra un motor de backtest o contra un adaptador de Interactive Brokers.

Fuente Mermaid

flowchart TB
  P[Parquet] --> M[puerto Mercado]
  R[Redis Streams] --> M
  M --> N[núcleo Ruptura]
  N --> E[puerto Ejecutor]
  E --> S[broker simulado]
  E --> B[broker real]

El núcleo de la lección es la estrategia de ruptura de la lección 7.1.2 reescrita para recibir una barra cada vez (un evento) y devolver intenciones: «quiero tener 1 contrato», «quiero tener 0». No decide cómo se ejecuta eso, ni sabe si la barra viene de un fichero o de una cola. Los eventos son dataclasses inmutables con un número de secuencia, que es lo que permite ordenarlos, deduplicarlos y reproducirlos:

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                    # la barra que lo provoca
    instrumento: str
    posicion: int
    precio_ref: float
    motivo: str

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

class Ruptura:
    """Ruptura de n barras con stop de k ATR (lección 7.1.2), una
    barra cada vez. Sin E/S: recibe eventos, devuelve objetivos."""

    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, "sesión"
        elif self.pos == 0:
            if b.cierre > max(self.maximos):
                self.pos, self.pico, motivo = 1, b.cierre, "ruptura"
        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)]

El deque de 30 máximos sustituye al iloc[i - n:i] de la versión de pandas: el núcleo guarda lo que necesita de su pasado y nada más. Los puertos son lo que el resto del programa le ofrece: una fuente de barras y un ejecutor de objetivos. Se escriben como Protocol (lección 0.4.2): cualquier clase con esos métodos sirve, sin heredar de nada. Con un adaptador que lee el Parquet y un ejecutor simulado, el núcleo hace un 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:
    """Adaptador: barras de un Parquet, con su número de secuencia."""

    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:
    """Adaptador: lleva la posición al objetivo al precio de ref."""

    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} entradas, en mercado el {fraccion:.1%} "
      "de las barras")
print({t.__name__: sum(isinstance(e, t) for e in diario)
       for t in (Barra, Objetivo, Ejecucion)})

OUT [02]Puertos, adaptadores y un backtest

226 entradas, en mercado el 40.7% de las barras
{'Barra': 20000, 'Objetivo': 451, 'Ejecucion': 451}

Las mismas 226 entradas y el mismo 40,7 % en mercado que el backtest de la lección 7.1.2 sobre las mismas 20.000 barras. Es la comprobación que importa al reorganizar código: el comportamiento no cambia. El diario reúne las 20.000 barras y los objetivos y ejecuciones que provocaron, en el orden en que ocurrieron.

La regla de dependencias, comprobada

Que el núcleo no dependa de la infraestructura es una regla fácil de romper: basta con que alguien importe el cliente de Redis «solo para una cosa» dentro de la estrategia. A partir de ahí, la estrategia ya no se puede probar sin Redis. La regla se comprueba como cualquier otra, con una herramienta. La celda escribe un paquete de juguete con la forma de un proyecto real y lee sus importaciones con el módulo ast, sin ejecutar nada:

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("núcleo limpio:", violaciones())
with open("bot/nucleo/ruptura.py", "a") as f:
    f.write("import redis\n"
            "from bot.adaptadores.redis_mercado import leer\n")
print("con un atajo:", violaciones())

OUT [03]Las importaciones del núcleo, leídas con ast

núcleo limpio: []
con un atajo: [('ruptura.py', 'redis'), ('ruptura.py', 'bot.adaptadores.redis_mercado')]

La lista de importaciones permitidas es una decisión de diseño escrita en código: el núcleo puede usar la biblioteca estándar que no hace E/S y su propio paquete. En un proyecto real, la herramienta habitual es import-linter, que lee contratos (forbidden, layers, independence…) desde el pyproject.toml y se ejecuta en pre-commit y en la integración continua como las comprobaciones de la lección 7.1.12Nota 2 · import-linter (https://import-linter.readthedocs.io/): tipos de contrato forbidden, protected, layers, independence y acyclic siblings. Consultada el 23 de octubre de 2026..

Un bus de eventos con cola

Dentro de un proceso, las piezas se comunican con un bus: quien produce un evento lo publica y quien lo necesita se suscribe por tipo. La lección 4.5.1 usó uno. Hay un detalle de diseño que decide si el sistema es correcto: qué pasa cuando un suscriptor publica a su vez. Un bus recursivo entrega el evento nuevo en el acto, dentro de la entrega del anterior; un bus con cola lo pone al final y lo entrega cuando termina el actual. La celda monta las dos versiones, con tres suscriptores: la estrategia (barras), el ejecutor (objetivos) y un control de riesgo (barras), suscrito después de la estrategia, que anota la posición que ve en cada barra:

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                       # se entregará en su turno
        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"estrategia ve la barra {b.seq}")
        for o in est.al_recibir(b):
            bus.publicar(o)

    def ejecutor(o):
        traza.append(f"objetivo {o.posicion} de la barra {o.seq}")
        for e in ejec.enviar(o):
            bus.publicar(e)

    def riesgo(b):
        traza.append(f"riesgo ve la barra {b.seq} con posición "
                     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("con cola:" if con_cola else "recursivo:", " | ".join(traza))

OUT [04]Recursión frente a cola

recursivo: estrategia ve la barra 170 | objetivo 1 de la barra 170 | riesgo ve la barra 170 con posición 1
con cola: estrategia ve la barra 170 | riesgo ve la barra 170 con posición 0 | objetivo 1 de la barra 170

Con el bus recursivo, el riesgo ve la barra de la entrada con la posición ya en 1: la estrategia publicó el objetivo, el ejecutor lo ejecutó y todo eso ocurrió antes de que la barra llegara al siguiente suscriptor. Para el riesgo, la ejecución que provoca la barra ha pasado antes que la barra. En este ejemplo es un rótulo equivocado; con un límite de exposición que se evalúa en cada barra, es una orden rechazada o aceptada por error, y el resultado depende del orden en que se suscribieron los manejadores. Con la cola, todos ven la barra antes de que exista lo que provoca, y el orden del diario es el orden causal.

El bus tiene otras dos responsabilidades. La primera, decidir qué hace con un suscriptor que falla. Un cálculo de métricas con un error (busca la referencia de la sesión antes de haberla guardado) falla en la primera barra de cada sesión; no debe parar el sistema, pero tampoco desaparecer en silencio. Un suscriptor crítico (estrategia, ejecutor, riesgo) que falla, en cambio, sí debe pararlo. La segunda, guardar el diario: con él se reconstruye el estado y se reproduce la sesión.

IN [05]python

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

def metrica_sesion(b):
    """Rentabilidad desde el primer cierre de la sesión. Error: en
    la primera barra de cada sesión aún no hay referencia."""
    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)         # no crítico
for b in barras:
    bus.publicar(b)
print("fallos aislados:", 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("posición desde el diario:", pos,
      "| igual a la del ejecutor:", 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("reproducción idéntica:", objetivos == originales,
      f"({len(objetivos)} objetivos)")

OUT [05]Fallos aislados y reproducción desde el diario

fallos aislados: 52 | ('Barra', 1, 'metrica_sesion', "KeyError('FUT')")
posición desde el diario: 1 | igual a la del ejecutor: True
reproducción idéntica: True (451 objetivos)

La primera barra de cada sesión falla en el suscriptor de métricas, cada fallo queda anotado con su número de secuencia y el error, y el sistema sigue; la posición final y los objetivos son los mismos que sin ese suscriptor. Reproducir las barras del diario con una estrategia nueva da exactamente los mismos objetivos: el núcleo es determinista, y el diario basta para reconstruir la sesión. Es la idea del event sourcing3Nota 3 · Martin Fowler, «Event Sourcing» (https://martinfowler.com/eaaDev/EventSourcing.html). Consultada el 23 de octubre de 2026.: el estado es una función de la secuencia de eventos, y la secuencia es lo que se guarda.

Colas entre procesos: Redis Pub/Sub y Streams

Cuando el lector de datos y la estrategia son procesos distintos (porque uno se reinicia sin el otro, porque hay varias estrategias leyendo los mismos datos, porque viven en máquinas distintas), el bus pasa a ser un servicio de mensajería. Redis, que muchos sistemas ya usan como caché, ofrece dos con garantías muy distintas. La lección usa fakeredis, que implementa los comandos de Redis en memoria con el mismo cliente redis-py; con un servidor real solo cambia la línea que crea la conexión.

Pub/Sub es difusión: el mensaje llega a quien esté suscrito en ese momento y Redis no lo guarda. La documentación lo llama semántica de «como mucho una vez»4Nota 4 · Redis, «Redis Pub/Sub» (https://redis.io/docs/latest/develop/pubsub/), sección «Delivery semantics»: «Redis' Pub/Sub exhibits at-most-once message delivery semantics». Consultada el 23 de octubre de 2026.. El experimento: 100 barras publicadas, y el suscriptor se cae y vuelve entre la 41 y la 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")              # el proceso cae
        recibidos += leer_pubsub(ps)
    if b.seq == 61:
        ps.subscribe("barras")                # y vuelve
    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), "recibidos; perdidos:", perdidos_pubsub[0], "a",
      perdidos_pubsub[-1], "| receptores por mensaje:",
      sorted(set(receptores)))

OUT [06]Pub/Sub con el suscriptor caído un rato

80 recibidos; perdidos: 41 a 60 | receptores por mensaje: [0, 1]

Veinte barras perdidas, sin error: publish devuelve el número de receptores, 0 mientras el suscriptor no está, y es la única pista. Pub/Sub sirve para lo que se puede perder (un panel que muestra el último precio, una notificación) y no para lo que hay que procesar entero.

Streams es un registro: cada mensaje se añade con un identificador creciente y se queda. Un grupo de consumidores reparte los mensajes entre los consumidores del grupo y recuerda dos cosas: hasta dónde ha entregado y qué mensajes entregados no se han confirmado todavía (la lista de pendientes, PEL). Un mensaje se confirma con XACK cuando su efecto está hecho5Nota 5 · Redis, «Redis Streams» (https://redis.io/docs/latest/develop/data-types/streams/): grupos de consumidores, lista de pendientes (PEL), XPENDING, XCLAIM, XAUTOCLAIM y el contador de entregas. Consultada el 23 de octubre de 2026.:

Fuente Mermaid

stateDiagram-v2
  [*] --> EnStream: XADD
  EnStream --> Pendiente: XREADGROUP
  Pendiente --> Confirmado: XACK
  Pendiente --> Pendiente: caída y relectura
  Pendiente --> Reclamado: XCLAIM
  Reclamado --> Confirmado: XACK

El experimento equivalente: el consumidor procesa las barras de una en una y se cae justo después de procesar la 41, antes de confirmarla. Al arrancar, primero lee sus propios pendientes (XREADGROUP con el identificador 0) y después los nuevos (>). Dos consumidores: uno que no recuerda lo que ha hecho y otro que guarda el efecto y el último número de secuencia procesado en la misma transacción de Redis (MULTI/EXEC):

IN [07]python

class Caida(BaseException):
    """El proceso muere: nada de lo que siga se ejecuta."""

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                               # ya estaba hecho
        with r.pipeline(transaction=True) as t:  # efecto + marca
            t.rpush("efectos", seq)
            t.set("ultimo_seq", seq)
            t.execute()
        procesados.append(seq)
        if seq == caer_en:
            raise Caida(f"cae tras procesar {seq}, sin XACK")

    for desde in ("0", ">"):                     # pendientes, nuevos
        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, "| pendientes:", [(p["message_id"],
              p["times_delivered"]) for p in pend])
    consumir(r, "c1", idempotente)               # rearranque
    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("idempotente:" if idempotente else "ingenuo:",
          len(set(efectos)), "barras con efecto; duplicadas:", dobles,
          "| pendientes al final:",
          r.xpending("barras", "estrategia")["pending"])

OUT [07]Streams: caída antes de XACK, con y sin idempotencia

cae tras procesar 41, sin XACK | pendientes: [('41-0', 1)]
ingenuo: 100 barras con efecto; duplicadas: [41] | pendientes al final: 0
cae tras procesar 41, sin XACK | pendientes: [('41-0', 1)]
idempotente: 100 barras con efecto; duplicadas: [] | pendientes al final: 0

Ninguna barra se pierde: al caer, la 41 queda pendiente del consumidor c1 con una entrega, y al arrancar se le vuelve a entregar. El consumidor ingenuo la procesa otra vez: su efecto está dos veces. El idempotente ve que la 41 ya está en ultimo_seq (escrito en la misma transacción que el efecto, así que o están los dos o ninguno) y solo la confirma. Es la regla de la lección 4.5.5 trasladada a la mensajería: la red y las colas dan «al menos una vez»; el consumidor idempotente convierte eso en el efecto de «exactamente una vez». Si c1 no volviera, otro consumidor del grupo podría quedarse sus pendientes con XCLAIM o XAUTOCLAIM, indicando cuánto tiempo deben llevar sin confirmar.

ZeroMQ: mensajería sin intermediario

ZeroMQ no es un servidor sino una biblioteca de sockets con patrones: los procesos se conectan directamente, sin intermediario que guarde nada. Eso lo hace rápido y sencillo de desplegar, y traslada a la aplicación lo que haga falta de persistencia. La celda usa el transporte inproc:// (entre hilos del mismo proceso; con tcp:// el código es el mismo) y tres patrones: PUSH/PULL para repartir trabajo, PUB/SUB para difundir precios con filtro por prefijo y el límite de cola (high water mark, HWM) cuando el suscriptor no lee:

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("suscripción recibida por el XPUB:", pub.recv())
for txt in ("ORSC 234.84", "ORUS 716.82", "ORSC 234.90"):
    pub.send_string(txt)
print("PUB/SUB con filtro:", drenar(sub))

def lento(hwm, conflate=False):
    """El SUB no lee mientras se publican 5.000 precios."""
    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()                                  # la suscripción
    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} recibidos, del {v[0]} al {v[-1]}")
print("CONFLATE:", lento(10, conflate=True))
for s in (push, *obreros, pub, sub):
    s.close(0)

OUT [08]ZeroMQ: reparto, difusión y límite de cola

PUSH/PULL: [['backtest 0', 'backtest 2', 'backtest 4'], ['backtest 1', 'backtest 3', 'backtest 5']]
suscripción recibida por el XPUB: b'\x01ORSC '
PUB/SUB con filtro: ['ORSC 234.84', 'ORSC 234.90']
HWM    1:    2 recibidos, del 0 al 1
HWM   10:   20 recibidos, del 0 al 19
HWM  100:  200 recibidos, del 0 al 199
HWM 1000: 2000 recibidos, del 0 al 1999
CONFLATE: ['4999']

Fig. — Un suscriptor de ZeroMQ que no lee mientras se publican 5.000 precios: con límite de cola recibe exactamente la suma de los dos límites, y son los precios más antiguos; con CONFLATE, solo el último

PUSH reparte por turnos (0, 2, 4 para un obrero; 1, 3, 5 para el otro): sirve para lanzar un barrido de backtests sobre varios procesos. El SUB solo recibe lo que empieza por ORSC ; el filtro se aplica en el lado del publicador, que recibe la suscripción como un mensaje (\x01 más el prefijo). Usar XPUB y esperar ese mensaje antes de publicar evita el problema del «suscriptor tardío» (slow joiner): con un PUB normal, lo publicado antes de que la suscripción llegue se pierde.

La parte que más importa en trading es la última. Con un suscriptor que no lee, el PUB llena la cola hasta el HWM de cada lado y después descarta lo nuevo, sin error, como describe la documentación de ZeroMQ para los sockets PUB en estado de silencio7Nota 7 · ZeroMQ, «zmq_socket» (https://libzmq.readthedocs.io/en/latest/zmq_socket.html): un socket ZMQ_PUB que llega al límite de un suscriptor descarta los mensajes para él mientras dure ese estado; y «zmq_setsockopt» (https://libzmq.readthedocs.io/en/latest/zmq_setsockopt.html) para ZMQ_SNDHWM, ZMQ_RCVHWM y ZMQ_CONFLATE. Consultadas el 23 de octubre de 2026.. En la figura, el número recibido es exactamente la suma de los dos límites, y lo recibido son los precios más antiguos: un consumidor lento que vuelve a leer opera con precios viejos y no se entera. Para precios, lo razonable suele ser lo contrario: ZMQ_CONFLATE guarda solo el último mensaje, que es lo que el SUB recibe (el 4999). Para órdenes o ejecuciones, ninguna de las dos cosas vale: no se puede perder ninguna, y hace falta un registro persistente.

Kafka: el registro distribuido

Kafka lleva la idea de Redis Streams a un sistema distribuido: un tema es un registro particionado y replicado en varios servidores, los mensajes se guardan durante el tiempo de retención configurado (días, meses o indefinidamente) y cada grupo de consumidores guarda su posición (offset) en cada partición. Los mensajes con la misma clave van a la misma partición, y Kafka garantiza que un consumidor lee cada partición en el orden en que se escribió8Nota 8 · Apache Kafka, «Introduction» (https://kafka.apache.org/intro/): mismos mensajes con la misma clave en la misma partición, leída en el mismo orden en que se escribió; y «Design» (https://kafka.apache.org/43/design/design/) para el productor idempotente. Consultadas el 23 de octubre de 2026.: con el instrumento como clave, todas las barras de ORSC llegan en orden aunque el tema tenga muchas particiones. El productor idempotente evita que un reintento escriba un mensaje dos veces: en el cliente de Java viene activado por defecto, mientras que en librdkafka, la biblioteca de C que usa confluent-kafka en Python, hay que pedirlo (enable.idempotence, por defecto false)9Nota 9 · Apache Kafka 4.3, «Producer Configs» (https://kafka.apache.org/43/configuration/producer-configs/): enable.idempotence, por defecto true si ninguna otra opción lo impide; librdkafka, «CONFIGURATION.md» (https://github.com/confluentinc/librdkafka/blob/master/CONFIGURATION.md): enable.idempotence, por defecto false. Consultadas el 23 de octubre de 2026.. Exige un clúster, así que el código es solo de lectura:

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()))      # idempotente, como antes
    cons.commit(message=m)               # después del efecto

La disciplina es la misma que con Streams: confirmar (commit) después del efecto y no antes, y consumidor idempotente. La ventaja propia de Kafka es la reproducción: un consumidor nuevo puede leer el tema desde el principio, y un backtest puede alimentarse exactamente de los mensajes que vio el sistema en vivo. El precio es operar un clúster (o pagar uno gestionado). Para un trader individual con un bot en un servidor, es casi siempre desproporcionado.

	Persistencia	Entrega	Orden	Reproducir	Operar
Bus en memoria	ninguna (diario propio)	síncrona	total	desde el diario	nada
Redis Pub/Sub	ninguna	como mucho una vez	por canal	no	un Redis
Redis Streams	en Redis	al menos una vez (grupos)	total por stream	sí, por id	un Redis
ZeroMQ	ninguna	descarta al llenarse (PUB)	por conexión	no	nada (biblioteca)
Kafka	registro replicado	al menos una vez; exactamente una con transacciones	por partición	sí, por offset	un clúster

El núcleo sobre Redis Streams

Para cerrar, el núcleo del principio, sin cambiar una línea, conectado a dos adaptadores nuevos: uno que lee las barras de un stream con un grupo de consumidores y otro que escribe los objetivos en otro stream con el número de secuencia como identificador. Redis rechaza un XADD con un identificador que no sea mayor que el último del stream, así que escribir dos veces el mismo objetivo es imposible: es la clave de idempotencia de la lección 4.5.5, aplicada por la cola. El proceso cae a mitad, en la barra 2.500, y al arrancar reconstruye el estado de la estrategia releyendo el stream de barras desde el principio:

IN [10]python

class MercadoStream:
    """Adaptador: barras del grupo; confirma cada una al volver."""

    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:
    """Adaptador: un objetivo por barra en el stream `objetivos`."""

    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           # ya enviado antes de caer
        return []

def caer(en):
    def anotar(ev):
        if isinstance(ev, Barra) and ev.seq == en:
            raise Caida(f"cae al recibir la barra {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, "| objetivos escritos:", r.xlen("objetivos"),
          "| pendientes:", pendientes)

# rearranque: estado reconstruido desde el stream de barras
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("rechazados por id repetido:", ejec2.rechazados)
print("objetivos en Redis:", len(en_redis), "| idénticos al proceso "
      "único:", en_redis == en_proceso)

OUT [10]El núcleo con dos adaptadores de Redis y una caída

cae al recibir la barra 2500 | objetivos escritos: 57 | pendientes: 1
rechazados por id repetido: 57
objetivos en Redis: 122 | idénticos al proceso único: True

Al caer, los objetivos de las barras anteriores a la 2.500 ya están en el stream y la barra 2.500 está entregada pero sin confirmar. Al arrancar, la estrategia nueva no sabe nada (su deque, su ATR y su máximo desde la entrada se perdieron con el proceso), así que relee el stream de barras hasta la última entregada; los objetivos que vuelve a producir chocan con los que ya estaban y Redis los rechaza. Después confirma la pendiente y sigue con las nuevas. El resultado coincide uno a uno con el del backtest en un solo proceso. Dos simplificaciones respecto a un sistema real: releer todo el stream no escala a meses de barras, y se sustituye por una instantánea periódica del estado de la estrategia más los eventos posteriores; y aquí el estado vive en la memoria del proceso, mientras que en real la posición la sabe el broker, y el arranque concilia con él (lección 4.5.5).

La figura de la cabecera resume los cuatro experimentos de entrega: con el consumidor caído o lento, Pub/Sub y ZeroMQ pierden mensajes sin error; Streams no pierde ninguno pero duplica si el consumidor no es idempotente; con idempotencia, cada barra tiene su efecto una sola vez.

Ejercicios

Añade al bus de la lección un suscriptor crítico que falle (por ejemplo, un riesgo que lanza una excepción cuando la posición es 1 y el cierre baja de un umbral) y comprueba que el bus con cola se detiene y que la cola queda con los eventos sin entregar. ¿Qué debería hacer el sistema a continuación?

Solución

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"barra {b.seq}: cierre {b.cierre}")

bus.suscribir(Barra, riesgo_duro, critico=True)
try:
    for b in barras:
        bus.publicar(b)
except LimiteDeRiesgo as e:
    print("parado:", e)
print("sin entregar en la cola:",
      [(type(ev).__name__, ev.seq, ev.posicion) for ev in bus.cola])
print("fallos aislados:", len(bus.fallidos))

OUT [11]Un fallo crítico detiene el bus

parado: barra 2427: cierre 5195.5
sin entregar en la cola: [('Objetivo', 2427, 0)]
fallos aislados: 0

El fallo de un suscriptor crítico sale del bus y detiene la publicación; lo que estuviera en la cola en ese momento queda sin entregar y a la vista. Aquí es el objetivo de salir (posición 0) que la propia barra 2.427 había provocado: la estrategia quería cerrar y el riesgo ha parado el sistema antes de que la orden saliera. Parar no deshace la posición, así que el sistema queda con un contrato y una salida pendiente. Debería pasar a un estado de bloqueo (lección 4.5.5) con un procedimiento explícito para la posición abierta (cerrarla, cubrirla o dejarla, decidido de antemano), avisar a una persona (lección 4.5.4) y no rearrancar solo. Lo que no debe hacer es lo que haría un except Exception genérico alrededor de todo: anotar el error y seguir operando como si el riesgo hubiera dicho que sí.

Con fakeredis, reproduce el caso de un consumidor que cae para siempre: c1 se queda con la barra 41 pendiente y un segundo consumidor, c2, la reclama con XAUTOCLAIM y la procesa. ¿Cuántas veces se ha entregado?

Solución

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 reclama:", [i for i, _ in reclamados])
p = r.xpending_range("barras", "estrategia", "-", "+", 10)[0]
print(p["consumer"], "| entregas:", p["times_delivered"])
consumir(r, "c2", idempotente=True)
print("pendientes:", r.xpending("barras", "estrategia")["pending"],
      "| efectos:", r.llen("efectos"))

OUT [12]Otro consumidor reclama los pendientes

c2 reclama: ['41-0']
c2 | entregas: 2
pendientes: 0 | efectos: 100

XAUTOCLAIM pasa a c2 los pendientes que llevan más de min_idle_time milisegundos sin confirmar (aquí 0, para que el ejemplo no dependa del reloj; en real, algo mayor que lo que tarda en procesarse un mensaje, para no robarle trabajo a un consumidor lento pero vivo). La entrega es la segunda. Como el consumidor es idempotente y la marca ultimo_seq se guardó con el efecto, c2 confirma la 41 sin repetirla y sigue con las demás: 100 efectos, uno por barra.

Un compañero propone publicar las órdenes del sistema por Redis Pub/Sub «porque es más rápido que Streams». ¿Qué le contestarías?

Solución

Que la velocidad no es el criterio para las órdenes, sino la garantía de entrega. Pub/Sub entrega como mucho una vez: si el gestor de órdenes se reinicia, se desconecta o tarda, las órdenes publicadas en ese intervalo desaparecen sin error (en la lección, 20 de 100 barras). Para órdenes hace falta un registro persistente con confirmación (Streams con grupo de consumidores, Kafka o una tabla en una base de datos), un consumidor idempotente y una clave de orden que salga de la intención (lección 4.5.5). Pub/Sub puede seguir sirviendo para lo que se puede perder, como difundir el último precio a un panel.

Alistair Cockburn, «Hexagonal architecture» (https://alistair.cockburn.us/hexagonal-architecture/). Consultada el 23 de octubre de 2026. ↩
import-linter (https://import-linter.readthedocs.io/): tipos de contrato forbidden, protected, layers, independence y acyclic siblings. Consultada el 23 de octubre de 2026. ↩
Martin Fowler, «Event Sourcing» (https://martinfowler.com/eaaDev/EventSourcing.html). Consultada el 23 de octubre de 2026. ↩
Redis, «Redis Pub/Sub» (https://redis.io/docs/latest/develop/pubsub/), sección «Delivery semantics»: «Redis' Pub/Sub exhibits at-most-once message delivery semantics». Consultada el 23 de octubre de 2026. ↩
Redis, «Redis Streams» (https://redis.io/docs/latest/develop/data-types/streams/): grupos de consumidores, lista de pendientes (PEL), XPENDING, XCLAIM, XAUTOCLAIM y el contador de entregas. Consultada el 23 de octubre de 2026. ↩
Redis, «Redis persistence» (https://redis.io/docs/latest/operate/oss_and_stack/management/persistence/): RDB y AOF. Consultada el 23 de octubre de 2026. ↩
ZeroMQ, «zmq_socket» (https://libzmq.readthedocs.io/en/latest/zmq_socket.html): un socket ZMQ_PUB que llega al límite de un suscriptor descarta los mensajes para él mientras dure ese estado; y «zmq_setsockopt» (https://libzmq.readthedocs.io/en/latest/zmq_setsockopt.html) para ZMQ_SNDHWM, ZMQ_RCVHWM y ZMQ_CONFLATE. Consultadas el 23 de octubre de 2026. ↩
Apache Kafka, «Introduction» (https://kafka.apache.org/intro/): mismos mensajes con la misma clave en la misma partición, leída en el mismo orden en que se escribió; y «Design» (https://kafka.apache.org/43/design/design/) para el productor idempotente. Consultadas el 23 de octubre de 2026. ↩
Apache Kafka 4.3, «Producer Configs» (https://kafka.apache.org/43/configuration/producer-configs/): enable.idempotence, por defecto true si ninguna otra opción lo impide; librdkafka, «CONFIGURATION.md» (https://github.com/confluentinc/librdkafka/blob/master/CONFIGURATION.md): enable.idempotence, por defecto false. Consultadas el 23 de octubre de 2026. ↩

Respuesta plegada en cada pregunta

Autoevaluación

En el bus de eventos recursivo de la lección, el gestor de riesgo, suscrito a las barras después de la estrategia, ve la posición 1 en la barra en la que se entra. ¿Por qué?
- APorque el riesgo calcula mal la posición
- BPorque el bus recursivo entrega el objetivo y la ejecución que provoca la barra antes de terminar de entregar la propia barra al resto de suscriptores: el riesgo ve un estado posterior a la barra que está evaluando
- CPorque la estrategia se ejecuta dos veces
- DPorque el riesgo está suscrito a las ejecuciones
Ver respuesta
Respuesta B. Porque el bus recursivo entrega el objetivo y la ejecución que provoca la barra antes de terminar de entregar la propia barra al resto de suscriptores: el riesgo ve un estado posterior a la barra que está evaluando
Con recursión, publicar dentro de un manejador entrega el evento nuevo en el acto (en profundidad). Con una cola, los eventos se entregan en el orden en que se publicaron: todos los suscriptores ven la barra antes de que exista la orden que provoca.
Un proceso suscrito a un canal de Redis Pub/Sub se reinicia durante 20 barras. ¿Qué recibe de esas 20 al volver?
- ALas 20, en orden
- BSolo la última
- CNinguna: Pub/Sub entrega como mucho una vez a quien esté suscrito en ese momento y no guarda nada
- DLas que quepan en el búfer de Redis
Ver respuesta
Respuesta C. Ninguna: Pub/Sub entrega como mucho una vez a quien esté suscrito en ese momento y no guarda nada
La documentación de Redis lo dice así: semántica de como mucho una vez. Para no perder mensajes con el consumidor caído hace falta un registro persistente: Redis Streams, Kafka o una base de datos.
Un consumidor de un grupo de Redis Streams procesa la barra 41 y se cae antes de XACK. ¿Qué pasa al arrancar?
- ALa barra 41 se ha perdido
- BRedis la reenvía a todos los consumidores
- CEl grupo salta a la 42
- DLa 41 sigue en la lista de pendientes del consumidor; al leer sus pendientes (o al reclamarla otro) se le vuelve a entregar, con el contador de entregas a 2: si el consumidor no es idempotente, la procesa dos veces
Ver respuesta
Respuesta D. La 41 sigue en la lista de pendientes del consumidor; al leer sus pendientes (o al reclamarla otro) se le vuelve a entregar, con el contador de entregas a 2: si el consumidor no es idempotente, la procesa dos veces
Los grupos de consumidores dan entrega al menos una vez. El duplicado lo evita el consumidor: guardando en la misma transacción el efecto y el último número de secuencia procesado, o rechazando efectos con una clave ya vista.
Un SUB de ZeroMQ deja de leer mientras el PUB publica 5.000 precios, con límites de cola (HWM) de 10 en cada lado. ¿Qué tiene en la cola cuando vuelve a leer?
- ALos 20 primeros: al llegar al límite, el PUB descarta los nuevos sin avisar, así que el consumidor lento se queda con los precios más viejos
- BLos 20 últimos
- CLos 5.000
- DNinguno
Ver respuesta
Respuesta A. Los 20 primeros: al llegar al límite, el PUB descarta los nuevos sin avisar, así que el consumidor lento se queda con los precios más viejos
Un PUB en estado de silencio descarta lo que ya no cabe; lo que estaba en la cola sigue ahí. Para precios, lo útil suele ser lo contrario, quedarse con el último: la opción ZMQ_CONFLATE guarda solo el mensaje más reciente.

Ideas clave

El núcleo de la estrategia recibe eventos y devuelve intenciones; no importa pandas, Redis ni el broker. Los puertos (Protocol) dicen qué necesita y los adaptadores lo conectan: Parquet en el backtest, una cola en real. Una comprobación con ast (o import-linter) vigila la regla.
Un bus de eventos debe entregar con cola, no por recursión: si no, un suscriptor ve un estado posterior al evento que procesa. Los fallos de un suscriptor no crítico se aíslan y se anotan; los de uno crítico paran el sistema.
El diario de eventos permite reconstruir el estado y reproducir la sesión exactamente; en la lección, la estrategia reproducida da los mismos objetivos.
Redis Pub/Sub entrega como mucho una vez; Redis Streams con grupos, al menos una vez; ZeroMQ no tiene intermediario ni persistencia y descarta lo que no cabe; Kafka guarda un registro particionado con orden por clave y reproducción desde cualquier posición.
Al menos una vez más un consumidor idempotente da el efecto de exactamente una vez: en la lección, tras una caída a mitad, el consumidor reconstruye su estado desde el stream y los objetivos coinciden uno a uno con los del proceso único.

Recursos

Un núcleo, tres mundos

La regla de dependencias, comprobada

Un bus de eventos con cola

Colas entre procesos: Redis Pub/Sub y Streams

ZeroMQ: mensajería sin intermediario

Kafka: el registro distribuido

El núcleo sobre Redis Streams

Ejercicios

Footnotes

Autoevaluación.

Ideas clave.

Recursos.

Research

Documentación

Datos

Autoevaluación

Ideas clave

Recursos