Gestión de órdenes: estados, fills parciales, cancelaciones y reconexiones.

Lo que el broker contesta

En la lección 4.5.1 el broker simulado contestaba siempre, en orden y una vez. Un broker real no promete nada de eso. La documentación de la API de Interactive Brokers lo dice sin rodeos: es normal recibir mensajes de estado duplicados, no está garantizado que llegue uno por cada cambio de estado (una orden a mercado que se ejecuta al instante puede no dar ninguno) y por eso recomienda vigilar también los mensajes de ejecución11Nota 1 · Interactive Brokers, «TWS API: Placing Orders», apartados sobre los estados de una orden y orderStatus (consultado el 2026-10-22; enlace en los recursos).. Los datos del nivel traen cuatro meses de mensajes de un broker así: vivo4_eventos_broker.csv, en el orden en que llegaron, con el número de secuencia (seq) que el broker pone a cada mensaje de cada cuenta.

import heapq

import numpy as np
import pandas as pd

ord_ = pd.read_csv("vivo4_ordenes.csv")
ev = pd.read_csv("vivo4_eventos_broker.csv")
ev["llegada"] = np.arange(len(ev))        # orden de llegada
for c in ("ts_broker", "ts_recibido"):
    ev[c] = pd.to_datetime(ev[c])
print(len(ord_), "órdenes;", len(ev), "mensajes")
pd.crosstab(ev["evento"], ev["modo"], margins=True)

398 órdenes; 983 mensajes
modo       paper  real  All
evento
aceptada     134   262  396
cancelada     16    34   50
ejecutada    118   232  350
parcial       65   120  185
rechazada      1     1    2
All          334   649  983

Cinco tipos de mensaje: aceptada, parcial (una ejecución que no completa la orden), ejecutada (la que la completa), cancelada y rechazada. Cada uno lleva la cantidad de esa ejecución (cantidad_ultima), la acumulada, la pendiente y el precio medio de todo lo ejecutado: el estado completo de la orden, no solo el cambio. Esa redundancia es la que salvará el sistema cuando falte un mensaje.

Una orden límite de la cuenta real que se llena en parte y se cancela a los siete minutos, y la orden a mercado que el OMS manda con el resto:

print(ord_.set_index("id_orden_cliente").loc[
    ["R00047", "R00048"], ["instrumento", "lado", "tipo", "cantidad",
                           "precio_limite", "orden_origen"]])
cortos = {"id_orden_cliente": "orden", "cantidad_ultima": "q",
          "precio_ultimo": "precio", "cantidad_acumulada": "acum",
          "cantidad_pendiente": "pend"}
(ev.loc[ev["id_orden_cliente"].isin(["R00047", "R00048"]),
        ["seq", "evento", *cortos]]
   .rename(columns=cortos).set_index("seq"))

                 instrumento   lado     tipo  cantidad  precio_limite orden_origen
id_orden_cliente
R00047                  ORIG  venta   limite       222         112.71          NaN
R00048                  ORIG  venta  mercado        91            NaN       R00047
        evento   orden    q  precio  acum  pend
seq
99    aceptada  R00047    0     NaN     0   222
110    parcial  R00047  131  112.71   131    91
112  cancelada  R00047    0     NaN   131     0
113   aceptada  R00048    0     NaN     0    91
114  ejecutada  R00048   91  112.71    91     0

R00047 vende 222 ORIG a 112,71 como mínimo. Se ejecutan 131 al precio límite; a los siete minutos el OMS pide cancelar y el broker confirma la cancelación con la acumulada (131) y la pendiente a cero. El OMS manda entonces R00048, a mercado, por las 91 que faltan, con orden_origen apuntando a la límite. Son tres reglas que la máquina de estados tiene que comprobar: la acumulada es la suma de las ejecuciones, la pendiente es lo pedido menos lo acumulado (cero al cancelar) y el resto es exactamente lo que no se ejecutó.

La máquina de estados de una orden

El ciclo de vida de una orden está estandarizado desde hace décadas en el protocolo FIX (el campo OrdStatus) y cada broker usa un dialecto de él; el research del sitio sobre máquinas de estados lo cuenta con detalle. Los estados de esta lección y sus equivalentes:

