Security and continuity: 2FA, IP allowlists, backups and failure procedures.

What can stop a trading system

Lesson 4.4.5 protected the broker credentials in the code: where they are stored, how they are kept out of the logs, what permissions they have and how they are rotated. Lesson 4.5.5 made the order manager survive a process crash, and 4.2.5 added the kill switch. This lesson looks at the deployed system of lessons 7.2.1 and 7.2.2 from the outside: who can get into the broker account or the server, what is lost if the disk dies, and what exactly the system, and the person watching it, does when a piece it does not control fails.

In the regulated world this has a name, and in the EU framework it is set out in two regulations. Commission Delegated Regulation (EU) 2017/589 (RTS 6) requires investment firms engaged in algorithmic trading to have documented “business continuity arrangements”, with adverse scenarios (unavailability of systems, external suppliers or data centres, loss or alteration of data), alternative arrangements to manage outstanding orders and positions, and an annual review and test (Article 14)11Note 1 · Commission Delegated Regulation (EU) 2017/589, Article 14, “Business continuity arrangements”, paragraphs 2(b) and (g) and 4 (OJ L 87, 31.3.2017; Spanish edition, PDF at https://www.boe.es/doue/2017/087/L00417-00448.pdf; English text on EUR-Lex: https://eur-lex.europa.eu/legal-content/EN/TXT/?uri=CELEX:32017R0589).. Regulation (EU) 2022/2554 (DORA), applicable since January 17, 2025, asks financial entities for response and recovery policies (Article 11) and for backup policies with periodic restoration tests on systems segregated from the source system (Article 12)22Note 2 · Regulation (EU) 2022/2554 (DORA), Articles 11, “Response and recovery”, 12, “Backup policies and procedures, restoration and recovery procedures and methods”, and 64 (application from January 17, 2025) (OJ L 333, 27.12.2022; Spanish edition, PDF at https://www.boe.es/doue/2022/333/L00001-00079.pdf; English text on EUR-Lex: https://eur-lex.europa.eu/legal-content/EN/TXT/?uri=CELEX:32022R2554).. An individual trader is subject to neither; the lists they contain are still a good guide to what to think about.

Second factor: TOTP per RFC 6238

Almost every broker and exchange offers a second factor with an authenticator app. What that app does fits in ten lines. HOTP (RFC 4226) computes an HMAC-SHA-1 of a counter with a shared key, takes 4 bytes at a position decided by the last nibble of the result (the “dynamic truncation”) and keeps the last digits. TOTP (RFC 6238) uses as the counter the number of 30-second steps since January 1, 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                       # dynamic truncation
    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)

A cryptographic implementation is validated against the standard's vectors, not by eye. RFC 4226 publishes the first ten HOTP values with the ASCII key “12345678901234567890” (appendix D), and RFC 6238 six instants with SHA-1, SHA-256 and SHA-512 in 8-digit codes (appendix B)33Note 3 · RFC 4226, appendix D, and RFC 6238, appendices A and 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), appendix 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"),
}
# seeds of the reference code (appendix A): the hash's length
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 HOTP values:", ok_hotp)
print("RFC 6238, 18 TOTP values:", aciertos, "of 18")
print("SHA-256 with the 20-byte seed:",
      totp(SEMILLA, 59, 8, "sha256"), "instead of 46119246")

RFC 4226, 10 HOTP values: True
RFC 6238, 18 TOTP values: 18 of 18
SHA-256 with the 20-byte seed: 32247374 instead of 46119246

All 28 checks pass. The third line shows a known trap in RFC 6238: the text of appendix B says the seed is the same 20-byte string for all three modes, but the reference program in appendix A, which generates the table, uses 32- and 64-byte seeds for SHA-256 and SHA-512. Anyone who implements it following only the text thinks their SHA-256 is wrong. That is why you validate against all the vectors and read the reference code.

Generating the code is the easy half. The verifier, the server side, is where the security decisions are, and the RFC and NIST's SP 800-63B-4 guideline set them44Note 4 · RFC 6238, sections 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”) and 6; NIST SP 800-63B-4 (https://pages.nist.gov/800-63-4/sp800-63b.html): key of at least 112 bits, limit of 100 consecutive failures and “OTP authentication is not phishing-resistant”. Accessed on October 23, 2026.:

