diff --git a/docs/biometric-benchmarks.md b/docs/biometric-benchmarks.md new file mode 100644 index 00000000..df9997db --- /dev/null +++ b/docs/biometric-benchmarks.md @@ -0,0 +1,161 @@ +# Biometric Engine — Benchmark Harness & Accuracy Protocol (WP7) + +**Status:** SYNTHETIC-COHORT decision-layer validation. This is a partial answer to +the Onfido/Entrust maturity gap: it validates that the engine's **decision layer** +(similarity math + configured thresholds) behaves correctly and measurably. It does +**not** claim production-grade biometric accuracy — see [LIMITATIONS](#limitations). + +## What is benchmarked + +The biometric engine (`services/biometric-engine/main.py`) turns model scores into +accept/reject decisions at three thresholds: + +| Threshold | Env var | Engine default | Decision governed | +| --- | --- | --- | --- | +| `MATCH_THRESHOLD` | `MATCH_THRESHOLD` | 0.40 | ArcFace cosine similarity accept/reject (`_match_faces`) | +| `LIVENESS_THRESHOLD` | `LIVENESS_THRESHOLD` | 0.72 | Passive/active liveness composite score | +| `ANTISPOOFING_THRESHOLD` | `ANTISPOOFING_THRESHOLD` | 0.60 | MiniFASNetV2 genuine/spoof classifier | + +The harness (`services/biometric-engine/benchmarks/run_benchmarks.py`) benchmarks the +layer where these thresholds live: **cosine similarity on 512-d embeddings and the +threshold decision rule**. + +**No model weights are loaded.** MediaPipe, InsightFace/ONNX Runtime and the +MiniFASNetV2 weights are not exercised: they require network downloads unavailable +offline, and they are upstream of the decision layer being validated. If the engine +module is importable in the environment, the engine's own `_cosine_similarity` is +used; otherwise the identical math (`dot / (|a|·|b|)`, zero-norm → 0.0) is applied +locally and the report records which implementation ran +(`similarity_implementation` field). Thresholds are **read from the engine source / +env vars at run time — never hardcoded** in the harness. + +## Protocol + +1. **Synthetic cohort (deterministic).** `N` identities (default 200, `--identities`) + × `K` samples each (default 10, `--samples`), 512-d L2-normalised embeddings + matching InsightFace `normed_embedding` shape. + - *Inter-class variance:* identity centres are uniform on the unit sphere + (`rng.normal` + normalise), so impostor cosine ≈ N(0, 1/512) (std ≈ 0.044). + - *Intra-class variance:* samples are `normalise(centre + σ_intra · N(0, I))` with + `σ_intra = sqrt((1/target − 1)/dim)` derived from `--target-genuine-cosine` + (default 0.65), so genuine-pair cosine is centred near the target with a spread + smaller than the impostor spread. + - One seeded `numpy.default_rng(seed)` (default seed `20260114`) drives every draw + in fixed order → the cohort is bit-for-bit reproducible. +2. **Pairs.** Genuine: every within-identity pair (C(K,2) per identity → 9,000 pairs + at defaults). Impostor: equal count of cross-identity pairs, sampled without + replacement from an independent seeded RNG. +3. **Metrics.** Genuine/impostor score distributions; FAR/FRR at the engine's live + `MATCH_THRESHOLD`; full ROC sweep over every observed score; EER (linear + interpolation across the FAR=FRR crossing); trapezoidal ROC AUC. +4. **Latency.** `perf_counter` timing of every similarity call → p50/p95/p99 in µs. +5. **Anti-spoofing decision summary.** The MiniFASNetV2 weights are unavailable + offline, so the classifier itself is **not** executed. Seeded Beta(8,2)/Beta(2,8) + score models for genuine/spoof presentations are pushed through the engine's + `ANTISPOOFING_THRESHOLD` decision rule and summarised (mean/std/percentiles/pass + rates). Decision-layer behaviour only. +6. **Outputs.** `benchmarks/report.json` (machine-readable: label, seed, params, + thresholds + provenance, metrics, latency, environment) and a markdown summary on + stdout. + +## How to run + +Fully offline — no network, no dataset downloads, only `numpy` required: + +```bash +cd services/biometric-engine +pip install numpy # the only dependency +python benchmarks/run_benchmarks.py +python -m unittest test_benchmarks -v +``` + +### CI wiring + +`.github/workflows/` exists in this repo. The ready-to-apply job below (validated +as YAML) could not be committed from the automation token used for this change +(`workflow` scope is required to modify GitHub Actions workflow files — the push +was rejected with HTTP 403 `insufficient scopes`). Apply this exact block to +`.github/workflows/ci.yml` (under `jobs:`) with an appropriately scoped token, or +run the command above manually / in any runner: + +```yaml + biometric-benchmarks: + name: biometric-benchmarks + runs-on: ubuntu-24.04 + # Non-blocking: synthetic-cohort decision-layer benchmark; must not gate PRs. + continue-on-error: true + steps: + - uses: actions/checkout@v7 + - uses: actions/setup-python@v6 + with: + python-version: "3.12" + - name: Install benchmark dependencies (numpy only — fully offline afterwards) + run: pip install numpy + - name: Run decision-layer benchmark (SYNTHETIC-COHORT) + working-directory: services/biometric-engine + run: python benchmarks/run_benchmarks.py --out benchmarks/report.json + - name: Run benchmark harness tests + working-directory: services/biometric-engine + run: python -m unittest test_benchmarks -v + - name: Upload benchmark report + if: always() + uses: actions/upload-artifact@v4 + with: + name: biometric-benchmark-report + path: services/biometric-engine/benchmarks/report.json + if-no-files-found: warn +``` + +## Results (real run, seed 20260114, 200 identities × 10 samples, 9,000 genuine / 9,000 impostor pairs) + +Thresholds read from the engine: MATCH_THRESHOLD=0.40, LIVENESS_THRESHOLD=0.72, +ANTISPOOFING_THRESHOLD=0.60. Environment: CPython 3.12, numpy 2.2, Linux x86_64, no +model weights loaded. + +| Metric | Value | +| --- | --- | +| FAR @ MATCH_THRESHOLD=0.40 | 0.000000 | +| FRR @ MATCH_THRESHOLD=0.40 | 0.000000 | +| EER | 0.000000 (crossing threshold 0.5445) | +| ROC AUC | 1.000000 | +| Genuine cosine (mean ± std) | 0.6493 ± 0.0215 | +| Impostor cosine (mean ± std) | −0.0008 ± 0.0438 | +| Similarity latency p50 / p95 / p99 | 8.7 / 13.7 / 20.4 µs per 512-d pair | +| Anti-spoof genuine pass rate @ 0.60 | 0.9314 (synthetic score model) | +| Anti-spoof spoof pass rate @ 0.60 | 0.0030 (synthetic score model) | + +**Reading these numbers honestly:** the synthetic cohort is well separated by +construction (genuine centre 0.65 vs impostor centre ~0.0 with std 0.044), so zero +errors at the operating point are expected and say *nothing* about real-world +accuracy. What they do establish: the threshold logic, ROC/EER computation, env-var +configuration path, and similarity code path are correct, deterministic, and fast +(~9–20 µs per comparison, so similarity is not a latency bottleneck). Degraded +regimes (motion blur, low light, lookalikes) can be explored by lowering +`--target-genuine-cosine` toward the threshold. + +## LIMITATIONS + +- **Synthetic-cohort results validate the decision layer only.** The cohort is + Gaussian noise on a sphere, not faces. Real FAR/FRR are dominated by the ArcFace + model's embedding quality on real imagery (pose, age, lighting, demographic + differentials), which this harness does not measure. +- **No iBeta PAD Level 1/2 certification is claimed or implied.** Presentation-attack + detection certification requires an accredited lab (e.g. iBeta) testing with real + attack-instrument datasets (print, replay, mask, deepfake) against ISO/IEC + 30107-3. That is a **roadmap item**, not claimed here. +- The anti-spoofing section models classifier *outputs* with synthetic Beta + distributions; the MiniFASNetV2 classifier itself is not executed offline, so no + statement is made about real spoof-detection accuracy. +- Liveness (landmark video) is recorded for threshold provenance only; its composite + scoring path is not exercised by embedding-pair benchmarks. +- Production accuracy sign-off against vendor benchmarks (Onfido/Entrust parity) + requires a labelled real-face evaluation set under a data-processing agreement — + also roadmap. + +## Roadmap to close the maturity gap + +1. Accredited iBeta PAD Level 1, then Level 2, evaluation of the liveness + + anti-spoofing stack. +2. Labelled real-face evaluation set (with consent/DPA) for production FAR/FRR at + the operating threshold; rerun this harness's metric code on those scores. +3. Demographic-cohort breakdown of FAR/FRR once a real dataset exists. diff --git a/services/biometric-engine/benchmarks/report.json b/services/biometric-engine/benchmarks/report.json new file mode 100644 index 00000000..4273ef28 --- /dev/null +++ b/services/biometric-engine/benchmarks/report.json @@ -0,0 +1,87 @@ +{ + "antispoofing_decision_summary": { + "genuine": { + "max": 0.998127339170327, + "mean": 0.8004245601245377, + "min": 0.29655156345273065, + "n": 5000, + "p05": 0.5725964643722944, + "p50": 0.8197472295386765, + "p95": 0.9597525297902306, + "pass_rate_at_threshold": 0.9314, + "std": 0.12004768213849419 + }, + "model": "MiniFASNetV2 score model (Beta(8,2) genuine / Beta(2,8) spoof) \u2014 classifier weights NOT loaded", + "note": "Decision-layer behaviour only; production PAD performance requires the real classifier and attack-instrument datasets.", + "spoof": { + "max": 0.7296393264747192, + "mean": 0.20125941429438696, + "min": 0.0014049231654741908, + "n": 5000, + "p05": 0.04088317499015958, + "p50": 0.1806916889584838, + "p95": 0.4319896540370375, + "pass_rate_at_threshold": 0.003, + "std": 0.12093182648610243 + }, + "threshold": 0.6 + }, + "benchmark": "biometric-engine decision-layer (similarity + thresholds)", + "environment": { + "cpu_count": 2, + "model_weights_loaded": false, + "network_required": false, + "numpy": "2.2.5", + "platform": "Linux-5.10.134-18.0.12.lifsea8.x86_64-x86_64-with-glibc2.36", + "python": "3.12.12" + }, + "label": "SYNTHETIC-COHORT", + "latency": { + "max_us": 1034.9660001338634, + "mean_us": 8.7105111125412, + "n_calls": 18000, + "p50_us": 8.489999800076475, + "p95_us": 12.835800202992674, + "p99_us": 20.30808987910858, + "unit": "microseconds per cosine-similarity call (512-d, CPU)" + }, + "liveness_threshold_note": "LIVENESS_THRESHOLD=0.72 governs landmark-video liveness composite scores; embedding-pair FAR/FRR does not exercise it. Recorded here for provenance only.", + "match_metrics": { + "eer": 0.0, + "eer_threshold": 0.5445313831985429, + "operating_point": { + "far": 0.0, + "frr": 0.0, + "genuine_mean": 0.6493054845127151, + "genuine_std": 0.021458777571238246, + "impostor_mean": -0.0007732291753746427, + "impostor_std": 0.043776815492409, + "threshold": 0.4 + }, + "roc_auc": 1.0, + "roc_points": 18002 + }, + "params": { + "cohort_model": "centres uniform on unit sphere (impostor cosine ~ N(0, 1/dim)); samples = normalise(centre + intra_std * N(0, I)) with intra_std = sqrt((1/target_genuine_cosine - 1)/dim), so genuine cosine is centred near target_genuine_cosine", + "embedding_dim": 512, + "impostor_ratio": 1.0, + "intra_std": 0.03242965760392317, + "n_genuine_pairs": 9000, + "n_identities": 200, + "n_impostor_pairs": 9000, + "samples_per_identity": 10, + "target_genuine_cosine": 0.65 + }, + "seed": 20260114, + "similarity_implementation": "local-identical-math (engine import unavailable: ModuleNotFoundError)", + "threshold_provenance": { + "ANTISPOOFING_THRESHOLD": "source:main.py", + "LIVENESS_THRESHOLD": "source:main.py", + "MATCH_THRESHOLD": "source:main.py" + }, + "thresholds": { + "ANTISPOOFING_THRESHOLD": 0.6, + "LIVENESS_THRESHOLD": 0.72, + "MATCH_THRESHOLD": 0.4 + } +} diff --git a/services/biometric-engine/benchmarks/run_benchmarks.py b/services/biometric-engine/benchmarks/run_benchmarks.py new file mode 100644 index 00000000..60101a5a --- /dev/null +++ b/services/biometric-engine/benchmarks/run_benchmarks.py @@ -0,0 +1,483 @@ +#!/usr/bin/env python3 +""" +BIS Biometric Engine — Decision-Layer Benchmark Harness (WP7) + +WHAT THIS BENCHMARKS +-------------------- +This harness validates the *decision layer* of the biometric engine: the +cosine-similarity math and the configured thresholds +(LIVENESS_THRESHOLD, MATCH_THRESHOLD, ANTISPOOFING_THRESHOLD) that turn +model scores into accept/reject decisions. + +It deliberately does NOT load MediaPipe, InsightFace/ONNX Runtime, or the +MiniFASNetV2 anti-spoofing weights: those model files require +network/dataset downloads that are unavailable offline, and they are not +the layer where the thresholds live. If the engine module can be imported +without heavy deps, the engine's own ``_cosine_similarity`` is used; +otherwise the identical math (dot / (|a| * |b|)) is applied locally and the +report records which implementation was used. Thresholds are always read +from the engine source (``services/biometric-engine/main.py``) or the same +environment variables the engine reads — they are never hardcoded here. + +COHORT MODEL (fully documented, fully synthetic) +------------------------------------------------ +Identity class centres are drawn uniformly on the unit 512-sphere +(``rng.normal`` + L2 normalise). For random unit vectors in 512-d the +inter-class cosine similarity has mean ~0 and std ~1/sqrt(512) ~ 0.044. +Within-class samples are ``normalise(centre + sigma_intra * N(0, I))`` where +``sigma_intra`` is derived from ``target_genuine_cosine`` (default 0.65) via +``sigma_intra = sqrt((1/target - 1) / dim)``: in 512-d the noise vector norm +squared is ``sigma_intra^2 * dim = (1/target - 1)``, so the expected genuine +cosine is ``1 / (1 + noise_norm^2) = target``. This places the genuine +distribution on the realistic side of, but close to, the engine's +MATCH_THRESHOLD so both decision tails are exercised. +ALL RESULTS ARE SYNTHETIC-COHORT — see docs/biometric-benchmarks.md. + +OFFLINE GUARANTEE: no network access, no dataset downloads, no model files. + +Usage: + python benchmarks/run_benchmarks.py [--identities 200] [--samples 10] + [--seed 20260114] [--out benchmarks/report.json] +""" + +from __future__ import annotations + +import argparse +import itertools +import json +import os +import platform +import re +import sys +import time +from typing import Any, Optional + +import numpy as np + +# ── Paths ───────────────────────────────────────────────────────────────────── +HERE = os.path.dirname(os.path.abspath(__file__)) +SERVICE_DIR = os.path.dirname(HERE) # services/biometric-engine +ENGINE_MAIN = os.path.join(SERVICE_DIR, "main.py") + +RESULTS_LABEL = "SYNTHETIC-COHORT" + +# Threshold names the engine configures via env (single source of truth: +# main.py). We never restate their defaults here. +THRESHOLD_ENV_VARS = ("LIVENESS_THRESHOLD", "MATCH_THRESHOLD", "ANTISPOOFING_THRESHOLD") + + +# ── Engine config ingestion ─────────────────────────────────────────────────── +def load_engine_thresholds(engine_main_path: str = ENGINE_MAIN) -> dict[str, Any]: + """ + Read the engine's decision thresholds WITHOUT hardcoded duplicates. + + Precedence (identical to the engine): + 1. Environment variables (LIVENESS_THRESHOLD / MATCH_THRESHOLD / + ANTISPOOFING_THRESHOLD) — the engine reads these at import time. + 2. The defaults parsed from the engine source file + (``NAME = float(os.getenv("NAME", ""))``). + + Returns a dict with the thresholds plus provenance for each value. + """ + with open(engine_main_path, "r", encoding="utf-8") as fh: + src = fh.read() + + thresholds: dict[str, float] = {} + provenance: dict[str, str] = {} + for name in THRESHOLD_ENV_VARS: + env_val = os.getenv(name) + if env_val is not None: + thresholds[name] = float(env_val) + provenance[name] = f"env:{name}" + continue + m = re.search( + rf'{name}\s*=\s*float\(\s*os\.getenv\(\s*"{name}"\s*,\s*"([0-9.eE+-]+)"\s*\)\s*\)', + src, + ) + if not m: + raise RuntimeError( + f"Could not locate {name} default in {engine_main_path}; " + "refusing to guess a threshold (no hardcoded duplicates)." + ) + thresholds[name] = float(m.group(1)) + provenance[name] = f"source:{os.path.basename(engine_main_path)}" + return {"thresholds": thresholds, "provenance": provenance} + + +# ── Similarity (decision layer) ─────────────────────────────────────────────── +def _local_cosine_similarity(a: np.ndarray, b: np.ndarray) -> float: + """Identical math to main._cosine_similarity (dot / (|a| * |b|)).""" + if a.shape != b.shape: + raise ValueError("embedding dimensions are inconsistent") + norm_a = float(np.linalg.norm(a)) + norm_b = float(np.linalg.norm(b)) + if norm_a == 0.0 or norm_b == 0.0: + return 0.0 + return float(np.dot(a, b)) / (norm_a * norm_b) + + +def get_cosine_similarity_fn() -> tuple[Any, str]: + """ + Prefer the engine's own ``_cosine_similarity``. Importing main.py pulls in + FastAPI/cv2/redis/etc., which are frequently unavailable offline; in that + case fall back to the byte-for-byte identical local math and record it. + """ + try: + sys.path.insert(0, SERVICE_DIR) + import main as engine_main # noqa: PLC0415 — intentional late import + + return engine_main._cosine_similarity, "engine:main._cosine_similarity" + except Exception as exc: # offline env without FastAPI/cv2/redis — expected + return _local_cosine_similarity, f"local-identical-math (engine import unavailable: {type(exc).__name__})" + + +# ── Synthetic cohort generator ──────────────────────────────────────────────── +def generate_cohort( + n_identities: int = 200, + samples_per_identity: int = 10, + dim: int = 512, + target_genuine_cosine: float = 0.65, + seed: int = 20260114, +) -> np.ndarray: + """ + Deterministic synthetic embedding cohort. + + Shape: (n_identities, samples_per_identity, dim). Every sample is an + L2-normalised dim-vector, matching InsightFace ``normed_embedding``. + + Inter-class variance: identity centres are uniform on the unit sphere, + so impostor cosine ~ N(0, 1/dim) (std ~0.044 for dim=512). + Intra-class variance: additive isotropic Gaussian noise of per-dimension + std ``sigma_intra = sqrt((1/target_genuine_cosine - 1) / dim)`` before + re-normalisation. The total noise energy relative to the unit centre is + ``sigma_intra^2 * dim = 1/target_genuine_cosine - 1``, so the genuine-pair + cosine distribution is centred near ``target_genuine_cosine``. + + Determinism: a single seeded ``numpy.default_rng`` drives every draw in a + fixed order, so the same seed reproduces the cohort bit-for-bit. + """ + if n_identities < 2: + raise ValueError("need at least 2 identities for impostor pairs") + if samples_per_identity < 2: + raise ValueError("need at least 2 samples per identity for genuine pairs") + if not 0.0 < target_genuine_cosine < 1.0: + raise ValueError("target_genuine_cosine must be in (0, 1)") + + intra_std = float(np.sqrt((1.0 / target_genuine_cosine - 1.0) / dim)) + rng = np.random.default_rng(seed) + centres = rng.normal(size=(n_identities, dim)) + centres /= np.linalg.norm(centres, axis=1, keepdims=True) + + cohort = np.empty((n_identities, samples_per_identity, dim), dtype=np.float64) + for i in range(n_identities): + noise = rng.normal(scale=intra_std, size=(samples_per_identity, dim)) + samples = centres[i] + noise + samples /= np.linalg.norm(samples, axis=1, keepdims=True) + cohort[i] = samples + return cohort + + +def derive_pairs( + cohort: np.ndarray, + impostor_ratio: float = 1.0, + seed: int = 20260114, +) -> tuple[list[tuple[int, int]], list[tuple[int, int]]]: + """ + Genuine pairs: every unordered within-identity sample pair (C(K, 2) each). + Impostor pairs: ``impostor_ratio * len(genuine)`` cross-identity pairs + sampled uniformly without replacement from a seeded RNG (independent of + the cohort RNG so pair sampling does not perturb cohort determinism). + Returns index pairs into the flattened (identity * K + sample) axis. + """ + n_id, k, _ = cohort.shape + genuine: list[tuple[int, int]] = [] + for i in range(n_id): + for a, b in itertools.combinations(range(k), 2): + genuine.append((i * k + a, i * k + b)) + + n_impostor = int(round(len(genuine) * impostor_ratio)) + rng = np.random.default_rng(seed + 1) + impostor: list[tuple[int, int]] = [] + seen: set[tuple[int, int]] = set() + while len(impostor) < n_impostor: + i, j = sorted(rng.choice(n_id, size=2, replace=False).tolist()) + a, b = int(rng.integers(k)), int(rng.integers(k)) + key = (i * k + a, j * k + b) + if key in seen: + continue + seen.add(key) + impostor.append(key) + return genuine, impostor + + +# ── Metrics ─────────────────────────────────────────────────────────────────── +def confusion_at_threshold( + genuine_scores: np.ndarray, impostor_scores: np.ndarray, threshold: float +) -> dict[str, float]: + """Accept iff score >= threshold. FAR = impostor accepts; FRR = genuine rejects.""" + far = float(np.mean(impostor_scores >= threshold)) if len(impostor_scores) else 0.0 + frr = float(np.mean(genuine_scores < threshold)) if len(genuine_scores) else 0.0 + return {"threshold": float(threshold), "far": far, "frr": frr} + + +def roc_sweep( + genuine_scores: np.ndarray, impostor_scores: np.ndarray +) -> list[dict[str, float]]: + """ + Proper ROC sweep over every candidate threshold (all observed scores, + plus endpoints above the max and at/below the min so (FAR=0,FRR=1) and + (FAR=1,FRR=0) are both represented). + """ + candidates = np.unique(np.concatenate([genuine_scores, impostor_scores])) + points = [{"threshold": float(np.nextafter(candidates.max(), np.inf)), "far": 0.0, "frr": 1.0}] + for t in candidates[::-1]: + points.append(confusion_at_threshold(genuine_scores, impostor_scores, float(t))) + points.append({"threshold": float(np.nextafter(candidates.min(), -np.inf)), "far": 1.0, "frr": 0.0}) + return points + + +def eer_from_roc(points: list[dict[str, float]]) -> dict[str, float]: + """ + EER = operating point minimising |FAR - FRR| over the ROC sweep, with + linear interpolation between the two sweep points that straddle the + FAR==FRR crossing (falls back to the closest discrete point). + """ + best = min(points, key=lambda p: abs(p["far"] - p["frr"])) + for hi, lo in zip(points, points[1:]): + d_hi = hi["far"] - hi["frr"] + d_lo = lo["far"] - lo["frr"] + if d_hi == 0.0: + return {"eer": hi["far"], "eer_threshold": hi["threshold"]} + if d_hi * d_lo < 0: # sign change -> interpolate + w = abs(d_hi) / (abs(d_hi) + abs(d_lo)) + eer = hi["far"] + w * (lo["far"] - hi["far"]) + thr = hi["threshold"] + w * (lo["threshold"] - hi["threshold"]) + return {"eer": float(eer), "eer_threshold": float(thr)} + return {"eer": float(best["far"]), "eer_threshold": float(best["threshold"])} + + +def auc_from_roc(points: list[dict[str, float]]) -> float: + """Trapezoidal AUC over TPR (= 1 - FRR) vs FAR, sorted by FAR.""" + pts = sorted(points, key=lambda p: p["far"]) + xs = [p["far"] for p in pts] + ys = [1.0 - p["frr"] for p in pts] + return float(np.trapezoid(ys, xs)) + + +def percentile(values: np.ndarray, q: float) -> float: + return float(np.percentile(values, q)) if len(values) else 0.0 + + +# ── Anti-spoofing decision summary ─────────────────────────────────────────── +def antispoofing_decision_summary( + threshold: float, seed: int, n_genuine: int = 5000, n_spoof: int = 5000 +) -> dict[str, Any]: + """ + DECISION-LAYER ONLY: the MiniFASNetV2 weights are not available offline, + so the classifier itself is not executed. Instead we model classifier + output scores with two seeded Beta distributions (genuine presentations + skew high, spoof presentations skew low) and summarise how the engine's + ANTISPOOFING_THRESHOLD decision rule behaves on them. + """ + rng = np.random.default_rng(seed + 2) + genuine_scores = rng.beta(8.0, 2.0, size=n_genuine) # skewed toward 1 + spoof_scores = rng.beta(2.0, 8.0, size=n_spoof) # skewed toward 0 + + def stats(x: np.ndarray) -> dict[str, float]: + return { + "n": int(len(x)), + "mean": float(np.mean(x)), + "std": float(np.std(x)), + "p05": percentile(x, 5), + "p50": percentile(x, 50), + "p95": percentile(x, 95), + "min": float(np.min(x)), + "max": float(np.max(x)), + } + + return { + "model": "MiniFASNetV2 score model (Beta(8,2) genuine / Beta(2,8) spoof) — classifier weights NOT loaded", + "threshold": threshold, + "genuine": {**stats(genuine_scores), "pass_rate_at_threshold": float(np.mean(genuine_scores >= threshold))}, + "spoof": {**stats(spoof_scores), "pass_rate_at_threshold": float(np.mean(spoof_scores >= threshold))}, + "note": "Decision-layer behaviour only; production PAD performance requires the real classifier and attack-instrument datasets.", + } + + +# ── Latency ─────────────────────────────────────────────────────────────────── +def measure_latency( + flat: np.ndarray, pairs: list[tuple[int, int]], sim_fn +) -> dict[str, float]: + """perf_counter timing of the embedding-similarity path (per pair, µs).""" + samples = np.empty(len(pairs), dtype=np.float64) + for idx, (a, b) in enumerate(pairs): + va, vb = flat[a], flat[b] + t0 = time.perf_counter() + sim_fn(va, vb) + samples[idx] = (time.perf_counter() - t0) * 1e6 + return { + "n_calls": int(len(samples)), + "mean_us": float(np.mean(samples)), + "p50_us": percentile(samples, 50), + "p95_us": percentile(samples, 95), + "p99_us": percentile(samples, 99), + "max_us": float(np.max(samples)), + "unit": "microseconds per cosine-similarity call (512-d, CPU)", + } + + +# ── Report assembly ─────────────────────────────────────────────────────────── +def run_benchmark( + n_identities: int = 200, + samples_per_identity: int = 10, + dim: int = 512, + target_genuine_cosine: float = 0.65, + impostor_ratio: float = 1.0, + seed: int = 20260114, +) -> dict[str, Any]: + cfg = load_engine_thresholds() + thresholds = cfg["thresholds"] + sim_fn, sim_impl = get_cosine_similarity_fn() + + cohort = generate_cohort(n_identities, samples_per_identity, dim, target_genuine_cosine, seed) + genuine_pairs, impostor_pairs = derive_pairs(cohort, impostor_ratio, seed) + flat = cohort.reshape(-1, dim) + + genuine_scores = np.array([sim_fn(flat[a], flat[b]) for a, b in genuine_pairs]) + impostor_scores = np.array([sim_fn(flat[a], flat[b]) for a, b in impostor_pairs]) + + match_threshold = thresholds["MATCH_THRESHOLD"] + operating = confusion_at_threshold(genuine_scores, impostor_scores, match_threshold) + roc = roc_sweep(genuine_scores, impostor_scores) + eer = eer_from_roc(roc) + auc = auc_from_roc(roc) + latency = measure_latency(flat, genuine_pairs + impostor_pairs, sim_fn) + + return { + "label": RESULTS_LABEL, + "benchmark": "biometric-engine decision-layer (similarity + thresholds)", + "seed": seed, + "params": { + "n_identities": n_identities, + "samples_per_identity": samples_per_identity, + "embedding_dim": dim, + "target_genuine_cosine": target_genuine_cosine, + "intra_std": float(np.sqrt((1.0 / target_genuine_cosine - 1.0) / dim)), + "impostor_ratio": impostor_ratio, + "n_genuine_pairs": len(genuine_pairs), + "n_impostor_pairs": len(impostor_pairs), + "cohort_model": ( + "centres uniform on unit sphere (impostor cosine ~ N(0, 1/dim)); " + "samples = normalise(centre + intra_std * N(0, I)) with " + "intra_std = sqrt((1/target_genuine_cosine - 1)/dim), so genuine " + "cosine is centred near target_genuine_cosine" + ), + }, + "thresholds": thresholds, + "threshold_provenance": cfg["provenance"], + "similarity_implementation": sim_impl, + "match_metrics": { + "operating_point": { + **operating, + "genuine_mean": float(np.mean(genuine_scores)), + "genuine_std": float(np.std(genuine_scores)), + "impostor_mean": float(np.mean(impostor_scores)), + "impostor_std": float(np.std(impostor_scores)), + }, + "eer": eer["eer"], + "eer_threshold": eer["eer_threshold"], + "roc_auc": auc, + "roc_points": len(roc), + }, + "latency": latency, + "antispoofing_decision_summary": antispoofing_decision_summary( + thresholds["ANTISPOOFING_THRESHOLD"], seed + ), + "liveness_threshold_note": ( + f"LIVENESS_THRESHOLD={thresholds['LIVENESS_THRESHOLD']} governs landmark-video " + "liveness composite scores; embedding-pair FAR/FRR does not exercise it. " + "Recorded here for provenance only." + ), + "environment": { + "python": platform.python_version(), + "platform": platform.platform(), + "numpy": np.__version__, + "cpu_count": os.cpu_count(), + "network_required": False, + "model_weights_loaded": False, + }, + } + + +# ── Markdown rendering ──────────────────────────────────────────────────────── +def render_markdown(report: dict[str, Any]) -> str: + m = report["match_metrics"] + op = m["operating_point"] + lat = report["latency"] + anti = report["antispoofing_decision_summary"] + t = report["thresholds"] + lines = [ + "## Biometric Decision-Layer Benchmark — SYNTHETIC-COHORT", + "", + f"- Seed: `{report['seed']}` | Identities: {report['params']['n_identities']} " + f"x {report['params']['samples_per_identity']} samples " + f"({report['params']['n_genuine_pairs']} genuine / {report['params']['n_impostor_pairs']} impostor pairs)", + f"- Thresholds (from engine): MATCH_THRESHOLD={t['MATCH_THRESHOLD']}, " + f"LIVENESS_THRESHOLD={t['LIVENESS_THRESHOLD']}, ANTISPOOFING_THRESHOLD={t['ANTISPOOFING_THRESHOLD']}", + f"- Similarity implementation: `{report['similarity_implementation']}`", + "", + "| Metric | Value |", + "| --- | --- |", + f"| FAR @ MATCH_THRESHOLD={t['MATCH_THRESHOLD']} | {op['far']:.6f} |", + f"| FRR @ MATCH_THRESHOLD={t['MATCH_THRESHOLD']} | {op['frr']:.6f} |", + f"| EER | {m['eer']:.6f} (threshold {m['eer_threshold']:.4f}) |", + f"| ROC AUC | {m['roc_auc']:.6f} |", + f"| Genuine cosine (mean +/- std) | {op['genuine_mean']:.4f} +/- {op['genuine_std']:.4f} |", + f"| Impostor cosine (mean +/- std) | {op['impostor_mean']:.4f} +/- {op['impostor_std']:.4f} |", + f"| Similarity latency p50 / p95 / p99 (us) | {lat['p50_us']:.1f} / {lat['p95_us']:.1f} / {lat['p99_us']:.1f} |", + f"| Anti-spoof genuine pass rate @ {anti['threshold']} | {anti['genuine']['pass_rate_at_threshold']:.4f} (synthetic scores) |", + f"| Anti-spoof spoof pass rate @ {anti['threshold']} | {anti['spoof']['pass_rate_at_threshold']:.4f} (synthetic scores) |", + "", + "> ALL RESULTS ARE SYNTHETIC-COHORT: they validate the decision layer only. " + "See docs/biometric-benchmarks.md LIMITATIONS.", + ] + return "\n".join(lines) + + +def main(argv: Optional[list[str]] = None) -> int: + parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + parser.add_argument("--identities", type=int, default=200) + parser.add_argument("--samples", type=int, default=10, help="samples per identity") + parser.add_argument("--dim", type=int, default=512) + parser.add_argument("--target-genuine-cosine", type=float, default=0.65, + help="target centre of the genuine-pair cosine distribution") + parser.add_argument("--impostor-ratio", type=float, default=1.0) + parser.add_argument("--seed", type=int, default=20260114) + parser.add_argument("--out", default=os.path.join(HERE, "report.json")) + parser.add_argument("--markdown-out", default=None, help="optional path to also write the markdown summary") + args = parser.parse_args(argv) + + report = run_benchmark( + n_identities=args.identities, + samples_per_identity=args.samples, + dim=args.dim, + target_genuine_cosine=args.target_genuine_cosine, + impostor_ratio=args.impostor_ratio, + seed=args.seed, + ) + + os.makedirs(os.path.dirname(os.path.abspath(args.out)), exist_ok=True) + with open(args.out, "w", encoding="utf-8") as fh: + json.dump(report, fh, indent=2, sort_keys=True) + md = render_markdown(report) + print(md) + print(f"\nReport written to {args.out}") + if args.markdown_out: + with open(args.markdown_out, "w", encoding="utf-8") as fh: + fh.write(md + "\n") + print(f"Markdown summary written to {args.markdown_out}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/services/biometric-engine/test_benchmarks.py b/services/biometric-engine/test_benchmarks.py new file mode 100644 index 00000000..757786bf --- /dev/null +++ b/services/biometric-engine/test_benchmarks.py @@ -0,0 +1,232 @@ +""" +BIS Biometric Engine — Benchmark Harness Tests (WP7) + +Validates the SYNTHETIC-COHORT decision-layer benchmark: + - cohort generator determinism (same seed -> identical cohort) + - pair derivation determinism and correctness + - metric correctness on hand-computed toy confusion sets (FAR/FRR/EER/AUC) + - thresholds are sourced from the engine config (no hardcoded duplicates) + - report.json schema shape + +Requires only numpy (no model weights, no network) — same offline guarantee +as the harness itself. +""" +import json +import os +import re +import sys +import tempfile +import unittest + +import numpy as np + +sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "benchmarks")) +import run_benchmarks as rb # noqa: E402 + +SEED = 20260114 + + +class TestCohortDeterminism(unittest.TestCase): + def test_same_seed_identical_cohort(self): + a = rb.generate_cohort(n_identities=8, samples_per_identity=4, dim=512, seed=SEED) + b = rb.generate_cohort(n_identities=8, samples_per_identity=4, dim=512, seed=SEED) + np.testing.assert_array_equal(a, b) + + def test_different_seed_differs(self): + a = rb.generate_cohort(n_identities=8, samples_per_identity=4, dim=512, seed=SEED) + b = rb.generate_cohort(n_identities=8, samples_per_identity=4, dim=512, seed=SEED + 1) + self.assertFalse(np.array_equal(a, b)) + + def test_embeddings_are_unit_norm(self): + cohort = rb.generate_cohort(n_identities=4, samples_per_identity=3, dim=512, seed=SEED) + norms = np.linalg.norm(cohort, axis=2) + np.testing.assert_allclose(norms, 1.0, atol=1e-12) + + def test_intra_variance_smaller_than_inter_variance(self): + # Genuine-pair cosine spread must be tighter than impostor-pair spread. + cohort = rb.generate_cohort(n_identities=60, samples_per_identity=6, dim=512, + target_genuine_cosine=0.65, seed=SEED) + genuine, impostor = rb.derive_pairs(cohort, impostor_ratio=1.0, seed=SEED) + flat = cohort.reshape(-1, 512) + g = np.array([rb._local_cosine_similarity(flat[a], flat[b]) for a, b in genuine]) + i = np.array([rb._local_cosine_similarity(flat[a], flat[b]) for a, b in impostor]) + self.assertLess(float(np.std(g)), float(np.std(i))) + self.assertGreater(float(np.mean(g)), 0.5) # centred near target 0.65 + self.assertAlmostEqual(float(np.mean(i)), 0.0, delta=0.02) + + def test_pair_derivation_deterministic_and_valid(self): + cohort = rb.generate_cohort(n_identities=6, samples_per_identity=4, dim=64, seed=SEED) + g1, i1 = rb.derive_pairs(cohort, seed=SEED) + g2, i2 = rb.derive_pairs(cohort, seed=SEED) + self.assertEqual(g1, g2) + self.assertEqual(i1, i2) + # Genuine: C(4,2)=6 per identity x 6 identities + self.assertEqual(len(g1), 6 * 6) + self.assertEqual(len(i1), len(g1)) # default impostor_ratio=1.0 + # Genuine pairs are within-identity; impostor pairs cross identities + k = 4 + for a, b in g1: + self.assertEqual(a // k, b // k) + for a, b in i1: + self.assertNotEqual(a // k, b // k) + self.assertEqual(len(set(i1)), len(i1)) # no duplicate impostor pairs + + +class TestToyMetrics(unittest.TestCase): + """Hand-computed FAR/FRR/EER/AUC on tiny confusion sets.""" + + def setUp(self): + # genuine: 0.9, 0.8, 0.7, 0.3 impostor: 0.5, 0.2, 0.15, 0.1 + self.genuine = np.array([0.9, 0.8, 0.7, 0.3]) + self.impostor = np.array([0.5, 0.2, 0.15, 0.1]) + + def test_confusion_at_threshold(self): + # threshold 0.6: impostor accepts = {0.5>=0.6? no} -> FAR=0/4=0.0 + # genuine rejects = {0.3} -> FRR=1/4=0.25 + r = rb.confusion_at_threshold(self.genuine, self.impostor, 0.6) + self.assertAlmostEqual(r["far"], 0.0) + self.assertAlmostEqual(r["frr"], 0.25) + # threshold 0.4: impostor accepts = {0.5} -> FAR=0.25; FRR=0.25 + r = rb.confusion_at_threshold(self.genuine, self.impostor, 0.4) + self.assertAlmostEqual(r["far"], 0.25) + self.assertAlmostEqual(r["frr"], 0.25) + + def test_roc_sweep_endpoints(self): + pts = rb.roc_sweep(self.genuine, self.impostor) + # Endpoint above max score: accept nothing -> FAR=0, FRR=1 + self.assertAlmostEqual(pts[0]["far"], 0.0) + self.assertAlmostEqual(pts[0]["frr"], 1.0) + # Endpoint below min score: accept everything -> FAR=1, FRR=0 + self.assertAlmostEqual(pts[-1]["far"], 1.0) + self.assertAlmostEqual(pts[-1]["frr"], 0.0) + # Every unique score appears as a candidate threshold + self.assertEqual(len(pts), 8 + 2) + + def test_eer_hand_computed(self): + # FAR/FRR by threshold (score >= t accepts): + # t in (0.5, 0.7]: FAR=0, FRR=0.25 -> |d|=0.25 + # t in (0.3, 0.5]: FAR=0.25, FRR=0.25 -> crossing, EER=0.25 + # t in (0.2, 0.3]: FAR=0.25, FRR=0 + pts = rb.roc_sweep(self.genuine, self.impostor) + res = rb.eer_from_roc(pts) + self.assertAlmostEqual(res["eer"], 0.25) + self.assertTrue(0.3 < res["eer_threshold"] <= 0.5) + + def test_eer_perfect_separation_is_zero(self): + genuine = np.array([0.9, 0.8, 0.7]) + impostor = np.array([0.3, 0.2, 0.1]) + res = rb.eer_from_roc(rb.roc_sweep(genuine, impostor)) + self.assertAlmostEqual(res["eer"], 0.0) + self.assertTrue(0.3 < res["eer_threshold"] <= 0.7) + + def test_auc_hand_computed(self): + # Perfect separation -> AUC 1.0 + genuine = np.array([0.9, 0.8, 0.7]) + impostor = np.array([0.3, 0.2, 0.1]) + self.assertAlmostEqual(rb.auc_from_roc(rb.roc_sweep(genuine, impostor)), 1.0) + # Fully overlapping identical distributions -> AUC 0.5 + same = np.array([0.5, 0.5]) + self.assertAlmostEqual(rb.auc_from_roc(rb.roc_sweep(same, same)), 0.5) + + +class TestThresholdSourcing(unittest.TestCase): + def test_thresholds_come_from_engine_source(self): + cfg = rb.load_engine_thresholds() + t = cfg["thresholds"] + self.assertEqual(set(t), {"LIVENESS_THRESHOLD", "MATCH_THRESHOLD", "ANTISPOOFING_THRESHOLD"}) + for name in t: + self.assertIn(name, cfg["provenance"]) + # The parsed value must equal the default literally declared in the + # engine source — proving the harness reads the real config. + with open(rb.ENGINE_MAIN, encoding="utf-8") as fh: + src = fh.read() + m = re.search( + r'MATCH_THRESHOLD\s*=\s*float\(\s*os\.getenv\(\s*"MATCH_THRESHOLD"\s*,\s*"([^"]+)"\s*\)\s*\)', + src, + ) + self.assertIsNotNone(m, "MATCH_THRESHOLD default not found in engine source") + self.assertEqual(float(m.group(1)), t["MATCH_THRESHOLD"]) + + def test_env_override_wins(self): + os.environ["MATCH_THRESHOLD"] = "0.55" + try: + cfg = rb.load_engine_thresholds() + self.assertEqual(cfg["thresholds"]["MATCH_THRESHOLD"], 0.55) + self.assertEqual(cfg["provenance"]["MATCH_THRESHOLD"], "env:MATCH_THRESHOLD") + finally: + del os.environ["MATCH_THRESHOLD"] + + +class TestReportSchema(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.report = rb.run_benchmark( + n_identities=12, samples_per_identity=4, dim=512, + target_genuine_cosine=0.65, impostor_ratio=1.0, seed=SEED, + ) + + def test_top_level_schema(self): + r = self.report + self.assertEqual(r["label"], "SYNTHETIC-COHORT") + for key in ("seed", "params", "thresholds", "threshold_provenance", + "similarity_implementation", "match_metrics", "latency", + "antispoofing_decision_summary", "environment"): + self.assertIn(key, r) + self.assertIsInstance(r["seed"], int) + self.assertFalse(r["environment"]["network_required"]) + self.assertFalse(r["environment"]["model_weights_loaded"]) + + def test_metrics_schema_and_ranges(self): + m = self.report["match_metrics"] + for key in ("operating_point", "eer", "eer_threshold", "roc_auc", "roc_points"): + self.assertIn(key, m) + op = m["operating_point"] + for key in ("threshold", "far", "frr", "genuine_mean", "impostor_mean"): + self.assertIn(key, op) + for v in (op["far"], op["frr"], m["eer"]): + self.assertGreaterEqual(v, 0.0) + self.assertLessEqual(v, 1.0) + self.assertGreaterEqual(m["roc_auc"], 0.0) + self.assertLessEqual(m["roc_auc"], 1.0) + # Threshold used at the operating point is the engine's MATCH_THRESHOLD + self.assertEqual(op["threshold"], self.report["thresholds"]["MATCH_THRESHOLD"]) + + def test_latency_schema(self): + lat = self.report["latency"] + self.assertGreater(lat["n_calls"], 0) + self.assertLessEqual(lat["p50_us"], lat["p95_us"]) + self.assertLessEqual(lat["p95_us"], lat["p99_us"]) + self.assertLessEqual(lat["p99_us"], lat["max_us"]) + + def test_antispoofing_summary_schema(self): + a = self.report["antispoofing_decision_summary"] + for side in ("genuine", "spoof"): + for key in ("n", "mean", "std", "p05", "p50", "p95", "min", "max", + "pass_rate_at_threshold"): + self.assertIn(key, a[side]) + self.assertEqual(a["threshold"], + self.report["thresholds"]["ANTISPOOFING_THRESHOLD"]) + + def test_report_is_json_serializable_and_deterministic(self): + r2 = rb.run_benchmark(n_identities=12, samples_per_identity=4, dim=512, + target_genuine_cosine=0.65, impostor_ratio=1.0, seed=SEED) + # Latency is wall-clock and varies; everything else must be identical. + for r in (self.report, r2): + r["latency"] = None + self.assertEqual(self.report, r2) + json.dumps(self.report) # must not raise + + def test_cli_writes_report_file(self): + with tempfile.TemporaryDirectory() as tmp: + out = os.path.join(tmp, "report.json") + rc = rb.main(["--identities", "8", "--samples", "3", "--seed", str(SEED), + "--out", out]) + self.assertEqual(rc, 0) + with open(out, encoding="utf-8") as fh: + report = json.load(fh) + self.assertEqual(report["label"], "SYNTHETIC-COHORT") + self.assertEqual(report["params"]["n_identities"], 8) + + +if __name__ == "__main__": + unittest.main()