Multiprocessing

Processos separados, memórias separadas

Threads não conseguem acelerar Python limitado pela CPU por causa do GIL, mas processos conseguem: cada um é um interpretador Python completo com seu próprio GIL, e o sistema operacional os executa em núcleos diferentes. O módulo multiprocessing e o mais amigável concurrent.futures.ProcessPoolExecutor iniciam um grupo de processos de trabalho e dão a cada um uma parte do trabalho. Um backtest sobre mil conjuntos de parâmetros, ou um Monte Carlo com milhões de trajetórias, pode usar os oito núcleos de um notebook e terminar em cerca de um oitavo do tempo.

# 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

O que cruza a fronteira: cópias, pickle, pedaços

Processos não compartilham variáveis. Tudo que você envia a um trabalhador, a função e seus argumentos, é serializado (pickle, em bytes), copiado e reconstruído, e o resultado volta do mesmo modo. Por isso só atravessam coisas serializáveis: funções de nível de módulo e dados simples sim; lambdas, funções aninhadas, locks e geradores não, o que falha com PicklingError ou AttributeError. Copiar custa: envie pouco e calcule muito. Dê a cada trabalhador um par de índices (start, stop) em vez de uma cópia dos dados quando ele puder carregá-los ou gerá-los, distribua poucos pedaços grandes em vez de um milhão de tarefas minúsculas e devolva resultados pequenos, como uma soma, não um array inteiro.

# 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

Fluxos aleatórios independentes e combinação de resultados parciais

Dois detalhes decidem se um Monte Carlo paralelo está certo. Primeiro, as sementes aleatórias: um trabalhador que começa com a mesma semente de outro produz os mesmos números «aleatórios», e dar seed + 0, seed + 1, ... cria fluxos que podem se sobrepor ou se correlacionar. np.random.SeedSequence(seed).spawn(n) deriva n sementes filhas estatisticamente independentes de um pai, de modo que toda a execução continua reprodutível a partir de um só número. Segundo, a combinação: cada trabalhador devolve um resumo pequeno e você os mescla. As contagens se somam, as médias se combinam como média ponderada pela contagem, e a variância do todo deve incluir a dispersão entre as médias dos trabalhadores, então não basta tirar a média das variâncias.

# 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

• Processos têm seu próprio GIL e memória, então Python limitado pela CPU pode usar todos os núcleos; um grupo (ProcessPoolExecutor) distribui o trabalho.
• Argumentos e resultados são serializados e copiados: envie pares de índices e resultados pequenos, em poucos pedaços grandes; lambdas e funções aninhadas não podem ser enviadas.
• Dê a cada trabalhador uma semente independente com SeedSequence.spawn, mescle exatamente as estatísticas parciais, proteja o ponto de entrada e meça o ganho real.

A seguir: bancos de dados SQL, armazenar e consultar dados de mercado com tabelas e consultas.

Lab: o que é avaliado

• Cortar o intervalo em fatias com chunk_bounds25 pts
• Combinar estatísticas parciais com combine_mean_var25 pts
• Dar a cada trabalhador sua semente com spawn_seeds25 pts
• Contar o custo de paralelizar com parallel_speedup25 pts