Pandas 完全指南

最后更新：2026-03-27

官方文档：https://pandas.pydata.org/docs/
适用版本：pandas 2.x（2026-05-08 核实）

相关文档：SQLAlchemy完全指南 MySQL基础完全指南

目录

1. Pandas完全指南 · 安装
2. Pandas完全指南 · 核心数据结构
3. Pandas完全指南 · 数据读写
4. Pandas完全指南 · 数据查看
5. Pandas完全指南 · 数据选取
6. Pandas完全指南 · 数据清洗
7. Pandas完全指南 · 数据变换
8. Pandas完全指南 · 字符串操作
9. Pandas完全指南 · 日期时间操作
10. Pandas完全指南 · 分组聚合
11. Pandas完全指南 · 合并与连接
12. Pandas完全指南 · 透视与重塑
13. Pandas完全指南 · 排序
14. Pandas完全指南 · 窗口函数
15. Pandas完全指南 · 性能优化
16. Pandas完全指南 · 常用配合库
17. Pandas完全指南 · 最佳实践
18. Pandas完全指南 · 应用场景
19. Pandas完全指南 · 常见陷阱与注意事项

安装

# 基础安装
pip install pandas

# 常用可选依赖
pip install openpyxl        # Excel 读写（.xlsx）
pip install xlrd            # Excel 读取旧格式（.xls）
pip install pyarrow         # Parquet 读写 / Arrow backend（pandas 2.0+）
pip install fastparquet     # Parquet 读写（另一实现）
pip install sqlalchemy      # read_sql / to_sql
pip install lxml            # read_html / XML 解析
pip install beautifulsoup4  # read_html HTML 解析备选
pip install tables          # HDF5 读写（PyTables）
pip install fsspec          # 远程文件系统（S3/GCS）

# 一次性安装所有常用依赖
pip install "pandas[all]"

版本确认：

import pandas as pd
print(pd.__version__)   # 例：2.2.0

核心数据结构

Series

Series 是带标签的一维数组，可存储任意类型数据。

创建 Series

import pandas as pd
import numpy as np

# 从列表创建
s = pd.Series([10, 20, 30, 40])

# 从字典创建（键成为索引）
s = pd.Series({"a": 1, "b": 2, "c": 3})

# 指定索引
s = pd.Series([10, 20, 30], index=["x", "y", "z"], name="score")

# 从标量广播
s = pd.Series(5, index=range(5))

# 从 NumPy 数组
s = pd.Series(np.arange(5), dtype="float32")

常用属性

s = pd.Series([10, 20, 30], index=["a", "b", "c"], name="score")
print(s.index)    # Index(['a', 'b', 'c'], dtype='object')
print(s.values)   # [10 20 30]
print(s.dtype)    # int64
print(s.shape)    # (3,)
print(s.name)     # score

DataFrame

DataFrame 是带标签的二维表格结构，每列可以是不同类型。

创建 DataFrame

# 从字典（最常用）
df = pd.DataFrame({
    "name": ["Alice", "Bob", "Charlie"],
    "age": [25, 30, 35],
    "score": [88.5, 92.0, 78.3],
})

# 从列表套字典
df = pd.DataFrame([
    {"name": "Alice", "age": 25},
    {"name": "Bob",   "age": 30},
])

# 从二维列表（需指定列名）
df = pd.DataFrame(
    [[1, 2, 3], [4, 5, 6]],
    columns=["a", "b", "c"],
    index=["row1", "row2"],
)

# 从 NumPy ndarray
arr = np.random.randn(4, 3)
df = pd.DataFrame(arr, columns=["x", "y", "z"])

# 从 records（列表套元组）
df = pd.DataFrame.from_records(
    [(1, "Alice"), (2, "Bob")],
    columns=["id", "name"],
)

# 从 CSV 文件（见数据读写章节）
df = pd.read_csv("data.csv")

常用属性

df.info()
# <class 'pandas.core.frame.DataFrame'>
# RangeIndex: 3 entries, 0 to 2
# Data columns (total 3 columns): ...

df.describe()
#        age      score
# count  3.0   3.000000
# mean  30.0  86.266667
# ...

print(df.dtypes)
# name      object
# age        int64
# score    float64

数据读写

read_csv()

读取 CSV 或文本文件。

df = pd.read_csv("data.csv")
df = pd.read_csv("data.csv", sep="\t", encoding="utf-8", nrows=1000)

# 分块读取大文件
for chunk in pd.read_csv("big.csv", chunksize=10000):
    process(chunk)

# 只读取指定列，并指定类型
df = pd.read_csv(
    "sales.csv",
    usecols=["date", "amount", "region"],
    dtype={"amount": "float32", "region": "category"},
    parse_dates=["date"],
)

read_excel()

读取 Excel 文件（.xlsx 需要 openpyxl，.xls 需要 xlrd）。

df = pd.read_excel("data.xlsx", sheet_name="Sheet1")

# 读取多个 sheet
sheets = pd.read_excel("data.xlsx", sheet_name=None)  # 返回 {name: df}

# 读取指定列范围
df = pd.read_excel("data.xlsx", usecols="A:E", nrows=100)

read_json()

读取 JSON 文件或字符串。

df = pd.read_json("data.json")
df = pd.read_json('{"a":[1,2],"b":[3,4]}')

# 读取 JSONL 格式（每行一个 JSON 对象）
df = pd.read_json("data.jsonl", lines=True)

