AI-Generated-Pygame速查指南

字数统计: 3k字   |   阅读时长≈ 12分

Pygame 速查手册

引言

Pygame 是基于 Simple DirectMedia Layer (SDL) 的跨平台 Python 模块，专为编写视频游戏设计。它提供了图形、声音、输入处理等游戏开发所需的核心功能，非常适合初学者快速上手 2D 游戏开发，也足够灵活支持一些复杂的项目。

本文旨在作为 Pygame 的快速参考指南，覆盖从初始化到高级特性的主要知识点。每个部分都包含关键函数、简短示例以及实用评论，帮助你在实际开发中快速查阅。

1. 安装与初始化

安装

1
pip install pygame

验证安装：

1
2
import pygame
print(pygame.version.ver)

初始化

使用 pygame.init() 一次性初始化所有模块，或单独初始化所需模块（如 pygame.display.init()）。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
import pygame

pygame.init()
screen = pygame.display.set_mode((800, 600))  # 创建窗口
pygame.display.set_caption("My Game")        # 设置标题
clock = pygame.time.Clock()                   # 用于控制帧率

running = True
while running:
    for event in pygame.event.get():
        if event.type == pygame.QUIT:
            running = False

    # 游戏逻辑与绘制
    pygame.display.flip()      # 更新整个屏幕
    clock.tick(60)              # 限制帧率为 60 FPS

pygame.quit()

评论：

  • pygame.init() 会尝试初始化所有模块，如果某个模块失败（例如缺少音频硬件），仍会继续执行，但后续使用该模块可能出错。建议根据需求单独初始化。
  • 主循环中必须调用 pygame.event.get() 来处理事件队列，否则窗口会无响应。
  • clock.tick(60) 不仅控制帧率，还返回上一帧经过的时间（毫秒），可用于基于时间的运动。

2. 核心概念

表面（Surface）

Surface 是 Pygame 中表示图像的对象。屏幕（display）本身就是一个 Surface，你可以创建额外的 Surface 用于绘图或缓存。

1
2
3
surf = pygame.Surface((50, 50))   # 创建一个 50x50 的透明表面
surf.fill((255, 0, 0))            # 填充红色
screen.blit(surf, (100, 100))      # 将表面绘制到屏幕的 (100,100) 位置

矩形（Rect）

Rect 对象存储矩形坐标和尺寸，提供许多便捷方法（碰撞检测、移动、合并等）。

1
2
3
4
rect = pygame.Rect(10, 20, 100, 50)   # x, y, width, height
rect.center = (400, 300)               # 修改中心位置
if rect.collidepoint(mouse_x, mouse_y):
    print("鼠标在矩形内")

颜色（Color）

Pygame 使用 (R, G, B) 三元组或 (R, G, B, A) 四元组表示颜色。pygame.Color 对象提供颜色名称和转换。

1
2
red = pygame.Color(255, 0, 0)
blue = pygame.Color("blue")           # 支持常见颜色名称

时钟（Clock）

用于控制时间，限制帧率，获取时间间隔。

事件（Event）

所有输入（键盘、鼠标、窗口事件）都以事件形式放入队列，由 pygame.event.get() 获取。

评论：

  • Surface 的 blit() 是绘制图像的核心方法，理解它的工作机制（源 Surface 和目标位置）至关重要。
  • Rect 对象的属性（如 x, y, top, left, center, size 等）都是可读写的，极大简化了坐标计算。
  • 事件处理通常放在主循环开头，避免累积导致输入延迟。

3. 图形绘制

Pygame 提供直接在 Surface 上绘制基本形状的函数（位于 pygame.draw 模块）。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
# 绘制矩形
pygame.draw.rect(screen, (0, 255, 0), (50, 50, 100, 100), 2)  # 绿色边框，线宽2

# 绘制圆形
pygame.draw.circle(screen, (255, 0, 0), (200, 200), 30)       # 红色实心圆

# 绘制线条
pygame.draw.line(screen, (0, 0, 255), (0, 0), (800, 600), 5)

# 绘制多边形
points = [(300, 300), (350, 350), (400, 300)]
pygame.draw.polygon(screen, (255, 255, 0), points)

# 绘制抗锯齿线（需要单独模块）
pygame.gfxdraw.line(screen, 100, 100, 200, 200, (255,255,255))

评论：

  • 实心形状通过省略线宽参数或设为 0 实现；线宽 >0 仅绘制边框。
  • 抗锯齿函数位于 pygame.gfxdraw，但需要 SDL_gfx 库支持，通常预装。对于简单抗锯齿，也可使用 pygame.draw.aaline 和 pygame.draw.aalines。
  • 频繁绘制大量形状时性能可能下降，可考虑使用 Sprite 或预渲染表面优化。

4. 图像处理

加载和显示

1
2
3
image = pygame.image.load("player.png")   # 支持 PNG, JPG, GIF 等
image = image.convert_alpha()              # 转换为显示格式，保留 Alpha 通道
screen.blit(image, (x, y))

变换

1
2
3
4
5
6
# 缩放
scaled = pygame.transform.scale(image, (new_width, new_height))
# 旋转（逆时针，角度）
rotated = pygame.transform.rotate(image, 45)
# 翻转（x轴，y轴）
flipped = pygame.transform.flip(image, True, False)

像素访问

1
2
3
pxarray = pygame.PixelArray(image)   # 将 Surface 转为二维数组
pxarray[10][20] = (255,0,0)           # 修改像素颜色
del pxarray                            # 释放数组，使原 Surface 可继续操作

评论：

  • convert() 和 convert_alpha() 将图像转换为屏幕像素格式，提高 blit 速度。不调用时，每次 blit 都会进行转换，影响性能。
  • 旋转操作会改变图像尺寸（生成的新矩形可能变大），通常结合 get_rect(center=原中心) 保持位置不变。
  • PixelArray 锁定 Surface，操作完毕后必须释放，否则无法对该 Surface 进行其他操作。

5. 文字显示

初始化字体模块

1
pygame.font.init()

使用系统字体

1
2
3
font = pygame.font.Font(None, 36)          # None 表示默认字体，36 字号
text_surface = font.render("Hello World", True, (255,255,255))
screen.blit(text_surface, (100,100))

加载自定义字体

1
font = pygame.font.Font("myfont.ttf", 24)

文字属性

1
2
# 渲染模式：True 表示抗锯齿，背景色可选
text_surface = font.render("Score: 10", True, (0,0,0), (255,255,255))

评论：

  • 字体渲染开销较大，建议将静态文字渲染一次并缓存 Surface，避免每帧渲染。
  • 默认字体通常不支持中文，如需显示中文，需加载中文字体文件，并确保字符串为 Unicode（如 u"你好" 或直接使用中文字符串）。
  • 抗锯齿渲染效果更平滑，但背景色必须与最终 blit 的背景一致，否则会有黑边。

6. 音频

初始化混音器

1
pygame.mixer.init(frequency=22050, size=-16, channels=2, buffer=512)

音效（短音频）

1
2
3
sound = pygame.mixer.Sound("explosion.wav")
sound.play()                     # 播放一次
sound.play(loops=3)               # 循环 3 次（总共播放 4 次）

背景音乐（长音频，如 MP3）

1
2
3
pygame.mixer.music.load("theme.mp3")
pygame.mixer.music.play(-1)        # -1 表示无限循环
pygame.mixer.music.set_volume(0.5) # 音量 0.0-1.0

评论：

  • 音效使用 Sound 对象，适合短小、频繁触发的声音；背景音乐使用 Music 流式播放，占用内存小。
  • 不同音频格式支持取决于 SDL 编译选项，一般 WAV、OGG 支持良好，MP3 可能受限。
  • 音效过多可能混叠，可通过为每个 Sound 设置独立的 Channel 进行管理。

7. 用户输入

事件处理（推荐）

1
2
3
4
5
6
7
for event in pygame.event.get():
    if event.type == pygame.KEYDOWN:
        if event.key == pygame.K_SPACE:
            print("Space pressed")
    elif event.type == pygame.MOUSEBUTTONDOWN:
        if event.button == 1:    # 左键
            print("Mouse clicked at", event.pos)

实时按键状态（连续响应）

1
2
3
4
5
6
keys = pygame.key.get_pressed()
if keys[pygame.K_LEFT]:
    player.x -= 5

mouse_x, mouse_y = pygame.mouse.get_pos()
pressed = pygame.mouse.get_pressed()  # 返回 (左键, 中键, 右键) 状态

评论：

  • 事件处理适合单次触发动作（如跳跃、菜单选择）；状态轮询适合持续移动（如按住方向键）。
  • 键盘常量和鼠标按钮常量可在 pygame.locals 中找到（但通常直接使用 pygame.K_xxx 即可，因为全局命名空间已导入）。
  • 注意 pygame.event.get() 会清空事件队列，如果需要同时处理事件和状态轮询，应先处理事件。

8. 精灵和组

Sprite 基类

1
2
3
4
5
6
7
8
9
10
11
12
class Player(pygame.sprite.Sprite):
    def __init__(self):
        super().__init__()
        self.image = pygame.Surface((50,50))
        self.image.fill((0,255,0))
        self.rect = self.image.get_rect()
        self.rect.center = (400,300)

    def update(self):
        keys = pygame.key.get_pressed()
        if keys[pygame.K_LEFT]:
            self.rect.x -= 5

精灵组

1
2
3
4
5
6
7
all_sprites = pygame.sprite.Group()
player = Player()
all_sprites.add(player)

# 在主循环中
all_sprites.update()          # 调用每个精灵的 update 方法
all_sprites.draw(screen)       # 绘制所有精灵到屏幕

碰撞检测

1
2
3
4
5
# 两个组之间的碰撞，返回字典 {精灵A: [精灵B, ...]}
collisions = pygame.sprite.groupcollide(group1, group2, True, True)

# 精灵与组的碰撞，返回碰撞到的精灵列表
hits = pygame.sprite.spritecollide(player, enemies, False)

评论：

  • Sprite 必须包含 image 和 rect 属性，Group.draw() 依赖它们。
  • 精灵组提供了批量更新和绘制，极大简化代码。
  • 碰撞检测默认基于 Rect 的矩形碰撞，如需更精确的像素级碰撞，可结合 pygame.sprite.collide_mask 或自定义碰撞函数。

9. 时间和动画

基于时间的运动

1
2
dt = clock.tick(60) / 1000.0   # 转换为秒
player.x += speed * dt

计时器与自定义事件

1
2
3
4
5
pygame.time.set_timer(pygame.USEREVENT + 1, 1000)  # 每秒触发一次事件

for event in pygame.event.get():
    if event.type == pygame.USEREVENT + 1:
        print("一秒过去了")

动画帧切换

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
class AnimatedSprite(pygame.sprite.Sprite):
    def __init__(self, frames):
        super().__init__()
        self.frames = frames
        self.current_frame = 0
        self.image = frames[0]
        self.rect = self.image.get_rect()
        self.last_update = pygame.time.get_ticks()
        self.frame_rate = 100  # 每 100 毫秒切换一帧

    def update(self):
        now = pygame.time.get_ticks()
        if now - self.last_update > self.frame_rate:
            self.last_update = now
            self.current_frame = (self.current_frame + 1) % len(self.frames)
            self.image = self.frames[self.current_frame]

评论：

  • 使用 dt 确保运动速度与帧率解耦，避免高帧率下速度过快。
  • 自定义事件非常适合实现定时器、刷怪等周期性逻辑。
  • 动画切换不依赖帧率，而是基于绝对时间，可保证动画速度一致。

10. 高级主题

全屏模式与分辨率切换

1
2
screen = pygame.display.set_mode((0,0), pygame.FULLSCREEN)  # 当前桌面分辨率
screen = pygame.display.set_mode((1280,720), pygame.FULLSCREEN | pygame.HWSURFACE)

双缓冲与页面翻转

pygame.display.flip() 交换前后缓冲区，实现平滑渲染。pygame.display.update(rect_list) 可更新部分区域优化性能。

自定义事件

事件类型从 pygame.USEREVENT 开始递增。

混合模式与特殊效果

blit 可以指定特殊标志（如 BLEND_ADD、BLEND_ALPHA_SDL2）实现颜色混合。

OpenGL 支持

Pygame 支持创建 OpenGL 上下文：

1
pygame.display.set_mode((800,600), pygame.OPENGL | pygame.DOUBLEBUF)

之后可使用 PyOpenGL 进行 3D 渲染，但此时 Pygame 的 2D 绘图函数无法使用。

评论：

  • 全屏模式下注意分辨率和硬件加速标志，不同平台支持有差异。
  • 混合模式可以实现简单的光照、遮罩效果，但性能开销较大，建议仅用于小区域。
  • 结合 OpenGL 可获得 3D 能力，但学习曲线陡峭，适合需要 3D 或复杂特效的项目。

11. 优化和调试

性能优化技巧

  • 使用 convert()/convert_alpha() 转换图像。
  • 尽量减少每帧的 Surface 创建，复用已有 Surface。
  • 对于静态背景，可将其绘制在一个背景 Surface 上，每帧直接 blit 这个背景，而不是重绘所有元素。
  • 使用 Dirty Rect 技术：仅更新变化的区域（pygame.display.update(rects)）。
  • 将频繁绘制且不变的图形缓存在临时 Surface 上。

调试工具

  • pygame.display.set_caption(str(clock.get_fps())) 在窗口标题显示帧率。
  • 使用 pygame.image.save(screen, "screenshot.png") 保存屏幕截图。
  • 异常处理：主循环中使用 try...except 捕获错误，防止窗口突然关闭。

常见陷阱

  • 事件队列阻塞：长时间运算不处理事件会导致窗口无响应，可适时调用 pygame.event.pump() 保持响应。
  • 图像格式不支持：某些 JPEG 可能无法加载，使用 PNG 格式更可靠。
  • 内存泄漏：精灵组中删除精灵后需从组中移除，否则仍被引用。
  • 音频格式问题：确保音频文件格式被 SDL 支持，必要时使用 OGG 或 WAV。

12. 资源和进一步学习

结语

Pygame 以其简洁的 API 和强大的功能，成为无数开发者踏入游戏开发领域的第一站。本文作为速查手册，涵盖了大多数常用模块，但真正的精进还需在项目中不断实践。希望这份指南能成为你开发路上的得力助手，快速解决常见问题，让你专注于创造有趣的游戏体验。

Happy Coding! 🎮

展开全文 >>

进化算法常见算子

字数统计: 2.6k字   |   阅读时长≈ 9分

进化算法常见算子

1. 前言

本文主要介绍进化算法各阶段常用的算子，并通过具体示例帮助理解每种算子的工作原理和适用场景。

2. 进化算法的结构及其算子

2.1. 进化算法的结构

进化算法可大致分为以下几个部分：

  • 表示：如何使用基因或染色体结构去表示一个解（如二进制串、实数向量、排列序列等）
  • 种群：基于种群的算法，维护一组候选解
  • 适应度函数：衡量解的质量，一般来讲，适应度越高越好
  • 初始化：随机或者基于先验知识生成初始种群
  • 父母选择机制：挑选适应性较好的个体用于繁衍后代
  • 变异与交叉算子：
    • 交叉重组：组合两个或多个父母的基因产生子代
    • 变异：对个体的基因进行微调，引入新的遗传信息
  • 幸存者选择机制：从父母和子代中选择合适的个体组成新一代种群
  • 终止条件：循环上述过程，直到达到目标解或满足预设的终止条件（如最大代数、计算时间、适应度阈值）

2.2. 进化算法算子

2.2.1 算子综览

  • 交叉重组：K点交叉、均匀交叉、算术交叉、模拟二进制交叉（SBX）等
  • 变异：位翻转、高斯变异、多项式变异、交换变异等
  • 选择：轮盘赌选择、锦标赛选择、截断选择、精英保留等
  • 代际更新策略：完全替换、精英保留、稳态替换、父子混合选择等
  • 特殊算子：多父母重组、反转变异、部分映射交叉（PMX）、顺序交叉（OX）等

2.2.2. 探索（Exploration）和利用（Exploitation）

  • 探索（Exploration）：全局搜索，探索整个搜索空间以发现新的有前景区域。主要通过变异算子和较大的交叉步长实现。
  • 利用（Exploitation）：局部搜索，在已知的良好区域进行精细搜索以逼近最优解。主要通过选择压力和局部交叉实现。

