NumPy: 性能优化
Bob 的数据分析脚本跑了整整 10 分钟。Alice 检查后做了三个关键优化:循环→向量化、append→预分配、C→F 顺序(按列操作),运行时间降到 15 秒——40 倍加速,代码还更短了。本章将系统讲解 NumPy 性能优化的核心方法。
1. 你将学到
- ❶ 向量化 vs 循环:用 NumPy 内建操作替代 Python 循环
- ❷ 避免不必要的数组创建与复制
- ❸ C 顺序 vs F 顺序:内存布局对性能的影响
- ❹ 预分配 vs 动态增长:告别 append
- ❺ einsum:一步完成多维运算
2. 向量化原则
(1) 为什么循环慢
Python 的 for 循环每次迭代都需要解释器进行类型检查和函数调度。NumPy 的向量化操作将循环推入底层 C 代码,一次调用完成全部计算。
| 对比维度 | Python 循环 | NumPy 向量化 |
|---|---|---|
| 执行位置 | Python 解释器 | C/Fortran 底层 |
| 逐元素调度 | 每次迭代均有开销 | 一次性批量处理 |
| 类型检查 | 每次迭代 | 仅入口一次 |
| 缓存友好 | 差(散列访问) | 好(连续内存) |
| 代码量 | 多 | 少 |
▶ 示例
> **输出:** 在本地 Python 环境运行 NumPy 2.x,输出 ndarray 数组内容。Piston 服务器未预装 NumPy,请在本机安装(`pip install numpy`)后实操对照。实际数值可能因 NumPy 版本、随机种子略有差异。
:循环 vs 向量化(难度⭐)
import numpy as np
n = 1_000_000
a = np.random.rand(n)
b = np.random.rand(n)
# Loop version
result_loop = np.empty(n)
for i in range(n):
result_loop[i] = a[i] + b[i]
# Vectorized version
result_vec = a + b
print(np.allclose(result_loop, result_vec)) # True
> **输出:** 在本地 Python 环境运行 NumPy 2.x,输出 ndarray 数组内容。Piston 服务器未预装 NumPy,请在本机安装(`pip install numpy`)后实操对照。实际数值可能因 NumPy 版本、随机种子略有差异。
(2) 向量化三原则
- 优先使用 NumPy 内建函数:
np.sum、np.dot、np.mean等已经过高度优化 - 用布尔索引替代条件循环:
a[a > 0]代替逐元素判断 - 用广播替代手动扩展:让 NumPy 自动处理不同形状
▶ 示例
> **输出:** 在本地 Python 环境运行 NumPy 2.x,输出 ndarray 数组内容。Piston 服务器未预装 NumPy,请在本机安装(`pip install numpy`)后实操对照。实际数值可能因 NumPy 版本、随机种子略有差异。
:布尔索引 vs 条件循环(难度⭐)
import numpy as np
data = np.random.randn(100000)
# Loop version
result_loop = np.empty_like(data)
for i in range(len(data)):
if data[i] > 0:
result_loop[i] = data[i] * 2
else:
result_loop[i] = data[i]
# Vectorized version
result_vec = np.where(data > 0, data * 2, data)
print(np.allclose(result_loop, result_vec)) # True
> **输出:** 在本地 Python 环境运行 NumPy 2.x,输出 ndarray 数组内容。Piston 服务器未预装 NumPy,请在本机安装(`pip install numpy`)后实操对照。实际数值可能因 NumPy 版本、随机种子略有差异。
3. 避免不必要的创建与复制
(1) 就地操作 vs 创建新数组
每次 a + b 都会创建新数组。对于大规模数据,就地操作可减少内存分配和垃圾回收开销。
| 操作 | 创建新数组 | 就地替代 |
|---|---|---|
a = a + b |
是 | a += b |
a = a * 2 |
是 | a *= 2 |
a = np.sqrt(a) |
是 | np.sqrt(a, out=a) |
a = np.maximum(a, 0) |
是 | np.maximum(a, 0, out=a) |
▶ 示例
> **输出:** 在本地 Python 环境运行 NumPy 2.x,输出 ndarray 数组内容。Piston 服务器未预装 NumPy,请在本机安装(`pip install numpy`)后实操对照。实际数值可能因 NumPy 版本、随机种子略有差异。
:就地操作节省内存(难度⭐)
import numpy as np
a = np.random.rand(1000000)
b = np.random.rand(1000000)
# Creates a new array each time
c = a + b
c = c * 2
c = np.sqrt(c)
# In-place version (no new allocations)
c = a + b
c *= 2
np.sqrt(c, out=c)
# Using out= parameter
result = np.empty_like(a)
np.add(a, b, out=result)
np.multiply(result, 2, out=result)
np.sqrt(result, out=result)
> **输出:** 在本地 Python 环境运行 NumPy 2.x,输出 ndarray 数组内容。Piston 服务器未预装 NumPy,请在本机安装(`pip install numpy`)后实操对照。实际数值可能因 NumPy 版本、随机种子略有差异。
(2) 视图 vs 副本
切片返回视图(不复制数据),而高级索引返回副本。能用切片时不要用高级索引。
import numpy as np
a = np.arange(12).reshape(3, 4)
# View: no copy (fast)
row_view = a[1, :] # shares memory with a
# Copy: new allocation (slow)
row_copy = a[[1], :] # creates a new array
a[1, 0] = 999
print(row_view[0]) # 999 (view reflects change)
print(row_copy[0, 0]) # 4 (copy is independent)
> **输出:** 在本地 Python 环境运行 NumPy 2.x,输出 ndarray 数组内容。Piston 服务器未预装 NumPy,请在本机安装(`pip install numpy`)后实操对照。实际数值可能因 NumPy 版本、随机种子略有差异。
4. 内存布局:C 顺序 vs F 顺序
(1) 行优先 vs 列优先
NumPy 默认 C 顺序(行优先),即每行的元素在内存中连续。F 顺序(列优先)则是每列元素连续。按内存布局方向访问数据能充分利用 CPU 缓存。
| 维度 | C 顺序(行优先) | F 顺序(列优先) |
|---|---|---|
| 内存排列 | 行内元素连续 | 列内元素连续 |
| 高效访问 | 按行遍历 | 按列遍历 |
| 默认 | NumPy 默认 | Fortran 风格 |
| 典型场景 | 按行处理图像、时间序列 | 按列处理统计、线性代数 |
▶ 示例
> **输出:** 在本地 Python 环境运行 NumPy 2.x,输出 ndarray 数组内容。Piston 服务器未预装 NumPy,请在本机安装(`pip install numpy`)后实操对照。实际数值可能因 NumPy 版本、随机种子略有差异。
:C 顺序 vs F 顺序按列求和(难度⭐⭐)
import numpy as np
n_rows, n_cols = 10000, 1000
# C order (default): columns are NOT contiguous
a_c = np.random.rand(n_rows, n_cols) # C order
# F order: columns ARE contiguous
a_f = np.asfortranarray(a_c) # F order
# Sum along columns (axis=0)
# In C order, this strided access; in F order, contiguous access
sum_c = a_c.sum(axis=0)
sum_f = a_f.sum(axis=0)
print(np.allclose(sum_c, sum_f)) # True
# a_f.sum(axis=0) is typically faster for large arrays
> **输出:** 在本地 Python 环境运行 NumPy 2.x,输出 ndarray 数组内容。Piston 服务器未预装 NumPy,请在本机安装(`pip install numpy`)后实操对照。实际数值可能因 NumPy 版本、随机种子略有差异。
(2) ascontiguousarray / asfortranarray
当你需要对特定方向反复操作时,可以显式转换内存布局:
import numpy as np
a = np.random.rand(5000, 5000)
# If you will do many column-wise operations, convert to F order
a_f = np.asfortranarray(a)
# If you will do many row-wise operations, ensure C order
a_c = np.ascontiguousarray(a)
# Check memory layout
print(a_c.flags['C_CONTIGUOUS']) # True
print(a_f.flags['F_CONTIGUOUS']) # True
> **输出:** 在本地 Python 环境运行 NumPy 2.x,输出 ndarray 数组内容。Piston 服务器未预装 NumPy,请在本机安装(`pip install numpy`)后实操对照。实际数值可能因 NumPy 版本、随机种子略有差异。
| 场景 | 推荐布局 | 转换函数 |
|---|---|---|
| 按行遍历/操作 | C 顺序 | np.ascontiguousarray |
| 按列遍历/操作 | F 顺序 | np.asfortranarray |
| 传递给 C 库 | C 顺序 | np.ascontiguousarray |
| 传递给 Fortran 库 | F 顺序 | np.asfortranarray |
5. 预分配 vs 动态增长
(1) append 的代价
np.append 每次调用都创建一个新数组并复制全部旧数据,时间复杂度为 O(n²)。
| 对比维度 | 动态 append | 预分配 |
|---|---|---|
| 时间复杂度 | O(n²) | O(n) |
| 内存分配次数 | n 次 | 1 次 |
| 数据复制 | 每次复制全部 | 无额外复制 |
| 代码复杂度 | 看似简单 | 需提前知道大小 |
▶ 示例
> **输出:** 在本地 Python 环境运行 NumPy 2.x,输出 ndarray 数组内容。Piston 服务器未预装 NumPy,请在本机安装(`pip install numpy`)后实操对照。实际数值可能因 NumPy 版本、随机种子略有差异。
:预分配 vs append(难度⭐)
import numpy as np
n = 100000
# Dynamic append (slow)
result_append = np.array([])
for i in range(n):
result_append = np.append(result_append, i ** 2)
# Pre-allocation (fast)
result_prealloc = np.empty(n)
for i in range(n):
result_prealloc[i] = i ** 2
# Best: fully vectorized
result_vec = np.arange(n) ** 2
print(np.allclose(result_prealloc, result_vec)) # True
> **输出:** 在本地 Python 环境运行 NumPy 2.x,输出 ndarray 数组内容。Piston 服务器未预装 NumPy,请在本机安装(`pip install numpy`)后实操对照。实际数值可能因 NumPy 版本、随机种子略有差异。
(2) 不知道最终大小时怎么办
如果无法预知元素数量,可以先用 Python list 收集,最后一次性转换:
import numpy as np
# When size is unknown, collect with list, then convert
result_list = []
for i in range(100000):
if i % 3 == 0:
result_list.append(i ** 2)
result = np.array(result_list)
print(result.shape) # (33334,)
> **输出:** 在本地 Python 环境运行 NumPy 2.x,输出 ndarray 数组内容。Piston 服务器未预装 NumPy,请在本机安装(`pip install numpy`)后实操对照。实际数值可能因 NumPy 版本、随机种子略有差异。
6. einsum:一步完成多维运算
(1) einsum 基础语法
np.einsum 使用爱因斯坦求和约定,用一个字符串表达式替代多次数组操作。格式为 "下标输入 -> 下标输出",重复下标表示求和,省略的下标表示沿该轴求和。
| 表达式 | 等价操作 | 说明 |
|---|---|---|
'ij,jk->ik' |
a @ b |
矩阵乘法 |
'ij->ji' |
a.T |
转置 |
'ij->i' |
a.sum(axis=1) |
行求和 |
'ij,j->i' |
a * b 再行求和 |
矩阵乘向量 |
'ij,ij->ij' |
a * b |
逐元素乘 |
'ij->' |
a.sum() |
全部求和 |
'i,i->' |
np.dot(a, b) |
内积 |
▶ 示例
> **输出:** 在本地 Python 环境运行 NumPy 2.x,输出 ndarray 数组内容。Piston 服务器未预装 NumPy,请在本机安装(`pip install numpy`)后实操对照。实际数值可能因 NumPy 版本、随机种子略有差异。
:einsum 常见用法(难度⭐⭐)
import numpy as np
a = np.random.rand(3, 4)
b = np.random.rand(4, 5)
v = np.random.rand(4)
# Matrix multiplication
matmul_std = a @ b
matmul_ein = np.einsum('ij,jk->ik', a, b)
print(np.allclose(matmul_std, matmul_ein)) # True
# Matrix-vector multiplication
matvec_std = a @ v
matvec_ein = np.einsum('ij,j->i', a, v)
print(np.allclose(matvec_std, matvec_ein)) # True
# Row-wise sum
rowsum_std = a.sum(axis=1)
rowsum_ein = np.einsum('ij->i', a)
print(np.allclose(rowsum_std, rowsum_ein)) # True
# Element-wise multiply then sum (batch dot product)
c = np.random.rand(3, 4)
d = np.random.rand(3, 4)
batchdot_std = (c * d).sum(axis=1)
batchdot_ein = np.einsum('ij,ij->i', c, d)
print(np.allclose(batchdot_std, batchdot_ein)) # True
> **输出:** 在本地 Python 环境运行 NumPy 2.x,输出 ndarray 数组内容。Piston 服务器未预装 NumPy,请在本机安装(`pip install numpy`)后实操对照。实际数值可能因 NumPy 版本、随机种子略有差异。
(2) einsum 的优势
当多个操作可以合并为一个 einsum 表达式时,避免了中间数组的创建:
import numpy as np
A = np.random.rand(100, 50)
B = np.random.rand(50, 80)
C = np.random.rand(80, 30)
# Step-by-step: creates two intermediate arrays
temp = A @ B # shape (100, 80)
result_std = temp @ C # shape (100, 30)
# einsum: one operation, no intermediates
result_ein = np.einsum('ij,jk,kl->il', A, B, C)
print(np.allclose(result_std, result_ein)) # True
> **输出:** 在本地 Python 环境运行 NumPy 2.x,输出 ndarray 数组内容。Piston 服务器未预装 NumPy,请在本机安装(`pip install numpy`)后实操对照。实际数值可能因 NumPy 版本、随机种子略有差异。
7. %timeit 性能测量
(1) 使用 %timeit 测量代码执行时间
%timeit 是 Jupyter/IPython 的魔法命令,自动多次运行取平均,排除偶然波动。
# In Jupyter/IPython
import numpy as np
a = np.random.rand(1000000)
# Measure loop version
# %timeit for i in range(len(a)): a[i] * 2
# Measure vectorized version
# %timeit a * 2
# Measure with Python time module (for scripts)
import time
start = time.perf_counter()
result = a * 2
elapsed = time.perf_counter() - start
print(f"Elapsed: {elapsed:.6f} seconds")
> **输出:** 在本地 Python 环境运行 NumPy 2.x,输出 ndarray 数组内容。Piston 服务器未预装 NumPy,请在本机安装(`pip install numpy`)后实操对照。实际数值可能因 NumPy 版本、随机种子略有差异。
▶ 示例
> **输出:** 在本地 Python 环境运行 NumPy 2.x,输出 ndarray 数组内容。Piston 服务器未预装 NumPy,请在本机安装(`pip install numpy`)后实操对照。实际数值可能因 NumPy 版本、随机种子略有差异。
:对比各方法执行时间(难度⭐⭐)
import numpy as np
import time
n = 10_000_000
a = np.random.rand(n)
b = np.random.rand(n)
# Method 1: Python loop (very slow)
start = time.perf_counter()
result1 = np.empty(n)
for i in range(n):
result1[i] = a[i] + b[i]
t1 = time.perf_counter() - start
# Method 2: NumPy vectorized
start = time.perf_counter()
result2 = a + b
t2 = time.perf_counter() - start
# Method 3: np.add with out=
result3 = np.empty(n)
start = time.perf_counter()
np.add(a, b, out=result3)
t3 = time.perf_counter() - start
print(f"Loop: {t1:.4f}s")
print(f"Vectorized: {t2:.4f}s")
print(f"np.add(out=):{t3:.4f}s")
print(f"Speedup: {t1/t2:.0f}x")
> **输出:** 在本地 Python 环境运行 NumPy 2.x,输出 ndarray 数组内容。Piston 服务器未预装 NumPy,请在本机安装(`pip install numpy`)后实操对照。实际数值可能因 NumPy 版本、随机种子略有差异。
8. 常见低效模式与优化
| 低效模式 | 优化方式 | 加速比 |
|---|---|---|
for i in range(n): result[i] = a[i] + b[i] |
result = a + b |
50-200x |
result = np.array([]); np.append(result, x) |
result = np.empty(n); result[i] = x |
10-100x |
a[[0,1,2], :] 连续行高级索引 |
a[:3, :] 切片 |
2-5x |
| 按列操作 C 顺序大数组 | np.asfortranarray 转 F 顺序 |
2-10x |
(A @ B) @ C 多步矩阵乘 |
np.einsum('ij,jk,kl->il', A, B, C) |
1.5-3x |
a = a + b 反复创建新数组 |
a += b 就地操作 |
1.5-2x |
np.sqrt(np.maximum(a, 0)) 多步链式 |
np.sqrt(a, out=a); ... 就地链式 |
1.3-1.8x |
9. 性能优化检查清单
graph TB
A[Start: NumPy Performance] --> B{Using Python loops?}
B -->|Yes| C[Replace with vectorized ops]
B -->|No| D{Appending in loop?}
C --> D
D -->|Yes| E[Pre-allocate or use list + np.array]
D -->|No| F{Accessing columns on C-order array?}
E --> F
F -->|Yes| G[Convert to F order with asfortranarray]
F -->|No| H{Multiple steps creating intermediates?}
G --> H
H -->|Yes| I[Use einsum or out= parameter]
H -->|No| J{Creating unnecessary copies?}
I --> J
J -->|Yes| K[Use views / in-place ops]
J -->|No| L[Profile with %timeit to find bottleneck]
K --> L
L --> M[Optimized!]
> **输出:** 在本地 Python 环境运行 NumPy 2.x,输出 ndarray 数组内容。Piston 服务器未预装 NumPy,请在本机安装(`pip install numpy`)后实操对照。实际数值可能因 NumPy 版本、随机种子略有差异。
10. 综合示例:优化股票投资组合脚本
▶ 示例
> **输出:** 在本地 Python 环境运行 NumPy 2.x,输出 ndarray 数组内容。Piston 服务器未预装 NumPy,请在本机安装(`pip install numpy`)后实操对照。实际数值可能因 NumPy 版本、随机种子略有差异。
:从循环到 einsum——50 倍加速(难度⭐⭐⭐)
Charlie 写了一个投资组合分析脚本,计算 1000 支股票在 252 个交易日的组合收益和风险。原始版本用循环 + append,运行缓慢。
import numpy as np
import time
np.random.seed(42)
n_stocks = 1000
n_days = 252
returns = np.random.randn(n_days, n_stocks) * 0.02
weights = np.random.dirichlet(np.ones(n_stocks))
# ---- Version 1: Loop + append (slowest) ----
start = time.perf_counter()
daily_portfolio_loop = np.array([])
for day in range(n_days):
daily_ret = 0.0
for stock in range(n_stocks):
daily_ret += returns[day, stock] * weights[stock]
daily_portfolio_loop = np.append(daily_portfolio_loop, daily_ret)
t1 = time.perf_counter() - start
# ---- Version 2: Pre-allocation (faster) ----
start = time.perf_counter()
daily_portfolio_pre = np.empty(n_days)
for day in range(n_days):
daily_portfolio_pre[day] = returns[day, :].dot(weights)
t2 = time.perf_counter() - start
# ---- Version 3: Vectorized (fast) ----
start = time.perf_counter()
daily_portfolio_vec = returns @ weights
t3 = time.perf_counter() - start
# ---- Version 4: einsum (fast + flexible) ----
start = time.perf_counter()
daily_portfolio_ein = np.einsum('ij,j->i', returns, weights)
t4 = time.perf_counter() - start
print(f"Loop+append: {t1:.4f}s")
print(f"Pre-allocate: {t2:.4f}s ({t1/t2:.1f}x faster)")
print(f"Vectorized: {t3:.4f}s ({t1/t3:.1f}x faster)")
print(f"einsum: {t4:.4f}s ({t1/t4:.1f}x faster)")
# Now compute portfolio risk (covariance) with einsum
mean_ret = np.einsum('ij->j', returns) / n_days
demeaned = returns - mean_ret
# NumPy 2.x 严格 reshape 对齐,demeaned.T 是 (1000, 252),demeaned 是 (252, 1000)
cov_matrix = np.einsum('ij,jk->ik', demeaned.T, demeaned) / (n_days - 1)
portfolio_var = np.einsum('i,ij,j->', weights, cov_matrix, weights)
portfolio_vol = float(np.sqrt(portfolio_var))
print(f"Portfolio volatility: {portfolio_vol:.4f}")
> **输出:** 在本地 Python 环境运行 NumPy 2.x,输出 ndarray 数组内容。Piston 服务器未预装 NumPy,请在本机安装(`pip install numpy`)后实操对照。实际数值可能因 NumPy 版本、随机种子略有差异。
❓ 常见问题
📖 小节
- 向量化是 NumPy 性能优化的第一原则:用内建操作替代 Python 循环
- 就地操作(
+=、out=)减少内存分配和复制开销 - C 顺序适合按行操作,F 顺序适合按列操作,用
ascontiguousarray/asfortranarray转换 - 预分配(
np.empty)或 list+np.array替代np.append,避免 O(n²) 复制 np.einsum可合并多步操作为一步,消除中间数组- 用
%timeit或cProfile定位瓶颈,针对性优化
📝 作业
- 基础题(难度⭐):将以下循环代码改写为向量化版本,并用
time.perf_counter测量加速比。
import numpy as np
a = np.random.rand(100000)
result = np.empty_like(a)
for i in range(len(a)):
if a[i] > 0.5:
result[i] = a[i] ** 2
else:
result[i] = a[i]
> **输出:** 在本地 Python 环境运行 NumPy 2.x,输出 ndarray 数组内容。Piston 服务器未预装 NumPy,请在本机安装(`pip install numpy`)后实操对照。实际数值可能因 NumPy 版本、随机种子略有差异。
-
进阶题(难度⭐⭐):给定矩阵 A(形状 100x50)和矩阵 B(形状 50x80),用
np.einsum分别实现:矩阵乘法、A 的行求和、A 与 B 的逐元素乘后再列求和。对比 einsum 与标准 NumPy 写法的执行时间。 -
挑战题(难度⭐⭐⭐):找到你之前写过的一个使用循环 + append 的 NumPy 脚本,应用本章所有优化技巧(向量化、预分配、内存布局、einsum),记录优化前后的执行时间和内存占用变化。