Concurrencia: threading, multiprocessing, asyncio y free-threading.

Esperar o calcular

Un programa puede ir lento por dos motivos muy distintos:

• Espera (limitado por E/S, entrada/salida): pedir precios a una API, recibir mensajes de un websocket, leer ficheros, escribir en una base de datos. El procesador está casi parado; el tiempo se va en esperar respuestas.
• Calcula (limitado por CPU): simular miles de trayectorias, optimizar parámetros, recalcular indicadores sobre años de datos. El procesador está al 100 %.

La concurrencia consiste en tener varias tareas en marcha a la vez; el paralelismo, en ejecutarlas literalmente al mismo tiempo en varios núcleos. Para esperar basta con concurrencia: mientras una petición espera, se lanza otra. Para calcular hace falta paralelismo. Python ofrece tres herramientas, y una cuarta que cambia las reglas:

Antes de todo esto hay una opción que no aparece en la tabla: vectorizar. Un cálculo que NumPy o Polars hacen sobre arrays enteros (lecciones 0.5.1 y 0.5.4) corre en C, suele ser decenas de veces más rápido que un bucle de Python y a menudo hace innecesario el paralelismo.

Todos los ejemplos de esta lección son deterministas: esperas simuladas con sleep, simulaciones con semilla y resultados que se recogen en el orden pedido. El orden en que terminan las tareas, en cambio, puede cambiar de una ejecución a otra, y ningún resultado debe depender de él.

Hilos para esperar a la red

La función descargar simula la petición del último cierre de un activo: espera un cuarto de segundo, como una llamada HTTP, y devuelve los cierres del 31 de diciembre de 2025 de cartera_sint.csv. Cuatro peticiones seguidas tardan cuatro cuartos de segundo:

import time

CIERRES = {"RVAR": 164.35, "RFIJ": 102.29, "ORO": 171.44,
           "TECH": 77.82}

def descargar(simbolo: str) -> tuple[str, float]:
    """Simula una petición HTTP: 0,25 s de espera de red."""
    time.sleep(0.25)
    return simbolo, CIERRES[simbolo]

simbolos = list(CIERRES)
t0 = time.perf_counter()
secuencial = [descargar(s) for s in simbolos]
t_secuencial = time.perf_counter() - t0
print(secuencial)

[('RVAR', 164.35), ('RFIJ', 102.29), ('ORO', 171.44), ('TECH', 77.82)]

Mientras descargar espera, el programa no hace nada. Con hilos, las cuatro esperas se solapan. concurrent.futures.ThreadPoolExecutor mantiene un grupo de hilos y les reparte las tareas; su método map funciona como el map normal y devuelve los resultados en el orden de la entrada, aunque las tareas terminen en otro:

from concurrent.futures import ThreadPoolExecutor

t0 = time.perf_counter()
with ThreadPoolExecutor(max_workers=4) as pool:
    en_hilos = list(pool.map(descargar, simbolos))
t_hilos = time.perf_counter() - t0

print(en_hilos == secuencial)
print("al menos el doble de rápido:", t_hilos < t_secuencial / 2)

True
al menos el doble de rápido: True

Los resultados son idénticos y el tiempo total se acerca al de una sola descarga. La lección no imprime los tiempos porque cambian en cada ejecución; la comparación, con un margen amplio, no. El with cierra el grupo de hilos al salir y espera a que terminen todas las tareas (lección 0.4.3).

Con una API real, el número de hilos lo limita el proveedor, no el ordenador: casi todas las APIs de datos y de brókeres fijan un máximo de peticiones por segundo, y superarlo lleva a errores 429 o a un bloqueo temporal.

Errores y orden

submit lanza una tarea y devuelve un futuro, un objeto que tendrá el resultado cuando la tarea termine. futuro.result() lo entrega o, si la tarea lanzó una excepción, la vuelve a lanzar en el hilo que pregunta. Recorrer los futuros en el orden en que se lanzaron mantiene la salida estable y permite tratar cada error por separado:

with ThreadPoolExecutor(max_workers=4) as pool:
    futuros = {s: pool.submit(descargar, s)
               for s in ["ORO", "XYZ", "TECH"]}

for simbolo, futuro in futuros.items():
    try:
        print(simbolo, futuro.result()[1])
    except KeyError as e:
        print(simbolo, "error:", repr(e))

ORO 171.44
XYZ error: KeyError('XYZ')
TECH 77.82

Un fallo no tumba las demás descargas. concurrent.futures.as_completed(futuros) entregaría los futuros según terminan, útil para mostrar progreso, pero en un orden que puede cambiar en cada ejecución.

Estado compartido: Lock