两者需要平衡：过度探索会导致收敛缓慢，过度利用则容易陷入局部最优。

2.2.3. 交叉重组算子

2.2.3.1. 离散编码的重组

适用于二进制编码、符号编码、排列编码等。

  1. 单点交叉

    • 操作：随机选择一个交叉点，将两个父母在该点之后的部分交换。
    • 示例：
      1
      2
      父母1: 101|0011 → 子代1: 101|1100
      父母2: 010|1100 → 子代2: 010|0011

  2. k点交叉

    • 操作：随机选择k个交叉点，将染色体分成k+1段，间隔交换这些段。
    • 示例（2点交叉）：
      1
      2
      父母1: 10|100|11 → 子代1: 10|011|11
      父母2: 01|011|00 → 子代2: 01|100|00

  3. 均匀交叉

    • 操作：对于每个基因位，以概率p（通常p=0.5）从父母1取值，否则从父母2取值。
    • 示例（p=0.5）：
      1
      2
      3
      4
      父母1: 1 0 1 0 0 1 1
      父母2: 0 1 0 1 1 0 0
      掩码:   1 0 1 0 1 1 0  (1表示取父1，0表示取父2)
      子代:   1 1 1 1 0 1 0
2.2.3.2. 连续编码的重组

适用于实数参数优化问题。

  1. 算术交叉

    • 操作：对父母向量进行加权平均。
    • 公式：
      设 $x_1$ 和 $x_2$ 为两个父母，$\alpha \in [0,1]$ 为权重：
      • 子代1：$x’ = \alpha x_1 + (1-\alpha) x_2$
      • 子代2：$x’’ = \alpha x_2 + (1-\alpha) x_1$
    • 示例（$\alpha=0.3$）：
      1
      2
      3
      4
      父母1: [2.0, 5.0]
      父母2: [8.0, 3.0]
      子代1: [0.3×2.0+0.7×8.0=6.2, 0.3×5.0+0.7×3.0=3.6]
      子代2: [0.3×8.0+0.7×2.0=3.8, 0.3×3.0+0.7×5.0=4.4]

  2. 模拟二进制交叉（SBX）

    • 特点：模拟二进制编码的单点交叉特性，使子代分布在父母周围，分布指数可调。
    • 公式（简化）：
      生成随机数 $u \in [0,1]$，根据分布因子 $\eta$ 计算扩展因子 $\beta$，然后：
      $x’_1 = 0.5[(1+\beta)x_1 + (1-\beta)x_2]$
      $x’_2 = 0.5[(1-\beta)x_1 + (1+\beta)x_2]$

  3. 启发式交叉

    • 操作：假设 $x_2$ 优于 $x_1$，则沿 $x_2$ 指向 $x_2 - x_1$ 的方向外推。
    • 公式：$x’ = x_2 + \alpha(x_2 - x_1)$，$\alpha \in [0,1]$

  4. 单纯形交叉

    • 操作：选择一组父母（>2），找出最优 $x_b$ 和最差 $x_w$，计算除最差点外其他点的重心 $x_c$，用 $x_c$ 替换 $x_w$。

2.2.4. 变异算子

2.2.4.1. 离散编码的变异

  1. 位翻转变异

    • 单点翻转：随机选择一个位，将0变1或1变0。
    • 示例：
      1
      2
      原个体: 1011001
      变异后: 1010001 (第4位翻转)
    • 多位翻转：按概率 $p_m$ 对每一位独立决定是否翻转，通常 $p_m = 1/d$（d为染色体长度）。

  2. 交换变异

    • 操作：随机选择两个位置，交换其值。
    • 示例（适用于排列编码，如TSP）：
      1
      2
      原路径: A-B-C-D-E-F
      变异后: A-E-C-D-B-F (交换B和E)

  3. 反转变异

    • 操作：随机选择一段子序列，将其顺序反转。
    • 示例：
      1
      2
      3
      原路径: A-B-C-D-E-F-G
      反转段: C-D-E → E-D-C
      变异后: A-B-E-D-C-F-G
2.2.4.2. 连续编码的变异

  1. 均匀变异

    • 操作：以概率 $p_m$ 将某维度的值替换为定义域内的随机值。
    • 公式：$x_i’ = U(L_i, U_i)$
    • 示例（定义域[0,10]）：
      1
      2
      原个体: [3.2, 5.7, 8.1]
      变异后: [3.2, 2.3, 8.1] (第2维随机变为2.3)
    • 注意：可能破坏优良解，通常用于早期探索。

  2. 高斯变异

    • 操作：在原值基础上添加服从高斯分布的随机扰动。
    • 公式：$x_i’ = x_i + N(0, \sigma)$
    • 示例（$\sigma=0.5$）：
      1
      2
      3
      原个体: [3.2, 5.7, 8.1]
      扰动:   [+0.3, -0.1, +0.4]
      变异后: [3.5, 5.6, 8.5]
    • 特点：小步长局部搜索，步长可自适应调整。

  3. 多项式变异

    • 特点：模拟二进制变异，分布指数可调，能在边界附近产生更多变异。

2.2.5. 选择机制

选择压力（Selection Pressure） 是驱动种群向更优区域进化的核心动力。选择压力过大易早熟，过小则收敛缓慢。

2.2.5.1. 父辈选择机制

  1. 轮盘赌选择（适应度比例选择）

    • 操作：个体被选中的概率与其适应度成正比。
    • 示例：设种群有3个个体，适应度为[5, 3, 2]，总适应度=10，则选中概率分别为[0.5, 0.3, 0.2]。
    • 问题：适应度不能为负；早期超级个体易主导种群；后期个体间适应度相近时选择压力消失。

  2. 锦标赛选择

    • 操作：随机抽取k个个体，从中选出最优者。
    • 示例（k=3）：随机抽取3个个体，比较适应度，胜者入选。
    • 特点：无需全局适应度归一化，选择压力可通过锦标赛规模k调节（k越大压力越大）。

  3. 排序选择

    • 操作：按适应度排序，根据排序位置分配选择概率。
    • 线性排序：设最优个体排序值为N，最差为1，则第i个体的选择概率为 $\frac{2s}{N(N+1)}(N-i+1)$，s为平均选择次数。
2.2.5.2. 幸存者选择机制

  1. 精英保留策略

    • 操作：将当前种群中最优的几个个体直接复制到下一代，不参与交叉变异。
    • 作用：保证算法单调不退化，防止最优解丢失。

  2. $(\mu+\lambda)$ 选择

    • 操作：从 $\mu$ 个父代和 $\lambda$ 个子代的并集中选出 $\mu$ 个最优个体。
    • 特点：保留父代中的优秀个体，收敛快，但可能早熟。

  3. $(\mu,\lambda)$ 选择

    • 操作：只从 $\lambda$ 个子代中选出 $\mu$ 个最优个体（通常 $\lambda > \mu$），父代全部淘汰。
    • 特点：允许暂时变差，有助于跳出局部最优。

  4. 基于年龄的选择

    • 操作：每个个体允许存活固定代数，到期后无论适应度如何都被淘汰。
    • 作用：保持种群活力，防止某些个体长期主导。

3. 算子选择的考量与建议

选择哪些算子以及如何设置参数，取决于问题的性质和优化目标：

3.1 问题特性与算子匹配

问题类型 推荐编码 常用交叉算子 常用变异算子
组合优化（如TSP） 排列编码 PMX, OX, 循环交叉 交换变异, 反转变异
函数优化（连续） 实数向量 SBX, 算术交叉 高斯变异, 多项式变异
特征选择 二进制编码 单点/均匀交叉 位翻转变异
调度问题 排列/实数 顺序交叉, 启发式交叉 插入变异, 扰动变异

3.2 参数设置建议

  • 交叉概率 $p_c$：通常设为0.7-0.95。过高会破坏优良模式，过低则搜索效率低。
  • 变异概率 $p_m$：二进制编码常取 $1/d$；连续编码常取0.01-0.1。自适应变异是更高级的策略。
  • 种群大小：一般取20-100，复杂问题可更大。过小易早熟，过大会降低收敛速度。
  • 选择压力调节：
    • 锦标赛规模：k=2-7，问题越复杂宜用小规模。
    • 精英数量：1-2个通常足够，过多会降低多样性。

3.3 常见组合模式

  1. 遗传算法（GA）经典组合：锦标赛选择 + 单点/均匀交叉 + 位翻转变异 + 精英保留
  2. 进化策略（ES）典型配置：$(\mu+\lambda)$ 或 $(\mu,\lambda)$ 选择 + 算术交叉/无交叉 + 自适应高斯变异
  3. 差分进化（DE）：DE算子（基于向量差分的变异） + 二项式交叉 + 贪婪选择

4. 总结

进化算法的算子种类繁多，每种算子都有其适用场景和特性：

  • 交叉算子负责从现有优良解中组合新解，主要贡献于利用能力。
  • 变异算子负责引入新的遗传信息，主要贡献于探索能力。
  • 选择机制调控进化方向，平衡探索与利用。

总的来说，进化算法的算子各种各样，但需要针对具体情况去进行选择。不适合具体情境的算子不但会对优化过程毫无作用，甚至可能破坏优秀解带来坏处。

展开全文 >>

进化算法

字数统计: 2.5k字   |   阅读时长≈ 9分

有信息搜索和无信息搜索

进化算法实际是一种从较高层面上去指导优化思路的元启发式搜索算法。从这个层面出发，我们将搜索方式总体的分成无信息搜索（Uninformed Search）和有信息搜索（Informed Search）。

无信息搜索指的是算法在搜索的过程中不依赖额外的路径信息，诸如BFS和DFS，其在搜索的过程中并不关心具体的树结构是什么样的。
相对的有信息搜索算法就是会去感知总体或局部搜索地图的结构，或者递进式的基于前面计算的结果去进行最优解的搜索。
more >>

展开全文 >>

AI-Generated-PySpark速查指南

字数统计: 2.5k字   |   阅读时长≈ 12分

PySpark速查

1. 引言

PySpark是Apache Spark的Python API，支持大规模数据处理的分布式计算。其核心抽象包括RDD（弹性分布式数据集）和DataFrame。RDD提供底层API，适合需要精细控制的数据处理；DataFrame则提供更高层的抽象，具备性能优化和类似SQL的操作接口。本文从环境搭建开始，全面覆盖RDD与DataFrame的基础操作，并深入SQL查询、用户自定义函数、性能优化、机器学习及流处理等进阶内容，每个知识点均配有代码示例和实用评论，帮助读者快速掌握PySpark的核心用法。

2. 环境搭建与配置

安装PySpark并初始化SparkSession：

1
pip install pyspark

1
2
3
4
5
6
7
from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .master("local[2]") \
    .appName("PySpark速查") \
    .getOrCreate()
sc = spark.sparkContext  # 获取SparkContext，RDD操作需要

评论：local[2]表示本地模式使用2个核心。生产环境通常通过--master指定集群管理器。SparkSession是Spark 2.0后统一入口，同时封装了SparkContext、SQLContext等。

3. RDD基础操作

3.1 创建RDD

RDD可从Python集合或外部文件创建。

1
2
3
4
5
6
7
8
# 从列表创建（parallelize）
rdd = sc.parallelize([1, 2, 3, 4, 5])
# 从元组创建
rdd = sc.parallelize(('cat', 'dog', 'fish'))
# 从集合（set）创建（自动去重）
rdd = sc.parallelize({'cat', 'dog', 'fish', 'cat'})
# 从文件创建（textFile）
rdd_text = sc.textFile("wordCount.txt")

评论：parallelize用于测试，大数据场景应使用textFile或后续的DataFrame读取接口。collect()可将RDD内容拉取到驱动节点，仅限小数据调试。

3.2 单RDD转换

转换操作返回新RDD，惰性执行。

3.2.1 map()

对每个元素应用函数。

1
2
3
4
5
6
7
rdd = sc.parallelize([1, 2, 3])
rdd.map(lambda x: x + 1).collect()  # [2, 3, 4]
# 字符串处理
rdd = sc.parallelize(['cat', 'dog'])
rdd.map(lambda x: 'object: ' + x).collect()  # ['object: cat', 'object: dog']
# 生成键值对
rdd.map(lambda x: (x, 1)).collect()  # [('cat', 1), ('dog', 1)]

3.2.2 filter()

返回使函数为真的元素。

1
2
rdd = sc.parallelize([3, 1, 2, 5, 5])
rdd.filter(lambda x: x > 2).collect()  # [3, 5, 5]

3.2.3 distinct()

去重。

1
2
rdd = sc.parallelize([3, 1, 2, 5, 5, 5])
rdd.distinct().collect()  # [1, 2, 3, 5]（顺序可能变化）

3.2.4 randomSplit()

按权重随机切分RDD。

1
2
3
4
rdd = sc.parallelize(range(10))
splits = rdd.randomSplit([0.6, 0.4], seed=42)
splits[0].collect()  # 第一部分
splits[1].collect()  # 第二部分

3.2.5 groupBy()

根据函数返回值分组。

1
2
3
4
rdd = sc.parallelize([1, 1, 2, 3, 5, 8])
grouped = rdd.groupBy(lambda x: x % 2)
result = [(k, list(v)) for k, v in grouped.collect()]
print(result)  # [(1, [1,1,3,5]), (0, [2,8])]

3.2.6 flatMap()

每个输入元素可生成多个输出元素，并展平。

1
2
3
rdd = sc.parallelize(['hello world', 'hi', 'dog'])
rdd.flatMap(lambda x: x.split(' ')).collect()
# ['hello', 'world', 'hi', 'dog']

3.3 多RDD转换

3.3.1 union()

合并两个RDD（不改变分区数）。

1
2
3
rdd1 = sc.parallelize([1, 2])
rdd2 = sc.parallelize([3, 4])
rdd1.union(rdd2).collect()  # [1, 2, 3, 4]

3.3.2 intersection()

返回两个RDD共有的元素（去重）。

1
2
3
rdd1 = sc.parallelize([1, 1, 2, 3])
rdd2 = sc.parallelize([3, 4, 5])
rdd1.intersection(rdd2).collect()  # [3]

3.3.3 subtract()

返回在第一个RDD中但不在第二个中的元素。

1
2
3
rdd1 = sc.parallelize([1, 1, 2, 3])
rdd2 = sc.parallelize([3, 4, 5])
rdd1.subtract(rdd2).collect()  # [1, 1, 2]

3.3.4 cartesian()

计算两个RDD的笛卡尔积。

1
2
3
rdd1 = sc.parallelize([1, 2])
rdd2 = sc.parallelize([3, 4])
rdd1.cartesian(rdd2).collect()  # [(1,3), (1,4), (2,3), (2,4)]

3.4 RDD动作

动作触发实际计算并返回结果到驱动。

3.4.1 first() / take(n)

1
2
3
rdd = sc.parallelize([21, 1, 2, 3])
rdd.first()      # 21
rdd.take(2)      # [21, 1]

3.4.2 takeOrdered(n, key)

返回排序后的前n个元素。

1
2
3
rdd = sc.parallelize([10, 1, 2, 9, 3])
rdd.takeOrdered(3)  # [1, 2, 3]（升序）
rdd.takeOrdered(3, lambda x: -x)  # [10, 9, 3]（降序）

3.5 基本统计

RDD提供便捷的统计方法（需数值型）。

1
2
3
4
5
6
7
rdd = sc.parallelize([1, 2, 3, 4, 5])
rdd.min()    # 1
rdd.max()    # 5
rdd.mean()   # 3.0
rdd.sum()    # 15
rdd.stdev()  # 样本标准差
rdd.count()  # 5

3.6 键值对RDD（PairRDD）

当RDD元素为(key, value)对时，可调用专属方法。

3.6.1 获取键和值

1
2
3
kv = sc.parallelize([(3,4), (3,6), (5,6), (1,2)])
kv.keys().collect()   # [3, 3, 5, 1]
kv.values().collect() # [4, 6, 6, 2]

3.6.2 filter() 基于键或值

1
2
kv.filter(lambda kv: kv[0] < 5).collect()   # 键小于5
kv.filter(lambda kv: kv[1] < 5).collect()   # 值小于5

3.6.3 sortBy() 排序

