#!/usr/bin/env python3 """Check Aner's distance-weighted KNN against a 1100-digit Decimal oracle. The fixtures use one-dimensional finite nonnegative training coordinates and an exact-zero query, so each input float is also its exact Euclidean distance. The oracle uses Decimal.from_float, not rounded decimal source literals. Temporary Aner source is deleted after execution. This requires only Python's standard library and a built Aner CLI. """ from __future__ import annotations import argparse import json import math import random import struct import subprocess import sys import tempfile from dataclasses import dataclass from decimal import Decimal, localcontext from pathlib import Path PRECISION = 1100 NORMAL_ULPS = 4 SUBNORMAL_ULPS = 1 @dataclass(frozen=True) class Case: name: str distances: list[float] labels: list[int] k: int subnormal: bool = False def cases() -> list[Case]: result = [ Case("rational_6_11", [1.0, 2.0, 3.0], [0, 1, 1], 3), Case("sparse_classes", [0.5, 1.5, 3.5], [30, 10, 10], 3), Case("zero_duplicates", [0.0, 0.0, 1.0, 2.0], [30, 10, 5, 30], 4), Case("zero_boundary", [0.0, 0.0, 0.0, 1.0], [30, 10, 20, 5], 2), Case("equal_distance_boundary", [1.0, 1.0, 1.0, 2.0], [30, 10, 20, 20], 2), Case("nearest_only", [1e300, 1e-300, 1.0], [30, 10, 20], 1), Case("wide_scales", [1e-300, 1e-200, 1e-100, 1.0, 1e100, 1e200, 1e300], [0, 1, 2, 3, 2, 1, 0], 7), Case("wide_class_sums", [1e-300, 2e-300, 3e-300, 1.0, 2.0, 1e300, 2e300], [0, 1, 1, 2, 2, 3, 3], 7), Case("large_distances", [1e300, 2e300, 3e300, 4e300], [0, 1, 1, 2], 4), Case("small_distances", [1e-300, 2e-300, 3e-300, 4e-300], [0, 1, 1, 2], 4), Case("unselected_classes", [1.0, 2.0, 3.0, 4.0], [0, 1, 2, 3], 2), ] rng = random.Random(20261030) for index in range(16): count = rng.randrange(4, 25) distances = [rng.uniform(0.125, 128.0) for _ in range(count)] labels = [rng.choice([0, 10, 30, 90]) for _ in distances] result.append(Case(f"random_reasonable_{index}", distances, labels, rng.randrange(1, count + 1))) for index in range(8): count = rng.randrange(5, 20) distances = [rng.uniform(1.0, 1.5) * 10.0 ** rng.randrange(-300, 301) for _ in range(count)] labels = [rng.choice([0, 10, 30, 90]) for _ in distances] result.append(Case(f"random_wide_{index}", distances, labels, rng.randrange(2, count + 1))) tiny = math.ulp(0.0) result.extend([ Case("minimum_subnormal", [tiny, 1.0], [0, 1], 2, True), Case("subnormal_final_normalization", [tiny, tiny, 0.7], [0, 1, 2], 3, True), Case("subnormal_class_aggregation", [tiny, tiny, 0.7, 0.7], [0, 1, 2, 2], 4, True), Case("subnormal_neighbors", [tiny, 2 * tiny, 3 * tiny], [0, 1, 1], 3, True), ]) return result def oracle(case: Case) -> tuple[list[int], list[float], int, bool]: classes = sorted(set(case.labels)) selected = sorted(range(len(case.distances)), key=lambda i: (case.distances[i], i))[:case.k] zero_only = case.distances[selected[0]] == 0.0 with localcontext() as context: context.prec = PRECISION scores = {label: Decimal(0) for label in classes} for index in selected: distance = case.distances[index] if zero_only: if distance == 0.0: scores[case.labels[index]] += 1 else: scores[case.labels[index]] += Decimal(1) / Decimal.from_float(distance) total = sum(scores.values(), Decimal(0)) probabilities = [float(scores[label] / total) for label in classes] ranked = sorted(classes, key=lambda label: (-scores[label], label)) winner = ranked[0] margin = scores[winner] - scores[ranked[1]] if len(ranked) > 1 else scores[winner] # Prediction uses computed Float64 vote scores. Assert the real-arithmetic # oracle's label only when rounding cannot plausibly reverse the vote; # exact integer zero-distance votes also have a directly defined tie rule. separated = margin > scores[winner] * Decimal.from_float(128 * sys.float_info.epsilon) check_label = zero_only or len(ranked) == 1 or separated return classes, probabilities, winner, check_label def source_for(fixtures: list[Case]) -> str: lines = ["import aner.ml", "import aner.tensor"] for index, case in enumerate(fixtures): features = "[" + ",".join("[" + repr(value) + "]" for value in case.distances) + "]" labels = "[" + ",".join("[" + repr(float(value)) + "]" for value in case.labels) + "]" lines.append(f'let m{index}=ml.fit_knn({features},{labels},{case.k},"distance")') lines.append(f"let p{index}=m{index}.predict_proba([[0.0]])") lines.append(f'print("case:{index}")') for column in range(len(set(case.labels))): lines.append(f"print(tensor.value(p{index},0,{column}))") lines.append(f"print(tensor.item(m{index}.predict([[0.0]])))") return "\n".join(lines) + "\n" def invoke(aner: Path, path: Path) -> subprocess.CompletedProcess[str]: return subprocess.run([str(aner), "run", str(path)], capture_output=True, text=True, timeout=45, check=False) def ulps(actual: float, expected: float) -> int: # All probabilities are finite and nonnegative, so IEEE binary64 bit patterns # are monotonically ordered. Normalize signed zero before measuring distance. actual_bits = struct.unpack(">Q", struct.pack(">d", actual if actual else 0.0))[0] expected_bits = struct.unpack(">Q", struct.pack(">d", expected if expected else 0.0))[0] return abs(actual_bits - expected_bits) def verify(aner: Path) -> dict: fixtures = cases() skipped: list[dict[str, str]] = [] with tempfile.TemporaryDirectory(prefix="aner-knn-weight-oracle-") as temporary: root = Path(temporary) probe = root / "subnormal.aner" probe.write_text(source_for([fixtures[-4]]), encoding="utf-8") result = invoke(aner, probe) if result.returncode: diagnostic = (result.stderr + result.stdout).strip() # Skip only a source-literal conversion failure. A KNN runtime error # here is a numerical regression and must not be hidden as a skip. if "R2301" in diagnostic or not any(token in diagnostic.lower() for token in ("literal", "range", "representable")): raise RuntimeError("subnormal fixture failed beyond language parsing: " + diagnostic) skipped = [{"case": case.name, "reason": diagnostic} for case in fixtures if case.subnormal] fixtures = [case for case in fixtures if not case.subnormal] program = root / "weights.aner" program.write_text(source_for(fixtures), encoding="utf-8") result = invoke(aner, program) if result.returncode: raise RuntimeError("Aner oracle fixture failed: " + (result.stderr + result.stdout).strip()) lines = iter(result.stdout.splitlines()) records = [] failures = [] maximum = {"normal": 0, "subnormal": 0} checked_labels = 0 probability_count = 0 for index, case in enumerate(fixtures): marker = next(lines, None) if marker != f"case:{index}": raise RuntimeError(f"expected case:{index}, received {marker!r}") classes, expected, expected_label, check_label = oracle(case) observed = [float(next(lines)) for _ in classes] predicted = float(next(lines)) case_maximum = 0 for label, actual, reference in zip(classes, observed, expected): probability_count += 1 category = "normal" if reference >= sys.float_info.min else "subnormal" bound = NORMAL_ULPS if category == "normal" else SUBNORMAL_ULPS if not math.isfinite(actual) or not 0.0 <= actual <= 1.0: failures.append({"case": case.name, "class": label, "actual": actual, "reason": "invalid probability"}) continue difference = ulps(actual, reference) maximum[category] = max(maximum[category], difference) case_maximum = max(case_maximum, difference) if difference > bound: failures.append({"case": case.name, "class": label, "actual": actual, "expected": reference, "actual_hex": actual.hex(), "expected_hex": reference.hex(), "ulps": difference, "bound": bound}) if check_label: checked_labels += 1 if predicted != expected_label: failures.append({"case": case.name, "predicted": predicted, "expected_label": expected_label, "reason": "well-separated or exact-zero vote"}) records.append({"case": case.name, "training_rows": len(case.distances), "k": case.k, "classes": classes, "max_probability_ulps": case_maximum, "label_checked": check_label, "predicted": predicted, "expected_probabilities": expected, "observed_probabilities": observed}) if next(lines, None) is not None: raise RuntimeError("unexpected trailing output from Aner") return {"passed": not failures, "oracle": "Decimal.from_float, 1D nonnegative distances, exact-zero query", "decimal_precision": PRECISION, "normal_ulp_bound": NORMAL_ULPS, "subnormal_ulp_bound": SUBNORMAL_ULPS, "cases_checked": len(fixtures), "probabilities_checked": probability_count, "labels_checked": checked_labels, "max_observed_ulps": maximum, "skipped": skipped, "failures": failures, "cases": records} def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--aner", required=True, type=Path, help="path to the built Aner CLI") parser.add_argument("--output", type=Path, help="write a new JSON report; existing files are refused") arguments = parser.parse_args() try: report = verify(arguments.aner.expanduser().resolve()) encoded = json.dumps(report, indent=2, allow_nan=False) + "\n" if arguments.output: with arguments.output.open("x", encoding="utf-8") as output: output.write(encoded) else: print(encoded, end="") summary = (f"KNN Decimal oracle: {report['cases_checked']} cases, {report['probabilities_checked']} probabilities, " f"max ULPs {report['max_observed_ulps']}, passed={report['passed']}") print(summary, file=sys.stderr) return 0 if report["passed"] else 1 except (OSError, ValueError, RuntimeError, subprocess.TimeoutExpired, StopIteration) as error: print(f"KNN Decimal oracle failed: {error}", file=sys.stderr) return 1 if __name__ == "__main__": raise SystemExit(main())