マルチプロセシング

別々のプロセス、別々のメモリ

GIL のため、スレッドは CPU バウンドの Python を速くできませんが、プロセスなら可能です。各プロセスは自分の GIL を持つ完全な Python インタプリタで、OS がそれらを別々のコアで動かします。multiprocessing モジュールと、より使いやすい concurrent.futures.ProcessPoolExecutor が、ワーカープロセスのプールを起動し、それぞれに仕事の一部を渡します。1000通りのパラメータでのバックテストや数百万パスのモンテカルロは、ノート PC の8コアすべてを使って、およそ8分の1の時間で終わらせられます。

# 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

境界を越えるもの：コピー、pickle、チャンク

プロセス同士は変数を共有しません。ワーカーに送るものすべて、つまり関数とその引数は、pickle（バイト列にシリアライズ）されてコピーされ、再構築されます。結果も同じ経路で戻ります。そのため pickle できるものだけが境界を越えられます。モジュール直下の関数や単純なデータは可能で、ラムダ、入れ子の関数、ロック、ジェネレータは不可で、PicklingError や AttributeError で失敗します。コピーにはコストがかかるので、送るのは少なく、計算は多くします。ワーカーがデータを自分で読み込める・生成できるなら、データのコピーではなく添字のペア (start, stop) を渡します。100万個の小さなタスクではなく数個の大きなチャンクを配り、配列全体ではなく合計のような小さな結果を返します。

# 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

独立した乱数列と部分結果の統合

並列モンテカルロが正しいかどうかは、2つの細部で決まります。1つ目は乱数シードです。他のワーカーと同じシードで始めたワーカーは同じ「乱数」を出しますし、seed + 0、seed + 1、…を渡すと、重なったり相関したりしうる乱数列ができます。np.random.SeedSequence(seed).spawn(n) は、1つの親から統計的に独立な n 個の子シードを導くので、実行全体は1つの数から再現できます。2つ目は統合です。各ワーカーが小さな要約を返し、それをまとめます。個数は足し算、平均は個数で重み付けした平均として合成し、全体の分散にはワーカーの平均どうしのばらつきを含める必要があるため、分散を単純に平均することはできません。

# 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

• プロセスは自分の GIL とメモリを持つので、CPU バウンドの Python も全コアを使える。プール（ProcessPoolExecutor）が仕事を配る
• 引数と結果は pickle されコピーされる。添字のペアと小さな結果を、少数の大きなチャンクで送る。ラムダや入れ子の関数は送れない
• SeedSequence.spawn で各ワーカーに独立したシードを渡し、部分統計を正確に統合し、入口を守り、実際の高速化を測る

次は SQL データベースです。テーブルとクエリで市場データを保存し、検索します。

ラボ：採点される内容

• chunk_bounds で範囲を区間に分ける25点
• combine_mean_var で部分統計を統合する25点
• spawn_seeds で各ワーカーに専用シードを渡す25点
• parallel_speedup で並列化のコストを見積もる25点