Logging profesional y testing con pytest.

Por qué no basta con print

Mientras se escribe un script, print es suficiente. Cuando el script se convierte en un sistema que opera solo, durante semanas, y algo sale mal a las 03:00, hace falta saber qué hizo, en qué orden y con qué datos. print no dice cuándo pasó algo, no distingue un detalle de un error grave, no se puede apagar por partes y escribe siempre en el mismo sitio.

El módulo logging de la biblioteca estándar resuelve las cuatro cosas. Cada mensaje tiene un nivel:

La configuración mínima, basicConfig, fija el nivel, el destino y el formato. Aquí el formato no lleva hora para que la salida de la lección sea reproducible; en producción sí la lleva, como se ve más abajo:

import logging
import sys

logging.basicConfig(
    level=logging.INFO, stream=sys.stdout, force=True,
    format="%(levelname)-8s %(name)s | %(message)s")

log = logging.getLogger("estrategia.cruce")
log.debug("media de 50 sesiones: %.2f", 141.87)
log.info("señal de compra en %s a %.2f", "SINT", 142.22)
log.warning("deslizamiento de %.1f pb (máximo %d)", 7.5, 5)
log.error("orden rechazada: %s", "margen insuficiente")

INFO     estrategia.cruce | señal de compra en SINT a 142.22
WARNING  estrategia.cruce | deslizamiento de 7.5 pb (máximo 5)
ERROR    estrategia.cruce | orden rechazada: margen insuficiente

El mensaje DEBUG no aparece: el nivel INFO es un umbral y todo lo que esté por debajo se descarta. Para investigar un problema basta con bajar el umbral a DEBUG, sin tocar ninguna línea más.

Los argumentos van aparte de la plantilla ("a %.2f", 142.22), no dentro de una f-string. Así el texto solo se formatea si el mensaje se emite, y todos los mensajes del mismo tipo comparten plantilla, lo que permite contarlos y buscarlos en un fichero de registro.

Una jerarquía de loggers

getLogger("estrategia.cruce") no crea un logger suelto: los puntos del nombre forman una jerarquía, como las carpetas de una ruta. estrategia.cruce es hijo de estrategia, que es hijo del logger raíz. Un logger sin nivel propio hereda el de su padre, y sus mensajes suben hasta los handlers de sus antepasados.

La costumbre es un logger por módulo, con logging.getLogger(__name__): el nombre del módulo da la jerarquía gratis. Así se puede, por ejemplo, ver con todo detalle la estrategia y solo los avisos del conector del bróker, que suele ser muy hablador:

logging.getLogger("broker").setLevel(logging.WARNING)
logging.getLogger("estrategia").setLevel(logging.DEBUG)

ordenes = logging.getLogger("broker.ordenes")
ordenes.info("latido del websocket")  # INFO < WARNING: no sale
ordenes.warning("respuesta lenta del bróker: %d ms", 850)
log.debug("media de 50 sesiones: %.2f", 141.87)

for nombre in ("broker.ordenes", "estrategia.cruce", "datos"):
    nivel = logging.getLogger(nombre).getEffectiveLevel()
    print(f"{nombre:<17} {logging.getLevelName(nivel)}")

WARNING  broker.ordenes | respuesta lenta del bróker: 850 ms
DEBUG    estrategia.cruce | media de 50 sesiones: 141.87
broker.ordenes    WARNING
estrategia.cruce  DEBUG
datos             INFO

broker.ordenes no tiene nivel propio y hereda WARNING de broker; estrategia.cruce hereda DEBUG de estrategia, así que ahora su mensaje de depuración sí sale; datos no tiene antepasado con nivel y hereda el INFO del raíz.

Para los errores con excepción, log.exception("mensaje") dentro de un except registra el mensaje con nivel ERROR y el traceback completo. Es la forma de no perder el contexto de un fallo sin parar el sistema.

Consola y fichero: handlers

Un handler decide adónde va un mensaje, con su propio nivel y su propio formato. La configuración habitual de un sistema de trading tiene dos: la consola, con avisos y errores para quien lo vigila, y un fichero con todo, para reconstruir después lo que pasó.

flowchart LR
  L[logger] --> HC[consola WARNING]
  L --> HF[fichero DEBUG]
  HC --> T[Terminal]
  HF --> F[sistema.log]

Para configuraciones con varias piezas, logging.config.dictConfig recibe un diccionario que se puede guardar en un YAML o un TOML (lección 0.4.4). El fichero usa RotatingFileHandler, que cambia de fichero al llegar a un tamaño y guarda unos cuantos anteriores, para que el registro no llene el disco. Y la hora va en UTC (lección 0.4.4), con un formateador que usa time.gmtime:

