From fd9a0974bd08795a1e271ed17fd0082502ffa23b Mon Sep 17 00:00:00 2001 From: JeffreyChen Date: Sun, 21 Jun 2026 22:59:39 +0800 Subject: [PATCH] Add distribution drift detection stats had A/B experiment tests but no Population Stability Index and no Kolmogorov-Smirnov two-sample test for the reference-vs-current "is today's data shaped like the baseline" check. Add psi, ks_two_sample, categorical_drift, and a detect_drift wrapper that pair with data_profile. Wired through facade, executor (AC_detect_drift / AC_categorical_drift), MCP, and the Script Builder with a headless test batch and EN/Zh docs. --- README.md | 7 ++ README/README_zh-CN.md | 7 ++ README/README_zh-TW.md | 7 ++ .../Eng/doc/new_features/v82_features_doc.rst | 42 +++++++ docs/source/Eng/eng_index.rst | 1 + .../Zh/doc/new_features/v82_features_doc.rst | 35 ++++++ docs/source/Zh/zh_index.rst | 1 + je_auto_control/__init__.py | 5 + .../gui/script_builder/command_schema.py | 23 ++++ je_auto_control/utils/data_drift/__init__.py | 6 + .../utils/data_drift/data_drift.py | 119 ++++++++++++++++++ .../utils/executor/action_executor.py | 26 ++++ .../utils/mcp_server/tools/_factories.py | 32 ++++- .../utils/mcp_server/tools/_handlers.py | 10 ++ .../headless/test_data_drift_batch.py | 93 ++++++++++++++ 15 files changed, 413 insertions(+), 1 deletion(-) create mode 100644 docs/source/Eng/doc/new_features/v82_features_doc.rst create mode 100644 docs/source/Zh/doc/new_features/v82_features_doc.rst create mode 100644 je_auto_control/utils/data_drift/__init__.py create mode 100644 je_auto_control/utils/data_drift/data_drift.py create mode 100644 test/unit_test/headless/test_data_drift_batch.py diff --git a/README.md b/README.md index 75f6e92c..4ad813d6 100644 --- 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 index 38c8b18e..5c8355bf 100644 --- 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 index f0e82ea9..c2ca650a 100644 --- 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 new file mode 100644 index 00000000..e74a88c3 --- /dev/null +++ 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 index 305de5b0..5542fcb0 100644 --- 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 new file mode 100644 index 00000000..9f06841b --- /dev/null +++ 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 index 400c8f97..10fb2deb 100644 --- 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 index 7e31f7b5..c8bb87a8 100644 --- 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 index 18fff497..f75c9c7a 100644 --- 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 new file mode 100644 index 00000000..acb74a98 --- /dev/null +++ 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 new file mode 100644 index 00000000..963ec392 --- /dev/null +++ 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 index 8ddbf1e5..e74eb49e 100644 --- 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 index 8b4db33d..2c1ef79f 100644 --- a/je_auto_control/utils/mcp_server/tools/_factories.py +++ b/je_auto_control/utils/mcp_server/tools/_factories.py @@ -3343,6 +3343,36 @@ def rate_limit_tools() -> List[MCPTool]: ] +def data_drift_tools() -> List[MCPTool]: + seq_schema = {"type": "array"} + return [ + MCPTool( + name="ac_detect_drift", + description=("Numeric distribution drift of 'current' vs " + "'reference': Population Stability Index (with verdict " + "at 'threshold') plus the KS two-sample test. Returns " + "{psi, drifted, ks}."), + input_schema=schema( + {"reference": seq_schema, "current": seq_schema, + "threshold": {"type": "number"}, "bins": {"type": "integer"}}, + ["reference", "current"]), + handler=h.detect_drift, + annotations=READ_ONLY, + ), + MCPTool( + name="ac_categorical_drift", + description=("Categorical drift of 'current' vs 'reference': " + "chi-square statistic and total-variation distance. " + "Returns {chi_square, total_variation, categories}."), + input_schema=schema( + {"reference": seq_schema, "current": seq_schema}, + ["reference", "current"]), + handler=h.categorical_drift, + annotations=READ_ONLY, + ), + ] + + def layered_config_tools() -> List[MCPTool]: layer_schema = {"type": "array", "items": {"type": "object"}} return [ @@ -5010,7 +5040,7 @@ def media_assert_tools() -> List[MCPTool]: feature_flag_tools, provenance_tools, json_contract_tools, chaos_tools, slo_tools, percentiles_tools, bulkhead_tools, http_cassette_tools, trace_context_tools, data_profile_tools, http_problem_tools, dotenv_tools, - sse_client_tools, layered_config_tools, + sse_client_tools, layered_config_tools, data_drift_tools, saga_tools, decision_table_tools, locator_repair_tools, pii_text_tools, sarif_tools, screen_record_tools, diff --git a/je_auto_control/utils/mcp_server/tools/_handlers.py b/je_auto_control/utils/mcp_server/tools/_handlers.py index 103da836..89ba2fe2 100644 --- a/je_auto_control/utils/mcp_server/tools/_handlers.py +++ b/je_auto_control/utils/mcp_server/tools/_handlers.py @@ -1766,6 +1766,16 @@ def explain_config(layers, key): return _explain_config(layers, key) +def detect_drift(reference, current, threshold=0.25, bins=10): + from je_auto_control.utils.executor.action_executor import _detect_drift + return _detect_drift(reference, current, threshold, bins) + + +def categorical_drift(reference, current): + from je_auto_control.utils.executor.action_executor import _categorical_drift + return _categorical_drift(reference, current) + + def build_provenance(paths, builder_id="je_auto_control"): from je_auto_control.utils.provenance import build_provenance, subject_for subjects = [subject_for(path) for path in paths] diff --git a/test/unit_test/headless/test_data_drift_batch.py b/test/unit_test/headless/test_data_drift_batch.py new file mode 100644 index 00000000..58249f01 --- /dev/null +++ b/test/unit_test/headless/test_data_drift_batch.py @@ -0,0 +1,93 @@ +"""Headless tests for distribution drift detection. Pure stdlib, no Qt.""" +import json + +import pytest + +import je_auto_control as ac +from je_auto_control.utils.data_drift import ( + categorical_drift, detect_drift, ks_two_sample, psi, +) + + +def test_psi_zero_for_identical_distributions(): + sample = [float(value) for value in range(100)] + assert psi(sample, sample) == pytest.approx(0.0, abs=1e-9) + + +def test_psi_grows_when_distribution_shifts(): + reference = [float(value) for value in range(100)] + shifted = [value + 100.0 for value in reference] + assert psi(reference, shifted) > 0.25 # large shift → significant PSI + + +def test_ks_identical_has_zero_statistic(): + sample = [float(value) for value in range(50)] + result = ks_two_sample(sample, sample) + assert result["statistic"] == pytest.approx(0.0) + assert result["p_value"] == pytest.approx(1.0) + + +def test_ks_detects_separation(): + low = [float(value) for value in range(50)] + high = [value + 1000.0 for value in low] + result = ks_two_sample(low, high) + assert result["statistic"] == pytest.approx(1.0) + assert result["p_value"] < 0.05 + + +def test_categorical_drift_summary(): + reference = ["a"] * 8 + ["b"] * 2 + current = ["a"] * 2 + ["b"] * 8 + result = categorical_drift(reference, current) + assert result["categories"] == 2 + assert result["total_variation"] == pytest.approx(0.6) + assert result["chi_square"] > 0 + + +def test_detect_drift_report_and_threshold(): + reference = [float(value) for value in range(100)] + same = list(reference) + assert detect_drift(reference, same)["drifted"] is False + shifted = [value + 100.0 for value in reference] + report = detect_drift(reference, shifted) + assert report["drifted"] is True and report["psi"] > 0.25 + assert "ks" in report + + +def test_empty_inputs_raise(): + with pytest.raises(ValueError): + psi([], [1.0]) + + +# --- wiring --------------------------------------------------------------- + +def test_executor_round_trip(): + reference = list(range(100)) + shifted = [value + 100 for value in reference] + rec = ac.execute_action([[ + "AC_detect_drift", + {"reference": json.dumps(reference), "current": json.dumps(shifted)}]]) + report = next(v for v in rec.values() if isinstance(v, dict)) + assert report["drifted"] is True + rec2 = ac.execute_action([[ + "AC_categorical_drift", + {"reference": json.dumps(["a", "a", "b"]), + "current": json.dumps(["b", "b", "a"])}]]) + cat = next(v for v in rec2.values() if isinstance(v, dict)) + assert cat["categories"] == 2 + + +def test_wiring(): + known = ac.executor.known_commands() + assert {"AC_detect_drift", "AC_categorical_drift"} <= set(known) + from je_auto_control.utils.mcp_server.tools import build_default_tool_registry + names = {t.name for t in build_default_tool_registry()} + assert {"ac_detect_drift", "ac_categorical_drift"} <= names + from je_auto_control.gui.script_builder.command_schema import _build_specs + specs = {s.command for s in _build_specs()} + assert {"AC_detect_drift", "AC_categorical_drift"} <= specs + + +def test_facade_exports(): + for attr in ("psi", "ks_two_sample", "categorical_drift", "detect_drift"): + assert hasattr(ac, attr) and attr in ac.__all__