Iterators, generators, decorators and context managers.

What for does under the hood

A for loop over a list of closes looks like a single, indivisible operation. It is not: Python asks the list for an iterator with iter() and asks it for values with next() until the iterator raises StopIteration. Doing it by hand shows the mechanism:

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 closes left")

99.14 99.25 101.02
StopIteration: no closes left

An iterable is any object that can provide an iterator (lists, tuples, dictionaries, files, strings). An iterator is the object that does the traversal: it remembers where it is and hands over the next value. The distinction matters because an iterator gets used up: once the last value has been delivered, it does not start over.

Any class with the __iter__ and __next__ methods is an iterator. Writing one by hand is tedious; generators do the same with much less code.

Generators: bars on demand

A function that contains yield is a generator. Calling it does not run it: it returns a generator object. Each next() runs the body up to the next yield, hands over the value and leaves the function frozen at that point, with its variables intact, until the next request. Printing inside makes the lazy execution visible:

def feed_simulado():
    print("  connecting to feed")
    yield 142.10
    print("  second tick")
    yield 142.15
    print("  feed closed")

feed = feed_simulado()
print("generator created")
print(next(feed))
print(next(feed))
print(list(feed))

generator created
  connecting to feed
142.1
  second tick
142.15
  feed closed
[]

"Connecting" does not appear when the generator is created, but on the first next(). The third pass (list(feed)) runs the rest of the body, finds no more yield and ends with an empty list.

This way of working fits market data. A file of bars can have millions of rows; a generator reads it one at a time, with the memory of a single bar, and hands over each bar already converted to Python types. The standard library's csv module takes care of the format (lesson 0.4.4 covers it in more detail):

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), "more bars")

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 more bars

The annotation Iterator[Barra] says what the generator produces (lesson 0.4.2). The last line counts the remaining bars with a generator expression, (1 for _ in flujo): the same syntax as a list comprehension, in parentheses, and without building any list.

The with open(...) inside the generator matters: the file is closed when the generator finishes. If it is abandoned halfway, the file stays open until Python collects the generator, a detail that in a process running for weeks can exhaust the file descriptors.

Rolling windows and consecutive pairs

Two operations appear in almost every indicator: looking at the last $n$ bars (averages, highs, volatility) and looking at each bar next to the previous one (returns, crossovers). Over a stream, without indexes, they are solved with two pieces of the standard library.

collections.deque(maxlen=n) is a fixed-size queue: when an element is added to a full queue, the oldest one drops out at the other end. It is a ready-made rolling window:

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

The first average comes out at bar 200: before that there is no complete window, just as with rolling(200) in lesson 0.1.1. The second deque is a common trick: consume an entire generator and keep only its last elements, without storing the rest.

itertools.pairwise (Python 3.10+) hands over each element together with the next one: (b0, b1), (b1, b2)… It is exactly what a return needs:

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"Worst day: {fecha} ({peor:.2%})")

[('2021-01-05', 0.00111), ('2021-01-06', -0.01884), ('2021-01-07', 0.01212)]
Worst day: 2022-04-22 (-4.81%)

The worst session of the five years, a −4.81% day, falls in April 2022, in the middle of that year's decline. islice slices an iterator as if it were a list ([:3]), without traversing the rest. The pieces chain together: each generator consumes the previous one and nobody stores the full history.

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

When the first day's bar arrives, min() has not yet seen anything; when the last one arrives, it has already compared them all. It is the same scheme as a real-time system that processes each bar as it arrives, and that is why generators are the foundation of many event-driven backtest engines (level 2).

caidas = (r for _, r in rentabilidades(barras("sint_diario.csv"))
          if r < -0.02)
print("Days with a drop > 2%:", sum(1 for _ in caidas))
print("Second pass:", sum(1 for _ in caidas))

Days with a drop > 2%: 45
Second pass: 0

The second count gives zero without any warning. If the data has to be traversed several times, it is stored in a list or the generator is created again, as has been done in every cell of this lesson.

Decorators: wrapping a function

In Python, functions are objects: they can be passed as arguments and returned from another function. A decorator is a function that takes a function and returns another that wraps it, usually to add something before or after the call. The @decorador syntax above a definition is shorthand for funcion = decorador(funcion). Several have already been used: @dataclass, @property, @classmethod.

A natural use in trading is validating inputs in a single place. A zero or negative price in a position sizing calculation should not produce an absurd quantity, but an error:

import functools

def positivos(func):
    """Rejects numeric arguments less than or equal to zero."""
    @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__}: non-positive value {valor}")
        return func(*args, **kwargs)
    return envoltura

