输入变了，模型就一定变差了吗？

"""Original, deterministic teaching simulation; no learned model or external data."""
import json
import math
from pathlib import Path

BUCKETS = ("below_0", "0_to_2.5", "2.5_to_5", "5_to_7.5", "7.5_to_10", "above_10", "missing")


def bucket(value):
    if value is None:
        return "missing"
    if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value):
        raise ValueError("feature must be finite numeric or None")
    if value < 0:
        return "below_0"
    if value < 2.5:
        return "0_to_2.5"
    if value < 5:
        return "2.5_to_5"
    if value < 7.5:
        return "5_to_7.5"
    if value <= 10:
        return "7.5_to_10"
    return "above_10"


def summarize(values):
    if not values:
        raise ValueError("empty observation window")
    counts = dict.fromkeys(BUCKETS, 0)
    for value in values:
        counts[bucket(value)] += 1
    return {"count": len(values), "counts": counts,
            "proportions": {name: count / len(values) for name, count in counts.items()}}


def total_variation(left, right):
    return sum(abs(left[name] - right[name]) for name in BUCKETS) / 2


def accuracy(values, labels):
    if (not values or len(values) != len(labels)
            or any(type(label) is not int or label not in (0, 1) for label in labels)):
        raise ValueError("aligned binary labels required")
    for value in values:
        bucket(value)  # Enforce the same feature contract as the input-only monitor.
    if any(value is None for value in values):
        raise ValueError("toy predictor has no missing-value rule")
    return sum(int(value >= 5) == label for value, label in zip(values, labels)) / len(values)


def must_reject(action):
    try:
        action()
    except ValueError:
        return
    raise AssertionError("invalid input accepted")


reference = [1] * 25 + [3] * 25 + [7] * 25 + [9] * 25
changed_mix = [1] * 40 + [3] * 40 + [7] * 10 + [9] * 10
changed_rule = list(reference)
windows = {"reference": reference, "changed_mix": changed_mix, "changed_rule": changed_rule}
baseline = summarize(reference)
monitor = {}
for name, values in windows.items():
    summary = summarize(values)
    monitor[name] = {**summary,
                     "input_tv": total_variation(baseline["proportions"], summary["proportions"]),
                     "quality_status": "unknown_without_labels"}

# The monitoring stage above never receives these labels.
labels = {"reference": [int(x >= 5) for x in reference],
          "changed_mix": [int(x >= 5) for x in changed_mix],
          "changed_rule": [int(x < 5) for x in changed_rule]}
delayed_quality = {name: accuracy(values, labels[name]) for name, values in windows.items()}
assert math.isclose(monitor["changed_mix"]["input_tv"], 0.3)
assert monitor["changed_rule"]["input_tv"] == 0
assert delayed_quality == {"reference": 1.0, "changed_mix": 1.0, "changed_rule": 0.0}
assert all(item["quality_status"] == "unknown_without_labels" for item in monitor.values())
assert summarize([None, -1, 11])["counts"]["missing"] == 1
assert [bucket(x) for x in (0, 2.5, 5, 7.5, 10)] == [
    "0_to_2.5", "2.5_to_5", "5_to_7.5", "7.5_to_10", "7.5_to_10"]
must_reject(lambda: summarize([]))
must_reject(lambda: summarize([float("nan")]))
must_reject(lambda: summarize([float("inf")]))
must_reject(lambda: summarize(["3"]))
must_reject(lambda: accuracy([float("nan")], [0]))
must_reject(lambda: accuracy([1], [False]))
must_reject(lambda: accuracy([1], [0.0]))
report = {"scope": "synthetic scalar rule; no training; no production-quality claim",
          "bin_policy": "fixed in advance; left closed, right open except final bin includes 10",
          "monitor_without_labels": monitor, "quality_after_labels_arrive": delayed_quality}
Path("drift-report.json").write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")
for name in windows:
    print(name, "input_tv=", round(monitor[name]["input_tv"], 3), "later_accuracy=", delayed_quality[name])
print("PASS: drift and quality are separate; drift-report.json saved")