import logging.config
import time
from pathlib import Path

class FormatoUTC(logging.Formatter):
    converter = time.gmtime

CONFIG = {
    "version": 1,
    "disable_existing_loggers": False,
    "formatters": {
        "consola": {"format": "%(levelname)-8s %(name)s | %(message)s"},
        "fichero": {
            "()": FormatoUTC,
            "fmt": ("%(asctime)s | %(levelname)s | %(name)s"
                    " | %(message)s"),
            "datefmt": "%Y-%m-%dT%H:%M:%SZ"},
    },
    "handlers": {
        "consola": {"class": "logging.StreamHandler",
                    "stream": "ext://sys.stdout",
                    "level": "WARNING", "formatter": "consola"},
        "fichero": {"class": "logging.handlers.RotatingFileHandler",
                    "filename": "sistema.log", "encoding": "utf-8",
                    "maxBytes": 5_000_000, "backupCount": 5,
                    "level": "DEBUG", "formatter": "fichero"},
    },
    "root": {"level": "DEBUG", "handlers": ["consola", "fichero"]},
}
logging.config.dictConfig(CONFIG)

log.info("sesión iniciada con %d símbolos", 3)
log.debug("tamaño calculado: %d títulos", 70)
log.warning("sin precio de ORO desde hace %d s", 45)

WARNING  estrategia.cruce | sin precio de ORO desde hace 45 s

En la consola solo aparece el aviso. El fichero tiene los tres mensajes, cada uno con su marca de tiempo en UTC. Para que la salida de la lección no cambie en cada ejecución, se imprime cada línea sin la hora:

for linea in Path("sistema.log").read_text("utf-8").splitlines():
    hora, resto = linea.split(" | ", 1)
    print(len(hora), "|", resto)

20 | INFO | estrategia.cruce | sesión iniciada con 3 símbolos
20 | DEBUG | estrategia.cruce | tamaño calculado: 70 títulos
20 | WARNING | estrategia.cruce | sin precio de ORO desde hace 45 s

Cada línea empieza con una marca de 20 caracteres (2026-10-21T09:30:00Z) y sigue con el nivel, el logger y el mensaje, separados por |: un formato fácil de filtrar con grep o de cargar en pandas.

Tests con pytest

Un test es código que comprueba que otro código hace lo que debe. Con pytest, un test es una función cuyo nombre empieza por test_ y que usa assert: si la condición es falsa, el test falla y pytest muestra los valores implicados. Los tests viven en ficheros test_*.py, separados del código que prueban.

La función que se va a probar es el drawdown máximo: la mayor caída desde un máximo previo, en tanto por uno. Con precios $p_t$ y el máximo acumulado $M_t = \max_{s \le t} p_s$:

La primera versión tiene un error a propósito, de los que se cometen de verdad: usa el máximo de toda la serie en lugar del máximo hasta ese día. En un proyecto real, el módulo y sus tests se escriben en el editor; aquí la celda los crea en la carpeta de trabajo:

Path("riesgo.py").write_text('''\
"""Métricas de riesgo sobre una lista de precios."""


def drawdown_maximo(precios: list[float]) -> float:
    """Mayor caída desde un máximo previo (≤ 0)."""
    if not precios:
        raise ValueError("serie vacía")
    if any(p <= 0 for p in precios):
        raise ValueError("precios no positivos")
    pico = max(precios)
    return min(p / pico - 1 for p in precios)
''', encoding="utf-8");

Los tests prueban un ejemplo calculado a mano, los casos límite (serie creciente, un solo precio, serie vacía, precios no válidos), una propiedad que cualquier drawdown debe cumplir y un caso con datos reales:

Path("test_riesgo.py").write_text('''\
import csv

import pytest

from riesgo import drawdown_maximo


@pytest.fixture
def cierres_sint() -> list[float]:
    with open("sint_diario.csv", newline="", encoding="utf-8") as f:
        return [float(fila["cierre"]) for fila in csv.DictReader(f)]


def test_ejemplo_a_mano():
    # pico 120, valle 90: -25 %
    resultado = drawdown_maximo([100, 120, 90, 130, 117])
    assert resultado == pytest.approx(-0.25)


def test_serie_creciente_sin_drawdown():
    assert drawdown_maximo([100, 101, 105, 110]) == 0


def test_un_solo_precio():
    assert drawdown_maximo([142.22]) == 0


def test_serie_vacia():
    with pytest.raises(ValueError, match="vacía"):
        drawdown_maximo([])


@pytest.mark.parametrize("malo", [0, -1.5])
def test_precios_no_positivos(malo):
    with pytest.raises(ValueError, match="no positivos"):
        drawdown_maximo([100, malo, 90])


def test_el_futuro_no_cuenta():
    # cuando cotizaba a 90, el pico conocido era 100
    assert drawdown_maximo([100, 90, 200]) == pytest.approx(-0.10)


def test_sint_caida_de_2022(cierres_sint):
    assert -0.35 < drawdown_maximo(cierres_sint) < -0.15
''', encoding="utf-8");