Las listas completas de Alpaca e Interactive Brokers tienen más estados (done_for_day, expired, replaced…) 22Nota 2 · Alpaca, «Orders at Alpaca», apartado «Order Lifecycle» (consultado el 2026-10-22; enlace en los recursos).; un sistema que opera órdenes del día sin modificarlas puede tratarlos como variantes de los de la tabla, pero tiene que reconocerlos: un estado desconocido es un error, no un else.

stateDiagram-v2
  [*] --> enviada
  enviada --> aceptada
  enviada --> rechazada
  aceptada --> parcial
  aceptada --> ejecutada
  parcial --> parcial
  parcial --> ejecutada
  aceptada --> pend_cancel
  parcial --> pend_cancel
  pend_cancel --> pend_cancel : fill
  pend_cancel --> cancelada
  pend_cancel --> ejecutada : carrera
  ejecutada --> [*]
  cancelada --> [*]
  rechazada --> [*]

pend_cancel es pendiente_cancelacion: el OMS ha pedido cancelar y el broker aún no ha contestado. No es un detalle: mientras la petición viaja, la orden puede seguir ejecutándose, y si se llena entera antes de que llegue la cancelación, el broker contesta con una ejecución, no con una cancelación (la «carrera»). Un sistema que diera la orden por cancelada al pedirlo mandaría el resto a mercado y compraría dos veces.

La máquina se programa como una tabla de transiciones y una clase que la aplica y comprueba las invariantes. Cualquier par (estado, evento) que no esté en la tabla es una transición inválida y lanza una excepción: no se ignora.

FINALES = {"ejecutada", "cancelada", "rechazada"}
T = {
    ("enviada", "aceptada"): "aceptada",
    ("enviada", "rechazada"): "rechazada",
    ("aceptada", "parcial"): "parcial",
    ("aceptada", "ejecutada"): "ejecutada",
    ("parcial", "parcial"): "parcial",
    ("parcial", "ejecutada"): "ejecutada",
    ("aceptada", "pedir_cancelacion"): "pendiente_cancelacion",
    ("parcial", "pedir_cancelacion"): "pendiente_cancelacion",
    ("pendiente_cancelacion", "parcial"): "pendiente_cancelacion",
    ("pendiente_cancelacion", "ejecutada"): "ejecutada",
    ("pendiente_cancelacion", "cancelada"): "cancelada",
}

class Invalida(Exception):
    pass

class Orden:
    def __init__(self, id_, cantidad):
        self.id, self.cantidad = id_, cantidad
        self.estado, self.acumulada, self.valor = "enviada", 0, 0.0
        self.ejecuciones = {}          # id_ejecucion -> (q, precio)

    def aplicar(self, e):
        clave = (self.estado, e.evento)
        if clave not in T:
            raise Invalida(f"{self.id}: «{e.evento}» en "
                           f"«{self.estado}»")
        if e.evento in ("parcial", "ejecutada"):
            if e.id_ejecucion in self.ejecuciones:
                raise Invalida(f"{self.id}: {e.id_ejecucion} repetida")
            nueva = self.acumulada + e.cantidad_ultima
            if nueva != e.cantidad_acumulada:
                raise Invalida(f"{self.id}: acumulada {self.acumulada}"
                               f" + {e.cantidad_ultima} != "
                               f"{e.cantidad_acumulada}")
            self.ejecuciones[e.id_ejecucion] = (e.cantidad_ultima,
                                                e.precio_ultimo)
            self.acumulada = e.cantidad_acumulada
            self.valor += e.cantidad_ultima * e.precio_ultimo
        if e.evento == "cancelada" and (
                e.cantidad_acumulada != self.acumulada):
            raise Invalida(f"{self.id}: acumulada {self.acumulada}"
                           f" != {e.cantidad_acumulada} al cancelar")
        if e.evento in ("parcial", "ejecutada", "cancelada"):
            pend = 0 if e.evento == "cancelada" else (
                self.cantidad - self.acumulada)
            if e.cantidad_pendiente != pend:
                raise Invalida(f"{self.id}: pendiente "
                               f"{e.cantidad_pendiente} != {pend}")
        self.estado = T[clave]

La orden empieza en enviada cuando el OMS la manda (no cuando el broker la acepta) y el OMS añade su propio evento, pedir_cancelacion, a la hora en que lo pidió (ts_cancelacion de vivo4_ordenes.csv). Los demás eventos son los del broker.

