12 种 Pandas 测试技巧，让数据处理少踩坑

Pandas 的 bug 有个特点，就是不会在控制台里大喊大叫，而是悄悄藏在 dtype 转换、索引操作、时区处理的某个角落，或者那种跑十万次才能复现一次的边界条件。所以如果你想找到和定位这种隐藏的BUG就需要一套相对简洁的测试手段能把大部分坑提前暴露出来。

下面这 12 个策略是实际项目里反复使用的测试方法，能让数据处理代码变得比较靠谱。

1) 用 Pytest Fixtures 做 DataFrame 工厂

弄几个小巧的 fixture "工厂"来生成样例数据，这样setup 代码会少写很多，测试逻辑反而能写更充分。

# conftest.py

import pandas as pd

import numpy as np

import pytest

@pytest.fixture

def sales_df():

  return pd.DataFrame({

      "order_id": [1, 2, 3],

      "country": ["IN", "US", "IN"],

      "amount": [99.0, 149.5, np.nan],

      "ts": pd.to_datetime(["2025-09-01", "2025-09-02", "2025-09-02"])

  })

然后在哪里都能直接用：

def test_revenue_total(sales_df):

  assert sales_df["amount"].sum(skipna=True) == 248.5

一个标准样本反复使用，减少重复代码，测试的样例也更容易读懂。

2) Schema 层来约束数据

Dtype 会漂，列也可能会丢。所以加个 schema 检查，这样违规的数据在边界就会被暴露出来。用pandera这类工具也行，或者自己写个轻量检查：

def assert_schema(df, expected):

  # expected: dict[column] -> dtype string, e.g. {"order_id": "int64", ...}

  assert set(df.columns) == set(expected), "Columns mismatchDEEPHUB"

  for c, dt in expected.items():

      assert str(df[c].dtype) == dt, f"{c} dtype mismatch: {df[c].dtype} != {dt}"

def test_schema(sales_df):

  assert_schema(sales_df, {

      "order_id": "int64",

      "country": "object",

      "amount": "float64",

      "ts": "datetime64[ns]"

  })

数据结构变化能第一时间发现，不会传到转换逻辑深处才暴露。

3) Property-Based Testing 检查不变量

有些规则应该对任意输入都成立，比如归一化之后总和还是 1。所以可以用 Hypothesis 自动生成各种输入来验证：

from hypothesis import given, strategies as st

import pandas as pd

import numpy as np

@given(st.lists(st.floats(allow_nan=False, width=32), min_size=1, max_size=50))

def test_normalize_preserves_sum(xs):

  s = pd.Series(xs, dtype="float32")

  total = float(s.sum())

  if total == 0:  # define behavior on zero-sum

      return

  normalized = s / total

  assert np.isclose(float(normalized.sum()), 1.0, atol=1e-6)

一个测试用例能自动覆盖几十种形状、数值范围和边界情况。

4) 参数化测试把边缘 case 都列出来

有一些经典的麻烦场景：空 DataFrame、单行数据、重复索引、全 null 列、混时区。

import pytest

import pandas as pd

import numpy as np

@pytest.mark.parametrize("df", [

  pd.DataFrame(columns=["a","b"]),

  pd.DataFrame({"a":[1], "b":[np.nan]}),

  pd.DataFrame({"a":[1,1], "b":[2,2]}).set_index("a"),

])

def test_transform_handles_edges(df):

  out = df.assign(b=df.get("b", pd.Series(dtype=float)).fillna(0.0))

  assert "b" in out

一次性把健壮性锁定，以后就不用反复调同样的边界问题了。

5) Golden Snapshot + 校验和来固定输出

复杂输出可以存个"黄金样本"，CI 里对比校验和。如果输出变了会直接报错。

import pandas as pd

import hashlib

def df_digest(df: pd.DataFrame) -> str:

  b = df.sort_index(axis=1).to_csv(index=False).encode()

  return hashlib.md5(b).hexdigest()

def test_output_snapshot(tmp_path, sales_df):

  out = (sales_df

         .assign(day=sales_df["ts"].dt.date)

         .groupby(["country","day"], as_index=False)["amount"].sum())

  expected = pd.read_parquet("tests/golden/agg.parquet")

  assert df_digest(out) == df_digest(expected)6) 固定随机数和时间

如果转换依赖随机或者当前时间，得把种子钉死。

import numpy as np

import pandas as pd

