Spaces:
Runtime error
Runtime error
| """Search for a rule variant that closes the architectural blindspot | |
| revealed by user's failure case 2026-06-03. | |
| Failure case (real OOD scan): | |
| v9c_p95 = 0.6328 (threshold 0.702 — silent, only 0.07 below) | |
| v8_area = 0 px (threshold 49 — silent, segmenter saw nothing) | |
| symmetry = 130 (threshold 83 — FIRED) | |
| andi_max = 0.000149 (threshold 1.36e-4 — FIRED) | |
| Current production rule: (v9c AND sym) OR (v8 AND andi) | |
| => both branches dead because the firing | |
| pair is (sym AND andi) — the diagonal. | |
| This script tests three families of fixes: | |
| 1. Diagonal-OR — add (sym AND andi) [and optionally (v9c AND andi)] | |
| as additional branches. | |
| 2. 2-of-4 voting — any two of four signals fire => tumor. | |
| 3. Soft-OR / weighted — tolerate a near-miss on v9c when sym+andi agree. | |
| For each rule we report the best operating point that: | |
| (a) keeps recall >= 95% on the 246-sample bench, | |
| (b) catches the user's failure case, | |
| (c) minimises FPR. | |
| """ | |
| from __future__ import annotations | |
| import csv | |
| from collections import defaultdict | |
| from itertools import product | |
| from pathlib import Path | |
| ROOT = Path(__file__).resolve().parent.parent | |
| SAMPLES = ROOT / 'samples' / 'ood' | |
| def load_rows(): | |
| base = list(csv.DictReader((SAMPLES / 'eval_v9c_ensemble_inputs.csv').open(encoding='utf-8'))) | |
| andi = {(r['source'], r['file']): r | |
| for r in csv.DictReader((SAMPLES / 'eval_v9b_andi_results.csv').open(encoding='utf-8'))} | |
| rows = [] | |
| for r in base: | |
| a = andi.get((r['source'], r['file'])) | |
| if not a: | |
| continue | |
| rows.append({ | |
| 'gt': r['gt'], | |
| 'v9c': float(r['v9c_p95']), | |
| 'v8': int(r['v8_area']), | |
| 'sym': float(r['sym_p95']), | |
| 'andi': float(a['max']), | |
| }) | |
| return rows | |
| # The user's actual failure case from their JSON dump | |
| USER_CASE = {'gt': 'tumor', 'v9c': 0.6328, 'v8': 0, 'sym': 130.0, 'andi': 0.000149} | |
| RULES = { | |
| # Current production | |
| 'baseline: (v9c AND sym) OR (v8 AND andi)': | |
| lambda c, v, s, a: (c and s) or (v and a), | |
| # Fix 1: add the missing diagonal pair | |
| 'fix1a: + (sym AND andi)': | |
| lambda c, v, s, a: (c and s) or (v and a) or (s and a), | |
| # Fix 1b: add ALL three missing pairs (full pairwise OR) | |
| 'fix1b: + (sym AND andi) + (v9c AND andi)': | |
| lambda c, v, s, a: (c and s) or (v and a) or (s and a) or (c and a), | |
| 'fix1c: + (sym AND andi) + (v9c AND v8)': | |
| lambda c, v, s, a: (c and s) or (v and a) or (s and a) or (c and v), | |
| 'fix1d: 2-of-{v9c,sym,andi} OR (v8 AND andi)': | |
| lambda c, v, s, a: (int(c)+int(s)+int(a) >= 2) or (v and a), | |
| # Fix 2: any 2 of 4 | |
| 'fix2: 2-of-4 voting': | |
| lambda c, v, s, a: (int(c)+int(v)+int(s)+int(a)) >= 2, | |
| # Stricter 3-of-4 (for completeness — should under-fire) | |
| 'fix2b: 3-of-4 voting': | |
| lambda c, v, s, a: (int(c)+int(v)+int(s)+int(a)) >= 3, | |
| # Soft-OR variants — any single anomaly signal + symmetry confirms | |
| 'fix3a: ((v9c OR andi) AND sym) OR (v8 AND andi)': | |
| lambda c, v, s, a: ((c or a) and s) or (v and a), | |
| 'fix3b: ((v9c OR andi) AND sym) OR (v8 AND (andi OR sym))': | |
| lambda c, v, s, a: ((c or a) and s) or (v and (a or s)), | |
| 'fix3c: any-OR (single signal triggers — most aggressive)': | |
| lambda c, v, s, a: c or v or s or a, | |
| } | |
| def _eval(rule, rows, tc, tv, ts, ta): | |
| TP = FN = FP = TN = 0 | |
| for r in rows: | |
| fires = rule(r['v9c'] > tc, r['v8'] >= tv, r['sym'] > ts, r['andi'] > ta) | |
| if r['gt'] == 'tumor': | |
| TP += fires; FN += not fires | |
| else: | |
| FP += fires; TN += not fires | |
| re = TP / (TP + FN) if TP + FN else 0 | |
| fp = FP / (FP + TN) if FP + TN else 0 | |
| pr = TP / (TP + FP) if TP + FP else 0 | |
| f1 = 2 * pr * re / (pr + re) if pr + re else 0 | |
| return re, fp, pr, f1 | |
| def main(): | |
| rows = load_rows() | |
| print(f'[loaded] {len(rows)} samples ' | |
| f'({sum(1 for r in rows if r["gt"]=="tumor")} tumor / ' | |
| f'{sum(1 for r in rows if r["gt"]=="no_tumor")} healthy)') | |
| print(f'[user case] v9c={USER_CASE["v9c"]} v8={USER_CASE["v8"]} ' | |
| f'sym={USER_CASE["sym"]} andi={USER_CASE["andi"]}') | |
| # Threshold grids — keep the same as production sweep | |
| v9c_g = sorted(set(round(r['v9c'], 3) for r in rows)) | |
| v8_g = [49, 99, 199, 499, 999, 1999, 4999, 9999] | |
| sym_g = sorted(set(round(r['sym'], 1) for r in rows if r['sym'] > 0)) | |
| andi_vals = sorted(r['andi'] for r in rows if r['andi'] > 0) | |
| andi_g = [andi_vals[int(len(andi_vals) * q)] | |
| for q in (0.05, 0.10, 0.25, 0.50, 0.75, 0.90, 0.95, 0.99)] | |
| n_grid = len(v9c_g) * len(v8_g) * len(sym_g) * len(andi_g) | |
| print(f'[grid] {n_grid:,} threshold combos per rule\n') | |
| print('=' * 110) | |
| print(f' {"rule":58s} {"target":12s} {"rec":>4s} {"FPR":>5s} {"prec":>5s} {"F1":>5s} catches?') | |
| print('=' * 110) | |
| overall_best = None | |
| for rname, rule in RULES.items(): | |
| # For each rule, find the best config that meets recall>=95% and | |
| # also catches the user case. | |
| best_with_user = None | |
| best_overall = None | |
| for tc, tv, ts, ta in product(v9c_g, v8_g, sym_g, andi_g): | |
| re, fp, pr, f1 = _eval(rule, rows, tc, tv, ts, ta) | |
| # "Best overall" — best F1 with recall >= 95 | |
| if re >= 0.95: | |
| key_overall = (fp, -f1) | |
| if best_overall is None or key_overall < best_overall[0]: | |
| best_overall = (key_overall, re, fp, pr, f1, tc, tv, ts, ta) | |
| # Does this config catch the user case? | |
| uc = USER_CASE | |
| catches = rule(uc['v9c'] > tc, uc['v8'] >= tv, uc['sym'] > ts, uc['andi'] > ta) | |
| if catches and re >= 0.95: | |
| key_user = (fp, -f1) | |
| if best_with_user is None or key_user < best_with_user[0]: | |
| best_with_user = (key_user, re, fp, pr, f1, tc, tv, ts, ta) | |
| # Display | |
| if best_overall: | |
| _, re, fp, pr, f1, tc, tv, ts, ta = best_overall | |
| catches_str = 'no' | |
| if best_with_user: | |
| catches_str = 'YES (same/diff config)' | |
| print(f' {rname:58s} {"re>=0.95":12s} ' | |
| f'{re:>3.0%} {fp:>4.0%} {pr:>4.0%} {f1:>5.3f} {catches_str}') | |
| if best_with_user and best_with_user is not best_overall: | |
| _, re, fp, pr, f1, tc, tv, ts, ta = best_with_user | |
| print(f' {"":58s} {"+catch":12s} ' | |
| f'{re:>3.0%} {fp:>4.0%} {pr:>4.0%} {f1:>5.3f} ' | |
| f'tc={tc:.3f} tv={tv} ts={ts} ta={ta:.2e}') | |
| # Track overall best across rules — F1-maximising with user-case catch | |
| if best_with_user: | |
| entry = (best_with_user[4], best_with_user[2], rname, best_with_user[5:]) | |
| if overall_best is None or entry > overall_best: | |
| overall_best = entry | |
| print() | |
| print('=' * 110) | |
| print('OPTIMAL RULE (highest F1 while catching the user failure case)') | |
| print('=' * 110) | |
| if overall_best is None: | |
| print(' no rule catches the user case at recall >= 95% — relax constraint') | |
| else: | |
| f1_opt, fp_opt, rname, (tc, tv, ts, ta) = overall_best | |
| re_opt = None | |
| for r in [overall_best]: | |
| re_opt = r | |
| # Re-eval to get re/pr cleanly | |
| rule = RULES[rname] | |
| re, fp, pr, f1 = _eval(rule, rows, tc, tv, ts, ta) | |
| print(f' rule: {rname}') | |
| print(f' thresholds: v9c>{tc} v8>={tv} sym>{ts} andi>{ta:.3e}') | |
| print(f' measured: recall={re:.0%} FPR={fp:.0%} prec={pr:.0%} F1={f1:.3f}') | |
| print(f' catches the user failure case: YES') | |
| # Sanity-check: how much does this differ from baseline? | |
| base_rule = RULES['baseline: (v9c AND sym) OR (v8 AND andi)'] | |
| re_b, fp_b, pr_b, f1_b = _eval(base_rule, rows, 0.702, 49, 83.0, 1.36e-4) | |
| print(f'\n vs baseline (current production): ' | |
| f'recall={re_b:.0%} FPR={fp_b:.0%} F1={f1_b:.3f}') | |
| print(f' delta: recall {re-re_b:+.0%} FPR {fp-fp_b:+.0%} F1 {f1-f1_b:+.3f}') | |
| if __name__ == '__main__': | |
| main() | |