Tolerancia a fallos: idempotencia, estado persistente y reconciliación al arrancar.

Dónde se rompe un sistema de órdenes

La lección 4.5.2 siguió la vida de una orden con el sistema en marcha: estados, ejecuciones parciales, reconexiones. Esta trata de lo que pasa cuando algo se rompe en medio. Un gestor de órdenes (OMS) recibe una señal, guarda lo que va a hacer, envía la orden, recibe el acuse y las ejecuciones y actualiza posición y efectivo. Entre cada dos pasos hay una ventana en la que el proceso puede morir o la red cortarse:

flowchart TB
  S[Señal] --> G[Guardar intención]
  G --> E[Enviar orden]
  E --> A[Acuse del broker]
  A --> X[Ejecución]
  X --> P[Guardar ejecución]
  P --> C[Conciliar extracto]

Los datos del sistema ficticio del nivel 4 (sintéticos) traen todos esos fallos plantados en la cuenta real U-0418272, del 13 de octubre al 31 de diciembre de 2025, con su respuesta conocida en vivo4_anomalias_verdad.csv. La lección 4.5.2 ya trató los que se ven en el flujo de mensajes mientras el sistema opera (duplicados, desorden, huecos y la orden reenviada); esta añade lo que pasa cuando el que se cae es el propio proceso y lo que solo aparece al comparar con el extracto del broker. Al final, compara la lista de hallazgos con esa verdad. La figura de la cabecera adelanta el resultado: el efectivo que el sistema cree tener, tal como lo cuentan los eventos y después de conciliar, frente al del extracto.

Idempotencia: la misma orden, una sola vez

La lección 3.7.4 enseñó el caso básico: un timeout no dice si la orden llegó, y la solución es enviarla con un identificador propio (clientOrderId) para que el broker rechace el reintento como duplicado. La 4.5.2 analizó el error del sistema de los datos: el 13 de noviembre, durante una desconexión, no llegó el acuse de R00113 y el OMS la reenvió como R00114, con otra clave, sin preguntar antes al broker. El resumen, para lo que sigue:

import numpy as np
import pandas as pd

ev = pd.read_csv("vivo4_eventos_broker.csv")
real = ev[ev["modo"] == "real"].reset_index(drop=True)
ords = pd.read_csv("vivo4_ordenes.csv").set_index("id_orden_cliente")
sen = pd.read_csv("vivo4_senales.csv").set_index("id_senal")

par = ords.loc[["R00113", "R00114"]]
print(par[["id_senal", "ts_envio", "lado", "cantidad", "motivo",
           "orden_origen"]].to_string())
fills = real[real["id_orden_cliente"].isin(par.index)
             & real["id_ejecucion"].notna()]
print(fills[["seq", "id_orden_cliente", "cantidad_ultima",
             "precio_ultimo"]].to_string(index=False))
s = sen.loc["S00215"]
print("señal S00215:", s["posicion_actual"], "->",
      s["posicion_objetivo"], s["instrumento"])

                 id_senal                     ts_envio   lado  cantidad                 motivo orden_origen
id_orden_cliente
R00113             S00215  2025-11-13T20:50:00.246596Z  venta         7             rebalanceo          NaN
R00114             S00215  2025-11-13T20:51:14.296596Z  venta         7  reenvio_sin_respuesta       R00113
 seq id_orden_cliente  cantidad_ultima  precio_ultimo
 309           R00114                7         717.07
 298           R00113                7         716.82
señal S00215: 147 -> 140 ORUS

La señal pedía vender 7 ORUS (de 147 a 140). Las dos órdenes se ejecutaron: 14 vendidos. El broker no hizo nada mal; recibió dos órdenes con claves distintas. Preguntar antes de reenviar (la solución de la 4.5.2) exige saber qué preguntar, y ahí entra cómo se genera la clave. Dos formas, con un broker simulado que pierde la respuesta de la primera orden y un OMS que, al no recibirla, muere y se reinicia antes de reenviar la señal pendiente:

import uuid

class BrokerSim:
    """Acepta órdenes; rechaza una clave repetida; puede perder
    la respuesta de la próxima orden (la orden sí queda)."""

    def __init__(self):
        self.ordenes, self.perder = {}, 0

    def enviar(self, clave, instrumento, cantidad):
        if clave in self.ordenes:
            return {"estado": "duplicada", "clave": clave}
        self.ordenes[clave] = (instrumento, cantidad)
        if self.perder:
            self.perder -= 1
            raise TimeoutError("sin respuesta")
        return {"estado": "aceptada", "clave": clave}

    def consultar(self, clave):
        return self.ordenes.get(clave)

def clave_aleatoria(senal, cuenta):
    return uuid.uuid4().hex                 # nueva en cada llamada

def clave_de_senal(senal, cuenta):
    return f"{cuenta}-{senal}"              # sale de la intención

