高级篇 / 第 20 章

第 20 章 · 数据分析：NumPy 与 Pandas

为什么需要 NumPy 与 Pandas

先看一个真实痛点。用纯 Python 处理一百万个数字求和：

import time, random

data = [random.random() for _ in range(1_000_000)]

t0 = time.perf_counter()
s = 0
for x in data:          # Python 层循环：每个元素都要一次解释器调用
    s += x
print(f"纯 Python 求和：{time.perf_counter()-t0:.3f}s")     # 约 0.05s（但换 numpy 后差距更大）

t0 = time.perf_counter()
mx = max(data)
print(f"max：{time.perf_counter()-t0:.3f}s")

核心问题不在于「Python 循环慢」本身，而在于三个结构性开销：每个元素在内存里是一个独立对象（不是连续的数值块）、每次运算都要经过解释器分发、循环体无法被 CPU 预取和向量化。NumPy 的 ndarray（n-dimensional array，N 维数组）正是一次性解决这三点：

1. 连续的内存布局：所有元素类型相同、紧密排列，CPU 缓存友好。
2. 按块计算：一次算完整列，循环发生在 C 层。
3. 广播（broadcasting）：不同形状的数组按规则自动对齐相加相乘，避免手写循环。

ndarray：NumPy 的核心对象

2.1 创建、属性与基本操作

ndarray_basics.py

import numpy as np

# ---- 创建方式 ----
a = np.array([1, 2, 3, 4, 5])
b = np.arange(0, 10, 2)          # [0 2 4 6 8]，等价 range 但返回 ndarray
c = np.linspace(0, 1, 5)         # [0.   0.25 0.5  0.75 1.  ] 含端点
d = np.zeros((3, 4))             # 全 0，(行=3, 列=4)
e = np.ones((2, 3), dtype=np.int8)
f = np.full((2, 2), 7)
g = np.eye(3)                    # 单位矩阵
h = np.random.default_rng(42)    # 推荐的新随机数生成器（见 2.6）

# ---- 关键属性 ----
m = np.array([[1, 2, 3],
              [4, 5, 6]])
print(m.ndim)      # 2 维数
print(m.shape)     # (2, 3) 每一维的长度
print(m.size)      # 6 元素总数
print(m.dtype)     # int64 元素类型
print(m.nbytes)    # 48 占用字节
print(m.itemsize)  # 8  单元素字节数
print(m.flags.c_contiguous)   # True 是否 C 连续（行优先）

# ---- 索引与切片 ----
print(m[0])          # [1 2 3]  取行
print(m[0, 1])       # 2       行列都取 → 等价 m[0][1] 但只取一次
print(m[:, 1])       # [2 5]   取列（冒号表示「所有」）
print(m[0, :])       # [1 2 3]
print(m[:, ::2])     # [[1 3]
                    #  [4 6]]  步长切片，等价 m[:, 0::2]

# ---- 布尔索引（真正的威力所在）----
x = np.array([3, -1, 0, 7, -5, 12])
mask = x > 0
print(x[mask])                    # [ 3  7 12]
print(x[x % 2 == 0])              # [ 0 12]   直接把条件写进去
# 等价的 np.where 版本
print(np.where(x > 0, x, 0))     # [3 0 0 7 0 12]

# ---- 花式索引 ----
idx = [0, 2, 4]
print(x[idx])                     # [ 3  0 -5]

# ---- 视图 vs 拷贝 ----
v = m[:, 0]        # 基本切片 → 视图，共享内存
v[0] = 99
print(m)            # [[99 2 3]
                    #  [ 4 5 6]]  ← m 也变了
c2 = m[:, 0].copy()  # 显式拷贝
c2[0] = 0
print(m[0, 0])       # 99，没变

# ---- 形状操作 ----
p = np.arange(12).reshape(3, 4)
print(p.T)                      # 转置 (4,3)
print(p.reshape(2, 6).shape)    # (2,6)
print(p.ravel())                 # 展平
print(p.reshape(2, -1).shape)   # -1 表示「自动推算」→ (2,6)

2.2 dtype：类型系统与类型转换

import numpy as np

# 整数数组参与浮点运算会自动提升为 float64（不会溢出，但会丢精度控制）
ii = np.array([1, 2, 3])
print(ii.dtype)                       # int64

big = np.array([2**62, 2**62], dtype=np.int64)
print(big + big)                      # 溢出警告，结果可能为负
# 解决办法：转成 float64 或 object
print((big.astype(np.float64) + big.astype(np.float64)))

# 显式转换
f = np.array([1.7, 2.2, -3.9])
print(f.astype(np.int32))             # [ 1  2 -3]  直接截断，不是四舍五入！
print(np.round(f).astype(int))        # [ 2  2 -4]  先四舍五入再转

# 字符串 dtype 的坑：定长会截断
s = np.array(["ab", "abcd"])
print(s.dtype)                        # <U4  长度按最长元素定
print(np.array(["ab", "abcdef"], dtype="<U3"))   # ['ab' 'abc'] 被截断！

# object 数组能放异构数据，但慢
o = np.array([1, "two", 3.0, None], dtype=object)
print(o.dtype, o)

# 类型转换的错误处理
try:
    np.array(["1", "x"]).astype(int)
except ValueError as e:
    print("转换失败：", e)

# 安全的做法：用 pd.to_numeric（下一章会大量用到）
# np.array(["1", "x"], dtype=float) 同样报错
print(np.array(["1.5", "2.5"]).astype(float))   # [1.5 2.5]

2.3 广播机制（broadcasting）

广播规则（从小到大对齐）：两个数组的各维，要么相等，要么其中一个在该维长度为 1（可被拉伸）。长度为 1 的维度被「广播」成对方的长度。

import numpy as np

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

# 规则 1：形状完全相同
print(a + a)          # 逐元素

# 规则 2：某一维为 1 → 拉伸
print(a + np.array([10, 20, 30]))     # (2,3) + (3,) → 每行都加上同一组
# [[11 22 33]
#  [14 25 36]]

print(a + 100)                      # 标量：广播到所有元素
print(a * np.array([[1], [2]]))      # (2,3) + (2,1) → 列方向拉伸
# [[2  4  6]
#  [10 12 14]]

# 常见用途：列归一化（每列减去均值再除标准差）
data = np.array([[170, 60], [180, 75], [165, 55]], dtype=float)
mean = data.mean(axis=0)             # 按列求均值 → (2,)
std = data.std(axis=0)
z = (data - mean) / std              # 广播：列方向拉伸 (3,2) - (2,)
print(np.round(z, 3))

# 行方向归一化
mean_r = data.mean(axis=1, keepdims=True)   # (3,1)，必须 keepdims 才能广播
print(np.round(data / mean_r, 3))
# [[1.    0.353]
#  [1.    0.417]
#  [1.    0.333]]

# 布尔运算与广播一起用：给负数置 0
x = np.array([-1, 2, -3, 4])
print(np.where(x < 0, 0, x))       # [0 2 0 4]

2.4 常用函数：数学、统计、线性代数

numpy_funcs.py

import numpy as np

x = np.array([3.0, 1.0, 4.0, 1.0, 5.0, 9.0, 2.0, 6.0])

# ---- 基本统计 ----
print(x.sum(), x.mean(), x.std(), x.var())
print(x.min(), x.max(), np.ptp(x))          # ptp = max - min
print(x.cumsum())                           # 累加
print(x.prod())                             # 连乘

# ---- 排序与去重 ----
print(np.sort(x))                           # 升序，返回新数组
print(np.argsort(x))                        # 升序索引：第 4 小的元素在下标 1
print(np.unique(np.array([1, 1, 2, 2, 3])))   # [1 2 3]
vals, idx, cnt = np.unique([1, 1, 2, 3], return_index=True, return_counts=True)
print(vals, cnt)                            # [1 2 3] [2 1 1]

# ---- 分位数 ----
print(np.percentile(x, 50))                 # 中位数
print(np.percentile(x, [25, 75]))            # 四分位
print(np.quantile(x, 0.9))

# ---- 三角/指数/对数 ----
print(np.sqrt(np.array([4, 9, 16])))
print(np.exp(np.array([0, 1, 2])))
print(np.log(np.array([1, np.e, np.e ** 2])))
print(np.abs(np.array([-1, 2, -3])))

# ---- 矩阵与线性代数 ----
m = np.array([[1, 2], [3, 4]])
print(m.T)                                   # 转置
print(np.dot(m, m))                          # 矩阵乘法
print(m @ np.array([1, 0]))                  # @ 是 dot 的运算符写法
print(np.linalg.det(m))                      # 行列式  -2.0
print(np.linalg.inv(m))                      # 逆矩阵
print(np.linalg.solve(m, np.array([1, 2])))  # 解线性方程组
print(np.linalg.norm(m))                     # 范数