# orient='records' 对应列表套字典格式
df = pd.read_json('[{"name":"Alice","age":25}]', orient="records")

read_parquet()

读取 Parquet 列式存储文件（需要 pyarrow 或 fastparquet）。

df = pd.read_parquet("data.parquet")

# 只读取部分列（Parquet 支持列裁剪，效率高）
df = pd.read_parquet("data.parquet", columns=["id", "amount"])

# 带过滤条件（分区裁剪）
df = pd.read_parquet("data.parquet", filters=[("year", "==", 2024)])

read_sql() / read_sql_query()

从数据库读取数据（需要 SQLAlchemy）。

from sqlalchemy import create_engine
engine = create_engine("postgresql+psycopg2://user:pass@localhost/db")

df = pd.read_sql("SELECT * FROM users", con=engine)
df = pd.read_sql_query("SELECT id, name FROM users WHERE age > 25", con=engine)
df = pd.read_sql_table("users", con=engine)

写出函数

to_csv()

df.to_csv("output.csv", index=False, encoding="utf-8-sig")  # utf-8-sig 避免 Excel 乱码

to_excel()

# 多 sheet 写出
with pd.ExcelWriter("output.xlsx", engine="openpyxl") as writer:
    df1.to_excel(writer, sheet_name="销售", index=False)
    df2.to_excel(writer, sheet_name="库存", index=False)

to_parquet()

to_json()

# 输出 JSONL 格式（常用于流式处理）
df.to_json("output.jsonl", orient="records", lines=True, force_ascii=False)

数据查看

head() / tail() / sample()

df.head(5)      # 前 5 行（默认）
df.tail(10)     # 后 10 行
df.sample(5)    # 随机 5 行
df.sample(frac=0.1, random_state=42)  # 随机 10%，固定随机种子

info() / describe() / value_counts()

df.info()                    # 列名、非空数、类型、内存
df.info(memory_usage="deep") # 精确内存统计

df.describe()                # 数值列统计摘要
df.describe(include="all")   # 包含字符串/category 列
df.describe(include="object")

df["city"].value_counts()               # 频次统计，降序
df["city"].value_counts(normalize=True) # 比例
df["city"].value_counts(dropna=False)   # 包含 NaN

shape / dtypes / columns / index

df.shape      # (100, 5) — (行数, 列数)
df.dtypes     # 每列数据类型
df.columns    # Index(['a', 'b', 'c'])
df.index      # RangeIndex(start=0, stop=100, step=1)

# 修改列名
df.columns = ["col1", "col2", "col3"]

# 列名转列表
cols = df.columns.tolist()

数据选取

这是 pandas 使用最频繁的核心章节。

[] 基础选取

# 选取单列 -> 返回 Series
s = df["name"]

# 选取多列 -> 返回 DataFrame
sub = df"name", "age"

# 行切片（仅支持整数或标签切片）
df[0:5]       # 前 5 行（整数切片，不含第 5 行）
df["a":"c"]   # 标签切片，含端点（适用于有字符串索引的 df）

注意：[] 用于行切片时是基于位置（整数）或标签，行为依索引类型而定，容易混淆，建议优先使用 loc / iloc。

loc[]

基于标签（行索引名和列名）选取数据。

# 单行
df.loc[0]
df.loc["alice"]

# 单个值
df.loc[0, "name"]

# 行范围（含端点）
df.loc[0:5]
df.loc["2024-01-01":"2024-01-31"]

# 行 + 列
df.loc[0:5, "name":"age"]
df.loc[[0, 2, 4], ["name", "score"]]

# 布尔数组
mask = df["age"] > 25
df.loc[mask]
df.loc[mask, "name"]

# 赋值（推荐方式）
df.loc[0, "score"] = 99.0
df.loc[df["age"] > 30, "grade"] = "senior"

iloc[]

基于整数位置选取数据（0-based，不含末端）。

# 单行
df.iloc[0]
df.iloc[-1]   # 最后一行

# 单个值
df.iloc[0, 2]  # 第 0 行第 2 列

# 行范围（不含末端）
df.iloc[0:5]
df.iloc[::2]   # 每隔一行

# 行 + 列
df.iloc[0:5, 0:3]
df.iloc[[0, 2, 4], [1, 3]]

# 赋值
df.iloc[0, 1] = 100

at[] / iat[]

单值快速访问，比 loc/iloc 更快（无需额外开销）。

df.at[0, "name"]        # 标签访问单值
df.iat[0, 2]            # 位置访问单值

df.at[0, "score"] = 95  # 赋值
df.iat[0, 2] = 95

布尔索引

# 单条件
df[df["age"] > 25]

# 多条件（必须用括号和位运算符）
df[(df["age"] > 25) & (df["city"] == "北京")]
df[(df["score"] < 60) | (df["score"] > 90)]
df[~(df["city"] == "北京")]  # 取反

# isin（IN 过滤）
df[df["city"].isin(["北京", "上海", "广州"])]

# between（区间过滤，含端点）
df[df["age"].between(20, 30)]

# str.contains（模糊匹配）
df[df["name"].str.contains("张")]

# notna / isna
df[df["score"].notna()]

query()

使用字符串表达式过滤，语法更简洁，支持链式调用。

df.query("age > 25")
df.query("city == '北京' and score >= 80")
df.query("city in ['北京', '上海']")
df.query("age.between(20, 30)")  # 需要 numexpr

# 引用外部变量（加 @ 前缀）
threshold = 25
df.query("age > @threshold")

# 列名含空格时用反引号
df.query("`first name` == 'Alice'")

