Iteradores, generadores, decoradores y context managers.

Qué hace for por debajo

Un bucle for sobre una lista de cierres parece una operación indivisible. No lo es: Python pide a la lista un iterador con iter() y le pide valores con next() hasta que el iterador lanza StopIteration. Hacerlo a mano muestra el mecanismo:

cierres = [99.14, 99.25, 101.02]
it = iter(cierres)
print(next(it), next(it), next(it))
try:
    next(it)
except StopIteration:
    print("StopIteration: no quedan cierres")

99.14 99.25 101.02
StopIteration: no quedan cierres

Un iterable es cualquier objeto que sabe dar un iterador (listas, tuplas, diccionarios, ficheros, cadenas). Un iterador es el objeto que recorre: guarda en qué punto va y entrega el siguiente valor. La distinción importa porque un iterador se gasta: una vez entregado el último valor, no vuelve a empezar.

Cualquier clase con los métodos __iter__ y __next__ es un iterador. Escribirla a mano es tedioso; los generadores hacen lo mismo con mucho menos código.

Generadores: barras bajo demanda

Una función que contiene yield es un generador. Al llamarla no se ejecuta: devuelve un objeto generador. Cada next() ejecuta el cuerpo hasta el siguiente yield, entrega el valor y deja la función congelada en ese punto, con sus variables intactas, hasta la próxima petición. La ejecución perezosa se ve imprimiendo dentro:

def feed_simulado():
    print("  conectando al feed")
    yield 142.10
    print("  segundo tick")
    yield 142.15
    print("  feed cerrado")

feed = feed_simulado()
print("generador creado")
print(next(feed))
print(next(feed))
print(list(feed))

generador creado
  conectando al feed
142.1
  segundo tick
142.15
  feed cerrado
[]

«Conectando» no aparece al crear el generador, sino con el primer next(). El tercer recorrido (list(feed)) ejecuta el resto del cuerpo, no encuentra más yield y termina con una lista vacía.

Esta forma de trabajar encaja con los datos de mercado. Un fichero de barras puede tener millones de filas; un generador lo lee de una en una, con la memoria de una sola barra, y entrega cada barra ya convertida a tipos de Python. El módulo csv de la biblioteca estándar se encarga del formato (la lección 0.4.4 lo ve con más detalle):

import csv
from collections.abc import Iterable, Iterator
from dataclasses import dataclass
from datetime import date

@dataclass(frozen=True, slots=True)
class Barra:
    fecha: date
    apertura: float
    maximo: float
    minimo: float
    cierre: float
    volumen: int

def barras(ruta: str) -> Iterator[Barra]:
    with open(ruta, newline="", encoding="utf-8") as f:
        for fila in csv.DictReader(f):
            yield Barra(date.fromisoformat(fila["fecha"]),
                        float(fila["apertura"]),
                        float(fila["maximo"]),
                        float(fila["minimo"]),
                        float(fila["cierre"]),
                        int(fila["volumen"]))

flujo = barras("sint_diario.csv")
print(type(flujo).__name__)
print(next(flujo))
print(next(flujo).cierre)
print(sum(1 for _ in flujo), "barras más")

generator
Barra(fecha=datetime.date(2021, 1, 4), apertura=100.02, maximo=100.64, minimo=99.07, cierre=99.14, volumen=2322609)
99.25
1301 barras más

La anotación Iterator[Barra] dice qué produce el generador (lección 0.4.2). La última línea cuenta las barras que quedan con una expresión generadora, (1 for _ in flujo): la misma sintaxis que una comprensión de lista, entre paréntesis, y sin construir ninguna lista.

El with open(...) dentro del generador es importante: el fichero se cierra cuando el generador termina. Si se abandona a medias, el fichero queda abierto hasta que Python recoge el generador, un detalle que en un proceso que corre semanas puede agotar los descriptores de fichero.

Ventanas móviles y pares consecutivos

Dos operaciones aparecen en casi cualquier indicador: mirar las últimas $n$ barras (medias, máximos, volatilidad) y mirar cada barra junto a la anterior (rentabilidades, cruces). Sobre un flujo, sin índices, se resuelven con dos piezas de la biblioteca estándar.

