前処理をテストする ── 壊れたデータを下流へ流さない

import numpy as np

def check_row_count(current: int, history: list[int],                    tolerance: float = 3.0) -> tuple[bool, str]:    """件数が過去の実績から外れていないかを見る。
    history: 直近の日次件数（新しい順でも古い順でもよい）    tolerance: 何倍のばらつきまで許すか    """    if len(history) < 7:        return True, "履歴が不足（判定なし）"
    h = np.array(history, dtype=float)    median = np.median(h)    # 外れ値に強い散らばりの指標を使う    mad = np.median(np.abs(h - median))    spread = mad * 1.4826 or 1.0      # 0除算を避ける
    z = abs(current - median) / spread
    if z > tolerance:        direction = "多すぎ" if current > median else "少なすぎ"        return False, (f"件数が{direction}: {current:,} 件 "                       f"(平常時 {median:,.0f} 件 / 乖離 {z:.1f})")    return True, f"正常: {current:,} 件"

history = [10200, 9800, 10100, 10500, 9900, 10300, 10000]for n in [10150, 5200, 0, 30000]:    ok, msg = check_row_count(n, history)    print(f"  {'OK ' if ok else 'NG '} {msg}")

from dataclasses import dataclass

@dataclassclass CheckResult:    name: str    passed: bool    message: str    severity: str    # "error" は止める / "warn" は通す

def run_checks(df, spec) -> list[CheckResult]:    results = []
    # 1. 必要な列があるか    missing = set(spec["required_columns"]) - set(df.columns)    results.append(CheckResult(        "必須列", not missing,        f"欠けている列: {sorted(missing)}" if missing else "すべて存在",        "error",    ))    if missing:        return results     # 列が無ければ以降は判定できない
    # 2. 欠損率    for col, max_null in spec.get("max_null_rate", {}).items():        rate = df[col].isna().mean()        results.append(CheckResult(            f"欠損率:{col}", rate <= max_null,            f"{rate:.1%}（上限 {max_null:.0%}）", "error",        ))
    # 3. 値の範囲    for col, (lo, hi) in spec.get("ranges", {}).items():        out = df[(df[col] < lo) | (df[col] > hi)][col]        results.append(CheckResult(            f"範囲:{col}", out.empty,            f"範囲外 {len(out)} 件" if len(out) else "すべて範囲内", "error",        ))
    # 4. 想定外の区分値（警告に留める）    for col, allowed in spec.get("allowed_values", {}).items():        unknown = set(df[col].dropna().unique()) - set(allowed)        results.append(CheckResult(            f"区分値:{col}", not unknown,            f"未知の値: {sorted(unknown)}" if unknown else "既知のみ", "warn",        ))
    # 5. 主キーの重複    if key := spec.get("primary_key"):        dup = df.duplicated(subset=key).sum()        results.append(CheckResult(            "主キー重複", dup == 0, f"重複 {dup} 件", "error",        ))
    return results