Las piezas de pytest que aparecen:

• pytest.approx compara floats con tolerancia. 90 / 120 - 1 no tiene por qué dar exactamente -0.25 en coma flotante (lección 0.3.1); un == directo haría el test frágil.
• pytest.raises comprueba que se lanza una excepción, y match que su mensaje contiene un texto.
• @pytest.mark.parametrize repite un test con varios valores: test_precios_no_positivos cuenta como dos tests.
• Una fixture (@pytest.fixture) prepara datos para los tests que la piden por nombre como argumento: test_sint_caida_de_2022 recibe la lista de cierres sin saber de dónde sale.

El último test no fija un número exacto, sino un intervalo: SINT tiene una caída en torno al 25 % en 2022 (lo dice el README de los datos). Es un test de cordura: no demuestra que la función sea correcta, pero falla si el resultado deja de tener sentido.

Ejecutar los tests

Desde la terminal, en un proyecto con pytest como dependencia de desarrollo, basta con:

uv add --dev pytest
uv run pytest            # todos los test_*.py del proyecto
uv run pytest -q tests/test_riesgo.py::test_serie_vacia

pytest busca los ficheros test_*.py, ejecuta cada test por separado, muestra un punto por test que pasa y una F por test que falla, con el detalle de cada fallo, y termina con un resumen y un código de salida: 0 si todo ha pasado, 1 si algo ha fallado.

En esta lección, pytest se llama desde Python con pytest.main, la misma ejecución sin salir del cuaderno. La función auxiliar se queda con las líneas de los tests fallidos, con la última línea del resumen sin el tiempo y con el código de salida. Además borra los módulos ya importados para que cada ejecución lea los ficheros nuevos; desde la terminal no hace falta, porque cada pytest es un proceso nuevo.

import contextlib
import io
import re

import pytest

sys.dont_write_bytecode = True  # sin cachés de .pyc entre versiones

def pasar_tests(*args: str) -> int:
    for modulo in ("riesgo", "test_riesgo"):
        sys.modules.pop(modulo, None)
    salida = io.StringIO()
    with contextlib.redirect_stdout(salida):
        codigo = pytest.main(["-q", "-p", "no:cacheprovider",
                              "-rf", "--tb=no", *args])
    lineas = salida.getvalue().strip().splitlines()
    for linea in lineas:
        if linea.startswith("FAILED"):
            print(linea)
    print(re.sub(r" in [\d.]+s$", "", lineas[-1]))
    print("código de salida:", int(codigo))
    return int(codigo)

pasar_tests("test_riesgo.py");

FAILED test_riesgo.py::test_ejemplo_a_mano - assert -0.3076923076923077 == -0.25 ± 2.5e-07
FAILED test_riesgo.py::test_serie_creciente_sin_drawdown - assert -0.09090909090909094 == 0
FAILED test_riesgo.py::test_el_futuro_no_cuenta - assert -0.55 == -0.1 ± 1.0e-07
FAILED test_riesgo.py::test_sint_caida_de_2022 - assert -0.35 < -0.36924001602778156
4 failed, 4 passed
código de salida: 1

Cuatro tests fallan, y los cuatro por el mismo motivo. En el ejemplo a mano, el máximo de toda la serie es 130, que llega después del valle de 90, así que la función mide 90 / 130 − 1 en lugar de 90 / 120 − 1. En la serie creciente, compara el primer precio con el último y ve una «caída» que nunca ocurrió. En test_el_futuro_no_cuenta, el 200 final convierte un −10 % en un −55 %. Y con datos reales, la caída de SINT sale del −36,9 %, fuera del intervalo razonable: el máximo global de la serie llegó años después del mínimo.

Es un sesgo de anticipación (lección 0.1.1) dentro de una métrica de riesgo: la función usa un precio que, en el momento que mide, todavía no existía. Un drawdown así exagera el riesgo de cualquier estrategia que haya subido mucho después de una caída. Sin el test de la serie creciente, nadie lo habría visto.

