12 種 Pandas 測試技巧，讓數據處理少踩坑
12 種測試實踐 —— fixtures、schemas、property-based tests、snapshots、performance guards —— 每週能省不少排查問題的時間
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 mismatch"
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] == 0NA 相關的 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
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
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 基本就消失了。
https://avoid.overfit.cn/post/6eca2c51cef244849e52ae02b932efa9
作者：Syntal
