"""Compare Jev Noul vs Choice and ask Jev to select a validation-set cutoff. Run with: python choice_threshold_benchmark.py --env-file /path/to/.env.local The test split is fresh (seed 18) and is never included in threshold selection. """ import argparse from concurrent.futures import ThreadPoolExecutor import json from pathlib import Path import time from urllib.error import HTTPError, URLError from urllib.request import Request, urlopen import numpy as np from sklearn.metrics import accuracy_score, confusion_matrix, roc_auc_score from sklearn.model_selection import train_test_split from benchmark import API_URL, datasets, read_api_key, row_dict SEED = 18 EXAMPLES_PER_CLASS = 8 CUTOFFS = [0.05, 0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 0.95] def ask(payload, api_key): request = Request( API_URL, data=json.dumps(payload).encode(), headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}, method="POST", ) for attempt in range(5): try: with urlopen(request, timeout=60) as response: return json.load(response) except HTTPError as exc: if exc.code not in (429, 500, 502, 503, 504) or attempt == 4: raise RuntimeError(f"Jev API returned HTTP {exc.code}") from exc except URLError: if attempt == 4: raise time.sleep(2 ** attempt) raise AssertionError("unreachable") def evaluate_row(row, names, examples, positive, negative, api_key): payload = { "model": "jev-latest", "state": { "task": f"Classify the sample as {positive} (positive) or {negative} (negative).", "labeled_training_examples": examples, "sample": row_dict(row, names), }, "questions": { "noul": { "type": "noul", "instructions": ( f"Based on the labeled examples and the sample's numeric features, " f"is the sample in the positive class ({positive}) rather than " f"the negative class ({negative})?" ), }, "choice": { "type": "choice", "instructions": ( "Based on the labeled examples and numeric features, " "which of the two classes best matches the sample?" ), "criteria": {"positive": positive, "negative": negative}, }, }, } response = ask(payload, api_key) return { "noul": float(response["answers"]["noul"]["noul"]), "choice": float(response["answers"]["choice"]["probabilities"]["positive"]), } def metrics(y, probabilities, cutoff): predictions = (np.asarray(probabilities) >= cutoff).astype(int) tn, fp, fn, tp = confusion_matrix(y, predictions, labels=[0, 1]).ravel() return { "accuracy": round(float(accuracy_score(y, predictions)), 4), "auc": round(float(roc_auc_score(y, probabilities)), 4), "sensitivity": round(float(tp / (tp + fn)), 4), "specificity": round(float(tn / (tn + fp)), 4), "confusion_matrix": [[int(tn), int(fp)], [int(fn), int(tp)]], } def code_best(summary): return max( CUTOFFS, key=lambda t: ( summary[str(t)]["accuracy"], summary[str(t)]["sensitivity"], -abs(t - 0.5), ), ) def jev_choose_cutoff(summary, api_key): keys = {f"cutoff_{str(t).replace('.', '_')}": t for t in CUTOFFS} response = ask({ "model": "jev-latest", "state": { "objective": ( "Choose the candidate cutoff with the highest validation accuracy. " "Break an accuracy tie using higher validation sensitivity, then " "closeness to 0.5. This is a benchmark, not a clinical decision." ), "validation_results": [ {"option": key, "cutoff": t, **summary[str(t)]} for key, t in keys.items() ], }, "questions": { "cutoff": { "type": "choice", "instructions": ( "Which cutoff best meets the stated validation objective? " "Choose only from the listed candidates." ), "criteria": { key: f"Use {t:.2f} as the positive-class probability cutoff" for key, t in keys.items() }, } }, }, api_key) answer = response["answers"]["cutoff"]["choice"] return keys[answer], response.get("model") def run(api_key): report = {} for name, X, y, feature_names, positive, negative in datasets(): X_dev, X_test, y_dev, y_test = train_test_split( X, y, test_size=0.25, stratify=y, random_state=SEED ) X_train, X_val, y_train, y_val = train_test_split( X_dev, y_dev, test_size=0.25, stratify=y_dev, random_state=SEED ) rng = np.random.default_rng(SEED) chosen = np.concatenate([ rng.choice(np.flatnonzero(y_train == label), EXAMPLES_PER_CLASS, replace=False) for label in (0, 1) ]) examples = [ {"features": row_dict(X_train[i], feature_names), "label": "positive" if y_train[i] else "negative"} for i in chosen ] print(f"Evaluating {name}: {len(X_val)} validation + {len(X_test)} test rows", flush=True) with ThreadPoolExecutor(max_workers=5) as executor: predictions = list(executor.map( lambda row: evaluate_row(row, feature_names, examples, positive, negative, api_key), np.concatenate([X_val, X_test]), )) val_rows = predictions[:len(X_val)] test_rows = predictions[len(X_val):] choice_val = [p["choice"] for p in val_rows] choice_test = [p["choice"] for p in test_rows] noul_test = [p["noul"] for p in test_rows] validation = {str(t): metrics(y_val, choice_val, t) for t in CUTOFFS} jev_cutoff, model = jev_choose_cutoff(validation, api_key) best_cutoff = code_best(validation) report[name] = { "positive": positive, "negative": negative, "n_train": len(X_train), "n_val": len(X_val), "n_test": len(X_test), "n_examples_for_jev": len(examples), "response_model": model, "cutoffs": {"jev_selected": jev_cutoff, "code_best": best_cutoff}, "validation_by_cutoff": validation, "test_metrics": { "noul_0_5": metrics(y_test, noul_test, 0.5), "choice_0_5": metrics(y_test, choice_test, 0.5), "choice_jev_cutoff": metrics(y_test, choice_test, jev_cutoff), "choice_code_cutoff": metrics(y_test, choice_test, best_cutoff), }, "test_rows": [ {"true_label": int(label), **prediction} for label, prediction in zip(y_test, test_rows) ], } print(f"Completed {name}: Jev cutoff {jev_cutoff}, code cutoff {best_cutoff}", flush=True) return report def main(): parser = argparse.ArgumentParser() parser.add_argument("--env-file", type=Path, required=True) parser.add_argument("--out", type=Path, default=Path("choice_threshold_results.json")) args = parser.parse_args() api_key = read_api_key(args.env_file) if not api_key: raise SystemExit("No Jev API key found") report = run(api_key) args.out.write_text(json.dumps(report, indent=2) + "\n") for name, item in report.items(): print(name, "cutoffs", item["cutoffs"], "test", item["test_metrics"]) if __name__ == "__main__": main()