NumPy: 性能优化

Bob 的数据分析脚本跑了整整 10 分钟。Alice 检查后做了三个关键优化:循环→向量化、append→预分配、C→F 顺序(按列操作),运行时间降到 15 秒——40 倍加速,代码还更短了。本章将系统讲解 NumPy 性能优化的核心方法。

1. 你将学到


2. 向量化原则

(1) 为什么循环慢

Python 的 for 循环每次迭代都需要解释器进行类型检查和函数调度。NumPy 的向量化操作将循环推入底层 C 代码,一次调用完成全部计算。

对比维度 Python 循环 NumPy 向量化
执行位置 Python 解释器 C/Fortran 底层
逐元素调度 每次迭代均有开销 一次性批量处理
类型检查 每次迭代 仅入口一次
缓存友好 差(散列访问) 好(连续内存)
代码量

▶ 示例

TEXT 📖 仅展示
> **输出:** 在本地 Python 环境运行 NumPy 2.x,输出 ndarray 数组内容。Piston 服务器未预装 NumPy,请在本机安装(`pip install numpy`)后实操对照。实际数值可能因 NumPy 版本、随机种子略有差异。

:循环 vs 向量化(难度⭐)

PYTHON
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
TEXT 📖 仅展示
> **输出:** 在本地 Python 环境运行 NumPy 2.x,输出 ndarray 数组内容。Piston 服务器未预装 NumPy,请在本机安装(`pip install numpy`)后实操对照。实际数值可能因 NumPy 版本、随机种子略有差异。

(2) 向量化三原则

  1. 优先使用 NumPy 内建函数np.sumnp.dotnp.mean 等已经过高度优化
  2. 用布尔索引替代条件循环a[a > 0] 代替逐元素判断
  3. 用广播替代手动扩展:让 NumPy 自动处理不同形状

▶ 示例

TEXT 📖 仅展示
> **输出:** 在本地 Python 环境运行 NumPy 2.x,输出 ndarray 数组内容。Piston 服务器未预装 NumPy,请在本机安装(`pip install numpy`)后实操对照。实际数值可能因 NumPy 版本、随机种子略有差异。

:布尔索引 vs 条件循环(难度⭐)

PYTHON
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
TEXT 📖 仅展示
> **输出:** 在本地 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)

▶ 示例

TEXT 📖 仅展示
> **输出:** 在本地 Python 环境运行 NumPy 2.x,输出 ndarray 数组内容。Piston 服务器未预装 NumPy,请在本机安装(`pip install numpy`)后实操对照。实际数值可能因 NumPy 版本、随机种子略有差异。

:就地操作节省内存(难度⭐)

PYTHON
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)
TEXT 📖 仅展示
> **输出:** 在本地 Python 环境运行 NumPy 2.x,输出 ndarray 数组内容。Piston 服务器未预装 NumPy,请在本机安装(`pip install numpy`)后实操对照。实际数值可能因 NumPy 版本、随机种子略有差异。

(2) 视图 vs 副本

切片返回视图(不复制数据),而高级索引返回副本。能用切片时不要用高级索引。

PYTHON
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)
TEXT 📖 仅展示
> **输出:** 在本地 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 风格
典型场景 按行处理图像、时间序列 按列处理统计、线性代数

▶ 示例

TEXT 📖 仅展示
> **输出:** 在本地 Python 环境运行 NumPy 2.x,输出 ndarray 数组内容。Piston 服务器未预装 NumPy,请在本机安装(`pip install numpy`)后实操对照。实际数值可能因 NumPy 版本、随机种子略有差异。

:C 顺序 vs F 顺序按列求和(难度⭐⭐)

PYTHON
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
TEXT 📖 仅展示
> **输出:** 在本地 Python 环境运行 NumPy 2.x,输出 ndarray 数组内容。Piston 服务器未预装 NumPy,请在本机安装(`pip install numpy`)后实操对照。实际数值可能因 NumPy 版本、随机种子略有差异。