Los hilos comparten la memoria del proceso. Si varios modifican la misma variable a la vez, una actualización puede pisar a otra: exposicion += importe lee, suma y escribe en pasos separados, y otro hilo puede colarse entre ellos. El error es intermitente, depende de la carga de la máquina y casi nunca sale en las pruebas. La protección es un threading.Lock: solo un hilo a la vez ejecuta el bloque with cerrojo:

import threading

exposicion = 0.0
cerrojo = threading.Lock()

def registrar_ejecucion(importe: float) -> None:
    global exposicion
    with cerrojo:
        exposicion += importe

importes = [7_782.0, 17_144.0, -3_500.0, 10_229.0] * 2_500
with ThreadPoolExecutor(max_workers=8) as pool:
    list(pool.map(registrar_ejecucion, importes))
print(f"{exposicion:,.2f}", f"{sum(importes):,.2f}")

79,137,500.00 79,137,500.00

Con el cerrojo, el total coincide siempre con la suma directa. El diseño más seguro, de todos modos, es no compartir: que cada tarea devuelva su resultado y que el hilo principal los sume al final, como en las descargas.

Procesos para calcular

Los hilos no aceleran el cálculo en Python con GIL (el Global Interpreter Lock): un cerrojo dentro del intérprete que deja ejecutar código Python a un solo hilo a la vez. Los hilos lo sueltan mientras esperan (por eso funcionan con la red), pero no mientras calculan. Para usar varios núcleos con código Python puro, la solución clásica son varios procesos, cada uno con su intérprete, su GIL y su memoria.

El cálculo de ejemplo es una simulación de Monte Carlo: para cada semilla, 2.000 años simulados de 252 sesiones con rentabilidades normales, y el peor drawdown encontrado. Es Python puro, sin NumPy, a propósito:

import random

def peor_drawdown(semilla: int, anios: int = 2_000) -> float:
    """Peor drawdown en `anios` años simulados de 252 sesiones."""
    rng = random.Random(semilla)
    peor = 0.0
    for _ in range(anios):
        valor = pico = 1.0
        for _ in range(252):
            valor *= 1 + rng.gauss(0.0004, 0.012)
            pico = max(pico, valor)
            peor = min(peor, valor / pico - 1)
    return peor

semillas = [1, 2, 3, 4]
peores = [peor_drawdown(s) for s in semillas]
print([f"{p:.2%}" for p in peores])

['-49.04%', '-44.49%', '-49.17%', '-48.19%']

Con una volatilidad diaria del 1,2 % (un 19 % anual), el peor de 2.000 años simulados pierde entre el 44 % y el 49 % desde su máximo según la semilla: los años malos de verdad son raros, pero existen. Cada semilla tarda una fracción de segundo; con cientos de semillas, el reparto entre núcleos empieza a importar.

Con multiprocessing, un Pool de procesos reparte las semillas y map devuelve los resultados en orden. Aquí entra una novedad de Python 3.14: en Linux, el método de arranque por defecto pasó de fork a forkserver.11Nota 1 · Documentación de multiprocessing en Python 3.14: «On POSIX platforms the default start method was changed from fork to forkserver to retain the performance but avoid common multithreaded process incompatibilities». En macOS y Windows sigue siendo spawn.

• fork copia el proceso padre tal cual, con todas sus funciones y datos. Es rápido, pero copiar un proceso que tiene hilos en marcha puede dejar el hijo bloqueado, y por eso dejó de ser el valor por defecto.
• forkserver arranca al principio un proceso servidor limpio, de un solo hilo, y crea los hijos a partir de él. Los hijos no heredan las funciones definidas en el programa: las reciben por nombre y las importan, así que tienen que estar en un módulo importable.
• spawn arranca un intérprete nuevo para cada hijo. Es el valor por defecto en Windows y macOS.

En un notebook, las funciones definidas en una celda no están en ningún fichero importable, así que con forkserver o spawn el pool falla. La salida en un cuaderno es pedir fork explícitamente con un contexto:

import multiprocessing as mp

print("por defecto:", mp.get_start_method())
ctx = mp.get_context("fork")
with ctx.Pool(processes=4) as pool:
    en_procesos = pool.map(peor_drawdown, semillas)
print(en_procesos == peores)

por defecto: forkserver
True

Mismos resultados, porque cada simulación depende solo de su semilla, no del proceso que la ejecuta ni del momento en que termina. Sin semillas por tarea, cada proceso generaría números distintos en cada ejecución.

En un script, la forma recomendada en cualquier plataforma es la otra: las funciones en el nivel superior del fichero y el arranque bajo if __name__ == "__main__":, para que los hijos puedan importar el fichero sin volver a lanzar el pool. Con ProcessPoolExecutor, la interfaz es la misma que con hilos:

