Professional logging and testing with pytest.

Why print is not enough

While you are writing a script, print is enough. When the script becomes a system that trades on its own, for weeks, and something goes wrong at 03:00, you need to know what it did, in what order and with what data. print does not say when something happened, does not distinguish a detail from a serious error, cannot be switched off piece by piece and always writes to the same place.

The standard library's logging module solves all four. Each message has a level:

The minimal configuration, basicConfig, sets the level, the destination and the format. Here the format has no time so that the lesson's output is reproducible; in production it does, as shown further down:

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("50-session average: %.2f", 141.87)
log.info("buy signal on %s at %.2f", "SINT", 142.22)
log.warning("slippage of %.1f bp (max %d)", 7.5, 5)
log.error("order rejected: %s", "insufficient margin")

INFO     estrategia.cruce | buy signal on SINT at 142.22
WARNING  estrategia.cruce | slippage of 7.5 bp (max 5)
ERROR    estrategia.cruce | order rejected: insufficient margin

The DEBUG message does not appear: the INFO level is a threshold and everything below it is discarded. To investigate a problem, it is enough to lower the threshold to DEBUG, without touching any other line.

The arguments go separately from the template ("at %.2f", 142.22), not inside an f-string. That way the text is only formatted if the message is emitted, and all messages of the same kind share a template, which makes it possible to count them and search for them in a log file.

A hierarchy of loggers

getLogger("estrategia.cruce") does not create a standalone logger: the dots in the name form a hierarchy, like the folders of a path. estrategia.cruce is a child of estrategia, which is a child of the root logger. A logger without its own level inherits its parent's, and its messages travel up to the handlers of its ancestors.

The convention is one logger per module, with logging.getLogger(__name__): the module name gives you the hierarchy for free. That way you can, for example, see the strategy in full detail and only the warnings of the broker connector, which tends to be very chatty:

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