# ---- 随机数（新 API，务必用 default_rng）----
rng = np.random.default_rng(2024)
print(rng.random())                           # [0,1) 单个
print(rng.normal(100, 15, size=5))            # 正态：均值100 标准差15，5 个
print(rng.integers(1, 7, size=10))           # 1~6 模拟掷骰
print(rng.choice(["A", "B", "C"], size=100, p=[.5, .3, .2]).tolist()[:10])
print(rng.permutation(10))                    # 0~9 的随机排列

# 可复现：同一个 seed 一定得到同样的序列（做实验与测试时必需）
r1 = np.random.default_rng(7).normal(size=3)
r2 = np.random.default_rng(7).normal(size(3))
print(np.allclose(r1, r2))                    # True

索引、视图与进阶索引

3.1 索引的两种范式

import numpy as np

a = np.arange(12)
v = a[2:6]        # 视图
c = a[2:6].copy() # 拷贝
v[0] = 100
print(a)          # [  0   1 100 103 104 105 ...] a 被改了
c[0] = 0
print(a[2])       # 100，拷贝不影响 a

# np.copyto / np.take：显式控制
idx = np.array([1, 3, 5])
print(np.take(a, idx))                  # 一定是拷贝
a[[1, 3, 5]] = -1                      # 左边是进阶索引 → 只改 a[[1,3,5]] 这一份拷贝，a 不变
print(a[1:6])                           # 确认未变

3.2 花式索引与配对采样

m = np.arange(12).reshape(3, 4)
print(m)
# [[ 0  1  2  3]
#  [ 4  5  6  7]
#  [ 8  9 10 11]]

# 取「列」：给一个一维索引数组
print(m[:, [1, 3]])
# [[ 1  3]
#  [ 5  7]
#  [ 9 11]]

# 取「行」
print(m[[0, 2], :])
# [[ 0  1  2  3]
#  [ 8  9 10 11]]

# 行列同时取：取交叉点（3 行 × 2 列 = 6 个元素）
print(m[[0, 2], [1, 3]])
# [[ 1  3]
#  [ 9 11]]

# 行列索引数组长度必须相同（配对），否则报错
# m[[0, 1, 2], [1, 2]]   → IndexError

# 用配对索引批量赋值（很实用的技巧）
mm = m.copy()
rows = np.array([0, 1, 2])
mm[rows, rows] = 99          # 把对角线置 99
print(mm)
# [[ 99  1  2  3]
#  [ 4 99  6  7]
#  [ 8  9 99 11]]

# 布尔矩阵掩码：找出所有 > 5 的元素
print(m[m > 5])
# [ 6  7  8  9 10 11]

3.3 用 reshape 跑 mini-batch：矩阵化思维

把「逐条处理数据」的循环改写成批量矩阵运算，是 NumPy 提速的核心技巧。

import numpy as np

# 逐条：给每个样本的每列做标准化
data = np.random.default_rng(0).normal(size=(10000, 20))   # 1 万样本 × 20 特征

# 慢：Python 循环
# slow = np.array([(row - row.mean()) / row.std() for row in data])

# 快：一次广播搞定
fast = (data - data.mean(axis=1, keepdims=True)) / data.std(axis=1, keepdims=True)
print(fast.shape, np.round(fast.mean(), 6))   # (10000, 20)，每行均值约 0

# 再进一步：整个数据集全局标准化（只需要两行）
z = (data - data.mean()) / data.std()

# 技巧：把 for 循环向量化时，先想清楚「每一步是对行、对列还是对整体」
# 对列 → axis=0；对行 → axis=1；对整体 → 不写 axis

Series：一维带标签的数据

Series 是 Pandas 的一维数据结构，可以理解为「带索引的一维数组」，但索引可以是任意类型（字符串、日期、甚至混合）。

import pandas as pd
import numpy as np

# ---- 创建 ----
s1 = pd.Series([10, 20, 30, 40])                       # 默认索引 0,1,2,3
s2 = pd.Series([10, 20, 30], index=["a", "b", "c"])    # 自定义索引
s3 = pd.Series({"苹果": 5.5, "香蕉": 3.2, "橙子": 4.8})  # dict → 索引为键
s4 = pd.Series(np.arange(4), index=pd.date_range("2024-01-01", periods=4))
s5 = pd.Series([1, 2, 3], index=["x", "y", "z"], name="销量", dtype="int64")

print(s1)
print(s3)
print("index:", s3.index.tolist())
print("values:", s3.values)
print("dtype:", s3.dtype, "| size:", s3.size)

# ---- 索引与定位 ----
print(s3["苹果"])               # 标签访问
print(s3.loc["香蕉"])            # 显式标签访问（推荐，含标签）
print(s3.iloc[0])                # 按位置（第 0 个）
print(s3.loc[["苹果", "橙子"]])   # 多标签
print(s3.iloc[0:2])              # 位置切片

# ---- 条件筛选 ----
print(s3[s3 > 4])              # 布尔 Series 掩码
print(s3[(s3 > 4) & (s3 < 5)])
# ⚠️ 链式比较必须用 & |，不能写 4 < s3 < 5（Series 不支持）
print(s3[(s3 >= 4) | (s3.index == "香蕉")])

# ---- 向量化运算 ----
prices = pd.Series({"苹果": 5.5, "香蕉": 3.2, "橙子": 4.8})
qty = pd.Series({"苹果": 10, "香蕉": 25, "梨子": 3})   # 故意少一个键
revenue = prices * qty
print(revenue)
# 苹果  55.0      ← 索引对齐，自动生效
# 香蕉  80.0
# 梨子    NaN      ← 不匹配的索引 → NaN，而不是报错！这是 Pandas 最重要的特性
# 橙子   NaN

# ---- 常用属性与方法 ----
print(revenue.describe())        # 描述统计
print(revenue.max(), revenue.idxmax())
print(revenue.isna())            # 哪些是缺失
print(revenue.dropna())
print(revenue.fillna(0))
print(revenue.astype(float).round(2))
print(revenue.sort_values(ascending=False))
print(revenue.sort_index())
print(revenue.value_counts())     # 频次统计
print(revenue.rename("revenue"))  # 改名字
print(revenue.map(lambda x: x * 1.13))   # 逐元素（实际用 .apply）
print(revenue.apply(lambda x: round(x, 1)))

# ---- 常用统计 ----
print(revenue.sum(), revenue.mean(), revenue.std())
print(revenue.cumsum())

4.1 Series 的类型转换与时间序列技巧

import pandas as pd

s = pd.Series(["1", "2", "不是数字"])
print(pd.to_numeric(s, errors="coerce"))      # 1.0 2.0 NaN
print(pd.to_numeric(s, errors="ignore"))      # 抛错（已废弃，别用）
print(s.str.strip().str.upper())              # 字符串方法有 .str 访问器
print(s.str.extract(r"(\d+)"))                # 正则提取
print(pd.to_datetime(["2024-01-01", "2024/02/01", "2024年03月01日"]))

# 索引可以排序后做查找
t = pd.Series(range(5), index=pd.date_range("2024-01-01", periods=5, freq="D"))
print(t.loc["2024-01-03":"2024-01-05"])       # 日期区间切片（时间序列专属能力）

DataFrame：二维表格与「不可变对象」的哲学

5.1 创建

import pandas as pd
import numpy as np

# 方式 1：dict of dict / dict of list（最常用）
df = pd.DataFrame({
    "name":   ["张三", "李四", "王五", "赵六"],
    "age":    [28, 34, 29, 41],
    "city":   ["北京", "上海", "北京", "广州"],
    "salary": [15000, 22000, 13500, 30000],
})
print(df)

# 方式 2：二维数组 + 列名 + 索引
df2 = pd.DataFrame(
    np.arange(12).reshape(3, 4),
    columns=["a", "b", "c", "d"],
    index=["r1", "r2", "r3"],
)

# 方式 3：从二维列表（要求各行列数一致）
df3 = pd.DataFrame([[1, "x"], [2, "y"]], columns=["num", "ch"])

# 方式 4：从已有对象
pd.DataFrame(s3)                 # Series → 单列 DataFrame
pd.DataFrame(s3.items())         # 由 (键, 值) 组成的 Series
pd.DataFrame(np.random.rand(100, 4), columns=list("abcd")).describe()

# 方式 5：从记录列表（爬虫常见：list[dict]）
records = [
    {"id": 1, "title": "Python 入门", "price": 59.0},
    {"id": 2, "title": "Pandas 实战",  "price": 79.0},
]
df4 = pd.DataFrame(records)