1
kv.sortBy(lambda kv: kv[0]).collect()   # 按键升序

3.6.4 mapValues() 仅对值操作

1
kv.mapValues(lambda v: v**2).collect()  # [(3,16), (3,36), (5,36), (1,4)]

3.6.5 reduceByKey() 按键聚合

1
2
from operator import add
kv.reduceByKey(add).collect()  # [(3,10), (5,6), (1,2)]

3.7 多键值对RDD转换

3.7.1 join()（内连接）

1
2
3
kv1 = sc.parallelize([(1,2), (3,4), (3,6)])
kv2 = sc.parallelize([(3,7)])
kv1.join(kv2).collect()  # [(3, (4,7)), (3, (6,7))]

3.7.2 leftOuterJoin() / rightOuterJoin()

1
2
kv1.leftOuterJoin(kv2).collect()
# [(1, (2, None)), (3, (4,7)), (3, (6,7))]

3.7.3 subtractByKey()

1
kv1.subtractByKey(kv2).collect()  # [(1,2)]

3.8 键值对RDD动作

3.8.1 countByKey() / countByValue()

1
2
kv = sc.parallelize([(1,2), (3,4), (3,6), (4,1), (5,6)])
kv.countByKey()   # defaultdict(int, {1:1, 3:2, 4:1, 5:1})

3.9 RDD惰性求值

转换操作是惰性的，只有动作触发才执行。

1
2
rdd = sc.textFile("nonexistent.txt")  # 此时不会报错
rdd.collect()  # 此时报错文件不存在

4. DataFrame操作

DataFrame以列方式组织，类似于关系表，内置Catalyst优化器。

4.1 创建DataFrame

4.1.1 从文件读取

1
2
3
4
# CSV（自动推断类型）
df = spark.read.csv("mtcars.csv", header=True, inferSchema=True)
# Parquet（推荐）
df = spark.read.parquet("data.parquet")

4.1.2 从Python集合创建

1
2
3
data = [['a', 1], ['b', 2]]
df = spark.createDataFrame(data, ['letter', 'number'])
df.show()

4.1.3 从Pandas DataFrame创建

1
2
3
import pandas as pd
pdf = pd.DataFrame({'x': [[1,2,3]], 'y': [['a','b','c']]})
df = spark.createDataFrame(pdf)

4.2 查看数据结构

1
2
3
df.columns        # 列名列表
df.dtypes         # 列名与类型元组列表
df.printSchema()  # 树形打印结构

4.3 基本操作

4.3.1 重命名列

1
df = df.withColumnRenamed('letter', 'char')

4.3.2 排序

1
df.sort('number', ascending=False).show()

4.3.3 删除列

1
df.drop('number').show()

4.3.4 过滤行

1
df.filter(df.number > 1).show()

4.3.5 选择列

1
2
df.select('letter').show()
df.select('letter', df.number * 10).show()

4.3.6 去重

1
df.select('number').distinct().show()

4.4 聚合函数

使用agg或直接调用内置函数。

1
2
from pyspark.sql.functions import max, min, sum, avg
df.agg(max('number'), min('number')).show()

4.5 DataFrame与RDD转换

4.5.1 DataFrame转RDD

1
2
rdd = df.rdd  # 返回RDD[Row]
rdd.take(2)   # [Row(letter='a', number=1), Row(letter='b', number=2)]

4.5.2 在RDD上使用map

1
rdd.map(lambda row: (row['letter'], row['number'])).collect()

4.5.3 flatMapValues示例

1
2
3
car_data = spark.createDataFrame([(2018, 'Mazda Ferrari')], ['year', 'brand'])
rdd = car_data.rdd.map(lambda x: (x.year, x.brand.split()))
rdd.flatMapValues(lambda brands: brands).collect()

4.5.4 countByKey / countByValue

1
rdd.map(lambda x: (x.year, x.brand)).countByKey()

4.6 合并与拆分列

4.6.1 合并多列

1
2
from pyspark.sql import Row
mtcars_rdd = df.rdd.map(lambda row: Row(model=row[0], values=row[1:]))

4.6.2 拆分列

1
2
mtcars_rdd2 = df.rdd.map(lambda row: Row(model=row[0], x1=row[1:5], x2=row[5:]))
new_df = spark.createDataFrame(mtcars_rdd2)

4.7 保存与加载

4.7.1 保存为CSV

1
df.coalesce(1).write.csv("output.csv", header=True, mode="overwrite")

4.7.2 保存为Parquet（推荐）

1
df.write.parquet("output.parquet", mode="overwrite")

4.7.3 加载Parquet

1
df = spark.read.parquet("output.parquet")

5. SQL查询

DataFrame可注册为临时视图，通过SQL进行复杂分析。

1
2
3
df.createOrReplaceTempView("people")
result = spark.sql("SELECT department, AVG(salary) FROM people GROUP BY department")
result.show()

Spark SQL支持大部分HiveQL语法，可与Hive元数据集成。评论：对于熟悉SQL的团队，SQL查询可降低学习成本；而DataFrame API则在编译时检查错误，适合程序化操作。两者可混合使用，视场景而定。

6. 用户自定义函数（UDF）

当内置函数无法满足需求时，可定义UDF。

6.1 普通UDF（性能较差）

1
2
3
4
5
6
7
8
from pyspark.sql.functions import udf
from pyspark.sql.types import StringType

def to_upper(s):
    return s.upper() if s else None

upper_udf = udf(to_upper, StringType())
df.withColumn("upper_name", upper_udf("name")).show()

6.2 Pandas UDF（矢量化，推荐）

1
2
3
4
5
6
from pyspark.sql.functions import pandas_udf
import pandas as pd

@pandas_udf(StringType())
def pandas_to_upper(s: pd.Series) -> pd.Series:
    return s.str.upper()

Pandas UDF使用Apache Arrow进行高效数据交换，性能提升显著。

7. 性能优化

7.1 缓存

1
2
df.cache()  # 或 df.persist(StorageLevel.MEMORY_AND_DISK)
df.unpersist()  # 释放缓存

7.2 分区控制

1
2
df.repartition(10)   # 增加分区（产生shuffle）
df.coalesce(5)       # 减少分区（避免全量shuffle）

7.3 广播变量

将小表广播到各节点，避免shuffle：

1
2
from pyspark.sql.functions import broadcast
df1.join(broadcast(df2), "key")

7.4 累加器

用于计数等聚合，但只能通过 SparkContext.accumulator 创建：

1
2
accum = sc.accumulator(0)
df.foreach(lambda row: accum.add(1) if row.value > 0 else None)

7.5 调整shuffle分区数

默认200，可通过配置调整：

1
spark.conf.set("spark.sql.shuffle.partitions", 50)

8. 机器学习（MLlib）

MLlib提供分布式机器学习算法，API与scikit-learn类似：

1
2
3
4
5
6
7
8
9
10
11
12
13
from pyspark.ml.feature import VectorAssembler
from pyspark.ml.regression import LinearRegression

# 准备特征向量
assembler = VectorAssembler(inputCols=["feature1", "feature2"], outputCol="features")
df = assembler.transform(df)

# 训练模型
lr = LinearRegression(featuresCol="features", labelCol="label")
model = lr.fit(df)

# 预测
predictions = model.transform(df)

MLlib支持Pipeline，方便构建完整工作流。注意：数据量极大时才体现出MLlib优势；小数据集上scikit-learn更轻量。

9. 流处理（Structured Streaming）

Structured Streaming基于DataFrame的流处理模型，支持事件时间、水印等：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
# 从Kafka读取流
stream_df = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "host:9092") \
    .option("subscribe", "topic") \
    .load()

# 转换
result = stream_df.selectExpr("CAST(value AS STRING)")

# 输出到控制台（调试）
query = result.writeStream \
    .outputMode("append") \
    .format("console") \
    .start()

query.awaitTermination()

评论：Structured Streaming提供 exactly-once 语义，容错机制完善。生产环境可输出到Parquet、Kafka等，并配合检查点实现故障恢复。

10. 总结

本文全面覆盖了PySpark的核心操作：从RDD和DataFrame的基础创建、转换、动作，到SQL查询、UDF、性能优化、机器学习和流处理。建议在实际应用中优先使用DataFrame API，充分利用Catalyst优化；RDD适合需要底层控制的场景。希望本文能成为你日常开发中的速查手册，助你高效处理大数据。更多细节可参考Spark官方文档。

展开全文 >>

(AI Generated)AI-Generated-Pandas速查指南

字数统计: 6.8k字   |   阅读时长≈ 36分

pandas速查：全面覆盖核心功能与高级技巧

1. 环境配置与基础

1.1 安装与依赖

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
# 基础安装
pip install pandas numpy matplotlib scipy

# 完整数据科学环境
pip install pandas numpy matplotlib seaborn scikit-learn jupyter

# 导入标准约定
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import warnings
warnings.filterwarnings('ignore')

# 设置显示选项
pd.set_option('display.max_columns', None)  # 显示所有列
pd.set_option('display.max_rows', 100)       # 最多显示100行
pd.set_option('display.float_format', '{:.2f}'.format)  # 格式化浮点数

1.2 核心数据结构详解

1.2.1 Series：带标签的一维数组

1
2
3
4
5
6
7
8
9
10
11
12
13
# 创建Series
s1 = pd.Series([1, 3, 5, np.nan, 6, 8])  # 从列表创建
s2 = pd.Series({'a': 1, 'b': 2, 'c': 3})  # 从字典创建
s3 = pd.Series(5, index=['a', 'b', 'c'])  # 标量值创建

# Series属性和方法
s1.values      # 底层numpy数组
s1.index       # 索引对象
s1.dtype       # 数据类型
s1.shape       # 形状
s1.size        # 元素数量
s1.name        # Series名称
s1.index.name  # 索引名称

1.2.2 DataFrame：二维表格型数据结构

1
2
3
4
5
6
7
8
# 多种创建方式
df1 = pd.DataFrame({'A': [1, 2, 3], 'B': [4, 5, 6]})  # 字典创建
df2 = pd.DataFrame(np.random.randn(3, 2), columns=['A', 'B'])  # numpy数组
df3 = pd.DataFrame([{'A': 1, 'B': 2}, {'A': 3, 'B': 4}])  # 字典列表

# 特殊DataFrame创建
pd.DataFrame.from_dict({'A': [1, 2], 'B': [3, 4]}, orient='index')
pd.DataFrame.from_records([(1, 'a'), (2, 'b')], columns=['id', 'name'])

1.2.3 Index：不可变的轴标签

1
2
3
4
5
# 索引类型
pd.Index([1, 2, 3])           # 基础索引
pd.RangeIndex(start=0, stop=5, step=1)  # 范围索引
pd.MultiIndex.from_tuples([('a', 1), ('b', 2)])  # 多级索引
pd.DatetimeIndex(['2023-01-01', '2023-01-02'])  # 时间索引

2. 数据创建与读取的高级技巧

2.1 动态数据结构创建

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# 从多个数据源合并创建
data_sources = {
    'sales': {'file': 'sales.csv', 'key': 'sale_id'},
    'customers': {'file': 'customers.csv', 'key': 'customer_id'},
    'products': {'file': 'products.csv', 'key': 'product_id'}
}

# 动态读取并合并
dfs = {}
for name, config in data_sources.items():
    df = pd.read_csv(config['file'])
    df['source'] = name  # 标记数据来源
    dfs[name] = df

# 创建统一的主数据表
master_df = pd.concat(dfs.values(), keys=dfs.keys(), names=['source', 'row_id'])

2.2 读取时直接处理管道

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
# 使用pipe的读取处理链
def create_processed_dataframe(filepath):
    """完整的读取-处理管道"""
    return (
        pd.read_csv(filepath)
        .pipe(lambda df: df if not df.empty else pd.DataFrame(columns=['default']))
        .pipe(standardize_column_names)
        .pipe(handle_missing_values, strategy='median')
        .pipe(encode_categorical_variables)
        .pipe(normalize_numeric_columns)
        .assign(processing_timestamp=pd.Timestamp.now())
        .set_index('id', verify_integrity=True)
    )

# 支持函数
def standardize_column_names(df):
    """标准化列名：小写，下划线分隔"""
    df.columns = df.columns.str.lower().str.replace(' ', '_')
    return df

def handle_missing_values(df, strategy='mean'):
    """处理缺失值"""
    numeric_cols = df.select_dtypes(include=[np.number]).columns
    if strategy == 'mean':
        df[numeric_cols] = df[numeric_cols].fillna(df[numeric_cols].mean())
    elif strategy == 'median':
        df[numeric_cols] = df[numeric_cols].fillna(df[numeric_cols].median())
    return df

2.3 读取不同类型文件的高级参数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
# CSV高级读取
df = pd.read_csv(
    'data.csv',
    encoding='utf-8-sig',  # 处理BOM
    sep=',',               # 分隔符
    header=0,              # 表头行
    index_col=0,           # 索引列
    usecols=['col1', 'col2'],  # 指定列
    dtype={'col1': 'int32', 'col2': 'category'},  # 数据类型
    parse_dates=['date_col'],  # 解析日期
    date_parser=lambda x: pd.to_datetime(x, format='%Y-%m-%d'),
    na_values=['NA', 'NULL', 'NaN', ''],  # 缺失值标识
    true_values=['Yes', 'True'],  # 布尔值转换
    false_values=['No', 'False'],
    skiprows=lambda x: x % 2 == 0,  # 跳过偶数行
    nrows=1000,  # 读取行数限制
    skipfooter=10,  # 跳过尾部行
    converters={'col': lambda x: x.strip()},  # 列转换器
    memory_map=True,  # 内存映射提高大文件读取速度
    verbose=True      # 显示读取信息
)

# Excel多工作表处理
excel_file = pd.ExcelFile('data.xlsx')
sheet_names = excel_file.sheet_names
dfs = {sheet: excel_file.parse(sheet) for sheet in sheet_names}

# 或使用read_excel的高级参数
df = pd.read_excel(
    'data.xlsx',
    sheet_name=None,  # 读取所有工作表
    header=[0, 1],    # 多级表头
    skiprows=range(1, 5),  # 跳过前5行
    usecols='A:C,E:G',  # 读取指定列
    converters={'Price': lambda x: x * 1.1}  # 转换函数
)

3. 数据查看与探索的全面方法

3.1 数据概览与统计

1
2
3
4
5
6
7
8
9
10
11
12
# 基础概览
df.info(verbose=True, memory_usage='deep')  # 详细内存使用
df.describe(include='all', percentiles=[.1, .5, .9])  # 完整统计
df.describe(exclude=[np.number])  # 排除数值列

# 内存使用分析
df.memory_usage(deep=True).sum()  # 总内存使用
df.memory_usage(deep=True).sort_values(ascending=False)  # 各列内存使用

# 数据类型分析
df.dtypes.value_counts()  # 各类型数量
df.select_dtypes(include=['object']).nunique()  # 文本列唯一值

3.2 数据质量检查

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
# 完整性检查
def check_data_quality(df):
    """全面数据质量检查"""
    quality_report = pd.DataFrame({
        'column': df.columns,
        'dtype': df.dtypes.values,
        'non_null_count': df.count().values,
        'null_count': df.isnull().sum().values,
        'null_percentage': (df.isnull().sum() / len(df) * 100).values,
        'unique_values': df.nunique().values,
        'most_frequent': [df[col].mode().iloc[0] if not df[col].mode().empty else None 
                         for col in df.columns],
        'most_freq_count': [df[col].value_counts().iloc[0] if not df[col].empty else 0 
                           for col in df.columns]
    })
    
    # 添加数据质量标记
    quality_report['quality_score'] = (
        (quality_report['null_percentage'] < 5).astype(int) * 3 +  # 缺失值少
        (quality_report['unique_values'] > 1).astype(int) * 2 +   # 有变化
        (quality_report['unique_values'] < len(df) * 0.9).astype(int) * 2  # 不过于分散
    )
    
    return quality_report.sort_values('quality_score', ascending=False)

# 执行检查
quality_df = check_data_quality(df)

4. 数据选择与切片的深度解析

4.1 选择方法的性能比较

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
# 各种选择方法的速度比较
import timeit

# 1. 直接索引（最快）
def method1(df):
    return df['column']

# 2. loc选择
def method2(df):
    return df.loc[:, 'column']

