基础教程

🧮 一、NumPy 是什么？

NumPy（Numerical Python）是 Python 科学计算的基础库，提供：

• 高性能的 N 维数组对象（ndarray）
• 广播（Broadcasting）机制
• 丰富的数学函数（线性代数、傅里叶变换、随机数等）
• 与 C/C++/Fortran 的高效集成

💡 为什么用 NumPy？
比 Python 原生列表快 10~100 倍，内存更紧凑，支持向量化操作（无需 for 循环）。

🚀 二、基础操作实战

1. 创建数组

import numpy as np

# 从列表创建
arr = np.array([1, 2, 3, 4])
print(arr)  # [1 2 3 4]

# 特殊数组
zeros = np.zeros((3, 4))        # 3x4 全零矩阵
ones = np.ones((2, 2))          # 全1
eye = np.eye(3)                 # 3x3 单位矩阵
rand = np.random.rand(2, 3)     # 0~1 均匀分布随机数
arange = np.arange(0, 10, 2)    # [0 2 4 6 8]
linspace = np.linspace(0, 1, 5) # [0.   0.25 0.5  0.75 1.  ]

2. 数组属性

a = np.array([[1, 2, 3], [4, 5, 6]])
print(a.shape)   # (2, 3)
print(a.ndim)    # 2
print(a.size)    # 6
print(a.dtype)   # int64

3. 索引与切片

a = np.array([[1, 2, 3], [4, 5, 6], [7, 8, 9]])

# 基础索引
print(a[1, 2])      # 6

# 切片（注意：切片是视图，修改会影响原数组！）
print(a[1:, :2])    # [[4 5], [7 8]]

# 布尔索引
mask = a > 5
print(a[mask])      # [6 7 8 9]

# 花式索引
rows = np.array([0, 2])
cols = np.array([1, 2])
print(a[rows, cols])  # [2 9]

🔧 三、核心操作实战

1. 向量化计算（避免 for 循环！）

# 计算两个向量的点积
a = np.array([1, 2, 3])
b = np.array([4, 5, 6])
dot = np.dot(a, b)  # 或 a @ b → 32

# 元素级运算
c = a + b           # [5 7 9]
d = a * b           # [4 10 18]
e = np.sqrt(a)      # [1.         1.41421356 1.73205081]

2. 广播（Broadcasting）

A = np.array([[1, 2, 3], [4, 5, 6]])  # shape (2,3)
b = np.array([10, 20, 30])            # shape (3,)

# b 自动广播为 (2,3)
C = A + b  # [[11 22 33], [14 25 36]]

3. 聚合与统计

arr = np.random.rand(4, 5)

print(arr.mean())        # 所有元素均值
print(arr.sum(axis=0))   # 每列求和 → shape (5,)
print(arr.max(axis=1))   # 每行最大值 → shape (4,)
print(np.std(arr))       # 标准差

4. 重塑与合并

a = np.arange(12)
b = a.reshape(3, 4)      # 重塑为 3x4

# 合并
x = np.array([[1, 2], [3, 4]])
y = np.array([[5, 6], [7, 8]])
vstack = np.vstack([x, y])  # 垂直堆叠 → 4x2
hstack = np.hstack([x, y])  # 水平堆叠 → 2x4

📊 四、实战案例

案例1：图像灰度化（模拟）

# 模拟一张 100x100 RGB 图像
image = np.random.randint(0, 256, size=(100, 100, 3), dtype=np.uint8)

# 灰度化公式：Y = 0.299*R + 0.587*G + 0.114*B
gray = np.dot(image[..., :3], [0.299, 0.587, 0.114])
print(gray.shape)  # (100, 100)

案例2：标准化数据（机器学习预处理）

# 模拟 1000 个样本，每个样本 5 个特征
data = np.random.randn(1000, 5)

# Z-score 标准化：(x - mean) / std
data_norm = (data - data.mean(axis=0)) / data.std(axis=0)

print(data_norm.mean(axis=0))  # 接近 [0,0,0,0,0]
print(data_norm.std(axis=0))   # 接近 [1,1,1,1,1]

案例3：求解线性方程组

# 求解 Ax = b
# 2x + y = 5
# x - 3y = -1

A = np.array([[2, 1], [1, -3]])
b = np.array([5, -1])

x = np.linalg.solve(A, b)
print(x)  # [2. 1.] → x=2, y=1

案例4：高效计算欧氏距离矩阵

