Multiprocessing

Procesos separados, memorias separadas

Los hilos no pueden acelerar el Python limitado por CPU a causa del GIL, pero los procesos sí: cada uno es un intérprete de Python completo con su propio GIL, y el sistema operativo los ejecuta en núcleos distintos. El módulo multiprocessing y el más amable concurrent.futures.ProcessPoolExecutor inician un grupo de procesos trabajadores y dan a cada uno una parte del trabajo. Un backtest sobre mil conjuntos de parámetros, o un Monte Carlo con millones de trayectorias, puede usar los ocho núcleos de un portátil y terminar en aproximadamente un octavo del tiempo.

# A pool of processes: each gets a slice of the work, computed on its own core
from concurrent.futures import ProcessPoolExecutor

def sum_of_squares(bounds):
    start, stop = bounds
    return sum(i * i for i in range(start, stop))

if __name__ == "__main__":                           # required on Windows and macOS, where workers re-import this file
    bounds = [(0, 250_000), (250_000, 500_000), (500_000, 750_000), (750_000, 1_000_000)]
    with ProcessPoolExecutor(max_workers=4) as pool:
        partials = list(pool.map(sum_of_squares, bounds))
    print(sum(partials))                             # 333332833333500000
    print(sum(partials) == sum(i * i for i in range(1_000_000)))        # True  the same answer as one process

Lo que cruza la frontera: copias, pickle, trozos

Los procesos no comparten variables. Todo lo que envías a un trabajador, la función y sus argumentos, se serializa (pickle, a bytes), se copia y se reconstruye, y el resultado vuelve del mismo modo. Por eso solo cruzan las cosas serializables: funciones de nivel de módulo y datos simples sí; lambdas, funciones anidadas, cerrojos y generadores no, lo que falla con una PicklingError o AttributeError. Copiar cuesta: envía poco y calcula mucho. Da a cada trabajador un par de índices (start, stop) en vez de una copia de los datos cuando pueda cargarlos o generarlos él mismo, reparte unos pocos trozos grandes en vez de un millón de tareas diminutas y devuelve resultados pequeños como una suma, no un array entero.

# Index pairs instead of data, and a check that something can be sent to another process
import pickle

def chunk_bounds(n, parts):
    size, extra = divmod(n, parts)
    bounds, start = [], 0
    for index in range(parts):
        stop = start + size + (1 if index < extra else 0)
        bounds.append((start, stop))
        start = stop
    return bounds

def is_picklable(obj):
    try:
        pickle.dumps(obj)
    except (pickle.PicklingError, AttributeError, TypeError):
        return False
    return True

print(chunk_bounds(1_000_000, 4))                    # [(0, 250000), (250000, 500000), (500000, 750000), (750000, 1000000)]
print(is_picklable({"AAPL": 184.25}), is_picklable(lambda x: x))      # True False

Flujos aleatorios independientes y combinación de resultados parciales

Dos detalles deciden si un Monte Carlo paralelo es correcto. Primero, las semillas aleatorias: un trabajador que empieza con la misma semilla que otro produce los mismos números «aleatorios», y dar seed + 0, seed + 1, ... crea flujos que pueden solaparse o correlacionarse. np.random.SeedSequence(seed).spawn(n) deriva n semillas hijas estadísticamente independientes de un padre, de modo que toda la ejecución sigue siendo reproducible a partir de un solo número. Segundo, la combinación: cada trabajador devuelve un resumen pequeño y se fusionan. Los recuentos se suman, las medias se combinan como media ponderada por el recuento y la varianza del conjunto debe incluir la dispersión entre las medias de los trabajadores, así que no basta con promediar las varianzas.

# Independent seeds for the workers, and merging their partial statistics exactly
import numpy as np

def spawn_seeds(seed, n):
    return [int(child.generate_state(1)[0]) for child in np.random.SeedSequence(seed).spawn(n)]

def combine_mean_var(partials):
    count = sum(n for n, _, _ in partials)
    mean = sum(n * m for n, m, _ in partials) / count
    variance = sum(n * (v + (m - mean) ** 2) for n, m, v in partials) / count
    return count, mean, variance

seeds = spawn_seeds(7, 4)
print(len(set(seeds)), spawn_seeds(7, 4) == seeds)   # 4 True   four different seeds, reproducible from one number

data = np.random.default_rng(seeds[0]).normal(0.0, 1.0, size=1_000)
partials = [(len(p), p.mean(), p.var()) for p in np.array_split(data, 4)]
count, mean, variance = combine_mean_var(partials)
print(count, np.isclose(mean, data.mean()), np.isclose(variance, data.var()))      # 1000 True True

# A parallel Monte Carlo: every worker has its own seed, and the run is still exactly reproducible
from concurrent.futures import ProcessPoolExecutor

import numpy as np

def simulate(job):
    seed, paths = job
    z = np.random.default_rng(seed).standard_normal(paths)
    payoff = np.maximum(100.0 * np.exp(-0.5 * 0.2 ** 2 + 0.2 * z) - 100.0, 0.0)   # at-the-money call, 20% volatility
    return paths, payoff.mean(), payoff.var()

if __name__ == "__main__":
    seeds = [int(child.generate_state(1)[0]) for child in np.random.SeedSequence(2024).spawn(4)]
    jobs = [(seed, 50_000) for seed in seeds]
    with ProcessPoolExecutor(max_workers=4) as pool:
        partials = list(pool.map(simulate, jobs))
    count = sum(n for n, _, _ in partials)
    price = sum(n * mean for n, mean, _ in partials) / count
    print(count, round(price, 1))                    # 200000 7.9   close to the exact Black-Scholes value of about 7.97
    print(partials == [simulate(job) for job in jobs])      # True  the same seeds give the same numbers in one process

• Los procesos tienen su propio GIL y memoria, así que Python limitado por CPU puede usar todos los núcleos; un grupo (ProcessPoolExecutor) reparte el trabajo.
• Argumentos y resultados se serializan y copian: envía pares de índices y resultados pequeños, en pocos trozos grandes; las lambdas y funciones anidadas no pueden enviarse.
• Da a cada trabajador una semilla independiente con SeedSequence.spawn, combina con exactitud las estadísticas parciales, protege el punto de entrada y mide la aceleración real.

A continuación: bases de datos SQL, almacenar y consultar datos de mercado con tablas y consultas.

Lab: qué se califica

• Cortar el rango en tramos con chunk_bounds25 pts
• Combinar estadísticas parciales con combine_mean_var25 pts
• Dar a cada trabajador su semilla con spawn_seeds25 pts
• Contar el coste de ir en paralelo con parallel_speedup25 pts