diff --git a/docs/research/NEXT_100_PR_MAP.md b/docs/research/NEXT_100_PR_MAP.md index d690f00d..b5d86e19 100644 --- a/docs/research/NEXT_100_PR_MAP.md +++ b/docs/research/NEXT_100_PR_MAP.md @@ -289,7 +289,7 @@ Track whether the pipeline actually improves across batches by capturing per-bat | ID | Task | Why it matters | Priority | |----|------|----------------|----------| | X1 | Add multi-batch learning record schema (MBL-) (complete). — src/openamp_foundry/evidence/multi_batch_learning_record.py: VALID_MBL_QUALITY_GRADES (5: A-D/N/A), VALID_BATCH_LEARNING_STATUSES (3), GRADE_A_HIT_RATE=0.40; hit_rate auto-computed from n_confirmed_hits/n_candidates_tested; grade auto-derived; no_wet_lab_data forces N/A grade; whr_ids list; dry_lab_only=True; 61 tests in tests/evidence/test_multi_batch_learning_record.py. | Per-batch snapshot of prediction quality (hit rate, AUROC, n_confirmed_hits) after wet-lab feedback; enables cross-batch comparison of whether the pipeline is learning; feeds into calibration improvement tracker. | C | -| X2 | Add calibration improvement tracker schema (CIT-). | Aggregates MBL records across batches; computes hit-rate trend direction (improving/stable/degrading/insufficient_data); minimum 2 batches required; flags when calibration is not producing measurable improvement. | C | +| X2 | Add calibration improvement tracker schema (CIT-) (complete). — src/openamp_foundry/evidence/calibration_improvement_tracker.py: VALID_CIT_TREND_DIRECTIONS (4: improving/stable/degrading/insufficient_data), VALID_CIT_SUMMARY_GRADES (5: A-D/N/A), MIN_BATCHES_FOR_TREND=2, IMPROVEMENT_THRESHOLD=0.05; BatchHitRateEntry helper; trend auto-computed from first→latest hit_rate delta; insufficient_data forces N/A grade; dry_lab_only=True; 49 tests. | Aggregates MBL records across batches; computes hit-rate trend direction (improving/stable/degrading/insufficient_data); minimum 2 batches required; flags when calibration is not producing measurable improvement. | C | | X3 | Add learning progress report schema (LPR-). | Human-readable summary of what the pipeline has learned from all batches to date; references CIT- for trend data; includes which candidate features proved predictive vs not; links to calibration decision logs. | C | | X4 | Add recalibration confidence certificate schema (RCC-). | Asserts with what confidence the current calibration weights are reliable based on cohort size, quality, and consistency across batches; A/B/C/D grade; prevents overconfident calibration claims. | C | | X5 | Add Phase X learning gate (XLG-). | Top-level gate asserting MBL + CIT + LPR + RCC all present; overall verdict: learning_verified/learning_in_progress/learning_not_started; closes Phase X; no calibration improvement claim is credible without passing this gate. | C | diff --git a/src/openamp_foundry/evidence/calibration_improvement_tracker.py b/src/openamp_foundry/evidence/calibration_improvement_tracker.py new file mode 100644 index 00000000..f8159cea --- /dev/null +++ b/src/openamp_foundry/evidence/calibration_improvement_tracker.py @@ -0,0 +1,191 @@ +"""CIT- calibration improvement tracker schema. + +Aggregates MBL- records across batches and computes whether the pipeline's +prediction quality is trending in the right direction. Requires at least 2 +batches with wet-lab data to compute a trend. Flags when calibration is not +producing measurable improvement. +""" + +from __future__ import annotations + +from dataclasses import dataclass + +VALID_CIT_TREND_DIRECTIONS: frozenset[str] = frozenset({ + "improving", + "stable", + "degrading", + "insufficient_data", +}) + +VALID_CIT_SUMMARY_GRADES: frozenset[str] = frozenset({ + "A", + "B", + "C", + "D", + "N/A", +}) + +MIN_BATCHES_FOR_TREND: int = 2 +IMPROVEMENT_THRESHOLD: float = 0.05 +DEGRADATION_THRESHOLD: float = -0.05 + + +@dataclass +class BatchHitRateEntry: + batch_id: str + mbl_id: str + hit_rate: float + quality_grade: str + + +@dataclass +class CalibrationImprovementTracker: + cit_id: str + pipeline_version: str + batch_entries: list[BatchHitRateEntry] + n_batches: int + n_batches_with_data: int + first_batch_hit_rate: float + latest_batch_hit_rate: float + hit_rate_delta: float + trend_direction: str + summary_grade: str + dry_lab_only: bool + limitations: list[str] + created_at: str + + +def validate_calibration_improvement_tracker(cit: CalibrationImprovementTracker) -> None: + if not cit.cit_id.startswith("CIT-"): + raise ValueError(f"cit_id must start with 'CIT-': {cit.cit_id!r}") + if not cit.pipeline_version: + raise ValueError("pipeline_version must be non-empty") + for entry in cit.batch_entries: + if not (0.0 <= entry.hit_rate <= 1.0): + raise ValueError( + f"hit_rate must be in [0, 1] for batch {entry.batch_id!r}: {entry.hit_rate}" + ) + if cit.n_batches != len(cit.batch_entries): + raise ValueError("n_batches must equal len(batch_entries)") + entries_with_data = [e for e in cit.batch_entries if e.quality_grade != "N/A"] + if cit.n_batches_with_data != len(entries_with_data): + raise ValueError("n_batches_with_data mismatch") + if cit.trend_direction not in VALID_CIT_TREND_DIRECTIONS: + raise ValueError( + f"trend_direction {cit.trend_direction!r} not in VALID_CIT_TREND_DIRECTIONS" + ) + if cit.summary_grade not in VALID_CIT_SUMMARY_GRADES: + raise ValueError( + f"summary_grade {cit.summary_grade!r} not in VALID_CIT_SUMMARY_GRADES" + ) + if cit.trend_direction == "insufficient_data" and cit.summary_grade != "N/A": + raise ValueError( + "summary_grade must be 'N/A' when trend_direction='insufficient_data'" + ) + if not cit.dry_lab_only: + raise ValueError("dry_lab_only must be True") + if not cit.limitations: + raise ValueError("limitations must be non-empty") + if not cit.created_at: + raise ValueError("created_at must be non-empty") + + +def _compute_trend( + entries_with_data: list[BatchHitRateEntry], +) -> tuple[float, float, float, str]: + if len(entries_with_data) < MIN_BATCHES_FOR_TREND: + first = entries_with_data[0].hit_rate if entries_with_data else 0.0 + latest = entries_with_data[-1].hit_rate if entries_with_data else 0.0 + return first, latest, 0.0, "insufficient_data" + first = entries_with_data[0].hit_rate + latest = entries_with_data[-1].hit_rate + delta = round(latest - first, 6) + if delta >= IMPROVEMENT_THRESHOLD: + direction = "improving" + elif delta <= DEGRADATION_THRESHOLD: + direction = "degrading" + else: + direction = "stable" + return first, latest, delta, direction + + +def _compute_summary_grade( + trend: str, + latest_hit_rate: float, +) -> str: + if trend == "insufficient_data": + return "N/A" + if trend == "improving" and latest_hit_rate >= 0.25: + return "A" + if trend in ("improving", "stable") and latest_hit_rate >= 0.10: + return "B" + if trend == "stable" or latest_hit_rate >= 0.05: + return "C" + return "D" + + +def build_calibration_improvement_tracker( + *, + cit_id: str, + pipeline_version: str, + batch_entry_dicts: list[dict], + limitations: list[str], + created_at: str, +) -> CalibrationImprovementTracker: + """Build a CalibrationImprovementTracker. + + batch_entry_dicts: list of dicts with keys: + batch_id, mbl_id, hit_rate, quality_grade + """ + entries = [ + BatchHitRateEntry( + batch_id=d["batch_id"], + mbl_id=d["mbl_id"], + hit_rate=float(d["hit_rate"]), + quality_grade=d["quality_grade"], + ) + for d in batch_entry_dicts + ] + entries_with_data = [e for e in entries if e.quality_grade != "N/A"] + first_hr, latest_hr, delta, trend = _compute_trend(entries_with_data) + grade = _compute_summary_grade(trend, latest_hr) + cit = CalibrationImprovementTracker( + cit_id=cit_id, + pipeline_version=pipeline_version, + batch_entries=entries, + n_batches=len(entries), + n_batches_with_data=len(entries_with_data), + first_batch_hit_rate=first_hr, + latest_batch_hit_rate=latest_hr, + hit_rate_delta=delta, + trend_direction=trend, + summary_grade=grade, + dry_lab_only=True, + limitations=limitations, + created_at=created_at, + ) + validate_calibration_improvement_tracker(cit) + return cit + + +def format_calibration_improvement_tracker(cit: CalibrationImprovementTracker) -> str: + lines = [ + f"Calibration Improvement Tracker — {cit.cit_id}", + f"Pipeline: {cit.pipeline_version}", + f"Trend: {cit.trend_direction} | Grade: {cit.summary_grade}", + f"Batches: {cit.n_batches} total, {cit.n_batches_with_data} with data", + f"Hit rate: first={cit.first_batch_hit_rate:.1%} " + f"latest={cit.latest_batch_hit_rate:.1%} " + f"delta={cit.hit_rate_delta:+.1%}", + ] + if cit.batch_entries: + lines.append("Batch history:") + for entry in cit.batch_entries: + lines.append( + f" {entry.batch_id} ({entry.mbl_id}): " + f"hr={entry.hit_rate:.1%} grade={entry.quality_grade}" + ) + lines.append(f"Created: {cit.created_at}") + lines.append(f"Limitations: {'; '.join(cit.limitations)}") + lines.append(f"dry_lab_only: {cit.dry_lab_only}") + return "\n".join(lines) diff --git a/tests/evidence/test_calibration_improvement_tracker.py b/tests/evidence/test_calibration_improvement_tracker.py new file mode 100644 index 00000000..f88da510 --- /dev/null +++ b/tests/evidence/test_calibration_improvement_tracker.py @@ -0,0 +1,315 @@ +"""Tests for CIT- calibration improvement tracker schema.""" + +import pytest +from openamp_foundry.evidence.calibration_improvement_tracker import ( + CalibrationImprovementTracker, + BatchHitRateEntry, + VALID_CIT_TREND_DIRECTIONS, + VALID_CIT_SUMMARY_GRADES, + MIN_BATCHES_FOR_TREND, + IMPROVEMENT_THRESHOLD, + DEGRADATION_THRESHOLD, + build_calibration_improvement_tracker, + format_calibration_improvement_tracker, + validate_calibration_improvement_tracker, +) + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +_TWO_IMPROVING = [ + {"batch_id": "BATCH-01", "mbl_id": "MBL-001", "hit_rate": 0.20, "quality_grade": "C"}, + {"batch_id": "BATCH-02", "mbl_id": "MBL-002", "hit_rate": 0.35, "quality_grade": "B"}, +] + +_TWO_STABLE = [ + {"batch_id": "BATCH-01", "mbl_id": "MBL-001", "hit_rate": 0.20, "quality_grade": "C"}, + {"batch_id": "BATCH-02", "mbl_id": "MBL-002", "hit_rate": 0.22, "quality_grade": "C"}, +] + +_TWO_DEGRADING = [ + {"batch_id": "BATCH-01", "mbl_id": "MBL-001", "hit_rate": 0.30, "quality_grade": "B"}, + {"batch_id": "BATCH-02", "mbl_id": "MBL-002", "hit_rate": 0.10, "quality_grade": "C"}, +] + +_ONE_ENTRY = [ + {"batch_id": "BATCH-01", "mbl_id": "MBL-001", "hit_rate": 0.30, "quality_grade": "B"}, +] + + +def _build(**kwargs): + defaults = dict( + cit_id="CIT-001", + pipeline_version="v1.0", + batch_entry_dicts=_TWO_IMPROVING, + limitations=["dry-lab only"], + created_at="2026-07-10", + ) + defaults.update(kwargs) + return build_calibration_improvement_tracker(**defaults) + + +# --------------------------------------------------------------------------- +# 1. Constants +# --------------------------------------------------------------------------- + + +def test_valid_cit_trend_directions_is_frozenset(): + assert isinstance(VALID_CIT_TREND_DIRECTIONS, frozenset) + + +def test_valid_cit_trend_directions_contains_improving(): + assert "improving" in VALID_CIT_TREND_DIRECTIONS + + +def test_valid_cit_trend_directions_contains_stable(): + assert "stable" in VALID_CIT_TREND_DIRECTIONS + + +def test_valid_cit_trend_directions_contains_degrading(): + assert "degrading" in VALID_CIT_TREND_DIRECTIONS + + +def test_valid_cit_trend_directions_contains_insufficient_data(): + assert "insufficient_data" in VALID_CIT_TREND_DIRECTIONS + + +def test_valid_cit_summary_grades_is_frozenset(): + assert isinstance(VALID_CIT_SUMMARY_GRADES, frozenset) + + +def test_valid_cit_summary_grades_contains_a(): + assert "A" in VALID_CIT_SUMMARY_GRADES + + +def test_valid_cit_summary_grades_contains_na(): + assert "N/A" in VALID_CIT_SUMMARY_GRADES + + +def test_min_batches_for_trend(): + assert MIN_BATCHES_FOR_TREND == 2 + + +def test_improvement_threshold(): + assert IMPROVEMENT_THRESHOLD == 0.05 + + +def test_degradation_threshold(): + assert DEGRADATION_THRESHOLD == -0.05 + + +# --------------------------------------------------------------------------- +# 2. build – happy paths +# --------------------------------------------------------------------------- + + +def test_build_returns_calibration_improvement_tracker(): + assert isinstance(_build(), CalibrationImprovementTracker) + + +def test_build_cit_id_stored(): + assert _build().cit_id == "CIT-001" + + +def test_build_pipeline_version_stored(): + assert _build().pipeline_version == "v1.0" + + +def test_build_dry_lab_only_true(): + assert _build().dry_lab_only is True + + +def test_build_improving_trend(): + r = _build(batch_entry_dicts=_TWO_IMPROVING) + assert r.trend_direction == "improving" + + +def test_build_stable_trend(): + r = _build(batch_entry_dicts=_TWO_STABLE) + assert r.trend_direction == "stable" + + +def test_build_degrading_trend(): + r = _build(batch_entry_dicts=_TWO_DEGRADING) + assert r.trend_direction == "degrading" + + +def test_build_insufficient_data_with_one_entry(): + r = _build(batch_entry_dicts=_ONE_ENTRY) + assert r.trend_direction == "insufficient_data" + + +def test_build_insufficient_data_grade_na(): + r = _build(batch_entry_dicts=_ONE_ENTRY) + assert r.summary_grade == "N/A" + + +def test_build_n_batches_matches_input(): + r = _build(batch_entry_dicts=_TWO_IMPROVING) + assert r.n_batches == 2 + + +def test_build_n_batches_with_data_counts_non_na(): + entries = [ + {"batch_id": "B1", "mbl_id": "M1", "hit_rate": 0.30, "quality_grade": "B"}, + {"batch_id": "B2", "mbl_id": "M2", "hit_rate": 0.0, "quality_grade": "N/A"}, + ] + r = _build(batch_entry_dicts=entries) + assert r.n_batches_with_data == 1 + + +def test_build_first_batch_hit_rate(): + r = _build(batch_entry_dicts=_TWO_IMPROVING) + assert abs(r.first_batch_hit_rate - 0.20) < 1e-4 + + +def test_build_latest_batch_hit_rate(): + r = _build(batch_entry_dicts=_TWO_IMPROVING) + assert abs(r.latest_batch_hit_rate - 0.35) < 1e-4 + + +def test_build_hit_rate_delta_improving(): + r = _build(batch_entry_dicts=_TWO_IMPROVING) + assert abs(r.hit_rate_delta - 0.15) < 1e-4 + + +def test_build_hit_rate_delta_degrading(): + r = _build(batch_entry_dicts=_TWO_DEGRADING) + assert r.hit_rate_delta < 0 + + +def test_build_batch_entries_are_batch_hit_rate_entry(): + for e in _build().batch_entries: + assert isinstance(e, BatchHitRateEntry) + + +def test_build_empty_entries_insufficient_data(): + r = _build(batch_entry_dicts=[]) + assert r.trend_direction == "insufficient_data" + + +def test_build_improving_summary_grade_a(): + entries = [ + {"batch_id": "B1", "mbl_id": "M1", "hit_rate": 0.10, "quality_grade": "C"}, + {"batch_id": "B2", "mbl_id": "M2", "hit_rate": 0.30, "quality_grade": "B"}, + ] + r = _build(batch_entry_dicts=entries) + assert r.summary_grade == "A" + + +def test_build_limitations_stored(): + assert _build().limitations == ["dry-lab only"] + + +def test_build_created_at_stored(): + assert _build().created_at == "2026-07-10" + + +# --------------------------------------------------------------------------- +# 3. validate – rejection cases +# --------------------------------------------------------------------------- + + +def test_validate_rejects_bad_cit_id_prefix(): + with pytest.raises(ValueError, match="CIT-"): + _build(cit_id="BAD-001") + + +def test_validate_rejects_empty_pipeline_version(): + with pytest.raises(ValueError): + _build(pipeline_version="") + + +def test_validate_rejects_invalid_trend_direction(): + cit = _build() + cit.trend_direction = "UNKNOWN" + with pytest.raises(ValueError, match="trend_direction"): + validate_calibration_improvement_tracker(cit) + + +def test_validate_rejects_invalid_summary_grade(): + cit = _build() + cit.summary_grade = "X" + with pytest.raises(ValueError, match="summary_grade"): + validate_calibration_improvement_tracker(cit) + + +def test_validate_rejects_grade_not_na_when_insufficient_data(): + cit = _build(batch_entry_dicts=_ONE_ENTRY) + cit.summary_grade = "A" + with pytest.raises(ValueError, match="N/A"): + validate_calibration_improvement_tracker(cit) + + +def test_validate_rejects_n_batches_mismatch(): + cit = _build() + cit.n_batches = 99 + with pytest.raises(ValueError, match="n_batches"): + validate_calibration_improvement_tracker(cit) + + +def test_validate_rejects_n_batches_with_data_mismatch(): + cit = _build() + cit.n_batches_with_data = 99 + with pytest.raises(ValueError, match="n_batches_with_data"): + validate_calibration_improvement_tracker(cit) + + +def test_validate_rejects_hit_rate_above_one_in_entry(): + entries = [ + {"batch_id": "B1", "mbl_id": "M1", "hit_rate": 1.5, "quality_grade": "A"}, + {"batch_id": "B2", "mbl_id": "M2", "hit_rate": 0.30, "quality_grade": "B"}, + ] + with pytest.raises(ValueError, match="hit_rate"): + _build(batch_entry_dicts=entries) + + +def test_validate_rejects_dry_lab_only_false(): + cit = _build() + cit.dry_lab_only = False + with pytest.raises(ValueError, match="dry_lab_only"): + validate_calibration_improvement_tracker(cit) + + +def test_validate_rejects_empty_limitations(): + with pytest.raises(ValueError, match="limitations"): + _build(limitations=[]) + + +def test_validate_rejects_empty_created_at(): + with pytest.raises(ValueError): + _build(created_at="") + + +# --------------------------------------------------------------------------- +# 4. format +# --------------------------------------------------------------------------- + + +def test_format_contains_cit_id(): + assert "CIT-001" in format_calibration_improvement_tracker(_build()) + + +def test_format_contains_pipeline_version(): + assert "v1.0" in format_calibration_improvement_tracker(_build()) + + +def test_format_contains_trend_direction(): + assert "improving" in format_calibration_improvement_tracker(_build()) + + +def test_format_contains_batch_id(): + assert "BATCH-01" in format_calibration_improvement_tracker(_build()) + + +def test_format_contains_limitations(): + assert "dry-lab only" in format_calibration_improvement_tracker(_build()) + + +def test_format_contains_dry_lab_only(): + assert "dry_lab_only: True" in format_calibration_improvement_tracker(_build()) + + +def test_format_is_string(): + assert isinstance(format_calibration_improvement_tracker(_build()), str)