• Key: random, with at least 112 bits of security (NIST); the RFC recommends the length of the HMAC output, 160 bits with SHA-1. It is shared once, usually with a QR code that encodes an otpauth:// URI.
• Window: the RFC recommends accepting at most one step of delay for the network; many verifiers accept one on each side for the phone's clock drift.
• Single use: after a successful validation, the same code is not accepted again (“MUST NOT”).
• Attempt limit: NIST requires limiting consecutive failed attempts on an authenticator to no more than 100 before disabling it.
• Constant-time comparison: hmac.compare_digest, so the response time does not leak how many digits match.

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 "locked"
        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 "reused"
                self.ultimo_paso, self.fallos = p, 0
                return "ok"
        self.fallos += 1
        return "failed"

clave = secrets.token_bytes(20)              # 160 bits
v = VerificadorTOTP(clave)
T = 1_766_000_000                            # injected clock
c = totp(clave, T)
print("current code:      ", v.verificar(c, T))
print("same code, +10 s:  ", v.verificar(c, T + 10))
print("phone clock -40 s: ",
      v.verificar(totp(clave, T + 60 - 40), T + 60))
print("phone clock -70 s: ",
      v.verificar(totp(clave, T + 120 - 70), T + 120))
for _ in range(100):
    v.verificar("000000", T + 200)
print("after 100 failures:", v.verificar(totp(clave, T + 230),
                                         T + 230))

p = 1 - (1 - 3 / 10 ** 6) ** 100
print(f"guessing with 100 attempts and 3 valid codes: {p:.4%}")
b32 = base64.b32encode(SEMILLA).decode().rstrip("=")
print(f"otpauth://totp/Broker:demo?secret={b32}&issuer=Broker")

current code:       ok
same code, +10 s:   reused
phone clock -40 s:  ok
phone clock -70 s:  failed
after 100 failures: locked
guessing with 100 attempts and 3 valid codes: 0.0300%
otpauth://totp/Broker:demo?secret=GEZDGNBVGY3TQOJQGEZDGNBVGY3TQOJQ&issuer=Broker

The one-step window accepts a phone whose clock is 40 seconds behind and rejects one that is 70 behind. The code reused 10 seconds later is rejected, which is what stops someone who saw it go by (a video, a shoulder, a log) from using it. After 100 failures the factor is locked even for the correct code: unlocking it is an account recovery process, not a retry. The probability of guessing blindly with 100 attempts, with three valid codes at a time, is 0.03%. The last line is the Google Authenticator app's format: the key in Base32 without padding, with the issuer55Note 5 · Google Authenticator, “Key Uri Format” (https://github.com/google/google-authenticator/wiki/Key-Uri-Format).; the one in the cell is the RFC's test seed, never a real one.

IP allowlists

Lesson 4.4.5 restricted the exchange API key to the server's IP. The same principle protects the server: SSH, the Grafana panel and any admin endpoint only accept connections from a few addresses. The standard library ships ipaddress, which understands IPv4 and IPv6 networks and their prefixes. Two typical mistakes: trusting the X-Forwarded-For header, which the client writes, and putting a much broader network than intended on the list:

import ipaddress as ipa

LISTA = [ipa.ip_network(r) for r in (
    "203.0.113.10/32",          # deployment VPS
    "198.51.100.0/24",          # office
    "2001:db8:10::/48",         # home IPv6
)]
PROXY = ipa.ip_address("192.0.2.5")         # our own proxy

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),             # the VPS
    ("::ffff:203.0.113.10", None),      # the VPS via mapped IPv6
    ("192.0.2.66", "198.51.100.7"),     # forged header
    ("192.0.2.5", "198.51.100.7"),      # through our own proxy
    ("2001:db8:10:4::2", None),         # home, IPv6
    ("2001:db8:11::2", None),           # another IPv6 network
]
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} naive={ing!s:<5} "
          f"good={permitido(s, x)}")

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