# 3. iloc选择
def method3(df):
    return df.iloc[:, 0]

# 4. 属性访问（仅限有效Python变量名）
def method4(df):
    return df.column

# 性能测试
df = pd.DataFrame(np.random.randn(1000000, 10), columns=[f'col_{i}' for i in range(10)])
for i, method in enumerate([method1, method2, method3, method4], 1):
    time_taken = timeit.timeit(lambda: method(df), number=100)
    print(f"Method {i}: {time_taken:.4f} seconds")

4.2 高级索引技巧

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
# 多重条件索引
df.query('A > 0 and B < 0')  # 使用query方法
df.loc[(df['A'] > 0) & (df['B'] < 0)]  # 布尔索引

# 使用between
df[df['value'].between(10, 20)]  # 包含边界
df[df['value'].between(10, 20, inclusive='neither')]  # 不包含边界

# 使用isin的多个条件
valid_categories = ['A', 'B', 'C']
valid_statuses = ['active', 'pending']
df[df['category'].isin(valid_categories) & df['status'].isin(valid_statuses)]

# 正则表达式索引
df.filter(regex='^user_')  # 选择以user_开头的列
df.filter(like='_count')   # 选择包含_count的列
df.filter(items=['col1', 'col2'])  # 精确选择

# 基于数据类型的索引
df.select_dtypes(include=['int64', 'float64'])  # 选择数值列
df.select_dtypes(exclude=['object', 'category'])  # 排除文本和分类列

4.3 多层索引的选择

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
# 创建多层索引
index = pd.MultiIndex.from_product([['A', 'B'], [1, 2]], names=['letter', 'number'])
df = pd.DataFrame({'value': np.random.randn(4)}, index=index)

# 多层索引选择
df.xs('A', level='letter')  # 选择字母为A的所有行
df.xs(1, level='number')    # 选择数字为1的所有行
df.loc[('A', 1)]            # 选择特定组合
df.loc[('A', slice(None))]  # 选择字母A的所有数字
df.loc[(slice(None), 1), :]  # 选择数字1的所有字母

# 使用IndexSlice
idx = pd.IndexSlice
df.loc[idx['A', :], :]  # 选择字母A的所有
df.loc[idx[:, 1], :]    # 选择数字1的所有

5. 数据清洗与预处理的专业技巧

5.1 高级缺失值处理

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
# 缺失值模式分析
import seaborn as sns
import matplotlib.pyplot as plt

def analyze_missing_patterns(df):
    """分析缺失值模式"""
    # 创建缺失值矩阵
    missing_matrix = df.isnull()
    
    # 可视化缺失值
    plt.figure(figsize=(10, 6))
    sns.heatmap(missing_matrix, cbar=False, cmap='viridis')
    plt.title('Missing Values Pattern')
    plt.show()
    
    # 缺失值相关性
    missing_corr = missing_matrix.corr()
    
    # 缺失值模式聚类
    from scipy.cluster import hierarchy
    linked = hierarchy.linkage(missing_corr, 'single')
    
    plt.figure(figsize=(10, 6))
    hierarchy.dendrogram(linked, labels=missing_corr.columns)
    plt.title('Missing Value Pattern Clustering')
    plt.show()
    
    return missing_corr

# 高级填充策略
def advanced_fillna(df):
    """基于数据关系的填充策略"""
    # 使用分组均值填充
    df_filled = df.copy()
    
    for col in df.columns:
        if df[col].isnull().any():
            # 如果有分组列，使用分组均值
            if 'group' in df.columns:
                df_filled[col] = df.groupby('group')[col].transform(
                    lambda x: x.fillna(x.mean())
                )
            # 使用时间序列方法填充时间相关数据
            elif 'date' in df.columns:
                df_filled[col] = df[col].interpolate(method='time')
            # 使用KNN填充
            else:
                from sklearn.impute import KNNImputer
                imputer = KNNImputer(n_neighbors=5)
                df_filled[col] = imputer.fit_transform(df[[col]])
    
    return df_filled

5.2 异常值检测与处理

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
def detect_and_handle_outliers(df, method='iqr', threshold=1.5):
    """异常值检测与处理"""
    numeric_cols = df.select_dtypes(include=[np.number]).columns
    
    outliers_info = {}
    
    for col in numeric_cols:
        if method == 'iqr':
            # IQR方法
            Q1 = df[col].quantile(0.25)
            Q3 = df[col].quantile(0.75)
            IQR = Q3 - Q1
            lower_bound = Q1 - threshold * IQR
            upper_bound = Q3 + threshold * IQR
            
            outliers = df[(df[col] < lower_bound) | (df[col] > upper_bound)]
        
        elif method == 'zscore':
            # Z-score方法
            from scipy import stats
            z_scores = np.abs(stats.zscore(df[col].dropna()))
            outliers = df[z_scores > threshold]
        
        elif method == 'mad':
            # MAD方法（中位数绝对偏差）
            median = df[col].median()
            mad = np.median(np.abs(df[col] - median))
            modified_z_scores = 0.6745 * (df[col] - median) / mad
            outliers = df[np.abs(modified_z_scores) > threshold]
        
        outliers_info[col] = {
            'count': len(outliers),
            'percentage': len(outliers) / len(df) * 100,
            'indices': outliers.index.tolist(),
            'values': outliers[col].tolist()
        }
    
    return outliers_info

# 异常值处理策略
def handle_outliers_strategy(df, strategy='cap', threshold=1.5):
    """处理异常值的多种策略"""
    df_clean = df.copy()
    numeric_cols = df.select_dtypes(include=[np.number]).columns
    
    for col in numeric_cols:
        Q1 = df[col].quantile(0.25)
        Q3 = df[col].quantile(0.75)
        IQR = Q3 - Q1
        lower_bound = Q1 - threshold * IQR
        upper_bound = Q3 + threshold * IQR
        
        if strategy == 'cap':
            # 封顶法：将异常值替换为边界值
            df_clean[col] = df_clean[col].clip(lower_bound, upper_bound)
        
        elif strategy == 'remove':
            # 删除法：删除异常值
            df_clean = df_clean[(df_clean[col] >= lower_bound) & 
                               (df_clean[col] <= upper_bound)]
        
        elif strategy == 'transform':
            # 变换法：使用对数变换
            df_clean[col] = np.log1p(df_clean[col] - df_clean[col].min() + 1)
        
        elif strategy == 'winsorize':
            # 缩尾法
            from scipy.stats.mstats import winsorize
            df_clean[col] = winsorize(df_clean[col], limits=[0.05, 0.05])
    
    return df_clean

5.3 数据标准化与编码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
from sklearn.preprocessing import StandardScaler, MinMaxScaler, RobustScaler
from sklearn.preprocessing import LabelEncoder, OneHotEncoder

def prepare_features(df, numeric_cols, categorical_cols):
    """特征工程：标准化和编码"""
    df_prepared = df.copy()
    
    # 数值特征标准化
    if numeric_cols:
        # 选择标准化方法
        scaler = StandardScaler()  # 或 MinMaxScaler(), RobustScaler()
        df_prepared[numeric_cols] = scaler.fit_transform(df[numeric_cols])
    
    # 分类特征编码
    if categorical_cols:
        # 对有序分类变量使用LabelEncoder
        for col in categorical_cols:
            if df[col].dtype.name == 'category' and df[col].cat.ordered:
                le = LabelEncoder()
                df_prepared[col] = le.fit_transform(df[col])
            else:
                # 对无序分类变量使用One-Hot编码
                dummies = pd.get_dummies(df[col], prefix=col, drop_first=True)
                df_prepared = pd.concat([df_prepared, dummies], axis=1)
                df_prepared = df_prepared.drop(col, axis=1)
    
    return df_prepared

6. 数据变形与重塑：pipe的全面应用

6.1 pipe方法深度解析

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
# pipe的基本用法
df.pipe(func, *args, **kwargs)

# pipe链式操作示例
result = (df
          .pipe(clean_data)
          .pipe(transform_data, method='log')
          .pipe(filter_data, min_value=0)
          .pipe(aggregate_data, by='category')
          .pipe(format_output, precision=2))

# 实际应用：创建数据处理管道类
class DataPipeline:
    def __init__(self, df):
        self.df = df.copy()
    
    @staticmethod
    def remove_outliers(df, column, threshold=3):
        """移除异常值"""
        z_scores = np.abs((df[column] - df[column].mean()) / df[column].std())
        return df[z_scores < threshold]
    
    @staticmethod
    def add_features(df):
        """添加衍生特征"""
        return df.assign(
            log_value=lambda x: np.log1p(x['value']),
            scaled_value=lambda x: (x['value'] - x['value'].min()) / 
                                   (x['value'].max() - x['value'].min()),
            interaction=lambda x: x['feature1'] * x['feature2']
        )
    
    @staticmethod
    def encode_categorical(df, columns):
        """编码分类变量"""
        for col in columns:
            if df[col].dtype == 'object':
                df[col] = pd.Categorical(df[col]).codes
        return df
    
    def run(self):
        """运行完整管道"""
        return (self.df
                .pipe(self.remove_outliers, 'value', threshold=2.5)
                .pipe(self.add_features)
                .pipe(self.encode_categorical, ['category', 'status'])
                .pipe(lambda x: x.dropna())
                .pipe(lambda x: x.reset_index(drop=True)))

# 使用管道
pipeline = DataPipeline(raw_df)
processed_df = pipeline.run()

6.2 复杂数据变形操作

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
# 1. 数据透视表的高级用法
pivot = df.pivot_table(
    values=['sales', 'profit'],
    index=['region', pd.Grouper(key='date', freq='M')],
    columns='product_category',
    aggfunc={'sales': 'sum', 'profit': 'mean'},
    fill_value=0,
    margins=True,
    margins_name='Total',
    observed=True
)

# 2. 交叉表的自定义
cross_tab = pd.crosstab(
    index=[df['year'], df['quarter']],
    columns=df['product_line'],
    values=df['revenue'],
    aggfunc='sum',
    normalize='index'  # 或 'columns', 'all'
)

# 3. 数据堆叠与解堆叠
# 宽表转长表
long_df = df.melt(
    id_vars=['id', 'name'],
    value_vars=['2020', '2021', '2022'],
    var_name='year',
    value_name='value'
)

# 长表转宽表
wide_df = long_df.pivot(
    index=['id', 'name'],
    columns='year',
    values='value'
).reset_index()

# 4. 多重索引的堆叠/解堆叠
df_multi = df.set_index(['region', 'product', 'date'])
stacked = df_multi.stack().reset_index()  # 堆叠
unstacked = stacked.set_index(['region', 'product', 'date']).unstack()  # 解堆叠

6.3 分组聚合的高级技巧

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
# 1. 自定义聚合函数
def weighted_average(df, value_col, weight_col):
    """计算加权平均值"""
    return np.average(df[value_col], weights=df[weight_col])

def top_n_values(series, n=3):
    """获取前N个值及其频率"""
    return series.value_counts().head(n).to_dict()

# 应用自定义聚合
agg_results = df.groupby('category').agg({
    'value': ['mean', 'std', lambda x: weighted_average(df, 'value', 'weight')],
    'name': lambda x: top_n_values(x, n=2)
})

# 2. 分组后应用复杂转换
def normalize_group(df):
    """组内标准化"""
    return (df - df.mean()) / df.std()

# 使用transform
df['normalized'] = df.groupby('group')['value'].transform(normalize_group)

# 3. 分组过滤
# 筛选组内数量大于阈值的组
filtered = df.groupby('category').filter(lambda x: len(x) > 10)

# 筛选组内平均值大于阈值的组
filtered = df.groupby('category').filter(lambda x: x['value'].mean() > 50)

# 4. 分组排名
df['rank_in_group'] = df.groupby('group')['score'].rank(method='dense', ascending=False)

7. 时间序列处理的完整指南

7.1 时间数据操作

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
# 创建时间范围
date_ranges = {
    'daily': pd.date_range('2023-01-01', periods=30, freq='D'),
    'business_daily': pd.date_range('2023-01-01', periods=30, freq='B'),
    'hourly': pd.date_range('2023-01-01', periods=24, freq='H'),
    'monthly': pd.date_range('2023-01-01', periods=12, freq='M'),
    'quarterly': pd.date_range('2023-01-01', periods=4, freq='Q'),
    'yearly': pd.date_range('2020-01-01', '2023-01-01', freq='Y')
}

# 时间属性提取
df['date'] = pd.to_datetime(df['date_string'])
df['year'] = df['date'].dt.year
df['quarter'] = df['date'].dt.quarter
df['month'] = df['date'].dt.month
df['week'] = df['date'].dt.isocalendar().week
df['day'] = df['date'].dt.day
df['dayofweek'] = df['date'].dt.dayofweek  # 周一=0
df['is_weekend'] = df['date'].dt.dayofweek >= 5
df['hour'] = df['date'].dt.hour
df['is_month_start'] = df['date'].dt.is_month_start
df['is_quarter_end'] = df['date'].dt.is_quarter_end

# 时间差计算
df['days_since_event'] = (df['date'] - pd.Timestamp('2023-01-01')).dt.days
df['hours_diff'] = df['end_time'] - df['start_time']
df['hours_diff'] = df['hours_diff'].dt.total_seconds() / 3600

7.2 时间序列分析

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
# 重采样分析
resample_methods = {
    'daily_sum': df.resample('D', on='date').sum(),
    'weekly_mean': df.resample('W-MON', on='date').mean(),  # 周一为每周起始
    'monthly_std': df.resample('M', on='date').std(),
    'quarterly_first': df.resample('Q', on='date').first(),
    'hourly_interpolated': df.resample('H', on='date').interpolate(method='linear')
}

# 滚动窗口统计
window_stats = {
    '7d_rolling_mean': df['value'].rolling(window='7D', min_periods=1).mean(),
    '30d_rolling_std': df['value'].rolling(window='30D', min_periods=7).std(),
    'exp_weighted_mean': df['value'].ewm(span=30, adjust=False).mean(),
    'expanding_mean': df['value'].expanding(min_periods=1).mean(),
    'centered_rolling': df['value'].rolling(window='14D', center=True).mean()
}

# 时间序列差分
df['value_diff'] = df['value'].diff()  # 一阶差分
df['value_pct_change'] = df['value'].pct_change()  # 百分比变化
df['value_log_diff'] = np.log(df['value']).diff()  # 对数差分

# 季节性分解
from statsmodels.tsa.seasonal import seasonal_decompose

def decompose_timeseries(df, column, freq=365):
    """时间序列分解"""
    decomposition = seasonal_decompose(
        df[column],
        model='additive',  # 或 'multiplicative'
        period=freq,
        extrapolate_trend='freq'
    )
    
    return {
        'observed': decomposition.observed,
        'trend': decomposition.trend,
        'seasonal': decomposition.seasonal,
        'residual': decomposition.resid
    }

8. 性能优化与内存管理

8.1 内存优化策略

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
def optimize_memory(df):
    """优化DataFrame内存使用"""
    df_optimized = df.copy()
    
    # 数值列优化
    for col in df.select_dtypes(include=[np.number]).columns:
        col_min = df[col].min()
        col_max = df[col].max()
        
        # 整数优化
        if np.issubdtype(df[col].dtype, np.integer):
            if col_min >= 0:
                if col_max < 255:
                    df_optimized[col] = df[col].astype(np.uint8)
                elif col_max < 65535:
                    df_optimized[col] = df[col].astype(np.uint16)
                elif col_max < 4294967295:
                    df_optimized[col] = df[col].astype(np.uint32)
                else:
                    df_optimized[col] = df[col].astype(np.uint64)
            else:
                if col_min > -128 and col_max < 127:
                    df_optimized[col] = df[col].astype(np.int8)
                elif col_min > -32768 and col_max < 32767:
                    df_optimized[col] = df[col].astype(np.int16)
                elif col_min > -2147483648 and col_max < 2147483647:
                    df_optimized[col] = df[col].astype(np.int32)
                else:
                    df_optimized[col] = df[col].astype(np.int64)
        # 浮点数优化
        else:
            if col_min > np.finfo(np.float16).min and col_max < np.finfo(np.float16).max:
                df_optimized[col] = df[col].astype(np.float16)
            elif col_min > np.finfo(np.float32).min and col_max < np.finfo(np.float32).max:
                df_optimized[col] = df[col].astype(np.float32)
    
    # 对象列优化
    for col in df.select_dtypes(include=['object']).columns:
        num_unique = df[col].nunique()
        num_total = len(df[col])
        
        if num_unique / num_total < 0.5:  # 唯一值比例小于50%
            df_optimized[col] = df[col].astype('category')
    
    # 日期时间优化
    for col in df.select_dtypes(include=['datetime64']).columns:
        df_optimized[col] = pd.to_datetime(df[col], format='%Y-%m-%d')
    
    return df_optimized