collections.deque(maxlen=n) es una cola de tamaño fijo: al añadir un elemento con la cola llena, sale el más antiguo por el otro extremo. Es una ventana móvil hecha:

from collections import deque

def medias_moviles(flujo: Iterable[Barra],
                   n: int) -> Iterator[tuple[date, float]]:
    ventana: deque[float] = deque(maxlen=n)
    for b in flujo:
        ventana.append(b.cierre)
        if len(ventana) == n:
            yield b.fecha, sum(ventana) / n

ultimas = deque(medias_moviles(barras("sint_diario.csv"), 200),
                maxlen=3)
for fecha, media in ultimas:
    print(fecha, round(media, 2))

2025-12-29 136.62
2025-12-30 136.66
2025-12-31 136.72

La primera media sale en la barra 200: antes no hay ventana completa, igual que con rolling(200) en la lección 0.1.1. La segunda deque es un truco habitual: consumir un generador entero y quedarse solo con sus últimos elementos, sin guardar el resto.

itertools.pairwise (Python 3.10+) entrega cada elemento junto al siguiente: (b0, b1), (b1, b2)… Es exactamente lo que necesita una rentabilidad:

from itertools import islice, pairwise

def rentabilidades(flujo: Iterable[Barra]
                   ) -> Iterator[tuple[date, float]]:
    for ayer, hoy in pairwise(flujo):
        yield hoy.fecha, hoy.cierre / ayer.cierre - 1

primeras = islice(rentabilidades(barras("sint_diario.csv")), 3)
print([(str(f), round(r, 5)) for f, r in primeras])
fecha, peor = min(rentabilidades(barras("sint_diario.csv")),
                  key=lambda x: x[1])
print(f"Peor día: {fecha} ({peor:.2%})")

[('2021-01-05', 0.00111), ('2021-01-06', -0.01884), ('2021-01-07', 0.01212)]
Peor día: 2022-04-22 (-4.81%)

La peor sesión de los cinco años, un −4,81 %, cae en abril de 2022, en plena caída de ese año. islice corta un iterador como si fuera una lista ([:3]), sin recorrer el resto. Las piezas se encadenan: cada generador consume el anterior y nadie guarda el histórico completo.

flowchart TB
  F[sint_diario.csv] --> B["barras()"]
  B --> R["rentabilidades()"]
  R --> M["min()"]

Cuando llega la barra del primer día, min() todavía no ha visto nada; cuando llega la última, ya ha comparado todas. Es el mismo esquema de un sistema en tiempo real que procesa cada barra al llegar, y por eso los generadores son la base de muchos motores de backtest orientados a eventos (nivel 2).

caidas = (r for _, r in rentabilidades(barras("sint_diario.csv"))
          if r < -0.02)
print("Días con caída > 2 %:", sum(1 for _ in caidas))
print("Segunda pasada:", sum(1 for _ in caidas))

Días con caída > 2 %: 45
Segunda pasada: 0

La segunda cuenta da cero sin ningún aviso. Si se necesita recorrer los datos varias veces, se guardan en una lista o se crea el generador de nuevo, como se ha hecho en cada celda de esta lección.

Decoradores: envolver una función

En Python las funciones son objetos: se pueden pasar como argumento y devolver desde otra función. Un decorador es una función que recibe una función y devuelve otra que la envuelve, normalmente para añadir algo antes o después de la llamada. La sintaxis @decorador encima de una definición es una abreviatura de funcion = decorador(funcion). Ya se han usado varios: @dataclass, @property, @classmethod.

Un uso natural en trading es validar entradas en un solo sitio. Un precio cero o negativo en el cálculo del tamaño de una posición no debe producir una cantidad absurda, sino un error:

import functools

def positivos(func):
    """Rechaza argumentos numéricos menores o iguales que cero."""
    @functools.wraps(func)
    def envoltura(*args, **kwargs):
        for valor in (*args, *kwargs.values()):
            if isinstance(valor, (int, float)) and valor <= 0:
                raise ValueError(
                    f"{func.__name__}: valor no positivo {valor}")
        return func(*args, **kwargs)
    return envoltura