# 方式 6：从嵌套 dict（会被 pandas 自动展开成多级列）
df5 = pd.DataFrame({"a": {"x": 1, "y": 2}, "b": {"x": 3, "y": 4}})
print(df5)
#    a  b
# x  1  3
# y  2  4

5.2 查看数据

import pandas as pd

df = pd.DataFrame({
    "name":   ["张三", "李四", "王五", "赵六", "钱七"],
    "age":    [28, 34, 29, 41, 23],
    "city":   ["北京", "上海", "北京", "广州", "北京"],
    "salary": [15000, 22000, 13500, 30000, 18000],
})

df.head()          # 前 5 行
df.tail(2)         # 后 2 行
df.shape           # (5, 4)
df.dtypes          # 每列的数据类型
df.info()          # 类型 + 非空数 + 内存占用
df.describe()      # 数值列的描述统计
df.columns         # 列名 Index
df.index           # 行索引
df.values          # 底层 ndarray（丢掉列名，不推荐用于后续处理）
df.axes            # [行索引, 列索引]
df.count()         # 每列非空个数
df.nunique()       # 每列不同取值个数
df.memory_usage(deep=True)

5.3 核心原则：DataFrame 是不可变视图

Pandas 的许多操作返回的是新对象而不是原地修改。这是它与 NumPy 切片视图最大的区别，也是初学者最常踩的坑。

import pandas as pd

df = pd.DataFrame({"a": [1, 2, 3]})

df["b"] = df["a"] * 2          # ✅ 新增列是原地生效的（赋值本身改的是原对象）
print(df)                       #    a  b
                                # 0  1  2
                                # 1  2  4
                                # 2  3  6

df["a"] = df["a"] * 10          # ✅ 覆盖已有列也是原地生效
print(df["a"].tolist())         # [10, 20, 30]

# ❌ 链式赋值的坑
try:
    df[df["a"] > 20]["c"] = 1          # SettingWithCopyWarning / 不生效
except Exception as e:
    print("失败：", e)

# ✅ 正确做法：用 loc 明确表达「在原表上赋值」
df.loc[df["a"] > 100, "c"] = 1          # 明确 loc → 对原对象生效
print(df)
#     a  b    c
# 0  10  2  NaN
# 1  20  4  NaN
# 2  30  6  1.0

5.4 选列与选行

import pandas as pd

df = pd.DataFrame({
    "name":  ["张三", "李四", "王五", "赵六"],
    "age":   [28, 34, 29, 41],
    "city":  ["北京", "上海", "北京", "广州"],
    "salary":[15000, 22000, 13500, 30000],
})
df.index = ["a", "b", "c", "d"]

df["salary"] > 2000                        # 得到布尔 Series
print(df[df["salary"] > 20000])             # 布尔行筛选
print(df[df["age"].between(28, 34)])          # 闭区间
print(df[df["city"].isin(["北京", "广州"])])    # 成员判断
print(df[df["name"].str.startswith("王")])     # 字符串方法
print(df[df["name"].str.contains("三|五", regex=True)])   # 正则

print(df.loc["b", "salary"])                  # 22000
print(df.loc[["a", "c"], ["name", "age"]])     # 行标签列表 × 列名列表
print(df.loc[df["age"] > 30, ["name", "salary"]])

# 排序
print(df.sort_values("salary", ascending=False))
print(df.sort_values(["city", "salary"], ascending=[True, False]))
print(df.sort_index(ascending=False))

5.5 新增、删除与重命名

import pandas as pd
import numpy as np

df = pd.DataFrame({
    "name":  ["张三", "李四", "王五"],
    "age":   [28, 34, 29],
    "salary":[15000, 22000, 13500],
})

# 新增列
df["bonus"] = df["salary"] * 0.1                       # 标量
df["level"] = np.where(df["salary"] > 18000, "高", "普通")   # 条件赋值
df["首字母"] = df["name"].str[0]                        # 中文首字

# 新增行
df.loc[3] = ["赵六", 41, 30000, 3000.0, "高", "赵"]     # 位置索引追加
df = df.reset_index(drop=True)                         # 重新编号 0,1,2,3

# 删除
df2 = df.drop(columns=["bonus"])                       # 删列
df3 = df.drop(index=[0, 1])                            # 删行
df4 = df.drop_duplicates(subset=["city"], keep="first") # 按列去重

# 重命名
df.rename(columns={"salary": "薪资", "age": "年龄"}, inplace=True)

# 用字典批量替换
mapping = {"张三": "Zhang San", "李四": "Li Si"}
df["name"] = df["name"].replace(mapping)

# 用 apply/map 做逐元素转换
df["薪资万"] = df["薪资"].apply(lambda v: round(v / 10000, 2))
df["name"] = df["name"].map(lambda n: n.upper())       # map 只作用于一个 Series

# 多列函数用 apply
df["数字列统计"] = df[["年龄", "薪资"]].apply(lambda col: col.max() - col.min(), axis=0)
# axis=0 逐列计算，返回 Series；axis=1 逐行计算，返回 Series

读写 CSV：数据进出口

6.1 写 CSV

import pandas as pd
import numpy as np

df = pd.DataFrame({
    "name":  ["张三", "李四", "王五"],
    "age":   [28, 34, 29],
    "score": [88.5, 92.0, 75.5],
})

# 默认：带索引
df.to_csv("out.csv")
# ✗ 不带索引（推荐，中文环境务必 encoding='utf-8-sig'）
df.to_csv("out.csv", index=False, encoding="utf-8-sig")

# 分隔符、日期、压缩
df.to_csv("out.tsv", sep="\t", index=False, encoding="utf-8-sig")
df.to_csv("out.csv.gz", index=False, encoding="utf-8-sig", compression="gzip")

# 指定列顺序 + 替换 NaN 表示
df.to_csv("out.csv", index=False, columns=["name", "age", "score"],
          na_rep="N/A", encoding="utf-8-sig")

6.2 读 CSV：常用参数全解

read_csv_demo.py · 需要先把上面的 CSV 写出来

import pandas as pd

# ---- 最简 ----
df = pd.read_csv("out.csv")
df = pd.read_csv("out.csv", encoding="utf-8-sig")

# ---- 常用参数 ----
df = pd.read_csv(
    "out.csv",
    encoding="utf-8-sig",
    sep=",",                    # 分隔符；制表符用 "\t"，也可以 auto 猜
    header=0,                   # 列名所在行；无表头写 header=None
    index_col=0,                # 把第几列当索引
    usecols=["name", "age"],    # 只读需要的列（省内存）
    usecols=lambda c: not c.startswith("Unnamed"),  # 排除空列
    nrows=1000,                 # 只读前 1000 行（探测数据必备）
    encoding_errors="replace",  # 解码错误的处理方式
    na_values=["N/A", "面议", "-", "NULL"],   # 自定义缺失值标记
    keep_default_na=True,       # 是否识别默认缺失值（NA、N/A、NaN、null 等）
    na_filter=True,             # 自动去掉全空行空列
    parse_dates=["日期"],        # 自动解析日期列
    thousands=",",              # 千分位分隔
    decimal=".",                # 小数点符号
    skiprows=2,                 # 跳过开头 2 行（如标题、单位说明）
    nrows_after_skiprows=None,
    low_memory=False,           # 混合类型时关闭分块推断，得到纯 object 列
    dtype={"age": "int32"},     # 指定列类型（最好的性能与正确性手段）
)

# ---- 表格预览（先看再读）----
head = pd.read_csv("out.csv", nrows=5)     # 前 5 行看结构
print(head.dtypes)
print(head.memory_usage(deep=True))         # 预估内存

# ---- 分块读取大文件 ----
total_rows = 0
chunks = pd.read_csv("big.csv", chunksize=100_000)
for chunk in chunks:
    total_rows += len(chunk)
    # 每个分块内部照样能 groupby，最后再 concat
print("总行数：", total_rows)

# 部分结果 = pd.concat(list_of_chunks, ignore_index=True)

# ---- 其他格式 ----
# df = pd.read_excel("a.xlsx", sheet_name="Sheet1", engine="openpyxl")
# df = pd.read_json("a.json")
# df = pd.read_html("https://example.com/table")[0]     # 直接读网页表格！
# df = pd.read_sql("SELECT * FROM t", conn)
# df = pd.read_parquet("a.parquet")                    # 最快的二进制格式
# df = pd.read_stata / read_sas / read_feather

6.3 Excel、数据库与网络表格

import pandas as pd

