Multiprocessing

Des processus séparés, des mémoires séparées

Les threads ne peuvent pas accélérer le Python lié au processeur à cause du GIL, mais les processus le peuvent : chacun est un interpréteur Python complet avec son propre GIL, et le système d'exploitation les exécute sur des cœurs différents. Le module multiprocessing et le plus agréable concurrent.futures.ProcessPoolExecutor lancent un groupe de processus de travail et donnent à chacun une part du travail. Un backtest sur mille jeux de paramètres, ou un Monte Carlo de millions de trajectoires, peut utiliser les huit cœurs d'un portable et se terminer en environ un huitième du temps.

# 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

Ce qui franchit la frontière : copies, pickle, morceaux

Les processus ne partagent pas de variables. Tout ce que vous envoyez à un travailleur, la fonction et ses arguments, est sérialisé (pickle, mis en octets), copié puis reconstruit, et le résultat revient de la même façon. Seules les choses sérialisables peuvent donc traverser : fonctions de niveau module et données simples oui, lambdas, fonctions imbriquées, verrous et générateurs non, ce qui échoue avec une PicklingError ou AttributeError. Copier a un coût : envoyez peu et calculez beaucoup. Donnez à chaque travailleur une paire d'indices (start, stop) plutôt qu'une copie des données quand il peut les charger ou les générer lui-même, distribuez quelques gros morceaux plutôt qu'un million de petites tâches, et renvoyez de petits résultats comme une somme, pas un tableau entier.

# 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

Flux aléatoires indépendants et combinaison des résultats partiels

Deux détails décident si un Monte Carlo parallèle est juste. D'abord, les graines aléatoires : un travailleur qui démarre avec la même graine qu'un autre produit les mêmes nombres « aléatoires », et donner seed + 0, seed + 1, ... crée des flux qui peuvent se recouvrir ou se corréler. np.random.SeedSequence(seed).spawn(n) dérive n graines filles statistiquement indépendantes d'un parent, si bien que toute l'exécution reste reproductible à partir d'un seul nombre. Ensuite, la combinaison : chaque travailleur renvoie un petit résumé et on les fusionne. Les effectifs s'additionnent, les moyennes se combinent en moyenne pondérée par l'effectif, et la variance de l'ensemble doit inclure la dispersion entre les moyennes des travailleurs : on ne peut donc pas simplement moyenner les variances.

# 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

• Les processus ont leur propre GIL et leur mémoire : du Python lié au processeur peut utiliser tous les cœurs ; un groupe (ProcessPoolExecutor) distribue le travail.
• Arguments et résultats sont sérialisés et copiés : envoyez des paires d'indices et de petits résultats, en quelques gros morceaux ; les lambdas et fonctions imbriquées ne passent pas.
• Donnez à chaque travailleur une graine indépendante avec SeedSequence.spawn, fusionnez exactement les statistiques partielles, protégez le point d'entrée et mesurez l'accélération réelle.

À suivre : les bases de données SQL, stocker et interroger des données de marché avec des tables et des requêtes.

Lab : ce qui est noté

• Découper la plage en tranches avec chunk_bounds25 pts
• Fusionner des statistiques partielles avec combine_mean_var25 pts
• Donner à chaque travailleur sa graine avec spawn_seeds25 pts
• Compter le coût du parallélisme avec parallel_speedup25 pts