@positivos
def titulos(capital: float, precio: float, riesgo: float = 0.01,
            stop: float = 0.05) -> int:
    """Títulos tales que un stop del `stop` pierda `riesgo`."""
    return int(capital * riesgo / (precio * stop))

print(titulos(50_000, 142.22))
print(titulos.__name__, "|", titulos.__doc__)
try:
    titulos(50_000, 0.0)
except ValueError as e:
    print(e)

70
titulos | Títulos tales que un stop del `stop` pierda `riesgo`.
titulos: valor no positivo 0.0

Con 50.000 € de capital, un 1 % de riesgo y un stop al 5 % por debajo de 142,22, el tamaño es de 70 títulos: si el stop salta, se pierden unos 500 €. Con precio cero, la función original dividiría por cero; la decorada lo rechaza con un mensaje que dice qué función y qué valor.

envoltura recibe *args y **kwargs (lección 0.3.5) para poder envolver cualquier función, tenga los argumentos que tenga. @functools.wraps(func) copia a la envoltura el nombre y la docstring de la original: sin él, titulos se llamaría envoltura en los registros y en los tracebacks.

Decoradores con argumentos

Cuando el decorador necesita parámetros, hace falta un nivel más: una función que recibe los parámetros y devuelve el decorador. El ejemplo clásico es reintentar una llamada que falla por la red, como pedir un precio a un bróker. El bróker simulado falla las dos primeras veces y responde a la tercera, siempre igual, para que la salida sea reproducible:

def reintentar(veces: int, errores=(ConnectionError,)):
    def decorador(func):
        @functools.wraps(func)
        def envoltura(*args, **kwargs):
            for intento in range(1, veces + 1):
                try:
                    return func(*args, **kwargs)
                except errores as e:
                    print(f"  intento {intento}/{veces}: {e}")
            raise ConnectionError(
                f"{func.__name__}: sin respuesta tras {veces}")
        return envoltura
    return decorador

respuestas = iter([False, False, True])

@reintentar(veces=3)
def precio_actual(simbolo: str) -> float:
    if not next(respuestas):
        raise ConnectionError(f"timeout pidiendo {simbolo}")
    return 142.22

print(precio_actual("SINT"))

  intento 1/3: timeout pidiendo SINT
  intento 2/3: timeout pidiendo SINT
142.22

Un reintento real espera entre intentos, cada vez más (backoff exponencial), para no saturar al servidor; aquí se omite para no medir tiempos. Lo que no se omite nunca es pensar qué se reintenta.

Context managers: limpiar pase lo que pase

with open(...) as f garantiza que el fichero se cierra al salir del bloque, también si dentro se lanza una excepción. Cualquier objeto con los métodos __enter__ y __exit__ sirve para with: es un context manager. __enter__ se ejecuta al entrar y su resultado es lo que recibe la variable de as; __exit__ se ejecuta al salir, siempre, y recibe la excepción si la hubo.

En trading, la limpieza importa: al terminar una sesión, por las buenas o por un error, no deben quedar órdenes pendientes sin nadie que las vigile. Una sesión operativa simulada:

class SesionOperativa:
    def __init__(self, cuenta: str):
        self.cuenta = cuenta
        self.pendientes: list[str] = []

    def __enter__(self):
        print(f"[{self.cuenta}] sesión abierta")
        return self

    def enviar(self, orden: str) -> None:
        self.pendientes.append(orden)
        print(f"[{self.cuenta}] enviada: {orden}")

    def ejecutada(self, orden: str) -> None:
        self.pendientes.remove(orden)

    def __exit__(self, tipo, valor, traza):
        for orden in self.pendientes:
            print(f"[{self.cuenta}] cancelada: {orden}")
        self.pendientes.clear()
        motivo = f"error: {valor}" if tipo else "normal"
        print(f"[{self.cuenta}] sesión cerrada ({motivo})")
        return False  # no suprime la excepción

with SesionOperativa("demo") as s:
    s.enviar("compra 70 SINT lim 141.50")
    s.enviar("venta 40 ORO stop 165.00")
    s.ejecutada("compra 70 SINT lim 141.50")