# Excel：需要 openpyxl / xlrd
with pd.ExcelWriter("report.xlsx", engine="openpyxl") as writer:
    df1.to_excel(writer, sheet_name="明细", index=False)
    summary.to_excel(writer, sheet_name="汇总")
    # 自定义表头样式
    # from openpyxl.styles import Font
    # ws = writer.sheets["明细"]
    # ws["A1"].font = Font(bold=True)

# 数据库：用第 17 章学的连接对象
import sqlite3
conn = sqlite3.connect("app.db")
df = pd.read_sql_query("SELECT * FROM users WHERE age > ?",
                       conn, params=(18,))
# ⚠️ 用参数化查询（params），不要用字符串拼接，否则有 SQL 注入风险
# 大表可分块：for chunk in pd.read_sql_query(..., chunksize=50000)

# 网页表格：read_html 会把所有表格都返回成一个列表
tables = pd.read_html("https://www.stats.gov.cn/sj/zxfb/")
print(len(tables))
df0 = tables[0]

# 网页表格的补充技巧：指定匹配的属性名，能避开「无关小表格」
# tables = pd.read_html(url, match="年份")
# tables = pd.read_html(url, attrs={"class": "data-table"})

6.4 序列化与网络

import pandas as pd, json

# 与 JSON 互转
json_str = df.to_json(orient="records", force_ascii=False, indent=2)
back = pd.read_json(json_str)
df.to_json("out.json", orient="records", force_ascii=False)

# 与 Excel/CSV 之外的通用序列化
df.to_pickle("out.pkl")      # 快但有安全风险，只用于自己产生的可信文件
df2 = pd.read_pickle("out.pkl")

# 与 SQL 引擎（推荐的高效存储）
# pip install duckdb
# import duckdb
# duckdb.sql("SELECT city, AVG(salary) FROM df GROUP BY city").df()

数据清洗：缺失值、重复值与类型转换

7.1 缺失值的四种表示

7.2 缺失值处理全策略

import pandas as pd
import numpy as np

df = pd.DataFrame({
    "name":   ["张三", "李四", None, "赵六", "王五"],
    "age":    [28.0, np.nan, 29.0, 41.0, np.nan],
    "city":   ["北京", "上海", None, "北京", "广州"],
    "salary": [15000.0, 22000.0, np.nan, 30000.0, 18000.0],
})

# ---- 1) 统计缺失 ----
print(df.isna().sum())                    # 每列缺失个数
print(df.isna().mean().round(2))           # 缺失比例
print(df.info(missing_counts=True))
print(df[df.isna().any(axis=1)])           # 任何一列缺失的行

# ---- 2) 删除 ----
print(df.dropna())                        # 删掉含缺失的任意一行的行
print(df.dropna(how="all"))               # 全是缺失才删
print(df.dropna(subset=["name", "age"]))  # 只看指定列
print(df.dropna(thresh=2))                # 至少保留 2 个非空才留
print(df.dropna(axis=1))                  # 删列（整列大量缺失时）

# ---- 3) 填充 ----
df1 = df.fillna(0)                                   # 标量
df2 = df.fillna({"age": 30, "city": "未知"})            # 按列指定
df3 = df.fillna(method="ffill")                      # 前值填充（forward fill）
df4 = df.fillna(method="bfill", limit=2)             # 后值填充，最多 2 个
df5 = df.ffill(axis=1)                              # 同行横向填充
df6 = df.groupby("city")["salary"].transform("mean")  # 分组均值填充（最常用）

# ---- 4) 插值（适合有时间趋势的连续序列）----
ts = pd.Series([1, 2, np.nan, 4, np.nan, 6],
               index=pd.date_range("2024-01-01", periods=6, freq="D"))
print(ts.interpolate())              # 线性插值
print(ts.interpolate(method="time")) # 按时间距离插值
print(ts.interpolate(limit_direction="both"))

# ---- 5) 替换与去除占位符 ----
df3 = df.replace({"city": {"": None, "未知城市": None}})
df4 = df.mask(df["age"] > 60)       # 条件置 NaN

# ---- 6) 可空整数（保留 int 语义的同时表达缺失）----
nullable = df["salary"].astype("Int64")
print(nullable.dropna().dtype)      # Int64（pandas 可空整数）
# ⚠️ 缺失的 Int64 列直接算 sum 会跳过；但整列全缺失时 dtype 变 float64

7.3 重复值的识别与处理

import pandas as pd

df = pd.DataFrame({
    "id":     [1, 1, 2, 3, 4, 4],
    "name":   ["张三", "张三", "李四", "王五", "赵六", "赵六"],
    "age":    [28, 28, 34, 29, 41, 42],
    "city":   ["北京", "北京", "上海", "北京", "广州", "广州"],
})

print(df.duplicated())                        # 整行完全重复的行（保留第一行）
print(df.duplicated(subset=["id"]))           # 只看某几列是否重复
print(df.duplicated(subset=["id"], keep="last"))   # 标记除最后一行外的
print(df.duplicated(subset=["id"], keep=False))    # 标记所有重复行
print(df.drop_duplicates())                   # 删除完全重复行
print(df.drop_duplicates(subset=["id"]))      # 按业务键去重
print(df.drop_duplicates(subset=["name"], keep="last"))   # 保留最后出现的

# 数据有「更新」时，通常保留最新记录
df_sorted = df.sort_values("age")
latest = df_sorted.drop_duplicates(subset=["id"], keep="last")

# 行列方向都要查
print(df.duplicated(axis=1, keep="first"))    # 重复的列

7.4 数据类型与异常值

import pandas as pd
import numpy as np

df = pd.DataFrame({
    "价格":   ["12.5", "8.0", "面议", "100.2"],
    "日期":   ["2024-01-05", "2024/02/06", "2024年03月07日", "2024-04-08"],
    "数量":   ["1", "2", "3", "很多"],
    "标签":   ["优", "良", "中", "差"],
})

# 类型转换：errors='coerce' 把无法转换的置为 NaN
df["价格"] = pd.to_numeric(df["价格"], errors="coerce")
df["数量"] = pd.to_numeric(df["数量"], errors="coerce")
df["日期"] = pd.to_datetime(df["日期"], format="mixed", errors="coerce")
# ⚠️ 日期格式混乱时需要 format='mixed'，否则会解析失败

# 分类类型：显著省内存，字符串重复多时效果明显
df["标签"] = df["标签"].astype("category")
print(df["标签"].dtype)                       # category
print(df["标签"].cat.categories.tolist())
# df["标签"].cat.codes        变成 0,1,2,3 整数编码
# df["标签"] = df["标签"].cat.add_categories("特差")

# 数值四舍五入与截断
df["价格"] = df["价格"].round(1)
df["价格"] = df["价格"].astype("Int64")        # 可空整数

# 异常值：IQR 法
s = pd.Series(np.random.default_rng(0).normal(100, 15, 1000))
q1, q3 = s.quantile(0.25), s.quantile(0.75)
iqr = q3 - q1
lo, hi = q1 - 1.5 * iqr, q3 + 1.5 * iqr
outliers = s[(s < lo) | (s > hi)]
print(f"正常范围 [{lo:.1f}, {hi:.1f}]，异常 {len(outliers)} 个")
s_clipped = s.clip(lower=lo, upper=hi)        # 截断（保留数据形状）

# z-score 法
z = (s - s.mean()) / s.std()
print(s[z.abs() > 3])                       # |z| > 3 视为异常

7.5 行列变换

import pandas as pd
import numpy as np

df = pd.DataFrame({
    "name": ["张三", "李四", "王五"],
    "age":  [28, 34, 29],
    "city": ["北京", "上海", "北京"],
})

# map：逐元素（Series 级）
df["city"] = df["city"].map({"北京": "Beijing", "上海": "Shanghai"})

# apply：逐 Series 或逐行（函数可更复杂）
df["age_desc"] = df["age"].apply(lambda a: "青年" if a < 30 else "中年")

# replace：全局替换
df = df.replace({"青年": "young", "中年": "middle"})

# melt：宽表 → 长表（做透视表的反向操作）
wide = pd.DataFrame({"name": ["A", "B"], "一月": [1, 2], "二月": [3, 4]})
long = wide.melt(id_vars="name", var_name="month", value_name="sales")
print(long)
#   name month  sales
# 0    A   一月      1
# 1    B   一月      2
# 2    A   二月      3
# 3    B   二月      4

# pivot：长表 → 宽表
print(long.pivot(index="name", columns="month", values="sales"))

# 三个以上变量时用 melt 配合 id_vars
# applymap / DataFrame.map：逐元素（2.1+ 用 df.map）
small = pd.DataFrame({"a": [1, 2], "b": [3, 4]})
print(small.map(lambda v: v * 10))

分组聚合与透视表

8.1 groupby 的心智模型