The naive verifier, which takes the header if there is one, lets in the request with the forged header and rejects the VPS when it arrives as a mapped IPv6 address. The good one trusts the header only when the connection comes from our own proxy and normalizes mapped addresses. An allowlist also reviews itself: an entry with too many addresses is usually a typo (a /16 where a /32 should be), and a private network on the list of a public server makes no sense:

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), "the whole Internet"))
        elif r.version == 4 and r.num_addresses > max_dir:
            out.append((str(r), f"{r.num_addresses:,} addresses"))
        elif r.version == 6 and r.prefixlen < 48:
            out.append((str(r), f"/{r.prefixlen} in IPv6"))
        if any(r.version == p.version and r.subnet_of(p)
               for p in PRIVADAS):
            out.append((str(r), "private network"))
    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 addresses'), ('10.0.0.0/8', '16,777,216 addresses'), ('10.0.0.0/8', 'private network'), ('0.0.0.0/0', 'the whole Internet'), ('2001:db8::/32', '/32 in IPv6')]

An allowlist has an operational cost: if your home IP changes, you are locked out of your own server. That is why a second path is planned before it is needed: the provider's web console (which does not go through the server's SSH), a jump host with a fixed IP or your own VPN. The “I am locked out” procedure is also part of continuity.

Backups that can be restored

For a bot, much of the state can be rebuilt: executions, positions and cash are in the broker statement, which is the accounting truth (lesson 4.5.5). What cannot be rebuilt is your own: the state of the order manager with its idempotency keys, the recorded signals and decisions, the trained models, the configuration. Those are the backups that matter. The process has four parts:

flowchart TB
  E[Live state] --> C[Consistent copy]
  C --> M[SHA-256 manifest]
  M --> T[Three destinations]
  T --> R[Restore]

The first trap is the copy itself. The SQLite documentation warns that copying a database file while a transaction is half done can produce a copy with part of the old content and part of the new, corrupt66Note 6 · SQLite, “How To Corrupt An SQLite Database File”, section 1.2 (https://www.sqlite.org/howtocorrupt.html), and “VACUUM” (VACUUM INTO) (https://www.sqlite.org/lang_vacuum.html).. The online backup API, Connection.backup in Python, or VACUUM INTO produce a consistent copy even if the bot keeps writing. The cell creates the bot's state with the 398 orders of the level 4 system, copies it that way and packages everything with a manifest:

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)  # copy from scratch
    with sqlite3.connect(origen / "oms.db") as src, \
            sqlite3.connect(st / "oms.db") as dst:
        src.backup(dst)                     # consistent while online
    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), "files;", "fingerprint", huella[:16])
print(hacer_copia(VIVO, Path("otra.tar.gz"))[0] == huella)

3 files; fingerprint 12aec368f12698b2
True

The package is reproducible: timestamps set to zero, empty owners, fixed order and gzip without a timestamp. Two backups of the same state give the same fingerprint, so a different fingerprint means a different state, not a different moment. The package's fingerprint is stored separately (in the backup log); the manifest, inside, holds the fingerprint of each file.

The 3-2-1 rule popularized by CISA, the US cybersecurity agency, sums up where to keep them: three copies of every important file (the original and two more), on two types of media, one of them off-site77Note 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”.. For a bot on a VPS: the server's disk, a volume or object storage from the provider, and a copy at another provider or at home. A copy on the same disk protects against nothing that happens to the disk. The cell simulates the three copies and two accidents: the one at the second destination loses a bit in transit and the one at the third was cut off halfway.

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                         # one bit flipped
(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])  # truncated

def verificar(paquete, huella):
    datos = paquete.read_bytes()
    if sha256(datos) != huella:
        return "fingerprint mismatch"
    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} does not match"
    return f"ok ({len(manif)} files)"

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

local    ok (3 files)
objetos  fingerprint mismatch
fuera    fingerprint mismatch

The package fingerprint detects both accidents before opening it. The manifest adds the file-by-file check, useful when the package has been rebuilt or when verifying a copy someone unpacked. But matching hashes only say the backup is the one that was made, not that it works. The real test is to restore it somewhere else and use it:

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("restored:", r.execute(consulta).fetchall())
print("original:", o.execute(consulta).fetchall())
r.close()
o.close()

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

The restored database passes PRAGMA integrity_check and gives the same orders and quantities as the original. filter="data" in extractall stops a tampered package from writing outside the destination folder or creating special files. This test is automated and scheduled (lesson 4.5.4): every week, restore the latest backup from each destination into a temporary folder and check it, with an alert if it fails. DORA asks EU financial entities for exactly that: periodic restoration tests, carried out on systems segregated from the source system.

