多进程

独立的进程，独立的内存

由于 GIL，线程无法加速 CPU 密集的 Python，但进程可以：每个进程都是拥有自己 GIL 的完整 Python 解释器，操作系统会把它们放在不同的核心上运行。multiprocessing 模块以及更友好的 concurrent.futures.ProcessPoolExecutor 会启动一池工作进程，并给每个进程分一部分工作。针对一千组参数的回测，或有数百万条路径的蒙特卡洛，可以用上笔记本电脑的全部八个核心，在大约八分之一的时间内完成。

# 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 的东西才能跨越边界：模块顶层的函数和普通数据可以，lambda、嵌套函数、锁和生成器不行，会以 PicklingError 或 AttributeError 失败。复制有代价，所以要少发多算：当工作进程能自己加载或生成数据时，给它一对下标 (start, stop) 而不是数据的副本；分发少量大块而不是一百万个小任务；返回像总和这样的小结果，而不是整个数组。

# 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

独立的随机流与合并部分结果

有两个细节决定并行蒙特卡洛是否正确。第一是随机种子：与另一个工作进程使用相同种子的进程会产生相同的“随机”数，而给各进程 seed + 0、seed + 1、…… 会得到可能重叠或相关的随机流。np.random.SeedSequence(seed).spawn(n) 从一个父种子派生出 n 个统计上独立的子种子，所以整个运行仍可由一个数字复现。第二是合并：每个工作进程返回一个小的汇总，再把它们合并。计数相加，均值按计数加权平均，而整体的方差必须包含各进程均值之间的离散程度，所以不能简单地把方差取平均。

# 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 并复制：用少量大块发送下标对和小结果；lambda 和嵌套函数无法发送。
• 用 SeedSequence.spawn 为每个工作进程分配独立种子，精确合并部分统计量，保护入口点，并测量真实的加速比。

下一节：SQL 数据库，用表和查询来存储和检索市场数据。

实验：评分内容

• 用 chunk_bounds 把范围切成若干段25 分
• 用 combine_mean_var 合并部分统计量25 分
• 用 spawn_seeds 为每个工作进程分配种子25 分
• 用 parallel_speedup 计算并行化的代价25 分