多条件过滤总结

# 方法一：布尔索引（最通用）
result = df[(df["age"] > 25) & (df["score"] > 80) & (df["city"] == "北京")]

# 方法二：query（最简洁）
result = df.query("age > 25 and score > 80 and city == '北京'")

# 方法三：np.logical_and.reduce（多条件组合）
import numpy as np
conditions = [df["age"] > 25, df["score"] > 80, df["city"] == "北京"]
result = df[np.logical_and.reduce(conditions)]

数据清洗

缺失值处理

isna() / notna()

df.isna()              # 逐元素 True/False
df.isna().sum()        # 每列缺失数量
df.isna().sum() / len(df)  # 缺失比例
df["score"].isna()
df["score"].notna()

fillna()

填充缺失值。

df.fillna(0)
df.fillna({"age": 0, "score": df["score"].mean()})
df["score"].fillna(df["score"].median(), inplace=True)
df.ffill()   # 前向填充（推荐替代 method='ffill'）
df.bfill()   # 后向填充

dropna()

删除含缺失值的行或列。

df.dropna()                          # 删除含任意 NaN 的行
df.dropna(subset=["name", "score"])  # 只检查这两列
df.dropna(how="all")                 # 全为 NaN 才删
df.dropna(thresh=3)                  # 保留至少有 3 个非 NaN 的行

重复值处理

duplicated()

df.duplicated()                    # 返回布尔 Series
df.duplicated(subset=["name"])     # 只按 name 判断重复
df[df.duplicated()]                # 查看重复行
df.duplicated().sum()              # 重复行数量

drop_duplicates()

df.drop_duplicates()
df.drop_duplicates(subset=["name", "date"], keep="last")

类型转换

astype()

df["age"].astype(int)
df["price"].astype("float32")
df["category"].astype("category")
df.astype({"age": int, "score": "float32"})

pd.to_numeric()

pd.to_numeric(df["score"], errors="coerce")  # 无法转换的变为 NaN

pd.to_datetime()

pd.to_datetime(df["date"])
pd.to_datetime(df["date"], format="%Y-%m-%d")
pd.to_datetime(df["ts"], unit="s")    # Unix 时间戳（秒）
pd.to_datetime(df["date"], errors="coerce")  # 非法日期转 NaT

重命名

rename()

df.rename(columns={"old_name": "new_name", "age": "年龄"})
df.rename(columns=str.lower)           # 全部列名转小写
df.rename(index={0: "first_row"})      # 重命名行索引

重置索引

reset_index()

df.reset_index()             # 原索引成为列，新增 RangeIndex
df.reset_index(drop=True)    # 丢弃原索引，常用于过滤后重置行号

set_index()

df.set_index("id")
df.set_index(["year", "month"])  # 多层索引
df.set_index("date", drop=True)

数据变换

apply()

对 DataFrame 的行或列（或 Series 的每个元素）应用函数。

# 对每列应用
df.apply(lambda col: col.max() - col.min())

# 对每行应用（axis=1）
df.apply(lambda row: row["score"] * 1.1 if row["grade"] == "A" else row["score"], axis=1)

# result_type='expand' 将返回的 list 展开为多列
df.apply(lambda row: [row["x"] + 1, row["y"] * 2], axis=1, result_type="expand")

# 传额外参数
def scale(col, factor):
    return col * factor

df.apply(scale, args=(2.0,))
df.apply(scale, factor=2.0)

map() / applymap() / pipe()

# Series.map()：逐元素映射（替代值 or 应用函数）
df["grade"].map({"A": 4.0, "B": 3.0, "C": 2.0})
df["score"].map(lambda x: round(x, 1))

# DataFrame.map()（pandas 2.1+，之前叫 applymap）：逐元素应用函数
df.map(lambda x: str(x).strip())

# applymap()：pandas < 2.1 的写法（2.1 起已弃用，改用 map）
df.applymap(lambda x: x ** 2)

# pipe()：链式传入整个 DataFrame，用于 method chaining
def normalize(df, col):
    df[col] = (df[col] - df[col].mean()) / df[col].std()
    return df

df.pipe(normalize, "score").pipe(normalize, "age")

向量化操作

pandas 支持直接在列之间做算术和比较运算，底层由 NumPy 加速，避免使用循环。

df["total"] = df["price"] * df["quantity"]
df["discount_price"] = df["price"] * (1 - df["discount_rate"])
df["is_adult"] = df["age"] >= 18
df["z_score"] = (df["score"] - df["score"].mean()) / df["score"].std()

# 字符串拼接
df["full_name"] = df["first_name"] + " " + df["last_name"]

# 条件赋值：np.where
import numpy as np
df["label"] = np.where(df["score"] >= 60, "pass", "fail")

# 多条件赋值：np.select
conditions = [df["score"] >= 90, df["score"] >= 60]
choices = ["优秀", "合格"]
df["grade"] = np.select(conditions, choices, default="不合格")

assign()

返回新 DataFrame（不修改原数据），适合链式操作。

df2 = (
    df
    .assign(total=df["price"] * df["quantity"])
    .assign(tax=lambda x: x["total"] * 0.13)
    .assign(final=lambda x: x["total"] + x["tax"])
)

列的增删改

# 新增列
df["new_col"] = 0
df["new_col"] = df["a"] + df["b"]

# 修改列
df["score"] = df["score"] * 1.1