# 100 个 2D 点
points = np.random.rand(100, 2)

# 计算所有点对之间的距离（不用双重 for 循环！）
diff = points[:, np.newaxis, :] - points[np.newaxis, :, :]  # shape (100,100,2)
dist_matrix = np.sqrt(np.sum(diff**2, axis=2))

print(dist_matrix.shape)  # (100, 100)

⚡ 五、性能技巧

📚 六、常用函数速查表

✅ 七、练习题（自测）

1. 创建一个 5x5 矩阵，对角线为 1，其余为 0（不用 eye）。
2. 找出数组 [3, 1, 4, 1, 5, 9, 2, 6] 中前 3 个最大值的索引。
3. 将二维数组每行归一化（使每行和为 1）。
4. 用 NumPy 实现 ReLU 激活函数：f(x) = max(0, x)。
5. 计算两个向量的余弦相似度。

答案可私信获取，或留言讨论！
当然！以下是对 NumPy 实战 的 深度扩展版，涵盖 高级特性、性能优化、实际工程场景、与生态工具集成，以及 常见陷阱与调试技巧，助你从“会用”进阶到“精通”。

中级操作教程

🔍 一、高级数组操作

1. 结构化数组（Structured Arrays）——处理异构数据

适用于类似表格但需高性能的场景（如传感器数据、日志记录）。

# 定义结构：姓名（字符串）、年龄（int）、身高（float）
dt = np.dtype([('name', 'U10'), ('age', 'i4'), ('height', 'f4')])
people = np.array([
    ('Alice', 25, 165.5),
    ('Bob', 30, 180.0),
    ('Charlie', 35, 175.2)
], dtype=dt)

print(people['name'])      # ['Alice' 'Bob' 'Charlie']
print(people[people['age'] > 28])  # 筛选年龄>28的人

💡 替代方案：Pandas 更适合复杂表格，但 NumPy 结构化数组在纯数值+少量字符串场景更快。

2. 广播机制深入理解

广播规则（从后往前对齐维度）：

• 维度为 1 的可拉伸
• 缺失维度自动补 1

A = np.random.rand(3, 4, 5)   # (3,4,5)
B = np.random.rand(4, 1)      # (4,1) → 广播为 (1,4,1) → (3,4,5)

C = A + B  # 成功！

常见错误：

X = np.ones((3, 2))
Y = np.ones((3,))  # (3,) 无法广播到 (3,2) → 需 reshape 为 (3,1)
Z = X + Y[:, np.newaxis]  # 正确

3. 高级索引：np.take, np.choose, np.select

# np.take：按索引从数组中取值（支持多维）
arr = np.array([[10, 20], [30, 40]])
indices = [0, 1, 0]
print(np.take(arr, indices, axis=0))  # 取第0、1、0行

# np.select：多条件赋值（替代多重 if-else）
x = np.array([-2, -1, 0, 1, 2])
condlist = [x < 0, x == 0, x > 0]
choicelist = [-1, 0, 1]
result = np.select(condlist, choicelist)
print(result)  # [-1 -1  0  1  1]

⚙️ 二、性能优化实战

1. 内存布局：C-order vs F-order

• C-order（行优先）：默认，arr[i, j] 连续访问 j 更快
• F-order（列优先）：适合 Fortran 风格或列操作多的场景

# 检查内存连续性
a = np.random.rand(1000, 1000)
print(a.flags['C_CONTIGUOUS'])  # True

# 转置后可能不连续
b = a.T
print(b.flags['C_CONTIGUOUS'])  # False → 影响性能！

# 强制连续化
b_cont = np.ascontiguousarray(b)

✅ 建议：在传递给 C/C++ 或深度学习框架前，确保数组是连续的。

2. 向量化替代循环（经典案例）

问题：计算两个数组的逐元素最大值，但跳过 NaN。

❌ 低效写法：

result = []
for i in range(len(a)):
    if np.isnan(a[i]) or np.isnan(b[i]):
        result.append(np.nan)
    else:
        result.append(max(a[i], b[i]))

✅ 高效写法：

result = np.maximum(a, b)  # 自动处理 NaN
# 或
result = np.where(np.isnan(a) | np.isnan(b), np.nan, np.maximum(a, b))

3. 使用 numba 或 Cython 加速瓶颈

from numba import jit

@jit(nopython=True)
def fast_sum(arr):
    total = 0.0
    for x in arr:
        total += x
    return total

# 比 np.sum 在某些小数组上更快（避免 NumPy 开销）

