"""Command-line interface: ``clfkit {train,report,predict,drift,similarity}``.""" from __future__ import annotations import argparse import json import sys import time from pathlib import Path import numpy as np import pandas as pd from . import pipeline from .data import clean_text_columns, load_csv, load_embeddings def _read_new(path: str, bundle) -> tuple[pd.DataFrame, np.ndarray | None]: """Read a batch to score: CSV with the training columns, or a .npy of embeddings.""" if path.endswith(".npy"): E = np.load(path) return pd.DataFrame(E, columns=[f"e{i}" for i in range(E.shape[1])]), None df = pd.read_csv(path) missing = [c for c in bundle.types.all() if c not in df.columns] if missing: raise SystemExit(f"error: columns missing from {path}: {missing}") y = df[bundle.target].astype(str).to_numpy() if bundle.target in df.columns else None return clean_text_columns(df, bundle.types), y def cmd_train(a) -> int: t0 = time.perf_counter() if a.embeddings: X, y, types = load_embeddings(a.embeddings, a.labels) target = "label" else: if not a.data or not a.target: raise SystemExit("error: give DATA.csv and --target (or --embeddings/--labels)") X, y, types = load_csv(a.data, a.target, a.text_col) target = a.target b = pipeline.train(X, y, types, target=target, model=a.model, calibration=a.calibration, test_size=a.test_size, calib_size=a.calib_size, seed=a.seed, compare=not a.no_compare) res = pipeline.evaluate(b, a.alpha, a.max_risk, a.delta) pipeline.save(b, a.out, res) _print_summary(res) print(f"[clfkit] saved to {a.out}/ in {time.perf_counter() - t0:.1f}s") return 0 def _print_summary(res: dict): m = res["test_metrics"] print(f"[clfkit] TEST accuracy={m['accuracy']:.3f} ECE={m['ece']:.3f} NLL={m['nll']:.3f} " f"(calibration: {res['calibration_used']})") for r in res["calibration"]: extra = f" T={r['T']:.2f}" if "T" in r else "" print(f"[clfkit] calib {r['method']:11s} ECE={r['ece']:.3f} NLL={r['nll']:.3f}{extra}") a = res["alpha"] for k, v in res["conformal"].items(): print(f"[clfkit] conformal {k:13s} coverage={v['coverage']:.3f} (target {1 - a:.2f}) " f"avg_size={v['avg_set_size']:.2f} worst_class={v['worst_class_coverage']:.3f}") s = res["selective"] if s["threshold"] is None: print(f"[clfkit] abstain: no threshold certifies risk<={s['target_risk']} (n={s['n_holdout']})") else: print(f"[clfkit] abstain if conf<{s['threshold']:.4f}: certified risk<={s['target_risk']} " f"(UB {s['certified']['risk_upper_bound']:.4f}); test coverage={s['test_coverage']:.3f} " f"test risk={s['test_selective_risk']:.4f}") if "certified" in res: c = res["certified"] print(f"[clfkit] logreg median certified L2 radius={c['median_radius']:.3f} (model feature space)") def cmd_report(a) -> int: b = pipeline.load(a.model_dir) res = pipeline.evaluate(b, a.alpha, a.max_risk, a.delta) drift_path = Path(a.model_dir) / "drift.json" drift_res = json.loads(drift_path.read_text()) if drift_path.exists() else None from .report import write_report out = Path(a.out) if a.out else Path(a.model_dir) / "report.html" write_report(res, b, out, drift_res) print(f"[clfkit] wrote {out} ({out.stat().st_size / 1024:.0f} KB)") return 0 def cmd_predict(a) -> int: b = pipeline.load(a.model_dir) X, y = _read_new(a.data, b) out, info = pipeline.predict(b, X, a.alpha, a.score, a.mondrian, a.max_risk, a.delta) out_path = Path(a.out) if a.out else Path(a.model_dir) / "predictions.csv" out.to_csv(out_path, index=False) lam = info["abstain_threshold"] print(f"[clfkit] {len(out)} rows -> {out_path}") print(f"[clfkit] sets: {a.score}{' (Mondrian)' if a.mondrian else ''} alpha={a.alpha}; " f"avg size {out['set_size'].mean():.2f}; abstain threshold " f"{'NONE (cannot certify -> abstain all)' if lam is None else f'{lam:.4f}'}; " f"abstained {out['abstain'].mean():.1%}") if y is not None: known = np.isin(y, b.classes) yy = y[known] acc = np.mean(out["pred_label"].to_numpy()[known] == yy) cov = np.mean([lab in s.split("|") for lab, s in zip(yy, out["prediction_set"].to_numpy()[known])]) ans = ~out["abstain"].to_numpy()[known] sel_err = 1 - np.mean(out["pred_label"].to_numpy()[known][ans] == yy[ans]) if ans.any() else float("nan") print(f"[clfkit] labels found: accuracy {acc:.3f}, set coverage {cov:.3f}, " f"error on answered {sel_err:.4f}") print(out.head(a.show).to_string(index=False)) return 0 def cmd_drift(a) -> int: b = pipeline.load(a.model_dir) X, _ = _read_new(a.data, b) d = pipeline.drift(b, X, level=a.level, seed=a.seed) out = Path(a.out) if a.out else Path(a.model_dir) / "drift.json" out.write_text(json.dumps(d, indent=2, default=pipeline._json_default)) sm = d["summary"] print(f"[clfkit] drift={'YES' if sm['drift'] else 'no'} alarms={sm['alarms']} -> {out}") print(f"[clfkit] score KS stat={d['scores']['statistic']:.3f} p={d['scores']['p']:.2e} " f"mean conf {d['scores']['mean_conf_ref']:.3f}->{d['scores']['mean_conf_new']:.3f}") print(f"[clfkit] C2ST acc={d['c2st']['accuracy']:.3f} p={d['c2st']['p']:.2e}") for r in d["features"][:5]: print(f"[clfkit] feature {r['feature']:24s} {r['test']:4s} p={r['p']:.2e} p_holm={r['p_holm']:.2e}") ls = d["label_shift"] pairs = ", ".join(f"{c}: {p:.2f}->{q:.2f}" for c, p, q in zip(ls["classes"], ls["p_ref"], ls["q_est"])) print(f"[clfkit] BBSE label proportions (ref->est): {pairs}") return 1 if (a.fail_on_drift and sm["drift"]) else 0 def cmd_similarity(a) -> int: from . import similarity as sim E = np.load(a.embeddings) _, y, _ = load_embeddings(a.embeddings, a.labels) if a.center: E = E - E.mean(0) pairs, rel = sim.sample_pairs(y, a.n_pairs, a.seed) cos = sim.pair_cosines(E, pairs) ns = sim.null_summary(cos[rel == 0], E.shape[1]) print(f"[clfkit] d={ns['dim']}: isotropic null sd={ns['isotropic_sd']:.4f}, " f"P(cos>=0.2 | isotropic)={ns['isotropic_p_at_0.2']:.2e}") print(f"[clfkit] empirical unrelated pairs: mean={ns['empirical_mean']:.3f} sd={ns['empirical_sd']:.3f} " f"q99={ns['empirical_q99']:.3f} frac>=0.2={ns['empirical_frac_above_0.2']:.3f}") r = sim.threshold_for_precision(cos, rel, a.target_precision, a.delta) if r["threshold"] is None: print(f"[clfkit] no threshold certifies precision>={a.target_precision}") else: c = r["certified"] print(f"[clfkit] threshold cos>={c['threshold']:.2f}: precision {c['precision']:.3f} " f"(lower bound {c['precision_lower_bound']:.3f}), recall {c['recall']:.3f}") return 0 def build_parser() -> argparse.ArgumentParser: p = argparse.ArgumentParser(prog="clfkit", description="Trustworthy classification toolkit") sub = p.add_subparsers(dest="cmd", required=True) def guarantees(sp): sp.add_argument("--alpha", type=float, default=0.1, help="conformal miscoverage (0.1 = 90%% sets)") sp.add_argument("--max-risk", type=float, default=0.05, help="target error rate among answered inputs") sp.add_argument("--delta", type=float, default=0.05, help="failure prob. of the abstention guarantee") t = sub.add_parser("train", help="fit model + calibration + conformal; save a model dir") t.add_argument("data", nargs="?", help="CSV file") t.add_argument("--target", help="label column") t.add_argument("--text-col", action="append", help="treat column as free text (repeatable)") t.add_argument("--embeddings", help=".npy embeddings instead of a CSV") t.add_argument("--labels", help="labels for --embeddings (.npy or one per line)") t.add_argument("--out", required=True, help="output model directory") t.add_argument("--model", default="auto", choices=["auto", "logreg", "hgb", "rf"]) t.add_argument("--no-compare", action="store_true", help="skip the model comparison (only allowed with an explicit --model)") t.add_argument("--calibration", default="temperature", choices=["temperature", "isotonic", "platt", "none"]) t.add_argument("--test-size", type=float, default=0.2) t.add_argument("--calib-size", type=float, default=0.2) t.add_argument("--seed", type=int, default=0) guarantees(t) t.set_defaults(func=cmd_train) r = sub.add_parser("report", help="write a self-contained HTML report") r.add_argument("model_dir") r.add_argument("--out", help="HTML path (default MODEL_DIR/report.html)") guarantees(r) r.set_defaults(func=cmd_report) pr = sub.add_parser("predict", help="label, prediction set and abstain flag for new rows") pr.add_argument("model_dir") pr.add_argument("data", help="CSV with the training feature columns (or .npy embeddings)") pr.add_argument("--score", default="aps", choices=["aps", "lac"]) pr.add_argument("--mondrian", action="store_true", help="class-conditional conformal") pr.add_argument("--out", help="output CSV (default MODEL_DIR/predictions.csv)") pr.add_argument("--show", type=int, default=5, help="rows to print") guarantees(pr) pr.set_defaults(func=cmd_predict) d = sub.add_parser("drift", help="compare a new unlabelled batch to the reference data") d.add_argument("model_dir") d.add_argument("data") d.add_argument("--level", type=float, default=0.01, help="overall alarm level") d.add_argument("--out", help="JSON path (default MODEL_DIR/drift.json; picked up by `report`)") d.add_argument("--fail-on-drift", action="store_true", help="exit code 1 if drift detected") d.add_argument("--seed", type=int, default=0) d.set_defaults(func=cmd_drift) s = sub.add_parser("similarity", help="calibrate a cosine-similarity threshold on labelled embeddings") s.add_argument("embeddings") s.add_argument("labels") s.add_argument("--target-precision", type=float, default=0.95) s.add_argument("--delta", type=float, default=0.05) s.add_argument("--n-pairs", type=int, default=6000) s.add_argument("--center", action="store_true", help="mean-centre embeddings first") s.add_argument("--seed", type=int, default=0) s.set_defaults(func=cmd_similarity) return p def main(argv=None) -> int: args = build_parser().parse_args(argv) return args.func(args) if __name__ == "__main__": sys.exit(main())