# 删除列
df.drop(columns=["col1", "col2"])        # 返回新 df
df.drop(columns="col1", inplace=True)    # 原地删除
del df["col1"]                           # 直接删除（原地）
df.pop("col1")                           # 删除并返回该列

# 插入列到指定位置
df.insert(loc=2, column="middle_col", value=0)

字符串操作

通过 Series.str 访问器对字符串列进行操作，自动处理 NaN（返回 NaN 而不是报错）。

str.contains()

df[df["name"].str.contains("张", na=False)]
df[df["email"].str.contains(r"@gmail\.com", regex=True, na=False)]

str.startswith() / str.endswith()

df[df["filename"].str.endswith((".csv", ".xlsx"), na=False)]

str.replace()

df["phone"].str.replace(r"\D", "", regex=True)           # 去除非数字
df["text"].str.replace("旧词", "新词", regex=False)      # 字面替换

str.strip() / str.lstrip() / str.rstrip()

df["name"].str.strip()
df["code"].str.strip("0")  # 去除前后的 "0"

str.split()

df["name"].str.split(" ")                        # 返回 Series of lists
df["name"].str.split(" ", expand=True)           # 展开为多列
df["date"].str.split("-", expand=True, n=2)      # 最多分 2 次

str.extract() / str.findall()

# extract：提取捕获组
df["date"].str.extract(r"(\d{4})-(\d{2})-(\d{2})")  # 返回 3 列
df["email"].str.extract(r"(?P<user>\w+)@(?P<domain>\w+)")  # 命名捕获组

# findall：返回所有匹配
df["text"].str.findall(r"\d+")

str.upper() / str.lower() / str.len()

df["name"].str.upper()
df["name"].str.lower()
df["name"].str.title()      # 首字母大写
df["text"].str.len()        # 字符串长度（NaN 返回 NaN）
df["text"].str.count("a")   # 字符出现次数

日期时间操作

pd.to_datetime()

见 Pandas完全指南 · 类型转换 章节中的参数表。

dt 访问器属性

通过 Series.dt 访问器操作 datetime 类型列。

# 先转换为 datetime 类型
df["date"] = pd.to_datetime(df["date"])

# 提取日期部分
df["date"].dt.year
df["date"].dt.month
df["date"].dt.day
df["date"].dt.hour
df["date"].dt.minute
df["date"].dt.second
df["date"].dt.microsecond
df["date"].dt.weekday    # 0=周一, 6=周日
df["date"].dt.day_name() # 星期名称（英文）
df["date"].dt.month_name()
df["date"].dt.dayofyear  # 一年中的第几天
df["date"].dt.quarter    # 季度（1-4）
df["date"].dt.is_leap_year
df["date"].dt.days_in_month

dt.strftime() / dt.floor() / dt.ceil() / dt.normalize()

df["date"].dt.strftime("%Y年%m月%d日")
df["date"].dt.floor("H")        # 向下取整到小时
df["date"].dt.ceil("T")         # 向上取整到分钟
df["date"].dt.normalize()       # 截断到日（时间部分归零）
df["date"].dt.tz_localize("Asia/Shanghai")    # 本地化时区
df["date"].dt.tz_convert("UTC")               # 转换时区

DateOffset / Timedelta

from pandas.tseries.offsets import MonthEnd, BDay

# Timedelta（固定时间差）
pd.Timedelta("1 days 2 hours")
pd.Timedelta(days=1, hours=2)
df["date"] + pd.Timedelta(days=7)

# DateOffset（日历感知）
df["date"] + pd.DateOffset(months=1)   # 下个月同日
df["date"] + MonthEnd(1)               # 下个月末
df["date"] + BDay(5)                   # 5 个工作日后

resample()

对时间序列进行重采样（必须以 datetime 为索引）。

# 设置时间索引
df = df.set_index("date")

# 按月聚合
df.resample("M").sum()
df.resample("M").agg({"amount": "sum", "count": "count"})

# 按周求均值
df["price"].resample("W").mean()

# 降采样（粗粒度）
df.resample("Q").last()      # 每季度最后一个值

# 升采样（细粒度）后填充
df.resample("D").ffill()     # 日频，前向填充

date_range()

生成日期范围。

pd.date_range("2024-01-01", "2024-12-31", freq="D")   # 全年每天
pd.date_range("2024-01-01", periods=12, freq="MS")     # 12 个月初
pd.date_range("2024-01-01", periods=5, freq="B")       # 5 个工作日

分组聚合

groupby()

g = df.groupby("city")
g = df.groupby(["city", "year"])
g = df.groupby("city", as_index=False)
g = df.groupby("city", sort=False, observed=True)

agg() / aggregate()

# 单函数
df.groupby("city")["score"].agg("mean")

# 多函数
df.groupby("city")["score"].agg(["mean", "std", "count"])

# 按列指定不同函数
df.groupby("city").agg({
    "score": ["mean", "max"],
    "age":   "median",
})

# 命名聚合（pandas 0.25+，推荐）
df.groupby("city").agg(
    avg_score=("score", "mean"),
    max_score=("score", "max"),
    count=("id", "count"),
)

transform()

对每组应用函数，返回与原 DataFrame 等长的结果（不降维）。

# 组内标准化（z-score）
df["score_z"] = df.groupby("city")["score"].transform(
    lambda x: (x - x.mean()) / x.std()
)

# 用组均值填充缺失值
df["score"] = df.groupby("city")["score"].transform(lambda x: x.fillna(x.mean()))

