你的測試套件是全綠的。你的日誌靜悄悄。你的儀表板上沒有任何紅線。然而,3% 的使用者收到的發票總額是負數,或者你的推薦模型默默地將已刪除的產品排在最前面,或者你的聚合 pipeline 對某個時區的退款重複計算。
這些是資料 bug。它們不會拋出例外。不會讓 pod 崩潰。它們穿過你可觀測性堆疊的每一層,因為每一層都假設資料是正確的。程式碼完全照預期執行。問題在於它寫下的內容根本是胡說。
統計除錯是一種將生產資料視為訊號、將 bug 視為該訊號中異常的實踐。你不是問「程式碼有沒有當掉?」,而是問「資料看起來跟平常一樣嗎?」當答案是否定的時候,你就找到了一個沒有任何stack trace會顯示的 bug。
統計除錯到底是什麼
統計除錯不是機器學習。你不需要神經網路。你需要的是一張直方圖,以及願意被它嚇到的勇氣。
核心概念是:正確的軟體會產生具有可預測統計特性的資料。使用者年齡集中在 18 到 80 歲之間。購買金額遵循對數常態分布。API 回應時間有長尾但中位數穩定。當這些特性發生偏移時,pipeline 中的某個環節動了手腳。一次新的部署、一個 schema migration、一個第三方 API 開始回傳空字串而不是 null。偏移是症狀,bug 是病因。
這與傳統除錯的方向相反。傳統除錯從錯誤開始,往回追溯到程式碼。統計除錯從資料開始,往回追溯到產生它的錯誤。
一個只有統計除錯才能抓到的 bug
這裡有一個真實的模式。一個支付服務重構了它的貨幣轉換邏輯。新程式碼通過了所有測試。整合測試模擬了匯率 API,驗證在模擬匯率下 100 美元會變成 85 歐元。沒有任何斷言失敗。
在生產環境中,匯率 API 偶爾會對次要貨幣回傳 null。舊程式碼會拋出錯誤並退回到快取匯率。新程式碼由一位不了解 fallback 機制的工程師撰寫,在 JavaScript 中將 null 強制轉型為 0,並以零匯率儲存交易。沒有例外,交易成功提交,使用者被收取零元。
你的錯誤追蹤工具什麼都沒看到。你的延遲圖表平坦如鏡。但你的 XOF 貨幣 exchange_rate 值分布,突然在零的位置出現了一根巨大的尖峰。一張直方圖可以在幾秒鐘內顯示這個問題。測試套件永遠找不到它。
如何在生產資料中偵測異常
統計除錯最簡單的形式是分布比較。你挑選一個指標,從歷史資料計算它的分布,然後與最近一小時的分布比較。如果兩者顯著不同,就是有東西改變了。
以下是一個具體的 Python 實作,使用 Kolmogorov-Smirnov 檢定,這是一種不需要對分布形狀做任何假設的非參數兩樣本比較方法。
import numpy as np
from scipy import stats
def detect_distribution_shift(
baseline: np.ndarray,
current: np.ndarray,
threshold: float = 0.05
) -> dict:
"""
Compare two samples using the two-sample KS test.
Returns whether the distributions differ significantly.
"""
# Drop NaNs; they are often the bug themselves
baseline = baseline[~np.isnan(baseline)]
current = current[~np.isnan(current)]
if len(baseline) == 0 or len(current) == 0:
return {"shift_detected": True, "reason": "empty_sample"}
statistic, p_value = stats.ks_2samp(baseline, current)
return {
"shift_detected": p_value < threshold,
"ks_statistic": statistic,
"p_value": p_value,
"baseline_mean": np.mean(baseline),
"current_mean": np.mean(current),
"baseline_std": np.std(baseline),
"current_std": np.std(current),
}
# Example: compare yesterday's purchase amounts to the last hour
baseline = np.random.lognormal(mean=3.0, sigma=1.0, size=10_000)
# Simulate the bug: 5% of transactions now have a zero amount
current = np.concatenate([
np.random.lognormal(mean=3.0, sigma=1.0, size=950),
np.zeros(50)
])
result = detect_distribution_shift(baseline, current)
print(result)
# {'shift_detected': True, 'ks_statistic': 0.052, ...}
這並不高級。這是一個從 1939 年就存在的兩樣本統計檢定。但它會抓到零匯率 bug、雙重計算退款 bug 與負數發票 bug,因為所有這些都會以可測量的方式改變資料的形狀。
關鍵在於選對指標。好的候選對象是任何應該保持穩定的東西:比率(refund_rate、cart_abandonment_rate)、邊界(age、price、quantity)、形狀(HTTP 狀態碼的分布、每小時註冊模式)與相關性(purchase_amount 與 session_duration)。如果你的程式碼正確,這些關係就是不變的。如果它們改變了,就是你的程式碼改變了它們。
分布比較的限制
KS 檢定有盲點。它對整體分布的偏移敏感,但可能會錯過沒有大幅移動全局形狀的局部異常。
假設你的 bug 只影響立陶宛的使用者,而且只在凌晨 2 點到 3 點之間。全球購買金額的分布看起來正常。這個 bug 被其他所有時區的雜訊掩蓋了。單一的全球比較抓不到它。
解決方法是分層(stratification)。與其只做一個全球測試,不如對資料的各個切片分別執行測試:按地理區域、裝置類型、使用者等級、一天中的小時。一個在全球視角下不可見的 bug,在正確的切片中可能會非常明顯。
from dataclasses import dataclass
from typing import Iterator
@dataclass
class DataSlice:
dimension: str # e.g. "country_code"
value: str # e.g. "LT"
baseline: np.ndarray
current: np.ndarray
def stratified_checks(
records: list[dict],
dimensions: list[str],
baseline_window: int,
current_window: int
) -> Iterator[DataSlice]:
"""Yield slices that differ significantly from baseline."""
for dim in dimensions:
for value in set(r[dim] for r in records):
baseline = np.array([
r["amount"] for r in records
if r[dim] == value and r["hour"] < baseline_window
])
current = np.array([
r["amount"] for r in records
if r[dim] == value and r["hour"] >= current_window
])
result = detect_distribution_shift(baseline, current)
if result["shift_detected"]:
yield DataSlice(dim, value, baseline, current)
這是以簡單性換取覆蓋率。你現在執行的是 N 個統計測試而非一個,這表示你需要關注多重比較修正。簡單的 Bonferroni 調整——將你的門檻除以切片數量——通常就足以控制誤報率。
發現偏移後該怎麼辦
統計測試不會告訴你資料為什麼改變。它只告訴你資料改變了。下一步是根本原因隔離,而最好的工具是差異分析。
你有兩個群體:偏移之前的資料與偏移之後的資料。在你能想到的每一個維度上比較它們。偏移是否集中在特定國家?特定的 API 版本?特定的資料庫 shard?相對差異最大的維度,通常就是 bug 所在的位置。
以下是一個輕量級的差異分析器:
def differential_analysis(
baseline_records: list[dict],
current_records: list[dict],
dimensions: list[str]
) -> list[dict]:
"""Find dimensions where the before/after ratios differ most."""
baseline_total = len(baseline_records)
current_total = len(current_records)
findings = []
for dim in dimensions:
baseline_counts = {}
current_counts = {}
for r in baseline_records:
baseline_counts[r[dim]] = baseline_counts.get(r[dim], 0) + 1
for r in current_records:
current_counts[r[dim]] = current_counts.get(r[dim], 0) + 1
for value in set(baseline_counts) | set(current_counts):
b_rate = baseline_counts.get(value, 0) / baseline_total
c_rate = current_counts.get(value, 0) / current_total
if b_rate > 0:
ratio = c_rate / b_rate
if ratio > 2.0 or ratio < 0.5:
findings.append({
"dimension": dim,
"value": value,
"baseline_rate": b_rate,
"current_rate": c_rate,
"ratio": ratio,
})
return sorted(findings, key=lambda x: abs(1 - x["ratio"]), reverse=True)
如果 api_version: v2.3 顯示零金額交易有 10 倍的飆升,而其他版本都平坦,你就已經將一個生產資料 bug 縮小到特定的部署。這比「某個地方有問題」是一個好得多的起點。
這方法抓不到的東西
統計除錯不能取代單元測試或靜態分析。它只抓特定類型的 bug:以統計異常形式呈現的無聲資料損毀。它不會抓到那些沒有以可測量方式改變資料的 bug。一個總是回傳正確答案但花十秒而非十毫秒的 bug,對分布比較來說是隱形的。一個交換日誌條目中兩個欄位但不影響業務邏輯的 bug 是隱形的。一個產生的錯誤答案與正確答案具有完全相同統計分布的 bug 是隱形的。
它也是本質上反應式的。你比較的是當前資料與歷史資料,這表示 bug 已經發生了。目標是將平均偵測時間從「客戶抱怨時」縮短到「同一個部署週期內」。
從哪裡開始
你不需要資料科學團隊。你只需要一個排程任務與一個告警。
在你的系統中挑選一個關鍵指標。order_total 是個好選擇。exchange_rate 也不錯。計算過去七天的分布作為baseline。每小時對最近一小時的資料執行 KS 檢定。如果測試失敗,就呼叫某人。
前幾週會很吵。你會調整門檻、增加分層維度,並學習哪些偏移是真實的 bug、哪些只是黑色星期五。這些雜訊是校準的代價。一旦校準完成,你就擁有了一個能抓到測試看不見的 bug 的安全網。
如果你想更深入,像 Great Expectations 與 Deequ 這樣的工具將這個模式形式化為可重複使用的資料品質套件。但核心概念只需要五十行 Python,而這五十行會找到你整個測試套件都漏掉的 bug。