from concurrent.futures import ProcessPoolExecutor

from montecarlo import peor_drawdown  # la función, en su módulo


def main() -> None:
    semillas = range(1, 33)
    with ProcessPoolExecutor() as pool:  # un proceso por núcleo
        peores = list(pool.map(peor_drawdown, semillas))
    print(f"peor de {len(peores)} simulaciones: {min(peores):.2%}")


if __name__ == "__main__":
    main()

Los procesos tienen costes que los hilos no tienen: arrancar cada uno lleva su tiempo, no comparten memoria y todo lo que entra o sale de ellos se serializa con pickle. Compensan cuando cada tarea calcula bastante y recibe y devuelve poco, como una simulación que devuelve un número. Repartir entre procesos un DataFrame de millones de filas para hacer una suma suele ir más lento que hacerla en uno.

asyncio: muchas esperas en un solo hilo

asyncio resuelve la espera de otra forma. En lugar de varios hilos, un solo hilo ejecuta un bucle de eventos que alterna entre corrutinas: funciones definidas con async def que, al llegar a un await, ceden el control mientras esperan. Es la base de los clientes de websockets y de las APIs asíncronas de casi todos los exchanges, y de los sistemas que escuchan muchos flujos de datos a la vez (nivel 3).

Tres exchanges ficticios, con latencias distintas, devuelven la mejor compra (bid) y la mejor venta (ask) de BTC. asyncio.gather lanza las tres consultas a la vez y devuelve los resultados en el orden de los argumentos:

import asyncio

LIBROS = {  # exchange: (latencia en s, bid, ask), ficticios
    "alfa": (0.30, 67_412.5, 67_415.0),
    "beta": (0.10, 67_418.0, 67_421.5),
    "gamma": (0.20, 67_409.0, 67_410.5),
}

async def mejor_precio(exchange: str) -> tuple[str, float, float]:
    latencia, bid, ask = LIBROS[exchange]
    await asyncio.sleep(latencia)  # la petición de red
    return exchange, bid, ask

async def consultar() -> list[tuple[str, float, float]]:
    return await asyncio.gather(*(mejor_precio(e) for e in LIBROS))

cotizaciones = asyncio.run(consultar())
for c in cotizaciones:
    print(c)
compra = min(cotizaciones, key=lambda c: c[2])
venta = max(cotizaciones, key=lambda c: c[1])
print(f"comprar en {compra[0]} a {compra[2]:,.1f}, "
      f"vender en {venta[0]} a {venta[1]:,.1f}")

('alfa', 67412.5, 67415.0)
('beta', 67418.0, 67421.5)
('gamma', 67409.0, 67410.5)
comprar en gamma a 67,410.5, vender en beta a 67,418.0

beta responde primero y alfa el último, pero gather devuelve los resultados en el orden pedido. El total tarda lo que la consulta más lenta, no la suma de las tres. Comprar en gamma a 67.410,5 y vender en beta a 67.418 deja 7,5 dólares por BTC antes de comisiones, que en la práctica se las llevarían enteras: es el tipo de cálculo que hace un arbitrajista, y la razón de que la latencia importe.

asyncio.run crea el bucle, ejecuta la corrutina y lo cierra: es la entrada de un programa asíncrono. Jupyter y Marimo ya tienen un bucle en marcha, y en ellos se escribe directamente await consultar() en la celda.

Un exchange lento no debe retrasar a los demás. asyncio.timeout pone un tiempo límite a una espera y lanza TimeoutError si se supera:

LIBROS["delta"] = (2.0, 67_420.0, 67_423.0)  # muy lento

async def con_limite(exchange: str, limite: float = 0.5):
    try:
        async with asyncio.timeout(limite):
            return await mejor_precio(exchange)
    except TimeoutError:
        return exchange, None, None

async def consultar_todos():
    return await asyncio.gather(*(con_limite(e) for e in LIBROS))

for c in asyncio.run(consultar_todos()):
    print(c)

('alfa', 67412.5, 67415.0)
('beta', 67418.0, 67421.5)
('gamma', 67409.0, 67410.5)
('delta', None, None)

delta se descarta a los 0,5 s y el resto llega a tiempo. Para lanzar tareas que deben cancelarse juntas si una falla, Python 3.11+ ofrece asyncio.TaskGroup, la forma estructurada de hacer lo mismo que gather.

El GIL y el Python sin GIL