def ejecutar(generar, reinicios=1):
    """Envía S00215; la respuesta se pierde y el proceso se
    reinicia `reinicios` veces antes de reintentar."""
    b = BrokerSim()
    b.perder = 1
    for _ in range(reinicios + 1):          # cada vuelta: un proceso
        clave = generar("S00215", "U-0418272")
        try:
            r = b.enviar(clave, "ORUS", -7)
        except TimeoutError:
            continue                        # el proceso «muere»
        if r["estado"] == "duplicada":
            r = {"estado": "ya existía", **r}
        break
    return len(b.ordenes), r["estado"]

for nombre, g in (("uuid aleatorio", clave_aleatoria),
                  ("clave de la señal", clave_de_senal)):
    print(f"{nombre:18} órdenes en el broker: %d · %s" % ejecutar(g))

uuid aleatorio     órdenes en el broker: 2 · aceptada
clave de la señal  órdenes en el broker: 1 · duplicada

Con un UUID aleatorio, el proceso nuevo genera otra clave y el broker acepta una segunda orden: es exactamente lo que pasó con R00113 y R00114. Con una clave que sale de la intención (cuenta y señal; si una señal puede producir varias órdenes, también un número de orden dentro de la señal), el reintento lleva la misma clave aunque lo haga otro proceso, el broker la reconoce y el sistema solo tiene que consultar el estado de la orden que ya existe. La alternativa equivalente es generar la clave aleatoria y guardarla en disco antes de enviar: el proceso nuevo la lee en vez de inventar otra.

Dos precauciones con los brokers reales. No todos rechazan una clave repetida de la misma forma, ni todos la admiten (Alpaca, por ejemplo, la describe como identificador único de la orden de hasta 128 caracteres11Nota 1 · Documentación de Alpaca, POST /v2/orders: client_order_id, «A unique identifier for the order. Automatically generated if not sent. (<= 128 characters)». La página no documenta la respuesta a una clave repetida. Consultada el 22 de octubre de 2026.; en ccxt depende del exchange): hay que comprobar en el entorno de pruebas qué responde el broker con una clave repetida antes de fiarse de ello. Y si el broker no la admite, la única defensa es consultar las órdenes abiertas y las ejecuciones del día antes de reenviar (lección 3.7.4).

Estado persistente: un evento, una transacción

El OMS guarda su estado en SQLite, que viene con Python: un fichero, transacciones ACID y ninguna instalación. Cuatro tablas bastan para lo que importa aquí: los eventos recibidos (clave: seq, el número de secuencia del broker), las ejecuciones (clave: id_ejecucion), las posiciones y el efectivo. La regla de diseño es que cada evento del broker se aplica en una sola transacción: o se guardan el evento, la ejecución, la posición y el efectivo, o no se guarda nada.

import sqlite3

ESQUEMA = """
CREATE TABLE IF NOT EXISTS eventos (
  seq INTEGER PRIMARY KEY, id_orden TEXT, evento TEXT,
  id_ejecucion TEXT);
CREATE TABLE IF NOT EXISTS ejecuciones (
  id_ejecucion TEXT PRIMARY KEY, id_orden TEXT, instrumento TEXT,
  cantidad INTEGER, precio REAL, comision REAL, seq INTEGER);
CREATE TABLE IF NOT EXISTS posiciones (
  instrumento TEXT PRIMARY KEY, cantidad INTEGER NOT NULL);
CREATE TABLE IF NOT EXISTS efectivo (
  id INTEGER PRIMARY KEY CHECK (id = 1), saldo REAL NOT NULL);
"""
SIGNO = ords["lado"].map({"compra": 1, "venta": -1})
INSTR = ords["instrumento"]

def abrir(ruta):
    con = sqlite3.connect(ruta, isolation_level=None)  # BEGIN a mano
    con.execute("PRAGMA journal_mode=WAL")
    con.execute("PRAGMA synchronous=FULL")
    con.executescript(ESQUEMA)
    con.execute("INSERT OR IGNORE INTO efectivo VALUES (1, 1e6)")
    return con

class Caida(BaseException):
    """El proceso muere: nadie hace ROLLBACK ni cierra nada."""

def aplicar(con, e, morir=False):
    con.execute("BEGIN IMMEDIATE")
    nuevo = con.execute(
        "INSERT OR IGNORE INTO eventos VALUES (?, ?, ?, ?)",
        (int(e.seq), e.id_orden_cliente, e.evento,
         e.id_ejecucion if isinstance(e.id_ejecucion, str) else None)
    ).rowcount
    if nuevo and isinstance(e.id_ejecucion, str):
        q = int(SIGNO[e.id_orden_cliente] * e.cantidad_ultima)
        ins = INSTR[e.id_orden_cliente]
        con.execute("INSERT INTO ejecuciones VALUES (?,?,?,?,?,?,?)",
                    (e.id_ejecucion, e.id_orden_cliente, ins, q,
                     e.precio_ultimo, e.comision, int(e.seq)))
        con.execute("""INSERT INTO posiciones VALUES (?, ?)
            ON CONFLICT(instrumento)
            DO UPDATE SET cantidad = cantidad + excluded.cantidad""",
            (ins, q))
        if morir:
            raise Caida(f"muere en seq {e.seq}, antes del COMMIT")
        con.execute("UPDATE efectivo SET saldo = saldo - ? - ?",
                    (q * e.precio_ultimo, e.comision))
    con.execute("COMMIT")
    return nuevo