from datetime import datetime

def stratified_sample(df, frac, rng):

  return df.groupby("country", group_keys=False).apply(lambda g: g.sample(frac=frac, random_state=rng))

def test_sample_is_deterministic(sales_df):

  rng = np.random.default_rng(42)

  a = stratified_sample(sales_df, 0.5, rng)

  rng = np.random.default_rng(42)

  b = stratified_sample(sales_df, 0.5, rng)

  pd.testing.assert_frame_equal(a.sort_index(), b.sort_index())

这个没什么说的，模型训练的时候也要固定随机种子

7) 明确测试 NA 的语义

NaN、None、pd.NA在不同操作下行为差异挺大的，这时候需要把预期行为显式写出来：

import pandas as pd

import numpy as np

def test_na_logic():

  s = pd.Series([1, np.nan, 3])

  s2 = s.fillna(0)

  assert s2.isna().sum() == 0

  assert s2.iloc[1] == 0

NA 相关的 bug 经常藏在 groupby、merge 和数学运算里，得当成一等公民来测。

8) 索引、排序、唯一性约束

函数如果保证"索引稳定"就测索引，依赖排序就断言排序状态。

def is_monotonic(df, column):

  return df[column].is_monotonic_increasing

def test_index_and_sort(sales_df):

  out = sales_df.sort_values(["ts","order_id"]).set_index("order_id")

  assert out.index.is_unique

  assert is_monotonic(out.reset_index(), "ts")

很多逻辑错误其实就是顺序错了或者不小心有重复。

9) 双实现交叉验证

聪明的向量化逻辑可以用"慢但一看就懂"的实现来验证：

import pandas as pd

def vectorized_net(df):

  return df.assign(net=df["amount"] - df["amount"].mean())

def slow_net(df):

  mean = df["amount"].mean()

  df = df.copy()

  df["net"] = df["amount"].apply(lambda x: x - mean)

  return df

def test_equivalence(sales_df):

  a = vectorized_net(sales_df)

  b = slow_net(sales_df)

  pd.testing.assert_series_equal(a["net"], b["net"], check_names=False)

防止向量化实现出现细微错误，同时保持性能优势。

10) 性能预算作为冒烟测试

不需要精确的 benchmark，设个大概的护栏就够了。用小规模代表性数据跑一下，给个时间上限：

import time

import pandas as pd

def test_runs_fast_enough(sales_df):

  small = pd.concat([sales_df]*2000, ignore_index=True)  # ~6k rows  hub

  t0 = time.perf_counter()

  _ = small.groupby("country", as_index=False)["amount"].sum()

  dt = time.perf_counter() - t0

  assert dt < 0.25  # budget for CI environment11) I/O 往返保证

CSV、Parquet、Arrow 格式往返可能会改类型，测一下关心的部分：

import pandas as pd

import numpy as np

def test_parquet_round_trip(tmp_path, sales_df):

  p = tmp_path / "sales.parquet"

  sales_df.to_parquet(p, index=False)

  back = pd.read_parquet(p)

  pd.testing.assert_frame_equal(

      sales_df.sort_index(axis=1),

      back.sort_index(axis=1),

      check_like=True

  )

"本地跑得好好的，生产环境因为 I/O 就挂了"这种谜之问题可能就出现在这里

12) Join 操作的基数和覆盖率检查

Merge 是数据质量最容易出问题的地方，基数、重复、覆盖率都得显式断言。

import pandas as pd

def test_merge_cardinality():

  left = pd.DataFrame({"id":[1,2,3], "x":[10,20,30]})

  right = pd.DataFrame({"id":[1,1,2], "y":[5,6,7]})

  out = left.merge(right, on="id", how="left")

  # Expect duplicated rows for id=1 because right has two matches   deephub

  assert (out["id"] == 1).sum() == 2

  # Coverage: every left id appears at least once

  assert set(left["id"]).issubset(set(out["id"]))

key 不唯一或者行数意外翻倍的时候能立刻发现。

小结

好的 Pandas 代码不光要写得聪明，更重要的是可预测。这 12 个策略能让正确性变成默认状态：fixtures 快速启动、schemas 早期失败、property-based tests 探索各种古怪情况、简单的性能预算阻止慢代码偷偷溜进来。本周先试两三个，接到 CI 里，那些神秘的数据 bug 基本就消失了。

作者：Syntal

点个在看你最好看！