# 内存使用比较
original_memory = df.memory_usage(deep=True).sum()
optimized_df = optimize_memory(df)
optimized_memory = optimized_df.memory_usage(deep=True).sum()
print(f"内存节省: {(original_memory - optimized_memory) / original_memory * 100:.1f}%")

8.2 性能优化技巧

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
# 1. 使用向量化操作替代循环
# 避免
for i in range(len(df)):
    df.loc[i, 'new_col'] = df.loc[i, 'col1'] * df.loc[i, 'col2']

# 推荐
df['new_col'] = df['col1'] * df['col2']

# 2. 使用eval进行表达式求值（大数据集）
result = df.eval('col1 + col2 * col3')

# 3. 使用query进行高效过滤
filtered = df.query('col1 > 0 and col2 < 0 and col3 in [1, 2, 3]')

# 4. 分块处理大数据
def process_large_file(filepath, chunk_size=10000):
    """分块处理大文件"""
    chunks = []
    for chunk in pd.read_csv(filepath, chunksize=chunk_size):
        # 对每个块进行处理
        processed_chunk = process_chunk(chunk)
        chunks.append(processed_chunk)
    
    return pd.concat(chunks, ignore_index=True)

# 5. 使用Dask处理超大数据集
import dask.dataframe as dd
dask_df = dd.read_csv('large_dataset.csv')
result = dask_df.groupby('category').mean().compute()

9. 数据导出与共享

9.1 多种格式导出

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
# 1. CSV导出
df.to_csv(
    'output.csv',
    index=False,                # 不保存索引
    encoding='utf-8-sig',       # 支持Excel中文
    sep=',',                    # 分隔符
    float_format='%.2f',        # 浮点数格式
    date_format='%Y-%m-%d',     # 日期格式
    quoting=csv.QUOTE_NONNUMERIC,  # 引用非数字
    compression='gzip'          # 压缩
)

# 2. Excel导出（多个工作表）
with pd.ExcelWriter('output.xlsx', engine='openpyxl') as writer:
    df1.to_excel(writer, sheet_name='Sheet1', index=False)
    df2.to_excel(writer, sheet_name='Sheet2', index=False)
    
    # 添加格式
    workbook = writer.book
    worksheet = writer.sheets['Sheet1']
    
    # 设置列宽
    for column in worksheet.columns:
        max_length = 0
        column_letter = column[0].column_letter
        for cell in column:
            try:
                if len(str(cell.value)) > max_length:
                    max_length = len(cell.value)
            except:
                pass
        adjusted_width = (max_length + 2)
        worksheet.column_dimensions[column_letter].width = adjusted_width

# 3. Parquet格式（高效存储）
df.to_parquet(
    'output.parquet',
    engine='pyarrow',
    compression='snappy',  # 或 'gzip', 'brotli'
    index=False
)

# 4. 数据库导出
df.to_sql(
    'table_name',
    con=engine,
    if_exists='replace',  # 或 'append', 'fail'
    index=False,
    chunksize=1000,       # 分批插入
    dtype={'col1': sqlalchemy.types.VARCHAR(255)}
)

# 5. JSON导出
df.to_json(
    'output.json',
    orient='records',      # 或 'split', 'index', 'columns', 'values', 'table'
    date_format='iso',
    double_precision=2,
    force_ascii=False
)

# 6. HTML导出（网页展示）
df.to_html(
    'output.html',
    index=False,
    classes='table table-striped table-bordered',
    border=0,
    float_format='{:,.2f}'.format,
    encoding='utf-8'
)

10. 实战应用：完整的数据分析管道

10.1 端到端分析项目

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
class CompleteDataAnalysis:
    """完整的数据分析管道"""
    
    def __init__(self, data_path):
        self.data_path = data_path
        self.results = {}
    
    def load_and_explore(self):
        """加载和探索数据"""
        # 智能检测文件类型并加载
        if self.data_path.endswith('.csv'):
            self.df = pd.read_csv(self.data_path, low_memory=False)
        elif self.data_path.endswith('.parquet'):
            self.df = pd.read_parquet(self.data_path)
        elif self.data_path.endswith('.xlsx'):
            self.df = pd.read_excel(self.data_path, sheet_name=None)
        
        # 基础探索
        self.results['shape'] = self.df.shape
        self.results['dtypes'] = self.df.dtypes.value_counts().to_dict()
        self.results['missing'] = self.df.isnull().sum().to_dict()
        self.results['stats'] = self.df.describe().to_dict()
        
        return self
    
    def clean_data(self):
        """数据清洗"""
        # 使用pipe链式清洗
        self.df = (self.df
                   .pipe(self._remove_duplicates)
                   .pipe(self._handle_missing)
                   .pipe(self._fix_data_types)
                   .pipe(self._remove_outliers))
        
        self.results['cleaning_report'] = {
            'rows_removed': self.results['shape'][0] - len(self.df),
            'columns_cleaned': list(self.df.columns)
        }
        
        return self
    
    def engineer_features(self):
        """特征工程"""
        # 添加时间特征
        if 'date' in self.df.columns:
            self.df = self.df.pipe(self._add_time_features)
        
        # 添加聚合特征
        self.df = self.df.pipe(self._add_aggregate_features)
        
        # 添加交互特征
        self.df = self.df.pipe(self._add_interaction_features)
        
        self.results['features_added'] = [
            col for col in self.df.columns if col not in self.original_columns
        ]
        
        return self
    
    def analyze(self):
        """数据分析"""
        # 相关性分析
        self.results['correlation'] = self.df.corr().to_dict()
        
        # 分组分析
        if 'category' in self.df.columns:
            self.results['group_analysis'] = self.df.groupby('category').agg({
                'value': ['mean', 'std', 'count']
            }).to_dict()
        
        # 趋势分析
        if 'date' in self.df.columns:
            self.results['trend_analysis'] = self._analyze_trends()
        
        return self
    
    def visualize(self):
        """数据可视化"""
        import matplotlib.pyplot as plt
        import seaborn as sns
        
        # 创建可视化
        fig, axes = plt.subplots(2, 2, figsize=(15, 10))
        
        # 1. 分布图
        if 'value' in self.df.columns:
            self.df['value'].hist(ax=axes[0, 0], bins=30)
            axes[0, 0].set_title('Value Distribution')
        
        # 2. 相关性热图
        numeric_cols = self.df.select_dtypes(include=[np.number]).columns
        if len(numeric_cols) > 1:
            sns.heatmap(self.df[numeric_cols].corr(), 
                       annot=True, fmt='.2f', 
                       ax=axes[0, 1], cmap='coolwarm')
            axes[0, 1].set_title('Correlation Heatmap')
        
        # 3. 时间序列图
        if 'date' in self.df.columns and 'value' in self.df.columns:
            self.df.set_index('date')['value'].plot(ax=axes[1, 0])
            axes[1, 0].set_title('Time Series Plot')
        
        # 4. 箱线图
        if 'category' in self.df.columns and 'value' in self.df.columns:
            sns.boxplot(x='category', y='value', data=self.df, ax=axes[1, 1])
            axes[1, 1].set_title('Value by Category')
            axes[1, 1].tick_params(axis='x', rotation=45)
        
        plt.tight_layout()
        plt.savefig('analysis_results.png', dpi=300, bbox_inches='tight')
        plt.show()
        
        return self
    
    def export_results(self):
        """导出结果"""
        # 导出处理后的数据
        self.df.to_csv('processed_data.csv', index=False)
        
        # 导出分析结果
        import json
        with open('analysis_results.json', 'w') as f:
            json.dump(self.results, f, indent=2, default=str)
        
        # 导出报告
        self._generate_report()
        
        return self
    
    # 辅助方法
    def _remove_duplicates(self, df):
        return df.drop_duplicates()
    
    def _handle_missing(self, df):
        # 根据列类型处理缺失值
        for col in df.columns:
            if df[col].dtype in [np.float64, np.int64]:
                df[col] = df[col].fillna(df[col].median())
            elif df[col].dtype == 'object':
                df[col] = df[col].fillna(df[col].mode()[0] if not df[col].mode().empty else 'Unknown')
        return df
    
    def _add_time_features(self, df):
        df['date'] = pd.to_datetime(df['date'])
        df['year'] = df['date'].dt.year
        df['month'] = df['date'].dt.month
        df['day'] = df['date'].dt.day
        df['dayofweek'] = df['date'].dt.dayofweek
        df['is_weekend'] = df['date'].dt.dayofweek >= 5
        return df

# 使用完整管道
analysis = CompleteDataAnalysis('data.csv')
(analysis.load_and_explore()
         .clean_data()
         .engineer_features()
         .analyze()
         .visualize()
         .export_results())

10.2 生产级数据处理框架

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
from abc import ABC, abstractmethod
from typing import Dict, Any, List, Optional
import logging

class DataProcessor(ABC):
    """数据处理器的抽象基类"""
    
    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.logger = logging.getLogger(self.__class__.__name__)
        self.results = {}
    
    @abstractmethod
    def process(self, data: pd.DataFrame) -> pd.DataFrame:
        pass
    
    def validate_input(self, data: pd.DataFrame) -> bool:
        """验证输入数据"""
        required_columns = self.config.get('required_columns', [])
        missing_cols = [col for col in required_columns if col not in data.columns]
        
        if missing_cols:
            self.logger.error(f"Missing required columns: {missing_cols}")
            return False
        
        return True

class DataPipeline:
    """生产级数据管道"""
    
    def __init__(self):
        self.processors: List[DataProcessor] = []
        self.data: Optional[pd.DataFrame] = None
        self.history: List[Dict[str, Any]] = []
    
    def add_processor(self, processor: DataProcessor) -> 'DataPipeline':
        """添加处理器"""
        self.processors.append(processor)
        return self
    
    def run(self, input_data: pd.DataFrame) -> pd.DataFrame:
        """运行管道"""
        self.data = input_data.copy()
        
        for i, processor in enumerate(self.processors):
            self.logger.info(f"Running processor {i+1}/{len(self.processors)}: {processor.__class__.__name__}")
            
            # 记录处理前状态
            snapshot = {
                'processor': processor.__class__.__name__,
                'rows_before': len(self.data),
                'columns_before': list(self.data.columns),
                'timestamp': pd.Timestamp.now()
            }
            
            # 执行处理
            try:
                self.data = processor.process(self.data)
                snapshot['success'] = True
            except Exception as e:
                self.logger.error(f"Processor failed: {e}")
                snapshot['success'] = False
                snapshot['error'] = str(e)
                raise
            
            # 记录处理后状态
            snapshot.update({
                'rows_after': len(self.data),
                'columns_after': list(self.data.columns),
                'processor_results': processor.results
            })
            
            self.history.append(snapshot)
        
        return self.data
    
    def get_report(self) -> Dict[str, Any]:
        """生成处理报告"""
        return {
            'total_processors': len(self.processors),
            'successful_processors': sum(1 for h in self.history if h['success']),
            'final_shape': self.data.shape if self.data is not None else None,
            'processing_history': self.history
        }

# 具体处理器实现
class MissingValueProcessor(DataProcessor):
    """缺失值处理器"""
    
    def process(self, data: pd.DataFrame) -> pd.DataFrame:
        strategies = self.config.get('strategies', {})
        
        for col in data.columns:
            if col in strategies:
                strategy = strategies[col]
                
                if strategy['method'] == 'mean':
                    data[col] = data[col].fillna(data[col].mean())
                elif strategy['method'] == 'median':
                    data[col] = data[col].fillna(data[col].median())
                elif strategy['method'] == 'mode':
                    data[col] = data[col].fillna(data[col].mode()[0])
                elif strategy['method'] == 'ffill':
                    data[col] = data[col].fillna(method='ffill')
                elif strategy['method'] == 'interpolate':
                    data[col] = data[col].interpolate()
                elif strategy['method'] == 'constant':
                    data[col] = data[col].fillna(strategy['value'])
        
        # 记录处理结果
        self.results['missing_values_filled'] = data.isnull().sum().to_dict()
        
        return data

class OutlierProcessor(DataProcessor):
    """异常值处理器"""
    
    def process(self, data: pd.DataFrame) -> pd.DataFrame:
        method = self.config.get('method', 'iqr')
        threshold = self.config.get('threshold', 1.5)
        
        numeric_cols = data.select_dtypes(include=[np.number]).columns
        
        for col in numeric_cols:
            if method == 'iqr':
                Q1 = data[col].quantile(0.25)
                Q3 = data[col].quantile(0.75)
                IQR = Q3 - Q1
                lower_bound = Q1 - threshold * IQR
                upper_bound = Q3 + threshold * IQR
                
                # 记录异常值
                outliers = data[(data[col] < lower_bound) | (data[col] > upper_bound)]
                self.results[f'{col}_outliers'] = {
                    'count': len(outliers),
                    'indices': outliers.index.tolist()
                }
                
                # 处理异常值
                if self.config.get('action') == 'remove':
                    data = data[(data[col] >= lower_bound) & (data[col] <= upper_bound)]
                elif self.config.get('action') == 'cap':
                    data[col] = data[col].clip(lower_bound, upper_bound)
        
        return data

# 使用生产级管道
pipeline = DataPipeline()

# 配置处理器
missing_config = {
    'strategies': {
        'age': {'method': 'median'},
        'salary': {'method': 'mean'},
        'department': {'method': 'mode'}
    }
}

outlier_config = {
    'method': 'iqr',
    'threshold': 3,
    'action': 'cap'
}

# 添加处理器
pipeline.add_processor(MissingValueProcessor(missing_config))
pipeline.add_processor(OutlierProcessor(outlier_config))

# 运行管道
result = pipeline.run(raw_data)
report = pipeline.get_report()

总结

这份全面的pandas速查笔记覆盖了从基础到高级的各种功能，特别强调了：

  1. pipe方法的强大应用：通过管道模式实现清晰的数据处理流程
  2. 性能优化策略：包括内存优化、向量化操作和分块处理
  3. 生产级代码模式：使用面向对象设计和错误处理
  4. 完整的数据分析流程：从数据加载到结果导出的端到端解决方案
  5. 高级时间序列处理：包括重采样、滚动窗口和季节性分解
  6. 数据质量保证：全面的数据验证和清洗策略

掌握这些技巧将使你能够高效地处理各种数据分析任务，从简单的数据清洗到复杂的生产级数据处理管道。记住，良好的数据习惯（如数据验证、错误处理和文档记录）与掌握技术技巧同样重要。

展开全文 >>

(AI Generated)NumPy速查指南：从入门到精通

字数统计: 10.2k字   |   阅读时长≈ 46分

NumPy速查指南：从入门到精通

摘要

NumPy（Numerical Python）是Python科学计算生态系统的基石，提供了高性能的多维数组对象和丰富的数学函数库。本文作为一份全面的NumPy速查指南，系统性地介绍了NumPy的核心概念、常用操作和高级特性，每个部分都包含详细的说明、使用场景分析和实用注意事项。无论是数据科学、机器学习还是科学计算领域的研究人员和工程师，本文都能为您提供快速参考和深入指导。

1. 引言

简要说明：NumPy是Python中用于科学计算的基础包，几乎所有数据科学工具都构建于其上。它提供了高效的N维数组对象、复杂的广播功能以及整合低级语言代码的能力。

重要性评价：

  • 性能优势：NumPy的数组操作在底层用C语言实现，比纯Python列表操作快10-100倍
  • 内存效率：NumPy数组在内存中连续存储，相比Python列表节省大量内存
  • 生态基础：Pandas、SciPy、Scikit-learn、TensorFlow等库都依赖NumPy

使用场景：

  • 数值计算和数学运算
  • 数据预处理和清洗
  • 机器学习算法的实现
  • 图像处理和多维数据操作
  • 科学研究和工程计算