⚠️ 注意：NumPy 本身已高度优化，仅在特定场景（如复杂逻辑+小数组）才需额外加速。

🧪 三、工程实战场景

场景1：时间序列滑动窗口（无需循环！）

def rolling_window(a, window):
    shape = a.shape[:-1] + (a.shape[-1] - window + 1, window)
    strides = a.strides + (a.strides[-1],)
    return np.lib.stride_tricks.as_strided(a, shape=shape, strides=strides)

# 示例：计算每3个点的移动平均
data = np.array([1, 2, 3, 4, 5, 6, 7])
windows = rolling_window(data, 3)
moving_avg = windows.mean(axis=1)  # [2. 3. 4. 5. 6.]

💡 应用于：金融K线、信号处理、特征工程。

场景2：图像批量旋转（使用 np.rot90 + 广播）

# 模拟 100 张 28x28 灰度图
images = np.random.rand(100, 28, 28)

# 将每张图顺时针旋转90度
rotated = np.rot90(images, k=-1, axes=(1, 2))  # k=-1 表示顺时针
print(rotated.shape)  # (100, 28, 28)

场景3：高效 one-hot 编码

labels = np.array([2, 0, 1, 3])  # 4 个样本，4 类
num_classes = 4

# 方法1：使用 np.eye
one_hot = np.eye(num_classes)[labels]

# 方法2：使用高级索引（更省内存）
one_hot = np.zeros((labels.size, num_classes))
one_hot[np.arange(labels.size), labels] = 1

🔗 四、与生态工具无缝集成

1. NumPy ↔ Pandas

import pandas as pd

df = pd.DataFrame({'A': [1,2,3], 'B': [4,5,6]})
arr = df.values          # 转为 NumPy 数组 (3,2)
df_back = pd.DataFrame(arr, columns=['A', 'B'])  # 转回

⚠️ 注意：Pandas 的 df.to_numpy() 更推荐（避免视图问题）。

2. NumPy ↔ PyTorch / TensorFlow

# NumPy → PyTorch
import torch
arr = np.random.rand(3, 4)
tensor = torch.from_numpy(arr)  # 共享内存！修改 tensor 会影响 arr

# PyTorch → NumPy
arr2 = tensor.numpy()  # 仅 CPU 张量支持

# TensorFlow 类似
import tensorflow as tf
tensor_tf = tf.constant(arr)
arr_tf = tensor_tf.numpy()

✅ 最佳实践：在数据加载管道中，用 NumPy 预处理，再转为张量输入模型。

3. 保存与加载大型数组

# 单个数组
np.save('data.npy', arr)
arr_loaded = np.load('data.npy')

# 多个数组（压缩）
np.savez_compressed('data.npz', arr1=arr, arr2=another_arr)
data = np.load('data.npz')
print(data['arr1'])

💡 .npy 是 NumPy 专有二进制格式，比 CSV 快 10 倍以上，且保留 dtype。

🚫 五、常见陷阱与调试技巧

调试技巧：

# 检查数组内存占用
print(arr.nbytes / 1024**2, "MB")

# 检查是否包含 NaN/Inf
print(np.any(np.isnan(arr)))
print(np.any(np.isinf(arr)))

# 打印完整数组（不省略）
np.set_printoptions(threshold=np.inf)

📈 六、进阶练习（挑战）

1. 实现卷积操作（不使用 scipy）：给定 5x5 图像和 3x3 卷积核，输出 3x3 特征图。
2. K-Means 聚类核心：用 NumPy 实现距离计算、中心更新（不调 sklearn）。
3. 稀疏矩阵模拟：用三个一维数组（行索引、列索引、值）表示稀疏矩阵，并实现矩阵-向量乘法。
4. 蒙特卡洛 π 估算：生成 100 万随机点，估算 π 值。
5. 自定义 ufunc：用 np.frompyfunc 创建一个逐元素判断质数的函数。

📚 推荐资源

• 官方文档：numpy.org/doc
• 《Python for Data Analysis》（Wes McKinney）— NumPy 章节
• 《Elegant SciPy》— NumPy 高级用法
• NumPy Cheat Sheet（GitHub 搜索）

项目实战

🧪 实战场景 1：金融量化 —— 向量化回测策略

目标：计算股票收益率、移动平均线交叉策略信号，无需循环。

import numpy as np
import matplotlib.pyplot as plt