# 添加组统计到原 df
df["city_avg"] = df.groupby("city")["score"].transform("mean")
df["city_rank"] = df.groupby("city")["score"].transform("rank", ascending=False)

filter()

过滤满足条件的组（整组保留或整组丢弃）。

# 保留组内记录数 >= 10 的组
df.groupby("city").filter(lambda g: len(g) >= 10)

# 保留组均值 > 80 的组
df.groupby("city").filter(lambda g: g["score"].mean() > 80)

apply() 在 GroupBy 上的用法

# 每组取 score 最高的 3 条
df.groupby("city").apply(lambda g: g.nlargest(3, "score"))

# 每组自定义处理
def process_group(g):
    g = g.copy()
    g["rank"] = g["score"].rank(ascending=False)
    return g

df.groupby("city").apply(process_group)

常用聚合函数汇总

合并与连接

merge()

类似 SQL JOIN，默认内连接。

# 基本用法
result = pd.merge(df_left, df_right, on="id")

# 不同列名的连接键
result = pd.merge(df_orders, df_users, left_on="user_id", right_on="id")

# 用索引连接
result = pd.merge(df_left, df_right, left_index=True, right_index=True)

# 左连接，重名列加后缀，添加来源标识
result = pd.merge(
    df_left, df_right,
    on="id",
    how="left",
    suffixes=("_left", "_right"),
    indicator=True,
)
# _merge 列值：'left_only' / 'right_only' / 'both'

# 验证连接基数
result = pd.merge(df_left, df_right, on="id", validate="m:1")

各种 how 的对比

# 找出左表独有的行（反连接）
result = pd.merge(df_left, df_right, on="id", how="left", indicator=True)
left_only = result[result["_merge"] == "left_only"].drop(columns="_merge")

join()

基于索引的连接，是 merge 的简化版。

df1.join(df2, how="inner")
df1.join(df2, on="id", rsuffix="_right")

concat()

沿轴拼接多个对象（行拼接或列拼接）。

# 纵向拼接（行堆叠）
df_all = pd.concat([df1, df2, df3], ignore_index=True)

# 带来源标识
df_all = pd.concat([df_2023, df_2024], keys=["2023", "2024"])

# 横向拼接（列合并）
df_wide = pd.concat([df_a, df_b], axis=1)

# 只保留共同列
df_all = pd.concat([df1, df2], join="inner", ignore_index=True)

透视与重塑

pivot_table()

类似 Excel 数据透视表，可处理重复值（通过聚合）。

pd.pivot_table(
    df,
    values="sales",
    index="region",
    columns="product",
    aggfunc="sum",
    fill_value=0,
    margins=True,
)

pivot()

将长格式转宽格式，不做聚合（值必须唯一，否则报错）。

df.pivot(index="date", columns="city", values="temperature")

melt()

将宽格式转长格式（逆透视）。

pd.melt(
    df,
    id_vars=["id", "name"],
    value_vars=["2022_sales", "2023_sales", "2024_sales"],
    var_name="year",
    value_name="sales",
)

stack() / unstack()

# stack()：将最内层列转为行（宽转长）
df.stack()

# unstack()：将最内层行索引转为列（长转宽）
df.unstack()
df.unstack(level=0)     # 指定层级
df.unstack(level="city")

crosstab()

交叉频率表，统计两个或多个因子的频次。

pd.crosstab(df["gender"], df["grade"])
pd.crosstab(df["gender"], df["grade"], normalize="index")   # 行比例
pd.crosstab(df["city"], df["grade"], margins=True)

排序

sort_values()

df.sort_values("score", ascending=False)
df.sort_values(["city", "score"], ascending=[True, False])
df.sort_values("name", key=lambda s: s.str.lower())  # 忽略大小写排序
df.sort_values("score", na_position="first")

sort_index()

df.sort_index()             # 按行索引升序
df.sort_index(ascending=False)
df.sort_index(axis=1)       # 按列名排序

窗口函数

rolling()

滑动窗口计算。

df["price"].rolling(window=5).mean()          # 5 日移动平均
df["price"].rolling(window=5, min_periods=1).mean()  # 不足 5 个也计算
df["price"].rolling(window=5, center=True).mean()    # 居中窗口

# 时间窗口（需 datetime 索引）
df["price"].rolling("7D").mean()   # 过去 7 天均值

# 多列同时计算
df"price", "volume".rolling(5).sum()

expanding()

扩张窗口（从起点到当前行，窗口不断增大）。

df["price"].expanding().mean()    # 累计均值
df["price"].expanding(min_periods=5).std()  # 至少 5 个点才开始计算

ewm()

指数加权移动（指数衰减权重）。

df["price"].ewm(span=20).mean()     # EMA-20（指数移动平均）
df["price"].ewm(halflife=5).std()

窗口函数常用聚合

roll = df["price"].rolling(20)
roll.mean()    # 移动平均
roll.sum()     # 移动求和
roll.std()     # 移动标准差
roll.min()     # 移动最小值
roll.max()     # 移动最大值
roll.var()     # 移动方差
roll.corr(df["volume"])  # 移动相关系数
roll.cov(df["volume"])   # 移动协方差
roll.apply(lambda x: x.iloc[-1] / x.iloc[0] - 1)  # 自定义（N 日涨幅）

性能优化

选择合适的数据类型

不合理的类型是内存浪费的主要来源。

# 查看内存占用
df.info(memory_usage="deep")
df.memory_usage(deep=True).sum() / 1024**2  # MB