Procesar los mensajes tal como llegan

El sistema más sencillo aplica cada mensaje en el orden de llegada. El flujo del OMS (envíos y peticiones de cancelación) se mezcla con el del broker por la hora del sistema: ts_envio y ts_cancelacion para los del OMS, ts_recibido para los del broker.

def eventos_oms(ordenes):
    """Los eventos que pone el propio OMS, con su hora."""
    filas = []
    for o in ordenes.itertuples():
        if pd.notna(o.ts_cancelacion):
            filas.append(dict(t=pd.Timestamp(o.ts_cancelacion),
                              id_orden_cliente=o.id_orden_cliente,
                              evento="pedir_cancelacion"))
    return pd.DataFrame(filas)

def procesar(flujo, ordenes):
    """flujo: tabla con t, id_orden_cliente, evento y campos."""
    libro = {o.id_orden_cliente: Orden(o.id_orden_cliente, o.cantidad)
             for o in ordenes.itertuples()}
    errores = []
    for e in flujo.itertuples():
        o = libro[e.id_orden_cliente]
        try:
            o.aplicar(e)
        except Invalida as x:
            errores.append((o.estado, e.evento, str(x)))
    return libro, errores

def tipo_error(estado, evento, texto):
    if estado == "enviada" and evento in ("parcial", "ejecutada"):
        return "ejecución antes de la aceptación"
    if estado in FINALES:
        return "mensaje tras un estado final"
    return "acumulada o pendiente que no cuadra"

def mezclar(broker, ordenes, col_t):
    b = broker.assign(t=broker[col_t])
    return (pd.concat([b, eventos_oms(ordenes)], ignore_index=True)
            .sort_values("t", kind="stable"))

libro_n, err_n = procesar(mezclar(ev, ord_, "ts_recibido"), ord_)
print(len(err_n), "errores")
print(pd.Series([tipo_error(*x) for x in err_n]).value_counts()
      .to_string())
print([x[2] for x in err_n[:2]])

53 errores
ejecución antes de la aceptación       28
acumulada o pendiente que no cuadra    21
mensaje tras un estado final            4
['P00007: «parcial» en «enviada»', 'P00007: acumulada 0 + 241.0 != 409.0']

Cincuenta y tres transiciones inválidas. Las 28 primeras son ejecuciones que llegan antes que la aceptación de su orden: el broker aceptó y ejecutó en milisegundos y los dos mensajes viajaron con latencias distintas. Las 21 de acumulada son casi todas consecuencia de esas: la máquina rechaza la ejecución adelantada y el siguiente mensaje de la orden ya no cuadra; entre ellas están también la de la ejecución parcial que llega después de la final (seq 393, anomalía A09 de los datos) y las de las dos ejecuciones que nunca llegaron. Las 4 últimas son los mensajes repetidos al reconectar, que llegan a órdenes ya terminadas. Con este procesamiento, un sistema que ignorase los errores tendría órdenes en estados absurdos; uno que contase todas las ejecuciones, posiciones falsas.

Un secuenciador por número de secuencia

El broker numera los mensajes de cada cuenta sin huecos. Eso permite reconstruir el orden: un secuenciador espera el seq siguiente al último entregado, retiene los que llegan adelantados y descarta los que ya entregó. Si el que falta no llega en un plazo, lo declara hueco y sigue: no se puede esperar para siempre.

flowchart TB
  L[Llega seq n] --> Q{n y esperado}
  Q -- menor --> D[Descartar repetido]
  Q -- igual --> E[Entregar y vaciar]
  Q -- mayor --> R[Retener]
  R --> V{Vence el plazo}
  V -- sí --> H[Declarar hueco]
  H --> E

El plazo se mide con el reloj de los mensajes (ts_recibido), no con el del ordenador: el secuenciador recibe la hora de cada llegada y comprueba los plazos vencidos antes de procesarla. Así el resultado es el mismo en cualquier máquina, cargada o no; en real, un temporizador del bucle de eventos le pasaría la hora cada poco.