# 模拟 252 个交易日的股价（随机游走）
np.random.seed(42)
prices = 100 * np.exp(np.cumsum(np.random.randn(252) * 0.01))

# 计算对数收益率
log_returns = np.diff(np.log(prices))

# 计算 20日 & 50日移动平均线（使用卷积）
def moving_average(x, window):
    return np.convolve(x, np.ones(window)/window, mode='valid')

ma20 = moving_average(prices, 20)
ma50 = moving_average(prices, 50)

# 对齐长度（ma50 更短）
start = len(prices) - len(ma50)
aligned_prices = prices[start:]
aligned_ma20 = ma20[-len(ma50):]

# 生成交易信号：金叉（买入）/死叉（卖出）
signal = np.zeros_like(aligned_ma20)
signal[1:] = np.where(
    (aligned_ma20[1:] > aligned_ma50[1:]) & (aligned_ma20[:-1] <= aligned_ma50[:-1]), 1,
    np.where(
        (aligned_ma20[1:] < aligned_ma50[1:]) & (aligned_ma20[:-1] >= aligned_ma50[:-1]), -1, 0
    )
)

# 计算策略收益（假设全仓）
strategy_returns = np.zeros_like(signal)
strategy_returns[1:] = signal[:-1] * np.diff(np.log(aligned_prices))

print(f"累计策略收益: {np.exp(np.sum(strategy_returns)) - 1:.2%}")

✅ 优势：比 Pandas .rolling().mean() 快 3~5 倍（无索引开销）。

📡 实战场景 2：信号处理 —— 傅里叶滤波去噪

目标：对含噪声的正弦信号进行频域滤波。

import numpy as np
import matplotlib.pyplot as plt

# 生成信号：10Hz + 50Hz 正弦波 + 噪声
fs = 1000  # 采样率
t = np.linspace(0, 1, fs, endpoint=False)
signal_clean = np.sin(2*np.pi*10*t) + 0.5*np.sin(2*np.pi*50*t)
signal_noisy = signal_clean + 0.8*np.random.randn(len(t))

# FFT 变换
fft_vals = np.fft.rfft(signal_noisy)
freqs = np.fft.rfftfreq(len(t), 1/fs)

# 设计低通滤波器：保留 < 20Hz
fft_filtered = fft_vals.copy()
fft_filtered[freqs > 20] = 0

# 逆变换
signal_filtered = np.fft.irfft(fft_filtered)

# 可视化（略）
# plt.plot(t[:100], signal_noisy[:100], label='Noisy')
# plt.plot(t[:100], signal_filtered[:100], label='Filtered')
# plt.legend()

✅ 关键点：rfft 用于实数信号，节省 50% 计算量。

🖼️ 实战场景 3：计算机视觉 —— 批量图像标准化与增强

目标：对 ImageNet 风格图像进行归一化 + 随机裁剪（向量化实现）。

def batch_normalize_and_crop(images, mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]):
    """
    images: (N, H, W, 3) uint8 [0,255]
    输出: (N, 224, 224, 3) float32 [-2, 2]
    """
    # 转为 float
    images = images.astype(np.float32) / 255.0
    
    # 归一化
    for i in range(3):
        images[..., i] = (images[..., i] - mean[i]) / std[i]
    
    # 随机中心裁剪（简化版：固定裁剪）
    h, w = images.shape[1:3]
    top = (h - 224) // 2
    left = (w - 224) // 2
    cropped = images[:, top:top+224, left:left+224, :]
    
    return cropped

# 模拟 32 张 256x256 RGB 图像
batch_images = np.random.randint(0, 256, (32, 256, 256, 3), dtype=np.uint8)
normalized = batch_normalize_and_crop(batch_images)
print(normalized.shape, normalized.dtype)  # (32, 224, 224, 3) float32

✅ 工业实践：在 PyTorch DataLoader 中，此类操作应在 CPU 上用 NumPy 预处理，避免 GPU 瓶颈。

🔬 实战场景 4：科学计算 —— 求解热传导方程（有限差分法）

目标：模拟 2D 平板热扩散过程。