1
2
3
4
5
import numpy as np  # 行业标准导入方式

# 验证安装和版本
print(f"NumPy版本: {np.__version__}")
print(f"NumPy配置: {np.__config__.show()}")

注意事项:

  • 始终使用import numpy as np约定，保证代码一致性
  • Numpy操作通常返回新数组，而非原地修改（除非特殊说明）

    2. 数组创建

    简要说明：NumPy提供了多种创建数组的方法，从简单的列表转换到复杂的特殊矩阵生成。
    性能考量：
  • 使用NumPy内置函数（如np.zeros()、np.ones()）比从Python列表转换更快
  • 对于大型数组，使用np.empty()可以避免初始化开销，但要小心未初始化的值

2.1 基础创建方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
# 从Python列表/元组创建 - 最常用方法
list_data = [1, 2, 3, 4, 5]
tuple_data = (6, 7, 8, 9, 10)

arr_from_list = np.array(list_data)        # 一维数组
arr_from_tuple = np.array(tuple_data)      # 同样可以

# 创建二维数组
matrix = np.array([[1, 2, 3], [4, 5, 6]])
print(f"二维数组形状: {matrix.shape}")  # (2, 3)

# 指定数据类型 - 重要优化手段
arr_float32 = np.array([1, 2, 3], dtype=np.float32)  # 32位浮点数
arr_int16 = np.array([1, 2, 3], dtype=np.int16)      # 16位整数

使用场景：

  • np.array()：从已有Python数据创建数组
  • 明确指定dtype可以减少内存使用并提高计算速度

2.2 特殊数组生成

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
# 生成全零数组 - 常用初始化方法
zeros_1d = np.zeros(5)                     # [0., 0., 0., 0., 0.]
zeros_2d = np.zeros((3, 4))                # 3行4列全零矩阵

# 生成全一数组 - 常用作初始值
ones_matrix = np.ones((2, 3))              # 2×3全一阵

# 生成空数组 - 不初始化值，最快但最危险
empty_arr = np.empty((2, 2))               # 值未定义，是内存中的随机值
print(f"空数组（随机值）:\n{empty_arr}")

# 生成单位矩阵 - 线性代数常用
identity = np.eye(3)                       # 3×3单位矩阵
print(f"单位矩阵:\n{identity}")

# 生成对角矩阵
diagonal = np.diag([1, 2, 3, 4])           # 对角线为1,2,3,4的对角矩阵

注意事项：

  • np.empty()返回的数组包含内存中的随机值，必须后续填充
  • np.zeros()和np.ones()默认创建float64类型数组

2.3 序列数组生成

1
2
3
4
5
6
7
8
9
10
11
12
13
# 等差数列 - 类似Python的range但更强大
arange_1 = np.arange(0, 10, 2)            # [0, 2, 4, 6, 8] - 不包括终点
arange_2 = np.arange(5)                   # [0, 1, 2, 3, 4] - 默认从0开始

# 等间距数列 - 控制点数而非步长
linspace = np.linspace(0, 1, 5)           # [0., 0.25, 0.5, 0.75, 1.]
logspace = np.logspace(0, 2, 5)           # 对数空间：[1., 3.16, 10., 31.62, 100.]

# 网格坐标生成 - 用于3D绘图和函数评估
x = np.linspace(-2, 2, 5)
y = np.linspace(-2, 2, 5)
X, Y = np.meshgrid(x, y)                  # 生成网格坐标
print(f"网格X坐标:\n{X}")

使用场景:

  • np.arange()：需要控制步长时使用
  • np.linspace()：需要精确控制点数时使用
  • np.meshgrid()：计算二维或三维函数时使用

2.4 随机数组生成

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
# 设置随机种子确保可重复性
np.random.seed(42)

# 均匀分布 [0, 1)
uniform_random = np.random.rand(3, 3)     # 3×3均匀分布随机矩阵

# 标准正态分布
normal_random = np.random.randn(1000)     # 1000个标准正态分布随机数

# 整数随机数
integers = np.random.randint(0, 100, 10)  # 10个[0,100)的随机整数

# 特定分布
poisson = np.random.poisson(5, 100)       # 泊松分布，λ=5
binomial = np.random.binomial(10, 0.5, 100)  # 二项分布，n=10, p=0.5

注意事项：

  • 始终设置随机种子以确保结构可重复
  • 注意不同分布的参数含义和范围
  • NumPy 1.17+推荐使用新的随机数生成器API

3. 数组属性和基本信息

简要说明：了解数组的属性和基本操作是有效使用NumPy的前提。这些属性提供了数组的元信息，对于调试和优化至关重要。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# 创建一个示例数组
arr = np.array([[1, 2, 3, 4], 
                [5, 6, 7, 8], 
                [9, 10, 11, 12]])

# 基本属性
print(f"数组维度: {arr.ndim}")           # 2 - 二维数组
print(f"数组形状: {arr.shape}")          # (3, 4) - 3行4列
print(f"元素总数: {arr.size}")           # 12 - 总元素个数
print(f"数据类型: {arr.dtype}")          # int64 - 64位整数
print(f"元素大小: {arr.itemsize}字节")    # 8 - 每个元素占8字节
print(f"数组大小: {arr.nbytes}字节")      # 96 - 总内存占用量

# 数组信息摘要
print(f"数组内存布局: {arr.flags}")
print(f"数组步长: {arr.strides}")        # (32, 8) - 行步长32字节，列步长8字节

重要概念：

  • 形状(shape)：数组每个维度的大小，对于理解数组结构至关重要
  • 数据类型(dtype)：决定内存使用和计算精度，选择合适的dtype是优化关键
  • 步长(strides)：内存中从一个元素到下一个元素的字节数，影响操作性能

内存布局说明：

1
2
3
4
5
6
# C顺序（行优先）和F顺序（列优先）
c_arr = np.array([[1, 2], [3, 4]], order='C')  # 默认，行优先
f_arr = np.array([[1, 2], [3, 4]], order='F')  # 列优先，Fortran风格

print(f"C顺序步长: {c_arr.strides}")  # (16, 8) - 行优先
print(f"F顺序步长: {f_arr.strides}")  # (8, 16) - 列优先

性能提示：

  • 对于行操作密集的任务，使用C顺序数组
  • 对于列操作密集的任务，使用F顺序数组
  • 使用arr.T（转置）不会复制数据，只是改变步长

4. 数组索引和切片

简要说明：NumPy提供了强大灵活的索引和切片功能，允许以各种方式访问和修改数组元素。理解这些机制对于高效数据操作至关重要

4.1 基础索引

1
2
3
4
5
6
7
8
9
10
arr = np.array([10, 20, 30, 40, 50, 60, 70, 80])

# 基础索引
print(arr[0])      # 10 - 第一个元素
print(arr[-1])     # 80 - 最后一个元素
print(arr[2:5])    # [30 40 50] - 切片（左闭右开）
print(arr[:3])     # [10 20 30] - 从头开始
print(arr[3:])     # [40 50 60 70 80] - 到末尾结束
print(arr[::2])    # [10 30 50 70] - 步长为2
print(arr[::-1])   # [80 70 60 50 40 30 20 10] - 反转数组

注意事项：

  • NumPy切片返回的视图而非副本，修改切片会影响原数组
  • 这与Python列表的切片行为不同（列表切片返回副本）

4.2 多维数组索引

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
matrix = np.array([[1, 2, 3, 4],
                   [5, 6, 7, 8],
                   [9, 10, 11, 12]])

# 单元素访问
print(matrix[1, 2])      # 7 - 第1行第2列（0-based）
print(matrix[1][2])      # 7 - 等价写法，但效率稍低

# 行和列切片
print("第0行:", matrix[0, :])        # [1 2 3 4]
print("第2列:", matrix[:, 2])        # [3 7 11]
print("前两行:", matrix[:2, :])      # [[1 2 3 4] [5 6 7 8]]
print("中间两列:", matrix[:, 1:3])   # [[2 3] [6 7] [10 11]]

# 子矩阵提取
submatrix = matrix[1:, 2:]           # 提取右下角2×2子矩阵
print("子矩阵:\n", submatrix)       # [[7 8] [11 12]]

视图与副本的关键区别：

1
2
3
4
5
6
7
8
9
10
11
# 切片创建视图
arr = np.array([1, 2, 3, 4, 5])
view = arr[1:4]           # 创建视图
view[0] = 99              # 修改视图
print("原数组:", arr)     # [1, 99, 3, 4, 5] - 原数组也被修改！

# 需要副本时使用copy()
arr = np.array([1, 2, 3, 4, 5])
copy = arr[1:4].copy()    # 创建副本
copy[0] = 99              # 修改副本
print("原数组:", arr)     # [1, 2, 3, 4, 5] - 原数组不受影响

4.3 布尔索引

简要说明：使用布尔数组作为索引，可以基于条件选择元素。这是NumPy最强大的特征之一。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
arr = np.array([3, 7, 1, 9, 5, 8, 2, 6])

# 基本布尔索引
mask = arr > 5
print("布尔掩码:", mask)           # [False  True False  True False  True False  True]
print("大于5的元素:", arr[mask])   # [7 9 8 6]

# 复合条件
print("大于3且小于8:", arr[(arr > 3) & (arr < 8)])  # [7 5 6]
print("小于3或大于7:", arr[(arr < 3) | (arr > 7)])  # [1 9 8 2]
print("不等于5:", arr[arr != 5])                    # 排除特定值

# 使用np.where
indices = np.where(arr > 5)
print("大于5的索引:", indices[0])   # [1 3 5 7]
print("对应值:", arr[indices])      # [7 9 8 6]

性能提示：

  • 布尔索引通常比循环块得多
  • 对于大型数组，考虑使用np.where()替代布尔索引提高性能

4.4 花式索引

简要说明：使用整数数组进行索引，可以实现复杂的元素选择。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
arr = np.array([10, 20, 30, 40, 50, 60])

# 使用整数列表索引
indices = [0, 2, 4]
print(arr[indices])                     # [10 30 50]

# 使用整数数组索引
idx_array = np.array([1, 3, 5])
print(arr[idx_array])                   # [20 40 60]

# 多维花式索引
matrix = np.array([[1, 2, 3],
                   [4, 5, 6],
                   [7, 8, 9]])

# 选择特定行
print("第0行和第2行:\n", matrix[[0, 2]])
# [[1 2 3]
#  [7 8 9]]

# 选择特定行和列
print("(0,1), (1,2), (2,0):", matrix[[0, 1, 2], [1, 2, 0]])  # [2 6 7]

使用场景：

  • 从数据集中选择非连续的子集
  • 重新排序数组元素
  • 实现查表功能

注意事项：

  • 花式索引返回的是副本而非视图
  • 与切片相比，花式索引的性能较差，特别是在选择大量元素时

5. 数组操作和变形

简要说明：NumPy提供了丰富的数组操作函数，可以改变数组的形状、连接数组、分割数组等

5.1 改变形状

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
arr = np.arange(12)  # [0, 1, 2, ..., 11]

# reshape - 改变形状（不复制数据）
reshaped = arr.reshape(3, 4)
print("3×4数组:\n", reshaped)
# [[ 0  1  2  3]
#  [ 4  5  6  7]
#  [ 8  9 10 11]]

# 自动推断维度
auto_reshaped = arr.reshape(3, -1)  # -1表示自动计算
print("自动推断形状:", auto_reshaped.shape)  # (3, 4)

# resize - 改变形状（可能复制数据）
arr_copy = arr.copy()
arr_copy.resize(2, 6)  # 原地操作，修改原数组
print("resize后:", arr_copy.shape)  # (2, 6)

# flatten和ravel - 展平数组
flattened = reshaped.flatten()  # 总是返回副本
raveled = reshaped.ravel()      # 尽可能返回视图
print("flatten:", flattened)
print("ravel:", raveled)

重要区别：

  • reshape()：返回视图，不改变原数组
  • resize()：原地修改数组，可能复制数据
  • flatten()：总是返回副本
  • ravel()：尽可能返回视图

5.2 转置和轴交换

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
matrix = np.array([[1, 2, 3],
                   [4, 5, 6]])

# 简单转置
transposed = matrix.T
print("转置矩阵:\n", transposed)
# [[1 4]
#  [2 5]
#  [3 6]]

# transpose函数（更通用）
transposed2 = np.transpose(matrix)
print("转置2:\n", transposed2)

# 高维数组的轴交换
arr_3d = np.arange(24).reshape(2, 3, 4)
print("原始形状:", arr_3d.shape)  # (2, 3, 4)

# 交换轴
swapped = np.transpose(arr_3d, (1, 0, 2))  # 交换轴0和轴1
print("交换后形状:", swapped.shape)  # (3, 2, 4)

# moveaxis - 移动轴到新位置
moved = np.moveaxis(arr_3d, 0, -1)  # 将第0轴移动到最后
print("移动轴后形状:", moved.shape)  # (3, 4, 2)

使用场景：

  • 转置：线性代数运算、改变数据方向
  • 轴交换：调整数据维度以适应不变通函数的要求

5.3 数组连接

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
a = np.array([[1, 2], [3, 4]])
b = np.array([[5, 6], [7, 8]])

# 垂直连接（增加行数）
vstacked = np.vstack((a, b))
print("垂直连接:\n", vstacked)
# [[1 2]
#  [3 4]
#  [5 6]
#  [7 8]]

# 水平连接（增加列数）
hstacked = np.hstack((a, b))
print("水平连接:\n", hstacked)
# [[1 2 5 6]
#  [3 4 7 8]]

# 深度连接（增加深度）
dstacked = np.dstack((a, b))
print("深度连接形状:", dstacked.shape)  # (2, 2, 2)

# concatenate - 通用连接函数
concat_axis0 = np.concatenate((a, b), axis=0)  # 沿轴0连接，同vstack
concat_axis1 = np.concatenate((a, b), axis=1)  # 沿轴1连接，同hstack

注意事项：

  • 连接数组时，出了连接轴外的其他维度必须匹配
  • 对于大型数组，连接操作可能设计数据复制，注意性能

5.4 数组分割

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
arr = np.arange(12).reshape(3, 4)
print("原始数组:\n", arr)

# 水平分割（按列）
hsplit_result = np.hsplit(arr, 2)  # 分成2个数组
print("水平分割结果:")
for i, part in enumerate(hsplit_result):
    print(f"部分{i}:\n{part}")

# 垂直分割（按行）
vsplit_result = np.vsplit(arr, 3)  # 分成3个数组
print("垂直分割结果:")
for i, part in enumerate(vsplit_result):
    print(f"部分{i}:\n{part}")

# split - 通用分割函数
# 等量分割
split_equal = np.split(arr, 3, axis=0)  # 沿轴0等分成3份

# 不等量分割
split_unequal = np.split(arr, [1, 3], axis=1)  # 在第1列和第3列处分割
print("不等量分割:")
for i, part in enumerate(split_unequal):
    print(f"部分{i}形状: {part.shape}")

性能提示：

  • 分割操作通常返回视图而非副本
  • 对于大型数组，避免频繁分割和重新连接

6. 数组运算

简要说明：NumPy提供了丰富的数学运算函数，支持逐元素运算、矩阵运算和统计运算等。

6.1 算数运算

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
a = np.array([1, 2, 3, 4])
b = np.array([5, 6, 7, 8])