class Secuenciador:
    def __init__(self, plazo=pd.Timedelta(seconds=2)):
        self.esperado, self.plazo = 1, plazo
        self.retenidos = {}          # seq -> (llegada, mensaje)
        self.salida, self.repetidos, self.huecos = [], [], []
        self.vistos = set()

    def _vaciar(self, t):
        while self.esperado in self.retenidos:
            _, m = self.retenidos.pop(self.esperado)
            self.salida.append((t, m))
            self.vistos.add(self.esperado)
            self.esperado += 1

    def vencer(self, t):
        """Declara los huecos cuyo plazo ha vencido antes de t."""
        while self.retenidos:
            limite = min(x for x, _ in self.retenidos.values()) \
                + self.plazo
            if limite > t:
                return
            nuevo = min(self.retenidos)
            self.huecos.extend(range(self.esperado, nuevo))
            self.esperado = nuevo
            self._vaciar(limite)

    def llegar(self, t, m):
        self.vencer(t)
        if m.seq in self.vistos or m.seq in self.retenidos:
            self.repetidos.append(m.seq)
            return
        if m.seq < self.esperado:            # llega tras su hueco
            self.salida.append((t, m))
            self.vistos.add(m.seq)
            return
        self.retenidos[m.seq] = (t, m)
        self._vaciar(t)

def secuenciar(mensajes):
    s = Secuenciador()
    for m in mensajes.itertuples():
        s.llegar(m.ts_recibido, m)
    s.vencer(pd.Timestamp.max.tz_localize("UTC"))
    filas = [m._asdict() | {"t": t} for t, m in s.salida]
    return pd.DataFrame(filas), s

sec, info = {}, {}
for modo, g in ev.groupby("modo", sort=False):
    sec[modo], info[modo] = secuenciar(g)
    print(f"{modo}: entregados {len(sec[modo])}, "
          f"repetidos {info[modo].repetidos}, "
          f"huecos {info[modo].huecos}")

paper: entregados 334, repetidos [], huecos []
real: entregados 645, repetidos [61, 62, 471, 475], huecos [202, 519]

En paper, los 334 mensajes se entregan sin repetidos ni huecos. En la cuenta real, el secuenciador descarta los cuatro mensajes repetidos y declara dos huecos, los seq 202 y 519: los dos mensajes que el broker no reenvió al reconectar. El mensaje 393, que llegó medio segundo tarde (después del 405), lo espera dentro del plazo de dos segundos y lo entrega en su sitio. ¿Cuánto cuesta esperar? El retraso entre la llegada y la entrega:

todo = pd.concat(sec.values(), ignore_index=True)
retraso = (todo["t"] - todo["ts_recibido"]).dt.total_seconds() * 1e3
print(f"Entregados en el acto: {(retraso == 0).mean():.1%}")
print("Retenidos:", (retraso > 0).sum(), "| mediana",
      f"{retraso[retraso > 0].median():.1f} ms | máximo",
      f"{retraso.max():,.0f} ms")

Entregados en el acto: 93.2%
Retenidos: 67 | mediana 7.1 ms | máximo 2,000 ms

El 93,2 % de los mensajes se entrega en el acto. Los 67 retenidos esperan una mediana de 7 ms (la ejecución que adelantó a su aceptación) y los peores, los de detrás de un hueco, el plazo completo: dos segundos. El plazo es un compromiso: más corto, y un mensaje que solo llega tarde se declara hueco; más largo, y el sistema se entera tarde de todo lo que hay detrás.

Recuperar lo que no llegó

Con los mensajes en orden, la máquina de estados ya solo encuentra lo que de verdad falta:

flujo = pd.concat([pd.concat(sec.values()), eventos_oms(ord_)],
                  ignore_index=True).sort_values("t", kind="stable")
libro, errores = procesar(flujo, ord_)
print(len(errores), "errores")
for x in errores:
    print(" ", x[2])
print(pd.Series({k: o.estado for k, o in libro.items()})
      .value_counts().to_dict())

2 errores
  R00079: acumulada 0 != 112.0 al cancelar
  R00214: acumulada 0 + 1016.0 != 1573.0
{'ejecutada': 345, 'cancelada': 49, 'rechazada': 2, 'pendiente_cancelacion': 1, 'aceptada': 1}

Dos errores, y son los dos huecos: en R00079 llega la cancelación con 112 acciones acumuladas cuando el sistema solo había contado 0; en R00214, la ejecución final con 1.573 acumuladas cuando había contado 0. Todas las demás órdenes recorren la máquina sin una sola transición inválida y terminan en un estado final, salvo esas dos, que se quedan donde las dejó el error.