@positivos
def titulos(capital: float, precio: float, riesgo: float = 0.01,
            stop: float = 0.05) -> int:
    """Shares such that a stop at `stop` loses `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 | Shares such that a stop at `stop` loses `riesgo`.
titulos: non-positive value 0.0

With €50,000 of capital, 1% risk and a stop 5% below 142.22, the size is 70 shares: if the stop is hit, about €500 is lost. With a zero price, the original function would divide by zero; the decorated one rejects it with a message that says which function and which value.

envoltura receives *args and **kwargs (lesson 0.3.5) so it can wrap any function, whatever its arguments. @functools.wraps(func) copies the original's name and docstring to the wrapper: without it, titulos would be called envoltura in logs and tracebacks.

Decorators with arguments

When the decorator needs parameters, one more level is required: a function that takes the parameters and returns the decorator. The classic example is retrying a call that fails because of the network, such as requesting a price from a broker. The simulated broker fails the first two times and responds on the third, always the same way, so the output is 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"  attempt {intento}/{veces}: {e}")
            raise ConnectionError(
                f"{func.__name__}: no response after {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 requesting {simbolo}")
    return 142.22

print(precio_actual("SINT"))

  attempt 1/3: timeout requesting SINT
  attempt 2/3: timeout requesting SINT
142.22

A real retry waits between attempts, longer each time (exponential backoff), so as not to overload the server; here it is left out to avoid measuring time. What is never left out is thinking about what gets retried.

Context managers: cleaning up no matter what

with open(...) as f guarantees that the file is closed when the block is exited, also if an exception is raised inside. Any object with the __enter__ and __exit__ methods works with with: it is a context manager. __enter__ runs on entry and its result is what the as variable receives; __exit__ runs on exit, always, and receives the exception if there was one.

In trading, cleanup matters: when a session ends, normally or because of an error, there must be no pending orders left with nobody watching them. A simulated trading session:

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

    def __enter__(self):
        print(f"[{self.cuenta}] session opened")
        return self

    def enviar(self, orden: str) -> None:
        self.pendientes.append(orden)
        print(f"[{self.cuenta}] sent: {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}] canceled: {orden}")
        self.pendientes.clear()
        motivo = f"error: {valor}" if tipo else "normal"
        print(f"[{self.cuenta}] session closed ({motivo})")
        return False  # does not suppress the exception

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

[demo] session opened
[demo] sent: buy 70 SINT lim 141.50
[demo] sent: sell 40 ORO stop 165.00
[demo] canceled: sell 40 ORO stop 165.00
[demo] session closed (normal)

The filled order leaves the list; the one still live is canceled on close. With an error mid-session, the close happens all the same and the exception continues on its way to whoever catches it:

try:
    with SesionOperativa("demo") as s:
        s.enviar("buy 100 TECH lim 77.50")
        raise RuntimeError("price feed down")
        s.enviar("sell 100 TECH lim 79.00")  # never reached
except RuntimeError as e:
    print("caught outside:", e)

[demo] session opened
[demo] sent: buy 100 TECH lim 77.50
[demo] canceled: buy 100 TECH lim 77.50
[demo] session closed (error: price feed down)
caught outside: price feed down

__exit__ returns False, so the exception is not lost. Returning True would suppress it: the program would carry on as if nothing had happened, which is almost always worse than a visible error.

For simple cases, contextlib.contextmanager turns a generator into a context manager: what comes before the yield is the entry, and what goes in the finally, the guaranteed exit. An example: suspending trading during the release of a macro data point and resuming it afterwards, no matter what:

from contextlib import contextmanager

estado = {"operativa": True}

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

with operativa_suspendida("US jobs report"):
    print("inside:", estado)
print("outside:", estado)

trading suspended: US jobs report
inside: {'operativa': False}
trading resumed
outside: {'operativa': True}

Without the try/finally, an exception inside the block would leave trading suspended forever.

Exercises

1. Write a generator barras_del_anio(flujo, anio) that only lets through the bars of that year. Use it with barras("sint_diario.csv") to count the sessions of 2022 and find that year's lowest close.

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. Write a decorator contar_llamadas that stores in the llamadas attribute of the decorated function how many times it has been called. Apply it to titulos (the undecorated function is in titulos.__wrapped__) and call it three times.

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. With @contextmanager, write registro_ordenes() that hands over an empty list and, on leaving the block, prints how many orders were added and their total amount. Test it with three (cantidad, precio) orders.

@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)} orders, amount {total:,.2f}")

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

3 orders, amount 21,166.20