# 基本算术运算（逐元素）
print("加法:", a + b)        # [6, 8, 10, 12]
print("减法:", a - b)        # [-4, -4, -4, -4]
print("乘法:", a * b)        # [5, 12, 21, 32] - 逐元素乘，不是矩阵乘！
print("除法:", b / a)        # [5., 3., 2.333..., 2.]
print("整数除法:", b // a)    # [5, 3, 2, 2]
print("幂运算:", a ** 2)     # [1, 4, 9, 16]
print("取余:", b % a)        # [0, 0, 1, 0]

# 原地运算（节省内存）
a += b  # 等价于 a = a + b，但原地操作
print("原地加法后:", a)

重要区别：

  • *运算符执行逐元素乘法而不是矩阵乘法
  • 矩阵乘法使用@运算付或np.dot()

6.2 矩阵运算

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
A = np.array([[1, 2], [3, 4]])
B = np.array([[5, 6], [7, 8]])

# 矩阵乘法
matrix_product = A @ B  # Python 3.5+推荐
matrix_product2 = np.dot(A, B)  # 传统方法
print("矩阵乘法结果:\n", matrix_product)
# [[19 22]
#  [43 50]]

# 内积和点积
v1 = np.array([1, 2, 3])
v2 = np.array([4, 5, 6])
print("点积:", np.dot(v1, v2))      # 32
print("内积:", np.inner(v1, v2))    # 32
print("外积:\n", np.outer(v1, v2))
# [[ 4  5  6]
#  [ 8 10 12]
#  [12 15 18]]

# 克罗内克积
kron = np.kron(A, B)
print("克罗内克积形状:", kron.shape)  # (4, 4)

6.3 统计运算

简要说明：NumPy提供了丰富的统计函数，可以计算数组的各种统计量。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
data = np.array([[1, 2, 3], 
                 [4, 5, 6], 
                 [7, 8, 9]])

# 基本统计量
print("总和:", np.sum(data))            # 45
print("平均值:", np.mean(data))         # 5.0
print("标准差:", np.std(data))          # 2.581...
print("方差:", np.var(data))            # 6.666...
print("最小值:", np.min(data))          # 1
print("最大值:", np.max(data))          # 9
print("中位数:", np.median(data))       # 5.0

# 沿特定轴计算
print("每列总和:", np.sum(data, axis=0))    # [12 15 18]
print("每行平均值:", np.mean(data, axis=1)) # [2. 5. 8.]

# 累积运算
print("累积和:", np.cumsum(data))      # [1, 3, 6, 10, 15, 21, 28, 36, 45]
print("累积积:", np.cumprod(data))     # [1, 2, 6, 24, 120, 720, 5040, 40320, 362880]

# 分位数
print("25%分位数:", np.percentile(data, 25))  # 3.5
print("75%分位数:", np.percentile(data, 75))  # 6.5

性能提示：

  • 对于大型数组，指定axis参数可以显著提供性能
  • 使用np.nanmean()，np.nanstd()等函数可以忽略NaN值

6.4 逻辑运算和比较

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
a = np.array([1, 2, 3, 4, 5])
b = np.array([5, 4, 3, 2, 1])

# 比较运算
print("等于:", a == b)          # [False False True False False]
print("不等于:", a != b)        # [True True False True True]
print("大于:", a > b)           # [False False False True True]
print("大于等于:", a >= b)      # [False False True True True]

# 逻辑运算
condition1 = a > 2
condition2 = b < 4
print("逻辑与:", condition1 & condition2)  # [False False True False False]
print("逻辑或:", condition1 | condition2)  # [False True True True True]
print("逻辑非:", ~condition1)             # [True True False False False]

# 判断函数
print("所有元素都大于0:", np.all(a > 0))   # True
print("有元素大于4:", np.any(a > 4))       # True
print("非零元素数量:", np.count_nonzero(a)) # 5

注意事项：

  • 使用&，|，~进行逻辑运算，而不是and，or，not
  • 这些逻辑运算符的优先级与Python的不同，建议使用括号

7. 广播机制

简要说明：广播是NumPy中处理不同形状数组进行算术运算的机制。它允许NumPy在执行元素级操作时扩展较小的数组。

广播规则：

  1. 如果两个数组的维度数不同，将维度较小的数组的形状前面补1
  2. 如果两个数组在某个维度上的大小不同，且其中一个为1，则该维度扩展以匹配另一个
  3. 如果两个数组在某个维度上的大小不同且都不为1，则广播失败
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
# 示例1：标量与数组
arr = np.array([1, 2, 3])
result = arr + 5  # 标量5被广播为[5, 5, 5]
print("标量广播:", result)  # [6, 7, 8]

# 示例2：列向量与行向量
col = np.array([[1], [2], [3]])   # 形状(3, 1)
row = np.array([4, 5, 6])         # 形状(3,)
result = col + row                # 广播为(3, 3)
print("列向量与行向量广播:\n", result)
# [[5 6 7]
#  [6 7 8]
#  [7 8 9]]

# 示例3：复杂广播
A = np.ones((2, 3, 4))   # 形状(2, 3, 4)
B = np.ones((3, 4))      # 形状(3, 4)
result = A + B           # B广播为(1, 3, 4)，然后为(2, 3, 4)
print("复杂广播后形状:", result.shape)  # (2, 3, 4)

使用场景：

  • 数组与标量的运算
  • 不同形状数组的逐元素运算
  • 标准化数据（如减去均值，除以标准差）

注意事项：

  • 广播不会实际复制数据，只是虚拟扩展，因此非常高效
  • 但错误的广播可能导致意外的结果，需要仔细检查形状
  • 使用np.newaxis或None可以增加维度以支持广播
1
2
3
4
5
6
# 显式增加维度
arr = np.array([1, 2, 3])
col_vector = arr[:, np.newaxis]  # 形状(3, 1)
row_vector = arr[np.newaxis, :]  # 形状(1, 3)
print("列向量形状:", col_vector.shape)
print("行向量形状:", row_vector.shape)

性能评价：

  • 广播避免了显式复制数据，非常内存高效
  • 但过度或不正确的广播可能导致性能下降
  • 在可能的情况下，尽量使用形状匹配的数组

8. 线性代数

简要说明：NumPy提供了丰富的线性代数函数，通过np.linalg模块访问。这些函数支持矩阵分解、求逆、特征值计算等高级操作

8.1 基本线性代数操作

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
A = np.array([[1, 2], [3, 4]])
B = np.array([[5, 6], [7, 8]])

# 矩阵乘法
print("矩阵乘法:\n", A @ B)

# 矩阵求逆
A_inv = np.linalg.inv(A)
print("逆矩阵:\n", A_inv)
print("验证A×A⁻¹:\n", A @ A_inv)  # 应接近单位矩阵

# 行列式
det = np.linalg.det(A)
print("行列式:", det)

# 矩阵的迹
trace = np.trace(A)
print("迹:", trace)

# 解线性方程组
# 解 Ax = b
b = np.array([5, 11])
x = np.linalg.solve(A, b)
print("方程解:", x)  # [1., 2.]
print("验证:", A @ x)  # 应接近[5., 11.]

8.2 矩阵分解

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
# 特征分解
A = np.array([[4, 2], [2, 4]])
eigenvalues, eigenvectors = np.linalg.eig(A)
print("特征值:", eigenvalues)
print("特征向量:\n", eigenvectors)

# 验证特征分解
# A·v = λ·v
for i in range(len(eigenvalues)):
    v = eigenvectors[:, i]
    λ = eigenvalues[i]
    print(f"验证特征向量{i}: {np.allclose(A @ v, λ * v)}")

# 奇异值分解(SVD)
U, S, Vh = np.linalg.svd(A)
print("U矩阵:\n", U)
print("奇异值:", S)
print("Vh矩阵:\n", Vh)

# QR分解
Q, R = np.linalg.qr(A)
print("Q矩阵（正交）:\n", Q)
print("R矩阵（上三角）:\n", R)
print("验证QR=A:", np.allclose(Q @ R, A))

# Cholesky分解（正定矩阵）
A_posdef = np.array([[4, 2], [2, 5]])
L = np.linalg.cholesky(A_posdef)
print("Cholesky分解L:\n", L)
print("验证L·Lᵀ=A:", np.allclose(L @ L.T, A_posdef))

8.3 矩阵范数和条件数

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

# 矩阵范数
print("Frobenius范数:", np.linalg.norm(A, 'fro'))
print("2-范数（谱范数）:", np.linalg.norm(A, 2))
print("1-范数:", np.linalg.norm(A, 1))
print("∞-范数:", np.linalg.norm(A, np.inf))

# 条件数（衡量矩阵求逆的稳定性）
cond = np.linalg.cond(A)
print("条件数:", cond)  # 值越大，矩阵越接近奇异

使用场景：

  • 特征分解：主成分分析(PCA)、振动分析
  • SVD：推荐系统、图像压缩
  • QR分解：最小二乘问题
  • Cholesky分解：优化问题、蒙特卡洛模拟

注意事项：

  • 不是所有矩阵都可逆，求逆前检查条件数
  • 对于大型稀疏矩阵，使用SciPy的稀疏线性代数模块
  • 特征值和特征向量可能是复数

9. 随机数生成

简要说明：NumPy提供了全面的随机数生成功能。从NumPy 1.17开始，推荐使用新的随机数生成器API

9.1 基础随机数生成

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
# 创建随机数生成器实例（推荐方式）
rng = np.random.default_rng(seed=42)

# 均匀分布
uniform = rng.uniform(0, 1, 10)  # 10个[0,1)的均匀分布随机数
print("均匀分布:", uniform[:5])

# 正态分布
normal = rng.normal(0, 1, 1000)  # 均值为0，标准差为1
print("正态分布均值:", np.mean(normal))
print("正态分布标准差:", np.std(normal))

# 整数随机数
integers = rng.integers(0, 100, 20)  # 20个[0,100)的随机整数
print("随机整数:", integers[:5])

# 从给定列表选择
choices = rng.choice(['A', 'B', 'C', 'D'], size=10, p=[0.1, 0.2, 0.3, 0.4])
print("加权随机选择:", choices)

9.2 特定分布

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
rng = np.random.default_rng(42)

# 二项分布
binomial = rng.binomial(10, 0.5, 1000)  # n=10, p=0.5
print("二项分布均值:", np.mean(binomial), "期望:", 10*0.5)

# 泊松分布
poisson = rng.poisson(5, 1000)  # λ=5
print("泊松分布均值:", np.mean(poisson))

# 指数分布
exponential = rng.exponential(scale=1.0, size=1000)
print("指数分布均值:", np.mean(exponential))

# 贝塔分布
beta = rng.beta(2, 5, 1000)  # α=2, β=5
print("贝塔分布均值:", np.mean(beta))

# 伽马分布
gamma = rng.gamma(shape=2, scale=2, size=1000)
print("伽马分布均值:", np.mean(gamma))

9.3 随机抽样和排列

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
data = np.arange(10)
rng = np.random.default_rng(42)

# 随机排列（返回新数组）
shuffled = rng.permutation(data)
print("随机排列:", shuffled)

# 打乱数组（原地）
data_copy = data.copy()
rng.shuffle(data_copy)
print("打乱后:", data_copy)

# 随机抽样
sample_without_replacement = rng.choice(data, size=5, replace=False)
print("无放回抽样:", sample_without_replacement)

sample_with_replacement = rng.choice(data, size=15, replace=True)
print("有放回抽样:", sample_with_replacement)

最佳实践：
1.总是设置随机种子以确保结果可重复
2.使用np.random.default_rng()而非旧的np.random.*函数
3.对于加密安全的应用，使用专门的加密安全随机数生成器

性能提示：

  • 一次生成大量随机数比多次生成少量随机数更高效
  • 对于蒙特卡洛模拟，考虑使用向量化操作

10. 输入输出操作

简要说明：NumPy提供了多种保存和加载数组的方法，支持文本格式和二进制格式

10.1 文本文件操作

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
# 创建示例数据
data = np.array([[1.0, 2.0, 3.0],
                 [4.0, 5.0, 6.0],
                 [7.0, 8.0, 9.0]])

# 保存为文本文件
np.savetxt('data.csv', data, delimiter=',', fmt='%.2f', header='x,y,z')

# 加载文本文件
loaded = np.loadtxt('data.csv', delimiter=',')
print("从CSV加载:\n", loaded)

# 处理带有缺失值的文件
# genfromtxt可以处理缺失值
data_with_nan = "1,2,3\n4,,6\n7,8,9"
loaded_nan = np.genfromtxt(data_with_nan.splitlines(), delimiter=',', 
                           filling_values=0)
print("处理缺失值:\n", loaded_nan)

10.2 二进制文件操作

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
# 保存单个数组为.npy格式
np.save('data.npy', data)

# 加载.npy文件
loaded_npy = np.load('data.npy')
print(".npy文件加载:\n", loaded_npy)

# 保存多个数组为.npz格式
np.savez('multi_data.npz', array1=data, array2=data*2)

# 加载.npz文件
loaded_npz = np.load('multi_data.npz')
print("数组1:\n", loaded_npz['array1'])
print("数组2:\n", loaded_npz['array2'])

# 保存为压缩格式
np.savez_compressed('compressed_data.npz', big_array=np.random.randn(1000, 1000))

格式比较：
格式|优点|缺点|适用场景
CSV|人类可读、通用|文件大，精度可能丢失|与其他软件交换数据
.npy|快速，保留所有信息|NumPy专用|NumPy数组长期保存|
.npz|保存多个数组|NumPy专用|保存多个相关数组

最佳实践：

  1. 对于NumPy内部使用，优先使用.npy/.npz格式
  2. 与其他软件交换数据，使用CSV或其他通用格式
  3. 大型数据集使用压缩格式节省空间

10.3 内存映射文件

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
# 创建内存映射文件（处理超大数组）
# 创建一个10GB的虚拟数组，但不立即加载到内存
shape = (10000, 10000)  # 10,000×10,000数组
dtype = np.float32
filename = 'large_array.dat'

# 创建内存映射
mmap_arr = np.memmap(filename, dtype=dtype, mode='w+', shape=shape)

# 像普通数组一样操作（数据自动写入文件）
mmap_arr[:1000, :1000] = np.random.randn(1000, 1000)

# 删除对象，确保数据写入文件
del mmap_arr

# 重新打开为只读
mmap_readonly = np.memmap(filename, dtype=dtype, mode='r', shape=shape)
print("内存映射数组形状:", mmap_readonly.shape)

使用场景：

  • 处理大于内存的数据集
  • 多个进程共享数据
  • 需要持久化的大型中间结果

11. 高级特性和技巧

11.1 结构化数组

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
# 定义数据类型
dtype = [('name', 'U10'),  # Unicode字符串，最大长度10
         ('age', 'i4'),     # 32位整数
         ('height', 'f4'),  # 32位浮点数
         ('weight', 'f4')]

# 创建结构化数组
people = np.array([('Alice', 25, 1.65, 58.5),
                   ('Bob', 30, 1.80, 75.2),
                   ('Charlie', 35, 1.75, 70.1)], dtype=dtype)

print("结构化数组:", people)
print("姓名字段:", people['name'])
print("平均年龄:", np.mean(people['age']))

# 条件选择
tall_people = people[people['height'] > 1.70]
print("身高>1.70的人:", tall_people['name'])

# 按字段排序
sorted_by_age = np.sort(people, order='age')
print("按年龄排序:", sorted_by_age)

使用场景：

  • 处理表格数据（类似Pandas DataFrame）
  • 处理异构数据
  • 与结构化数据文件（如HDF5）交互

11.2 掩码数组

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
import numpy.ma as ma

# 创建带有无效值的数据
data = np.array([1, 2, 3, -999, 5, -999, 7])
print("原始数据:", data)

# 创建掩码数组
masked_data = ma.masked_values(data, -999)
print("掩码数组:", masked_data)
print("掩码:", masked_data.mask)

# 计算统计量（自动忽略掩码值）
print("均值:", masked_data.mean())
print("标准差:", masked_data.std())

# 填充掩码值
filled_data = masked_data.filled(0)
print("填充后:", filled_data)

# 掩码数组运算
masked_data2 = ma.masked_values([1, -999, 3, 4, -999], -999)
result = masked_data + masked_data2
print("掩码数组相加:", result)

使用场景：

  • 处理缺失值或无效值
  • 遥感图像处理（云层遮挡）
  • 实验数据中的异常值处理

11.3向量化操作

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
# 非向量化方法（慢）
def slow_calc(x, y):
    result = np.zeros_like(x)
    for i in range(len(x)):
        for j in range(len(y)):
            result[i] = x[i]**2 + y[j]**2
    return result

# 向量化方法（快）
def fast_calc(x, y):
    # 利用广播
    x_squared = x[:, np.newaxis]**2  # 形状(n, 1)
    y_squared = y[np.newaxis, :]**2  # 形状(1, m)
    return x_squared + y_squared     # 广播为(n, m)

# 性能测试
x = np.random.randn(1000)
y = np.random.randn(1000)

import time
start = time.time()
slow_result = slow_calc(x[:10], y[:10])  # 只测试小规模
print(f"循环耗时: {time.time()-start:.4f}秒")

start = time.time()
fast_result = fast_calc(x, y)
print(f"向量化耗时: {time.time()-start:.4f}秒")

# 验证结果
print("结果是否一致:", np.allclose(slow_result, fast_result[:10, :10]))

性能提升技巧：

  1. 尽量使用NumPy内置函数而非Python循环
  2. 利用广播机制避免显式循环
  3. 使用np.vectorize()装饰Python函数（有限优化）
  4. 对于复杂操作，考虑使用Numba或Cython

11.4 通用函数(ufunc)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
# 创建自定义ufunc
def my_func(x, y):
    """自定义函数：返回x² + y²"""
    return x**2 + y**2

# 使用frompyfunc创建ufunc
my_ufunc = np.frompyfunc(my_func, 2, 1)  # 2个输入，1个输出

# 使用自定义ufunc
result = my_ufunc(np.array([1, 2, 3]), np.array([4, 5, 6]))
print("自定义ufunc结果:", result)

# ufunc方法
arr = np.array([1, 2, 3, 4, 5])

# reduce: 累积应用
sum_result = np.add.reduce(arr)  # 等价于np.sum(arr)
print("reduce求和:", sum_result)

# accumulate: 累积结果
cumsum_result = np.add.accumulate(arr)  # 等价于np.cumsum(arr)
print("accumulate累积和:", cumsum_result)

# outer: 外积
outer_result = np.multiply.outer(arr, arr)
print("外积形状:", outer_result.shape)

# reduceat: 在指定位置分段reduce
indices = [0, 2, 4]
reduceat_result = np.add.reduceat(arr, indices)
print("reduceat结果:", reduceat_result)  # [1+2, 3+4, 5]

高级应用：

  • 自定义ufunc可以用于任意元素级操作
  • ufunc方法提供了灵活的归约和累积操作
  • 结合np.vectorize()可以处理非标量返回值的函数

12. 性能优化和最佳实践

简要说明：NumPy虽然高效，但不当使用仍可能导致性能问题。了解NumPy的内部机制和最佳实践至关重要

12.1 内存布局优化

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
# C顺序（行优先）和F顺序（列优先）的性能差异
size = 1000
c_arr = np.ones((size, size), order='C')  # C顺序
f_arr = np.ones((size, size), order='F')  # F顺序

# 行操作性能比较
import time

start = time.time()
row_sum_c = np.sum(c_arr, axis=1)  # 按行求和
c_row_time = time.time() - start

start = time.time()
row_sum_f = np.sum(f_arr, axis=1)  # 按行求和
f_row_time = time.time() - start

print(f"C顺序行操作耗时: {c_row_time:.4f}秒")
print(f"F顺序行操作耗时: {f_row_time:.4f}秒")

# 列操作性能比较
start = time.time()
col_sum_c = np.sum(c_arr, axis=0)  # 按列求和
c_col_time = time.time() - start

start = time.time()
col_sum_f = np.sum(f_arr, axis=0)  # 按列求和
f_col_time = time.time() - start

print(f"C顺序列操作耗时: {c_col_time:.4f}秒")
print(f"F顺序列操作耗时: {f_col_time:.4f}秒")

优化建议：

  1. 根据主要访问模式选择内存布局
  2. 行操作多用C顺序，列操作多用F顺序
  3. 使用np.ascontiguousarray()和np.asfortranarray()转换布局

12.2 避免不必要的复制

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
arr = np.arange(10)

# 视图操作（不复制数据）
view1 = arr[:5]          # 切片创建视图
view2 = arr[::2]         # 步长切片创建视图
view3 = arr.T            # 转置创建视图
view4 = arr.reshape(2, 5)  # 重塑形状创建视图

# 检查是否为视图
print("arr.base:", arr.base)          # None（原始数组）
print("view1.base is arr:", view1.base is arr)  # True

# 复制操作（复制数据）
copy1 = arr.copy()                    # 显式复制
copy2 = arr[[0, 2, 4, 6, 8]]          # 花式索引创建副本
copy3 = np.array(arr)                 # 从数组创建新数组

print("copy1.base is arr:", copy1.base is arr)  # False

最佳实践：

  1. 尽量使用视图而非副本
  2. 必要时使用.copy显示创建副本
  3. 注意哪些操作创建视图，哪些创建副本

12.3 使用高效的数据类型

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
# 不同数据类型的比较
large_array = np.random.randn(1000000)

# float64（默认）
arr_float64 = large_array.astype(np.float64)
print(f"float64占用内存: {arr_float64.nbytes / 1024 / 1024:.2f} MB")

# float32（半精度）
arr_float32 = large_array.astype(np.float32)
print(f"float32占用内存: {arr_float32.nbytes / 1024 / 1024:.2f} MB")

# 性能比较
import time

start = time.time()
result64 = np.sum(arr_float64)
time64 = time.time() - start

start = time.time()
result32 = np.sum(arr_float32)
time32 = time.time() - start

print(f"float64计算时间: {time64:.6f}秒")
print(f"float32计算时间: {time32:.6f}秒")
print(f"精度损失: {abs(result64 - result32)/result64:.2%}")

数据类型选择指南：

  • 深度学习：float/fp16（混合精度训练）
  • 科学计算：float64（高精度需求）
  • 嵌入式系统：float/int（内存限制）
  • 图象处理：uint8（0-255像素值）

12.4 使用Numba加速

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
# 安装: pip install numba
from numba import jit, vectorize
import numpy as np

# 使用jit装饰器加速Python函数
@jit(nopython=True)
def numba_sum(arr):
    total = 0.0
    for i in range(arr.shape[0]):
        total += arr[i]
    return total

# 创建大型数组
large_arr = np.random.randn(10000000)

# 比较性能
import time

start = time.time()
numpy_sum = np.sum(large_arr)
numpy_time = time.time() - start

start = time.time()
numba_result = numba_sum(large_arr)
numba_time = time.time() - start

print(f"NumPy求和: {numpy_sum:.6f}, 耗时: {numpy_time:.4f}秒")
print(f"Numba求和: {numba_result:.6f}, 耗时: {numba_time:.4f}秒")
print(f"加速比: {numpy_time/numba_time:.2f}x")

# 使用vectorize创建ufunc
@vectorize(['float64(float64, float64)'], nopython=True)
def numba_vectorized(x, y):
    return x**2 + y**2

# 测试向量化函数
a = np.random.randn(10000)
b = np.random.randn(10000)

result = numba_vectorized(a, b)
print(f"向量化函数结果形状: {result.shape}")

Numba适用场景：

  1. NumPy没有提供的复杂逐元素操作
  2. 需要与现有的Python代码集成
  3. 算法原型到生产代码的快速迁移

注意事项：

  • 第一次调用有编译开销
  • 支持的数据类型和NumPy函数有限
  • 需要安装LLVM依赖

13. 常见问题与解决方案

简要说明：
在实际使用NumPy时，经常会遇到一些常见问题。了解这些问题的解决方案可以提高开发效率。

13.1 处理NaN和无穷值

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
# 创建包含特殊值的数据
arr = np.array([1, 2, np.nan, 4, np.inf, -np.inf, 7])

# 检测特殊值
print("NaN值:", np.isnan(arr))
print("无穷值:", np.isinf(arr))
print("有限值:", np.isfinite(arr))
print("非NaN有限值:", np.isfinite(arr) & ~np.isnan(arr))

# 安全计算（忽略特殊值）
print("忽略NaN的均值:", np.nanmean(arr))
print("忽略NaN的标准差:", np.nanstd(arr))
print("忽略NaN的最大值:", np.nanmax(arr))

# 替换特殊值
arr_clean = np.nan_to_num(arr, nan=0.0, posinf=999, neginf=-999)
print("清理后:", arr_clean)

# 删除包含NaN的行
arr_2d = np.array([[1, 2, 3],
                   [4, np.nan, 6],
                   [7, 8, 9]])
clean_rows = arr_2d[~np.isnan(arr_2d).any(axis=1)]
print("删除NaN行后:\n", clean_rows)

最佳实践：

  1. 在处理数据前先检查特殊值
  2. 使用np.nan*函数进行安全计算
  3. 根据应用场景决定如何处理特殊值

13.2 数组排序和搜索

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
arr = np.array([3, 1, 4, 1, 5, 9, 2, 6])

# 排序
sorted_arr = np.sort(arr)  # 返回排序后的副本
print("排序后:", sorted_arr)

arr.sort()  # 原地排序
print("原地排序后:", arr)

# 获取排序索引
arr = np.array([3, 1, 4, 1, 5, 9, 2, 6])
indices = np.argsort(arr)
print("排序索引:", indices)
print("按索引排序:", arr[indices])

# 部分排序
partial_indices = np.argpartition(arr, 3)  # 前3个最小值的索引
print("部分排序索引:", partial_indices[:3])
print("前3个最小值:", arr[partial_indices[:3]])

# 搜索
print("最大值索引:", np.argmax(arr))
print("最小值索引:", np.argmin(arr))

# 条件搜索
print("大于5的索引:", np.where(arr > 5)[0])
print("大于5的值:", arr[arr > 5])

# 最接近值搜索
target = 4.5
closest_idx = np.argmin(np.abs(arr - target))
print(f"最接近{target}的值:", arr[closest_idx])

性能提示：

  • 对于大型数组，部分排序比完全排序更快
  • np.where()返回元组，需要索引时用[0]
  • 搜索排序数组可以使用np.searchsorted()，效率更高

13.3 数据类型转换问题

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
# 隐式转换
arr_int = np.array([1, 2, 3], dtype=np.int32)
arr_float = np.array([1.5, 2.5, 3.5], dtype=np.float64)

result = arr_int + arr_float  # int32会提升为float64
print("结果类型:", result.dtype)  # float64

# 显式转换
arr = np.array([1.5, 2.7, 3.2])
arr_int = arr.astype(np.int32)  # 向下取整
print("转换为int32:", arr_int)  # [1, 2, 3]

# 安全转换
arr_large = np.array([100000, 200000, 300000], dtype=np.int32)
arr_small = arr_large.astype(np.int16)  # 可能溢出！
print("危险转换:", arr_small)  # 溢出值

# 安全转换函数
safe_converted = np.asarray(arr_large, dtype=np.int16)  # 同样危险
print("安全转换:", safe_converted)

# 最佳实践：先检查范围
if np.all(arr_large <= np.iinfo(np.int16).max) and \
   np.all(arr_large >= np.iinfo(np.int16).min):
    safe = arr_large.astype(np.int16)
    print("安全转换结果:", safe)
else:
    print("值超出范围，不能安全转换")

转换规则：

  1. 运算时，数据类型向更精确、范围更大的类型提升
  2. astype()总是返回副本，即使类型相同
  3. 向下转换可能丢失精度或溢出

13.4 广播错误排查

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
# 常见的广播错误
A = np.ones((3, 4, 5))
B = np.ones((3, 5))

try:
    result = A + B  # 错误：形状不兼容
except ValueError as e:
    print(f"广播错误: {e}")

# 调试广播
def debug_broadcast(shape1, shape2):
    """检查两个形状是否可以广播"""
    # 对齐形状
    s1 = list(shape1)
    s2 = list(shape2)
    
    # 补齐维度
    while len(s1) < len(s2):
        s1.insert(0, 1)
    while len(s2) < len(s1):
        s2.insert(0, 1)
    
    print(f"对齐后形状1: {s1}")
    print(f"对齐后形状2: {s2}")
    
    # 检查每个维度
    for i, (d1, d2) in enumerate(zip(s1, s2)):
        if d1 != d2 and d1 != 1 and d2 != 1:
            print(f"第{i}维不兼容: {d1} != {d2}")
            return False
    
    print("形状可以广播")
    # 计算广播后形状
    broadcast_shape = [max(d1, d2) for d1, d2 in zip(s1, s2)]
    print(f"广播后形状: {tuple(broadcast_shape)}")
    return True

# 测试
debug_broadcast((3, 4, 5), (3, 5))
debug_broadcast((3, 1, 5), (1, 4, 5))

广播调试技巧：

  1. 使用np.broadcast_to()显式扩展数组
  2. 使用np.newaxis增加维度
  3. 使用np.squeeze()去除大小为1的维度

14. 结语和资源推荐

14.1 总结

NumPy是Python科学计算的基石，掌握其核心功能对于任何涉及数值计算的工作都至关重要。本文涵盖了从基础到高级的NumPy知识，包括：

  1. 数组创建和操作：多种创建数组的方法和灵活的索引切片
  2. 数组运算和广播：高效的逐元素运算和形状自动扩展
  3. 线性代数和随机数：丰富的数学函数库
  4. 性能优化：内存布局、数据类型选择和加速技巧
  5. 高级特性：结构化数组、掩码数组和向量化操作
  6. 常见问题解决：NaN处理、排序搜索和错误排查

14.2 最佳实践回顾

  1. 优先使用向量化操作：避免Python循环，使用NumPy内置函数
  2. 注意内存使用：选择合适的dtype，使用视图而非副本
  3. 利用广播机制：但要注意形状兼容性
  4. 处理特殊值：使用安全的nan函数处理缺失值
  5. 优化数据布局：根据访问模式选择C或F顺序

14.3 学习资源推荐

官方资源：

  • NumPy官方文档
  • NumPy用户指南
  • NumPy API参考
    书籍推荐：
  • 《Python for Data Analysis》by Wes McKinney：包含NumPy和Pandas
  • 《Elegant SciPy》by Juan Nunez-Iglesias：NumPy在科学计算中的应用
  • 《From Python to NumPy》by Nicolas P. Rougier：深入理解NumPy内部机制

在线课程：

  • Coursera：Python for Data Science and AI (IBM)
  • edX：Introduction to Computational Thinking and Data Science (MIT)
  • DataCamp：Introduction to NumPy

进阶学习：

  • SciPy：建立在NumPy之上的科学计算库
  • Numba：NumPy代码的即时编译加速
  • Cython：将Python/NumPy代码编译为C扩展
  • CuPy：GPU加速的NumPy替代品

14.4 速查卡片

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
# 附录：快速参考卡片

import numpy as np

# ============ 创建数组 ============
np.array([1,2,3])           # 从列表
np.zeros((3,4))             # 全零数组
np.ones((2,3))              # 全一数组
np.empty((2,2))             # 空数组（未初始化）
np.arange(0,10,2)           # 等差数列
np.linspace(0,1,5)          # 等间隔数列
np.random.rand(3,3)         # [0,1)均匀分布

# ============ 数组属性 ============
arr.shape                   # 数组形状
arr.ndim                    # 数组维度
arr.dtype                   # 数据类型
arr.size                    # 元素总数
arr.nbytes                  # 内存占用量

# ============ 索引切片 ============
arr[0]                      # 第一个元素
arr[-1]                     # 最后一个元素
arr[1:4]                    # 切片
arr[arr > 5]                # 布尔索引
arr[[0,2,4]]                # 花式索引
arr[:, 1]                   # 第二列

# ============ 数组操作 ============
arr.reshape(3,4)            # 改变形状
arr.T                       # 转置
np.concatenate([a,b])       # 连接数组
np.split(arr, 3)            # 分割数组

# ============ 数学运算 ============
np.add(x, y)                # 加法
np.multiply(x, y)           # 乘法
np.dot(A, B)                # 矩阵乘法
np.sum(arr)                 # 求和
np.mean(arr)                # 平均值
np.std(arr)                 # 标准差

# ============ 线性代数 ============
np.linalg.inv(A)            # 矩阵求逆
np.linalg.det(A)            # 行列式
np.linalg.eig(A)            # 特征分解
np.linalg.solve(A, b)       # 解线性方程组

# ============ 文件操作 ============
np.save('data.npy', arr)    # 保存为.npy
np.load('data.npy')         # 加载.npy文件
np.savetxt('data.txt', arr) # 保存为文本
np.loadtxt('data.txt')      # 加载文本文件

展开全文 >>

数据的表示与运算

字数统计: 1.4k字   |   阅读时长≈ 5分

一、数据表示的基本概念

在计算机系统中，所有信息——无论是数字、文字、图像还是声音——最终都必须以二进制的形式表示和处理。因此，如何高效、准确地将各类数据映射为二进制形式，是计算机组成原理中的核心问题之一。数据的表示不仅影响存储效率，还直接关系到运算的正确性与系统的性能。
more >>

展开全文 >>

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia-plus根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true