groupby 本质是一个「拆分 → 应用 → 合并（split-apply-combine）」的过程。理解这三步，剩下的 API 都是细节。

import pandas as pd
import numpy as np

df = pd.DataFrame({
    "城市":   ["北京", "北京", "上海", "上海", "广州", "广州"],
    "部门":   ["技术", "销售", "技术", "技术", "销售", "技术"],
    "姓名":   ["张三", "李四", "王五", "赵六", "钱七", "孙八"],
    "销售额": [12000, 8000, 15000, 9000, 6000, 11000],
    "成本":   [3000, 2000, 4000, 2500, 1500, 3200],
})

# 分组求和
g = df.groupby("城市")["销售额"].sum()
print(g)
g = df.groupby("城市").agg({"销售额": "sum", "成本": "sum"})
print(g)
g = df.groupby("城市").agg(总额=("销售额", "sum"), 均值=("销售额", "mean"), 笔数=("销售额", "size"))
print(g)

# 多列分组：默认返回 MultiIndex
g2 = df.groupby(["城市", "部门"])["销售额"].agg(["sum", "mean", "count"])
print(g2)

# 自定义聚合函数
g3 = df.groupby("城市").agg(
    最高=("销售额", "max"),
    最低=("销售额", "min"),
    极差=("销售额", lambda s: s.max() - s.min()),
    总额=lambda s: s.sum(),
)

# 一次算多个统计量（describe 最省事）
print(df.groupby("城市")["销售额"].describe())

# named aggregation：给结果列起中文名（推荐写法）
print(df.groupby("城市").agg(
    总销售额=("销售额", "sum"),
    总成本=("成本", "sum"),
))
print(df.groupby("城市").agg(
    毛利=("销售额", lambda s: s.sum()),
))

# ---- transform：聚合值广播回每一行（关键技巧）----
df["城市平均"] = df.groupby("城市")["销售额"].transform("mean")
df["相对水平"] = df["销售额"] / df["城市平均"]
print(df[["城市", "销售额", "城市平均", "相对水平"]].round(2))

# 常见用途：组内去重（每城取销售额最高的人）
idx = df.groupby("城市")["销售额"].idxmax()
print(df.loc[idx])
# idxmax 返回「原始行标签」，所以配合 .loc 用，不要用 .iloc

# ---- apply：分组后对整个子表处理 ----
def high_low(sub: pd.DataFrame) -> pd.Series:
    return pd.Series({"最高": sub["销售额"].max(),
                      "最低": sub["销售额"].min(),
                      "极差": sub["销售额"].max() - sub["销售额"].min()})

print(df.groupby("城市").apply(high_low))
# ⚠️ apply 比 agg 慢，能用 agg 就别用 apply

8.2 常用聚合函数速查

8.3 透视表

import pandas as pd
import numpy as np

df = pd.DataFrame({
    "城市":  ["北京", "北京", "上海", "上海", "广州", "广州"],
    "季度":  ["Q1", "Q2", "Q1", "Q2", "Q1", "Q2"],
    "产品线": ["硬件", "软件", "硬件", "软件", "硬件", "软件"],
    "销售额": [12000, 8000, 15000, 9000, 6000, 11000],
})

# ---- pivot_table：更灵活，推荐 ----
pt = df.pivot_table(index="城市", columns="季度",
                    values="销售额", aggfunc="sum", fill_value=0)
print(pt)
# 城市   Q1     Q2
# 上海  15000   9000
# 北京  12000   8000
# 广州   6000  11000

# 多行多列
pt2 = df.pivot_table(index=["城市", "产品线"], columns="季度",
                     values="销售额", aggfunc=["sum", "mean"],
                     fill_value=0, margins=True, margins_name="合计")
print(pt2)

# ---- pivot：更接近 Excel 的透视表 ----
# 要求「城市×季度」组合唯一，否则报 ValueError: Index contains duplicate entries
pt3 = df.pivot(index="城市", columns="季度", values="销售额")
print(pt3)
# 有重复时先聚合：df.pivot_table(...).reset_index().pivot(...)

# ---- crosstab：列联表，统计出现次数（很适合分类变量）----
ct = pd.crosstab(df["城市"], df["产品线"])
print(ct)
# crosstab 内部就是 pivot_table(values=1, aggfunc=len)

# ---- get_dummies：把分类变量转成 0/1 列（机器学习特征工程常用）----
d = pd.get_dummies(df["城市"], prefix="city", dtype=int)
print(pd.concat([df, d], axis=1))

8.4 concat、merge 与 join：合并数据集

import pandas as pd

orders = pd.DataFrame({
    "订单号": ["A001", "A002", "A003", "A004"],
    "客户ID": [1, 2, 1, 3],
    "金额":   [100, 200, 150, 300],
})
customers = pd.DataFrame({
    "客户ID": [1, 2, 3],
    "姓名":   ["张三", "李四", "王五"],
    "城市":   ["北京", "上海", "广州"],
})
extra = pd.DataFrame({
    "订单号": ["A001", "A003"],
    "销售员": ["小赵", "小钱"],
})

# ---- merge：按键连接（SQL 的 JOIN）----
print(pd.merge(orders, customers, on="客户ID", how="inner"))
print(pd.merge(orders, customers, on="客户ID", how="left"))
# 用左表全部行，右表缺失处填 NaN —— 做数据富化最常用

# 左右键名不同
print(pd.merge(orders, customers, left_on="客户ID", right_on="客户ID"))

# 多键连接
# pd.merge(df1, df2, on=["年份", "月份"])

# 按索引连接
d1 = pd.DataFrame({"v": [1, 2]}, index=[0, 1])
d2 = pd.DataFrame({"w": [3, 4]}, index=[0, 1])
print(pd.merge(d1, d2, left_index=True, right_index=True, how="outer"))

# ---- concat：纵向/横向拼接 ----
a = pd.DataFrame({"v": [1, 2]})
b = pd.DataFrame({"v": [3, 4]})
print(pd.concat([a, b], ignore_index=True))          # 上下拼
print(pd.concat([a, b], keys=["上半年", "下半年"]))   # 带分组键

c = pd.DataFrame({"w": [9, 9]})
print(pd.concat([a, c], axis=1))                      # 左右拼（按索引对齐）

# ---- merge 的坑 ----
# 1) 一对多会让行数膨胀，先检查
print(orders.merge(customers, on="客户ID").shape)   # (4,5)
# 2) 键的 dtype 必须一致：int64 vs object 匹配不上
# 3) 连接后列名冲突会加 _x/_y 后缀，可用 suffixes=() 自定义
print(pd.merge(orders, orders, on="订单号", suffixes=("", "_副本")))

8.5 apply 的四个维度

import pandas as pd

df = pd.DataFrame({"a": [1, 2], "b": [3, 4], "c": [5, 6]})

print(df.apply(sum))                       # axis=0 默认，逐列 → Series
print(df.apply(lambda c: c.sum(), axis=0))
print(df.apply(lambda r: r["a"] + r["b"], axis=1))   # 逐行
print(df.map(lambda v: v * 2))            # 逐元素

# 向量化替代 apply（差 10~100 倍速度）
df["s1"] = df["a"] + df["b"]              # ✅ 快
df["s2"] = df.apply(lambda r: r["a"] + r["b"], axis=1)  # ❌ 慢，⚠️ 未来版本会被弃用

时间序列处理

9.1 生成时间索引

import pandas as pd
import numpy as np

# 频率别名表（选摘）
FREQ = {
    "D": "日", "B": "工作日", "W": "周", "M": "月末", "MS": "月初",
    "Q": "季末", "QS": "季初", "H": "小时", "T": "分钟", "S": "秒",
    "Y"/"YE": "年末", "W-MON": "每月第一个周一",
}

idx = pd.date_range("2024-01-01", periods=10, freq="D")
print(idx[:3])

# 工作日序列（跳过周末）—— 金融/办公场景非常常用
bd = pd.date_range("2024-01-01", "2024-01-31", freq="B")
print(bd[:5])

# 月末、季末
me = pd.date_range("2024-01-01", periods=6, freq="ME")
qe = pd.date_range("2024-01-01", periods=4, freq="QE")

# 从已有列创建时间索引
df = pd.DataFrame({
    "日期": ["2024-01-01", "2024-01-02", "2024-01-03"],
    "销售额": [100, 200, 300],
})
df["日期"] = pd.to_datetime(df["日期"])
df = df.set_index("日期")
print(df.index.dtype)     # datetime64[ns]
print(df["日期"].dt.year)  # .dt 访问器：提取日期分量

9.2 dt 访问器

import pandas as pd