(2) ascontiguousarray / asfortranarray

当你需要对特定方向反复操作时,可以显式转换内存布局:

PYTHON
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
TEXT 📖 仅展示
> **输出:** 在本地 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 次
数据复制 每次复制全部 无额外复制
代码复杂度 看似简单 需提前知道大小

▶ 示例

TEXT 📖 仅展示
> **输出:** 在本地 Python 环境运行 NumPy 2.x,输出 ndarray 数组内容。Piston 服务器未预装 NumPy,请在本机安装(`pip install numpy`)后实操对照。实际数值可能因 NumPy 版本、随机种子略有差异。

:预分配 vs append(难度⭐)

PYTHON
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
TEXT 📖 仅展示
> **输出:** 在本地 Python 环境运行 NumPy 2.x,输出 ndarray 数组内容。Piston 服务器未预装 NumPy,请在本机安装(`pip install numpy`)后实操对照。实际数值可能因 NumPy 版本、随机种子略有差异。

(2) 不知道最终大小时怎么办

如果无法预知元素数量,可以先用 Python list 收集,最后一次性转换:

PYTHON
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,)
TEXT 📖 仅展示
> **输出:** 在本地 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) 内积

▶ 示例

TEXT 📖 仅展示
> **输出:** 在本地 Python 环境运行 NumPy 2.x,输出 ndarray 数组内容。Piston 服务器未预装 NumPy,请在本机安装(`pip install numpy`)后实操对照。实际数值可能因 NumPy 版本、随机种子略有差异。

:einsum 常见用法(难度⭐⭐)

PYTHON
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
TEXT 📖 仅展示
> **输出:** 在本地 Python 环境运行 NumPy 2.x,输出 ndarray 数组内容。Piston 服务器未预装 NumPy,请在本机安装(`pip install numpy`)后实操对照。实际数值可能因 NumPy 版本、随机种子略有差异。

(2) einsum 的优势

当多个操作可以合并为一个 einsum 表达式时,避免了中间数组的创建:

PYTHON
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
TEXT 📖 仅展示
> **输出:** 在本地 Python 环境运行 NumPy 2.x,输出 ndarray 数组内容。Piston 服务器未预装 NumPy,请在本机安装(`pip install numpy`)后实操对照。实际数值可能因 NumPy 版本、随机种子略有差异。

7. %timeit 性能测量

(1) 使用 %timeit 测量代码执行时间

%timeit 是 Jupyter/IPython 的魔法命令,自动多次运行取平均,排除偶然波动。

PYTHON
# 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")
TEXT 📖 仅展示
> **输出:** 在本地 Python 环境运行 NumPy 2.x,输出 ndarray 数组内容。Piston 服务器未预装 NumPy,请在本机安装(`pip install numpy`)后实操对照。实际数值可能因 NumPy 版本、随机种子略有差异。

▶ 示例

TEXT 📖 仅展示
> **输出:** 在本地 Python 环境运行 NumPy 2.x,输出 ndarray 数组内容。Piston 服务器未预装 NumPy,请在本机安装(`pip install numpy`)后实操对照。实际数值可能因 NumPy 版本、随机种子略有差异。

:对比各方法执行时间(难度⭐⭐)

PYTHON
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")
TEXT 📖 仅展示
> **输出:** 在本地 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. 性能优化检查清单

100%
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!]
TEXT 📖 仅展示
> **输出:** 在本地 Python 环境运行 NumPy 2.x,输出 ndarray 数组内容。Piston 服务器未预装 NumPy,请在本机安装(`pip install numpy`)后实操对照。实际数值可能因 NumPy 版本、随机种子略有差异。

10. 综合示例:优化股票投资组合脚本

▶ 示例

TEXT 📖 仅展示
> **输出:** 在本地 Python 环境运行 NumPy 2.x,输出 ndarray 数组内容。Piston 服务器未预装 NumPy,请在本机安装(`pip install numpy`)后实操对照。实际数值可能因 NumPy 版本、随机种子略有差异。