ordenes = logging.getLogger("broker.ordenes")
ordenes.info("websocket heartbeat")  # INFO < WARNING: not emitted
ordenes.warning("slow broker response: %d ms", 850)
log.debug("50-session average: %.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 | slow broker response: 850 ms
DEBUG    estrategia.cruce | 50-session average: 141.87
broker.ordenes    WARNING
estrategia.cruce  DEBUG
datos             INFO

broker.ordenes has no level of its own and inherits WARNING from broker; estrategia.cruce inherits DEBUG from estrategia, so its debug message now does come out; datos has no ancestor with a level and inherits INFO from the root.

For errors with an exception, log.exception("message") inside an except records the message at ERROR level along with the full traceback. It is the way to keep the context of a failure without stopping the system.

Console and file: handlers

A handler decides where a message goes, with its own level and its own format. The usual setup of a trading system has two: the console, with warnings and errors for whoever is watching it, and a file with everything, to reconstruct later what happened.

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

For configurations with several pieces, logging.config.dictConfig takes a dictionary that can be stored in a YAML or a TOML file (lesson 0.4.4). The file uses RotatingFileHandler, which switches to a new file when a size is reached and keeps a few previous ones, so the log does not fill the disk. And the time is in UTC (lesson 0.4.4), with a formatter that uses 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("session started with %d symbols", 3)
log.debug("computed size: %d shares", 70)
log.warning("no ORO price for %d s", 45)

WARNING  estrategia.cruce | no ORO price for 45 s

Only the warning appears in the console. The file has all three messages, each with its UTC timestamp. So that the lesson's output does not change on every run, each line is printed without the time:

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

20 | INFO | estrategia.cruce | session started with 3 symbols
20 | DEBUG | estrategia.cruce | computed size: 70 shares
20 | WARNING | estrategia.cruce | no ORO price for 45 s

Each line starts with a 20-character timestamp (2026-10-21T09:30:00Z) followed by the level, the logger and the message, separated by |: a format that is easy to filter with grep or to load into pandas.

Tests with pytest

A test is code that checks that other code does what it should. With pytest, a test is a function whose name starts with test_ and that uses assert: if the condition is false, the test fails and pytest shows the values involved. Tests live in test_*.py files, separate from the code they test.

The function to be tested is the maximum drawdown: the largest drop from a previous peak, as a fraction. With prices $p_t$ and the running maximum $M_t = \max_{s \le t} p_s$:

The first version has a deliberate bug, of the kind that really gets made: it uses the maximum of the whole series instead of the maximum up to that day. In a real project, the module and its tests are written in the editor; here the cell creates them in the working folder:

Path("riesgo.py").write_text('''\
"""Risk metrics on a list of prices."""


def drawdown_maximo(precios: list[float]) -> float:
    """Largest drop from a previous peak (≤ 0)."""
    if not precios:
        raise ValueError("empty series")
    if any(p <= 0 for p in precios):
        raise ValueError("non-positive prices")
    pico = max(precios)
    return min(p / pico - 1 for p in precios)
''', encoding="utf-8");

The tests check a hand-computed example, the edge cases (rising series, a single price, empty series, invalid prices), a property that any drawdown must satisfy and a case with real data:

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():
    # peak 120, trough 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="empty"):
        drawdown_maximo([])


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


def test_el_futuro_no_cuenta():
    # when it traded at 90, the known peak was 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");

The pytest pieces that appear:

• pytest.approx compares floats with a tolerance. 90 / 120 - 1 does not necessarily give exactly -0.25 in floating point (lesson 0.3.1); a direct == would make the test fragile.
• pytest.raises checks that an exception is raised, and match that its message contains a given text.
• @pytest.mark.parametrize repeats a test with several values: test_precios_no_positivos counts as two tests.
• A fixture (@pytest.fixture) prepares data for the tests that request it by name as an argument: test_sint_caida_de_2022 receives the list of closes without knowing where it comes from.

The last test does not pin an exact number, but an interval: SINT has a drop of around 25% in 2022 (the data README says so). It is a sanity test: it does not prove the function is correct, but it fails if the result stops making sense.

Running the tests

From the terminal, in a project with pytest as a development dependency, all it takes is:

uv add --dev pytest
uv run pytest            # all test_*.py files in the project
uv run pytest -q tests/test_riesgo.py::test_serie_vacia

pytest finds the test_*.py files, runs each test separately, shows a dot for each test that passes and an F for each test that fails, with the details of each failure, and ends with a summary and an exit code: 0 if everything passed, 1 if something failed.

In this lesson, pytest is called from Python with pytest.main, the same run without leaving the notebook. The helper function keeps the lines of the failed tests, the last line of the summary without the time, and the exit code. It also removes already imported modules so that each run reads the new files; from the terminal this is not needed, because each pytest is a new process.

import contextlib
import io
import re

import pytest

sys.dont_write_bytecode = True  # no .pyc caches between versions

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("exit code:", 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
exit code: 1

Four tests fail, and all four for the same reason. In the hand-computed example, the maximum of the whole series is 130, which comes after the trough of 90, so the function measures 90 / 130 − 1 instead of 90 / 120 − 1. In the rising series, it compares the first price with the last and sees a "drop" that never happened. In test_el_futuro_no_cuenta, the final 200 turns a −10% into a −55%. And with real data, SINT's drop comes out at −36.9%, outside the reasonable interval: the series' global maximum came years after the low.

It is look-ahead bias (lesson 0.1.1) inside a risk metric: the function uses a price that, at the moment it measures, did not exist yet. A drawdown like this exaggerates the risk of any strategy that rose a lot after a drop. Without the rising-series test, nobody would have seen it.

Fixing and rerunning the tests

The correct version walks through the prices in order and updates the peak only with what has already been seen:

Path("riesgo.py").write_text('''\
"""Risk metrics on a list of prices."""


def drawdown_maximo(precios: list[float]) -> float:
    """Largest drop from a previous peak (≤ 0)."""
    if not precios:
        raise ValueError("empty series")
    if any(p <= 0 for p in precios):
        raise ValueError("non-positive prices")
    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
exit code: 0

Eight tests (seven functions, one of them with two parameters) and exit code 0. From here on, every change to riesgo.py is validated with a single command. That is the real value of tests: not so much finding today's bug as preventing it from coming back tomorrow, when someone "optimizes" the function.

In a project, the usual layout separates the code from the tests and keeps the pytest configuration in pyproject.toml (lesson 0.2.2):

mi-sistema/
├── pyproject.toml        # [tool.pytest.ini_options] testpaths = ["tests"]
├── src/mi_sistema/
│   └── riesgo.py
└── tests/
    ├── conftest.py       # fixtures shared by several files
    └── test_riesgo.py

What to test first in trading code, in order of damage avoided: money calculations (P&L, commissions, position sizes), risk metrics, the time alignment of signals and positions (the shift(1) from lesson 0.1.1) and configuration validation (lesson 0.4.2).

Exercises

1. Add to riesgo.py a function rentabilidad_total(precios) that returns precios[-1] / precios[0] - 1 and raises ValueError with fewer than two prices. Write two tests (one with pytest.approx and another with pytest.raises) in test_total.py and run them.

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("two prices are needed")
    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
exit code: 0

2. Configure a riesgo logger so that a function vigilar(dd, limite) logs a WARNING if the drawdown exceeds half the limit and a CRITICAL if it exceeds the whole limit. Test it with −0.05, −0.12 and −0.22 and a limit of −0.20. (The console in the previous configuration shows WARNING and above.)

log_riesgo = logging.getLogger("riesgo")

def vigilar(dd: float, limite: float) -> None:
    if dd <= limite:
        log_riesgo.critical("drawdown %.1f%%: limit breached, "
                            "stop the system", dd * 100)
    elif dd <= limite / 2:
        log_riesgo.warning("drawdown %.1f%%: half the limit",
                           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%: half the limit
CRITICAL riesgo | drawdown -22.0%: limit breached, stop the system

3. With parametrize, write a test that checks, for three different series, that the drawdown is never positive or less than −1. Why is it a property and not an example?

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
exit code: 0