Corregir y volver a pasar los tests

La versión correcta recorre los precios en orden y actualiza el pico solo con lo ya visto:

Path("riesgo.py").write_text('''\
"""Métricas de riesgo sobre una lista de precios."""


def drawdown_maximo(precios: list[float]) -> float:
    """Mayor caída desde un máximo previo (≤ 0)."""
    if not precios:
        raise ValueError("serie vacía")
    if any(p <= 0 for p in precios):
        raise ValueError("precios no positivos")
    pico, peor = precios[0], 0.0
    for p in precios:
        pico = max(pico, p)
        peor = min(peor, p / pico - 1)
    return peor
''', encoding="utf-8")

pasar_tests("test_riesgo.py");

8 passed
código de salida: 0

Ocho tests (siete funciones, una de ellas con dos parámetros) y código de salida 0. A partir de aquí, cada cambio en riesgo.py se valida con un solo comando. Ese es el valor real de los tests: no tanto encontrar el error de hoy como impedir que vuelva mañana, cuando alguien «optimice» la función.

En un proyecto, la organización habitual separa el código de los tests y deja la configuración de pytest en pyproject.toml (lección 0.2.2):

mi-sistema/
├── pyproject.toml        # [tool.pytest.ini_options] testpaths = ["tests"]
├── src/mi_sistema/
│   └── riesgo.py
└── tests/
    ├── conftest.py       # fixtures compartidas por varios ficheros
    └── test_riesgo.py

Qué conviene probar primero en código de trading, por orden de daño evitado: los cálculos de dinero (P&L, comisiones, tamaños de posición), las métricas de riesgo, la alineación temporal de señales y posiciones (el shift(1) de la lección 0.1.1) y la validación de configuraciones (lección 0.4.2).

Ejercicios

1. Añade a riesgo.py una función rentabilidad_total(precios) que devuelva precios[-1] / precios[0] - 1 y lance ValueError con menos de dos precios. Escribe dos tests (uno con pytest.approx y otro con pytest.raises) en test_total.py y pásalos.

with open("riesgo.py", "a", encoding="utf-8") as f:
    f.write('''

def rentabilidad_total(precios: list[float]) -> float:
    if len(precios) < 2:
        raise ValueError("hacen falta dos precios")
    return precios[-1] / precios[0] - 1
''')

Path("test_total.py").write_text('''\
import pytest

from riesgo import rentabilidad_total


def test_total():
    assert rentabilidad_total([100, 90, 110]) == pytest.approx(0.10)


def test_un_precio():
    with pytest.raises(ValueError):
        rentabilidad_total([100])
''', encoding="utf-8")

sys.modules.pop("test_total", None)
pasar_tests("test_total.py");

2 passed
código de salida: 0

2. Configura un logger riesgo que, al llamar a una función vigilar(dd, limite), registre un WARNING si el drawdown supera la mitad del límite y un CRITICAL si lo supera entero. Prueba con −0,05, −0,12 y −0,22 y un límite de −0,20. (La consola de la configuración anterior muestra WARNING y superiores.)

log_riesgo = logging.getLogger("riesgo")

def vigilar(dd: float, limite: float) -> None:
    if dd <= limite:
        log_riesgo.critical("drawdown %.1f%%: límite superado, "
                            "parar el sistema", dd * 100)
    elif dd <= limite / 2:
        log_riesgo.warning("drawdown %.1f%%: mitad del límite",
                           dd * 100)
    else:
        log_riesgo.info("drawdown %.1f%%", dd * 100)

for dd in (-0.05, -0.12, -0.22):
    vigilar(dd, limite=-0.20)

WARNING  riesgo | drawdown -12.0%: mitad del límite
CRITICAL riesgo | drawdown -22.0%: límite superado, parar el sistema

3. Con parametrize, escribe un test que compruebe para tres series distintas que el drawdown nunca es positivo ni menor que −1. ¿Por qué es una propiedad y no un ejemplo?

Path("test_propiedad.py").write_text('''\
import pytest

from riesgo import drawdown_maximo


@pytest.mark.parametrize("serie", [
    [100, 50, 25, 12.5],
    [1, 2, 3],
    [142.22, 139.98, 145.10, 101.30, 150.00],
])
def test_entre_menos_uno_y_cero(serie):
    assert -1 < drawdown_maximo(serie) <= 0
''', encoding="utf-8")

sys.modules.pop("test_propiedad", None)
pasar_tests("test_propiedad.py");

3 passed
código de salida: 0