df = pd.DataFrame({
    "订单时间": ["2024-03-05 14:23:11", "2024-03-06 09:05:00", "2024-12-31 23:59:59"],
})
df["订单时间"] = pd.to_datetime(df["订单时间"])

df["年"]   = df["订单时间"].dt.year
df["月"]   = df["订单时间"].dt.month
df["日"]   = df["订单时间"].dt.day
df["星期"] = df["订单时间"].dt.dayofweek        # 0=周一 … 6=周日
df["月名"] = df["订单时间"].dt.month_name()
df["季度"] = df["订单时间"].dt.quarter
df["时分"] = df["订单时间"].dt.hour
df["周几"] = df["订单时间"].dt.day_name(locale="zh_CN")
df["是月末"] = df["订单时间"].dt.is_month_end
df["是工作日"] = df["订单时间"].dt.dayofweek < 5
print(df)

# 归一化：截断到某个粒度
ts = pd.Series([1, 2, 3, 4],
               index=pd.to_datetime(["2024-01-01 01:00", "2024-01-01 13:00",
                                     "2024-01-02 02:00", "2024-01-03 09:00"]))
print(ts.to_period("D").value_counts().sort_index())   # 按天统计
print(ts.resample("ME").sum())                        # 按月求和

9.3 重采样与滚动窗口

import pandas as pd
import numpy as np

# 构造每分钟一笔的成交数据
n = 24 * 60
idx = pd.date_range("2024-01-01", periods=n, freq="min")
rng = np.random.default_rng(42)
ts = pd.Series(rng.normal(1000, 200, n), index=idx, name="成交额")
ts.index.name = "时间"

# ---- resample：把高频聚合到低频（必须有 DatetimeIndex）----
daily = ts.resample("D").agg({"成交额": ["sum", "mean", "max", "count"]})
print(daily)
daily.columns = ["总成交额", "平均", "最大", "笔数"]
print(daily["总成交额"].sum())

# ⚠️ 会产生空桶（缺失的时间段）。用 fill 策略：
ts.resample("h").sum().fillna(0)              # 填 0（计数类指标合适）
ts.resample("h").mean().interpolate()         # 插值（均值类指标合适）
ts.resample("h").agg(["mean", "std"]).round(2)

# 按周、按季
week = ts.resample("W-MON").agg(["sum", "count"])
month = ts.resample("ME").sum()

# 分组 + 重采样（每个城市各自的日成交）
city_ts = ts.to_frame().assign(城市=np.random.default_rng(0).choice(["北京","上海"], n))
res = city_ts.groupby("城市").resample("D")["成交额"].sum()
print(res)

# ---- rolling：滚动窗口（窗口随时间滑动，不是固定分组）----
s = pd.Series(range(1, 11), dtype=float)
print(s.rolling(3).sum())            # [nan 3. 6. 9. ...]  窗口不足处为 NaN
print(s.rolling(3, min_periods=1).sum())   # 不足时也算
print(s.rolling(3).mean())
print(s.rolling(3).std())

# 常用：20 日均线（金融分析基础）
price = pd.Series(rng.normal(100, 3, 120),
                  index=pd.date_range("2024-01-01", periods=120, freq="D"),
                  name="close")
ma20 = price.rolling(window=20).mean()
ma60 = price.rolling(window=60).mean()
print(price.tail(3).round(2))

# 指数加权移动平均：对近端数据给更高权重
ewm = price.ewm(span=10, adjust=False).mean()

# 差分与变化率
print(price.diff().tail())            # 环比差值
print(price.pct_change().tail())      # 环比百分比

# ---- shift / lag ----
print(price.shift(1).tail())          # 前移一天：今天的值用昨天的（做预测的特征）
print(price.shift(-1).head())         # 后移一天

# ---- expanding：累积窗口 ----
print(s.expanding().sum())            # 前缀和

9.4 时间序列的完整分析流程

import pandas as pd
import numpy as np

# 1) 准备：日成交额数据，索引为日期
n = 365
idx = pd.date_range("2023-01-01", periods=n, freq="D")
rng = np.random.default_rng(7)
trend = np.linspace(0, 500, n)
season = 300 * np.sin(np.arange(n) / 365 * 2 * np.pi)
noise = rng.normal(0, 120, n)
df = pd.DataFrame({
    "日期": idx,
    "销售额": 1000 + trend + season + noise,
    "渠道": np.where(idx.dayofweek < 5, "工作日", "周末"),
})
df["日期"] = pd.to_datetime(df["日期"])
df = df.set_index("日期").sort_index()

# 2) 检查：是否有缺失日期（节假日停售等）
print("缺失天数：", len(pd.date_range(df.index.min(), df.index.max(), freq="D")) - len(df))

# 3) 月度趋势
monthly = df["销售额"].resample("ME").agg(["sum", "mean", "std"])
monthly["环比"] = monthly["sum"].pct_change().round(4)
print(monthly.round(1))

# 4) 同比（去年同期）→ 同比列
df["年"] = df.index.year
df["月"] = df.index.month
yoy = df.groupby(["月"])["销售额"].mean().unstack()
print(yoy.round(1))

# 5) 渠道对比
print(df.groupby("渠道")["销售额"].agg(["mean", "std", "count"]).round(1))

# 6) 移动平均去噪
df["MA7"] = df["销售额"].rolling(7).mean()
df["MA30"] = df["销售额"].rolling(30).mean()

# 7) 简单预测：用最近 7 天均值 + 趋势外推
last = df["销售额"].tail(30)
trend7 = np.polyfit(np.arange(len(last)), last.values, 1)
next_idx = pd.date_range(df.index.max() + pd.Timedelta(days=1), periods=7, freq="D")
pred = np.polyval(trend7, np.arange(len(last), len(last) + 7))
forecast = pd.Series(pred, index=next_idx, name="预测销售额")
print(forecast.round(1))

# 完整时序表
report = pd.concat([df[["销售额", "MA7"]], forecast.to_frame()], axis=1)
print(report.tail(10).round(1))

完整实战：从脏数据到结论

下面把本章所有知识串成一个端到端的分析流程。数据用代码生成，读者可以完整跑通；换成第 19 章爬来的真实 CSV 即可直接复用流程。

"""sales_analysis.py —— 销售数据端到端分析

流程：生成/读取 → 探查 → 清洗 → 特征工程 → 分组聚合 → 透视表 → 时序 → 导出
"""
from __future__ import annotations

import numpy as np
import pandas as pd

pd.set_option("display.unicode.east_asian_width", True)   # 中文对齐好看
pd.set_option("display.max_columns", 20)
pd.set_option("display.width", 160)


# ---------- 1. 造一份「脏」的原始数据 ----------
def make_raw_data(n: int = 2000) -> pd.DataFrame:
    rng = np.random.default_rng(42)
    cities = ["北京", "上海", "广州", "深圳", "杭州"]
    cats = ["电子", "服饰", "食品", "图书", "家居"]

    df = pd.DataFrame({
        "订单号": [f"O{i:06d}" for i in range(n)],
        "下单时间": pd.date_range("2024-01-01", periods=n, freq="h").astype(str),
        "城市": rng.choice(cities, n),
        "品类": rng.choice(cats, n),
        "渠道": rng.choice(["app", "web", "小程序", None], n, p=[.4, .35, .25, .0]),
        "数量": rng.integers(1, 6, n),
        "单价": np.round(rng.normal(200, 80, n), 2),
        "折扣": rng.choice([1.0, 0.9, 0.8, 0.7], n),
        "状态": rng.choice(["已完成", "已完成", "已完成", "已退款", "待付款"], n,
                          p=[.7, .05, .05, .1, .1]),
    })
    # 人为制造脏数据
    df.loc[rng.choice(df.index, 60, replace=False), "单价"] = "面议"
    df.loc[rng.choice(df.index, 40, replace=False), "渠道"] = None
    df.loc[rng.choice(df.index, 15, replace=False), "下单时间"] = "无效时间"
    df.loc[rng.choice(df.index, 20, replace=False), "单价"] *= 100   # 异常值
    # 制造重复订单
    dup = df.sample(30, random_state=1)
    return pd.concat([df, dup], ignore_index=True)