El mensaje siguiente al hueco trae lo necesario para reconstruir lo perdido. La cantidad es la diferencia entre la acumulada del mensaje y lo contado más la ejecución del propio mensaje; el precio sale del precio medio:

con $A$ y $\overline{p}$ la acumulada y el precio medio del mensaje, $a$ y $V$ la cantidad y el valor contados hasta entonces, y $q$, $p$ la ejecución del propio mensaje (cero en una cancelación).

class OrdenRecuperable(Orden):
    def aplicar(self, e):
        q = e.cantidad_ultima if e.evento in ("parcial",
                                               "ejecutada") else 0
        if e.cantidad_acumulada > self.acumulada + q:
            perdida = e.cantidad_acumulada - self.acumulada - q
            px = ((e.precio_medio * e.cantidad_acumulada - self.valor
                   - q * (e.precio_ultimo if q else 0)) / perdida)
            self.ejecuciones[f"inferida-{e.seq}"] = (perdida,
                                                     round(px, 2))
            self.acumulada += perdida
            self.valor += perdida * px
            self.inferida = (perdida, round(px, 2))
        super().aplicar(e)

def procesar_rec(flujo, ordenes):
    libro = {o.id_orden_cliente: OrdenRecuperable(o.id_orden_cliente,
                                                  o.cantidad)
             for o in ordenes.itertuples()}
    errores = []
    for e in flujo.itertuples():
        try:
            libro[e.id_orden_cliente].aplicar(e)
        except Invalida as x:
            errores.append(str(x))
    return libro, errores

libro_r, err_r = procesar_rec(flujo, ord_)
print(len(err_r), "errores")
for k, o in libro_r.items():
    if hasattr(o, "inferida"):
        print(k, o.estado, "inferida:", o.inferida)
verdad = pd.read_csv("vivo4_anomalias_verdad.csv")
verdad.loc[verdad["tipo"] == "falta_ejecucion",
           ["id_anomalia", "id_orden_cliente", "descripcion"]]

0 errores
R00079 cancelada inferida: (112.0, 99.06)
R00214 ejecutada inferida: (557.0, 114.03)
   id_anomalia id_orden_cliente                                        descripcion