# 整数降级
df["age"] = df["age"].astype("int8")         # -128 ~ 127
df["count"] = df["count"].astype("uint16")   # 0 ~ 65535

# 浮点降级
df["ratio"] = df["ratio"].astype("float32")  # 精度约 7 位

# 字符串列转 category（低基数列效果显著）
df["city"] = df["city"].astype("category")
df["status"] = df["status"].astype("category")

使用 category 类型节省内存

适用于基数（唯一值数量）远小于行数的列。

# 转为 category
df["region"] = df["region"].astype("category")

# 内存对比
import sys
s_str = pd.Series(["北京", "上海", "广州"] * 100000)
s_cat = s_str.astype("category")
print(s_str.memory_usage(deep=True))   # ~2.4 MB
print(s_cat.memory_usage(deep=True))   # ~0.3 MB

# category 排序
from pandas.api.types import CategoricalDtype
cat_type = CategoricalDtype(categories=["低", "中", "高"], ordered=True)
df["level"] = df["level"].astype(cat_type)
df.sort_values("level")  # 按自定义顺序排序

eval() / query() 替代向量运算

使用 numexpr 引擎，对大 DataFrame 可节省内存并提速。

# 普通写法（会创建中间数组）
result = df["a"] + df["b"] * df["c"]

# eval 写法（原地计算，节省内存）
result = df.eval("a + b * c")
df.eval("total = a + b * c", inplace=True)

# 引用外部变量
factor = 1.1
df.eval("score_adj = score * @factor", inplace=True)

适用场景：行数 > 10 万、表达式涉及多列运算时效果明显；简单操作反而因解析开销更慢。

chunksize 处理大文件

def process_large_csv(filepath, chunksize=50000):
    results = []
    for chunk in pd.read_csv(filepath, chunksize=chunksize):
        # 每块独立处理
        filtered = chunk[chunk["amount"] > 100]
        agg = filtered.groupby("city")["amount"].sum()
        results.append(agg)
    return pd.concat(results).groupby(level=0).sum()

避免 iterrows()

iterrows() 逐行迭代，性能极差（每行创建 Series，开销大）。

# 错误示范（慢）
for idx, row in df.iterrows():
    df.at[idx, "total"] = row["price"] * row["qty"]

# 正确做法：向量化
df["total"] = df["price"] * df["qty"]

# 如必须遍历，用 itertuples（比 iterrows 快 10x 以上）
for row in df.itertuples(index=True, name="Row"):
    print(row.price, row.qty)

# 或用 apply（次优）
df["total"] = df.apply(lambda r: r["price"] * r["qty"], axis=1)

inplace=True 的真实效果

inplace=True 并不总是节省内存，也不总是更快。

# inplace=True 实际上仍会在内部创建新对象再赋值回去
df.drop(columns=["col"], inplace=True)
# 等效于
df = df.drop(columns=["col"])

# pandas 2.0+ CoW（写时复制）机制下，inplace 行为更一致
# 建议：明确赋值，避免依赖 inplace 的副作用
df = df.dropna().reset_index(drop=True)

使用 PyArrow backend（pandas 2.0+）

# 创建时指定 ArrowDtype
import pyarrow as pa
df = pd.read_csv("data.csv", dtype_backend="pyarrow")

# 手动转换
df = df.convert_dtypes(dtype_backend="pyarrow")

# PyArrow 类型的优势：更低内存占用、更快 IO、原生支持 null（非 NaN）

Copy-on-Write（pandas 2.0+ CoW 机制）

pandas 2.0 引入 CoW，2.0 起可选，3.0 起强制启用。

# 启用 CoW（pandas 2.0/2.1 中需手动开启）
pd.options.mode.copy_on_write = True

# CoW 下，所有切片都是写时复制视图
subset = df"a", "b"   # 不立即复制
subset["a"] = 0            # 触发复制，不影响原 df

# 明确需要独立副本时
copy = df.copy()

# CoW 的主要影响：彻底消除 SettingWithCopyWarning
# 不再需要 df.loc[mask, "col"] = val 来避免链式赋值警告

常用配合库

NumPy

import numpy as np

# DataFrame -> ndarray
arr = df.to_numpy()
arr = df.values        # 旧写法，不推荐
arr = df["score"].to_numpy()

# ndarray -> DataFrame
df = pd.DataFrame(arr, columns=["a", "b", "c"])

# NumPy ufunc 直接应用于 Series
np.sqrt(df["score"])
np.log(df["price"])
np.clip(df["age"], a_min=18, a_max=65)

# NumPy 函数与 pandas 互操作
np.where(df["score"] >= 60, "pass", "fail")
np.select([df["score"] >= 90, df["score"] >= 60], ["A", "B"], default="C")

Matplotlib / Seaborn

df.plot() 基于 Matplotlib，是快速可视化的便捷方式。

import matplotlib.pyplot as plt

df["score"].plot(kind="hist", bins=20, title="Score Distribution")
plt.show()

df.plot(kind="scatter", x="age", y="score", figsize=(8, 5))
plt.show()

# 多列折线图
df"price", "cost".plot(figsize=(12, 4), grid=True)
plt.tight_layout()
plt.show()

# Seaborn
import seaborn as sns
sns.heatmap(df.corr(), annot=True, fmt=".2f", cmap="coolwarm")
sns.boxplot(data=df, x="city", y="score")

SQLAlchemy

见 SQLAlchemy完全指南。

from sqlalchemy import create_engine

engine = create_engine("mysql+pymysql://user:pass@localhost/db")

