8.3 向量化改写实战与性能清单


文档摘要

8.3 向量化改写实战与性能清单 本节摘要:把全册知识收束为三张改写模板(双重循环、循环 append、逐元素条件分支)和一张性能排查清单。案例从 200 倍提速的完整改写过程开始,落到可以贴在工位的十二项清单——每一项都对应前面某一章的原理。 改写实战:蒙特卡洛期权定价 完整过程。背景:用蒙特卡洛方法估计亚式期权收益,朴素版是三层循环,一天要跑几百次,性能不可接受。 操作,第一步看朴素版: 第二步找循环体里的依赖:每条路径内部有状态传递(s 逐步相乘),路径之间完全独立。独立方向先向量化,路径内的累积交给矩阵乘: 结果与解读:约 200 倍提速,两次价格差异来自随机路径不同(0.92 量级一致)。

8.3 向量化改写实战与性能清单

本节摘要:把全册知识收束为三张改写模板(双重循环、循环 append、逐元素条件分支)和一张性能排查清单。案例从 200 倍提速的完整改写过程开始,落到可以贴在工位的十二项清单——每一项都对应前面某一章的原理。

改写实战:蒙特卡洛期权定价

完整过程。背景:用蒙特卡洛方法估计亚式期权收益,朴素版是三层循环,一天要跑几百次,性能不可接受。

操作,第一步看朴素版:

import numpy as np import time rng = np.random.default_rng(7) n_path, n_step = 10_000, 100 # 朴素版:逐路径逐步推进(反面教材) t0 = time.perf_counter() payoffs = [] for i in range(n_path): s = 100.0 total = 0.0 for j in range(n_step): s *= 1.0 + 0.0001 + rng.normal(0, 0.02) total += s avg = total / n_step payoffs.append(max(avg - 100.0, 0.0)) price1 = np.mean(payoffs) t1 = time.perf_counter() print("循环版:", round(t1 - t0, 3), "秒,价格:", round(price1, 4)) # 输出示例:循环版: 1.946 秒,价格: 0.9217

第二步找循环体里的依赖:每条路径内部有状态传递(s 逐步相乘),路径之间完全独立。独立方向先向量化,路径内的累积交给矩阵乘:

# 向量化版:所有路径同步推进 t0 = time.perf_counter() steps = 1.0 + 0.0001 + rng.normal(0, 0.02, size=(n_step, n_path)) paths = 100.0 * np.cumprod(steps, axis=0) # 路径内连乘 avg_price = paths.mean(axis=0) # 路径内平均 price2 = np.clip(avg_price - 100.0, 0, None).mean() t1 = time.perf_counter() print("向量化版:", round(t1 - t0, 3), "秒,价格:", round(price2, 4)) # 输出示例:向量化版: 0.009 秒,价格: 0.9183

结果与解读:约 200 倍提速,两次价格差异来自随机路径不同(0.92 量级一致)。改写的内存代价:steps 与 paths 各 (100, 10000) 约 8MB,账本可控;若路径数再大一个量级,按 8.2 节分块生成即可。变式:对数收益率的路径可把连乘换成 cumsum 再整体 exp,数值更稳。

三张改写模板

模板一,逐元素条件分支换 where 或 clip:

import numpy as np x = np.random.rand(1_000_000) # 反面:for + if # 正面: y1 = np.where(x > 0.5, 1.0, 0.0) y2 = np.clip(x, 0.2, 0.8) # 截断类分支更用 clip

模板二,双重循环换广播(5.2 节距离案例的抽象):

q = np.random.rand(5, 3) r = np.random.rand(8, 3) # 反面:for i for j 算 dist[i,j] # 正面: dist = np.linalg.norm(q[:, None, :] - r[None, :, :], axis=2) print(dist.shape) # (5, 8)

模板三,循环收集换一次聚合:

values = np.random.rand(1000, 100) # 反面:for row: out.append(row.mean()) # 正面: out = values.mean(axis=1) print(out.shape) # (1000,)

性能排查清单

性能排查清单

验收方法:计时三件套

改写必须以测量收尾,凭感觉的优化是玄学:

import numpy as np import time def bench(fn, *args, repeat=5): times = [] for _ in range(repeat): t0 = time.perf_counter() fn(*args) times.append(time.perf_counter() - t0) return min(times) # 取最小值最能代表纯计算能力 a = np.random.rand(2_000_000) print("sum 耗时:", round(bench(lambda x: x.sum(), a) * 1000, 3), "毫秒") # 输出示例:sum 耗时: 0.812 毫秒

取 min 是为排除操作系统抖动与垃圾回收的干扰;要更严谨可换专业基准测试库,但手写三件套(perf_counter、多次重复、取最优)已覆盖日常需求。改写前后各测一次,比值就是真实收益,写进注释留给下一位读者。

💡 关键直觉:性能问题很少是"NumPy 不够快",几乎总是"代码没按 NumPy 的方式写"。清单过一遍,九成慢代码当场现形;剩下的一成才是算法与库的边界问题。

反面案例:向量化不是永远赢

保持诚实,也看一个向量化反而吃亏的场景,理解工具的边界:

import numpy as np import time # 场景:一亿个元素里只想找第一个超过阈值的下标,找到就停 rng = np.random.default_rng(12) big = rng.rand(100_000_000) big[-1] = 2.0 # 把目标藏在最后,最坏情况 # 向量化版:必须扫完全部一亿个 t0 = time.perf_counter() idx1 = np.argmax(big > 1.0) t1 = time.perf_counter() print("向量化:", round(t1 - t0, 3), "秒,下标:", idx1) # 约 0.25 秒 # 循环版:逐块找,找到即提前退出 t0 = time.perf_counter() idx2 = None step = 1_000_000 for start in range(0, big.size, step): hits = np.flatnonzero(big[start:start + step] > 1.0) if hits.size: idx2 = start + hits[0] break t1 = time.perf_counter() print("分块早停:", round(t1 - t0, 3), "秒,下标:", idx2) # 目标靠前时毫秒级

结果与解读:目标在末尾时两者相当;目标一旦靠前,分块早停版立刻反超几个数量级。教训是向量化摊薄的是"每元素成本",而提前退出省的是"总元素数"——两者属于不同的优化维度,真实工程里经常组合使用。清单第一项因此要补一句注脚:流程上有提前退出语义时,分块加早停也是合格的"向量化思维"。

本节要点回顾

  • 改写流程:先找循环体内的独立性方向,独立维度整体向量化,路径内依赖交给 cumprod、cumsum 或矩阵乘
  • 三张模板:条件分支换 where/clip、双重循环换广播、循环收集换一次聚合
  • 十二项清单:代码层五项、内存层五项、进出层两项,每项对应前文一章
  • 验收纪律:perf_counter 加重复取最小值,改一条测一条,收益写进注释
  • 全册主线回顾:从第 1 章的连续内存,到 strides、视图拷贝、广播、账本,最终都汇入"预判每个操作的代价"这一种判断力

恭喜走完全册。回头看导读的知识地图,每一条连线现在都应该是你已经亲手验证过的结论。


作者与出处
原作者: 灏天文库
来源:灏天文库
整理: 灏天文库整理
由灏天文库平台收录,内容或由平台用户上传,仅供学习交流
发布者: 作者: 灏天文库 转发
评论区 (0)
U