Seguridad y continuidad: 2FA, IP, copias y procedimientos ante fallos.

Qué puede parar un sistema de trading

La lección 4.4.5 protegió las credenciales del broker en el código: dónde se guardan, cómo no se escapan a los logs, qué permisos tienen y cómo se rotan. La 4.5.5 hizo que el gestor de órdenes sobreviviera a una caída del proceso, y la 4.2.5 añadió el kill switch. Esta lección mira el sistema desplegado de las lecciones 7.2.1 y 7.2.2 desde fuera: quién puede entrar en la cuenta del broker o en el servidor, qué se pierde si el disco muere, y qué hace exactamente el sistema, y la persona que lo vigila, cuando falla una pieza que no controla.

En el mundo regulado esto tiene nombre. El Reglamento Delegado (UE) 2017/589 (RTS 6) exige a las empresas de servicios de inversión con negociación algorítmica «dispositivos de continuidad de las actividades» documentados, con escenarios adversos (indisponibilidad de sistemas, proveedores o centros de datos, pérdida o alteración de datos), dispositivos alternativos para gestionar órdenes y posiciones pendientes, y revisión y prueba anuales (artículo 14)11Nota 1 · Reglamento Delegado (UE) 2017/589 de la Comisión, artículo 14, «Dispositivos de continuidad de las actividades», apartados 2 b) y g) y 4 (DOUE L 87 de 31.3.2017, PDF en https://www.boe.es/doue/2017/087/L00417-00448.pdf).. El Reglamento (UE) 2022/2554 (DORA), aplicable desde el 17 de enero de 2025, pide a las entidades financieras políticas de respuesta y recuperación (artículo 11) y de copias de seguridad con pruebas periódicas de restauración en sistemas separados del de origen (artículo 12)22Nota 2 · Reglamento (UE) 2022/2554 (DORA), artículos 11, «Respuesta y recuperación», 12, «Políticas y procedimientos de respaldo y procedimientos y métodos de restablecimiento y recuperación», y 64 (aplicación desde el 17 de enero de 2025) (DOUE L 333 de 27.12.2022, PDF en https://www.boe.es/doue/2022/333/L00001-00079.pdf).. Un particular no está sujeto a ninguno de los dos; las listas que contienen son, aun así, una buena guía de qué pensar.

Segundo factor: TOTP según el RFC 6238

Casi todos los brokers y exchanges ofrecen un segundo factor con una aplicación de autenticación. Lo que hace esa aplicación cabe en diez líneas. HOTP (RFC 4226) calcula un HMAC-SHA-1 de un contador con una clave compartida, toma 4 bytes en una posición que decide el último nibble del resultado (el «truncado dinámico») y se queda con los últimos dígitos. TOTP (RFC 6238) usa como contador el número de pasos de 30 segundos desde el 1 de enero de 1970:

import hashlib
import hmac
import struct

def hotp(clave, contador, digitos=6, alg="sha1"):
    mac = hmac.new(clave, struct.pack(">Q", contador), alg).digest()
    o = mac[-1] & 0x0F                       # truncado dinámico
    n = int.from_bytes(mac[o:o + 4]) & 0x7FFFFFFF
    return str(n % 10 ** digitos).zfill(digitos)

def totp(clave, t, digitos=6, alg="sha1", paso=30, t0=0):
    return hotp(clave, (int(t) - t0) // paso, digitos, alg)

Una implementación criptográfica se valida contra los vectores del estándar, no a ojo. El RFC 4226 publica los diez primeros valores HOTP con la clave ASCII «12345678901234567890» (apéndice D) y el RFC 6238, seis instantes con SHA-1, SHA-256 y SHA-512 en códigos de 8 dígitos (apéndice B)33Nota 3 · RFC 4226, apéndice D, y RFC 6238, apéndices A y B (https://www.rfc-editor.org/rfc/rfc6238).:

SEMILLA = b"12345678901234567890"
RFC4226 = ["755224", "287082", "359152", "969429", "338314",
           "254676", "287922", "162583", "399871", "520489"]
ok_hotp = [hotp(SEMILLA, i) for i in range(10)] == RFC4226

RFC6238 = {          # t: (SHA1, SHA256, SHA512), apéndice B
    59: ("94287082", "46119246", "90693936"),
    1111111109: ("07081804", "68084774", "25091201"),
    1111111111: ("14050471", "67062674", "99943326"),
    1234567890: ("89005924", "91819424", "93441116"),
    2000000000: ("69279037", "90698825", "38618901"),
    20000000000: ("65353130", "77737706", "47863826"),
}
# semillas del código de referencia (apéndice A): longitud del hash
SEMILLAS = {"sha1": SEMILLA, "sha256": (SEMILLA * 2)[:32],
            "sha512": (SEMILLA * 4)[:64]}
aciertos = sum(totp(SEMILLAS[a], t, 8, a) == esperado[i]
               for t, esperado in RFC6238.items()
               for i, a in enumerate(SEMILLAS))
print("RFC 4226, 10 valores HOTP:", ok_hotp)
print("RFC 6238, 18 valores TOTP:", aciertos, "de 18")
print("SHA-256 con la semilla de 20 bytes:",
      totp(SEMILLA, 59, 8, "sha256"), "en vez de 46119246")

RFC 4226, 10 valores HOTP: True
RFC 6238, 18 valores TOTP: 18 de 18
SHA-256 con la semilla de 20 bytes: 32247374 en vez de 46119246

Las 28 comprobaciones pasan. La tercera línea muestra una trampa conocida del RFC 6238: el texto del apéndice B dice que la semilla es la misma cadena de 20 bytes para los tres modos, pero el programa de referencia del apéndice A, que genera la tabla, usa semillas de 32 y 64 bytes para SHA-256 y SHA-512. Quien implementa siguiendo solo el texto cree que su SHA-256 está mal. Por eso hay que validar contra todos los vectores y leer el código de referencia.

Generar el código es la mitad fácil. El verificador, el lado del servidor, es donde están las decisiones de seguridad, y el RFC y la guía SP 800-63B-4 del NIST las fijan44Nota 4 · RFC 6238, secciones 5.2 («at most one time step is allowed as the network delay»; «The verifier MUST NOT accept the second attempt of the OTP after the successful validation») y 6; NIST SP 800-63B-4 (https://pages.nist.gov/800-63-4/sp800-63b.html): clave de al menos 112 bits, límite de 100 fallos consecutivos y «OTP authentication is not phishing-resistant». Consultados el 23 de octubre de 2026.:

• Clave: aleatoria, con al menos 112 bits de seguridad (NIST); el RFC recomienda la longitud de la salida del HMAC, 160 bits con SHA-1. Se comparte una vez, normalmente con un código QR que codifica un URI otpauth://.
• Ventana: el RFC recomienda aceptar como mucho un paso de retraso por la red; muchos verificadores aceptan uno a cada lado por el desfase de reloj del teléfono.
• Un solo uso: tras una validación correcta, el mismo código no se acepta otra vez («MUST NOT»).
• Límite de intentos: NIST exige limitar los fallos consecutivos de un autenticador a no más de 100 antes de desactivarlo.
• Comparación en tiempo constante: hmac.compare_digest, para no filtrar por el tiempo de respuesta cuántos dígitos coinciden.

import base64
import secrets

class VerificadorTOTP:
    def __init__(self, clave, ventana=1, max_fallos=100):
        self.clave, self.ventana = clave, ventana
        self.max_fallos, self.fallos = max_fallos, 0
        self.ultimo_paso = -1

    def verificar(self, codigo, ahora):
        if self.fallos >= self.max_fallos:
            return "bloqueado"
        paso = int(ahora) // 30
        for d in range(-self.ventana, self.ventana + 1):
            p = paso + d
            if hmac.compare_digest(hotp(self.clave, p), codigo):
                if p <= self.ultimo_paso:
                    return "reutilizado"
                self.ultimo_paso, self.fallos = p, 0
                return "ok"
        self.fallos += 1
        return "fallo"

clave = secrets.token_bytes(20)              # 160 bits
v = VerificadorTOTP(clave)
T = 1_766_000_000                            # reloj inyectado
c = totp(clave, T)
print("código actual:      ", v.verificar(c, T))
print("mismo código, +10 s:", v.verificar(c, T + 10))
print("reloj del móvil -40 s:",
      v.verificar(totp(clave, T + 60 - 40), T + 60))
print("reloj del móvil -70 s:",
      v.verificar(totp(clave, T + 120 - 70), T + 120))
for _ in range(100):
    v.verificar("000000", T + 200)
print("tras 100 fallos:    ", v.verificar(totp(clave, T + 230),
                                         T + 230))

p = 1 - (1 - 3 / 10 ** 6) ** 100
print(f"adivinar con 100 intentos y 3 códigos válidos: {p:.4%}")
b32 = base64.b32encode(SEMILLA).decode().rstrip("=")
print(f"otpauth://totp/Broker:demo?secret={b32}&issuer=Broker")

código actual:       ok
mismo código, +10 s: reutilizado
reloj del móvil -40 s: ok
reloj del móvil -70 s: fallo
tras 100 fallos:     bloqueado
adivinar con 100 intentos y 3 códigos válidos: 0.0300%
otpauth://totp/Broker:demo?secret=GEZDGNBVGY3TQOJQGEZDGNBVGY3TQOJQ&issuer=Broker

La ventana de un paso acepta un móvil con el reloj 40 segundos atrasado y rechaza uno con 70. El código reutilizado 10 segundos después se rechaza, que es lo que impide que alguien que lo ha visto pasar (un vídeo, un hombro, un log) lo use. Tras 100 fallos el factor queda bloqueado incluso para el código correcto: desbloquearlo es un proceso de recuperación de cuenta, no un reintento. La probabilidad de acertar a ciegas con 100 intentos, con tres códigos válidos a la vez, es del 0,03 %. La última línea es el formato de la aplicación de Google Authenticator: la clave en Base32 sin relleno, con el emisor55Nota 5 · Google Authenticator, «Key Uri Format» (https://github.com/google/google-authenticator/wiki/Key-Uri-Format).; la de la celda es la semilla de prueba del RFC, nunca una real.

Listas blancas de IP

La lección 4.4.5 restringió la clave de API del exchange a la IP del servidor. El mismo principio protege el servidor: el SSH, el panel de Grafana y cualquier endpoint de administración solo aceptan conexiones de unas pocas direcciones. La biblioteca estándar trae ipaddress, que entiende redes IPv4 e IPv6 y sus prefijos. Dos errores típicos: fiarse de la cabecera X-Forwarded-For, que escribe el cliente, y meter en la lista una red mucho más amplia de lo previsto:

import ipaddress as ipa

LISTA = [ipa.ip_network(r) for r in (
    "203.0.113.10/32",          # VPS de despliegue
    "198.51.100.0/24",          # oficina
    "2001:db8:10::/48",         # IPv6 de casa
)]
PROXY = ipa.ip_address("192.0.2.5")         # el proxy propio

def ip_cliente(socket_ip, xff=None):
    ip = ipa.ip_address(socket_ip)
    if getattr(ip, "ipv4_mapped", None):
        ip = ip.ipv4_mapped                 # ::ffff:a.b.c.d
    if ip == PROXY and xff:
        return ipa.ip_address(xff.split(",")[-1].strip())
    return ip

def permitido(socket_ip, xff=None, lista=LISTA):
    ip = ip_cliente(socket_ip, xff)
    return any(ip.version == n.version and ip in n for n in lista)

ACCESOS = [
    ("203.0.113.10", None),             # el VPS
    ("::ffff:203.0.113.10", None),      # el VPS por IPv6 mapeada
    ("192.0.2.66", "198.51.100.7"),     # cabecera falsa
    ("192.0.2.5", "198.51.100.7"),      # vía el proxy propio
    ("2001:db8:10:4::2", None),         # casa, IPv6
    ("2001:db8:11::2", None),           # otra red IPv6
]
for s, x in ACCESOS:
    ingenuo = ipa.ip_address(x or s)
    ing = any(ingenuo.version == n.version and ingenuo in n
              for n in LISTA)
    print(f"{s:<22} xff={str(x):<13} ingenuo={ing!s:<5} "
          f"bueno={permitido(s, x)}")

203.0.113.10           xff=None          ingenuo=True  bueno=True
::ffff:203.0.113.10    xff=None          ingenuo=False bueno=True
192.0.2.66             xff=198.51.100.7  ingenuo=True  bueno=False
192.0.2.5              xff=198.51.100.7  ingenuo=True  bueno=True
2001:db8:10:4::2       xff=None          ingenuo=True  bueno=True
2001:db8:11::2         xff=None          ingenuo=False bueno=False

El verificador ingenuo, que toma la cabecera si existe, deja entrar la petición con la cabecera falsa y rechaza el VPS cuando llega como dirección IPv6 mapeada. El bueno cree la cabecera solo cuando la conexión viene del proxy propio y normaliza las direcciones mapeadas. Una lista blanca también se revisa a sí misma: una entrada con demasiadas direcciones suele ser un error de escritura (un /16 donde iba un /32), y una red privada en la lista de un servidor público no tiene sentido:

PRIVADAS = [ipa.ip_network(r) for r in (
    "10.0.0.0/8", "172.16.0.0/12", "192.168.0.0/16", "fc00::/7")]

def revisar_lista(redes, max_dir=256):
    out = []
    for r in map(ipa.ip_network, redes):
        if r.prefixlen == 0:
            out.append((str(r), "todo Internet"))
        elif r.version == 4 and r.num_addresses > max_dir:
            out.append((str(r), f"{r.num_addresses:,} direcciones"))
        elif r.version == 6 and r.prefixlen < 48:
            out.append((str(r), f"/{r.prefixlen} en IPv6"))
        if any(r.version == p.version and r.subnet_of(p)
               for p in PRIVADAS):
            out.append((str(r), "red privada"))
    return out

print(revisar_lista(["203.0.113.10/32", "198.51.0.0/16",
                     "10.0.0.0/8", "0.0.0.0/0", "2001:db8::/32"]))

[('198.51.0.0/16', '65,536 direcciones'), ('10.0.0.0/8', '16,777,216 direcciones'), ('10.0.0.0/8', 'red privada'), ('0.0.0.0/0', 'todo Internet'), ('2001:db8::/32', '/32 en IPv6')]

La lista blanca tiene un coste operativo: si la IP de casa cambia, uno se queda fuera de su propio servidor. Por eso se prevé un segundo camino antes de necesitarlo: la consola web del proveedor (que no pasa por el SSH del servidor), un equipo de salto con IP fija o una VPN propia. El procedimiento «me he quedado fuera» también es parte de la continuidad.

Copias de seguridad que se pueden restaurar

Para un bot, buena parte del estado se puede reconstruir: las ejecuciones, posiciones y efectivo están en el extracto del broker, que es la verdad contable (lección 4.5.5). Lo que no se reconstruye es lo propio: el estado del gestor de órdenes con sus claves de idempotencia, las señales y decisiones registradas, los modelos entrenados, la configuración. Esas son las copias que importan. El proceso tiene cuatro partes:

flowchart TB
  E[Estado en vivo] --> C[Copia coherente]
  C --> M[Manifiesto SHA-256]
  M --> T[Tres destinos]
  T --> R[Restauración]

La primera trampa es la copia en sí. La documentación de SQLite advierte que copiar el fichero de una base de datos mientras una transacción está a medias puede producir una copia con parte del contenido viejo y parte del nuevo, corrupta66Nota 6 · SQLite, «How To Corrupt An SQLite Database File», apartado 1.2 (https://www.sqlite.org/howtocorrupt.html), y «VACUUM» (VACUUM INTO) (https://www.sqlite.org/lang_vacuum.html).. La API de copia en línea, Connection.backup en Python, o VACUUM INTO producen una copia coherente aunque el bot siga escribiendo. La celda crea el estado del bot con las 398 órdenes del sistema del nivel 4, lo copia así y empaqueta todo con un manifiesto:

import io
import json
import sqlite3
import tarfile
import gzip
from pathlib import Path

import pandas as pd

VIVO = Path("vivo")
(VIVO / "config").mkdir(parents=True, exist_ok=True)
con = sqlite3.connect(VIVO / "oms.db")
pd.read_csv("vivo4_ordenes.csv").to_sql("ordenes", con,
                                        index=False)
con.commit()
(VIVO / "config/bot.yaml").write_text("modo: real\nriesgo: 0.02\n")
(VIVO / "modelo.json").write_text(json.dumps({"pesos": [0.4, 0.6]}))

def sha256(b):
    return hashlib.sha256(b).hexdigest()

def hacer_copia(origen, destino):
    st = Path("staging")
    (st / "config").mkdir(parents=True, exist_ok=True)
    (st / "oms.db").unlink(missing_ok=True)  # copia desde cero
    with sqlite3.connect(origen / "oms.db") as src, \
            sqlite3.connect(st / "oms.db") as dst:
        src.backup(dst)                     # coherente en línea
    for f in ("config/bot.yaml", "modelo.json"):
        (st / f).write_bytes((origen / f).read_bytes())
    ficheros = sorted(p for p in st.rglob("*") if p.is_file())
    manif = {p.relative_to(st).as_posix(): sha256(p.read_bytes())
             for p in ficheros}
    buf = io.BytesIO()
    with gzip.GzipFile(fileobj=buf, mode="wb", mtime=0) as gz, \
            tarfile.open(fileobj=gz, mode="w") as tar:
        datos = json.dumps(manif, indent=1).encode()
        for nombre, contenido in [("MANIFIESTO.json", datos)] + [
                (k, (st / k).read_bytes()) for k in manif]:
            ti = tarfile.TarInfo(nombre)
            ti.size, ti.mtime, ti.mode = len(contenido), 0, 0o600
            tar.addfile(ti, io.BytesIO(contenido))
    destino.write_bytes(buf.getvalue())
    return sha256(buf.getvalue()), manif

huella, manif = hacer_copia(VIVO, Path("copia.tar.gz"))
print(len(manif), "ficheros;", "huella", huella[:16])
print(hacer_copia(VIVO, Path("otra.tar.gz"))[0] == huella)

3 ficheros; huella a46e3b029b1a31a1
True

El paquete es reproducible: fechas a cero, propietarios vacíos, orden fijo y gzip sin marca de tiempo. Dos copias del mismo estado dan la misma huella, así que una huella distinta significa un estado distinto, no otro momento. La huella del paquete se guarda aparte (en el registro de copias); el manifiesto, dentro, con la de cada fichero.

La regla 3-2-1 que divulga CISA, la agencia de ciberseguridad de Estados Unidos, resume dónde guardarlas: tres copias de cada fichero importante (la original y dos más), en dos tipos de soporte, una de ellas fuera del sitio77Nota 7 · CISA (US-CERT), «Data Backup Options» (https://www.cisa.gov/sites/default/files/publications/data_backup_options.pdf): «3 – Keep 3 copies of any important file: 1 primary and 2 backups. 2 – Keep the files on 2 different media types […]. 1 – Store 1 copy offsite».. Para un bot en un VPS: el disco del servidor, un volumen o almacenamiento de objetos del proveedor y una copia en otro proveedor o en casa. Una copia en el mismo disco no protege de nada que le pase al disco. La celda simula las tres copias y dos accidentes: la del segundo destino pierde un bit en el transporte y la del tercero se cortó a medias.

import shutil

DESTINOS = {d: Path(d) for d in ("local", "objetos", "fuera")}
for d, p in DESTINOS.items():
    p.mkdir(exist_ok=True)
    shutil.copy("copia.tar.gz", p / "copia.tar.gz")
b = bytearray((DESTINOS["objetos"] / "copia.tar.gz").read_bytes())
b[len(b) // 2] ^= 0x01                         # un bit cambiado
(DESTINOS["objetos"] / "copia.tar.gz").write_bytes(b)
f = DESTINOS["fuera"] / "copia.tar.gz"
f.write_bytes(f.read_bytes()[: f.stat().st_size // 3])  # cortada

def verificar(paquete, huella):
    datos = paquete.read_bytes()
    if sha256(datos) != huella:
        return "huella distinta"
    with tarfile.open(fileobj=io.BytesIO(datos)) as tar:
        manif = json.load(tar.extractfile("MANIFIESTO.json"))
        for nombre, h in manif.items():
            if sha256(tar.extractfile(nombre).read()) != h:
                return f"{nombre} no coincide"
    return f"ok ({len(manif)} ficheros)"

for d, p in DESTINOS.items():
    print(f"{d:<8}", verificar(p / "copia.tar.gz", huella))

local    ok (3 ficheros)
objetos  huella distinta
fuera    huella distinta

La huella del paquete detecta los dos accidentes antes de abrirlo. El manifiesto añade la comprobación fichero a fichero, útil cuando el paquete se ha reconstruido o cuando se verifica una copia que alguien descomprimió. Pero que los hashes coincidan solo dice que la copia es la que se hizo, no que sirva. La prueba de verdad es restaurar en otro sitio y usarla:

RESTAURADO = Path("restaurado")
with tarfile.open(DESTINOS["local"] / "copia.tar.gz") as tar:
    tar.extractall(RESTAURADO, filter="data")
r = sqlite3.connect(RESTAURADO / "oms.db")
o = sqlite3.connect(VIVO / "oms.db")
print("integrity_check:",
      r.execute("PRAGMA integrity_check").fetchone()[0])
consulta = ("SELECT modo, COUNT(*), SUM(cantidad) FROM ordenes "
            "GROUP BY modo ORDER BY modo")
print("restaurado:", r.execute(consulta).fetchall())
print("original:  ", o.execute(consulta).fetchall())
r.close()
o.close()

integrity_check: ok
restaurado: [('paper', 135, 73817), ('real', 263, 128062)]
original:   [('paper', 135, 73817), ('real', 263, 128062)]

La base de datos restaurada pasa PRAGMA integrity_check y da las mismas órdenes y cantidades que la original. filter="data" en extractall impide que un paquete manipulado escriba fuera de la carpeta de destino o cree ficheros especiales. Esta prueba se automatiza y se programa (lección 4.5.4): cada semana, restaurar la última copia de cada destino en una carpeta temporal y comprobarla, con alerta si falla. DORA pide a las entidades financieras exactamente eso, pruebas periódicas de restauración y hacerlas en sistemas separados del de origen.

Dos cifras ordenan la política. El objetivo de punto de recuperación (RPO) es el momento al que hay que poder volver tras un fallo; el de tiempo de recuperación (RTO), cuánto puede durar la recuperación antes de causar daño (definiciones de NIST SP 800-34)88Nota 8 · NIST CSRC, glosario: «Recovery Time Objective» y «Recovery Point Objective», con fuente NIST SP 800-34 Rev. 1 (https://csrc.nist.gov/glossary/term/recovery_point_objective).. Con una copia diaria tras el cierre, el RPO es de un día: se pierde lo del día, que en un bot se reconstruye en buena parte desde el extracto. El RTO depende de lo ensayado: si restaurar exige recordar comandos a las tres de la madrugada, no hay RTO.

Procedimientos ante fallos como tablas de decisión

Un procedimiento escrito en prosa («si se cae el broker, valorar la situación y actuar con prudencia») no sirve en el momento en que se necesita. Lo que sirve es una tabla: para cada tipo de incidente y estado del sistema, qué se hace. Escrita como datos, una tabla se puede comprobar con código: que cubre todas las combinaciones y que ninguna regla queda tapada por una anterior. El sistema tiene cuatro estados, los del kill switch de la lección 4.2.5 con uno intermedio:

stateDiagram-v2
  [*] --> OPERANDO
  OPERANDO --> DEGRADADO
  DEGRADADO --> OPERANDO
  DEGRADADO --> SOLO_REDUCIR
  SOLO_REDUCIR --> PARADO
  DEGRADADO --> PARADO
  PARADO --> OPERANDO

DEGRADADO no abre posiciones nuevas pero mantiene lo que hay; SOLO_REDUCIR solo acepta órdenes que reducen riesgo; PARADO cancela lo que pueda y espera a una persona. La vuelta a OPERANDO pasa siempre por una conciliación (lección 4.5.5). La tabla usa tres condiciones: el incidente, si hay órdenes vivas y si el mercado está abierto.

from itertools import product

INCIDENTES = ["broker_caido", "datos_congelados", "datos_anomalos",
              "red_caida", "credencial_expuesta", "disco_lleno"]
DOMINIO = {"incidente": INCIDENTES, "ordenes_vivas": [True, False],
           "mercado_abierto": [True, False]}
X = "*"                                      # cualquier valor
TABLA = [  # (incidente, órdenes vivas, mercado abierto) -> acción
    (("credencial_expuesta", X, X), "PARADO",
     "revocar la clave en la consola del broker; rotar (4.4.5)"),
    ((X, X, False), "DEGRADADO",
     "registrar; resolver y conciliar antes de la apertura"),
    (("broker_caido", True, True), "DEGRADADO",
     "no reenviar; esperar 90 s; después, mesa del broker"),
    (("broker_caido", False, True), "DEGRADADO",
     "esperar la reconexión; conciliar al volver"),
    (("datos_congelados", X, True), "SOLO_REDUCIR",
     "cancelar límites vivos; fuente de reserva"),
    (("datos_anomalos", X, True), "DEGRADADO",
     "descartar el dato; contrastar con la fuente secundaria"),
    (("red_caida", True, True), "DEGRADADO",
     "canal alternativo (móvil) para cancelar o confirmar"),
    (("red_caida", False, True), "DEGRADADO",
     "esperar; al volver, conciliar antes de operar"),
]

def coincide(patron, caso):
    return all(p == X or p == c for p, c in zip(patron, caso))

def decidir(caso, tabla=TABLA):
    for patron, estado, accion in tabla:
        if coincide(patron, caso):
            return estado, accion
    return None

def comprobar(tabla):
    casos = list(product(*DOMINIO.values()))
    huecos = [c for c in casos if decidir(c, tabla) is None]
    usadas = {next(i for i, (p, _, _) in enumerate(tabla)
                   if coincide(p, c))
              for c in casos if c not in huecos}
    tapadas = [i for i in range(len(tabla)) if i not in usadas]
    return len(casos), huecos, tapadas

total, huecos, tapadas = comprobar(TABLA)
print(total, "casos;", len(huecos), "huecos:",
      sorted({h[0] for h in huecos}), "| reglas tapadas:", tapadas)

24 casos; 2 huecos: ['disco_lleno'] | reglas tapadas: []

Veinticuatro combinaciones y dos huecos: nadie escribió qué hacer con el disco lleno con el mercado abierto. Es el tipo de olvido que la comprobación encuentra y una lectura no. La versión corregida añade la regla, y para ver la segunda comprobación, alguien pega por error una regla general antes de una específica:

TABLA.append((("disco_lleno", X, True), "SOLO_REDUCIR",
              "rotar logs y liberar espacio; alerta"))
print(comprobar(TABLA)[1:])
mala = TABLA[:2] + [(("broker_caido", X, X), "PARADO",
                     "parar todo")] + TABLA[2:]
print(comprobar(mala)[1:])
print(decidir(("broker_caido", True, True), mala))

([], [])
([], [3, 4])
('PARADO', 'parar todo')

Con la regla nueva no quedan huecos ni reglas tapadas. En la tabla mala, la regla general de la posición 2 tapa a las dos específicas del broker (posiciones 3 y 4): ante una caída del broker con órdenes vivas, el sistema pararía todo en vez de esperar y conciliar, que con órdenes en el aire es justo lo peligroso. La comprobación lo señala antes de que pase.

Un simulacro con desconexiones reales

El sistema ficticio del nivel 4 dejó en vivo4_conexion.csv su registro de conexión con el broker: inicio y fin de cada sesión y, en siete ocasiones, un latido perdido, una desconexión y una reconexión. El simulacro pasa esos siete incidentes por la tabla, con un reloj que avanza según el registro, y apunta qué hace el sistema en cada momento. Primero, las órdenes que estaban en el aire al cortarse la conexión:

cx = pd.read_csv("vivo4_conexion.csv", parse_dates=["ts"])
ords = pd.read_csv("vivo4_ordenes.csv", parse_dates=["ts_envio"])
inc = []
for (modo, dia), g in cx.groupby(["modo", cx["ts"].dt.date]):
    if "latido_perdido" not in set(g["evento"]):
        continue
    def primero(evento, desde=g["ts"].min()):
        return g.loc[(g["evento"] == evento) & (g["ts"] >= desde),
                     "ts"].iloc[0]
    ev = {"latido_perdido": primero("latido_perdido")}
    ev["desconectado"] = primero("desconectado", ev["latido_perdido"])
    ev["reconectado"] = primero("reconectado", ev["desconectado"])
    o = ords[(ords["modo"] == modo) & (ords["ts_envio"].dt.date == dia)
             & (ords["ts_envio"] < ev["desconectado"])]
    ultimo = o["ts_envio"].max()
    inc.append({"fecha": str(dia), "modo": modo,
                "en_el_aire": len(o),
                "tras_envio_s": (ev["desconectado"] - ultimo)
                .total_seconds(),
                "caida_s": (ev["reconectado"] - ev["desconectado"])
                .total_seconds(),
                "aviso_s": (ev["desconectado"] - ev["latido_perdido"])
                .total_seconds()})
INC = pd.DataFrame(inc)
print(INC.round(1).to_string(index=False))

     fecha  modo  en_el_aire  tras_envio_s  caida_s  aviso_s
2025-09-23 paper           0           NaN     42.0      3.0
2025-10-21  real           9           0.1     92.0      3.0
2025-11-03  real           2           0.4    137.0      3.0
2025-11-13  real           7           1.2     72.0      3.0
2025-12-03  real          10           0.1    117.0      3.0
2025-12-09  real           7           0.2     57.0      3.0
2025-12-15  real           5           0.1    107.0      3.0

Las seis desconexiones de la cuenta real empiezan entre 0,1 y 1,2 segundos después del último envío de las órdenes del cierre (a las 15:50 de Nueva York), con entre 2 y 10 órdenes en el aire; la de paper ocurrió un día sin órdenes. Duran entre 42 y 137 segundos, y el latido perdido avisa siempre 3 segundos antes del corte. Ahora el simulacro: el sistema pasa a DEGRADADO con el latido perdido, consulta la tabla al cortarse la conexión y, si a los 90 segundos no ha vuelto, escala a la mesa del broker; al reconectar, concilia y vuelve a OPERANDO.

ESCALADO_S = 90

def simulacro(fila):
    t, log = 0.0, []
    log.append((t, "latido perdido", "DEGRADADO"))
    t += fila.aviso_s
    estado, accion = decidir(("broker_caido", fila.en_el_aire > 0,
                              True))
    log.append((t, "desconexión: " + accion, estado))
    if fila.caida_s > ESCALADO_S:
        log.append((t + ESCALADO_S, "llamar a la mesa: confirmar "
                    f"{fila.en_el_aire} órdenes por teléfono",
                    "DEGRADADO"))
    t += fila.caida_s
    log.append((t, "reconexión: conciliar por clave (4.5.5)",
                "OPERANDO"))
    return log

for _, fila in INC.iterrows():
    print(fila.fecha, fila.modo)
    for t, que, estado in simulacro(fila):
        print(f"  +{t:5.0f} s  {estado:<10} {que}")
    if fila.fecha == "2025-11-03":
        break
esc = INC[INC["caida_s"] > ESCALADO_S]
print("escalados a la mesa:", len(esc), "de", len(INC),
      list(esc["fecha"]))

2025-09-23 paper
  +    0 s  DEGRADADO  latido perdido
  +    3 s  DEGRADADO  desconexión: esperar la reconexión; conciliar al volver
  +   45 s  OPERANDO   reconexión: conciliar por clave (4.5.5)
2025-10-21 real
  +    0 s  DEGRADADO  latido perdido
  +    3 s  DEGRADADO  desconexión: no reenviar; esperar 90 s; después, mesa del broker
  +   93 s  DEGRADADO  llamar a la mesa: confirmar 9 órdenes por teléfono
  +   95 s  OPERANDO   reconexión: conciliar por clave (4.5.5)
2025-11-03 real
  +    0 s  DEGRADADO  latido perdido
  +    3 s  DEGRADADO  desconexión: no reenviar; esperar 90 s; después, mesa del broker
  +   93 s  DEGRADADO  llamar a la mesa: confirmar 2 órdenes por teléfono
  +  140 s  OPERANDO   reconexión: conciliar por clave (4.5.5)
escalados a la mesa: 4 de 7 ['2025-10-21', '2025-11-03', '2025-12-03', '2025-12-15']

La celda imprime las líneas de tiempo de los primeros incidentes hasta el de mayor duración, el 3 de noviembre, con 137 segundos de corte y 2 órdenes en el aire. Con el umbral de 90 segundos, cuatro de los siete incidentes habrían escalado a la mesa del broker. La figura de la cabecera los ordena por fecha con el umbral marcado. La elección del umbral es un compromiso que el simulacro hace visible: más bajo, más llamadas por cortes que se resuelven solos; más alto, menos margen antes del cierre, cuando una orden al cierre ya no se puede cancelar.

El simulacro se cierra con lo que dijo la conciliación. La lección 4.5.5 encontró trece anomalías en la cuenta real; ¿cuántas cayeron en días de desconexión?

anom = pd.read_csv("vivo4_anomalias_verdad.csv")
reales = INC[INC["modo"] == "real"]["fecha"]
cruce = anom[anom["fecha"].isin(reales)]
print(cruce.groupby("fecha")["tipo"].apply(list).to_string())
print(len(cruce), "de", len(anom), "anomalías en días de corte;",
      "días de corte sin anomalía:",
      sorted(set(reales) - set(cruce["fecha"])))

fecha
2025-10-21    [ejecucion_duplicada, ejecucion_duplicada]
2025-11-03                             [falta_ejecucion]
2025-11-13                             [orden_duplicada]
2025-12-03           [correccion_precio, fuera_de_orden]
2025-12-09    [ejecucion_duplicada, ejecucion_duplicada]
2025-12-15                             [falta_ejecucion]
9 de 13 anomalías en días de corte; días de corte sin anomalía: []

Los seis cortes de la cuenta real dejaron rastro en la conciliación: ejecuciones duplicadas que el broker reenvió al reconectar, ejecuciones que no llegó a reenviar, un mensaje fuera de orden y la orden duplicada R00114, que el gestor reenvió durante el corte del 13 de noviembre sin preguntar. Nueve de las trece anomalías están en esos seis días. La conclusión práctica del simulacro es doble: el procedimiento correcto ante una desconexión no es actuar deprisa sino no empeorar las cosas (no reenviar) y conciliar al volver; y los días de incidente son los días en que hay que mirar la conciliación con más cuidado.

Un simulacro de verdad se hace con personas: se anuncia (o no), se provoca el fallo en el entorno de paper (cortar la red del contenedor con docker network disconnect, parar la base de datos, poner el disco al 100 %), se cronometra cuánto se tarda en detectar y en llegar a un estado seguro, y se anota lo que el procedimiento no previó. Lo que salga mal en el simulacro es lo que habría salido mal de verdad.

Ejercicios

1. Añade al verificador TOTP un límite de ritmo: como mucho 5 intentos por minuto. Comprueba que un atacante que prueba un código por segundo durante 10 minutos hace como mucho 50 intentos.

class VerificadorConRitmo(VerificadorTOTP):
    def __init__(self, clave, por_minuto=5, **kw):
        super().__init__(clave, **kw)
        self.por_minuto, self.intentos = por_minuto, []

    def verificar(self, codigo, ahora):
        self.intentos = [t for t in self.intentos if ahora - t < 60]
        if len(self.intentos) >= self.por_minuto:
            return "ritmo"
        self.intentos.append(ahora)
        return super().verificar(codigo, ahora)

v = VerificadorConRitmo(secrets.token_bytes(20))
r = [v.verificar(f"{i:06d}", T + i) for i in range(600)]
print({k: r.count(k) for k in sorted(set(r))})

{'fallo': 50, 'ritmo': 550}

2. Añade a la tabla de decisión la condición «posiciones abiertas» (sí o no) y comprueba cuántos huecos aparecen si no cambias las reglas. ¿Por qué no aparece ninguno?

DOMINIO["posiciones"] = [True, False]
TABLA4 = [((*p, X), e, a) for p, e, a in TABLA]
print(comprobar(TABLA4))
del DOMINIO["posiciones"]

(48, [], [])

3. El VPS del bot se borra por un error del proveedor un martes a las 17:00, hora de Madrid, con el mercado de Nueva York abierto. Las copias son diarias, a las 23:00, en almacenamiento de objetos del mismo proveedor y en casa. Escribe los pasos hasta volver a operar y di qué se pierde.