From 5c256c4672ade1a3a7277fa4a3efa11dfef010e4 Mon Sep 17 00:00:00 2001 From: Alex Rubinsteyn Date: Thu, 1 Oct 2026 13:00:14 -0400 Subject: [PATCH 1/4] Evaluate occurrence policies with named criteria and retained audit evidence --- .github/workflows/tests.yml | 2 +- CHANGELOG.md | 12 + RELEASING.md | 2 +- docs/api.md | 22 ++ docs/combined-sources.md | 123 ++++++++- scripts/check_vaxrank_candidates.py | 118 ++++++++ tests/test_policy_evaluation.py | 169 ++++++++++++ tests/test_selection_criteria.py | 144 ++++++++++ tests/test_selection_policy.py | 4 +- tests/test_twin_conformance.py | 8 +- topiary/__init__.py | 23 +- topiary/candidates.py | 26 +- topiary/criterion_evaluation.py | 165 +++++++++++ topiary/policy_evaluation.py | 407 ++++++++++++++++++++++++++++ topiary/ranking/__init__.py | 3 +- topiary/ranking/apply.py | 58 +++- topiary/ranking/nodes.py | 34 ++- topiary/ranking/parser.py | 22 +- topiary/selection_criteria.py | 165 +++++++++++ topiary/selection_policy.py | 121 ++++++++- 20 files changed, 1577 insertions(+), 51 deletions(-) create mode 100644 tests/test_policy_evaluation.py create mode 100644 tests/test_selection_criteria.py create mode 100644 topiary/criterion_evaluation.py create mode 100644 topiary/policy_evaluation.py create mode 100644 topiary/selection_criteria.py diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 5fbe95a..08ed8f6 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -17,7 +17,7 @@ jobs: python-version: "3.12" - name: Install the released downstream consumer run: | - bash scripts/pip_install.sh pytest 'vaxrank==3.32.0' -e . + bash scripts/pip_install.sh pytest 'vaxrank==3.35.0' -e . python -m pip check - name: Verify table features through DSL scoring and vaccine construction run: python -m pytest scripts/check_vaxrank_candidates.py -q diff --git a/CHANGELOG.md b/CHANGELOG.md index 2da5d93..ed80dc6 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.90.0 + +- Evaluate saved policies at explicit occurrence/allele identity before selecting + representatives. Retain all evidence, concrete genotype declarations, + source-local model choices, score fill/gates and exact replay context (#444). +- Add reusable named DSL criteria, explicit composition and ordered tie-breaks, + with pass/fail/unknown/not-applicable/not-evaluated audit records. Schema 2 + preserves expanded definitions; schema-1 definitions keep their digests (#445). +- Fix context derivation with genotype callbacks and changed mappings (#447). +- Verify released Vaxrank 3.35.0 scoring, window selection, peptide/mRNA + construction and native dataset reload against the shared policy evaluator. + ## 5.89.0 - Save and replay named `SelectionPolicy` definitions with complete DSL expressions, diff --git a/RELEASING.md b/RELEASING.md index fdf0f5f..666ebf9 100644 --- a/RELEASING.md +++ b/RELEASING.md @@ -57,7 +57,7 @@ CI installs a pinned published Vaxrank release and runs the real candidate scoring and vaccine-construction workflow. To repeat it locally: ```bash -"$PYTHON" -m pip install 'vaxrank==3.32.0' +"$PYTHON" -m pip install 'vaxrank==3.35.0' "$PYTHON" -m pip check "$PYTHON" -m pytest scripts/check_vaxrank_candidates.py -q ``` diff --git a/docs/api.md b/docs/api.md index ab0123e..13915ed 100644 --- a/docs/api.md +++ b/docs/api.md @@ -663,3 +663,25 @@ available_properties() ``` Groups: `"core"`, `"manufacturability"`, `"immunogenicity"`. See [Peptide Properties](properties.md) for details. + +### Occurrence policy and criterion evaluation + +- `evaluate_selection_policy(result, policy, group_keys=..., alleles=..., + source_contexts=...)` returns `PolicyEvaluation` with complete `evidence`, + `occurrences`, `selected` and criterion `audit` views. +- `replay_selection_policy(evidence)` repeats evaluation from the embedded + definition and materialized runtime context without prediction. +- `select_policy_representatives(evaluation, candidate_keys=..., strata=...)` + selects actual occurrences and links alternatives without adding support. +- `describe_evaluation_context(context)` materializes grouping, model choices + and per-occurrence genotype declarations for replay. +- `candidate_identifier(sample, peptide, allele)` shares combined/projected + pMHC identity, using mhcgnomes for allele canonicalization. +- `SelectionCriterion`, `RankingTerm`, and `resolve_selection_expression` + define and expand named DSL criteria with role/cycle validation. +- `evaluate_selection_criteria(policy, context, references=...)` exposes named + measurements and reasons; `evaluate_filter(frame, expression, context=...)` + exposes group decisions without discarding rows. + +See [the complete source-policy workflow](combined-sources.md#named-criteria-and-audit-decisions) +for schema compatibility, source-local defaults, unknown handling and persistence. diff --git a/docs/combined-sources.md b/docs/combined-sources.md index 4afb43c..67797f1 100644 --- a/docs/combined-sources.md +++ b/docs/combined-sources.md @@ -153,14 +153,121 @@ fill, window rules, RNA weighting and construct settings must also be frozen in its bundle. A cutoff inside a score expression must not become a destructive pre-filter when capturing that baseline. -The representative ranking above is one view, not the complete construction -interface. Shared occurrence-level evaluation with retained alternatives is -tracked in [#444](https://github.com/openvax/topiary/issues/444). Optional named -criteria, references and pass/fail/unknown audit records are tracked in -[#445](https://github.com/openvax/topiary/issues/445); schema 1 currently accepts -direct DSL expressions and rejects unknown criteria/registry fields. These -layers will reuse the same DSL rather than require a second evaluator in -Vaxrank. +The representative ranking above is one view. `evaluate_selection_policy` +retains every input row and evaluates explicit occurrence identities before +representative selection or vaccine-window construction: + +```python +from topiary import ( + SelectionPolicy, evaluate_selection_policy, replay_selection_policy, + select_policy_representatives, read_tsv, +) + +policy = SelectionPolicy( + "example-baseline", "(affinity.value < 5000) * affinity.value.logistic_normalized(350, 150)", + score_fill=0.0, min_score=1e-5, duplicates="best", +) +evaluated = evaluate_selection_policy(combined, policy) +occurrences = evaluated.occurrences # includes excluded and unscorable alternatives +eligible = evaluated.selected # no candidate collapse +representatives = select_policy_representatives(evaluated) +evaluated.evidence.to_tsv("complete-evaluation.tsv") +replayed = replay_selection_policy(read_tsv("complete-evaluation.tsv")) +``` + +This example freezes a scoring convention; it does not establish an official +`openvax-v1` recipe or a calibrated probability of immunogenicity. The cutoff +remains inside the score, with a separate inclusive minimum-score gate. +`raw_score` preserves missingness even when `score_fill=0` supplies an effective +zero. Pre-filtered groups are not scored and cannot be restored by filling. +With no minimum gate, missing scores remain eligible but unranked, matching the +existing candidate-ranker convention. + +For direct Vaxrank frames, pass +`group_keys=["prediction_id", "peptide", "peptide_offset", "allele"]` and the +consumer's per-occurrence `alleles` mapping/callback. The callback is evaluated +once per peptide identity and saved as concrete declarations, never Python code. +Projected allele groups carry `supporting_rows` links, not duplicated RNA counts. +`evidence_rows` links the group's original observations. These positions refer +to `evaluated.evidence.long_df`; retain that complete long-form result when +saving an evaluation. Input columns and metadata remain available there. + +Use `source_contexts={label: {...}}` when sources have independent model defaults. +Every source label must be named; each mapping may replace `default_methods`, +`default_versions`, `kind_support`, and `alleles`. Filtering and scoring share +those choices. Definitions remain separate from runtime contexts, which are +stored alongside decisions in `extra["policy_evaluation"]`. Replay uses the +complete evidence and stored contexts and never invokes prediction. Model +defaults resolve ambiguity in the existing DSL; they are not a requirement that +all input measurements come from the selected model. + +`select_policy_representatives` chooses an actual eligible occurrence, without +summing support, and records all alternative occurrence IDs, including excluded +alternatives. Direct consumers supply their `candidate_keys` and `strata` if +combined-source candidate columns are absent. Missing values sort last; stable +input order breaks exact ties. Topiary owns these generic decisions; Vaxrank +still owns window geometry, source admission and construct assembly. + +### Named criteria and audit decisions + +Criteria reuse existing DSL expressions through an explicit namespace. This +example is synthetic policy content, not a recommended processing weight: + +```python +from topiary import SelectionCriterion, RankingTerm + +binding = SelectionCriterion("binding", "affinity.value < 500", "eligibility") +processing = SelectionCriterion( + "processing", "peptide_view(proteasome_cleavage.score)", "score", +) +policy = SelectionPolicy( + "example-processing", 'criterion("processing")', + filter_by='criterion("binding")', criteria=(binding, processing), + ranking_by=(RankingTerm("n_rna_alt", ascending=False),), + unknown="exclude", duplicates="best", +) +evaluated = evaluate_selection_policy(combined, policy) +audit = evaluated.audit +``` + +Eligibility criteria compose with explicit `&`, `|`, and `~`; score terms +combine through explicit arithmetic; `ranking_by` lists ordered expressions +and directions after the primary score. YAML overrides do none of this +implicitly. Predicate references may reference predicates, score terms may +reference score terms, and ranking terms may reference score or ranking terms. +An input column called `binding` remains a column; only `criterion("binding")` +means the named criterion. Unknown/cyclic references, duplicate names, wrong +roles and non-boolean eligibility outputs raise. Optional `applies_to` is an +eligibility expression; false applicability is recorded as `not_applicable`, +and its reference remains unknown rather than becoming an observed failure. + +Audit rows retain occurrence identity, raw row links, criterion name and role, +value, status and reason: + +| Status | Meaning | +| --- | --- | +| `pass` / `fail` | An applicable predicate evaluated true / false | +| `unknown` | Missing column/input, missing/ambiguous/conflicting model evidence, unknown applicability, or an out-of-domain calculation | +| `value` | An observed numeric term, including zero | +| `not_applicable` | The applicability predicate was observed false | +| `not_evaluated` | Unreferenced, or scoring skipped after a pre-filter | + +Named predicates use three-valued boolean logic. `unknown="exclude"`, +`"include"`, or `"error"` controls the decision on an unknown final eligibility +result, while audit values remain unknown. Direct-expression policies without +criteria keep historical DSL comparison behavior. A vector expression with an +ambiguous/conflicting model is unknown for that evaluation context; separate +source contexts prevent one source's model selection from being imposed on +another. Unexpected programming/DSL errors still raise. + +Schema 2 persists criteria, their complete expansions, references, ordered +terms, fill/gate settings and unknown handling in the policy digest. Original +schema-1 definitions retain their original serialization and digest. Definitions +are self-contained; a mutable registry cannot change a saved recipe. Export +`evaluated.evidence`, not a filtered ranking, to reproduce rejected alternatives. +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. ## Identities and source evidence diff --git a/scripts/check_vaxrank_candidates.py b/scripts/check_vaxrank_candidates.py index 3419adc..4652f9f 100644 --- a/scripts/check_vaxrank_candidates.py +++ b/scripts/check_vaxrank_candidates.py @@ -100,3 +100,121 @@ def test_candidate_features_reach_vaxrank_scoring_and_vaccine_construction(antig vaccines.sort(key=lambda v: v.target_epitope_score, reverse=True) assert vaccines[0].antigen.amino_acids == ("SIINFEKL" if policy == "original" else "GILGFVFTL") assert len(vaccines) == 2 # two pipelines did not become four vaccine targets + + +def test_occurrence_policy_matches_vaxrank_sources_windows_constructs_and_native_reload(tmp_path): + """Real Vaxrank consumer operations over retained heterogeneous evidence.""" + from dataclasses import replace + import json + from pathlib import Path + from varcode import Variant + from topiary import ( + CachedPredictor, ProteinFragment, TopiaryPredictor, TopiaryResult, + SelectionCriterion, evaluate_selection_policy, replay_selection_policy, + read_lens, read_pvacseq, + ) + from vaxrank.epitope_dataset import EpitopeDataset + from vaxrank.epitope_dsl import ( + default_score_expr, genotype_lookup, score_predictions, + prediction_group_columns, resolve_default_methods, resolve_default_versions, + epitopes_for_ranking, + ) + from vaxrank.core_logic import vaccine_peptides_from_epitopes + from vaxrank.mutant_protein_fragment import MutantProteinFragment + from vaxrank.peptide import assemble_peptide_constructs, PeptideConstructConfig + from vaxrank.mrna import assemble_mrna_constructs, RNAConstructConfig + from vaxrank.native_serialization import to_native_json + + root = Path(__file__).resolve().parents[1] / "tests/data" + translation = json.loads((root / "osteosarc_shared/translation-v1.json").read_text()) + prediction = json.loads((root / "osteosarc_shared/prediction-contract-v1.json").read_text()) + fragment = ProteinFragment.from_dict(translation["fragment"]) + # This versioned fixture is reconstructed from VCF/BAM in the full suite. + # Its numeric binding predictions are deliberately synthetic and cached. + direct = TopiaryPredictor(models=CachedPredictor(pd.DataFrame(prediction["rows"])), + only_novel_epitopes=True).predict_from_fragments([fragment]) + # Carry the original translated context alongside the measurement table. + direct["source_sequence"] = fragment.sequence + combined = combine_sources({ + "direct": TopiaryResult(direct), + "normalized": source(), + "lens": read_lens(root / "lens/sample_v1_4.tsv"), + "pvacseq": read_pvacseq(root / "pvacseq/mhc_i_all_epitopes.tsv"), + }, sample_name="fixture-patient") + dataset = EpitopeDataset.from_topiary(combined) + frame = dataset.scoring_frame() + keys = prediction_group_columns(frame) + cfg = EpitopeConfig() + policy = SelectionPolicy("frozen-vaxrank", default_score_expr(cfg), score_fill=0., + min_score=cfg.min_epitope_score, duplicates="best") + contexts, expected = {}, [] + for label, part in frame.groupby("source_label", sort=False): + contexts[label] = dict( + default_methods=resolve_default_methods(cfg, part), + default_versions=resolve_default_versions(cfg, part), + alleles=genotype_lookup(dataset.epitopes, keys)) + expected.append(score_predictions(dataset.epitopes, cfg, topiary_df=part)) + evaluation = evaluate_selection_policy(TopiaryResult(frame, metadata=combined.metadata), policy, + group_keys=keys, source_contexts=contexts) + original_scores = pd.concat(expected).sort_index() + actual_scores = evaluation.occurrences.set_index(keys).score.sort_index() + pd.testing.assert_series_equal(actual_scores, original_scores, check_names=False, check_exact=True) + + def transfer_scores(result): + # Match score_predictions: pre-filtered groups have no score entry. + retained = result.occurrences.loc[result.occurrences.filter_retained] + records = retained.set_index(keys).score.to_dict() + return [replace(epitope, per_allele_scores={ + allele: score for (*identity, allele), score in records.items() + if tuple(identity) == epitope.prediction_group_key}) for epitope in dataset.epitopes] + + source_ids = set(frame.loc[frame.source_label.eq("direct"), "prediction_id"]) + variant = Variant("12", 5494381, "A", "G") + native_fragment = MutantProteinFragment( + variant=variant, gene_name=fragment.gene, amino_acids=fragment.sequence, + mutant_amino_acid_start_offset=10, mutant_amino_acid_end_offset=11, + supporting_reference_transcripts=[], n_overlapping_reads=14, n_alt_reads=9, + n_ref_reads=5, n_alt_reads_supporting_protein_sequence=9) + + def construct(epitopes): + selected = [epitope for epitope in epitopes_for_ranking(epitopes, cfg) + if epitope.prediction_id in source_ids] + windows = vaccine_peptides_from_epitopes(variant, native_fragment, selected, vaccine_peptide_length=11) + assert windows + ranked = [(variant, windows)] + peptide = assemble_peptide_constructs(ranked, PeptideConstructConfig( + min_antigen_length_aa=9, max_antigen_length_aa=11)) + mrna = assemble_mrna_constructs(ranked, RNAConstructConfig( + signal_peptide="", include_mitd=False, poly_a_length=0, optimize_linkers=False, + min_antigen_length_aa=9, max_antigen_length_aa=11)) + assert peptide and mrna + return [window.amino_acids for window in windows], peptide, to_native_json(mrna) + + old_epitopes = [] + for label, part in frame.groupby("source_label", sort=False): + ids = set(part.prediction_id) + old_epitopes.extend(attach_per_allele_scores( + [e for e in dataset.epitopes if e.prediction_id in ids], cfg, topiary_df=part)) + original = construct(old_epitopes) + assert construct(transfer_scores(evaluation)) == original + + criterion = SelectionCriterion("late_occurrence", "peptide_offset >= 10", "eligibility") + changed_policy = replace(policy, name="late-context", criteria=(criterion,), + filter_by='criterion("late_occurrence")') + changed = evaluate_selection_policy(TopiaryResult(frame, metadata=combined.metadata), changed_policy, + group_keys=keys, source_contexts=contexts) + changed_epitopes = transfer_scores(changed) + assert construct(changed_epitopes)[0] != original[0] + excluded = changed.audit.loc[changed.audit.status.eq("fail")] + assert not excluded.empty + assert excluded.criterion.eq("late_occurrence").all() + assert excluded.reason.eq("predicate_false").all() + saved = EpitopeDataset(result=changed.evidence, epitopes=tuple(changed_epitopes), config=cfg, + selection={"audit": changed.audit.to_json(orient="records")}) + path = tmp_path / "native-vaxrank.tsv" + saved.save(path) + restored = EpitopeDataset.load(path) + replay = replay_selection_policy(restored.result) + assert restored.selection == saved.selection + pd.testing.assert_frame_equal(replay.occurrences, changed.occurrences, check_exact=True) + assert construct(restored.epitopes) == construct(changed_epitopes) diff --git a/tests/test_policy_evaluation.py b/tests/test_policy_evaluation.py new file mode 100644 index 0000000..5a91a58 --- /dev/null +++ b/tests/test_policy_evaluation.py @@ -0,0 +1,169 @@ +"""Policy evaluation retains occurrences and reuses the public DSL semantics.""" +from dataclasses import replace +import json + +import numpy as np +import pandas as pd +import pytest + +from topiary import ( + EvalContext, SelectionPolicy, TopiaryResult, apply_filter, combine_sources, + evaluate_filter, evaluate_selection_policy, parse, replay_selection_policy, + select_policy_representatives, +) +from .test_candidate_tables import source +from .test_twin_conformance import DELIMITED_IO_TWINS, OCCURRENCE_POLICY_TWINS, FILTER_DECISION_TWINS + + +@pytest.mark.parametrize("filter_by", [None, "n_rna_alt >= 10", "affinity.value < 100", "n_rna_alt > 100"]) +@pytest.mark.parametrize("score_fill,min_score", [(None, None), (0., 1e-5), (0., 0.)]) +def test_occurrence_and_existing_dsl_paths_agree(filter_by, score_fill, min_score): + frame = source(values=(50., np.nan)).df + policy = SelectionPolicy("baseline", "affinity.value", filter_by=filter_by, + score_fill=score_fill, min_score=min_score) + keys = ["source_sequence_name", "peptide", "peptide_offset", "allele"] + evaluate, make_context = OCCURRENCE_POLICY_TWINS + decide_filter, filter_rows = FILTER_DECISION_TWINS + evaluation = evaluate(TopiaryResult(frame), policy, group_keys=keys) + kept = filter_rows(frame, filter_by, group_keys=keys) + direct_ctx = make_context(kept, group_keys=keys) + direct = parse(policy.score_by).eval(direct_ctx).reindex(direct_ctx.group_index) + if score_fill is not None: + direct = direct.fillna(score_fill) + decisions = evaluation.occurrences.set_index(keys) + pd.testing.assert_series_equal(decisions["score"].reindex(direct.index), direct, check_names=False) + expected = direct.index if min_score is None else direct[direct.ge(min_score)].index + assert set(evaluation.selected.set_index(keys).index) == set(expected) + pd.testing.assert_frame_equal(evaluation.evidence.df, frame) + filters = decide_filter(frame, filter_by, group_keys=keys) + ctx = EvalContext(frame, group_keys=keys) + pd.testing.assert_frame_equal(frame[filters.retained.to_numpy()[ctx.row_group_codes()]].reset_index(drop=True), kept) + + +def test_different_flanks_offsets_and_genes_keep_alternatives_and_replay(tmp_path): + first = source(peptides=("SIINFEKL", "SIINFEKL"), values=(50., 50.), + peptide_offset=[0, 20], n_flank=["A", "GG"], gene=["G1", "G2"], + protein_sequence=["SIINFEKL", "M" * 20 + "SIINFEKL"]) + combined = combine_sources({"first": first, "repeat": first}, sample_name="p") + policy = SelectionPolicy("context", "n_rna_alt", duplicates="best") + evaluation = evaluate_selection_policy(combined, policy) + best = select_policy_representatives(evaluation) + assert best.score.tolist() == [15.] + assert len(best.alternative_occurrences.iloc[0]) == 4 + assert len(evaluation.occurrences) == 4 + row = best.evidence_rows.iloc[0][0] + assert combined.df.iloc[row].peptide_offset == 20 + assert combined.df.iloc[row].n_rna_alt == 15 # never 30 from repeated discovery + changed = evaluate_selection_policy(combined, replace(policy, filter_by="gene == 'G1'")) + assert select_policy_representatives(changed).score.tolist() == [5.] + for suffix, writer, _, reader in DELIMITED_IO_TWINS: + path = tmp_path / ("retained." + suffix) + writer(evaluation.evidence, path) + restored = reader(path) + assert len(restored.df) == len(combined.df) + replay = replay_selection_policy(restored) + pd.testing.assert_frame_equal(replay.occurrences, evaluation.occurrences, check_exact=True) + pd.testing.assert_frame_equal(select_policy_representatives(replay), best, check_exact=True) + + +def test_sparse_genotypes_materialize_once_and_project_without_borrowing(tmp_path): + frame = source(kind="proteasome_cleavage", allele="", values=(.2, .8), score=[.2, .8]).df + frame["prediction_id"] = ["one", "two"] + keys = ["prediction_id", "peptide", "peptide_offset", "allele"] + calls = [] + def genotype(row): + calls.append(row) + return ["HLA-A*02:01"] if row["prediction_id"] == "one" else ["HLA-B*07:02"] + policy = SelectionPolicy("processing", "peptide_view(proteasome_cleavage.score)") + evaluation = evaluate_selection_policy(TopiaryResult(frame), policy, group_keys=keys, alleles=genotype) + assert len(calls) == 2 + projected = evaluation.occurrences.query("allele != ''") + assert projected[["prediction_id", "allele", "score"]].values.tolist() == [ + ["one", "HLA-A*02:01", .2], ["two", "HLA-B*07:02", .8]] + assert projected.evidence_rows.tolist() == [[], []] + assert projected.supporting_rows.tolist() == [[0], [1]] + assert len(evaluation.evidence.df) == 2 + replay = replay_selection_policy(evaluation.evidence) + assert len(calls) == 2 + pd.testing.assert_frame_equal(replay.occurrences, evaluation.occurrences) + assert json.dumps(evaluation.evidence.extra, allow_nan=False) + + +def test_source_local_models_versions_used_for_filter_and_score(tmp_path): + frames = pd.concat([source(values=(50., 500.)).df, + source(values=(500., 50.), predictor_version="2").df, + source(values=(20., 800.), prediction_method_name="other").df], ignore_index=True) + evidence = combine_sources({"one": TopiaryResult(frames), "two": TopiaryResult(frames)}, sample_name="p") + policy = SelectionPolicy("local", "affinity.value", filter_by="affinity.value < 100", ascending=True) + contexts = { + "one": dict(default_methods={"affinity": "original"}, default_versions={("affinity", "original"): "2"}), + "two": dict(default_methods={"affinity": "other"}, default_versions={("affinity", "other"): "1"}), + } + evaluation = evaluate_selection_policy(evidence, policy, source_contexts=contexts) + assert evaluation.selected[["source_label", "peptide", "score"]].values.tolist() == [ + ["one", "GILGFVFTL", 50.], ["two", "SIINFEKL", 20.]] + for suffix, writer, _, reader in DELIMITED_IO_TWINS: + path = tmp_path / ("models." + suffix) + writer(evaluation.evidence, path) + pd.testing.assert_frame_equal(replay_selection_policy(reader(path)).occurrences, + evaluation.occurrences, check_exact=True) + + +def test_v1_policy_preserves_definition_and_digest(): + original = SelectionPolicy("v1", "affinity.value").to_dict() + original["schema_version"] = 1 + for field in ("score_fill", "min_score", "criteria", "ranking_by", "unknown", "expanded"): + del original[field] + restored = SelectionPolicy.from_dict(original) + assert restored.to_dict() == original + assert restored.score_fill is None and restored.min_score is None + + +def test_empty_evidence_and_invalid_runtime_context(): + evidence = combine_sources({}, sample_name="p") + evaluation = evaluate_selection_policy(evidence, SelectionPolicy("empty", "affinity.value")) + assert evaluation.selected.empty + assert replay_selection_policy(evaluation.evidence).occurrences.empty + with pytest.raises(ValueError, match="every source"): + evaluate_selection_policy(combine_sources({"one": source()}, sample_name="p"), + evaluation.policy, source_contexts={}) + + +def test_context_derivation_accepts_callbacks_and_rebuilds_changed_genotypes(): + frame = source(allele="", kind="proteasome_cleavage").df + callback = lambda keys: ["HLA-A*02:01"] + ctx = EvalContext(frame, alleles=callback) + expected = ctx.group_index + derived = ctx.derive(filter_context=True) + assert derived.group_index.equals(expected) + apply_filter(frame, "n_rna_alt >= 0", context=ctx) + first = {"SIINFEKL": ["HLA-A*02:01"]} + ctx = EvalContext(frame, alleles=first) + assert "HLA-A*02:01" in ctx.group_index.get_level_values("allele") + changed = ctx.derive(alleles={"SIINFEKL": ["HLA-B*07:02"]}) + assert "HLA-A*02:01" not in changed.group_index.get_level_values("allele") + assert "HLA-B*07:02" in changed.group_index.get_level_values("allele") + + +def test_projected_combined_candidates_use_the_same_public_identity(): + from topiary import candidate_identifier + combined = combine_sources({"processing": source(allele="", kind="proteasome_cleavage")}, sample_name="p") + evaluation = evaluate_selection_policy(combined, SelectionPolicy("projection", "peptide_view(proteasome_cleavage.score)"), + alleles=["HLA-A*02:01"]) + representatives = select_policy_representatives(evaluation) + assert len(representatives) == 2 + assert set(representatives.candidate_id) == { + candidate_identifier("p", peptide, "A0201") for peptide in combined.df.peptide} + assert representatives.source_label.eq("processing").all() + + +@pytest.mark.parametrize("expression", ["(n_rna_alt + 3).sqrt()", "n_rna_alt / (2 * 3)", + "n_rna_alt - (2 - 3)", "(1 + 2).logistic(2, 1)", + "affinity.value < 100"]) +def test_saved_expansion_preserves_original_arithmetic_and_legacy_boolean_scores(expression): + from topiary import evaluate_scores + frame = source().df + policy = SelectionPolicy("grouping", expression) + actual = evaluate_selection_policy(TopiaryResult(frame), policy).occurrences.score + expected = evaluate_scores(frame, expression) + pd.testing.assert_series_equal(actual, expected, check_names=False, check_exact=True) diff --git a/tests/test_selection_criteria.py b/tests/test_selection_criteria.py new file mode 100644 index 0000000..8126959 --- /dev/null +++ b/tests/test_selection_criteria.py @@ -0,0 +1,144 @@ +"""Named criteria compose explicitly and preserve evidence states in audits.""" +from dataclasses import replace + +import numpy as np +import pandas as pd +import pytest + +from topiary import ( + EvalContext, RankingTerm, SelectionCriterion, SelectionPolicy, TopiaryResult, + evaluate_selection_policy, evaluate_selection_criteria, parse, + replay_selection_policy, resolve_selection_policy, select_policy_representatives, +) +from .test_candidate_tables import source +from .test_twin_conformance import DELIMITED_IO_TWINS + + +BINDING = SelectionCriterion("binding", "affinity.value < 500", "eligibility") +PROCESSING = SelectionCriterion("processing", "peptide_view(proteasome_cleavage.score)", "score") + + +def test_reusable_binding_and_opt_in_processing_change_only_referencing_policy(tmp_path): + frame = pd.concat([ + source(values=(50., 100.)).df, + source(kind="proteasome_cleavage", allele="", score=[.1, .9]).df, + ], ignore_index=True) + policy = SelectionPolicy("binding-v1", "1 / affinity.value", filter_by='criterion("binding")', + criteria=(BINDING, PROCESSING), strata=()) + baseline = evaluate_selection_policy(TopiaryResult(frame), policy) + changed = evaluate_selection_policy(TopiaryResult(frame), replace( + policy, name="processing-v2", score_by='criterion("processing")')) + assert select_policy_representatives(baseline, candidate_keys=["peptide", "allele"]).peptide.iloc[0] == "SIINFEKL" + assert select_policy_representatives(changed, candidate_keys=["peptide", "allele"]).peptide.iloc[0] == "GILGFVFTL" + assert baseline.audit.query("criterion == 'processing'").status.eq("not_evaluated").all() + assert changed.audit.query("criterion == 'processing'").status.eq("value").all() + for suffix, writer, _, reader in DELIMITED_IO_TWINS: + path = tmp_path / ("criteria." + suffix) + writer(changed.evidence, path) + restored = replay_selection_policy(reader(path)) + pd.testing.assert_frame_equal(restored.occurrences, changed.occurrences, check_exact=True) + assert restored.policy.sha256 == changed.policy.sha256 + + +@pytest.mark.parametrize("unknown,expected", [("exclude", [0]), ("include", [0, 2, 3])]) +def test_pass_fail_unknown_zero_and_not_applicable_remain_distinct(unknown, expected): + frame = pd.concat([source().df.iloc[[0]]] * 4, ignore_index=True) + frame["prediction_id"] = range(4) + frame["measurement"] = [0., -1., np.nan, 1.] + frame["applicable"] = [1, 1, 1, 0] + criterion = SelectionCriterion("measurement", "measurement >= 0", "eligibility", "applicable == 1") + score = SelectionCriterion("observed", "measurement", "score") + policy = SelectionPolicy("states", 'criterion("observed")', filter_by='criterion("measurement")', + criteria=(criterion, score), unknown=unknown) + keys = ["prediction_id", "peptide", "allele"] + evaluation = evaluate_selection_policy(TopiaryResult(frame), policy, group_keys=keys) + audit = evaluation.audit + assert audit.query("criterion == 'measurement'").status.tolist() == ["pass", "fail", "unknown", "not_applicable"] + assert audit.query("criterion == 'measurement'").reason.tolist() == [ + "predicate_true", "predicate_false", "missing_evidence", "applicability_false"] + observed = audit.query("criterion == 'observed'") + assert observed.iloc[0].status == "value" and observed.iloc[0].reason == "observed_zero" + assert observed.iloc[1].status == "not_evaluated" and observed.iloc[1].reason == "filtered_before_scoring" + assert evaluation.selected.prediction_id.tolist() == expected + assert len(evaluation.evidence.df) == 4 + with pytest.raises(ValueError, match="Unknown eligibility"): + evaluate_selection_policy(TopiaryResult(frame), replace(policy, unknown="error"), group_keys=keys) + + +@pytest.mark.parametrize("expression,reason", [ + ("not_supplied >= 0", "missing_column"), + ("affinity.value >= 0", "missing_evidence"), + ("(-1).sqrt() >= 0", "out_of_domain"), +]) +def test_unknown_diagnostics(expression, reason): + frame = source(values=(np.nan, np.nan)).df + policy = SelectionPolicy("unknown", "1", filter_by='criterion("check")', + criteria=(SelectionCriterion("check", expression, "eligibility"),)) + evaluation = evaluate_selection_policy(TopiaryResult(frame), policy) + assert evaluation.audit.status.eq("unknown").all() + assert evaluation.audit.reason.eq(reason).all() + assert evaluation.selected.empty + + +def test_ambiguity_and_conflicting_measurements_are_not_observed_false(): + for frame, reason in ( + (pd.concat([source().df, source(prediction_method_name="other").df]), "ambiguous_model"), + (pd.concat([source().df, source(values=(40., 400.)).df]), "conflicting_evidence"), + ): + # Numeric criteria do not inherit directional filter auto-aggregation. + policy = SelectionPolicy("audit", 'criterion("binding_value")', + criteria=(SelectionCriterion("binding_value", "affinity.value", "score"),)) + evaluation = evaluate_selection_policy(TopiaryResult(frame), policy) + assert evaluation.audit.status.eq("unknown").all() + assert evaluation.audit.reason.eq(reason).all() + + +def test_transitive_and_or_criteria_and_ordered_ties(): + criteria = (BINDING, + SelectionCriterion("rna", "n_rna_alt >= 10", "eligibility"), + SelectionCriterion("both", 'criterion("binding") & criterion("rna")', "eligibility"), + SelectionCriterion("reads", "n_rna_alt", "ranking")) + policy = SelectionPolicy("compose", "1", filter_by='criterion("binding") | criterion("rna")', + criteria=criteria, ranking_by=(RankingTerm('criterion("reads")'),), + strata=(), duplicates="best") + frame = source(peptides=("SIINFEKL", "SIINFEKL"), values=(50., 100.), peptide_offset=[0, 20]).df + evaluation = evaluate_selection_policy(TopiaryResult(frame), policy) + best = select_policy_representatives(evaluation, candidate_keys=["peptide", "allele"]) + assert best.peptide_offset.tolist() == [20] + assert len(best.alternative_occurrences.iloc[0]) == 2 + assert evaluation.audit.query("criterion == 'both'").status.eq("not_evaluated").all() + both = evaluate_selection_policy(TopiaryResult(frame), replace(policy, filter_by='criterion("both")')) + assert both.selected.peptide_offset.tolist() == [20] + resolved = resolve_selection_policy({k: v for k, v in policy.to_dict().items() if k != "expanded"}) + assert resolved.sha256 == policy.sha256 + saved = policy.to_dict() + saved["expanded"]["score_by"]["expression"] = "999" + with pytest.raises(ValueError, match="expanded"): + SelectionPolicy.from_dict(saved) + + +@pytest.mark.parametrize("criteria,expression,role", [ + ((BINDING, BINDING), 'criterion("binding")', "eligibility"), + ((BINDING,), 'criterion("missing")', "eligibility"), + ((BINDING,), 'criterion("binding")', "score"), + ((SelectionCriterion("a", 'criterion("b")', "score"), + SelectionCriterion("b", 'criterion("a")', "score")), 'criterion("a")', "score"), + ((SelectionCriterion("a", "affinity.value < 500", "score"),), 'criterion("a")', "score"), +]) +def test_duplicate_unresolved_cyclic_and_wrong_role_references_rejected(criteria, expression, role): + from topiary import resolve_selection_expression + with pytest.raises(ValueError): + resolve_selection_expression(expression, criteria, role=role) + + +def test_named_references_do_not_shadow_input_columns_and_direct_dsl_stays_unchanged(): + frame = source(values=(np.nan, 50.), binding=[5., 10.]).df + policy = SelectionPolicy("columns", "binding", criteria=(BINDING,)) + result = evaluate_selection_policy(TopiaryResult(frame), policy) + assert result.occurrences.score.tolist() == [5., 10.] + ctx = EvalContext(frame) + predicate = parse("~(affinity.value < 500)") + assert predicate.eval(ctx).tolist() == [True, False] + assert pd.isna(predicate.eval(ctx.derive(preserve_unknown=True)).iloc[0]) + audit = evaluate_selection_criteria(policy, ctx, references=["binding"]) + assert audit.decisions.status.tolist() == ["unknown", "pass"] diff --git a/tests/test_selection_policy.py b/tests/test_selection_policy.py index 4f5ce20..d364d96 100644 --- a/tests/test_selection_policy.py +++ b/tests/test_selection_policy.py @@ -60,7 +60,7 @@ def test_invalid_definition_rejected(settings): @pytest.mark.parametrize("change", [ - {"schema_version": 2}, {"schema_version": True}, {"unexpected": "setting"}, + {"schema_version": 3}, {"schema_version": True}, {"unexpected": "setting"}, {"default_versions": {}}, {"default_versions": [{"kind": "affinity", "method": "original"}]}, {"default_versions": [{"kind": "affinity", "method": "original", "version": "1"}] * 2}, @@ -137,7 +137,7 @@ def test_yaml_composition_resolves_once_and_preserves_null_vs_inheritance(tmp_pa def test_saved_complete_defaults_survive_changed_constructor_defaults(monkeypatch): saved = SelectionPolicy("frozen-v1", "affinity.value").to_dict() - monkeypatch.setattr(SelectionPolicy.__init__, "__defaults__", (None, True, "best", (), None, None)) + monkeypatch.setattr(SelectionPolicy.__init__, "__defaults__", (None, True, "best", (), None, None, None, None, (), (), "exclude")) assert SelectionPolicy("new-defaults", "affinity.value").ascending is True for resolve in POLICY_MAPPING_TWINS: assert resolve(saved).to_dict() == saved diff --git a/tests/test_twin_conformance.py b/tests/test_twin_conformance.py index 3f3253f..a5d7207 100644 --- a/tests/test_twin_conformance.py +++ b/tests/test_twin_conformance.py @@ -48,7 +48,8 @@ to_tsv, to_csv, read_tsv, read_csv, read_isovar_hypotheses, combine_sources, rank_candidates, rank_with_policy, evaluate_scores, - SelectionPolicy, resolve_selection_policy, + SelectionPolicy, resolve_selection_policy, evaluate_selection_policy, + EvalContext, evaluate_filter, Affinity, Column, apply_filter, apply_sort, ) from topiary.io_isovar import _check_isovar @@ -148,6 +149,11 @@ def _sv_support_record(support): # or encountered in an already-composed authoring configuration. POLICY_MAPPING_TWINS = (SelectionPolicy.from_dict, resolve_selection_policy) +# Full occurrence decisions must match the public context/filter path, including +# sparse genotype projection. Driven by test_policy_evaluation.py. +OCCURRENCE_POLICY_TWINS = (evaluate_selection_policy, EvalContext) +FILTER_DECISION_TWINS = (evaluate_filter, apply_filter) + # The CLI serializes the same exhaustive SV evidence policy as the public API. # Driven together in test_consumer_workflows, including absent protein rows. SV_INTEREST_REPORT_TWINS = (build_sv_interest_report, sv_interest_main) diff --git a/topiary/__init__.py b/topiary/__init__.py index 03bd7f0..1b1abc7 100644 --- a/topiary/__init__.py +++ b/topiary/__init__.py @@ -2,7 +2,13 @@ from .serialization import normalize_python_types from .report_geometry import map_peptide_intervals, mutation_intervals_from_positions from .annotations import join_annotations -from .candidates import combine_sources, protein_evidence_view, rank_candidates, rescore_candidates +from .candidates import candidate_identifier, combine_sources, protein_evidence_view, rank_candidates, rescore_candidates +from .selection_criteria import SelectionCriterion, RankingTerm, resolve_selection_expression +from .criterion_evaluation import CriterionEvaluation, evaluate_selection_criteria +from .policy_evaluation import ( + PolicyEvaluation, evaluate_selection_policy, replay_selection_policy, + select_policy_representatives, describe_evaluation_context, +) from .selection_policy import ( SelectionPolicy, resolve_selection_policy, read_selection_policy, write_selection_policy, rank_with_policy, @@ -67,6 +73,7 @@ class_i, class_ii, evaluate_scores, + evaluate_filter, geomean, maximum, mean, @@ -195,7 +202,7 @@ encode_amino_acids, ) -__version__ = "5.89.0" +__version__ = "5.90.0" __all__ = [ "normalize_isovar_rna_support", @@ -216,6 +223,18 @@ "union_rna_observations", "protein_evidence_view", "rank_candidates", + "SelectionCriterion", + "RankingTerm", + "resolve_selection_expression", + "CriterionEvaluation", + "evaluate_selection_criteria", + "candidate_identifier", + "PolicyEvaluation", + "evaluate_selection_policy", + "replay_selection_policy", + "select_policy_representatives", + "describe_evaluation_context", + "evaluate_filter", "SelectionPolicy", "resolve_selection_policy", "read_selection_policy", diff --git a/topiary/candidates.py b/topiary/candidates.py index 6d22c4a..3f68d58 100644 --- a/topiary/candidates.py +++ b/topiary/candidates.py @@ -64,6 +64,30 @@ def _allele(value): return mhcgnomes.parse(str(value)).to_string() if is_stated(value) else "" +def candidate_identifier(sample, peptide, allele): + """Return the stable sample/peptide/canonical-MHC candidate identity. + + Parameters + ---------- + sample, peptide, allele : str or None + Explicit sample, peptide and allele. Missing peptide or allele returns + None: supporting evidence alone is not a pMHC candidate. Missing sample + for a stated peptide/allele raises ValueError. + + Returns + ------- + str or None + The identity used by combine_sources and projected policy groups. + Alleles are canonicalized through mhcgnomes. + """ + allele = _allele(allele) + if not allele or not is_stated(peptide): + return None + if not is_stated(sample): + raise ValueError("Candidate identity requires a stated sample") + return _identity([sample, peptide, allele]) + + def combine_sources(sources, *, sample_name=None): """Combine prediction tables without running models or merging evidence. @@ -146,7 +170,7 @@ def combine_sources(sources, *, sample_name=None): for column, source_column in SOURCE_PREDICTION_COLUMNS.items(): frame[source_column] = frame[column] if column in frame else None frame["candidate_id"] = [ - _identity([sample, peptide, allele]) if allele and is_stated(peptide) else None + candidate_identifier(sample, peptide, allele) for sample, peptide, allele in zip(samples, frame.peptide, frame.candidate_allele) ] # Differing annotations/abundance are separate source observations, diff --git a/topiary/criterion_evaluation.py b/topiary/criterion_evaluation.py new file mode 100644 index 0000000..8a9fb2a --- /dev/null +++ b/topiary/criterion_evaluation.py @@ -0,0 +1,165 @@ +"""Named-criterion measurements and reasons, evaluated by the Topiary DSL.""" + +from dataclasses import dataclass + +import numpy as np +import pandas as pd + +from .ranking import DSLNode, PeptideView, parse +from .ranking.apply import _check_boolean_like, _collect_column_names, _collect_kinds +from .selection_criteria import _References + + +class _MeasuredCriterion(DSLNode): + def __init__(self, values, node): + self.values = values + self.prediction_kinds = _collect_kinds(node) + + def child_nodes(self): + return [] + + def eval(self, ctx): + return self.values.reindex(ctx.group_index) + + +@dataclass +class CriterionEvaluation: + """Named measurements and group-indexed audit records. + + ``values`` maps criterion names to nullable per-group series, and + ``decisions`` contains group keys, criterion identity, role, status, value, + reason and detail. Use ``expression`` to compose the measured terms through + the existing DSL without introducing input columns or evaluating predictors. + """ + + values: dict + decisions: pd.DataFrame + definitions: dict + + def expression(self, expression): + """Resolve an explicit DSL expression against these measured criteria.""" + return parse(expression, criteria={name: _MeasuredCriterion(value, self.definitions[name]) + for name, value in self.values.items()}) + + +def evaluate_selection_criteria(policy, context, *, references): + """Measure referenced criteria and retain distinct evidence states. + + Parameters + ---------- + policy : SelectionPolicy + Complete definitions, already validated for roles and cycles. + context : EvalContext + Context on this evaluation stage's evidence. Predicate evaluation uses + nullable boolean semantics; the caller decides whether unknown excludes. + references : sequence of str + Criteria used in this stage. Dependencies are evaluated recursively. + Unreferenced definitions are reported as not_evaluated, not failures. + + Returns + ------- + CriterionEvaluation + Values plus per-group records. False applicability is not_applicable; + absent measurements remain unknown; observed numeric zero is a value. + Missing columns, ambiguous/conflicting models and out-of-domain results + have distinct reasons. Unrecognized DSL errors still raise. A model + ambiguity affecting a vector expression marks that expression unknown + for its context; use source_contexts to express independent choices. + Empty contexts return empty records with the same columns. + """ + ctx = context.derive(preserve_unknown=True) + definitions = {item.name: item for item in policy.criteria} + expanded = policy.expanded + nodes = {name: parse(record["expression"]["expression"]) for name, record in expanded["criteria"].items()} + values, records = {}, [] + index = ctx.group_index + + def measured(text): + node = parse(text, criteria=_References(lambda name: _MeasuredCriterion(measure(name), nodes[name]))) + missing = sorted(_collect_column_names(node) - set(ctx.df)) + if missing: + return pd.Series(np.nan, index=index), "missing_column", ", ".join(missing) + try: + with np.errstate(invalid="ignore", divide="ignore", over="ignore"): + result = node.eval(ctx).reindex(index) + except ValueError as error: + message = str(error) + if "Conflicting prediction measurements" in message: + reason = "conflicting_evidence" + elif "Ambiguous" in message: + reason = "ambiguous_model" + elif "predictions" in message and ("No " in message or "not found" in message): + reason = "missing_model" + else: + raise + return pd.Series(np.nan, index=index), reason, message + return result, None, None + + def measure(name): + if name in values: + return values[name] + if name not in definitions: + raise ValueError(f"Unresolved criterion {name!r}") + criterion = definitions[name] + applicable = pd.Series(True, index=index, dtype="boolean") + if criterion.applies_to is not None: + applicable, _, _ = measured(criterion.applies_to) + _check_boolean_like(applicable) + applicable = applicable.astype("boolean") + result, failure, detail = measured(criterion.expression) + if criterion.role == "eligibility": + _check_boolean_like(result) + result = result.astype("boolean") + else: + result = pd.to_numeric(result, errors="raise") + if pd.api.types.is_bool_dtype(result): + raise ValueError(f"Criterion {name!r} requires numeric {criterion.role} values, not booleans") + numeric = pd.to_numeric(result, errors="raise").to_numpy(dtype=float, na_value=np.nan) + nonfinite = ~np.isfinite(numeric) & ~np.isnan(numeric) + # Missing leaf inputs distinguish absent evidence from undefined math. + leaf_missing = pd.Series(False, index=index) + if failure is None and result.isna().any(): + node = nodes[name] + stack = [node] + while stack: + child = stack.pop() + children = [] if isinstance(child, PeptideView) else child.child_nodes() + if children: + stack.extend(children) + else: + try: + leaf_missing |= child.eval(ctx).reindex(index).isna() + except ValueError: + leaf_missing[:] = True + result = result.mask(nonfinite).where(applicable.fillna(False)) + values[name] = result + identities = index.to_frame(index=False).to_dict("records") + for i, identity in enumerate(identities): + value = result.iloc[i] + applies = applicable.iloc[i] + if pd.isna(applies): + status, reason = "unknown", "unknown_applicability" + elif not applies: + status, reason = "not_applicable", "applicability_false" + elif failure: + status, reason = "unknown", failure + elif pd.isna(value): + status = "unknown" + reason = "out_of_domain" if nonfinite[i] or not leaf_missing.iloc[i] else "missing_evidence" + elif criterion.role == "eligibility": + status, reason = ("pass", "predicate_true") if value else ("fail", "predicate_false") + else: + status, reason = "value", "observed_zero" if value == 0 else "observed_value" + records.append(dict(identity, criterion=name, role=criterion.role, value=value, + status=status, reason=reason, detail=detail)) + return result + + for name in references: + measure(name) + for name, criterion in definitions.items(): + if name not in values: + for identity in index.to_frame(index=False).to_dict("records"): + records.append(dict(identity, criterion=name, role=criterion.role, value=None, + status="not_evaluated", reason="not_referenced", detail=None)) + columns = [*ctx.group_keys, "criterion", "role", "value", "status", "reason", "detail"] + return CriterionEvaluation(values, pd.DataFrame(records, columns=columns), nodes) diff --git a/topiary/policy_evaluation.py b/topiary/policy_evaluation.py new file mode 100644 index 0000000..8c727bd --- /dev/null +++ b/topiary/policy_evaluation.py @@ -0,0 +1,407 @@ +"""Occurrence-level policy evaluation with retained evidence and replay context.""" + +from collections.abc import Mapping +from copy import deepcopy +from dataclasses import dataclass +import hashlib +import json + +import numpy as np +import pandas as pd + +from .ranking import EvalContext, as_dsl_node, is_stated +from .ranking.apply import evaluate_filter +from .ranking.nodes import _PeptideAlleleLookup, _peptide_keys +from .result import TopiaryResult +from .selection_policy import SelectionPolicy +from .serialization import normalize_python_types +from .criterion_evaluation import evaluate_selection_criteria + + +_DECISION_COLUMNS = ["occurrence_id", "evidence_rows", "supporting_rows", "filter_value", + "filter_retained", "raw_score", "score", "eligible", "reason"] + + +def _json_value(value): + value = normalize_python_types(value) + if isinstance(value, Mapping): + return {key: _json_value(item) for key, item in value.items()} + if isinstance(value, (list, tuple)): + return [_json_value(item) for item in value] + return None if value is None or pd.isna(value) else value + + +def _key(values): + return json.dumps(_json_value(values), sort_keys=True, separators=(",", ":"), allow_nan=False) + + +def describe_evaluation_context(context): + """Materialize a DSL runtime context for portable occurrence replay. + + Parameters + ---------- + context : EvalContext + Context on the complete source frame. Genotype callbacks are called + once per peptide identity, including peptides later excluded. + + Returns + ------- + dict + JSON-compatible grouping, model defaults, kind support, and allele + declarations. No Python callable is retained. Empty frames yield empty + allele declarations. None and an explicitly empty declaration differ. + Model defaults remain ambiguity-resolution choices, not strict model + requirements; the retained evidence records the actual measurements. + """ + if not isinstance(context, EvalContext): + raise TypeError("context must be an EvalContext") + alleles = context.alleles + declaration = None + if alleles is not None: + keys = _peptide_keys(context.group_keys) + if callable(alleles) or isinstance(alleles, Mapping): + lookup = _PeptideAlleleLookup(alleles, keys) + rows = context.key_frame[keys].drop_duplicates().to_dict("records") + declaration = [dict(keys=row, alleles=list(lookup.for_peptide(row))) for row in rows] + lookup.check_all_used() + else: + # A flat declaration is also valid for an allele-only grouping. + declaration = list(alleles) + return _json_value(dict( + group_keys=list(context.group_keys), default_methods=dict(context.default_methods), + default_versions=[dict(kind=k, method=m, version=v) + for (k, m), v in sorted(context.default_versions.items())], + kind_support=deepcopy(context.kind_support), alleles=declaration, + )) + + +def _context_options(record): + alleles = record["alleles"] + if alleles and isinstance(alleles[0], dict): + lookup = {_key(item["keys"]): item["alleles"] for item in alleles} + alleles = lambda keys: lookup.get(_key(keys), ()) + elif alleles == []: + alleles = lambda keys: () + return dict( + group_keys=record["group_keys"], alleles=alleles, + default_methods=record["default_methods"], kind_support=record["kind_support"], + default_versions={(item["kind"], item["method"]): item["version"] + for item in record["default_versions"]}, + ) + + +@dataclass +class PolicyEvaluation: + """Full source evidence and occurrence decisions from a selection policy. + + ``evidence`` is a TopiaryResult containing every original row, with policy, + runtime context and decisions in ``extra['policy_evaluation']``. Save this + object through Topiary's CSV/TSV writers to retain rejected alternatives. + ``occurrences`` has one row per evaluation group, including projected allele + groups with no direct measurement. Row links are positional in + ``evidence.long_df``; supporting rows are references, never additional + biological observations or additive read support. + """ + + evidence: TopiaryResult + occurrences: pd.DataFrame + + @property + def selected(self): + """Eligible occurrence records, without collapsing candidates.""" + return self.occurrences.loc[self.occurrences.eligible].copy() + + @property + def policy(self): + """The immutable effective policy evaluated against this evidence.""" + return SelectionPolicy.from_dict(self.evidence.extra["policy_evaluation"]["definition"]) + + @property + def audit(self): + """One criterion decision per occurrence, with all identity/row links.""" + records = [] + for row in self.occurrences.to_dict("records"): + criteria = row.pop("criteria", []) + records.extend(dict(row, **criterion) for criterion in criteria) + return pd.DataFrame(records) + + +def _partition_decisions(frame, positions, context, policy, partition): + options = _context_options(context) + ctx = EvalContext(frame, **options) + if set(ctx.group_keys) & set(_DECISION_COLUMNS): + raise ValueError("group_keys collide with reserved policy decision columns") + expanded = policy.expanded + filter_criteria = None + filter_node = policy.filter_by + if policy.criteria: + ctx = ctx.derive(preserve_unknown=True) + filter_refs = [] if expanded["filter_by"] is None else expanded["filter_by"]["references"] + filter_criteria = evaluate_selection_criteria(policy, ctx.derive(filter_context=True), references=filter_refs) + filter_node = None if policy.filter_by is None else filter_criteria.expression(policy.filter_by) + filtered = evaluate_filter(frame, filter_node, context=ctx, unknown=policy.unknown) + retained = filtered.retained.to_numpy() + kept = frame.iloc[np.flatnonzero(retained[ctx.row_group_codes()])] + score_ctx = EvalContext(kept, **options, preserve_unknown=bool(policy.criteria)) + score_criteria = None + score_node = as_dsl_node(expanded["score_by"]["expression"]) + if policy.criteria: + score_refs = list(dict.fromkeys([ + *expanded["score_by"]["references"], + *(name for term in expanded["ranking_by"] for name in term["references"]), + ])) + score_criteria = evaluate_selection_criteria(policy, score_ctx, references=score_refs) + score_node = score_criteria.expression(policy.score_by) + raw = score_node.eval(score_ctx).reindex(score_ctx.group_index) + raw = pd.to_numeric(raw, errors="raise").reindex(ctx.group_index) + effective = raw.fillna(policy.score_fill) if policy.score_fill is not None else raw.copy() + # A removed occurrence must never be restored by score filling. + effective = effective.where(filtered.retained) + groups = ctx.group_index.to_frame(index=False) + tie_scores = [] + for term, resolved in zip(policy.ranking_by, expanded["ranking_by"]): + node = score_criteria.expression(term.expression) if score_criteria else as_dsl_node(resolved["expression"]) + tie_scores.append(node.eval(score_ctx).reindex(ctx.group_index)) + criterion_records = {} + for measurements in (filter_criteria, score_criteria): + if measurements is None: + continue + for record in measurements.decisions.to_dict("records"): + identity = _key({key: record.pop(key) for key in ctx.group_keys}) + slot = criterion_records.setdefault(identity, {}) + if record["status"] != "not_evaluated" or record["criterion"] not in slot: + slot[record["criterion"]] = record + peptide_keys = _peptide_keys(ctx.group_keys) + by_peptide = {} + for row, position in zip(ctx.key_frame.to_dict("records"), positions): + identity = _key([row[k] for k in peptide_keys]) + by_peptide.setdefault(identity, []).append(int(position)) + by_group = {} + for code, position in zip(ctx.row_group_codes(), positions): + by_group.setdefault(int(code), []).append(int(position)) + decisions = [] + for i, identity in enumerate(groups.to_dict("records")): + rows = by_group.get(i, []) + support = by_peptide.get(_key([identity[k] for k in peptide_keys]), []) + score = effective.iloc[i] + eligible = bool(retained[i] and (policy.min_score is None or ( + pd.notna(score) and score >= policy.min_score))) + reason = ("filtered" if not retained[i] else + "below_min_score" if pd.notna(score) and not eligible else + "missing_score" if pd.isna(raw.iloc[i]) else "eligible") + record = dict(identity) + # Preserve source identity/annotations which are constant in a group. + local = frame.loc[rows if rows else support] + for column in ("candidate_id", "candidate_mhc_class", "candidate_sample", "source_observation_id", "source_label"): + if column in local and column not in record: + values = local[column].drop_duplicates() + if len(values) > 1: + raise ValueError(f"Occurrence group collapses distinct {column}; include it in group_keys") + record[column] = values.iloc[0] if len(values) else None + if not rows and is_stated(record.get("candidate_sample")): + from .candidates import candidate_identifier + from .io_pvacseq import derive_mhc_class + record["candidate_id"] = candidate_identifier(record["candidate_sample"], identity.get("peptide"), identity.get("allele")) + record["candidate_mhc_class"] = derive_mhc_class(pd.Series([identity.get("allele")])).iloc[0] + record.update( + occurrence_id=hashlib.sha256(_key([partition, identity]).encode()).hexdigest(), + evidence_rows=rows, supporting_rows=support, + filter_value=filtered.value.iloc[i], filter_retained=bool(retained[i]), + raw_score=raw.iloc[i], score=score, eligible=eligible, reason=reason, + ) + for j, values in enumerate(tie_scores): + record[f"ranking_{j}"] = values.iloc[i] + if policy.criteria: + audit = criterion_records.get(_key(identity), {}) + record["criteria"] = [] + for criterion in policy.criteria: + detail = dict(audit.get(criterion.name, dict( + criterion=criterion.name, role=criterion.role, value=None, + status="not_evaluated", reason="not_referenced", detail=None))) + if detail["status"] == "not_evaluated" and not retained[i] and criterion.name in score_refs: + detail["reason"] = "filtered_before_scoring" + record["criteria"].append(detail) + decisions.append(record) + return decisions + + +def evaluate_selection_policy(result, policy, *, group_keys=None, alleles=None, + kind_support=None, source_contexts=None, provenance=None): + """Evaluate a saved policy before selecting candidate representatives. + + Parameters + ---------- + result : TopiaryResult + Complete long/wide evidence, including enriched source columns. A + combined result is optional when explicit group_keys identify the input. + Empty inputs with their identity schema return empty decisions. + policy : SelectionPolicy + Filter, scoring, fill and post-score minimum. No predictor is invoked. + group_keys : sequence of str, optional + Occurrence identity; for Vaxrank typically prediction_id, peptide, + peptide_offset, allele. None uses EvalContext's inference. + alleles : sequence, mapping or callable, optional + Per-occurrence genotype, with exactly EvalContext's semantics. A + callback is materialized before evaluation and is not saved as code. + kind_support : mapping, optional + Model MHC context. None uses the source result's recorded metadata. + source_contexts : mapping, optional + Source-label to context overrides (default_methods, default_versions, + kind_support, alleles). If supplied, every input source must be named; + each source is evaluated separately. Values replace global defaults, + rather than implicitly merging model choices across sources. + provenance : mapping, optional + JSON-compatible derivation information, outside the policy digest. + + Returns + ------- + PolicyEvaluation + All evidence plus decisions including rejected, missing-score, zero + and projected groups. Raw scores for pre-filtered groups are missing: + they were not scored. Missing scores stay eligible when min_score is + None, matching candidate ranking's explicit unranked state. Set a gate + to exclude them; score_fill is applied before the gate. Raw evidence + and the input object remain unchanged. + """ + from . import __version__ + + if not isinstance(result, TopiaryResult) or not isinstance(policy, SelectionPolicy): + raise TypeError("Expected a TopiaryResult and SelectionPolicy") + if provenance is not None and not isinstance(provenance, Mapping): + raise ValueError("provenance must be a mapping or None") + derivation = json.loads(json.dumps(provenance, allow_nan=False)) + frame = result.long_df.copy().reset_index(drop=True) + if not frame.columns.is_unique: + raise ValueError("Evidence must have distinct column names") + options = dict(group_keys=group_keys, alleles=alleles, + default_methods=policy.default_methods, default_versions=policy.default_versions, + kind_support=result._kind_support() if kind_support is None else kind_support) + if source_contexts is None: + partitions = [(None, np.arange(len(frame)), {})] + else: + if not isinstance(source_contexts, Mapping) or "source_label" not in frame: + raise ValueError("source_contexts requires a mapping and a source_label column") + if not frame.source_label.map(is_stated).all() or set(source_contexts) != set(frame.source_label): + raise ValueError("source_contexts must name every source_label exactly once") + partitions = [(label, np.flatnonzero(frame.source_label.eq(label)), settings) + for label, settings in source_contexts.items()] + executions, decisions = [], [] + for label, positions, overrides in partitions: + if not isinstance(overrides, Mapping) or set(overrides) - { + "default_methods", "default_versions", "kind_support", "alleles"}: + raise ValueError("Unsupported source context settings") + part = frame.iloc[positions] + context = describe_evaluation_context(EvalContext(part, **dict(options, **overrides))) + executions.append(dict(source_label=label, context=context)) + decisions.extend(_partition_decisions(part, positions, context, policy, label)) + occurrences = pd.DataFrame(decisions) + if not decisions: + columns = list(dict.fromkeys([*(group_keys or EvalContext(frame).group_keys), *_DECISION_COLUMNS])) + occurrences = pd.DataFrame(columns=columns).astype({"eligible": bool, "filter_retained": bool}) + extra = deepcopy(result.extra) + extra["policy_evaluation"] = dict( + schema_version=1, definition=policy.to_dict(), sha256=policy.sha256, + provenance=derivation, topiary_version=__version__, + input_topiary_version=result.topiary_version, partitions=executions, + occurrences=_json_value(occurrences.to_dict("records")), + ) + evidence = TopiaryResult(frame, metadata=result.metadata, form="long", extra=extra) + return PolicyEvaluation(evidence, occurrences) + + +def replay_selection_policy(result): + """Re-evaluate a saved occurrence policy using its retained evidence. + + Parameters + ---------- + result : TopiaryResult + Evidence exported from PolicyEvaluation.evidence and optionally read + back through CSV/TSV. A representative-only export is insufficient. + + Returns + ------- + PolicyEvaluation + Fresh decisions using the saved definition and materialized contexts. + Missing/unsupported metadata or a mismatched policy digest raises. + Empty saved evaluations replay as empty. Predictors are never called. + """ + record = result.extra.get("policy_evaluation", {}) + if record.get("schema_version") != 1: + raise ValueError("Missing or unsupported policy_evaluation metadata") + policy = SelectionPolicy.from_dict(record["definition"]) + if policy.sha256 != record["sha256"]: + raise ValueError("Saved selection policy digest does not match its definition") + partitions = record["partitions"] + if len(partitions) == 1 and partitions[0]["source_label"] is None: + options = _context_options(partitions[0]["context"]) + groups = options.pop("group_keys") + # Source-local model choices are already part of this runtime context. + from dataclasses import replace + runtime_policy = replace(policy, default_methods=options.pop("default_methods"), + default_versions=options.pop("default_versions")) + replay = evaluate_selection_policy(result, runtime_policy, group_keys=groups, + provenance=record["provenance"], **options) + replay.evidence.extra["policy_evaluation"].update(definition=policy.to_dict(), sha256=policy.sha256) + return replay + groups = [item["context"]["group_keys"] for item in partitions] + if groups and any(keys != groups[0] for keys in groups): + raise ValueError("Saved source contexts must share group_keys") + contexts = {} + for item in partitions: + context = _context_options(item["context"]) + context.pop("group_keys") + contexts[item["source_label"]] = context + return evaluate_selection_policy(result, policy, group_keys=groups[0] if groups else None, + source_contexts=contexts, provenance=record["provenance"]) + + +def select_policy_representatives(evaluation, *, candidate_keys=("candidate_id",), strata=None): + """Select actual scored occurrences without adding repeated evidence. + + Parameters + ---------- + evaluation : PolicyEvaluation + Complete occurrence decisions. This object is not narrowed or changed. + candidate_keys : sequence of str + Columns identifying candidates in the occurrence table. For a direct + consumer, pass its explicit candidate identity (e.g. peptide, allele). + Rows with absent candidate identity are supporting evidence, not targets. + strata : sequence of str, optional + Independent partitions; None uses the saved policy's strata. + + Returns + ------- + pandas.DataFrame + One eligible occurrence per candidate and stratum, with alternative + occurrence IDs and an explicit selection rationale. Best/worst/error + duplicate handling and score direction come from the saved policy. + Missing scores sort last. Empty input returns an empty selection. + Evidence links identify real input rows; counts are never summed. + """ + policy = evaluation.policy + frame = evaluation.selected + strata = policy.strata if strata is None else strata + if isinstance(candidate_keys, str) or isinstance(strata, str) or not candidate_keys: + raise ValueError("candidate_keys and strata must be column-name sequences") + keys = list(dict.fromkeys([*strata, *candidate_keys])) + if set(keys) - set(frame): + raise ValueError(f"Unknown representative keys: {sorted(set(keys) - set(frame))}") + frame = frame.loc[frame[list(candidate_keys)].map(is_stated).all(axis=1)] + all_alternatives = { + _key(list(identity) if isinstance(identity, tuple) else [identity]): group.occurrence_id.tolist() + for identity, group in evaluation.occurrences.groupby(keys, sort=False, dropna=False) + } + rows = [] + for _, group in frame.groupby(keys, sort=False, dropna=False): + if policy.duplicates == "error" and group.score.nunique(dropna=False) > 1: + raise ValueError("Candidate occurrences have conflicting scores; choose duplicates='best' or 'worst'") + ascending = policy.ascending if policy.duplicates != "worst" else not policy.ascending + columns = ["score", *(f"ranking_{i}" for i in range(len(policy.ranking_by)))] + directions = [ascending, *(term.ascending for term in policy.ranking_by)] + row = group.sort_values(columns, ascending=directions, na_position="last", kind="stable").iloc[0].copy() + row["alternative_occurrences"] = all_alternatives[_key([row[key] for key in keys])] + row["representative_reason"] = (f"{policy.duplicates}; ordered keys {columns}; " + "missing last; stable input order breaks ties") + rows.append(row) + return pd.DataFrame(rows, columns=[*frame.columns, "alternative_occurrences", "representative_reason"]).sort_values( + "score", ascending=policy.ascending, na_position="last", kind="stable").reset_index(drop=True) diff --git a/topiary/ranking/__init__.py b/topiary/ranking/__init__.py index dae28df..d8a5617 100644 --- a/topiary/ranking/__init__.py +++ b/topiary/ranking/__init__.py @@ -88,7 +88,7 @@ shuffled, wt, ) -from .apply import apply_filter, apply_sort, evaluate_scores +from .apply import apply_filter, apply_sort, evaluate_scores, evaluate_filter from .parser import as_dsl_node, as_dsl_nodes, parse __all__ = [ @@ -138,6 +138,7 @@ "apply_filter", "apply_sort", "evaluate_scores", + "evaluate_filter", "parse", "as_dsl_node", "as_dsl_nodes", diff --git a/topiary/ranking/apply.py b/topiary/ranking/apply.py index d29a0ca..7d6217d 100644 --- a/topiary/ranking/apply.py +++ b/topiary/ranking/apply.py @@ -131,6 +131,7 @@ def _collect_kinds(node): kind = getattr(n, "kind", None) if kind is not None: kinds.add(_kind_value(kind)) + kinds.update(getattr(n, "prediction_kinds", ())) stack.extend(n.child_nodes()) return kinds @@ -358,22 +359,61 @@ def apply_filter(df, node, *, group_keys=None, default_methods=None, _check_group_keys(df, group_keys) return df if node is None else df.reset_index(drop=True) - _validate_columns(df, node) ctx = _resolve_context( df, context, filter_context=True, group_keys=group_keys, default_methods=default_methods, kind_support=kind_support, alleles=alleles, default_versions=default_versions, ) - # Reindex defensively so a misbehaving node (index mismatch) surfaces - # as NaN → False rather than silently picking up rows from a - # different MultiIndex alignment. + decisions = evaluate_filter(df, node, context=ctx) + keep = decisions.retained.to_numpy()[ctx.row_group_codes()] + return df[keep].reset_index(drop=True) + + +def evaluate_filter(df, node, *, context=None, group_keys=None, + default_methods=None, default_versions=None, + kind_support=None, alleles=None, unknown="exclude"): + """Evaluate a filter without discarding observations. + + Parameters + ---------- + df : pandas.DataFrame + Long-form evidence, including an empty frame with its identity columns. + node : str or DSLNode or None + Boolean expression; None retains every group. + context : EvalContext, optional + Context built on this frame. Mutually exclusive with the other context + options, which have the same meaning as in :func:`apply_filter`. + group_keys, default_methods, default_versions, kind_support, alleles : optional + Explicit identity, model choices, and per-occurrence allele context. + unknown : {'exclude', 'include', 'error'} + Decision for missing predicate values. To preserve unknown comparisons, + pass a context with preserve_unknown=True. Default retains legacy rules. + + Returns + ------- + pandas.DataFrame + Group-indexed ``value`` and boolean ``retained`` columns. Retention + uses exactly :func:`apply_filter` semantics, including preservation of + allele-free supporting evidence for surviving peptides. Thus a retained + supporting row need not itself have a true filter value. + """ + if unknown not in {"exclude", "include", "error"}: + raise ValueError("unknown must be exclude, include, or error") + ctx = _resolve_context( + df, context, filter_context=True, group_keys=group_keys, + default_methods=default_methods, default_versions=default_versions, + kind_support=kind_support, alleles=alleles, + ) + if node is None or df.empty: + return pd.DataFrame({"value": True, "retained": True}, index=ctx.group_index) + node = as_dsl_node(node) + _validate_columns(df, node) values = node.eval(ctx).reindex(ctx.group_index) _check_boolean_like(values) - mask = values.fillna(False).astype(bool).to_numpy() - mask = _keep_allele_free_evidence(ctx, node, mask) - - keep = mask[ctx.row_group_codes()] - return df[keep].reset_index(drop=True) + if unknown == "error" and values.isna().any(): + raise ValueError("Unknown eligibility evidence under unknown='error'") + mask = _keep_allele_free_evidence(ctx, node, values.fillna(unknown == "include").astype(bool).to_numpy()) + return pd.DataFrame({"value": values, "retained": mask}, index=ctx.group_index) def apply_sort(df, sort_nodes, sort_direction="auto", *, group_keys=None, diff --git a/topiary/ranking/nodes.py b/topiary/ranking/nodes.py index 29e1bcc..bc7c490 100644 --- a/topiary/ranking/nodes.py +++ b/topiary/ranking/nodes.py @@ -1086,18 +1086,22 @@ class EvalContext: for ``>``/``>=``) instead of raising on ambiguity. :func:`apply_filter` sets this to ``True`` automatically; :func:`apply_sort` leaves it ``False`` so sort stays strict. + preserve_unknown : bool, optional + Preserve missing comparisons and categorical predicates as nullable + booleans, with three-valued AND/OR/NOT. Default False keeps historical + filter behavior. Named-policy audits opt in explicitly. """ __slots__ = ( "_source_df", "group_keys", "default_methods", "default_versions", "filter_context", "kind_support", "alleles", "_group_index", "_key_frame", "_group_tuples_cache", - "_group_codes_cache", "_df", "_method_override", + "_group_codes_cache", "_df", "_method_override", "preserve_unknown", ) def __init__( self, df, group_keys=None, default_methods=None, filter_context=False, - kind_support=None, alleles=None, default_versions=None, + kind_support=None, alleles=None, default_versions=None, preserve_unknown=False, ): self._source_df = df if group_keys is None: @@ -1112,6 +1116,7 @@ def __init__( if default_versions else {} ) self.filter_context = filter_context + self.preserve_unknown = preserve_unknown # mhctools >=3.13.7 per-(model, kind) metadata. Optional; when # provided (typically from ``TopiaryPredictor.kind_support``), # nodes that care about allele dependence (e.g. @@ -1147,7 +1152,7 @@ def derive(self, **overrides) -> "EvalContext": """ unknown = set(overrides) - { "df", "group_keys", "default_methods", "default_versions", - "filter_context", "kind_support", "alleles", + "filter_context", "kind_support", "alleles", "preserve_unknown", } if unknown: raise TypeError( @@ -1170,11 +1175,12 @@ def derive(self, **overrides) -> "EvalContext": ), kind_support=overrides.get("kind_support", self.kind_support), alleles=alleles, + preserve_unknown=overrides.get("preserve_unknown", self.preserve_unknown), ) reshaped = ( df is not self._source_df or list(group_keys) != list(self.group_keys) - or list(alleles or ()) != list(self.alleles or ()) + or alleles is not self.alleles ) if not reshaped: derived._key_frame = self._key_frame @@ -1792,7 +1798,12 @@ def eval(self, ctx: EvalContext) -> pd.Series: vals = member.groupby( [df[k] for k in ctx.group_keys], sort=False, dropna=False ).any() - return vals.reindex(ctx.group_index).fillna(False).astype("boolean") + result = vals.reindex(ctx.group_index).fillna(False).astype("boolean") + if ctx.preserve_unknown: + known = df[self.col_name].notna().groupby( + [df[k] for k in ctx.group_keys], sort=False, dropna=False).any() + result = result.where(known.reindex(ctx.group_index, fill_value=False)) + return result def __repr__(self): prefix = "~" if self.negate else "" @@ -1860,6 +1871,8 @@ def eval(self, ctx: EvalContext) -> pd.Series: mask = vals.isin(self.values) if self.negate: mask = ~mask + if ctx.preserve_unknown: + mask = mask.astype("boolean").where(vals.notna()) return mask def __invert__(self): @@ -3563,7 +3576,8 @@ def eval(self, ctx: EvalContext) -> pd.Series: b = self.right.eval(ctx) # pandas comparison returns False for NaN comparisons — matches # the intended "missing values fail the filter" behavior. - return self.op(a, b) + result = self.op(a, b) + return result.astype("boolean").where(a.notna() & b.notna()) if ctx.preserve_unknown else result def _should_auto_aggregate(self, ctx): """Gate check for the narrow auto-aggregation scope (issue #118). @@ -3645,7 +3659,8 @@ def _auto_aggregate(self, ctx): left_agg = left_df.max(axis=1, skipna=True) right_agg = right_df.min(axis=1, skipna=True) - return self.op(left_agg, right_agg) + result = self.op(left_agg, right_agg) + return result.astype("boolean").where(left_agg.notna() & right_agg.notna()) if ctx.preserve_unknown else result def __repr__(self): sym = _CMP_SYMBOLS.get(self.op, "?") @@ -3694,9 +3709,10 @@ def eval(self, ctx: EvalContext) -> pd.Series: # Policy: NaN is treated as False. Naive `astype(bool)` coerces # NaN / None to True (any object is truthy), so we explicitly # map NaN → False per-dtype before applying the boolean op. + convert = (lambda value: value.astype("boolean")) if ctx.preserve_unknown else _as_bool_series if self.op is operator.invert: - return ~_as_bool_series(self.children[0].eval(ctx)) - values = [_as_bool_series(c.eval(ctx)) for c in self.children] + return ~convert(self.children[0].eval(ctx)) + values = [convert(c.eval(ctx)) for c in self.children] if self.op is operator.and_: result = values[0] for v in values[1:]: diff --git a/topiary/ranking/parser.py b/topiary/ranking/parser.py index d3ee97c..1934bca 100644 --- a/topiary/ranking/parser.py +++ b/topiary/ranking/parser.py @@ -261,9 +261,10 @@ def _as_bool_node(node): class _Parser: """Recursive-descent parser for the full DSL (arithmetic + booleans).""" - def __init__(self, text): + def __init__(self, text, criteria=None): self.tokenizer = _Tokenizer(text) self.text = text + self.criteria = criteria def parse(self) -> DSLNode: node = self._or() @@ -426,6 +427,17 @@ def _atom(self): self.tokenizer.advance() args = self._call_args() return _AGGREGATION_FUNCS[name](*args) + if name == "criterion" and self.tokenizer.peek_at(1)[0] == "LPAREN": + self.tokenizer.advance() + self.tokenizer.expect("LPAREN") + identifier = self.tokenizer.expect("STRING")[1] + self.tokenizer.expect("RPAREN") + if self.criteria is None: + raise ValueError(f"Unresolved criterion {identifier!r}; supply named criteria") + try: + return _parser_as_node(self.criteria[identifier]) + except KeyError: + raise ValueError(f"Unresolved criterion {identifier!r}") from None if name == "peptide_view": self.tokenizer.advance() self.tokenizer.expect("LPAREN") @@ -803,13 +815,17 @@ def _prediction_kind_for_name(self, name): return KIND_ALIASES.get(name.strip().lower()) -def parse(text: str) -> DSLNode: +def parse(text: str, *, criteria=None) -> DSLNode: """Parse a DSL string into a :class:`DSLNode`. Supports the full grammar: arithmetic, comparisons, boolean combinators, transforms, aggregations, scoped fields. + ``criteria`` optionally maps stable names to already-resolved DSL nodes + for explicit ``criterion("name")`` references. Bare identifiers continue + to name input columns. Unknown references raise ValueError. No registry is + consulted and no Python expression is evaluated. """ - return _Parser(text).parse() + return _Parser(text, criteria=criteria).parse() def as_dsl_node(expression) -> DSLNode: diff --git a/topiary/selection_criteria.py b/topiary/selection_criteria.py new file mode 100644 index 0000000..dc29500 --- /dev/null +++ b/topiary/selection_criteria.py @@ -0,0 +1,165 @@ +"""Named criteria resolve into the existing DSL without a mutable registry.""" + +from collections.abc import Mapping +from dataclasses import dataclass +import re + +from .ranking import BoolOp, Comparison, Includes, IsIn, parse +from .ranking.parser import _Tokenizer + + +@dataclass(frozen=True) +class SelectionCriterion: + """A reusable, explicitly scoped criterion. + + Parameters + ---------- + name : str + Stable identifier, referenced only by ``criterion("name")``. A same-named + input column continues to be a column, so references cannot shadow data. + expression : str + Existing Topiary DSL expression, optionally referencing other criteria. + role : {'eligibility', 'score', 'ranking'} + Predicate, numeric score term, or numeric ordered tie-break expression. + Eligibility references may compose only predicates; score references + only score terms; ranking may reuse score or ranking terms. + applies_to : str, optional + Eligibility expression defining applicability. False means not + applicable; missing means applicability unknown. None applies to all + groups. Non-applicable references remain unknown, never observed false. + """ + + name: str + expression: str + role: str + applies_to: str | None = None + + def __post_init__(self): + for label in ("name", "expression", "role", "applies_to"): + value = getattr(self, label) + if label == "applies_to" and value is None: + continue + if not isinstance(value, str) or not value.strip(): + raise ValueError(f"Criterion {label} must be a nonempty string") + if self.role not in {"eligibility", "score", "ranking"}: + raise ValueError("Criterion role must be eligibility, score, or ranking") + if not re.fullmatch(r"[A-Za-z][A-Za-z0-9_.-]*", self.name): + raise ValueError("Criterion names must be stable identifiers: letters, digits, dots, underscores or hyphens") + + def to_dict(self): + """Return the complete portable definition, including applicability.""" + return dict(name=self.name, expression=self.expression, role=self.role, applies_to=self.applies_to) + + +@dataclass(frozen=True) +class RankingTerm: + """One ordered numeric tie-break expression and explicit direction. + + ``expression`` is a DSL string, optionally containing named score/ranking + references. ``ascending`` must be a bool. Terms follow the primary policy + score in the supplied order, missing values last, with stable ties. + """ + + expression: str + ascending: bool = False + + def __post_init__(self): + if not isinstance(self.expression, str) or not self.expression.strip(): + raise ValueError("Ranking expression must be a nonempty string") + if type(self.ascending) is not bool: + raise ValueError("Ranking direction must be a boolean") + + def to_dict(self): + """Return a portable expression/direction record.""" + return dict(expression=self.expression, ascending=self.ascending) + + +class _References(Mapping): + def __init__(self, lookup): + self.lookup = lookup + + def __getitem__(self, name): + return self.lookup(name) + + def __iter__(self): + return iter(()) + + def __len__(self): + return 0 + + +def resolve_selection_expression(expression, criteria=(), *, role="score"): + """Expand named references, validating dependencies and output roles. + + Parameters + ---------- + expression : str + Explicit DSL arithmetic/AND/OR expression; no combination is inferred. + criteria : sequence of SelectionCriterion + Complete definitions. Empty is valid for a direct expression. Duplicate + names, unknown references, cyclic dependencies, and wrong roles raise. + role : {'eligibility', 'score', 'ranking'} + Expected output role. The numeric/boolean runtime type is also checked + when a policy is evaluated, since input columns have no static type. + + Returns + ------- + dict + Expanded DSL ``expression`` and ordered ``references`` (including + transitive dependencies). Resolution uses only supplied definitions; + no registry or input column is consulted. Applicability dependencies + are validated and included even though the expanded expression itself + describes only the measurement, not its applicability mask. + """ + definitions = {} + for criterion in criteria: + if not isinstance(criterion, SelectionCriterion): + raise ValueError("criteria must contain SelectionCriterion definitions") + if criterion.name in definitions: + raise ValueError(f"Duplicate criterion identifier {criterion.name!r}") + definitions[criterion.name] = criterion + if role not in {"eligibility", "score", "ranking"}: + raise ValueError("Unknown expression role") + visited, active = [], [] + + def expand(text, expected): + expanded_references = {} + def lookup(name): + if name not in definitions: + raise ValueError(f"Unresolved criterion {name!r}") + criterion = definitions[name] + allowed = {"ranking", "score"} if expected == "ranking" else {expected} + if criterion.role not in allowed: + raise ValueError(f"Criterion {name!r} has role {criterion.role}, expected {expected}") + if name in active: + raise ValueError(f"Cyclic criterion references: {' -> '.join([*active, name])}") + if name not in visited: + visited.append(name) + active.append(name) + node, expanded = expand(criterion.expression, criterion.role) + expanded_references[name] = expanded + if criterion.applies_to is not None: + expand(criterion.applies_to, "eligibility") + active.pop() + return node + node = parse(text, criteria=_References(lookup)) + boolean = isinstance(node, (Comparison, BoolOp, IsIn, Includes)) + if expected != "eligibility" and boolean and active: + raise ValueError(f"Expected numeric {expected} expression, got a predicate") + # Preserve source parentheses and numeric literals. repr(node) is a + # human display, not a lossless serialization of arithmetic grouping. + tokens = _Tokenizer(text).tokens[:-1] + rendered, i = [], 0 + while i < len(tokens): + if (tokens[i][0] == "IDENT" and tokens[i][1].lower() == "criterion" + and i + 3 < len(tokens) and tokens[i + 1][0] == "LPAREN"): + rendered.append("(" + expanded_references[tokens[i + 2][1]] + ")") + i += 4 + else: + kind, value = tokens[i] + rendered.append(repr(value) if kind == "STRING" else value) + i += 1 + return node, " ".join(rendered) + + _, expanded = expand(expression, role) + return dict(expression=expanded, references=visited) diff --git a/topiary/selection_policy.py b/topiary/selection_policy.py index df4ea33..aa6fc29 100644 --- a/topiary/selection_policy.py +++ b/topiary/selection_policy.py @@ -2,17 +2,20 @@ from collections.abc import Mapping from copy import deepcopy -from dataclasses import dataclass +from dataclasses import dataclass, field import hashlib import json +import math from pathlib import Path from types import MappingProxyType -from .ranking import as_dsl_node +from .selection_criteria import SelectionCriterion, RankingTerm, resolve_selection_expression -_POLICY_FIELDS = {"schema_version", "name", "score_by", "filter_by", "ascending", +_V1_FIELDS = {"schema_version", "name", "score_by", "filter_by", "ascending", "duplicates", "strata", "default_methods", "default_versions"} +_POLICY_FIELDS = _V1_FIELDS | {"score_fill", "min_score", "criteria", "ranking_by", "unknown", + "expanded"} def _text(value, label): @@ -52,6 +55,21 @@ class SelectionPolicy: default_versions : mapping of (str, str) to str, optional Explicit (kind, method)-to-version selections. Unknown source versions remain unknown. JSON represents these tuple keys as records. + score_fill : float, optional + Replacement for missing occurrence scores, after evaluation. None + preserves missingness. Raw scores always remain in the audit view. + min_score : float, optional + Inclusive minimum on the effective occurrence score. This gate runs + after scoring, independently of the pre-filter; None adds no gate. + criteria : sequence of SelectionCriterion + Embedded named criteria. Explicit criterion("name") references reuse + them; unused definitions are audited as not evaluated. + ranking_by : sequence of RankingTerm + Ordered tie-break expressions following the primary score. + unknown : {'exclude', 'include', 'error'} + Decision on unknown named eligibility evidence. Its audit value remains + unknown under every decision. Direct expression policies retain legacy + filter semantics when criteria is empty. Notes ----- @@ -70,17 +88,37 @@ class SelectionPolicy: strata: tuple[str, ...] = ("candidate_mhc_class",) default_methods: Mapping | None = None default_versions: Mapping | None = None + score_fill: float | None = None + min_score: float | None = None + criteria: tuple = () + ranking_by: tuple = () + unknown: str = "exclude" + _schema_version: int = field(default=2, init=False, repr=False, compare=False) def __post_init__(self): _text(self.name, "name") + for label, expected in (("criteria", SelectionCriterion), ("ranking_by", RankingTerm)): + supplied = getattr(self, label) + if not isinstance(supplied, (list, tuple)) or any(not isinstance(item, expected) for item in supplied): + raise ValueError(f"{label} must be a sequence of {expected.__name__}") + object.__setattr__(self, label, tuple(supplied)) + if self.unknown not in {"exclude", "include", "error"}: + raise ValueError("unknown must be exclude, include, or error") for label in ("score_by", "filter_by"): value = getattr(self, label) if value is None and label == "filter_by": continue try: - as_dsl_node(_text(value, label)) + resolve_selection_expression(_text(value, label), self.criteria, + role="eligibility" if label == "filter_by" else "score") except (ValueError, SyntaxError) as error: raise ValueError(f"{label}: {error}") from error + # Validate every definition, including unused ones, so a saved recipe + # cannot carry a hidden unresolved dependency or cycle. + for criterion in self.criteria: + resolve_selection_expression(f"criterion({criterion.name!r})", self.criteria, role=criterion.role) + for term in self.ranking_by: + resolve_selection_expression(term.expression, self.criteria, role="ranking") if type(self.ascending) is not bool: raise ValueError("ascending must be a boolean") if not isinstance(self.duplicates, str) or self.duplicates not in {"error", "best", "worst"}: @@ -91,6 +129,12 @@ def __post_init__(self): if len(set(strata)) != len(strata): raise ValueError("strata must name distinct columns") object.__setattr__(self, "strata", strata) + for label in ("score_fill", "min_score"): + value = getattr(self, label) + if value is not None: + if isinstance(value, bool) or not isinstance(value, (float, int)) or not math.isfinite(value): + raise ValueError(f"{label} must be a finite number or None") + object.__setattr__(self, label, float(value)) for label in ("default_methods", "default_versions"): supplied = getattr(self, label) if supplied is None: @@ -110,7 +154,7 @@ def __post_init__(self): object.__setattr__(self, label, MappingProxyType(copied)) def to_dict(self): - """Return an independent, JSON-compatible schema-version-1 definition. + """Return an independent, JSON-compatible versioned definition. Explicit None selections remain None; empty mappings remain empty. Every setting is written, including defaults, so a future constructor @@ -122,8 +166,8 @@ def to_dict(self): Complete definition with string keys, suitable for JSON or YAML. """ versions = self.default_versions - return dict( - schema_version=1, name=self.name, score_by=self.score_by, + definition = dict( + schema_version=self._schema_version, name=self.name, score_by=self.score_by, filter_by=self.filter_by, ascending=self.ascending, duplicates=self.duplicates, strata=list(self.strata), default_methods=None if self.default_methods is None else dict(sorted(self.default_methods.items())), @@ -131,12 +175,31 @@ def to_dict(self): dict(kind=kind, method=method, version=version) for (kind, method), version in sorted(versions.items())], ) + if self._schema_version >= 2: + definition.update(score_fill=self.score_fill, min_score=self.min_score, + criteria=[item.to_dict() for item in self.criteria], + ranking_by=[item.to_dict() for item in self.ranking_by], + unknown=self.unknown, expanded=self.expanded) + return definition + + @property + def expanded(self): + """Complete DSL expansion and references, independent of any registry.""" + def resolve(expression, role): + return None if expression is None else resolve_selection_expression(expression, self.criteria, role=role) + return dict( + filter_by=resolve(self.filter_by, "eligibility"), score_by=resolve(self.score_by, "score"), + ranking_by=[resolve(term.expression, "ranking") for term in self.ranking_by], + criteria={item.name: dict(expression=resolve(item.expression, item.role), + applies_to=resolve(item.applies_to, "eligibility")) for item in self.criteria}, + ) @classmethod def from_dict(cls, definition): """Load a complete saved definition, rejecting unknown/missing fields. - Only schema version 1 is accepted. Duplicate model/version selections + Schema versions 1 and 2 are accepted. Version 1 retains its original + definition and digest. Duplicate model/version selections raise instead of silently taking the last record. Parameters @@ -152,13 +215,25 @@ def from_dict(cls, definition): """ if not isinstance(definition, Mapping): raise ValueError("SelectionPolicy definition must be a mapping") - missing, unknown = _POLICY_FIELDS - set(definition), set(definition) - _POLICY_FIELDS + version = definition.get("schema_version") + if type(version) is not int or version not in (1, 2): + raise ValueError("Unsupported selection policy schema_version; expected 1 or 2") + fields = _V1_FIELDS if version == 1 else _POLICY_FIELDS + missing, unknown = fields - set(definition), set(definition) - fields if missing or unknown: raise ValueError(f"SelectionPolicy fields: missing={sorted(missing)!r}, unknown={sorted(unknown, key=str)!r}") - if type(definition["schema_version"]) is not int or definition["schema_version"] != 1: - raise ValueError("Unsupported selection policy schema_version; expected 1") values = dict(definition) del values["schema_version"] + expanded = values.pop("expanded", None) + if version >= 2: + for label, constructor in (("criteria", SelectionCriterion), ("ranking_by", RankingTerm)): + records = values[label] + if not isinstance(records, list): + raise ValueError(f"Saved {label} must be a list") + try: + values[label] = tuple(constructor(**record) for record in records) + except (TypeError, ValueError) as error: + raise ValueError(f"Invalid saved {label}: {error}") from error versions = values["default_versions"] if versions is not None: if not isinstance(versions, list): @@ -172,7 +247,11 @@ def from_dict(cls, definition): raise ValueError(f"Duplicate version selection for {key!r}") decoded[key] = record["version"] values["default_versions"] = decoded - return cls(**values) + policy = cls(**values) + if version >= 2 and expanded != policy.expanded: + raise ValueError("Saved expanded definitions do not match criterion references") + object.__setattr__(policy, "_schema_version", version) + return policy @property def sha256(self): @@ -214,8 +293,22 @@ def resolve_selection_policy(configuration): missing = {"name", "score_by"} - set(configuration) if unknown or missing: raise ValueError(f"SelectionPolicy fields: missing={sorted(missing)!r}, unknown={sorted(unknown, key=str)!r}") - defaults = SelectionPolicy(configuration["name"], configuration["score_by"]).to_dict() + defaults = SelectionPolicy(configuration["name"], "0").to_dict() + if configuration.get("schema_version") == 1: + defaults = {key: value for key, value in defaults.items() if key in _V1_FIELDS} defaults.update(configuration) + if defaults.get("schema_version") == 2 and "expanded" not in configuration: + # Resolve authoring references after all consumer overrides, not against + # the temporary constructor used to materialize scalar defaults above. + settings = dict(defaults) + settings.pop("schema_version") + settings.pop("expanded") + settings["criteria"] = tuple(SelectionCriterion(**item) for item in settings["criteria"]) + settings["ranking_by"] = tuple(RankingTerm(**item) for item in settings["ranking_by"]) + settings["default_versions"] = None if settings["default_versions"] is None else { + (item["kind"], item["method"]): item["version"] for item in settings["default_versions"]} + # from_dict below still checks duplicate version records. + defaults["expanded"] = SelectionPolicy(**settings).expanded return SelectionPolicy.from_dict(defaults) @@ -311,6 +404,8 @@ def rank_with_policy(result, policy, *, provenance=None): raise TypeError("policy must be a SelectionPolicy") if not isinstance(result, TopiaryResult): raise TypeError("result must be a TopiaryResult from combine_sources") + if policy.score_fill is not None or policy.min_score is not None or policy.criteria or policy.ranking_by: + raise ValueError("Occurrence score fill/gates require evaluate_selection_policy, then select_policy_representatives") if provenance is not None and not isinstance(provenance, Mapping): raise ValueError("provenance must be a JSON-compatible mapping or None") derivation = None if provenance is None else json.loads(json.dumps(dict(provenance), allow_nan=False)) From 06611d83d83f9a26a864dffcde992b1499a645b1 Mon Sep 17 00:00:00 2001 From: Alex Rubinsteyn Date: Thu, 1 Oct 2026 13:03:34 -0400 Subject: [PATCH 2/4] Keep projected allele identities distinct and audit nonfinite comparisons --- tests/test_policy_evaluation.py | 12 ++++++++++++ tests/test_selection_criteria.py | 1 + topiary/policy_evaluation.py | 4 ++++ topiary/ranking/nodes.py | 14 ++++++++++---- 4 files changed, 27 insertions(+), 4 deletions(-) diff --git a/tests/test_policy_evaluation.py b/tests/test_policy_evaluation.py index 5a91a58..f9c15ee 100644 --- a/tests/test_policy_evaluation.py +++ b/tests/test_policy_evaluation.py @@ -167,3 +167,15 @@ def test_saved_expansion_preserves_original_arithmetic_and_legacy_boolean_scores actual = evaluate_selection_policy(TopiaryResult(frame), policy).occurrences.score expected = evaluate_scores(frame, expression) pd.testing.assert_series_equal(actual, expected, check_names=False, check_exact=True) + + +def test_projection_does_not_inherit_one_of_several_observed_candidate_ids(): + frame = pd.concat([source(allele="HLA-A*02:01").df, source(allele="HLA-B*07:02").df], ignore_index=True) + combined = combine_sources({"multiple": TopiaryResult(frame)}, sample_name="p") + policy = SelectionPolicy("projected", "1") + evaluation = evaluate_selection_policy(combined, policy, alleles=["HLA-C*07:01"]) + projected = evaluation.occurrences.query("allele == 'HLA-C*07:01'") + assert len(projected) == 2 + assert projected.candidate_id.notna().all() + assert not set(projected.candidate_id) & set(combined.df.candidate_id) + assert len(select_policy_representatives(evaluation)) == 6 diff --git a/tests/test_selection_criteria.py b/tests/test_selection_criteria.py index 8126959..13d3388 100644 --- a/tests/test_selection_criteria.py +++ b/tests/test_selection_criteria.py @@ -69,6 +69,7 @@ def test_pass_fail_unknown_zero_and_not_applicable_remain_distinct(unknown, expe ("not_supplied >= 0", "missing_column"), ("affinity.value >= 0", "missing_evidence"), ("(-1).sqrt() >= 0", "out_of_domain"), + ("(1 / 0) > 0", "out_of_domain"), ]) def test_unknown_diagnostics(expression, reason): frame = source(values=(np.nan, np.nan)).df diff --git a/topiary/policy_evaluation.py b/topiary/policy_evaluation.py index 8c727bd..23f80c7 100644 --- a/topiary/policy_evaluation.py +++ b/topiary/policy_evaluation.py @@ -193,6 +193,10 @@ def _partition_decisions(frame, positions, context, policy, partition): # Preserve source identity/annotations which are constant in a group. local = frame.loc[rows if rows else support] for column in ("candidate_id", "candidate_mhc_class", "candidate_sample", "source_observation_id", "source_label"): + if not rows and column in {"candidate_id", "candidate_mhc_class"}: + # Supporting rows may name several other alleles. The projected + # candidate's identity is derived from its own allele below. + continue if column in local and column not in record: values = local[column].drop_duplicates() if len(values) > 1: diff --git a/topiary/ranking/nodes.py b/topiary/ranking/nodes.py index bc7c490..2df6506 100644 --- a/topiary/ranking/nodes.py +++ b/topiary/ranking/nodes.py @@ -3576,8 +3576,15 @@ def eval(self, ctx: EvalContext) -> pd.Series: b = self.right.eval(ctx) # pandas comparison returns False for NaN comparisons — matches # the intended "missing values fail the filter" behavior. - result = self.op(a, b) - return result.astype("boolean").where(a.notna() & b.notna()) if ctx.preserve_unknown else result + return self._compare(a, b, ctx) + + def _compare(self, left, right, ctx): + result = self.op(left, right) + if not ctx.preserve_unknown: + return result + valid = np.isfinite(left.to_numpy(dtype=float, na_value=np.nan)) & np.isfinite( + right.to_numpy(dtype=float, na_value=np.nan)) + return result.astype("boolean").where(valid) def _should_auto_aggregate(self, ctx): """Gate check for the narrow auto-aggregation scope (issue #118). @@ -3659,8 +3666,7 @@ def _auto_aggregate(self, ctx): left_agg = left_df.max(axis=1, skipna=True) right_agg = right_df.min(axis=1, skipna=True) - result = self.op(left_agg, right_agg) - return result.astype("boolean").where(left_agg.notna() & right_agg.notna()) if ctx.preserve_unknown else result + return self._compare(left_agg, right_agg, ctx) def __repr__(self): sym = _CMP_SYMBOLS.get(self.op, "?") From 9d47c43c10da3aead8e5a7cdb2557e59c698ce7d Mon Sep 17 00:00:00 2001 From: Alex Rubinsteyn Date: Thu, 1 Oct 2026 13:15:08 -0400 Subject: [PATCH 3/4] Validate composed criterion output roles rather than banning cross-role arithmetic --- docs/combined-sources.md | 5 +++-- tests/test_selection_criteria.py | 19 +++++++++++++++++++ topiary/selection_criteria.py | 14 ++++++++------ 3 files changed, 30 insertions(+), 8 deletions(-) diff --git a/docs/combined-sources.md b/docs/combined-sources.md index 67797f1..1d040bb 100644 --- a/docs/combined-sources.md +++ b/docs/combined-sources.md @@ -233,8 +233,9 @@ audit = evaluated.audit Eligibility criteria compose with explicit `&`, `|`, and `~`; score terms combine through explicit arithmetic; `ranking_by` lists ordered expressions and directions after the primary score. YAML overrides do none of this -implicitly. Predicate references may reference predicates, score terms may -reference score terms, and ranking terms may reference score or ranking terms. +implicitly. Numeric references can be compared to form predicates, and predicates can +participate in explicit score arithmetic. A bare numeric reference is not an +eligibility predicate; a bare predicate is not a named numeric score term. An input column called `binding` remains a column; only `criterion("binding")` means the named criterion. Unknown/cyclic references, duplicate names, wrong roles and non-boolean eligibility outputs raise. Optional `applies_to` is an diff --git a/tests/test_selection_criteria.py b/tests/test_selection_criteria.py index 13d3388..afd3822 100644 --- a/tests/test_selection_criteria.py +++ b/tests/test_selection_criteria.py @@ -143,3 +143,22 @@ def test_named_references_do_not_shadow_input_columns_and_direct_dsl_stays_uncha assert pd.isna(predicate.eval(ctx.derive(preserve_unknown=True)).iloc[0]) audit = evaluate_selection_criteria(policy, ctx, references=["binding"]) assert audit.decisions.status.tolist() == ["unknown", "pass"] + + +def test_numeric_criteria_can_be_compared_and_predicates_used_in_explicit_arithmetic(): + quality = SelectionCriterion("quality", "1 / affinity.value", "score") + eligible = SelectionCriterion("good", 'criterion("quality") > 0.01', "eligibility") + weighted = SelectionCriterion("gated", 'criterion("good") * criterion("quality")', "score") + policy = SelectionPolicy("typed-composition", 'criterion("gated")', + criteria=(quality, eligible, weighted), min_score=0.001) + evaluation = evaluate_selection_policy(source(), policy) + assert evaluation.occurrences.score.tolist() == [0.02, 0.0] + assert evaluation.selected.peptide.tolist() == ["SIINFEKL"] + filtered = evaluate_selection_policy(source(), replace(policy, filter_by='criterion("quality") > 0.01')) + assert filtered.selected.peptide.tolist() == ["SIINFEKL"] + with pytest.raises(ValueError, match="explicit comparison"): + replace(policy, filter_by='criterion("quality")') + with pytest.raises(ValueError, match="numeric"): + replace(policy, score_by='criterion("good")') + restored = replay_selection_policy(evaluation.evidence) + pd.testing.assert_frame_equal(restored.occurrences, evaluation.occurrences) diff --git a/topiary/selection_criteria.py b/topiary/selection_criteria.py index dc29500..c05f7b3 100644 --- a/topiary/selection_criteria.py +++ b/topiary/selection_criteria.py @@ -21,8 +21,8 @@ class SelectionCriterion: Existing Topiary DSL expression, optionally referencing other criteria. role : {'eligibility', 'score', 'ranking'} Predicate, numeric score term, or numeric ordered tie-break expression. - Eligibility references may compose only predicates; score references - only score terms; ranking may reuse score or ranking terms. + References retain their roles: compare numeric terms to produce a + predicate, or use explicit arithmetic to turn a predicate into a score. applies_to : str, optional Eligibility expression defining applicability. False means not applicable; missing means applicability unknown. None applies to all @@ -124,13 +124,11 @@ def resolve_selection_expression(expression, criteria=(), *, role="score"): def expand(text, expected): expanded_references = {} + reference_roles = {} def lookup(name): if name not in definitions: raise ValueError(f"Unresolved criterion {name!r}") criterion = definitions[name] - allowed = {"ranking", "score"} if expected == "ranking" else {expected} - if criterion.role not in allowed: - raise ValueError(f"Criterion {name!r} has role {criterion.role}, expected {expected}") if name in active: raise ValueError(f"Cyclic criterion references: {' -> '.join([*active, name])}") if name not in visited: @@ -138,13 +136,17 @@ def lookup(name): active.append(name) node, expanded = expand(criterion.expression, criterion.role) expanded_references[name] = expanded + reference_roles[id(node)] = criterion.role if criterion.applies_to is not None: expand(criterion.applies_to, "eligibility") active.pop() return node node = parse(text, criteria=_References(lookup)) boolean = isinstance(node, (Comparison, BoolOp, IsIn, Includes)) - if expected != "eligibility" and boolean and active: + referenced_role = reference_roles.get(id(node)) + if expected == "eligibility" and referenced_role in {"score", "ranking"}: + raise ValueError("A numeric criterion needs an explicit comparison for eligibility") + if expected != "eligibility" and ((boolean and active) or referenced_role == "eligibility"): raise ValueError(f"Expected numeric {expected} expression, got a predicate") # Preserve source parentheses and numeric literals. repr(node) is a # human display, not a lossless serialization of arithmetic grouping. From a74b455a08fb33d9183d7fedfded3e9c0c3509ad Mon Sep 17 00:00:00 2001 From: Alex Rubinsteyn Date: Thu, 1 Oct 2026 13:18:28 -0400 Subject: [PATCH 4/4] Persist representative choices and their audit links through replay --- docs/combined-sources.md | 5 ++++- tests/test_selection_criteria.py | 23 +++++++++++++++++++++++ topiary/policy_evaluation.py | 27 ++++++++++++++++++++++----- 3 files changed, 49 insertions(+), 6 deletions(-) diff --git a/docs/combined-sources.md b/docs/combined-sources.md index 1d040bb..36cb472 100644 --- a/docs/combined-sources.md +++ b/docs/combined-sources.md @@ -203,7 +203,10 @@ all input measurements come from the selected model. `select_policy_representatives` chooses an actual eligible occurrence, without summing support, and records all alternative occurrence IDs, including excluded -alternatives. Direct consumers supply their `candidate_keys` and `strata` if +alternatives. It records the choice and its runtime keys in the evaluation +evidence metadata; replay repeats that explicit choice and `audit` links +criteria to the selected occurrence and rationale. Direct consumers supply +their `candidate_keys` and `strata` if combined-source candidate columns are absent. Missing values sort last; stable input order breaks exact ties. Topiary owns these generic decisions; Vaxrank still owns window geometry, source admission and construct assembly. diff --git a/tests/test_selection_criteria.py b/tests/test_selection_criteria.py index afd3822..bf367f7 100644 --- a/tests/test_selection_criteria.py +++ b/tests/test_selection_criteria.py @@ -162,3 +162,26 @@ def test_numeric_criteria_can_be_compared_and_predicates_used_in_explicit_arithm replace(policy, score_by='criterion("good")') restored = replay_selection_policy(evaluation.evidence) pd.testing.assert_frame_equal(restored.occurrences, evaluation.occurrences) + + +def test_representative_choice_runtime_keys_and_audit_rationale_survive_reload(tmp_path): + frame = source(peptides=("SIINFEKL", "SIINFEKL"), peptide_offset=[0, 20]).df + frame["prediction_id"] = ["first", "second"] + criterion = SelectionCriterion("reads", "n_rna_alt", "score") + policy = SelectionPolicy("choice", 'criterion("reads")', criteria=(criterion,), duplicates="best", strata=()) + evaluation = evaluate_selection_policy(TopiaryResult(frame), policy, + group_keys=["prediction_id", "peptide", "peptide_offset", "allele"]) + selected = select_policy_representatives(evaluation, candidate_keys=["peptide", "allele"], strata=[]) + assert selected.prediction_id.tolist() == ["second"] + assert evaluation.audit.representative_occurrence_id.eq(selected.occurrence_id.iloc[0]).all() + assert evaluation.audit.representative_reason.str.contains("ordered keys").all() + record = evaluation.evidence.extra["policy_evaluation"]["representative_selection"] + assert record["candidate_keys"] == ["peptide", "allele"] and record["strata"] == [] + for suffix, writer, _, reader in DELIMITED_IO_TWINS: + path = tmp_path / ("choice." + suffix) + writer(evaluation.evidence, path) + reloaded = reader(path) + assert reloaded.extra["policy_evaluation"]["representative_selection"] == record + replay = replay_selection_policy(reloaded) + assert replay.evidence.extra["policy_evaluation"]["representative_selection"] == record + pd.testing.assert_frame_equal(replay.audit, evaluation.audit) diff --git a/topiary/policy_evaluation.py b/topiary/policy_evaluation.py index 23f80c7..7a8510a 100644 --- a/topiary/policy_evaluation.py +++ b/topiary/policy_evaluation.py @@ -120,8 +120,13 @@ def policy(self): def audit(self): """One criterion decision per occurrence, with all identity/row links.""" records = [] + selection = self.evidence.extra["policy_evaluation"].get("representative_selection", {}) + choices = {alternative: (row["occurrence_id"], row["representative_reason"]) + for row in selection.get("records", []) for alternative in row["alternative_occurrences"]} for row in self.occurrences.to_dict("records"): criteria = row.pop("criteria", []) + chosen, rationale = choices.get(row["occurrence_id"], (None, None)) + row.update(representative_occurrence_id=chosen, representative_reason=rationale) records.extend(dict(row, **criterion) for criterion in criteria) return pd.DataFrame(records) @@ -336,6 +341,13 @@ def replay_selection_policy(result): if policy.sha256 != record["sha256"]: raise ValueError("Saved selection policy digest does not match its definition") partitions = record["partitions"] + def finish(replay): + selection = record.get("representative_selection") + if selection is not None: + select_policy_representatives(replay, candidate_keys=selection["candidate_keys"], + strata=selection["strata"]) + return replay + if len(partitions) == 1 and partitions[0]["source_label"] is None: options = _context_options(partitions[0]["context"]) groups = options.pop("group_keys") @@ -346,7 +358,7 @@ def replay_selection_policy(result): replay = evaluate_selection_policy(result, runtime_policy, group_keys=groups, provenance=record["provenance"], **options) replay.evidence.extra["policy_evaluation"].update(definition=policy.to_dict(), sha256=policy.sha256) - return replay + return finish(replay) groups = [item["context"]["group_keys"] for item in partitions] if groups and any(keys != groups[0] for keys in groups): raise ValueError("Saved source contexts must share group_keys") @@ -355,8 +367,8 @@ def replay_selection_policy(result): context = _context_options(item["context"]) context.pop("group_keys") contexts[item["source_label"]] = context - return evaluate_selection_policy(result, policy, group_keys=groups[0] if groups else None, - source_contexts=contexts, provenance=record["provenance"]) + return finish(evaluate_selection_policy(result, policy, group_keys=groups[0] if groups else None, + source_contexts=contexts, provenance=record["provenance"])) def select_policy_representatives(evaluation, *, candidate_keys=("candidate_id",), strata=None): @@ -365,7 +377,9 @@ def select_policy_representatives(evaluation, *, candidate_keys=("candidate_id", Parameters ---------- evaluation : PolicyEvaluation - Complete occurrence decisions. This object is not narrowed or changed. + Complete occurrence decisions. Records the selection in this object's + evidence metadata, without narrowing observations or changing scores. + Save evaluation.evidence to persist the choice and its runtime keys. candidate_keys : sequence of str Columns identifying candidates in the occurrence table. For a direct consumer, pass its explicit candidate identity (e.g. peptide, allele). @@ -407,5 +421,8 @@ def select_policy_representatives(evaluation, *, candidate_keys=("candidate_id", row["representative_reason"] = (f"{policy.duplicates}; ordered keys {columns}; " "missing last; stable input order breaks ties") rows.append(row) - return pd.DataFrame(rows, columns=[*frame.columns, "alternative_occurrences", "representative_reason"]).sort_values( + selected = pd.DataFrame(rows, columns=[*frame.columns, "alternative_occurrences", "representative_reason"]).sort_values( "score", ascending=policy.ascending, na_position="last", kind="stable").reset_index(drop=True) + evaluation.evidence.extra["policy_evaluation"]["representative_selection"] = dict( + candidate_keys=list(candidate_keys), strata=list(strata), records=_json_value(selected.to_dict("records"))) + return selected