El GIL simplificó durante décadas el intérprete y las extensiones en C, a cambio de que los hilos no pudieran calcular en paralelo. El PEP 703 propuso hacerlo opcional y se aceptó con la condición de un despliegue gradual y reversible.22Nota 2 · PEP 703, estado Final. El Steering Council lo aceptó «with clear proviso: that the rollout be gradual and break as little as possible». Python 3.13 incluyó una compilación experimental sin GIL, y en Python 3.14 esa compilación pasó a estar soportada oficialmente (PEP 779), aunque sigue siendo opcional: el Python que se instala por defecto lleva GIL.33Nota 3 · PEP 779, estado Final, aceptado el 16 de junio de 2025; «What's New In Python 3.14»: «Free-threaded Python is officially supported» y «The performance penalty on single-threaded code in free-threaded mode is now roughly 5-10%» (https://docs.python.org/3.14/whatsnew/3.14.html). Hacerla la opción por defecto queda para un PEP futuro.

La compilación sin GIL se llama free-threading y su ejecutable lleva una t (python3.14t). Con ella, los hilos ejecutan código Python en paralelo, y una simulación como la de la lección escala con hilos, sin los costes de los procesos. Tiene contrapartidas: según las notas de la versión, el código de un solo hilo va entre un 5 % y un 10 % más lento, y una extensión en C que no declare compatibilidad vuelve a activar el GIL al importarse, con un aviso.

Desde Python se comprueba qué compilación se está usando y si el GIL está activo:

import sys
import sysconfig

print(sys.version_info[:3])
print("compilación free-threading:",
      bool(sysconfig.get_config_var("Py_GIL_DISABLED")))
print("GIL activo:", sys._is_gil_enabled())

(3, 14, 8)
compilación free-threading: False
GIL activo: True

El curso usa la compilación normal, con GIL. Para probar la otra, uv instala las dos variantes:44Nota 4 · Documentación de uv, «Python versions»: uv instala variantes free-threaded desde CPython 3.13; en 3.14, uv python install 3.14 sigue prefiriendo la compilación con GIL (https://docs.astral.sh/uv/concepts/python-versions/).

uv python install 3.14t
uv run --python 3.14t python simular_hilos.py
PYTHON_GIL=1 uv run --python 3.14t python simular_hilos.py  # GIL forzado

Con free-threading, el código que comparte estado entre hilos necesita más que nunca los Lock: sin GIL, las condiciones de carrera que antes eran raras se vuelven frecuentes.

Qué usar en cada caso

El árbol de decisión, de arriba abajo:

flowchart TB
  Q{¿Espera o calcula?} -->|espera| N{¿Muchas a la vez?}
  N -->|pocas| T[Hilos]
  N -->|cientos| A[asyncio]
  Q -->|calcula| V{¿Se vectoriza?}
  V -->|sí| NP[NumPy o Polars]
  V -->|no| P[Procesos]
  P -.->|3.14t| H[Hilos sin GIL]

Para pedir precios de unas decenas de símbolos a una API REST, hilos; para mantener abiertos cientos de websockets, asyncio; para un cálculo que se puede expresar con arrays, NumPy o Polars antes que cualquier otra cosa; para un cálculo en Python puro que no se vectoriza (simulaciones con lógica de trading barra a barra, optimizaciones), procesos, o hilos en una compilación sin GIL. Y en todos los casos, resultados que no dependan del orden en que terminan las tareas.

Ejercicios

1. Repite las descargas con max_workers=2. ¿Cambian los resultados? ¿Qué esperas que pase con el tiempo total?

t0 = time.perf_counter()
with ThreadPoolExecutor(max_workers=2) as pool:
    dos_hilos = list(pool.map(descargar, simbolos))
t_dos = time.perf_counter() - t0
print(dos_hilos == secuencial)
print("más lento que con 4 hilos:", t_dos > t_hilos)

True
más lento que con 4 hilos: True

2. Un exchange admite como máximo dos peticiones simultáneas. Con asyncio.Semaphore(2), consulta alfa, beta y gamma sin pasar de dos a la vez y comprueba que el resultado es el mismo que antes.

async def consultar_con_limite(maximo: int = 2):
    semaforo = asyncio.Semaphore(maximo)

    async def una(exchange: str):
        async with semaforo:  # espera turno si ya hay `maximo`
            return await mejor_precio(exchange)

    nombres = ["alfa", "beta", "gamma"]
    return await asyncio.gather(*(una(e) for e in nombres))

print(asyncio.run(consultar_con_limite()) == cotizaciones)

True

3. Con el pool de procesos, lanza peor_drawdown para las semillas 1 a 4 con 500 años en lugar de 2.000 (usa pool.starmap con pares (semilla, anios)). ¿Por qué el peor drawdown es menos extremo?

with ctx.Pool(processes=4) as pool:
    cortos = pool.starmap(peor_drawdown, [(s, 500) for s in semillas])
print([f"{p:.2%}" for p in cortos])

['-49.04%', '-41.01%', '-43.26%', '-38.45%']