# 读取
df = pd.read_sql("SELECT * FROM orders WHERE status = 'paid'", con=engine)

# 写入
df.to_sql(
    name="orders_backup",
    con=engine,
    if_exists="replace",   # 'fail' / 'replace' / 'append'
    index=False,
    chunksize=1000,
    method="multi",        # 批量插入，更快
)

to_sql() 重点参数：

PyArrow

import pyarrow as pa
import pyarrow.parquet as pq

# pandas -> Arrow Table
table = pa.Table.from_pandas(df)

# Arrow Table -> pandas
df = table.to_pandas()

# 直接读写 Parquet
pq.write_table(table, "data.parquet", compression="snappy")
table = pq.read_table("data.parquet", columns=["id", "amount"])
df = table.to_pandas()

# 使用 pandas 接口（底层走 pyarrow）
df.to_parquet("data.parquet", engine="pyarrow", compression="snappy")
df = pd.read_parquet("data.parquet", engine="pyarrow")

Dask

Dask 提供与 pandas 几乎相同的 API，支持并行和大于内存的数据集。

import dask.dataframe as dd

# 读取（懒执行）
ddf = dd.read_csv("large_*.csv")
ddf = dd.from_pandas(df, npartitions=4)

# 操作（与 pandas 语法基本相同）
result = ddf[ddf["score"] > 80].groupby("city")["score"].mean()

# 触发计算
result_df = result.compute()

# 转回 pandas
df = ddf.compute()

Polars

Polars 是 Rust 编写的高性能 DataFrame 库，与 pandas 互转方便。

import polars as pl

# pandas -> Polars
pl_df = pl.from_pandas(df)

# Polars -> pandas
df = pl_df.to_pandas()

# 通过 Arrow 互转（零拷贝，更高效）
arrow_table = df_pandas.to_arrow()  # pandas 2.0+ 方法（实验性）
pl_df = pl.from_arrow(arrow_table)

详见 Polars完全指南。

最佳实践

链式操作（Method Chaining）

链式写法让数据处理流程更清晰，避免中间变量。

result = (
    pd.read_csv("sales.csv", parse_dates=["date"])
    .rename(columns=str.lower)
    .dropna(subset=["amount", "product"])
    .assign(
        month=lambda df: df["date"].dt.to_period("M"),
        amount_rmb=lambda df: df["amount"] * 7.2,
    )
    .query("amount > 100")
    .groupby(["month", "product"])
    .agg(total=("amount_rmb", "sum"), cnt=("amount_rmb", "count"))
    .reset_index()
    .sort_values("total", ascending=False)
)

链式操作中，pipe() 可插入自定义函数：

def add_rank(df, col):
    df = df.copy()
    df["rank"] = df[col].rank(ascending=False, method="dense").astype(int)
    return df

result = df.query("score > 60").pipe(add_rank, "score")

避免 SettingWithCopyWarning

该警告提示你正在对切片的副本赋值，修改可能不会反映到原 df。

# 错误示范（链式赋值，触发警告）
df[df["score"] > 60]["grade"] = "pass"  # 不生效

# 正确做法：用 loc 赋值
df.loc[df["score"] > 60, "grade"] = "pass"

# 需要修改子集时，显式 copy
subset = df[df["city"] == "北京"].copy()
subset["discount"] = 0.9

# pandas 2.0+ 启用 CoW 后，上述问题从根本上消除
pd.options.mode.copy_on_write = True

使用 pd.api.types 做类型检查

from pandas.api.types import (
    is_numeric_dtype,
    is_string_dtype,
    is_datetime64_any_dtype,
    is_categorical_dtype,
    is_integer_dtype,
    is_float_dtype,
)

for col in df.columns:
    if is_numeric_dtype(df[col]):
        df[col] = df[col].fillna(0)
    elif is_string_dtype(df[col]):
        df[col] = df[col].fillna("")

大数据集分块处理模式

import pandas as pd

def etl_pipeline(input_path, output_path, chunksize=100_000):
    first = True
    for chunk in pd.read_csv(input_path, chunksize=chunksize):
        # 处理每块
        chunk = (
            chunk
            .dropna(subset=["id", "amount"])
            .assign(amount=pd.to_numeric(chunk["amount"], errors="coerce"))
            .dropna(subset=["amount"])
        )
        # 追加写出
        chunk.to_csv(
            output_path,
            mode="w" if first else "a",
            header=first,
            index=False,
        )
        first = False

明确使用 copy() 避免隐式引用

# 切片是视图，修改会影响原 df（pandas CoW 前的行为）
subset = df"a", "b"
subset["a"] = 0   # 可能触发警告，且行为不确定

# 明确复制
subset = df"a", "b".copy()
subset["a"] = 0   # 安全，不影响原 df

应用场景

数据清洗流水线示例

import pandas as pd

def clean_user_data(filepath):
    df = (
        pd.read_csv(filepath, dtype={"phone": str})
        .rename(columns=str.strip)              # 去除列名空白
        .rename(columns=str.lower)              # 列名小写
        .drop_duplicates(subset=["user_id"])    # 去重
        .dropna(subset=["user_id", "email"])    # 删除关键字段为空的行
        .assign(
            email=lambda df: df["email"].str.strip().str.lower(),
            phone=lambda df: df["phone"].str.replace(r"\D", "", regex=True),
            age=lambda df: pd.to_numeric(df["age"], errors="coerce"),
            created_at=lambda df: pd.to_datetime(df["created_at"], errors="coerce"),
        )
        .loc[lambda df: df["age"].between(0, 150)]   # 过滤异常年龄
        .reset_index(drop=True)
    )
    return df