Two figures shape the policy. The recovery point objective (RPO) is the point in time you must be able to go back to after a failure; the recovery time objective (RTO), how long recovery can take before it causes damage (definitions from NIST SP 800-34)88Note 8 · NIST CSRC, glossary: “Recovery Time Objective” and “Recovery Point Objective”, sourced from NIST SP 800-34 Rev. 1 (https://csrc.nist.gov/glossary/term/recovery_point_objective).. With a daily backup after the close, the RPO is one day: the day's work is lost, and for a bot much of it can be rebuilt from the statement. The RTO depends on what has been rehearsed: if restoring means remembering commands at three in the morning, there is no RTO.

Failure procedures as decision tables

A procedure written in prose (“if the broker goes down, assess the situation and act prudently”) is useless at the moment it is needed. What works is a table: for each type of incident and system state, what to do. Written as data, a table can be checked with code: that it covers every combination and that no rule is shadowed by an earlier one. The system has four states, those of the lesson 4.2.5 kill switch plus an intermediate one:

stateDiagram-v2
  [*] --> RUNNING
  RUNNING --> DEGRADED
  DEGRADED --> RUNNING
  DEGRADED --> REDUCE_ONLY
  REDUCE_ONLY --> HALTED
  DEGRADED --> HALTED
  HALTED --> RUNNING

DEGRADED opens no new positions but keeps what is there; REDUCE_ONLY only accepts orders that reduce risk; HALTED cancels what it can and waits for a person. The return to RUNNING always goes through a reconciliation (lesson 4.5.5). The table uses three conditions: the incident, whether there are live orders and whether the market is open.

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 = "*"                                      # any value
TABLA = [  # (incident, live orders, market open) -> action
    (("credencial_expuesta", X, X), "HALTED",
     "revoke the key in the broker console; rotate (4.4.5)"),
    ((X, X, False), "DEGRADED",
     "log it; resolve and reconcile before the open"),
    (("broker_caido", True, True), "DEGRADED",
     "do not resend; wait 90 s; then, the broker's desk"),
    (("broker_caido", False, True), "DEGRADED",
     "wait for reconnection; reconcile on return"),
    (("datos_congelados", X, True), "REDUCE_ONLY",
     "cancel live limit orders; backup source"),
    (("datos_anomalos", X, True), "DEGRADED",
     "discard the data point; check against the secondary source"),
    (("red_caida", True, True), "DEGRADED",
     "alternative channel (phone) to cancel or confirm"),
    (("red_caida", False, True), "DEGRADED",
     "wait; on return, reconcile before trading"),
]

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, "cases;", len(huecos), "gaps:",
      sorted({h[0] for h in huecos}), "| shadowed rules:", tapadas)

24 cases; 2 gaps: ['disco_lleno'] | shadowed rules: []

Twenty-four combinations and two gaps: nobody wrote what to do with a full disk while the market is open. It is the kind of omission that the check finds and a read-through does not. The corrected version adds the rule, and to show the second check, someone mistakenly pastes a general rule before a specific one:

TABLA.append((("disco_lleno", X, True), "REDUCE_ONLY",
              "rotate logs and free up space; alert"))
print(comprobar(TABLA)[1:])
mala = TABLA[:2] + [(("broker_caido", X, X), "HALTED",
                     "halt everything")] + TABLA[2:]
print(comprobar(mala)[1:])
print(decidir(("broker_caido", True, True), mala))

([], [])
([], [3, 4])
('HALTED', 'halt everything')

With the new rule there are no gaps or shadowed rules left. In the bad table, the general rule at position 2 shadows the two specific broker rules (positions 3 and 4): faced with a broker outage with live orders, the system would halt everything instead of waiting and reconciling, which with orders in flight is exactly the dangerous thing. The check flags it before it happens.

A drill with real disconnections

The fictional level 4 system left its connection log with the broker in vivo4_conexion.csv: start and end of each session and, on seven occasions, a missed heartbeat, a disconnection and a reconnection. The drill runs those seven incidents through the table, with a clock that advances according to the log, and records what the system does at each moment. First, the orders that were in flight when the connection dropped:

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