[demo] sesión abierta
[demo] enviada: compra 70 SINT lim 141.50
[demo] enviada: venta 40 ORO stop 165.00
[demo] cancelada: venta 40 ORO stop 165.00
[demo] sesión cerrada (normal)

La orden ejecutada sale de la lista; la que seguía viva se cancela al cerrar. Con un error a mitad de sesión, el cierre ocurre igual y la excepción sigue su camino hasta quien la capture:

try:
    with SesionOperativa("demo") as s:
        s.enviar("compra 100 TECH lim 77.50")
        raise RuntimeError("feed de precios caído")
        s.enviar("venta 100 TECH lim 79.00")  # no llega
except RuntimeError as e:
    print("capturada fuera:", e)

[demo] sesión abierta
[demo] enviada: compra 100 TECH lim 77.50
[demo] cancelada: compra 100 TECH lim 77.50
[demo] sesión cerrada (error: feed de precios caído)
capturada fuera: feed de precios caído

__exit__ devuelve False, así que la excepción no se pierde. Devolver True la suprimiría: el programa seguiría como si nada, que es casi siempre peor que un error visible.

Para casos sencillos, contextlib.contextmanager convierte un generador en un context manager: lo que va antes del yield es la entrada, y lo que va en el finally, la salida garantizada. Un ejemplo: suspender la operativa durante la publicación de un dato macro y reactivarla después, pase lo que pase:

from contextlib import contextmanager

estado = {"operativa": True}

@contextmanager
def operativa_suspendida(motivo: str):
    estado["operativa"] = False
    print(f"operativa suspendida: {motivo}")
    try:
        yield
    finally:
        estado["operativa"] = True
        print("operativa reactivada")

with operativa_suspendida("dato de empleo de EE. UU."):
    print("dentro:", estado)
print("fuera:", estado)

operativa suspendida: dato de empleo de EE. UU.
dentro: {'operativa': False}
operativa reactivada
fuera: {'operativa': True}

Sin el try/finally, una excepción dentro del bloque dejaría la operativa suspendida para siempre.

Ejercicios

1. Escribe un generador barras_del_anio(flujo, anio) que deje pasar solo las barras de ese año. Úsalo con barras("sint_diario.csv") para contar las sesiones de 2022 y encontrar el cierre mínimo de ese año.

def barras_del_anio(flujo: Iterable[Barra],
                    anio: int) -> Iterator[Barra]:
    for b in flujo:
        if b.fecha.year == anio:
            yield b

de_2022 = list(barras_del_anio(barras("sint_diario.csv"), 2022))
minima = min(de_2022, key=lambda b: b.cierre)
print(len(de_2022), minima.fecha, minima.cierre)

260 2022-10-20 94.45

2. Escribe un decorador contar_llamadas que guarde en el atributo llamadas de la función decorada cuántas veces se ha llamado. Aplícalo a titulos (la función sin decorar está en titulos.__wrapped__) y llámala tres veces.

def contar_llamadas(func):
    @functools.wraps(func)
    def envoltura(*args, **kwargs):
        envoltura.llamadas += 1
        return func(*args, **kwargs)
    envoltura.llamadas = 0
    return envoltura

titulos_contados = contar_llamadas(titulos.__wrapped__)
for precio in (142.22, 77.82, 171.44):
    titulos_contados(50_000, precio)
print(titulos_contados.llamadas)

3

3. Con @contextmanager, escribe registro_ordenes() que entregue una lista vacía y, al salir del bloque, imprima cuántas órdenes se añadieron y su importe total. Prueba con tres órdenes (cantidad, precio).

@contextmanager
def registro_ordenes():
    ordenes: list[tuple[int, float]] = []
    try:
        yield ordenes
    finally:
        total = sum(c * p for c, p in ordenes)
        print(f"{len(ordenes)} órdenes, importe {total:,.2f}")

with registro_ordenes() as reg:
    reg.append((70, 142.22))
    reg.append((100, 77.82))
    reg.append((20, 171.44))

3 órdenes, importe 21,166.20