diff --git a/CHANGELOG.md b/CHANGELOG.md index 1cd51a7..2350e18 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,18 @@ between published tags; older pre-5.0 notes are retained below. For current interfaces, see the [consumer guide](docs/consumer-guide.md). +## 5.93.0 + +- Report policy coverage separately from score filling and acceptance, with + explicit occurrence and model-request denominators, missing/failure reasons, + and distinct raw-row counts that do not multiply allele-free projections (#457). +- Compare complete policy membership and raw scores on the shared assessable + set; retain unknown criteria, effective scores, definitions and provenance. + Typed CSV/TSV preserves reports and their replay inputs. +- Exercise the released Vaxrank `openvax-v1` bundle with a synthetic context + overlay: missing short-flank evidence changes the chosen vaccine window while + shared assessable scores remain unchanged. No default recipe changes. + ## 5.92.0 - Add all-candidate self matching within an explicit same-length Hamming radius diff --git a/docs/combined-sources.md b/docs/combined-sources.md index 7fdb211..ce0335f 100644 --- a/docs/combined-sources.md +++ b/docs/combined-sources.md @@ -272,6 +272,81 @@ The Vaxrank consumer test carries these records through native dataset save/load and verifies frozen scores, selected windows, peptide constructs and mRNA constructs, plus a changed criterion that changes the selected window. +### Coverage and policy comparisons + +Coverage does not depend on whether a policy fills a missing score with zero, +includes unknown eligibility, or rejects a candidate. Use retained evaluations: + +```python +from topiary import policy_coverage, summarize_policy_coverage, compare_policy_evaluations + +coverage = policy_coverage(evaluated) +summary = summarize_policy_coverage(coverage) +reasons = summarize_policy_coverage(coverage, by=["level", "criterion", "reason"]) +comparison = compare_policy_evaluations(evaluated, another_evaluation) +shared = comparison.df.query("assessment_set == 'both'") +coverage.to_tsv("coverage.tsv") +comparison.to_tsv("policy-comparison.tsv") +``` + +`policy_coverage` emits one score and each named criterion per occurrence. Native +`pass`, `fail` and numeric `value` are all assessed; `unknown` becomes missing. +Not-applicable and not-evaluated remain distinct. A measured zero is assessed; +a zero from `score_fill` retains its missing raw score. The original status, +reason, criterion value, raw/effective score and eligibility remain visible. + +By default the denominator contains evaluated groups. Supply `universe`, a +DataFrame of unique identity keys that includes every evaluated group, to keep +expected inputs that produced no evaluation. Absent groups are not evaluated. +Keys default to the retained occurrence grouping; explicit `keys` must uniquely +identify occurrences. Do not drop flanks, offsets, samples or sources when that +would merge different occurrences. + +Returned prediction rows alone cannot reveal failed or unattempted requests. +For model coverage, pass an explicit `prediction_requests` DataFrame: one row +per expected occurrence/model/field assessment, with the identity keys plus +canonical `kind`, `prediction_method_name`, `predictor_version`, and numeric +`field` (for example `score`). A null version matches only an unstated version. +Include `allele_set` for joint-genotype evidence. An optional diagnostic `status` +(`missing`, `failed`, `not_applicable`, or `not_evaluated`), `reason`, and `detail` +can preserve explanations such as `insufficient_c_terminal_context` or a backend +failure when no row exists. Diagnostics cannot override observed measurements. +Topiary does not infer a context/domain failure from an absent value. + +Model records describe **availability**, not which model contributed to a +criterion. They match exact model/version, occurrence context and canonical MHC +scope; allele-free measurements can project, while allele-credited and joint +measurements retain their scope. A criterion can combine multiple models or +annotations, and this report does not invent lineage for that expression. + +The summary separates requested assessment counts from unique occurrences, +unique peptide/allele/genotype candidates, and the union of referenced raw +prediction rows. Repeated source discoveries can describe one candidate; +allele projections can reference the same raw row. Finite raw fields are counted +separately as `n_observed_prediction_rows`, even if conflicting values make the +assessment unknown. Requests with no output contribute to assessment counts, +not raw-row counts. `assessed_fraction` explicitly uses **all assessments** as +its denominator, including inapplicable and unevaluated requests. Group by +`reason` or any occurrence key to inspect the cause or location of missingness. + +Comparison preserves both memberships, eligibility, raw/effective scores and +criterion decisions. `assessment_set` separates `both`, `left_only`, +`right_only`, and `neither`; a finite raw score plus no unknown referenced +criterion is required for assessability. Unused and inapplicable criteria do +not disqualify an otherwise assessed score. Prefiltered groups have no assessed +score. `raw_score_delta` is right minus left **only on the shared assessable +set**; it does not normalize different policy scales or add scientific weights. + +Reports retain definitions, digests, contexts, provenance and explicit reporting +inputs in `extra["policy_coverage"]` or `extra["policy_comparison"]`. Save the +underlying `evaluated.evidence` too: replay it with `replay_selection_policy`, +then regenerate coverage using the saved `keys`, `universe` and +`prediction_requests` (convert the latter two record lists to DataFrames when +not null). CSV/TSV readers may add their normal per-row file `source` label; +coverage counts and replay are unchanged. Reports run no predictors and change +no selection. Vaxrank owns vaccine-window/construct choices and its frozen +`openvax-v1` bundle. + ## Identities and source evidence | Column | Meaning | diff --git a/scripts/check_vaxrank_candidates.py b/scripts/check_vaxrank_candidates.py index 87c604d..4906997 100644 --- a/scripts/check_vaxrank_candidates.py +++ b/scripts/check_vaxrank_candidates.py @@ -344,3 +344,63 @@ def construct(epitopes): assert restored.selection == saved.selection pd.testing.assert_frame_equal(replay.occurrences, changed.occurrences, check_exact=True) assert construct(restored.epitopes) == construct(changed_epitopes) + + +def test_short_context_coverage_explains_vaxrank_window_change(): + from dataclasses import replace + from topiary import (SelectionCriterion, TopiaryResult, evaluate_selection_policy, + policy_coverage, summarize_policy_coverage, compare_policy_evaluations) + from tests.test_policy_coverage import short_context_inputs + from vaxrank.config.loader import load_vaxrank_config, extract_epitope_config_kwargs + from vaxrank.epitope_dsl import prediction_group_columns + from vaxrank.window_selection import WindowSelection, select_windows + + bundle = load_vaxrank_config(config_path="builtin:openvax-v1") + cfg = EpitopeConfig(**extract_epitope_config_kwargs(bundle)) + baseline = SelectionPolicy.from_dict(cfg.selection_policy) + overlay = replace(baseline, name="synthetic-context-overlay", + score_by=baseline.score_by + ' * criterion("processing")', criteria=baseline.criteria + ( + SelectionCriterion("processing", "peptide_view(proteasome_cleavage.score)", "score"),)) + evidence = short_context_inputs() + frame = evidence.df + keys = prediction_group_columns(frame) + measured = frame.loc[frame.kind.eq("pMHC_affinity")].to_dict("records") + rows = [] + for row in measured: + prediction = Prediction(peptide=row["peptide"], allele=row["allele"], kind="pMHC_affinity", + value=row["value"], score=row["score"], predictor_name="original", predictor_version="1") + rows.append(dict(peptide=row["peptide"], source=row["source_sequence"], offset=row["peptide_offset"], + n_flank=row["n_flank"], c_flank=row["c_flank"], prediction_id=row["prediction_id"], + mutant=prediction, source_class="mutation", overlaps_targetable=True, + occurs_in_reference=False, occurs_in_non_CTA_reference=False, patient_alleles=[row["allele"]])) + epitopes = candidate_epitopes_from_rows(rows) + evaluations, choices = [], [] + for policy in (baseline, overlay): + configured = EpitopeConfig(**dict(extract_epitope_config_kwargs(bundle), selection_policy=policy.to_dict())) + scored = attach_per_allele_scores(epitopes, configured, topiary_df=frame) + evaluated = evaluate_selection_policy(TopiaryResult(frame), policy, group_keys=keys) + evaluations.append(evaluated) + candidates = [] + for row in measured: + sequence = row["source_sequence"] + antigen = VaccineAntigen(kind="mutation", amino_acids=sequence, + targetable_mask=TargetableMask((AminoAcidInterval(row["peptide_offset"], row["peptide_offset"] + 8),)), + tumor_specificity=TumorSpecificityAttestation(status="admitted", evidence_kind="synthetic_fixture", + evidence_source="check_vaxrank_candidates", patient_specific=True, rationale_code="test_only"), + source_identifier=row["prediction_id"]) + candidates.append(VaccinePeptide(antigen=antigen, + epitopes=[epitope for epitope in scored if epitope.prediction_id == row["prediction_id"]], + combined_score_expr="target_epitope_score", ranking_rules=("target_epitope_score",))) + chosen = select_windows(candidates, WindowSelection(self_weight=0., serum_weight=0.), preferred_length=13, limit=1) + choices.append(chosen[0].amino_acids) + assert choices == ["SIINFEKL", "AAASIINFEKLGGG"] + requested = evaluations[1].occurrences.query("allele != ''")[keys].assign(kind="proteasome_cleavage", + prediction_method_name="context-model", predictor_version="1", field="score") + requested["status"], requested["reason"] = ["missing", None], ["insufficient_c_terminal_context", None] + coverage = policy_coverage(evaluations[1], prediction_requests=requested) + summary = summarize_policy_coverage(coverage, by=["level", "reason"]) + assert summary.query("reason == 'insufficient_c_terminal_context'").n_missing.item() == 1 + comparison = compare_policy_evaluations(*evaluations).df + assert comparison.query("prediction_id == 'short'").assessment_set.item() == "left_only" + shared = comparison.query("assessment_set == 'both'") + assert shared.prediction_id.tolist() == ["long"] and shared.raw_score_delta.eq(0.).all() diff --git a/tests/test_consumer_workflows.py b/tests/test_consumer_workflows.py index f5210b4..0405f33 100644 --- a/tests/test_consumer_workflows.py +++ b/tests/test_consumer_workflows.py @@ -3523,3 +3523,43 @@ def forbidden(*args, **kwargs): assert restored.df.loc[restored.df.source_label.eq("exacto"), "candidate_id"].isna().all() assert restored.extra == combined.extra assert rank_candidates(restored, expression, **options).candidate_id.tolist() == pooled.candidate_id.tolist() + + +def test_short_context_coverage_explains_changed_selection_and_replays(tmp_path, pandas_string_inference): + from dataclasses import replace + from topiary import (SelectionCriterion, SelectionPolicy, evaluate_selection_policy, policy_coverage, + summarize_policy_coverage, compare_policy_evaluations, replay_selection_policy) + from .test_policy_coverage import short_context_inputs, KEYS + from .test_twin_conformance import DELIMITED_IO_TWINS + + evidence = short_context_inputs() + baseline = SelectionPolicy("synthetic-baseline", 'criterion("binding")', score_fill=0., min_score=.01, + criteria=(SelectionCriterion("binding", "1 / affinity.value", "score"),)) + overlay = replace(baseline, name="context-overlay", score_by=baseline.score_by + ' * criterion("processing")', + criteria=baseline.criteria + (SelectionCriterion("processing", "peptide_view(proteasome_cleavage.score)", "score"),)) + left = evaluate_selection_policy(evidence, baseline, group_keys=KEYS) + right = evaluate_selection_policy(evidence, overlay, group_keys=KEYS) + assert left.selected.sort_values("score", ascending=False).prediction_id.iloc[0] == "short" + assert right.selected.prediction_id.tolist() == ["long"] + requested = right.occurrences.query("allele != ''")[KEYS].assign( + kind="proteasome_cleavage", prediction_method_name="context-model", predictor_version="1", field="score") + requested["status"] = ["missing", None] + requested["reason"] = ["insufficient_c_terminal_context", None] + coverage = policy_coverage(right, prediction_requests=requested) + summary = summarize_policy_coverage(coverage, by=["level", "reason"]) + assert summary.query("reason == 'insufficient_c_terminal_context'").n_missing.item() == 1 + model = summarize_policy_coverage(coverage).query("level == 'prediction'").iloc[0] + assert model.n_assessments == 2 and model.assessed_fraction == .5 + comparison = compare_policy_evaluations(left, right) + assert comparison.df.query("prediction_id == 'short'").assessment_set.item() == "left_only" + shared = comparison.df.query("assessment_set == 'both'") + assert shared.prediction_id.tolist() == ["long"] + assert shared.raw_score_delta.eq(0.).all() # selection changed through missingness + for suffix, writer, _, reader in DELIMITED_IO_TWINS: + path = tmp_path / ("context." + suffix) + writer(right.evidence, path) + replay = replay_selection_policy(reader(path)) + repeated = policy_coverage(replay, prediction_requests=requested) + pd.testing.assert_frame_equal(repeated.df, coverage.df, check_exact=True) + pd.testing.assert_frame_equal(summarize_policy_coverage(repeated), summarize_policy_coverage(coverage)) + pd.testing.assert_frame_equal(compare_policy_evaluations(left, replay).df, comparison.df, check_exact=True) diff --git a/tests/test_policy_coverage.py b/tests/test_policy_coverage.py new file mode 100644 index 0000000..390b311 --- /dev/null +++ b/tests/test_policy_coverage.py @@ -0,0 +1,335 @@ +"""Coverage denominators and common-set comparisons retain unknown evidence.""" +from dataclasses import replace + +import numpy as np +import pandas as pd +import pytest + +from topiary import ( + SelectionCriterion, SelectionPolicy, TopiaryResult, compare_policy_evaluations, + evaluate_selection_policy, policy_coverage, replay_selection_policy, summarize_policy_coverage, +) +from .test_candidate_tables import source +from .test_twin_conformance import DELIMITED_IO_TWINS + +pytestmark = pytest.mark.usefixtures("pandas_string_inference") + +KEYS = ["prediction_id", "peptide", "allele"] + + +def evaluation(values=(0., np.nan), policy=None, **columns): + frame = source(values=values, prediction_id=["zero", "missing"], **columns) + policy = policy or SelectionPolicy("coverage", 'criterion("binding")', score_fill=0., strata=(), + criteria=(SelectionCriterion("binding", "affinity.value", "score"),)) + return evaluate_selection_policy(frame, policy, group_keys=KEYS) + + +def requests(evaluated, **columns): + frame = evaluated.occurrences[KEYS].copy() + for key, value in dict(kind="pMHC_affinity", prediction_method_name="original", predictor_version="1", + field="value", **columns).items(): + frame[key] = value + return frame + + +def test_zero_is_assessed_but_filled_zero_remains_missing(): + evaluated = evaluation() + coverage = policy_coverage(evaluated, prediction_requests=requests(evaluated)) + assert coverage.df.score.eq(0).all() + assert coverage.df.eligible.all() + assert coverage.df.query("prediction_id == 'zero'").status.eq("assessed").all() + assert coverage.df.query("prediction_id == 'missing'").status.eq("missing").all() + summary = summarize_policy_coverage(coverage) + assert summary.n_assessments.tolist() == [2, 2, 2] + assert summary.n_assessed.eq(1).all() and summary.n_missing.eq(1).all() + assert summary.assessed_fraction.eq(.5).all() + assert summary.n_prediction_rows.tolist() == [0, 0, 2] + assert coverage.extra["policy_coverage"]["evaluation"]["definition"]["score_fill"] == 0. + + +def test_missing_inputs_requests_and_diagnostics_are_in_denominator(): + evaluated = evaluation() + expected = pd.concat([evaluated.occurrences[KEYS], pd.DataFrame([dict( + prediction_id="absent", peptide="ELAGIGILT", allele="HLA-A*02:01")])], ignore_index=True) + requested = requests(evaluated).reindex(range(3)) + requested.loc[2] = [*expected.iloc[2], "pMHC_affinity", "original", "1", "value"] + requested["status"] = [None, "missing", "failed"] + requested["reason"] = [None, "insufficient_context", "backend_failure"] + coverage = policy_coverage(evaluated, universe=expected, prediction_requests=requested) + absent = coverage.df.query("prediction_id == 'absent'") + assert absent.status.tolist() == ["not_evaluated", "not_evaluated", "failed"] + summary = summarize_policy_coverage(coverage, by=["level", "reason"]) + assert summary.query("reason == 'insufficient_context'").n_missing.item() == 1 + assert summary.query("reason == 'backend_failure'").n_failed.item() == 1 + assert summarize_policy_coverage(coverage, by=["level"]).n_assessments.eq(3).all() + requested.loc[0, ["status", "reason"]] = ["failed", "backend_failure"] + with pytest.raises(ValueError, match="contradicts"): + policy_coverage(evaluated, universe=expected, prediction_requests=requested) + + +def test_native_failure_inapplicability_and_unused_are_not_missing(): + policy = SelectionPolicy("filter", 'criterion("binding")', filter_by='criterion("keep")', + criteria=(SelectionCriterion("keep", "n_rna_alt > 10", "eligibility"), + SelectionCriterion("binding", "affinity.value", "score", applies_to="n_rna_alt < 0"), + SelectionCriterion("unused", "affinity.value", "score"))) + coverage = policy_coverage(evaluation(policy=policy)) + by = coverage.df.set_index(["prediction_id", "level", "criterion"]) + assert by.loc[("zero", "criterion", "keep"), "native_status"] == "fail" + assert by.loc[("zero", "criterion", "keep"), "status"] == "assessed" + assert by.loc[("missing", "criterion", "binding"), "status"] == "not_applicable" + assert by.loc[("zero", "criterion", "binding"), "reason"] == "filtered_before_scoring" + assert by.loc[("missing", "criterion", "unused"), "reason"] == "not_referenced" + + +def test_model_version_field_and_allele_are_exact(): + evaluated = evaluation(values=(0., 3.)) + requested = requests(evaluated) + requested.loc[0, "predictor_version"] = "2" + requested.loc[1, "field"] = "absent_column" + coverage = policy_coverage(evaluated, prediction_requests=requested) + assert coverage.df.query("level == 'prediction'").reason.tolist() == ["no_prediction_rows", "missing_field"] + frame = evaluated.evidence.df.copy() + second_allele = frame.iloc[[0]].assign(allele="HLA-B*07:02", value=np.nan) + frame = pd.concat([frame, second_allele], ignore_index=True) + evaluated = evaluate_selection_policy(TopiaryResult(frame), evaluated.policy, group_keys=KEYS) + coverage = policy_coverage(evaluated, prediction_requests=requests(evaluated)) + row = coverage.df.query("level == 'prediction' and allele == 'HLA-B*07:02'").iloc[0] + assert row.status == "missing" and row.evidence_rows == [2] + + +def test_projection_and_duplicate_rows_do_not_inflate_assessments(): + frame = source(kind="proteasome_cleavage", allele="", prediction_id=["one", "two"], score=[.2, .8]).df + frame = pd.concat([frame, frame.iloc[[0]]], ignore_index=True) + evaluated = evaluate_selection_policy(TopiaryResult(frame), + SelectionPolicy("processing", "peptide_view(proteasome_cleavage.score)"), + group_keys=KEYS, alleles=["HLA-A*02:01", "HLA-B*07:02"]) + requested = requests(evaluated) + requested["kind"] = "proteasome_cleavage" + requested["field"] = "score" + coverage = policy_coverage(evaluated, prediction_requests=requested) + summary = summarize_policy_coverage(coverage).query("level == 'prediction'").iloc[0] + assert summary.n_assessed == summary.n_occurrences == 6 + assert summary.n_prediction_rows == 3 # three stored rows, not nine projected links + assert coverage.df.query("level == 'prediction' and prediction_id == 'one'").evidence_rows.tolist() == [[0, 2]] * 3 + + +def test_haplotype_is_not_borrowed_from_presenter(): + frame = source(prediction_id=["zero", "missing"], allele_set="HLA-A*02:01,HLA-B*07:02").df + evaluated = evaluate_selection_policy(TopiaryResult(frame), SelectionPolicy("haplotype", "peptide_view(affinity.value)"), + group_keys=KEYS + ["allele_set"]) + keys = KEYS + ["allele_set"] + requested = evaluated.occurrences[keys].assign(kind="pMHC_affinity", prediction_method_name="original", + predictor_version="1", field="value") + coverage = policy_coverage(evaluated, prediction_requests=requested) + assert coverage.df.query("level == 'prediction'").status.eq("assessed").all() + # An explicit narrower grouping cannot silently treat a missing genotype as per-allele. + requested = requested.drop(columns="allele_set") + coverage = policy_coverage(evaluated, keys=KEYS, prediction_requests=requested) + assert coverage.df.query("level == 'prediction'").status.eq("missing").all() + + +def test_conflicting_predictions_remain_unknown(): + evaluated = evaluation() + frame = pd.concat([evaluated.evidence.df, evaluated.evidence.df.iloc[[0]].assign(value=100.)], ignore_index=True) + evaluated = evaluate_selection_policy(TopiaryResult(frame), evaluated.policy, group_keys=KEYS) + coverage = policy_coverage(evaluated, prediction_requests=requests(evaluated)) + assert coverage.df.query("level == 'prediction' and prediction_id == 'zero'").reason.item() == "conflicting_evidence" + + +def test_comparison_preserves_union_and_only_compares_common_assessable_scores(): + left = evaluation(values=(0., 3.)) + right = evaluation(values=(1., np.nan)) + comparison = compare_policy_evaluations(left, right) + assert comparison.df.assessment_set.tolist() == ["both", "left_only"] + assert comparison.df.raw_score_delta.iloc[0] == 1. + assert pd.isna(comparison.df.raw_score_delta.iloc[1]) + assert comparison.df.right_score.iloc[1] == 0. + assert comparison.df.right_unknown_criteria.iloc[1] == ["binding"] + assert comparison.extra["policy_comparison"]["left"]["sha256"] == left.policy.sha256 + frame = right.evidence.df.copy() + frame.loc[1, "prediction_id"] = "other" + right = evaluate_selection_policy(TopiaryResult(frame), right.policy, group_keys=KEYS) + comparison = compare_policy_evaluations(left, right) + assert comparison.df.left_present.tolist() == [True, True, False] + assert comparison.df.right_present.tolist() == [True, False, True] + assert comparison.df.assessment_set.tolist() == ["both", "left_only", "neither"] + + +def test_unknown_included_filter_is_not_fully_assessable(): + policy = SelectionPolicy("include", "1", filter_by='criterion("binding")', unknown="include", + criteria=(SelectionCriterion("binding", "affinity.value < 100", "eligibility"),)) + evaluated = evaluation(policy=policy) + assert evaluated.occurrences.score.eq(1).all() + comparison = compare_policy_evaluations(evaluated, evaluated) + assert comparison.df.assessment_set.tolist() == ["both", "neither"] + + +@pytest.mark.parametrize("suffix,writer,_,reader", DELIMITED_IO_TWINS) +def test_coverage_and_comparison_roundtrip_replay(tmp_path, suffix, writer, _, reader, pandas_string_inference): + evaluated = evaluation() + requested = requests(evaluated) + coverage = policy_coverage(evaluated, prediction_requests=requested) + path = tmp_path / ("coverage." + suffix) + writer(coverage, path) + restored = reader(path) + pd.testing.assert_frame_equal(restored.df[coverage.df.columns], coverage.df, check_exact=True) + pd.testing.assert_frame_equal(summarize_policy_coverage(restored), summarize_policy_coverage(coverage)) + evidence_path = tmp_path / ("evidence." + suffix) + writer(evaluated.evidence, evidence_path) + replay = replay_selection_policy(reader(evidence_path)) + metadata = restored.extra["policy_coverage"] + regenerated = policy_coverage(replay, keys=metadata["keys"], + prediction_requests=pd.DataFrame(metadata["prediction_requests"])) + pd.testing.assert_frame_equal(regenerated.df, coverage.df, check_exact=True) + comparison = compare_policy_evaluations(evaluated, replay) + writer(comparison, path) + pd.testing.assert_frame_equal(reader(path).df[comparison.df.columns], comparison.df, check_exact=True) + + +def test_empty_and_invalid_denominators(): + evaluated = evaluation() + empty = evaluate_selection_policy(TopiaryResult(evaluated.evidence.df.iloc[:0]), evaluated.policy, group_keys=KEYS) + assert policy_coverage(empty).df.empty + assert summarize_policy_coverage(policy_coverage(empty)).empty + assert compare_policy_evaluations(empty, empty).df.empty + with pytest.raises(ValueError, match="include every"): + policy_coverage(evaluated, universe=evaluated.occurrences[KEYS].iloc[:1]) + with pytest.raises(ValueError, match="uniquely"): + policy_coverage(evaluated, keys=["allele"]) + requested = requests(evaluated) + with pytest.raises(ValueError, match="Duplicate prediction"): + policy_coverage(evaluated, prediction_requests=pd.concat([requested, requested])) + with pytest.raises(ValueError, match="Missing prediction"): + policy_coverage(evaluated, prediction_requests=requested.drop(columns="field")) + with pytest.raises(ValueError, match="universe"): + policy_coverage(evaluated, prediction_requests=requested.assign(prediction_id="absent")) + with pytest.raises(ValueError, match="groupings differ"): + compare_policy_evaluations(evaluated, evaluate_selection_policy(evaluated.evidence, evaluated.policy)) + + +def short_context_inputs(): + """Synthetic retained outputs: the short source has no processing result.""" + frame = source(peptides=("SIINFEKL", "SIINFEKL"), values=(50., 100.), prediction_id=["short", "long"], + peptide_offset=[0, 3], n_flank=["", "AAA"], c_flank=["", "GGG"], + source_sequence=["SIINFEKL", "AAASIINFEKLGGG"]).df + processing = frame.iloc[[1]].assign(kind="proteasome_cleavage", prediction_method_name="context-model", + allele="", value=1., score=1.) + return TopiaryResult(pd.concat([frame, processing], ignore_index=True)) + + +@pytest.mark.parametrize("values", [(0., np.nan), (2., 3.)]) +@pytest.mark.parametrize("filtered", [False, True]) +def test_report_and_comparison_share_assessability(values, filtered): + from .test_twin_conformance import POLICY_COVERAGE_TWINS + report, compare = POLICY_COVERAGE_TWINS + evaluated = evaluation(values=values) + if filtered: + evaluated = evaluation(values=values, policy=replace(evaluated.policy, filter_by="n_rna_alt > 10")) + coverage = report(evaluated).df + comparable = compare(evaluated, evaluated).df + for row in comparable.to_dict("records"): + group = coverage[coverage.prediction_id.eq(row["prediction_id"])] + expected = group.loc[group.level.eq("score"), "status"].eq("assessed").all() + expected &= ~group.loc[group.level.eq("criterion"), "status"].eq("missing").any() + assert row["left_assessable"] == row["right_assessable"] == bool(expected) + + +def test_allele_credited_peptide_rows_do_not_broadcast(): + frame = source(kind="proteasome_cleavage", prediction_id=["one", "two"], score=[.2, .8]).df + evaluated = evaluate_selection_policy(TopiaryResult(frame), + SelectionPolicy("processing", "peptide_view(proteasome_cleavage.score)"), + group_keys=KEYS, alleles=["HLA-A*02:01", "HLA-B*07:02"]) + requested = requests(evaluated) + requested["kind"], requested["field"] = "proteasome_cleavage", "score" + coverage = policy_coverage(evaluated, prediction_requests=requested) + predictions = coverage.df.query("level == 'prediction'") + assert predictions.query("allele == 'HLA-A*02:01'").status.eq("assessed").all() + assert predictions.query("allele == 'HLA-B*07:02'").status.eq("missing").all() + + +def test_repeated_discoveries_keep_occurrences_but_share_candidate_count(): + from topiary import combine_sources + combined = combine_sources({"one": source(), "two": source(allele="A0201")}, sample_name="patient") + evaluated = evaluate_selection_policy(combined, SelectionPolicy("binding", "affinity.value"), + group_keys=["candidate_sample", "source_label", "peptide", "allele"], source_contexts={"one": {}, "two": {}}) + coverage = policy_coverage(evaluated) + summary = summarize_policy_coverage(coverage).iloc[0] + assert summary.n_occurrences == summary.n_assessments == 4 + assert summary.n_candidates == 2 + assert coverage.extra["policy_coverage"]["evaluation"]["evidence_extra"]["combined_sources"] + + +def test_report_retains_filter_and_acceptance_decisions(): + evaluated = evaluation() + coverage = policy_coverage(evaluated).df + assert coverage.query("prediction_id == 'missing'").decision_reason.eq("missing_score").all() + assert coverage.filter_value.eq(True).all() + comparison = compare_policy_evaluations(evaluated, evaluated).df + assert comparison.left_filter_value.tolist() == evaluated.occurrences.filter_value.tolist() + frame = evaluated.evidence.df.assign(left_score=["one", "two"]) + renamed = evaluate_selection_policy(TopiaryResult(frame), evaluated.policy, + group_keys=["left_score", "peptide", "allele"]) + with pytest.raises(ValueError, match="reserved comparison"): + compare_policy_evaluations(renamed, renamed) + + +def test_haplotype_request_scope_is_retained_and_canonical_duplicates_rejected(): + frame = source(prediction_id=["zero", "missing"], allele_set="HLA-A*02:01,HLA-B*07:02").df + evaluated = evaluate_selection_policy(TopiaryResult(frame), SelectionPolicy("haplotype", "peptide_view(affinity.value)"), + group_keys=KEYS) + requested = requests(evaluated).assign(allele_set="B0702,A0201") + coverage = policy_coverage(evaluated, prediction_requests=requested).df.query("level == 'prediction'") + assert coverage.requested_allele_set.eq("HLA-A*02:01,HLA-B*07:02").all() + with pytest.raises(ValueError, match="Duplicate prediction"): + policy_coverage(evaluated, prediction_requests=pd.concat([ + requested, requested.assign(allele_set="HLA-A*02:01,HLA-B*07:02")])) + + +def test_missing_model_request_uses_its_own_scope_declaration(): + frame = source(prediction_id=["zero", "missing"]) + support = {"requested": {"pMHC_presentation": {"mhc_dependence": "none"}}, + "unrelated": {"pMHC_presentation": {"mhc_dependence": "single_allele"}}} + evaluated = evaluate_selection_policy(frame, SelectionPolicy("constant", "1"), group_keys=KEYS, + kind_support=support) + requested = requests(evaluated) + requested["kind"], requested["prediction_method_name"] = "pMHC_presentation", "requested" + coverage = policy_coverage(evaluated, prediction_requests=requested).df.query("level == 'prediction'") + assert coverage.status.eq("missing").all() + assert coverage.mhc_dependence.eq("none").all() + assert coverage.evidence_rows.tolist() == [[], []] + + +@pytest.mark.parametrize("version", [None, "", "nan", "null", ""]) +def test_unnamed_versions_use_the_existing_public_identity_rule(version): + from topiary import is_named_version + assert not is_named_version(version) + evaluated = evaluation(values=(0., 3.), predictor_version=version) + requested = requests(evaluated) + requested["predictor_version"] = None + coverage = policy_coverage(evaluated, prediction_requests=requested).df.query("level == 'prediction'") + assert coverage.status.eq("assessed").all() + assert coverage.predictor_version.isna().all() + requested["predictor_version"] = "1" + assert policy_coverage(evaluated, prediction_requests=requested).df.query("level == 'prediction'").status.eq("missing").all() + + +def test_absent_version_column_is_unnamed_and_duplicate_spellings_are_rejected(pandas_string_inference): + original = evaluation(values=(0., 3.)) + evaluated = evaluate_selection_policy(TopiaryResult(original.evidence.df.drop(columns="predictor_version")), + original.policy, group_keys=KEYS) + requested = requests(evaluated) + requested["predictor_version"] = None + assert policy_coverage(evaluated, prediction_requests=requested).df.query("level == 'prediction'").status.eq("assessed").all() + with pytest.raises(ValueError, match="Duplicate prediction"): + policy_coverage(evaluated, prediction_requests=pd.DataFrame([ + *requested.to_dict("records"), *requested.assign(predictor_version="").to_dict("records")])) + + +@pytest.mark.parametrize("version", ["NA", "unknown"]) +def test_literal_named_versions_are_not_collapsed_to_missing(version): + evaluated = evaluation(values=(0., 3.), predictor_version=version) + requested = requests(evaluated) + requested["predictor_version"] = version + assert policy_coverage(evaluated, prediction_requests=requested).df.query("level == 'prediction'").status.eq("assessed").all() + requested["predictor_version"] = None + assert policy_coverage(evaluated, prediction_requests=requested).df.query("level == 'prediction'").status.eq("missing").all() diff --git a/tests/test_twin_conformance.py b/tests/test_twin_conformance.py index 5b4d26d..36fafb7 100644 --- a/tests/test_twin_conformance.py +++ b/tests/test_twin_conformance.py @@ -49,7 +49,7 @@ read_isovar_hypotheses, combine_sources, rank_candidates, rank_with_policy, evaluate_scores, SelectionPolicy, resolve_selection_policy, evaluate_selection_policy, - EvalContext, evaluate_filter, + EvalContext, evaluate_filter, policy_coverage, compare_policy_evaluations, predict_peptide_occurrences, match_self_peptides, self_matches_in_windows, Affinity, Column, apply_filter, apply_sort, @@ -168,6 +168,9 @@ def _sv_support_record(support): OCCURRENCE_POLICY_TWINS = (evaluate_selection_policy, EvalContext) FILTER_DECISION_TWINS = (evaluate_filter, apply_filter) +# Coverage summaries and common-set comparisons must agree on assessability. +POLICY_COVERAGE_TWINS = (policy_coverage, compare_policy_evaluations) + # Explicit occurrence and named-peptide calls agree when context is the same. # Driven by test_peptide_occurrences.py; source coordinates differ legitimately. PEPTIDE_OCCURRENCE_TWINS = ( diff --git a/topiary/__init__.py b/topiary/__init__.py index a269121..36d046f 100644 --- a/topiary/__init__.py +++ b/topiary/__init__.py @@ -9,6 +9,7 @@ PolicyEvaluation, evaluate_selection_policy, replay_selection_policy, select_policy_representatives, describe_evaluation_context, ) +from .policy_coverage import policy_coverage, summarize_policy_coverage, compare_policy_evaluations from .selection_policy import ( SelectionPolicy, resolve_selection_policy, read_selection_policy, write_selection_policy, rank_with_policy, @@ -205,7 +206,7 @@ encode_amino_acids, ) -__version__ = "5.92.0" +__version__ = "5.93.0" __all__ = [ "normalize_isovar_rna_support", @@ -232,6 +233,9 @@ "CriterionEvaluation", "evaluate_selection_criteria", "candidate_identifier", + "policy_coverage", + "summarize_policy_coverage", + "compare_policy_evaluations", "PolicyEvaluation", "evaluate_selection_policy", "replay_selection_policy", diff --git a/topiary/policy_coverage.py b/topiary/policy_coverage.py new file mode 100644 index 0000000..4dbb45d --- /dev/null +++ b/topiary/policy_coverage.py @@ -0,0 +1,383 @@ +"""Coverage and comparisons over retained policy decisions, without inference.""" + +from copy import deepcopy + +import numpy as np +import pandas as pd + +from .policy_evaluation import PolicyEvaluation, _json_value, _key +from .ranking import is_named_version, is_stated, mhc_dependence, prediction_field_values, prediction_mhc_scope +from .result import TopiaryResult + + +_STATUSES = ("assessed", "missing", "failed", "not_applicable", "not_evaluated") +_MODEL_KEYS = ["kind", "prediction_method_name", "predictor_version", "field"] +_COLUMNS = ["occurrence_id", "level", "criterion", "role", *_MODEL_KEYS, "mhc_dependence", "requested_allele_set", "status", "reason", + "native_status", "value", "detail", "raw_score", "score", "eligible", + "filter_value", "filter_retained", "decision_reason", "evidence_rows", "observed_evidence_rows"] + +_COMPARISON_FIELDS = ["present", "assessable", "score_status", "score_reason", "unknown_criteria", "criteria", + "occurrence_id", "raw_score", "score", "eligible", "reason", "filter_value", + "filter_retained", "evidence_rows", "supporting_rows"] +_COMPARISON_COLUMNS = [*[name + "_" + key for name in ("left", "right") for key in _COMPARISON_FIELDS], + "assessment_set", "raw_score_delta"] + + +def _metadata(evaluation): + if not isinstance(evaluation, PolicyEvaluation): + raise TypeError("evaluation must be a PolicyEvaluation") + return evaluation.evidence.extra["policy_evaluation"] + + +def _report_context(evaluation): + # Retain definitions/provenance once, without duplicating every occurrence + # already represented in the report's records. + context = {key: value for key, value in _metadata(evaluation).items() if key != "occurrences"} + context["evidence_sources"] = evaluation.evidence.sources + context["evidence_extra"] = {key: value for key, value in evaluation.evidence.extra.items() + if key != "policy_evaluation"} + return deepcopy(context) + + +def _identity_keys(evaluation, keys): + metadata = _metadata(evaluation) + if keys is None: + groups = [part["context"]["group_keys"] for part in metadata["partitions"]] + if any(group != groups[0] for group in groups[1:]): + raise ValueError("Different source groupings require explicit coverage keys") + keys = list(groups[0]) if groups else [] + if any(part["source_label"] is not None for part in metadata["partitions"]) and "source_label" not in keys: + keys.insert(0, "source_label") + keys = list(keys) + if not keys or len(set(keys)) != len(keys) or set(keys) & set(_COLUMNS): + raise ValueError("keys must be unique identity columns, distinct from report columns") + _index(evaluation.occurrences, keys) + return keys + + +def _index(frame, keys): + if not frame.columns.is_unique: + raise ValueError("Identity frames must have distinct column names") + missing = set(keys) - set(frame.columns) + if missing: + raise ValueError(f"Missing identity columns: {sorted(missing)}") + rows = frame.to_dict("records") + identities = [_key([row[key] for key in keys]) for row in rows] + if len(set(identities)) != len(identities): + raise ValueError("keys must uniquely identify occurrences; include source and context identity") + return dict(zip(identities, rows)) + + +def _universe(evaluation, keys, universe): + observed = _index(evaluation.occurrences, keys) + expected = observed if universe is None else _index(universe, keys) + if set(observed) - set(expected): + raise ValueError("universe must include every evaluated occurrence") + return expected, observed + + +def _score_status(row): + if row is None: + return "not_evaluated", "no_evaluation" + if not row["filter_retained"]: + return "not_evaluated", "filtered_before_scoring" + if pd.isna(row["raw_score"]) or not np.isfinite(row["raw_score"]): + return "missing", "missing_score" + return "assessed", "observed_zero" if row["raw_score"] == 0 else "observed_value" + + +def _base(identity, row): + return dict(identity, occurrence_id=None if row is None else row["occurrence_id"], + raw_score=None if row is None else row["raw_score"], + score=None if row is None else row["score"], + eligible=False if row is None else row["eligible"], + filter_value=None if row is None else row["filter_value"], + filter_retained=None if row is None else row["filter_retained"], + decision_reason="no_evaluation" if row is None else row["reason"], + evidence_rows=[], observed_evidence_rows=[]) + + +def _version_key(value): + return str(value).strip() if is_named_version(value) else None + + +def _prediction_record(evaluation, row, request): + # Model availability is independent of which DSL expression read the model. + frame = evaluation.evidence.long_df + positions = [] if row is None else row["supporting_rows"] + selected = frame.iloc[positions] + for key in _MODEL_KEYS[:-1]: + wanted = _key(request[key]) + if key == "predictor_version": + versions = selected[key] if key in selected else pd.Series(None, index=selected.index, dtype=object) + selected = selected.loc[versions.map(lambda version: _key(_version_key(version))).eq(wanted)] + elif key not in selected: + selected = selected.iloc[:0] + else: + selected = selected.loc[selected[key].map(_key).eq(wanted)] + support = {} + for partition in _metadata(evaluation)["partitions"]: + if partition["source_label"] is None or partition["source_label"] == (row or request).get("source_label"): + support = partition["context"]["kind_support"] + break + # A request that returned no rows still names its model. Give the public + # resolver that declaration so unrelated models cannot decide its scope. + scope_rows = selected if len(selected) else pd.DataFrame([request]) + dependence = mhc_dependence(request["kind"], kind_support=support, rows=scope_rows) + scope = prediction_mhc_scope(request.get("allele"), dependence=dependence, + allele_set=request.get("allele_set")) + if scope is None: + selected = selected.iloc[:0] + else: + mask = [] + for item in selected.to_dict("records"): + matches = prediction_mhc_scope(item.get("allele"), dependence=dependence, + allele_set=item.get("allele_set")) == scope + # Peptide-level rows explicitly credited to one allele retain that + # restriction, matching peptide_view; allele-free rows can project. + if dependence == "none" and is_stated(item.get("allele")): + matches = matches and prediction_mhc_scope(item["allele"], dependence="single_allele") == ( + prediction_mhc_scope(request.get("allele"), dependence="single_allele")) + mask.append(matches) + selected = selected.loc[np.asarray(mask, dtype=bool)] + links = [int(position) for position in selected.index] + status, reason, value, detail = "missing", "no_prediction_rows", None, None + field = request["field"] + observed_links = [] + if field in selected: + numeric = pd.to_numeric(selected[field], errors="coerce") + observed_links = [int(index) for index in selected.index[np.isfinite(numeric).fillna(False)]] + if len(selected): + reason = "missing_field" if field not in selected else "missing_evidence" + if field in selected: + try: + values = prediction_field_values(selected.assign(_coverage_group=0), field, + group_keys=["_coverage_group"]) + value = values.iloc[0] + if not pd.isna(value) and np.isfinite(value): + status, reason = "assessed", "observed_zero" if value == 0 else "observed_value" + except ValueError as exc: + reason, detail = "conflicting_evidence", str(exc) + declared = request.get("status") + if declared is not None and not pd.isna(declared): + if declared not in _STATUSES[1:]: + raise ValueError("Request diagnostics must declare a non-assessed coverage status") + if status == "assessed": + raise ValueError("Request diagnostic contradicts an observed prediction") + if not isinstance(request.get("reason"), str) or not request["reason"]: + raise ValueError("Request diagnostics require a nonempty reason") + status, reason = declared, request["reason"] + detail = request.get("detail", detail) + return dict(level="prediction", **{key: request[key] for key in _MODEL_KEYS}, + status=status, native_status=status, reason=reason, detail=detail, + mhc_dependence=dependence, requested_allele_set=(",".join(scope[1:]) if scope and dependence == "haplotype" else None), + value=value, evidence_rows=links, observed_evidence_rows=observed_links) + + +def policy_coverage(evaluation, *, keys=None, universe=None, prediction_requests=None): + """Retain coverage independently of policy acceptance and score filling. + + Parameters + ---------- + evaluation : PolicyEvaluation + Complete evaluation, including rejected occurrences and raw evidence. + keys : sequence of str, optional + Unique occurrence identity. Defaults to retained grouping, plus source + label for partitioned evaluations. Incompatible groupings require keys. + universe : pandas.DataFrame, optional + Expected identities, including all evaluated occurrences. Absent input + groups are explicitly not evaluated. None uses evaluated groups only; + no Cartesian product of peptides, alleles or models is guessed. + prediction_requests : pandas.DataFrame, optional + Exact expected model assessments: identity keys, ``kind``, + ``prediction_method_name``, ``predictor_version`` and numeric ``field``. + Model names/kinds use retained canonical spellings; a null version means + an unstated version, never a wildcard. Include ``allele_set`` for a + haplotype request. Optional ``status``, ``reason`` and ``detail`` carry + explicit missing/failed/not_applicable/not_evaluated diagnostics. A + diagnostic contradicting a measured value is rejected. Requests report + model availability, not criterion-to-model lineage. None omits this + level: returned rows alone cannot establish a requested denominator. + + Returns + ------- + TopiaryResult + One score and each named criterion per expected occurrence, plus one + record per explicit prediction request. ``status`` distinguishes + assessed/missing/failed/not_applicable/not_evaluated; native decisions, + reasons, raw and effective scores, and eligibility stay separate. + Prediction links are positions in the retained evidence (shared across + allele-free projections), not new observations. Empty input produces a + typed table with no records. Metadata retains policy/context/provenance + and the exact reporting inputs for CSV/TSV persistence and replay. + """ + keys = _identity_keys(evaluation, keys) + expected, observed = _universe(evaluation, keys, universe) + records = [] + for identity, item in expected.items(): + row = observed.get(identity) + base = _base({key: item[key] for key in keys}, row) + status, reason = _score_status(row) + records.append(dict(base, level="score", status=status, native_status=status, + reason=reason, value=None if row is None else row["raw_score"])) + criteria = (row.get("criteria", []) if row is not None else + [dict(criterion=criterion.name, status="not_evaluated", reason="no_evaluation") + for criterion in evaluation.policy.criteria]) + for criterion in criteria: + native = criterion["status"] + status = "assessed" if native in {"pass", "fail", "value"} else "missing" if native == "unknown" else native + records.append(dict(base, **dict(criterion, level="criterion", native_status=native, status=status))) + requests = [] if prediction_requests is None else prediction_requests.to_dict("records") + seen = set() + for request in requests: + missing = set([*keys, *_MODEL_KEYS]) - set(request) + if missing: + raise ValueError(f"Missing prediction request columns: {sorted(missing)}") + for key in ("kind", "prediction_method_name", "field"): + if not isinstance(request[key], str) or not request[key].strip(): + raise ValueError(f"Prediction request {key} must be a nonempty string") + request = dict(request, predictor_version=_version_key(request["predictor_version"])) + identity = _key([request[key] for key in keys]) + if identity not in expected: + raise ValueError("Prediction requests must belong to the explicit occurrence universe") + row = observed.get(identity) + prediction = _prediction_record(evaluation, row, request) + request_id = _key([identity, *[request[key] for key in _MODEL_KEYS], + prediction["mhc_dependence"], prediction["requested_allele_set"]]) + if request_id in seen: + raise ValueError("Duplicate prediction request") + seen.add(request_id) + base = _base({key: request[key] for key in keys}, row) + records.append(dict(base, **prediction)) + columns = list(dict.fromkeys([*keys, *_COLUMNS, *(key for row in records for key in row)])) + frame = pd.DataFrame(records, columns=columns) + return TopiaryResult(frame, form="long", extra={"policy_coverage": _json_value(dict( + schema_version=1, keys=keys, evaluation=_report_context(evaluation), + universe=None if universe is None else universe.to_dict("records"), + prediction_requests=None if prediction_requests is None else requests, + ))}) + + +def summarize_policy_coverage(coverage, *, by=None): + """Count assessment coverage with explicit, independent denominators. + + Parameters + ---------- + coverage : TopiaryResult + Output of :func:`policy_coverage`, including after typed CSV/TSV reload. + by : sequence of str, optional + Grouping columns. Defaults to level, criterion, kind, method, version + and field; add ``reason`` or occurrence keys to inspect missingness. + An empty list gives one overall group (across assessment types). + + Returns + ------- + pandas.DataFrame + ``n_assessments`` and counts for every status; ``assessed_fraction`` uses + all requested assessments, including inapplicable/unevaluated ones. + ``n_occurrences`` counts distinct report identities; ``n_candidates`` + counts distinct peptide/canonical-allele/genotype identities when available (or + occurrence identities otherwise), across repeated source discoveries. + ``n_prediction_rows`` counts the union of linked raw row positions, + never their sum across projections; ``n_observed_prediction_rows`` counts + the subset with finite numeric fields (even when rows conflict). These + raw-row counts cannot include requests that returned no rows. Equal repeated raw rows remain raw + rows, not independent biological evidence. Empty input has no groups. + """ + metadata = coverage.extra["policy_coverage"] + keys = metadata["keys"] + frame = coverage.df + by = ["level", "criterion", *_MODEL_KEYS] if by is None else list(by) + if len(set(by)) != len(by) or set(by) - set(frame.columns): + raise ValueError("by must name unique coverage columns") + count_columns = ["n_assessments", "n_occurrences", "n_candidates", "n_prediction_rows", "n_observed_prediction_rows", + *["n_" + status for status in _STATUSES], "assessed_fraction"] + records = [] + groups = frame.groupby(by, dropna=False, sort=False) if by else [((), frame)] + for identity, group in groups: + if group.empty: + continue + identity = identity if isinstance(identity, tuple) else (identity,) + counts = group.status.value_counts() + candidate_keys = [key for key in ("sample_name", "candidate_sample", "peptide", "allele", "allele_set") if key in keys] + # Without peptide identity a subset of dimensions cannot define a candidate. + if "peptide" not in candidate_keys: + candidate_keys = keys + candidates = group[candidate_keys].drop_duplicates().to_dict("records") + for candidate in candidates: + if "allele" in candidate: + dependence = "haplotype" if is_stated(candidate.get("allele_set")) else "single_allele" + candidate["allele"] = prediction_mhc_scope(candidate["allele"], dependence=dependence, + allele_set=candidate.pop("allele_set", None)) + records.append(dict(zip(by, identity), n_assessments=len(group), + n_occurrences=len(group[keys].drop_duplicates()), + n_candidates=len({_key(candidate) for candidate in candidates}), + n_prediction_rows=len({index for links in group.evidence_rows for index in links}), + n_observed_prediction_rows=len({index for links in group.observed_evidence_rows for index in links}), + **{"n_" + status: int(counts.get(status, 0)) for status in _STATUSES}, + assessed_fraction=float(counts.get("assessed", 0) / len(group)))) + return pd.DataFrame(records, columns=[*by, *count_columns]) + + +def compare_policy_evaluations(left, right, *, keys=None, universe=None): + """Align full membership and compare raw scores only where both are assessable. + + Parameters + ---------- + left, right : PolicyEvaluation + Retained evaluations. They may have different policies and input rows. + keys : sequence of str, optional + Unique shared occurrence identity. Defaults to identical saved grouping; + ambiguous or incompatible identities raise rather than joining loosely. + universe : pandas.DataFrame, optional + Expected identities including both evaluations. None uses their union. + + Returns + ------- + TopiaryResult + One record per identity, preserving both memberships, raw/effective + scores, eligibility and native criterion decisions. ``assessment_set`` + is both/left_only/right_only/neither. Assessability requires an observed + finite raw score and no unknown referenced criterion; not-applicable + and unused criteria do not disqualify it. Prefiltered groups are not + score-assessable. ``raw_score_delta`` (right minus left) exists only for + both, with no normalization across policy scales. Metadata preserves + both definitions, digests, runtime contexts and execution provenance. + Empty inputs produce an empty table; absent groups remain explicit. + """ + left_keys = _identity_keys(left, keys) + right_keys = _identity_keys(right, keys) + if left_keys != right_keys: + raise ValueError("Comparison groupings differ; supply explicit keys") + keys = left_keys + if set(keys) & set(_COMPARISON_COLUMNS): + raise ValueError("keys collide with reserved comparison columns") + lhs, rhs = _index(left.occurrences, keys), _index(right.occurrences, keys) + expected = {**lhs, **rhs} if universe is None else _index(universe, keys) + if (set(lhs) | set(rhs)) - set(expected): + raise ValueError("universe must include every evaluated occurrence") + records = [] + for identity, item in expected.items(): + record = {key: item[key] for key in keys} + for name, rows in (("left", lhs), ("right", rhs)): + row = rows.get(identity) + status, reason = _score_status(row) + criteria = [] if row is None else row.get("criteria", []) + unknown = [criterion["criterion"] for criterion in criteria if criterion["status"] == "unknown"] + record.update({name + "_" + key: value for key, value in dict( + present=row is not None, assessable=status == "assessed" and not unknown, + score_status=status, score_reason=reason, unknown_criteria=unknown, criteria=_json_value(criteria), + **{key: None if row is None else row[key] for key in + ("occurrence_id", "raw_score", "score", "eligible", "reason", "filter_value", + "filter_retained", "evidence_rows", "supporting_rows")}, + ).items()}) + a, b = record["left_assessable"], record["right_assessable"] + record["assessment_set"] = "both" if a and b else "left_only" if a else "right_only" if b else "neither" + record["raw_score_delta"] = record["right_raw_score"] - record["left_raw_score"] if a and b else None + records.append(record) + columns = [*keys, *_COMPARISON_COLUMNS] + return TopiaryResult(pd.DataFrame(records, columns=columns), form="long", extra={ + "policy_comparison": _json_value(dict(schema_version=1, keys=keys, + left=_report_context(left), right=_report_context(right), + universe=None if universe is None else universe.to_dict("records"))), + })