The six disconnections of the live account (real in the data) start between 0.1 and 1.2 seconds after the last send of the closing orders (at 15:50 New York time), with between 2 and 10 orders in flight; the paper one happened on a day without orders. They last between 42 and 137 seconds, and the missed heartbeat always gives a 3-second warning before the drop. Now the drill: the system moves to DEGRADED on the missed heartbeat, looks up the table when the connection drops and, if it is not back after 90 seconds, escalates to the broker's desk; on reconnection, it reconciles and returns to RUNNING.

ESCALADO_S = 90

def simulacro(fila):
    t, log = 0.0, []
    log.append((t, "missed heartbeat", "DEGRADED"))
    t += fila.aviso_s
    estado, accion = decidir(("broker_caido", fila.en_el_aire > 0,
                              True))
    log.append((t, "disconnection: " + accion, estado))
    if fila.caida_s > ESCALADO_S:
        log.append((t + ESCALADO_S, "call the desk: confirm "
                    f"{fila.en_el_aire} orders by phone",
                    "DEGRADED"))
    t += fila.caida_s
    log.append((t, "reconnection: reconcile by key (4.5.5)",
                "RUNNING"))
    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("escalated to the desk:", len(esc), "of", len(INC),
      list(esc["fecha"]))

2025-09-23 paper
  +    0 s  DEGRADED   missed heartbeat
  +    3 s  DEGRADED   disconnection: wait for reconnection; reconcile on return
  +   45 s  RUNNING    reconnection: reconcile by key (4.5.5)
2025-10-21 real
  +    0 s  DEGRADED   missed heartbeat
  +    3 s  DEGRADED   disconnection: do not resend; wait 90 s; then, the broker's desk
  +   93 s  DEGRADED   call the desk: confirm 9 orders by phone
  +   95 s  RUNNING    reconnection: reconcile by key (4.5.5)
2025-11-03 real
  +    0 s  DEGRADED   missed heartbeat
  +    3 s  DEGRADED   disconnection: do not resend; wait 90 s; then, the broker's desk
  +   93 s  DEGRADED   call the desk: confirm 2 orders by phone
  +  140 s  RUNNING    reconnection: reconcile by key (4.5.5)
escalated to the desk: 4 of 7 ['2025-10-21', '2025-11-03', '2025-12-03', '2025-12-15']

The cell prints the timelines of the first incidents up to the longest one, on November 3, with 137 seconds of outage and 2 orders in flight. With the 90-second threshold, four of the seven incidents would have escalated to the broker's desk. The figure at the top orders them by date with the threshold marked. Choosing the threshold is a trade-off that the drill makes visible: lower, more calls for outages that resolve themselves; higher, less margin before the close, when a market-on-close order can no longer be canceled.

The drill closes with what the reconciliation said. Lesson 4.5.5 found thirteen anomalies in the live account; how many fell on disconnection days?

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), "of", len(anom), "anomalies on outage days;",
      "outage days without anomalies:",
      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 of 13 anomalies on outage days; outage days without anomalies: []

All six outages of the live account left a trace in the reconciliation: duplicate executions that the broker resent on reconnection, executions it never resent, an out-of-order message and the duplicate order R00114, which the manager resent during the November 13 outage without asking. Nine of the thirteen anomalies fall on those six days. The practical conclusion of the drill is twofold: the right procedure for a disconnection is not to act fast but to avoid making things worse (do not resend) and reconcile on return; and incident days are the days when the reconciliation needs the closest look.

A real drill is done with people: it is announced (or not), the failure is triggered in the paper environment (cut the container's network with docker network disconnect, stop the database, fill the disk to 100%), the time to detect and to reach a safe state is measured, and whatever the procedure did not foresee is written down. Whatever goes wrong in the drill is what would have gone wrong for real.

Exercises

1. Add a rate limit to the TOTP verifier: at most 5 attempts per minute. Check that an attacker who tries one code per second for 10 minutes makes at most 50 attempts.

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 "rate_limited"
        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))})

{'failed': 50, 'rate_limited': 550}

2. Add the condition “open positions” (yes or no) to the decision table and check how many gaps appear if you do not change the rules. Why does none appear?

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

(48, [], [])

3. The bot's VPS is wiped by a provider error on a Tuesday at 17:00 Madrid time, with the New York market open. Backups are daily, at 23:00, in the same provider's object storage and at home. Write the steps until you are trading again and say what is lost.