def estado(con):
    pos = dict(con.execute("SELECT instrumento, cantidad FROM "
                           "posiciones WHERE cantidad != 0"))
    saldo = con.execute("SELECT saldo FROM efectivo").fetchone()[0]
    n = con.execute("SELECT COUNT(*) FROM ejecuciones").fetchone()[0]
    return pos, round(saldo, 2), n

Tres detalles. isolation_level=None desactiva las transacciones implícitas del módulo sqlite3 para abrirlas y cerrarlas a mano (BEGIN IMMEDIATE toma el bloqueo de escritura al empezar). INSERT OR IGNORE sobre la clave primaria hace que un evento repetido no haga nada: rowcount vale 0 y no se toca la posición. Y el modo WAL con synchronous=FULL hace que un COMMIT terminado esté en disco aunque después se vaya la luz; SQLite garantiza que una transacción a medias se deshace al abrir de nuevo el fichero22Nota 2 · SQLite, «Atomic Commit In SQLite» (https://www.sqlite.org/atomiccommit.html) y «Write-Ahead Logging» (https://www.sqlite.org/wal.html): una transacción confirmada sobrevive a una caída del proceso o del sistema, y una sin confirmar se deshace al volver a abrir la base de datos.. La excepción Caida hereda de BaseException para que ningún except Exception la capture: representa la muerte del proceso, no un error que el código gestiona.

Primero, la referencia: todos los eventos de la cuenta real, en el orden en que llegaron, sin caídas.

from pathlib import Path

def limpiar(ruta):
    for sufijo in ("", "-wal", "-shm"):
        Path(ruta + sufijo).unlink(missing_ok=True)

limpiar("oms.db")
con = abrir("oms.db")
aplicados = sum(aplicar(con, e) for e in real.itertuples())
ref_pos, ref_saldo, ref_n = estado(con)
con.close()
print(len(real), "mensajes;", aplicados, "aplicados;",
      ref_n, "ejecuciones")
print("efectivo:", ref_saldo)
print(dict(sorted(ref_pos.items())))

649 mensajes; 645 aplicados; 348 ejecuciones
efectivo: 981849.26
{'ORAU': 488, 'ORIG': -557, 'ORTN': 112}

Seiscientos cuarenta y nueve mensajes, 645 aplicados y 348 ejecuciones: los cuatro restantes son repeticiones exactas de mensajes ya recibidos, con el mismo seq y la misma ejecución, que la clave primaria descarta. Son las ejecuciones que el broker reenvió al reconectar el 21 de octubre y el 9 de diciembre. Contarlas dos veces habría añadido títulos y gastos que no existen.

El proceso muere a mitad de un evento

Ahora la caída. Un OMS ingenuo actualiza la posición y el efectivo en una transacción y apunta el evento como procesado en otra; el transaccional lo hace todo en una. En los dos casos el proceso muere en el evento número 292 de la lista (la venta de 923 ORIG de R00109, el 6 de noviembre), se reabre el fichero como lo haría un proceso nuevo y el broker reenvía los mensajes desde el último que el sistema no había confirmado:

def aplicar_ingenuo(con, e, morir=False):
    """Posición y efectivo en una transacción; la marca del evento,
    en otra. Sin claves únicas en las ejecuciones."""
    hecho = con.execute("SELECT 1 FROM eventos WHERE seq = ?",
                        (int(e.seq),)).fetchone()
    if hecho:
        return 0
    if isinstance(e.id_ejecucion, str):
        q = int(SIGNO[e.id_orden_cliente] * e.cantidad_ultima)
        con.execute("BEGIN")
        con.execute("""INSERT INTO posiciones VALUES (?, ?)
            ON CONFLICT(instrumento)
            DO UPDATE SET cantidad = cantidad + excluded.cantidad""",
            (INSTR[e.id_orden_cliente], q))
        con.execute("UPDATE efectivo SET saldo = saldo - ? - ?",
                    (q * e.precio_ultimo, e.comision))
        con.execute("COMMIT")
        if morir:
            raise Caida(f"muere en seq {e.seq}, entre los COMMIT")
    con.execute("INSERT INTO eventos VALUES (?, ?, ?, ?)",
                (int(e.seq), e.id_orden_cliente, e.evento, None))
    return 1

K = 292                               # el evento en que muere
print(real.loc[K, ["seq", "id_orden_cliente", "evento",
                   "cantidad_ultima"]].to_dict(),
      INSTR[real.loc[K, "id_orden_cliente"]])

for nombre, fn in (("ingenuo", aplicar_ingenuo),
                   ("transaccional", aplicar)):
    limpiar("oms.db")
    con = abrir("oms.db")
    try:
        for i, e in enumerate(real.itertuples()):
            fn(con, e, morir=(i == K))
    except Caida as c:
        print(f"{nombre}: {c}")
    con.close()                       # el SO cierra el fichero
    con = abrir("oms.db")             # un proceso nuevo
    for e in real.iloc[K:].itertuples():   # el broker reenvía
        fn(con, e)
    pos, saldo, _ = estado(con)
    con.close()
    dif = {k: pos.get(k, 0) - ref_pos.get(k, 0)
           for k in set(pos) | set(ref_pos)
           if pos.get(k, 0) != ref_pos.get(k, 0)}
    print(f"  diferencia con la referencia: posiciones {dif}, "
          f"efectivo {saldo - ref_saldo:+,.2f}")

{'seq': 292, 'id_orden_cliente': 'R00109', 'evento': 'ejecutada', 'cantidad_ultima': 923} ORIG
ingenuo: muere en seq 292, entre los COMMIT
  diferencia con la referencia: posiciones {'ORIG': -923}, efectivo +105,197.23
transaccional: muere en seq 292, antes del COMMIT
  diferencia con la referencia: posiciones {}, efectivo +0.00

El ingenuo había confirmado la ejecución pero no la marca de evento procesado; al volver, el evento reenviado parece nuevo y la ejecución se cuenta otra vez: el sistema cree haber vendido 923 ORIG más de los que vendió y tener 105.197,23 USD de efectivo que no existen. El transaccional muere con la ejecución y la posición escritas pero sin COMMIT: SQLite las descarta, el evento reenviado se aplica entero una vez y el estado coincide al céntimo con el de la referencia. Una variante del mismo problema es morir después del COMMIT pero antes de avisar al broker de que el mensaje se procesó: el broker lo reenvía, y INSERT OR IGNORE lo descarta. Es la idea general: la red entrega «al menos una vez», y el procesamiento idempotente convierte eso en un efecto de «exactamente una vez».

Lo que llega del broker: duplicados, desorden y huecos

La lección 4.5.2 construyó un secuenciador que ordena los mensajes por seq, descarta repetidos y detecta huecos mientras el sistema opera. Para la conciliación basta con tener la lista de lo que encontró, en una celda: los duplicados (que la clave primaria ya descartó), las ejecuciones que llegan después de otra posterior de la misma orden (la cantidad acumulada retrocede, que es lo que rompe una máquina de estados) y los huecos de seq.

dup = real[real.duplicated(["seq", "evento", "id_ejecucion"])]
unicos = real.drop(dup.index)
fills = unicos[unicos["evento"].isin(["parcial", "ejecutada"])]
previo = (fills.groupby("id_orden_cliente")["cantidad_acumulada"]
          .transform(lambda s: s.cummax().shift()))
desorden = fills[fills["cantidad_acumulada"] < previo]
huecos = sorted(set(range(1, real["seq"].max() + 1))
                - set(real["seq"]))
tarde = (unicos["seq"] < unicos["seq"].cummax().shift()).sum()

print("duplicados:", dup[["seq", "id_ejecucion"]]
      .to_records(index=False).tolist())
print("llegan después de un seq mayor:", int(tarde))
print("ejecución que retrocede:",
      desorden[["seq", "id_orden_cliente", "id_ejecucion",
                "cantidad_acumulada"]].to_dict("records"))
print("huecos en seq:", huecos)

duplicados: [(61, 'E710000210.01'), (62, 'E710000168.01'), (471, 'E710001365.01'), (475, 'E710001386.01')]
llegan después de un seq mayor: 47
ejecución que retrocede: [{'seq': 393, 'id_orden_cliente': 'R00162', 'id_ejecucion': 'E710001134.01', 'cantidad_acumulada': 857}]
huecos en seq: [202, 519]

Cuatro duplicados, 47 mensajes que llegan después de otro con número mayor (casi todos acuses adelantados por su ejecución, o el lote que el broker reenvía tras una reconexión) y uno solo que importa: la ejecución E710001134.01 de R00162, con 857 títulos acumulados, llega cuando ya había llegado otra de la misma orden con 1.509. Un OMS que procese por orden de llegada pondría la orden en 857 después de haberla dado por ejecutada; uno que ordene por seq antes de aplicar la máquina de estados, no. Y dos huecos, el 202 y el 519: dos mensajes que no llegaron y que el broker no reenvió. La 4.5.2 reconstruyó su cantidad y su precio a partir de la acumulada del mensaje siguiente; el extracto lo confirma o lo corrige.

Conciliación con el extracto: ejecuciones

El extracto del broker (vivo4_extracto_*) es la verdad contable de la cuenta: lo que el broker liquida, cobra y custodia. La conciliación compara cuatro cosas, de lo más concreto a lo más agregado: ejecuciones, posiciones, efectivo y, al final, órdenes frente a señales.

ex = pd.read_csv("vivo4_extracto_operaciones.csv")
ej = (unicos[unicos["id_ejecucion"].notna()]
      [["id_ejecucion", "id_orden_cliente", "cantidad_ultima",
        "precio_ultimo", "comision", "seq"]])
m = ex.merge(ej, on="id_ejecucion", how="outer", indicator=True,
             suffixes=("_ext", "_ev"))
print(m["_merge"].value_counts().to_dict())
solo_ext = m[m["_merge"] == "left_only"]
cliente_de = (real.dropna(subset=["id_orden_broker"])
              .drop_duplicates("id_orden_broker")
              .set_index("id_orden_broker")["id_orden_cliente"])
print(solo_ext[["fecha_operacion", "id_ejecucion", "instrumento",
                "lado", "cantidad", "precio"]].to_string(index=False))
ambos = m[m["_merge"] == "both"]
otro_px = ambos[(ambos["precio"] - ambos["precio_ultimo"]).abs() > 1e-9]
otro_q = ambos[ambos["cantidad"] != ambos["cantidad_ultima"]]
otra_com = ambos[(ambos["comision_ext"] - ambos["comision_ev"]).abs()
                 > 1e-9]
print("precio distinto:", otro_px[["fecha_operacion", "id_ejecucion",
      "cantidad", "precio", "precio_ultimo"]].to_dict("records"))
print("cantidad distinta:", len(otro_q),
      "| comisión distinta:", len(otra_com))

{'both': 348, 'left_only': 2, 'right_only': 0}
fecha_operacion  id_ejecucion instrumento   lado  cantidad  precio
     2025-11-03 E710000553.01        ORTN  venta       112   99.06
     2025-12-15 E710001498.01        ORIG compra       557  114.03
precio distinto: [{'fecha_operacion': '2025-12-03', 'id_ejecucion': 'E710001162.01', 'cantidad': 170, 'precio': 115.34, 'precio_ultimo': 115.29}]
cantidad distinta: 0 | comisión distinta: 0

Trescientas cuarenta y ocho ejecuciones casan por identificador. Dos están en el extracto y no en los eventos: son exactamente los dos huecos, una venta parcial de 112 ORTN del 3 de noviembre y una compra parcial de 557 ORIG del 15 de diciembre, que el broker no reenvió al reconectar. Ninguna está en los eventos y no en el extracto. Y una casa por identificador pero no por precio: el broker liquidó E710001162.01 a 115,34 en vez de los 115,29 del mensaje de ejecución, una corrección posterior de 8,50 USD en 170 títulos. Cantidades y comisiones coinciden en todas.

Posiciones: lo que el broker hizo sin avisar

Con las ejecuciones del extracto incorporadas (las que faltaban y el precio corregido), las posiciones del sistema deberían ser las del extracto cada día. Las posiciones se reconstruyen sumando ejecuciones por fecha y se comparan con vivo4_extracto_posiciones.csv, que da la posición de cierre de cada día:

ej_f = ej.assign(
    fecha=ords.loc[ej["id_orden_cliente"], "fecha"].to_numpy(),
    instrumento=INSTR[ej["id_orden_cliente"]].to_numpy(),
    q=(SIGNO[ej["id_orden_cliente"]].to_numpy()
       * ej["cantidad_ultima"].to_numpy()))
falta = solo_ext.assign(
    fecha=solo_ext["fecha_operacion"],
    q=np.where(solo_ext["lado"] == "compra", 1, -1)
      * solo_ext["cantidad"])
todas = pd.concat([ej_f[["fecha", "instrumento", "q"]],
                   falta[["fecha", "instrumento", "q"]]])
pos_ext = (pd.read_csv("vivo4_extracto_posiciones.csv")
           .pivot(index="fecha", columns="instrumento",
                  values="cantidad").fillna(0))
pos_sis = (todas.pivot_table(index="fecha", columns="instrumento",
                             values="q", aggfunc="sum")
           .reindex(pos_ext.index).fillna(0).cumsum())
dif = (pos_ext - pos_sis.reindex(columns=pos_ext.columns,
                                 fill_value=0))
dif = dif.loc[:, (dif != 0).any()]
print(dif[(dif != 0).any(axis=1)].head(3))
d0 = dif.index[(dif != 0).any(axis=1)][0]
sis = pos_ext.loc[d0, "ORAU"] - dif.loc[d0, "ORAU"]
x = pos_ext.loc[d0, "ORAU"] / sis
px = pd.read_csv("vivo4_extracto_posiciones.csv").query(
    "instrumento == 'ORAU'").set_index("fecha")["precio_cierre"]
print(f"{d0}: razón {x:.4f}; cierre {px.shift().loc[d0]} -> "
      f"{px.loc[d0]} (razón {px.shift().loc[d0] / px.loc[d0]:.3f})")
efe = pd.read_csv("vivo4_extracto_efectivo.csv")
print(efe[efe["concepto"] == "desdoblamiento"].to_string(index=False))

instrumento   ORAU
fecha
2025-11-17   374.0
2025-11-18   374.0
2025-11-19   374.0
2025-11-17: razón 3.0000; cierre 369.01 -> 124.15 (razón 2.972)
     fecha       concepto instrumento  importe   saldo
2025-11-17 desdoblamiento        ORAU      0.0 6297.88

Todas las posiciones cuadran todos los días salvo una, ORAU, desde el 17 de noviembre: el extracto tiene 374 títulos más, exactamente el doble de los 187 que había. La razón es 3,0000, el precio de cierre pasa de 369,01 a 124,15 (una razón de 2,97 que combina el desdoblamiento y el movimiento del día) y el extracto de efectivo tiene una línea desdoblamiento sin importe. Es un desdoblamiento 3 por 1 que el sistema no conocía. La diferencia sigue los días siguientes con el mismo tamaño: las órdenes posteriores de ORAU se calcularon con la posición mala.

Un desdoblamiento no lo manda el broker como ejecución, y un sistema que no lo incorpore calcula mal todo lo que depende de la posición: el peso real de ORAU era el triple del que creía, y la orden siguiente para llevarlo al objetivo compró más de la cuenta. Los proveedores de datos y los brokers publican las operaciones corporativas por adelantado (lección 1.4.3); el sistema debe leerlas antes de la apertura y ajustar posición y precios de referencia.

Efectivo: intereses, tasas y correcciones

El efectivo del sistema es el inicial más los importes de las ejecuciones. Comparado con el saldo del extracto al final de cada día, con las ejecuciones corregidas, las diferencias que quedan son movimientos que no son operaciones:

imp = ex.groupby("fecha_operacion")["importe_neto"].sum()
saldo_ext = efe.groupby("fecha")["saldo"].last()
saldo_ops = 1e6 + imp.reindex(saldo_ext.index, fill_value=0).cumsum()
resto = (saldo_ext - saldo_ops).round(2)
saltos = resto.diff().fillna(resto).round(2)
print(saltos[saltos != 0].to_dict())
otros = efe[~efe["concepto"].isin(["ingreso", "operaciones",
                                    "comisiones"])]
print(otros.groupby("fecha")["importe"].sum().round(2).to_dict())

# lo mismo con las ejecuciones de los EVENTOS, sin corregir
ev_imp = (ej_f.assign(imp=-ej_f["q"] * ej_f["precio_ultimo"]
                      - ej_f["comision"])
          .groupby("fecha")["imp"].sum())
ingenuo = 1e6 + ev_imp.reindex(saldo_ext.index, fill_value=0).cumsum()
d_ing = (saldo_ext - ingenuo).round(2)
print("diferencia sin corregir, al final:", d_ing.iloc[-1],
      "| corregida:", resto.iloc[-1])

{'2025-10-31': 164.77, '2025-11-28': 42.07, '2025-12-31': 276.76}
{'2025-10-31': 164.77, '2025-11-17': 0.0, '2025-11-28': 42.07, '2025-12-31': 276.76}
diferencia sin corregir, al final: -51947.56 | corregida: 483.6

Con las ejecuciones del extracto, el efectivo solo se separa del sistema tres días, los tres últimos días hábiles de mes: +164,77, +42,07 y +276,76 USD. Coinciden con la suma de las líneas intereses y tasa_datos_mercado del extracto (174,77 − 10, 52,07 − 10, 286,76 − 10): intereses del efectivo y una tasa de datos de mercado que el broker abona y carga sin mandar ningún evento. Con las ejecuciones de los eventos sin corregir, la diferencia final es de −51.947,56 USD: la compra de ORIG que el sistema no vio (−63.516,66 con su comisión), la venta de ORTN que tampoco vio (+11.094,00 neta), la corrección de precio de 170 ORRE (−8,50) y los intereses y tasas (+483,60). La figura de la cabecera sigue las dos curvas día a día: la de los eventos tal cual salta en cada fallo; la conciliada solo se mueve el último día de cada mes, por importes que el extracto explica.

Órdenes frente a señales

Lo que falta no está en el broker: es la orden duplicada del principio. Eventos y extracto coinciden en ella porque las dos órdenes existieron. Solo se ve al comparar, para cada señal, lo ejecutado con lo que pedía:

o = ords[ords["modo"] == "real"].copy()
lleno = ej_f.groupby("id_orden_cliente")["q"].sum()
# las dos ejecuciones del extracto, a su orden (por id de broker)
for _, f in solo_ext.iterrows():
    cli = cliente_de[f["id_orden_broker"]]
    lleno[cli] = lleno.get(cli, 0) + (1 if f["lado"] == "compra"
                                      else -1) * f["cantidad"]
o["ejecutado"] = lleno.reindex(o.index).fillna(0)
por_senal = o.groupby("id_senal")["ejecutado"].sum()
pide = (sen["posicion_objetivo"] - sen["posicion_actual"])
desv = (por_senal - pide.reindex(por_senal.index)).astype(int)
for s, d in desv[desv != 0].items():
    os_ = o[o["id_senal"] == s]
    print(s, sen.loc[s, "instrumento"], f"{d:+d}",
          os_.index.tolist(), os_["motivo"].tolist())
rech = real[real["evento"] == "rechazada"]
print(rech[["id_orden_cliente", "motivo"]].to_string(index=False))

S00125 ORIG -1260 ['R00007'] ['rebalanceo']
S00215 ORUS -7 ['R00113', 'R00114'] ['rebalanceo', 'reenvio_sin_respuesta']
id_orden_cliente                       motivo
          R00007 poder de compra insuficiente

Dos señales no quedaron como pedían. S00125 no ejecutó nada porque su orden fue rechazada por poder de compra insuficiente: un descuadre real, pero ya explicado y comunicado por el broker (es una de las alertas de la lección 4.5.4). S00215 tiene 7 ORUS vendidos de más, con dos órdenes, la segunda marcada como reenvio_sin_respuesta: la orden duplicada. Nada en el broker estaba mal; el error estaba en el OMS.

Las trece, una a una

Juntando todas las comprobaciones, la conciliación produce una lista de hallazgos con su tipo, su fecha y sus identificadores. La respuesta conocida de los datos permite comprobar si la lista es completa y si sobra algo:

fecha_seq = dict(zip(real["seq"], real["ts_broker"].str[:10]))
h = []
for _, r in dup.iterrows():
    h.append(("ejecucion_duplicada", fecha_seq[r["seq"]],
              r["id_orden_cliente"], r["id_ejecucion"]))
for _, r in desorden.iterrows():
    h.append(("fuera_de_orden", fecha_seq[r["seq"]],
              r["id_orden_cliente"], r["id_ejecucion"]))
for _, r in solo_ext.iterrows():
    cli = cliente_de[r["id_orden_broker"]]
    h.append(("falta_ejecucion", r["fecha_operacion"], cli,
              r["id_ejecucion"]))
for _, r in otro_px.iterrows():
    h.append(("correccion_precio", r["fecha_operacion"],
              r["id_orden_cliente"], r["id_ejecucion"]))
h.append(("desdoblamiento", d0, None, None))
split_f = efe.loc[efe["concepto"] == "desdoblamiento", "fecha"].values
for f in otros["fecha"].unique():
    if f not in split_f:
        h.append(("tasas_e_intereses", f, None, None))
for s, d in desv[desv != 0].items():
    os_ = o[o["id_senal"] == s]
    if (os_["motivo"] == "reenvio_sin_respuesta").any():
        h.append(("orden_duplicada", os_["fecha"].iloc[0],
                  os_.index[-1], None))

hall = pd.DataFrame(h, columns=["tipo", "fecha", "id_orden_cliente",
                                "id_ejecucion"])
verdad = pd.read_csv("vivo4_anomalias_verdad.csv")
claves = ["tipo", "fecha", "id_orden_cliente", "id_ejecucion"]
cruce = verdad.merge(hall, on=claves, how="outer", indicator=True)
print(cruce["_merge"].value_counts().to_dict())
print(cruce.sort_values("fecha")[["id_anomalia", "tipo", "fecha",
      "id_orden_cliente"]].to_string(index=False))

{'both': 13, 'left_only': 0, 'right_only': 0}
id_anomalia                tipo      fecha id_orden_cliente
        A02 ejecucion_duplicada 2025-10-21           R00024
        A01 ejecucion_duplicada 2025-10-21           R00030
        A03   tasas_e_intereses 2025-10-31              NaN
        A04     falta_ejecucion 2025-11-03           R00079
        A05     orden_duplicada 2025-11-13           R00114
        A06      desdoblamiento 2025-11-17              NaN
        A07   tasas_e_intereses 2025-11-28              NaN
        A08   correccion_precio 2025-12-03           R00166
        A09      fuera_de_orden 2025-12-03           R00162
        A10 ejecucion_duplicada 2025-12-09           R00195
        A11 ejecucion_duplicada 2025-12-09           R00198
        A12     falta_ejecucion 2025-12-15           R00214
        A13   tasas_e_intereses 2025-12-31              NaN

Trece hallazgos y trece anomalías plantadas, y las trece casan por tipo, fecha, orden y ejecución: ni falta ninguna ni sobra ninguna. La figura las ordena por la comprobación que las descubre: cinco (la corrección de precio, el desdoblamiento y los tres cierres de mes con intereses y tasas) no dejan rastro en el flujo de mensajes y solo aparecen contra el extracto, y la orden duplicada solo aparece contra las señales. El rechazo de S00125 no aparece porque la regla de órdenes duplicadas exige un reenvío; un rechazo lo comunica el broker y va por otro camino (la alerta). Que la lista coincida con la verdad depende de que los datos están hechos para eso; en una cuenta real, la conciliación da una lista y alguien tiene que decidir qué es cada cosa. Lo que se puede automatizar es la clasificación de las firmas conocidas, que es lo que hace la celda.

Qué hacer con cada diferencia

Encontrar la diferencia es la mitad. La otra es decidir, para cada tipo, si el sistema la corrige solo, la corrige y avisa o se para:

La regla de la última fila es la más importante: si al arrancar queda una diferencia que no encaja en ninguna firma conocida, el sistema no opera. Operar sobre un estado falso propaga el error (lo que pasó con ORAU durante dos semanas). El arranque de un sistema en vivo, por tanto, es una pequeña máquina de estados:

stateDiagram-v2
  [*] --> CARGANDO
  CARGANDO --> CONCILIANDO
  CONCILIANDO --> OPERANDO
  CONCILIANDO --> CORRIGIENDO
  CORRIGIENDO --> CONCILIANDO
  CONCILIANDO --> BLOQUEADO
  BLOQUEADO --> CONCILIANDO
  OPERANDO --> CONCILIANDO

CARGANDO lee el estado del disco; CONCILIANDO pide al broker órdenes abiertas, ejecuciones y posiciones (y, si hay extracto nuevo, lo compara); con todo explicado pasa a OPERANDO; con diferencias de firma conocida, a CORRIGIENDO, que aplica la corrección y vuelve a conciliar; con algo sin explicar, a BLOQUEADO, que manda una alerta crítica (lección 4.5.4) y espera a una persona. Desde OPERANDO se vuelve a conciliar tras cada reconexión y cada mañana. Es el mismo principio que el interruptor de la lección 4.2.5, aplicado al estado en vez de al resultado. NautilusTrader hace este paso al arrancar su nodo en vivo, como se verá en la lección 4.5.6.

Ejercicios

1. Sustituye en aplicar la clave de las ejecuciones por un INSERT sin OR IGNORE y vuelve a procesar todos los eventos de la cuenta real sin caídas. ¿Qué pasa con los duplicados? ¿Por qué es preferible que falle a que los cuente dos veces?

def aplicar_estricto(con, e):
    con.execute("BEGIN IMMEDIATE")
    try:
        con.execute("INSERT INTO eventos VALUES (?, ?, ?, ?)",
                    (int(e.seq), e.id_orden_cliente, e.evento, None))
        con.execute("COMMIT")
        return "ok"
    except sqlite3.IntegrityError as err:
        con.execute("ROLLBACK")
        return f"seq {e.seq}: {err}"

limpiar("estricto.db")
con = abrir("estricto.db")
r = [aplicar_estricto(con, e) for e in real.itertuples()]
con.close()
print([x for x in r if x != "ok"])

['seq 61: UNIQUE constraint failed: eventos.seq', 'seq 62: UNIQUE constraint failed: eventos.seq', 'seq 471: UNIQUE constraint failed: eventos.seq', 'seq 475: UNIQUE constraint failed: eventos.seq']

2. Escribe una función que, dada la tabla de posiciones diarias del extracto y la del sistema, detecte desdoblamientos sin mirar el extracto de efectivo: una diferencia que aparece un día sin ejecuciones del instrumento y cuya razón con la posición del sistema es un entero o el inverso de un entero.

def desdoblamientos(pos_ext, pos_sis, ejecs):
    out = []
    for ins in pos_ext.columns:
        d = pos_ext[ins] - pos_sis.get(ins, 0)
        nuevos = d[(d != 0) & (d.shift(fill_value=0) == 0)]
        for fecha in nuevos.index:
            hubo = ((ejecs["instrumento"] == ins)
                    & (ejecs["fecha"] == fecha)).any()
            antes = pos_sis.loc[fecha, ins]
            if hubo or antes == 0:
                continue
            r = pos_ext.loc[fecha, ins] / antes
            for k in (r, 1 / r):
                if abs(k - round(k)) < 1e-9 and round(k) > 1:
                    out.append((ins, fecha, f"{r:.4g}"))
    return out

desdoblamientos(pos_ext, pos_sis, todas)

[('ORAU', '2025-11-17', '3')]

3. Un compañero propone no conciliar con el extracto y fiarse de los eventos, porque «el broker siempre manda todo». Con los datos de esta lección, ¿qué le responderías?