# ---------- 2. 清洗 ----------
def clean(df: pd.DataFrame) -> pd.DataFrame:
    df = df.copy()

    # 2.1 类型转换：无法转换的置为 NaN，而不是报错
    df["单价"] = pd.to_numeric(df["单价"], errors="coerce")
    df["数量"] = pd.to_numeric(df["数量"], errors="coerce")
    df["下单时间"] = pd.to_datetime(df["下单时间"], errors="coerce", format="mixed")

    # 2.2 去重（按订单号）
    before = len(df)
    df = df.drop_duplicates(subset=["订单号"], keep="first")
    print(f"去重：{before} -> {len(df)}")

    # 2.3 缺失处理：先看缺失情况再决定
    print(df.isna().sum().to_dict())
    df["渠道"] = df["渠道"].fillna("未知")
    # 关键字段缺失直接丢（业务判断）
    df = df.dropna(subset=["下单时间", "单价", "数量"])

    # 2.4 异常值裁剪（IQR 法）
    q1, q3 = df["单价"].quantile([0.25, 0.75])
    iqr = q3 - q1
    lo, hi = q1 - 1.5 * iqr, q3 + 1.5 * iqr
    n_out = ((df["单价"] < lo) | (df["单价"] > hi)).sum()
    print(f"异常单价：{n_out} 条，范围 [{lo:.1f}, {hi:.1f}]")
    df["单价"] = df["单价"].clip(lo, hi)

    # 2.5 只保留有效状态（分析销售时排除退款与待付）
    df = df[df["状态"] == "已完成"].copy()
    return df


# ---------- 3. 特征工程 ----------
def add_features(df: pd.DataFrame) -> pd.DataFrame:
    df["金额"] = (df["单价"] * df["数量"] * df["折扣"]).round(2)
    ts = df["下单时间"]
    df["年"] = ts.dt.year
    df["月"] = ts.dt.month
    df["日期"] = ts.dt.normalize()
    df["月标签"] = ts.dt.to_period("M").astype(str)
    df["季度"] = ts.dt.quarter
    df["星期"] = ts.dt.dayofweek
    df["是否周末"] = ts.dt.dayofweek >= 5
    df["是月初3天"] = ts.dt.day <= 3
    # 分类变量低基数字段转 category 省内存
    for c in ["城市", "品类", "渠道", "状态"]:
        df[c] = df[c].astype("category")
    return df


# ---------- 4. 分组聚合 ----------
def aggregate(df: pd.DataFrame) -> pd.DataFrame:
    return (df.groupby(["城市", "品类"], observed=True)
              .agg(订单数=("订单号", "nunique"),
                   总金额=("金额", "sum"),
                   平均金额=("金额", "mean"),
                   最大金额=("金额", "max"),
                   总数量=("数量", "sum"))
              .round(2)
              .reset_index())


# ---------- 5. 透视表与交叉分析 ----------
def pivot(df: pd.DataFrame):
    p1 = df.pivot_table(index="城市", columns="品类", values="金额",
                        aggfunc="sum", fill_value=0, margins=True,
                        margins_name="合计").round(0)
    p2 = pd.crosstab(df["城市"], df["渠道"])
    p3 = pd.crosstab(df["月"], df["是否周末"], normalize="index").round(3)
    return p1, p2, p3


# ---------- 6. 时间序列 ----------
def timeseries(df: pd.DataFrame):
    daily = df.groupby("日期")["金额"].agg(["sum", "mean", "count"])
    daily.columns = ["日总额", "日均单额", "订单数"]
    daily["MA7"] = daily["日总额"].rolling(7, min_periods=1).mean().round(2)
    daily["环比"] = daily["日总额"].pct_change().round(4)
    monthly = daily.resample("ME")[["日总额", "订单数"]].sum().round(2)
    monthly.index.name = "月份"
    return daily, monthly


# ---------- 7. 描述统计与排名 ----------
def report(df: pd.DataFrame, agg: pd.DataFrame):
    print("\n===== 整体描述统计 =====")
    print(df[["数量", "单价", "金额"]].describe().round(2))

    print("\n===== 各城市金额占比 =====")
    share = df.groupby("城市", observed=True)["金额"].sum()
    print((share / share.sum()).sort_values(ascending=False).round(4).to_frame("占比"))

    print("\n===== TOP5 订单 =====")
    print(df.nlargest(5, "金额")[["订单号", "城市", "品类", "金额", "下单时间"]])

    print("\n===== 城市 × 品类 聚合（分组结果）=====")
    print(agg.head(10))

    print("\n===== 帕累托：累计占比 =====")
    r = agg.groupby("城市", observed=True)["总金额"].sum().sort_values(ascending=False)
    cum = r.cumsum() / r.sum()
    print(pd.DataFrame({"金额": r.round(0), "累计占比": cum.round(4)}))


# ---------- 8. 导出 ----------
def export(daily: pd.DataFrame, agg: pd.DataFrame, p1: pd.DataFrame):
    with pd.ExcelWriter("销售分析报告.xlsx", engine="openpyxl") as w:
        agg.to_excel(w, sheet_name="城市品类汇总", index=False)
        p1.to_excel(w, sheet_name="透视表")
        daily.to_excel(w, sheet_name="每日趋势")
    daily.to_csv("每日趋势.csv", encoding="utf-8-sig")
    print("\n已导出：销售分析报告.xlsx / 每日趋势.csv")


# ---------- main ----------
def main():
    raw = make_raw_data()
    print(f"原始数据：{raw.shape}")
    print(raw.head())

    df = clean(raw)
    print(f"清洗后：{df.shape}")

    df = add_features(df)
    agg = aggregate(df)
    p1, p2, p3 = pivot(df)
    daily, monthly = timeseries(df)

    report(df, agg)
    print("\n===== 透视表（城市 × 品类）=====")
    print(p1)
    print("\n===== 城市 × 渠道 =====")
    print(p2)
    print("\n===== 月内周末占比 =====")
    print(p3.head())
    print("\n===== 月度趋势 =====")
    print(monthly)

    export(daily, agg, p1)


if __name__ == "__main__":
    main()

性能优化与常见陷阱

11.1 让 Pandas 变快的七个手段

import numpy as np
import pandas as pd

df = pd.DataFrame({
    "a": np.random.default_rng(0).integers(0, 100, 200_000),
    "b": np.random.default_rng(1).normal(0, 1, 200_000),
})

# 1) 省内存
df["a"] = df["a"].astype("int32")
df["b"] = df["b"].astype("float32")
print(df.memory_usage(deep=True).sum() / 1024, "KB")

# 2) 多分支条件用 np.select（极快）
cond = df["b"] > 1
df["档位"] = np.select(
    [df["a"] > 80, df["a"] > 50, df["a"] > 20],
    ["高", "中", "低"],
    default="其他",
)

# 3) 批量取值用 at/iat
idx = df.index[:1000]
vals = df["a"].to_numpy()
out = vals * 2          # 交给 NumPy，最快

# 4) 聚合时只选需要的列（提前裁剪能省大量内存）
small = df[["a", "档位"]]
print(small.groupby("档位", observed=True)["a"].mean())

# 5) merge 前先看匹配率
left = pd.DataFrame({"k": np.arange(100_000)})
right = pd.DataFrame({"k": np.arange(0, 200_000, 2)})
m = left.merge(right, on="k", how="left", indicator=True)
print(m["_merge"].value_counts())
# left_only 99950, both 50000 → 说明 right 覆盖率只有 50%，要先补数据

11.2 十个高频陷阱速查

11.3 什么时候该换工具

# DuckDB：直接对 CSV/Parquet 做 SQL 聚合，内存友好
# pip install duckdb
import duckdb

result = duckdb.sql("""
    SELECT 城市, 品类, SUM(单价 * 数量 * 折扣) AS 总金额
    FROM 'sales.csv'
    WHERE 状态 = '已完成'
    GROUP BY 城市, 品类
    ORDER BY 总金额 DESC
    LIMIT 10
""").df()
print(result)

本章小结

• 分工：NumPy 负责「算」（ndarray、向量化、广播），Pandas 负责「表」（Series、DataFrame、索引对齐、分组聚合）。Pandas 底层基于 NumPy。
• ndarray：统一 dtype + 连续内存；基本切片返回视图，花式/布尔索引返回拷贝；批量计算前先想清楚是对行（axis=1）、对列（axis=0）还是对整体。
• 广播规则是「逐维相等或其一为 1」；归一化时用 keepdims=True 避免形状错误。dtype 要警惕字符串截断、astype(int) 截断而非四舍五入、整数溢出。
• 随机数一律用 np.random.default_rng(seed)，可复现、可并行。
• Series 的一切运算按索引自动对齐，缺失处填 NaN；布尔条件用 & | 加括号；字符串方法挂在 .str 下。
• DataFrame 是不可变视图：只有赋值语句原地生效，链式筛选赋值要用 .loc/.at；选列用 []，定位用 .loc(标签) 与 .iloc(位置)。
• 读写 CSV：写中文文件用 encoding="utf-8-sig"；读时用 usecols、nrows、parse_dates、dtype、na_values、chunksize 控制成本。
• 清洗三板斧：to_numeric(errors="coerce") 治脏类型、dropna/fillna/ffill/interpolate 治缺失、drop_duplicates 治重复；异常值先区分业务异常与统计异常，IQR 与 z-score 是常用判据。
• groupby 是拆分-应用-合并；agg 优于 apply；transform 把聚合值广播回每一行；idxmax 取组内极值所在行。
• pivot_table（可聚合、自动填 NaN）优于 pivot（会因重复报错），crosstab 适合分类计数，melt 是其逆操作。
• 合并用 merge（按键，SQL JOIN 语义，how 决定保留哪边），concat（拼接）；注意键的 dtype 一致与一对多膨胀。
• 时间序列：DatetimeIndex + .dt 访问器提取日期分量，resample 切桶聚合（配 fillna/interpolate），rolling/ewm/expanding 做窗口与平滑，shift/diff/pct_change 做滞后与变化率。
• 性能：选对 dtype、提前裁剪、向量化替代 apply、np.select 做多分支、to_numpy 交给 NumPy；数据量超过内存时用 chunksize 或 DuckDB/Polars。
• 方法论：raw → clean → features → analyze → export 五段式，每个阶段纯函数化；原始数据先存档，筛选逻辑单独调试。

