Multiprocessing

Getrennte Prozesse, getrennter Speicher

Threads können CPU-gebundenes Python wegen der GIL nicht beschleunigen, Prozesse aber schon: Jeder ist ein vollständiger Python-Interpreter mit eigener GIL, und das Betriebssystem führt sie auf verschiedenen Kernen aus. Das Modul multiprocessing und das freundlichere concurrent.futures.ProcessPoolExecutor starten einen Pool von Arbeitsprozessen und geben jedem einen Teil der Arbeit. Ein Backtest über tausend Parametersätze oder ein Monte-Carlo mit Millionen Pfaden kann alle acht Kerne eines Laptops nutzen und in etwa einem Achtel der Zeit fertig sein.

# 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

Was die Grenze überquert: Kopien, Pickling, Teile

Prozesse teilen keine Variablen. Alles, was Sie an einen Worker senden, die Funktion und ihre Argumente, wird gepickelt (zu Bytes serialisiert), kopiert und wieder aufgebaut, und das Ergebnis kommt auf dem gleichen Weg zurück. Es können also nur picklebare Dinge die Grenze überqueren: Funktionen auf Modulebene und einfache Daten ja, Lambdas, verschachtelte Funktionen, Locks und Generatoren nein, was mit PicklingError oder AttributeError scheitert. Kopieren kostet: Senden Sie wenig und rechnen Sie viel. Geben Sie jedem Worker ein Indexpaar (start, stop) statt einer Datenkopie, wenn er die Daten selbst laden oder erzeugen kann, verteilen Sie wenige große Teile statt einer Million winziger Aufgaben und liefern Sie kleine Ergebnisse wie eine Summe zurück, kein ganzes Array.

# 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

Unabhängige Zufallsströme und das Zusammenführen von Teilergebnissen

Zwei Details entscheiden, ob ein paralleles Monte-Carlo stimmt. Erstens die Zufalls-Seeds: Ein Worker, der mit demselben Seed startet wie ein anderer, erzeugt dieselben „zufälligen“ Zahlen, und seed + 0, seed + 1, ... ergibt Ströme, die sich überlappen oder korrelieren können. np.random.SeedSequence(seed).spawn(n) leitet n statistisch unabhängige Kind-Seeds aus einem Eltern-Seed ab, sodass der ganze Lauf aus einer einzigen Zahl reproduzierbar bleibt. Zweitens das Zusammenführen: Jeder Worker liefert eine kleine Zusammenfassung, die Sie vereinen. Anzahlen addieren sich, Mittelwerte kombinieren sich als mit der Anzahl gewichteter Durchschnitt, und die Varianz des Ganzen muss die Streuung zwischen den Worker-Mittelwerten enthalten, man kann die Varianzen also nicht einfach mitteln.

# 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

• Prozesse haben eigene GIL und eigenen Speicher, daher kann CPU-gebundenes Python alle Kerne nutzen; ein Pool (ProcessPoolExecutor) verteilt die Arbeit.
• Argumente und Ergebnisse werden gepickelt und kopiert: Senden Sie Indexpaare und kleine Ergebnisse in wenigen großen Teilen; Lambdas und verschachtelte Funktionen lassen sich nicht senden.
• Geben Sie jedem Worker mit SeedSequence.spawn einen unabhängigen Seed, führen Sie Teilstatistiken exakt zusammen, sichern Sie den Einstiegspunkt und messen Sie den echten Speed-up.

Als Nächstes: SQL-Datenbanken, Marktdaten mit Tabellen und Abfragen speichern und abfragen.

Lab: Was bewertet wird

• Mit chunk_bounds den Bereich in Abschnitte teilen25 Pkt.
• Mit combine_mean_var Teilstatistiken zusammenführen25 Pkt.
• Mit spawn_seeds jedem Worker einen eigenen Seed geben25 Pkt.
• Mit parallel_speedup die Kosten der Parallelisierung beziffern25 Pkt.