def solve_heat_equation(Lx, Ly, dx, dy, dt, T_final, alpha=0.01):
    """
    Lx, Ly: 区域大小
    dx, dy: 空间步长
    dt: 时间步长
    T_final: 终止时间
    alpha: 热扩散系数
    """
    nx, ny = int(Lx/dx), int(Ly/dy)
    nt = int(T_final/dt)
    
    # 初始化温度场（中心加热）
    T = np.zeros((nx, ny))
    T[nx//4:3*nx//4, ny//4:3*ny//4] = 100.0
    
    # 稳定性条件：alpha * dt / dx^2 <= 0.25
    r_x = alpha * dt / dx**2
    r_y = alpha * dt / dy**2
    
    for _ in range(nt):
        T_new = T.copy()
        # 内部点更新（向量化！）
        T_new[1:-1, 1:-1] = (
            T[1:-1, 1:-1] +
            r_x * (T[2:, 1:-1] - 2*T[1:-1, 1:-1] + T[:-2, 1:-1]) +
            r_y * (T[1:-1, 2:] - 2*T[1:-1, 1:-1] + T[1:-1, :-2])
        )
        T = T_new
    
    return T

# 运行模拟
temp_field = solve_heat_equation(Lx=1.0, Ly=1.0, dx=0.02, dy=0.02, dt=0.001, T_final=0.1)

✅ 性能提示：使用 numba.jit 可将此循环加速 100 倍以上。

🗃️ 实战场景 5：大数据分块处理 —— 内存映射（memmap）

目标：处理 10GB 的特征矩阵，内存仅 8GB。

# 创建超大数组（不加载到内存）
filename = 'huge_dataset.dat'
shape = (1_000_000, 512)
dtype = np.float32

# 写入（分块）
fp = np.memmap(filename, dtype=dtype, mode='w+', shape=shape)
chunk_size = 10_000
for i in range(0, shape[0], chunk_size):
    end = min(i + chunk_size, shape[0])
    fp[i:end] = np.random.randn(end - i, shape[1])
fp.flush()  # 确保写入磁盘

# 读取并处理（分块标准化）
fp_read = np.memmap(filename, dtype=dtype, mode='r', shape=shape)
means = np.zeros(shape[1])
for i in range(0, shape[0], chunk_size):
    end = min(i + chunk_size, shape[0])
    chunk = fp_read[i:end]
    means += chunk.sum(axis=0)
means /= shape[0]

# 再次遍历进行标准化
fp_norm = np.memmap('normalized.dat', dtype=dtype, mode='w+', shape=shape)
for i in range(0, shape[0], chunk_size):
    end = min(i + chunk_size, shape[0])
    chunk = fp_read[i:end]
    fp_norm[i:end] = (chunk - means) / chunk.std(axis=0)

✅ 适用场景：推荐系统特征、基因组数据、遥感图像。

🧩 实战场景 6：自定义通用函数（ufunc）

目标：创建一个支持广播的“安全除法”函数（0/0 → 0）。

def safe_divide(x, y):
    """x / y，当 y=0 时返回 0"""
    with np.errstate(divide='ignore', invalid='ignore'):
        result = np.true_divide(x, y)
        result[~np.isfinite(result)] = 0  # NaN 和 Inf 设为 0
    return result

# 转为 ufunc（支持广播和 out 参数）
safe_divide_ufunc = np.frompyfunc(safe_divide, 2, 1)

# 测试
a = np.array([1, 2, 0])
b = np.array([2, 0, 0])
print(safe_divide_ufunc(a, b))  # [0.5 0.0 0.0]

💡 更高效方式：直接用 np.where(y != 0, x / y, 0)。

📊 性能对比表（NumPy vs 原生 Python）

测试环境：Intel i7, Python 3.10, NumPy 1.24

🛠️ 工业级最佳实践清单

1. 始终指定 dtype：避免默认 int64/float64 浪费内存。
2. 预分配结果数组：用 np.empty + out 参数避免临时数组。
3. 避免 np.append/np.concatenate 在循环中：先收集再合并。
4. 用 np.einsum 替代复杂 dot/transpose：更清晰且更快。
5. 调试时用 np.seterr(all='raise')：让 NaN/Inf 立即报错。
6. 大数组用 np.memmap：突破内存限制。
7. 与 C/C++ 交互用 ctypes 或 Cython：避免复制。

📥 附：完整项目结构建议

numpy-projects/
├── data/                  # 原始数据
├── src/
│   ├── finance/           # 量化策略
│   ├── signal/            # 信号处理
│   ├── vision/            # 图像预处理
│   └── utils.py           # 通用 NumPy 工具函数
├── notebooks/             # 探索性分析
├── tests/                 # 单元测试（用 pytest + numpy.testing）
└── requirements.txt

参考：https://numpy.org/doc/stable/user/index.html