练习题

1. ndarray 与广播专项。给定 a = np.arange(12).reshape(3, 4)，要求：① 用广播把每一列归一化到 0~1（min-max scaling），不使用任何循环；② 取出所有偶数位置的元素；③ 用 np.select 把元素分成「小于 5」「5~10」「大于 10」三档并转成文字标签；④ 说明第 ① 步中若漏掉 keep_dims 会报什么错。
   参考答案
   ① rng = a.max(0) - a.min(0); norm = (a - a.min(0)) / rng（min(0)/max(0) 返回 (4,)，靠广播自动按列拉伸）。
   ② a[(np.arange(a.shape)[:, None] + np.arange(a.shape)[None, :]) % 2 == 0]，或简单用 a[a % 2 == 0]（等价于按值筛偶数）。
   ③ lbl = np.select([a < 5, a <= 10], ["小", "中"], default="大")。
   ④ 若写 row_min = a.min(1) 得 (3,)，再做 (a - row_min) / (a.max(1) - row_min) 会因 (3,3) 与 (3,4) 不兼容而报 operands could not be broadcast together；加 keepdims=True 得 (3,1) 即可。
2. Series 索引对齐实验。创建两个 Series（索引分别为 ['a','b','c'] 和 ['b','c','d']），完成：① 相加、② 相乘、③ 求并集索引的结果、④ 用 .add(..., fill_value=0) 消除 NaN；解释每一步的输出并说明索引对齐规则。
   参考答案要点相加结果索引为 ['a','b','c','d']，只有 b、c 是双方共有的非 NaN 值；.add(other, fill_value=0) 把只在一边存在的标签视为 0，从而得到完整的对齐结果。这条「对齐」规则在 merge、concat、算术运算中一致存在，是 Pandas 的核心语义。
3. 读取并清洗真实 CSV。构造一个含中文、缺失值、重复行、混乱日期格式与脏数值（如「面议」）的 orders.csv（UTF-8 编码），要求用 read_csv 一次读入并完成：指定 na_values、转换数值与日期类型、去重、按业务键填充缺失、删除关键字段缺失的行。每一步打印处理前后的行数与缺失数。
   参考答案要点关键参数：encoding="utf-8-sig"、na_values=["面议","-","暂无"]、parse_dates=["下单时间"]（格式混乱时先不 parse，读入后 pd.to_datetime(..., format="mixed", errors="coerce")）；drop_duplicates(subset=["订单号"])；用 df.isna().sum() 在每步前后打印。务必先 df.head() 与 df.dtypes 探查再动手。
4. 分组聚合与 transform。对一张销售表，要求：① 按城市分组算总金额、订单数、平均客单价（count/mean）；② 用 transform("mean") 给每一行加上「所在城市的平均金额」和「相对城市的倍数」；③ 用 idxmax 找出每个城市金额最高的一条订单；④ 用 agg 一次性算出 max/min/极差并用中文列名。
   参考答案要点② df["城市均值"] = df.groupby("城市")["金额"].transform("mean")；③ best = df.loc[df.groupby("城市")["金额"].idxmax()]（注意是 .loc 不是 .iloc）；④ df.groupby("城市").agg(最高=("金额","max"), 最低=("金额","min"), 极差=("金额", lambda s: s.max()-s.min()))，这叫命名聚合（named aggregation）。
5. 透视表三连。用同一张销售表：① pivot_table 输出城市 × 品类的金额总和，含行列合计（margins）；② crosstab 输出城市 × 渠道的订单数；③ melt 把 ① 的结果再变回长表。比较三者适用场景。
   参考答案要点① pivot_table(index="城市", columns="品类", values="金额", aggfunc="sum", fill_value=0, margins=True, margins_name="合计")；若 品类 是 category 类型，需加 observed=True 避免出现未观测组合的全 0 列；② pd.crosstab(df["城市"], df["渠道"])；③ pd.melt(表, id_vars="城市", var_name="品类", value_name="金额")。结论：透视表给人看报表，crosstab 适合看分布，melt 是透视的逆操作用于绘图与建模。
6. 时间序列全流程。生成一年（365 天）的日销售数据（含周末效应与轻微上升趋势），要求：① 转成 DatetimeIndex；② 计算 7 日与 30 日移动平均；③ 按月 resample 求和并计算环比；④ 找出销售额最高的 5 天和最低的 5 天；⑤ 用 shift(1) 构造「昨日销售额」特征列。
   参考答案要点② s.rolling(7).mean()、s.rolling(30, min_periods=1).mean()；③ m = s.resample("ME").sum()，环比 = m.pct_change().round(4)（重采样后若有空月份需先 .fillna(0) 或 reindex(pd.period_range(...))）；④ s.nlargest(5) / s.nsmallest(5)；⑤ df["昨日"] = df["销售额"].shift(1)（首行会变 NaN，这是正确行为，不要用 0 填充）。
7. 多表合并实战。有三张表：orders（订单号、客户ID、金额）、customers（客户ID、姓名、注册日期）、regions（客户ID、区域）。要求：① 用 merge left 把客户信息与区域信息依次并到订单上；② 用 indicator=True 统计有多少订单没匹配到客户；③ 把注册日期算成「注册月数」；④ 导出为 utf-8-sig 的 CSV 并用 Excel 打开验证中文不乱码。
   参考答案要点① pd.merge(orders, customers, on="客户ID", how="left").merge(regions, on="客户ID", how="left")；② 加入 indicator="_m" 后 res["_m"].value_counts()，left_only 的行即未匹配；③ (pd.Timestamp.today() - df["注册日期"]).dt.days / 30.4 或用 pd.Period 做差；④ df.to_csv("out.csv", index=False, encoding="utf-8-sig")，BOM 让 Excel 正确识别中文。
8. 性能对比实验。构造 50 万行的 DataFrame（含一个数值列与一个分类列），分别用三种方式求「按分类分组的均值」：① groupby.mean()；② apply 逐行；③ np.bincount 或 np.add.at。用 time.perf_counter 计时并输出三者倍数关系，并解释瓶颈在哪里。
   参考答案要点计时模板：t0=time.perf_counter(); r=...; print(f"{time.perf_counter()-t0:.3f}s")。结论：groupby 基于哈希表，接近 O(n)；逐行 apply 有 50 万次 Python 函数调用，慢 2 个数量级；把分类列先转成 category 或用 pd.factorize 得到整数编码后交给 NumPy 聚合最快。额外加分：给出 df.memory_usage(deep=True) 在 astype("category") 前后的对比。
9. （选做）端到端分析报告。把第 10 节的 sales_analysis.py 跑通，然后：① 把造数据换成读取第 19 章爬虫产出的真实 CSV（字段名不同处做 rename 映射）；② 增加一个「复购率」指标（同一客户ID 在 30 天内有 ≥2 次订单的比例，用 groupby+rolling 或时间差分实现）；③ 增加一个帕累托分析（城市/品类贡献度的累计占比，找出贡献 80% 的头部）；④ 生成一页含透视表与时序汇总的 Excel。
   参考答案要点复购率：df.sort_values(["客户ID","下单时间"]); df["间隔天数"] = df.groupby("客户ID")["下单时间"].diff().dt.days; 复购率 = (df["间隔天数"] <= 30).sum() / df["客户ID"].nunique()；帕累托：g = df.groupby("品类")["金额"].sum().sort_values(ascending=False); cum = g.cumsum()/g.sum(); 头部 = g[cum <= 0.8].index；导出沿用 pd.ExcelWriter 多 sheet 写法。评分要点是流程完整性与代码组织，而非数值结果。