:从循环到 einsum——50 倍加速(难度⭐⭐⭐)

Charlie 写了一个投资组合分析脚本,计算 1000 支股票在 252 个交易日的组合收益和风险。原始版本用循环 + append,运行缓慢。

PYTHON
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}")
TEXT 📖 仅展示
> **输出:** 在本地 Python 环境运行 NumPy 2.x,输出 ndarray 数组内容。Piston 服务器未预装 NumPy,请在本机安装(`pip install numpy`)后实操对照。实际数值可能因 NumPy 版本、随机种子略有差异。

❓ 常见问题

Q 向量化一定比循环快吗?
A 绝大多数情况是的。但当数组极小(如 < 10 个元素)或操作本身无法向量化(如每个元素依赖前一个元素的计算结果)时,循环可能相当甚至更快,因为 NumPy 函数调用本身有固定开销。
Q einsum 比普通矩阵乘快多少?
A 对于简单矩阵乘法(如 A @ B),einsum 与 @ 性能接近。einsum 的优势在于合并多步操作(避免中间数组)和表达复杂的多维收缩。对于三矩阵连乘,einsum 可比两步 @ 快 1.5-3 倍。
Q 什么时候需要考虑 C/F 顺序?
A 当数组较大(> 10000 元素)且你主要沿某个轴反复操作时。按行遍历用 C 顺序(默认),按列遍历用 F 顺序。小数组差异不明显,不必纠结。
Q 除了本章方法,还能更快吗?
A 可以。进一步优化手段包括:使用 Numba JIT 编译、Cython 编写关键路径、多线程(np.dot 自动调用 BLAS 多线程)、GPU 加速(CuPy)、以及选择更优算法(如用 np.searchsorted 替代循环查找)。
Q 如何找到代码的性能瓶颈?
A 用 %timeit 测量各步骤耗时,或用 cProfile / line_profiler 获取逐行耗时。先用 %timeit 定位慢的代码段,再针对性优化,不要盲目优化。
Q np.empty 比 np.zeros 快多少?值得用吗?
A np.empty 跳过初始化,比 zeros 快约 2-5 倍。但如果你会立刻填入数据,用 empty 节省的时间通常可忽略。如果数组很大且不会立刻全部赋值,empty 可能残留旧值导致隐蔽 bug,需谨慎。

📖 小节


📝 作业

  1. 基础题(难度⭐):将以下循环代码改写为向量化版本,并用 time.perf_counter 测量加速比。
PYTHON
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]
TEXT 📖 仅展示
> **输出:** 在本地 Python 环境运行 NumPy 2.x,输出 ndarray 数组内容。Piston 服务器未预装 NumPy,请在本机安装(`pip install numpy`)后实操对照。实际数值可能因 NumPy 版本、随机种子略有差异。
  1. 进阶题(难度⭐⭐):给定矩阵 A(形状 100x50)和矩阵 B(形状 50x80),用 np.einsum 分别实现:矩阵乘法、A 的行求和、A 与 B 的逐元素乘后再列求和。对比 einsum 与标准 NumPy 写法的执行时间。

  2. 挑战题(难度⭐⭐⭐):找到你之前写过的一个使用循环 + append 的 NumPy 脚本,应用本章所有优化技巧(向量化、预分配、内存布局、einsum),记录优化前后的执行时间和内存占用变化。

Web-Tutorial.com

Web-Tutorial 技术团队

由多位开发者共同维护的编程教程平台。每篇教程由对应领域的开发者编写和审核,确保内容准确可靠。如发现任何问题,欢迎向我们反馈。

100%

🙏 帮我们做得更好

我们是刚上线的编程教程站,几个人的小团队,精力有限。页面虽经检查,难免还有疏漏——链接失效、排版错乱、内容有误、语言生硬……

如果您发现了,麻烦告诉我们,我们会在收到反馈后第一时间进行修复,再次感谢您的光临 🙏