3          A04           R00079  La ejecución parcial E710000553.01 (112 título...
11         A12           R00214  La ejecución parcial E710001498.01 (557 título...

Cero errores, y las dos ejecuciones inferidas son exactamente las de la verdad de los datos: 112 acciones a 99,06 en R00079 y 557 a 114,03 en R00214. La inferencia es buena porque el broker manda el estado completo en cada mensaje; con un broker que solo enviara cambios no habría nada que despejar. Aun así, una ejecución inferida no es una ejecución confirmada: no tiene id_ejecucion ni comisión. El sistema debe marcarla, pedir al broker las ejecuciones del día (todas las APIs tienen esa consulta) y sustituirla por la real; si no cuadra, es un descuadre para la reconciliación (lección 4.5.5).

Fills parciales y cancelaciones

Que una orden se ejecute en varias partes es lo normal, no la excepción. Con el libro ya reconstruido:

n_ej = pd.Series({k: len(o.ejecuciones) for k, o in libro_r.items()})
tabla = ord_.assign(ejecuciones=ord_["id_orden_cliente"].map(n_ej))
print(pd.crosstab(tabla["tipo"], tabla["ejecuciones"]))
lim = tabla[tabla["tipo"] == "limite"].copy()
fin = lim["id_orden_cliente"].map(lambda k: libro_r[k].estado)
acu = lim["id_orden_cliente"].map(lambda k: libro_r[k].acumulada)
lim["destino"] = np.where(fin == "ejecutada", "llena",
                          np.where(acu > 0, "parcial", "nada"))
pd.crosstab(lim["modo"], lim["destino"], margins=True)

ejecuciones   0    1   2   3   4
tipo
cierre        0   56   0   0   0
limite       18   75   7   5   6
mercado       2  165  25  22  17
destino  llena  nada  parcial  All
modo
paper       22     6       10   38
real        39    12       22   73
All         61    18       32  111

Las órdenes al cierre (MOC) se ejecutan siempre en una sola parte, en la subasta. Las de mercado grandes se reparten en dos a cuatro ejecuciones; las límite, en una casi siempre, porque se llenan contra órdenes que llegan al precio. Lo que distingue a las límite es su destino: en la cuenta real, 39 de 73 se llenaron enteras, 22 en parte y 12 nada. Las 34 que no se llenaron enteras se cancelaron a los siete minutos y su resto se mandó a mercado. La regla del OMS para ese resto se comprueba sobre todas:

resto = ord_[ord_["motivo"] == "resto_de_limite"]
pedido = resto["cantidad"].to_numpy()
falta = resto["orden_origen"].map(
    lambda k: libro_r[k].cantidad - libro_r[k].acumulada).to_numpy()
cancel = resto["orden_origen"].map(lambda k: libro_r[k].estado)
print(len(resto), "restos;", "todos de órdenes canceladas:",
      (cancel == "cancelada").all(), "| cantidad correcta:",
      (pedido == falta).all())

50 restos; todos de órdenes canceladas: True | cantidad correcta: True

Los 50 restos (16 en paper, 34 en real) salen de una límite cancelada y piden justo lo que faltaba. Fíjate en que la comprobación usa la acumulada reconstruida: R00079 es una de esas límites, y sin la ejecución inferida de 112 acciones el resto habría parecido 112 acciones demasiado grande. Un OMS que calculase el resto con lo que ha contado, y no con la acumulada que confirma el broker en la cancelación, habría vendido 154 ORTN en vez de 42: 112 de más.

Los rechazos son pocos y conviene tratarlos como lo que son, una respuesta definitiva:

r = ev[ev["evento"] == "rechazada"]
r[["id_orden_cliente", "ts_broker", "motivo"]].merge(
    ord_[["id_orden_cliente", "instrumento", "cantidad"]])

  id_orden_cliente                        ts_broker  ... instrumento cantidad
0           P00005 2025-09-02 19:50:00.862198+00:00  ...        ORTL     1639
1           R00007 2025-10-14 13:35:00.585315+00:00  ...        ORIG     1260

[2 rows x 5 columns]

Dos compras rechazadas por falta de poder de compra, una en cada cuenta y las dos en los primeros días, cuando la cartera se monta desde efectivo. Reintentar la misma orden daría el mismo rechazo: el OMS no reintenta, y la siguiente señal recalcula la diferencia con lo que de verdad hay. Un rechazo por una causa transitoria (el mercado aún no ha abierto, un límite de peticiones) se trata distinto, y por eso el motivo debe guardarse con la orden.

Desconexiones y reconexiones

La conexión con el broker tiene su propio registro (vivo4_conexion.csv): cada sesión empieza con conectado y termina con desconectado («fin de sesión»), y entre medias hay siete cortes. Cada corte sigue el mismo guion: el sistema deja de recibir el latido del broker, a los tres segundos lo anota (latido_perdido), la conexión se cierra, el sistema vuelve a conectar con una sesión nueva y el broker reenvía lo que el sistema se perdió.

con = pd.read_csv("vivo4_conexion.csv", parse_dates=["ts"])
cortes = con[con["evento"] != "conectado"]
cortes = cortes[cortes["detalle"] != "fin de sesión"]
filas = []
for (modo, dia), g in cortes.groupby(["modo", cortes["ts"].dt.date]):
    t_d = g.loc[g["evento"] == "desconectado", "ts"].iloc[0]
    t_r = g.loc[g["evento"] == "reconectado", "ts"].iloc[0]
    cuenta = ev[ev["modo"] == modo]
    rep = cuenta[(cuenta["ts_broker"] < t_r)
                 & (cuenta["ts_recibido"] >= t_r)]
    seqs = range(rep["seq"].min(), rep["seq"].max() + 2) \
        if len(rep) else []
    filas.append(dict(modo=modo, dia=dia,
                      segundos=int((t_r - t_d).total_seconds()),
                      reenviados=len(rep),
                      repetidos=int(rep["seq"].duplicated().sum()),
                      huecos=[h for h in info[modo].huecos
                              if h in seqs]))
pd.DataFrame(filas)

    modo         dia  segundos  reenviados  repetidos huecos
0  paper  2025-09-23        42           0          0     []
1   real  2025-10-21        92          16          2     []
2   real  2025-11-03       137           4          0  [202]
3   real  2025-11-13        72          14          0     []
4   real  2025-12-03       117          18          0     []
5   real  2025-12-09        57          15          2     []
6   real  2025-12-15       107           6          0  [519]

Cortes de entre 42 y 137 segundos, todos a la hora más delicada: justo cuando el sistema manda las órdenes del cierre. El de paper cayó en un día sin órdenes y no dejó nada que reenviar. En los de la cuenta real, el broker reenvía al reconectar entre 4 y 18 mensajes generados durante el corte; en dos de ellos manda dos veces dos ejecuciones (los repetidos que descartó el secuenciador) y en otros dos se deja uno (los huecos 202 y 519). La figura de la cabecera resume qué dejó cada corte: repetidos, huecos, un mensaje fuera de orden y una orden duplicada.

La orden duplicada es la más cara, y no es culpa del broker. El 13 de noviembre, el OMS envió R00113 (venta de 7 ORUS) a las 20:50:00, cuando ya no llegaba el latido y dos segundos antes de que se cerrara la conexión. Al reconectar, sin la aceptación, la dio por perdida y la reenvió como R00114:

dup = ord_.set_index("id_orden_cliente").loc[["R00113", "R00114"]]
print(dup[["ts_envio", "lado", "cantidad", "motivo", "orden_origen"]])
t_rec = pd.Timestamp("2025-11-13 20:51:14", tz="UTC")
m = ev[ev["id_orden_cliente"].isin(["R00113", "R00114"])]
hora = lambda c: m[c].dt.strftime("%H:%M:%S.%f").str[:12]
m.assign(broker=hora("ts_broker"), recibido=hora("ts_recibido"))[
    ["seq", "id_orden_cliente", "evento", "broker", "recibido"]]

                                     ts_envio   lado  cantidad                 motivo orden_origen
id_orden_cliente
R00113            2025-11-13T20:50:00.246596Z  venta         7             rebalanceo          NaN
R00114            2025-11-13T20:51:14.296596Z  venta         7  reenvio_sin_respuesta       R00113
     seq id_orden_cliente     evento        broker      recibido
628  308           R00114   aceptada  20:51:14.345  20:51:14.375
629  309           R00114  ejecutada  20:51:14.389  20:51:14.422
631  295           R00113   aceptada  20:50:00.290  20:51:14.586
634  298           R00113  ejecutada  20:50:00.343  20:51:14.706

El broker había aceptado y ejecutado R00113 a las 20:50:00, pero sus mensajes no llegaron antes del corte. El OMS reenvió a las 20:51:14,296, justo al reconectar, y los mensajes de R00113 llegaron 290 ms después del reenvío. Resultado: dos ventas de 7 ORUS donde la señal pedía una. La alternativa correcta es preguntar antes de reenviar. Con el estado que tenía el broker en el instante del reenvío:

def estado_en_broker(id_cliente, t):
    """Lo que devolvería la consulta de la orden por su id de cliente
    (simulada con los mensajes que el broker ya había generado)."""
    m = ev[(ev["id_orden_cliente"] == id_cliente)
           & (ev["ts_broker"] <= t)].sort_values("seq")
    return "desconocida" if m.empty else m["evento"].iloc[-1]

def al_reconectar(en_vuelo, t):
    for id_cliente in en_vuelo:
        estado = estado_en_broker(id_cliente, t)
        accion = "enviar con el mismo id" if (
            estado == "desconocida") else "no reenviar"
        print(f"{id_cliente}: el broker dice «{estado}» -> {accion}")

al_reconectar(["R00113"], pd.Timestamp("2025-11-13 20:51:14.296",
                                       tz="UTC"))

R00113: el broker dice «ejecutada» -> no reenviar

La consulta habría dicho «ejecutada» y no habría habido reenvío. Si el broker no conociera la orden, lo seguro es enviarla con el mismo identificador de cliente: si la primera llegó, un broker que exige identificadores de cliente únicos rechaza la segunda por duplicada. Es la clave de idempotencia, que la lección 3.7.4 mostró con ccxt y que la lección 4.5.5 desarrolla junto con el estado persistente. El orden de las operaciones al reconectar queda así:

flowchart TB
  R[Reconectar] --> RE[Recibir reenvío]
  RE --> S[Secuenciar]
  S --> C[Consultar en vuelo]
  C --> E[Pedir ejecuciones]
  E --> O[Volver a operar]

Recibir y secuenciar el reenvío, consultar las órdenes que siguen sin estado final, pedir las ejecuciones del día para confirmar las inferidas y, solo entonces, volver a mandar órdenes. Mientras tanto, la señal puede seguir calculando, pero el OMS no envía.

El latido, con reloj inyectado

La desconexión se detecta por el latido (heartbeat): el broker manda un mensaje cada poco aunque no haya nada que decir, y si pasan tres segundos sin ninguno, la conexión se da por muerta. Como el secuenciador, el vigilante del latido no debe leer el reloj del ordenador: recibe la hora como argumento, y así se prueba con horas inventadas:

class Latido:
    def __init__(self, limite=pd.Timedelta(seconds=3)):
        self.limite, self.ultimo, self.caido = limite, None, False

    def recibir(self, t):
        self.ultimo, self.caido = t, False

    def comprobar(self, t):
        if not self.caido and t - self.ultimo >= self.limite:
            self.caido = True
            return f"{t.time()} sin latido desde {self.ultimo.time()}"
        return None

t0 = pd.Timestamp("2025-12-15 20:49:50", tz="UTC")
lat = Latido()
for s in range(0, 6):                      # latidos cada segundo
    lat.recibir(t0 + pd.Timedelta(seconds=s))
for s in range(6, 12):                     # el broker calla
    aviso = lat.comprobar(t0 + pd.Timedelta(seconds=s))
    if aviso:
        print(aviso)

20:49:58 sin latido desde 20:49:55

El último latido llega a las 20:49:55 y el vigilante salta a las 20:49:58: la hora a la que el registro de conexión de los datos anota «sin latido del broker en 3 s» en el corte del 15 de diciembre. La misma clase, en producción, recibe la hora de un temporizador del bucle de eventos; en las pruebas, la que se quiera, sin esperar ni un segundo.

Ejercicios

1. Cambia el plazo del secuenciador a 0,2 segundos. ¿Qué pasa con el mensaje 393 de la cuenta real? ¿Lo trata bien la máquina de estados?

s2 = Secuenciador(plazo=pd.Timedelta(seconds=0.2))
for m in ev[ev["modo"] == "real"].itertuples():
    s2.llegar(m.ts_recibido, m)
s2.vencer(pd.Timestamp.max.tz_localize("UTC"))
print("huecos:", s2.huecos)
orden_393 = [m.seq for _, m in s2.salida].index(393)
print("posición de entrega del 393:", orden_393, "| antes van:",
      [m.seq for _, m in s2.salida][orden_393 - 3:orden_393])
sal = pd.DataFrame([m._asdict() | {"t": t} for t, m in s2.salida])
ord_r = ord_[ord_["modo"] == "real"]
fl = pd.concat([sal, eventos_oms(ord_r)], ignore_index=True)
_, err2 = procesar_rec(fl.sort_values("t", kind="stable"), ord_r)
print(err2)

huecos: [202, 295, 296, 297, 298, 299, 300, 301, 302, 303, 304, 305, 306, 307, 393, 398, 399, 400, 401, 402, 403, 404, 519]
posición de entrega del 393: 403 | antes van: [402, 403, 404]
['R00162: «parcial» en «ejecutada»']

2. Comprueba que, en el libro reconstruido, ninguna orden en estado final tiene cantidad pendiente y que la acumulada de cada orden es la suma de sus ejecuciones (incluidas las inferidas).

mal = []
for k, o in libro_r.items():
    suma = sum(q for q, _ in o.ejecuciones.values())
    if suma != o.acumulada:
        mal.append((k, "suma"))
    if o.estado == "ejecutada" and o.acumulada != o.cantidad:
        mal.append((k, "ejecutada incompleta"))
    if o.estado not in FINALES:
        mal.append((k, o.estado))
print(len(libro_r), "órdenes;", len(mal), "incumplen:", mal)

398 órdenes; 0 incumplen: []

3. Una orden límite está en pendiente_cancelacion y llega una ejecución que la completa. ¿Qué debe hacer el OMS con el resto que pensaba enviar a mercado?