日志分析示例

import pandas as pd

# 读取 Nginx 日志
df = pd.read_csv(
    "access.log",
    sep=r"\s+",
    header=None,
    names=["ip", "dash1", "user", "time", "request", "status", "size"],
    usecols=["ip", "time", "request", "status", "size"],
    engine="python",
)

df["time"] = pd.to_datetime(df["time"], format="[%d/%b/%Y:%H:%M:%S")
df["status"] = pd.to_numeric(df["status"], errors="coerce")
df["size"] = pd.to_numeric(df["size"].replace("-", "0"), errors="coerce")

# 按小时统计请求量
hourly = df.set_index("time").resample("H").agg(
    requests=("status", "count"),
    errors=("status", lambda x: (x >= 500).sum()),
    traffic_mb=("size", lambda x: x.sum() / 1024**2),
)

# 找出访问量最多的 IP
top_ips = df["ip"].value_counts().head(10)

时序数据处理示例

import pandas as pd

df = pd.read_csv("stock.csv", parse_dates=["date"])
df = df.set_index("date").sort_index()

# 计算技术指标
df["ma5"]  = df["close"].rolling(5).mean()
df["ma20"] = df["close"].rolling(20).mean()
df["ema12"] = df["close"].ewm(span=12).mean()
df["ema26"] = df["close"].ewm(span=26).mean()
df["macd"] = df["ema12"] - df["ema26"]
df["signal"] = df["macd"].ewm(span=9).mean()

# 月度统计
monthly = df["close"].resample("M").agg(
    open="first",
    high="max",
    low="min",
    close="last",
)

# 计算月涨跌幅
monthly["return"] = monthly["close"].pct_change()

报表生成示例

import pandas as pd

df = pd.read_excel("sales.xlsx", parse_dates=["order_date"])

# 销售汇总透视表
pivot = pd.pivot_table(
    df,
    values=["amount", "qty"],
    index=pd.Grouper(key="order_date", freq="Q"),
    columns="product_category",
    aggfunc={"amount": "sum", "qty": "sum"},
    fill_value=0,
    margins=True,
    margins_name="合计",
)

# 格式化并导出
with pd.ExcelWriter("report.xlsx", engine="openpyxl") as writer:
    pivot.to_excel(writer, sheet_name="季度销售", float_format="%.2f")
    df.groupby("region")["amount"].describe().to_excel(writer, sheet_name="区域统计")

常见陷阱与注意事项

SettingWithCopyWarning

# 触发警告的典型写法（链式赋值）
df[df["score"] > 60]["grade"] = "pass"  # 无效赋值，且触发警告

# 正确写法
df.loc[df["score"] > 60, "grade"] = "pass"

原因：df[df["score"] > 60] 可能返回视图也可能返回副本，pandas 无法确定，因此发出警告。

Chained Indexing 问题

# 危险：连续两次 []，行为不确定
df["city"]["Beijing"]   # 不推荐

# 安全：一次性用 loc 完成
df.loc["Beijing", "city"]

inplace 不总是节省内存

# 错误认知：inplace=True 会在原地修改，节省内存
df.sort_values("score", inplace=True)

# 实际：pandas 内部仍创建新对象再替换，内存峰值不变
# 另外 inplace=True 阻断 method chaining，建议放弃使用
df = df.sort_values("score")

groupby 后 index 变化

# 默认 as_index=True，分组键成为索引
g = df.groupby("city")["score"].mean()
print(g.index)   # Index(['北京', '上海', ...], name='city')

# 解决方法一：reset_index()
result = df.groupby("city")["score"].mean().reset_index()

# 解决方法二：as_index=False
result = df.groupby("city", as_index=False)["score"].mean()

# 多级索引时
result = df.groupby(["city", "year"])["score"].mean()
result.xs("北京", level="city")     # 取某级索引

merge 后列名冲突

# 两表都有 name 列，合并后自动加后缀
result = pd.merge(df_left, df_right, on="id")
# 产生 name_x, name_y

# 指定自定义后缀
result = pd.merge(df_left, df_right, on="id", suffixes=("_customer", "_order"))

# 提前重命名，避免冲突
df_right = df_right.rename(columns={"name": "order_name"})

datetime 时区问题

# 无时区 datetime（naive）
dt = pd.Timestamp("2024-01-01 08:00:00")

# 附加时区（localize）
dt_tz = dt.tz_localize("Asia/Shanghai")

# 转换时区
dt_utc = dt_tz.tz_convert("UTC")

# 时区比较：naive 与 tz-aware 不能直接比较
# 统一策略：所有 datetime 都存储为 UTC，展示时转为本地时区

# 读取时指定 UTC
df["ts"] = pd.to_datetime(df["ts"], utc=True)

# 混合时区列会被存为 object 类型，需统一处理

浮点数精度

# 浮点数比较不可靠
0.1 + 0.2 == 0.3   # False

# 安全比较方式
import numpy as np
np.isclose(df["a"], df["b"])                    # 元素级
np.allclose(df["a"].values, df["b"].values)     # 全部

# 金融场景建议用 Decimal 或存整数（分）
# 或转为 string 再比较哈希

# pandas 浮点显示精度
pd.set_option("display.float_format", "{:.4f}".format)

参见

Polars完全指南
SQLAlchemy完全指南
FastAPI完全指南