Merge pull request #290 from Integration-Automation/feat/data-drift-batch

JE-Chen · web-flow · commit 29a29207d6ed · 2026-06-21T23:04:41.000+08:00
Add distribution drift detection
diff --git a/README.md b/README.md
@@ -13,6 +13,7 @@
 
 ## Table of Contents
 
+- [What's new (2026-06-21) — Distribution Drift Detection](#whats-new-2026-06-21--distribution-drift-detection)
 - [What's new (2026-06-21) — Layered Configuration Resolver](#whats-new-2026-06-21--layered-configuration-resolver)
 - [What's new (2026-06-21) — Server-Sent Events (SSE) Client Parser](#whats-new-2026-06-21--server-sent-events-sse-client-parser)
 - [What's new (2026-06-21) — Dotenv (.env) Parsing](#whats-new-2026-06-21--dotenv-env-parsing)
@@ -134,6 +135,12 @@
 
 ---
 
+## What's new (2026-06-21) — Distribution Drift Detection
+
+Check whether today's data is shaped like the baseline. Full reference: [`docs/source/Eng/doc/new_features/v82_features_doc.rst`](docs/source/Eng/doc/new_features/v82_features_doc.rst).
+
+- **`psi` / `ks_two_sample` / `categorical_drift` / `detect_drift`** (`AC_detect_drift`, `AC_categorical_drift`): `stats` had A/B experiment tests but no Population Stability Index and no KS two-sample test for reference-vs-current distributions. This adds PSI (quantile-binned log-ratio), the KS statistic with a Kolmogorov p-value, and a categorical chi-square + total-variation summary — pairing with `data_profile`. `detect_drift` gives a one-call `{psi, drifted, ks}` verdict. Pure-stdlib, deterministic.
+
 ## What's new (2026-06-21) — Layered Configuration Resolver
 
 Compose config with `defaults < file < env < CLI` precedence. Full reference: [`docs/source/Eng/doc/new_features/v81_features_doc.rst`](docs/source/Eng/doc/new_features/v81_features_doc.rst).
diff --git a/README/README_zh-CN.md b/README/README_zh-CN.md
@@ -12,6 +12,7 @@
 
 ## 目录
 
+- [本次更新 (2026-06-21) — 分布漂移检测](#本次更新-2026-06-21--分布漂移检测)
 - [本次更新 (2026-06-21) — 分层配置解析器](#本次更新-2026-06-21--分层配置解析器)
 - [本次更新 (2026-06-21) — Server-Sent Events (SSE) 客户端解析器](#本次更新-2026-06-21--server-sent-events-sse-客户端解析器)
 - [本次更新 (2026-06-21) — Dotenv (.env) 解析](#本次更新-2026-06-21--dotenv-env-解析)
@@ -133,6 +134,12 @@
 
 ---
 
+## 本次更新 (2026-06-21) — 分布漂移检测
+
+检查今天的数据形状是否与基准一致。完整参考:[`docs/source/Zh/doc/new_features/v82_features_doc.rst`](../docs/source/Zh/doc/new_features/v82_features_doc.rst)。
+
+- **`psi` / `ks_two_sample` / `categorical_drift` / `detect_drift`**(`AC_detect_drift`、`AC_categorical_drift`):`stats` 有 A/B 实验检定,但没有 Population Stability Index,也没有针对 reference-vs-current 分布的 KS 双样本检定。本功能加入 PSI(分位分箱的 log-ratio)、KS 统计量与 Kolmogorov p 值,以及类别卡方 + total-variation 摘要 —— 与 `data_profile` 搭配。`detect_drift` 给出一次性的 `{psi, drifted, ks}` 判定。纯标准库、确定。
+
 ## 本次更新 (2026-06-21) — 分层配置解析器
 
 以 `defaults < file < env < CLI` 优先级组合配置。完整参考:[`docs/source/Zh/doc/new_features/v81_features_doc.rst`](../docs/source/Zh/doc/new_features/v81_features_doc.rst)。
diff --git a/README/README_zh-TW.md b/README/README_zh-TW.md
@@ -12,6 +12,7 @@
 
 ## 目錄
 
+- [本次更新 (2026-06-21) — 分布漂移偵測](#本次更新-2026-06-21--分布漂移偵測)
 - [本次更新 (2026-06-21) — 分層設定解析器](#本次更新-2026-06-21--分層設定解析器)
 - [本次更新 (2026-06-21) — Server-Sent Events (SSE) 用戶端解析器](#本次更新-2026-06-21--server-sent-events-sse-用戶端解析器)
 - [本次更新 (2026-06-21) — Dotenv (.env) 解析](#本次更新-2026-06-21--dotenv-env-解析)
@@ -133,6 +134,12 @@
 
 ---
 
+## 本次更新 (2026-06-21) — 分布漂移偵測
+
+檢查今天的資料形狀是否與基準一致。完整參考:[`docs/source/Zh/doc/new_features/v82_features_doc.rst`](../docs/source/Zh/doc/new_features/v82_features_doc.rst)。
+
+- **`psi` / `ks_two_sample` / `categorical_drift` / `detect_drift`**(`AC_detect_drift`、`AC_categorical_drift`):`stats` 有 A/B 實驗檢定,但沒有 Population Stability Index,也沒有針對 reference-vs-current 分布的 KS 雙樣本檢定。本功能加入 PSI(分位分箱的 log-ratio)、KS 統計量與 Kolmogorov p 值,以及類別卡方 + total-variation 摘要 —— 與 `data_profile` 搭配。`detect_drift` 給出一次性的 `{psi, drifted, ks}` 判定。純標準函式庫、具決定性。
+
 ## 本次更新 (2026-06-21) — 分層設定解析器
 
 以 `defaults < file < env < CLI` 優先序組合設定。完整參考:[`docs/source/Zh/doc/new_features/v81_features_doc.rst`](../docs/source/Zh/doc/new_features/v81_features_doc.rst)。
diff --git a/docs/source/Eng/doc/new_features/v82_features_doc.rst b/docs/source/Eng/doc/new_features/v82_features_doc.rst
@@ -0,0 +1,42 @@
+Distribution Drift Detection
+============================
+
+``stats`` has two-sample tests for A/B *experiment* outcomes (proportions and
+means), but no Population Stability Index and no Kolmogorov-Smirnov two-sample
+test for the canonical "is today's data shaped like the baseline" check. This
+adds PSI, KS, and a categorical-drift summary that pair with ``data_profile``.
+
+Pure standard library (``math`` / ``bisect`` / ``collections`` + reuse of
+``stats.percentile``); imports no ``PySide6``. Every function is pure
+(sequences in, dict/float out), so it is fully deterministic in CI.
+
+Headless API
+------------
+
+.. code-block:: python
+
+    from je_auto_control import psi, ks_two_sample, categorical_drift, detect_drift
+
+    score = psi(reference, current)               # Population Stability Index
+    ks = ks_two_sample(reference, current)        # {statistic, p_value}
+    report = detect_drift(reference, current)     # {psi, drifted, ks}
+
+    cat = categorical_drift(ref_labels, cur_labels)
+    # {chi_square, total_variation, categories}
+
+``psi`` bins ``current`` against ``reference`` quantile edges and sums the
+log-ratio contribution per bin (0 for identical distributions, growing as they
+diverge). ``ks_two_sample`` returns the maximum empirical-CDF gap and a p-value
+from the Kolmogorov distribution. ``categorical_drift`` compares label
+frequencies via a chi-square statistic and the total-variation distance.
+``detect_drift`` wraps the numeric path into one report with a ``drifted``
+verdict at ``threshold`` (default ``0.25``).
+
+Executor commands
+-----------------
+
+``AC_detect_drift`` takes ``reference`` / ``current`` numeric lists (and
+optional ``threshold`` / ``bins``) and returns ``{psi, drifted, ks}``.
+``AC_categorical_drift`` returns the categorical summary. Both are exposed as
+MCP tools (``ac_detect_drift`` / ``ac_categorical_drift``) and as Script Builder
+commands under **Data**.
diff --git a/docs/source/Eng/eng_index.rst b/docs/source/Eng/eng_index.rst
@@ -104,6 +104,7 @@ Comprehensive guides for all AutoControl features.
    doc/new_features/v79_features_doc
    doc/new_features/v80_features_doc
    doc/new_features/v81_features_doc
+   doc/new_features/v82_features_doc
    doc/ocr_backends/ocr_backends_doc
    doc/observability/observability_doc
    doc/operations_layer/operations_layer_doc
diff --git a/docs/source/Zh/doc/new_features/v82_features_doc.rst b/docs/source/Zh/doc/new_features/v82_features_doc.rst
@@ -0,0 +1,35 @@
+分布漂移偵測
+==========
+
+``stats`` 具備針對 A/B *實驗*結果(比例與平均)的雙樣本檢定,但沒有 Population Stability Index,也沒有
+Kolmogorov-Smirnov 雙樣本檢定來做經典的「今天的資料形狀是否與基準一致」檢查。本功能加入 PSI、KS,以及
+與 ``data_profile`` 搭配的類別漂移摘要。
+
+純標準函式庫(``math`` / ``bisect`` / ``collections`` + 重用 ``stats.percentile``);不匯入 ``PySide6``。
+每個函式皆為純函式(輸入序列、輸出 dict/float),因此在 CI 中完全具決定性。
+
+無頭 API
+--------
+
+.. code-block:: python
+
+    from je_auto_control import psi, ks_two_sample, categorical_drift, detect_drift
+
+    score = psi(reference, current)               # Population Stability Index
+    ks = ks_two_sample(reference, current)        # {statistic, p_value}
+    report = detect_drift(reference, current)     # {psi, drifted, ks}
+
+    cat = categorical_drift(ref_labels, cur_labels)
+    # {chi_square, total_variation, categories}
+
+``psi`` 以 ``reference`` 的分位邊界將 ``current`` 分箱,並加總每箱的 log-ratio 貢獻(分布相同為 0,
+分歧越大值越大)。``ks_two_sample`` 回傳最大經驗 CDF 差距與 Kolmogorov 分布的 p 值。``categorical_drift``
+以卡方統計量與 total-variation 距離比較類別頻率。``detect_drift`` 把數值路徑包成一份報告,並以 ``threshold``
+(預設 ``0.25``)給出 ``drifted`` 判定。
+
+執行器命令
+----------
+
+``AC_detect_drift`` 接受 ``reference`` / ``current`` 數值清單(以及選用的 ``threshold`` / ``bins``)
+回傳 ``{psi, drifted, ks}``。``AC_categorical_drift`` 回傳類別摘要。兩者皆以 MCP 工具
+(``ac_detect_drift`` / ``ac_categorical_drift``)以及 Script Builder 中 **Data** 分類下的命令提供。
diff --git a/docs/source/Zh/zh_index.rst b/docs/source/Zh/zh_index.rst
@@ -104,6 +104,7 @@ AutoControl 所有功能的完整使用指南。
    doc/new_features/v79_features_doc
    doc/new_features/v80_features_doc
    doc/new_features/v81_features_doc
+   doc/new_features/v82_features_doc
    doc/ocr_backends/ocr_backends_doc
    doc/observability/observability_doc
    doc/operations_layer/operations_layer_doc
diff --git a/je_auto_control/__init__.py b/je_auto_control/__init__.py
@@ -159,6 +159,10 @@
 )
 # Data profiling: per-column stats + schema inference from observed rows
 from je_auto_control.utils.data_profile import infer_schema, profile_rows
+# Distribution drift: PSI, KS two-sample, categorical drift
+from je_auto_control.utils.data_drift import (
+    categorical_drift, detect_drift, ks_two_sample, psi,
+)
 # i18n / l10n testing: pseudo-localize, overflow + catalog checks
 from je_auto_control.utils.i18n_test import (
     check_catalog, check_overflow, pseudo_localize, pseudo_localize_catalog,
@@ -823,6 +827,7 @@ def start_autocontrol_gui(*args, **kwargs):
     "merge_results", "shard_flows",
     "extract_fields", "mask_rows", "validate_rows",
     "infer_schema", "profile_rows",
+    "categorical_drift", "detect_drift", "ks_two_sample", "psi",
     "check_catalog", "check_overflow", "pseudo_localize",
     "pseudo_localize_catalog",
     "Checkpoint", "CheckpointStore", "run_resumable",
diff --git a/je_auto_control/gui/script_builder/command_schema.py b/je_auto_control/gui/script_builder/command_schema.py
@@ -1697,6 +1697,29 @@ def _add_resilience_specs(specs: List[CommandSpec]) -> None:
         ),
         description="Show the value and winning layer for a dotted config key.",
     ))
+    specs.append(CommandSpec(
+        "AC_detect_drift", "Data", "Data Drift: Detect (PSI + KS)",
+        fields=(
+            FieldSpec("reference", FieldType.STRING,
+                      placeholder="[1.0, 2.0, 3.0, ...]"),
+            FieldSpec("current", FieldType.STRING,
+                      placeholder="[1.1, 2.2, 9.9, ...]"),
+            FieldSpec("threshold", FieldType.FLOAT, optional=True,
+                      default=0.25),
+            FieldSpec("bins", FieldType.INT, optional=True, default=10),
+        ),
+        description="Numeric drift: Population Stability Index + KS two-sample.",
+    ))
+    specs.append(CommandSpec(
+        "AC_categorical_drift", "Data", "Data Drift: Categorical",
+        fields=(
+            FieldSpec("reference", FieldType.STRING,
+                      placeholder='["a", "b", "a", ...]'),
+            FieldSpec("current", FieldType.STRING,
+                      placeholder='["a", "c", "c", ...]'),
+        ),
+        description="Categorical drift: chi-square + total-variation distance.",
+    ))
     specs.append(CommandSpec(
         "AC_rate_limit", "Flow", "Rate Limit (Token Bucket)",
         fields=(
diff --git a/je_auto_control/utils/data_drift/__init__.py b/je_auto_control/utils/data_drift/__init__.py
@@ -0,0 +1,6 @@
+"""Distribution drift detection for AutoControl data checks."""
+from je_auto_control.utils.data_drift.data_drift import (
+    categorical_drift, detect_drift, ks_two_sample, psi,
+)
+
+__all__ = ["categorical_drift", "detect_drift", "ks_two_sample", "psi"]
diff --git a/je_auto_control/utils/data_drift/data_drift.py b/je_auto_control/utils/data_drift/data_drift.py
@@ -0,0 +1,119 @@
+"""Detect distribution drift between a reference and a current sample.
+
+``stats`` has two-sample tests for A/B *experiment* outcomes (proportions /
+means), but no Population Stability Index and no Kolmogorov-Smirnov two-sample
+test for the canonical "is today's data shaped like the baseline" check. This
+adds PSI, KS, and a categorical-drift summary that pair with ``data_profile``.
+
+Pure standard library (``math`` / ``bisect`` / ``collections`` + reuse of
+``stats.percentile``); imports no ``PySide6``. Every function is pure
+(sequences in, dict/float out) so it is fully deterministic in CI.
+"""
+import bisect
+import math
+from collections import Counter
+from typing import Any, Dict, List, Sequence
+
+from je_auto_control.utils.stats.stats import percentile
+
+_EPSILON = 1e-6
+
+
+def _require(reference: Sequence[Any], current: Sequence[Any]) -> None:
+    if not reference or not current:
+        raise ValueError("reference and current must both be non-empty")
+
+
+def _bin_edges(reference: Sequence[float], bins: int) -> List[float]:
+    return [percentile(reference, index * 100.0 / bins)
+            for index in range(1, bins)]
+
+
+def _bucket_fractions(values: Sequence[float],
+                      edges: Sequence[float]) -> List[float]:
+    counts = [0] * (len(edges) + 1)
+    for value in values:
+        counts[bisect.bisect_right(edges, value)] += 1
+    total = len(values) or 1
+    return [count / total for count in counts]
+
+
+def psi(reference: Sequence[float], current: Sequence[float],
+        bins: int = 10) -> float:
+    """Population Stability Index of ``current`` against ``reference`` bins."""
+    _require(reference, current)
+    edges = _bin_edges([float(value) for value in reference], bins)
+    ref_fractions = _bucket_fractions([float(v) for v in reference], edges)
+    cur_fractions = _bucket_fractions([float(v) for v in current], edges)
+    score = 0.0
+    for ref_part, cur_part in zip(ref_fractions, cur_fractions):
+        ref_part = max(ref_part, _EPSILON)
+        cur_part = max(cur_part, _EPSILON)
+        score += (cur_part - ref_part) * math.log(cur_part / ref_part)
+    return score
+
+
+def _ks_statistic(reference: Sequence[float],
+                  current: Sequence[float]) -> float:
+    ref_sorted = sorted(float(value) for value in reference)
+    cur_sorted = sorted(float(value) for value in current)
+    n_ref, n_cur = len(ref_sorted), len(cur_sorted)
+    statistic = 0.0
+    for value in sorted(set(ref_sorted) | set(cur_sorted)):
+        cdf_ref = bisect.bisect_right(ref_sorted, value) / n_ref
+        cdf_cur = bisect.bisect_right(cur_sorted, value) / n_cur
+        statistic = max(statistic, abs(cdf_ref - cdf_cur))
+    return statistic
+
+
+def _kolmogorov(lam: float) -> float:
+    if lam <= 0:
+        return 1.0
+    total = 0.0
+    for term_index in range(1, 101):
+        term = 2 * (-1) ** (term_index - 1) * math.exp(
+            -2 * term_index * term_index * lam * lam)
+        total += term
+        if abs(term) < 1e-10:
+            break
+    return min(1.0, max(0.0, total))
+
+
+def ks_two_sample(reference: Sequence[float],
+                  current: Sequence[float]) -> Dict[str, float]:
+    """Two-sample Kolmogorov-Smirnov test: ``{statistic, p_value}``."""
+    _require(reference, current)
+    statistic = _ks_statistic(reference, current)
+    n_ref, n_cur = len(reference), len(current)
+    en = math.sqrt(n_ref * n_cur / (n_ref + n_cur))
+    p_value = _kolmogorov((en + 0.12 + 0.11 / en) * statistic)
+    return {"statistic": statistic, "p_value": p_value}
+
+
+def categorical_drift(reference: Sequence[Any],
+                      current: Sequence[Any]) -> Dict[str, float]:
+    """Chi-square statistic and total-variation distance over categories."""
+    _require(reference, current)
+    ref_counts, cur_counts = Counter(reference), Counter(current)
+    ref_total = sum(ref_counts.values())
+    cur_total = sum(cur_counts.values())
+    categories = set(ref_counts) | set(cur_counts)
+    chi_square = 0.0
+    total_variation = 0.0
+    for category in categories:
+        ref_part = ref_counts.get(category, 0) / ref_total
+        cur_part = cur_counts.get(category, 0) / cur_total
+        expected = ref_part * cur_total
+        if expected > 0:
+            chi_square += (cur_counts.get(category, 0) - expected) ** 2 / expected
+        total_variation += abs(ref_part - cur_part)
+    return {"chi_square": chi_square, "total_variation": 0.5 * total_variation,
+            "categories": len(categories)}
+
+
+def detect_drift(reference: Sequence[float], current: Sequence[float], *,
+                 threshold: float = 0.25, bins: int = 10) -> Dict[str, Any]:
+    """Numeric drift report: PSI (with verdict) plus the KS test."""
+    score = psi(reference, current, bins)
+    return {"psi": score, "drifted": score >= threshold,
+            "ks": ks_two_sample(reference, current)}
diff --git a/je_auto_control/utils/executor/action_executor.py b/je_auto_control/utils/executor/action_executor.py
@@ -3075,6 +3075,30 @@ def _explain_config(layers: Any, key: str) -> Dict[str, Any]:
                       "layer": trace.layer}}
 
 
+def _detect_drift(reference: Any, current: Any,
+                  threshold: Any = 0.25, bins: Any = 10) -> Dict[str, Any]:
+    """Adapter: numeric distribution drift report (PSI + KS)."""
+    import json
+    from je_auto_control.utils.data_drift import detect_drift
+    if isinstance(reference, str):
+        reference = json.loads(reference)
+    if isinstance(current, str):
+        current = json.loads(current)
+    return detect_drift(reference, current,
+                        threshold=float(threshold), bins=int(bins))
+
+
+def _categorical_drift(reference: Any, current: Any) -> Dict[str, Any]:
+    """Adapter: categorical distribution drift summary."""
+    import json
+    from je_auto_control.utils.data_drift import categorical_drift
+    if isinstance(reference, str):
+        reference = json.loads(reference)
+    if isinstance(current, str):
+        current = json.loads(current)
+    return categorical_drift(reference, current)
+
+
 def _percentiles(samples: Any, qs: Any = None) -> Dict[str, Any]:
     """Adapter: exact percentiles of a numeric sample list (or JSON string)."""
     import json
@@ -4161,6 +4185,8 @@ def __init__(self):
             "AC_parse_sse": _parse_sse,
             "AC_resolve_config": _resolve_config,
             "AC_explain_config": _explain_config,
+            "AC_detect_drift": _detect_drift,
+            "AC_categorical_drift": _categorical_drift,
             "AC_unified_diff": _unified_diff,
             "AC_apply_unified": _apply_unified,
             "AC_three_way_merge": _three_way_merge,
diff --git a/je_auto_control/utils/mcp_server/tools/_factories.py b/je_auto_control/utils/mcp_server/tools/_factories.py
diff --git a/je_auto_control/utils/mcp_server/tools/_handlers.py b/je_auto_control/utils/mcp_server/tools/_handlers.py
diff --git a/test/unit_test/headless/test_data_drift_batch.py b/test/unit_test/headless/test_data_drift_batch.py
