diff --git a/CHANGELOG.md b/CHANGELOG.md index fb87a14..2da5d93 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,16 @@ between published tags; older pre-5.0 notes are retained below. For current interfaces, see the [consumer guide](docs/consumer-guide.md). +## 5.89.0 + +- Save and replay named `SelectionPolicy` definitions with complete DSL expressions, + method/version selections, ranking settings, and a content digest. Ranked + exports retain the policy, derivation and separate execution provenance. + Public mapping IO accepts composed consumer configuration and freezes defaults + without running predictors or changing scientific defaults (#442, #443). +- Preserve binary floating-point measurements and annotations through typed + CSV/TSV reload, keeping threshold decisions and scores reproducible (#441). + ## 5.88.1 - Prefer the current predictor DataFrame API for cache fallbacks, retaining diff --git a/docs/api.md b/docs/api.md index b4189f5..ab0123e 100644 --- a/docs/api.md +++ b/docs/api.md @@ -431,6 +431,17 @@ always `Column("x") == "y"` (see below). ### Context options +For portable policy definitions, `SelectionPolicy` stores explicit filter/score +strings and model/version selections. `resolve_selection_policy(mapping)` fills +authoring defaults after a consumer has composed its overrides; +`SelectionPolicy.from_dict(mapping)` strictly reloads a complete definition. +`to_dict()` and `sha256` expose the effective settings and content identity. +`read_selection_policy(path)` / `write_selection_policy(policy, path)` provide +JSON IO; writes refuse existing files. `rank_with_policy(result, policy, +provenance=None)` wraps `rank_candidates` and returns a `TopiaryResult` carrying +the effective definition and separate provenance/execution metadata. See the +[policy workflow and consumer boundary](combined-sources.md#named-selection-policies). + `apply_filter`, `apply_sort` and `evaluate_scores` share five keyword-only options, all forwarded to `EvalContext`: diff --git a/docs/combined-sources.md b/docs/combined-sources.md index 053ba2f..4afb43c 100644 --- a/docs/combined-sources.md +++ b/docs/combined-sources.md @@ -50,6 +50,118 @@ Unknown predictor names and versions remain unknown. The ordinary expression `affinity.value` continues to work; adding provenance does not require a version-qualified expression for this simple case. +## Named selection policies + +`SelectionPolicy` saves the exact candidate filter, score, model/version +selections, ranking direction, duplicate policy and strata under a stable name. +It has no implicit scientific recipe. Topiary does not currently ship an +official `openvax-v1` preset; that name can identify a reviewed definition you +freeze, and later definitions can use new names. + +```python +from topiary import ( + SelectionPolicy, read_selection_policy, write_selection_policy, + rank_with_policy, read_tsv, +) + +# Illustrative settings, not a calibrated biological model. +policy = SelectionPolicy( + name="example-v1", score_by="1 / affinity.value", + filter_by="n_rna_alt >= 5", duplicates="best", +) +write_selection_policy(policy, "example-v1.json") # refuses to overwrite +combined.to_tsv("all-evidence.tsv") +ranked = rank_with_policy(combined, policy) +ranked.to_tsv("ranked.tsv") + +replayed = rank_with_policy( + read_tsv("all-evidence.tsv"), read_selection_policy("example-v1.json"), +) +``` + +The immutable policy has `to_dict()` / `from_dict()` for complete, strict, +schema-versioned mappings and `sha256` for the canonical definition, including +its name. Every default is saved explicitly. Changing an expression or model +selection changes the digest even if the name remains `openvax-v1`. Runtime +versions and input facts live separately in +`ranked.extra['selection_policy']['execution']`; authoring history can be passed +as `provenance=` and is stored alongside the definition, outside its digest. +Record actual base digests and ordered config-file hashes there when available. + +`rank_with_policy` delegates to `rank_candidates`: it filters first, then scores +surviving evidence and chooses one representative per candidate/stratum. Unknown +filter results exclude groups under the existing DSL's evidence-retention +rules. Missing scores remain unranked; zero scores remain zero. It performs no +predictor calls, automatic method/version preference selection, or score filling. +Explicit model selections follow the same DSL rules as direct calls. To choose +input-dependent defaults, call `resolve_default_methods` and +`resolve_default_versions` explicitly and include their results in the effective +policy before saving. Unstated historical versions remain unknown. Distinct +source-local selections need separate effective policies/invocations; do not +apply one source's choice globally. + +CSV/TSV exports retain the full definition, digest, derivation and execution +record in long and wide form. Numeric measurements and annotations round-trip +exactly as binary floats. Preserve the **full evidence** as well: a filtered +ranking cannot restore excluded observations or discarded alternatives. Replay +also needs the recorded Topiary version when exact execution semantics matter. + +Proteasome cleavage and whole-peptide half-life can already be named explicitly, +for example `peptide_view(proteasome_cleavage.score)` and +`peptide_view(serum_half_life.value)`. Saving a policy does not enable these +predictors or add processing weights to existing defaults. Typed extracellular +site-level evidence remains [#288](https://github.com/openvax/topiary/issues/288). + +### Composed consumer configuration + +Vaxrank owns YAML composition and window/construct configuration. Its repeated +`--config openvax-v1.yaml --config overrides.yaml` workflow merges mappings +left to right and replaces lists/scalars. The Topiary integration boundary is +the **already-composed policy subtree**: + +```python +from topiary import resolve_selection_policy + +# `merged` comes from the consumer's config loader, after all overrides. +policy = resolve_selection_policy(merged["selection_policy"]) +saved_settings = { + "selection_policy": policy.to_dict(), + "selection_policy_sha256": policy.sha256, + "selection_policy_provenance": config_provenance, + "vaccine_settings": effective_vaccine_settings, +} +# Reload complete definitions strictly; never apply newer authoring defaults. +policy = SelectionPolicy.from_dict(saved_settings["selection_policy"]) +``` + +`resolve_selection_policy` accepts a JSON- or YAML-decoded mapping with required +`name` and `score_by`. It fills omitted optional settings only after composition. +An omitted filter in an override inherits the base filter; an explicit YAML +`filter_by: null` removes it. Replacing a filter expression does not AND it with +the old expression. Model selections are mappings; version selections are lists +of `{kind, method, version}` records, so an override replaces that entire list. +Topiary does not merge files or parse unrelated consumer settings. + +Vaxrank adoption of this subtree remains +[Vaxrank #497](https://github.com/openvax/vaxrank/issues/497); the example above +describes the consumer boundary, not a newly supported Vaxrank configuration +key. Its current occurrence-level scorer can consume `policy.score_by`, +`policy.filter_by`, and the selection mappings through the existing public +`EvalContext` / `apply_filter` APIs, retaining its explicit grouping and +per-occurrence `alleles` lookup. Vaxrank's post-score minimum gate, missing-score +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. + ## Identities and source evidence | Column | Meaning | diff --git a/docs/ranking.md b/docs/ranking.md index f300f76..f73c296 100644 --- a/docs/ranking.md +++ b/docs/ranking.md @@ -19,6 +19,11 @@ df = apply_sort(df, [Presentation.score, Affinity.score]) `TopiaryPredictor(filter_by=..., sort_by=[...])` applies them automatically during prediction. +For reusable named configurations and exact save/replay, see +[selection policies](combined-sources.md#named-selection-policies). A policy +stores explicit expressions and settings without changing the DSL or enabling +new predictors. + ## Group identity Expressions evaluate per *group*, not per row: one peptide-allele group can span several rows (one per predictor and kind). `apply_filter` keeps or drops whole groups, `apply_sort` orders groups, and `evaluate_scores` gives every row its group's score. diff --git a/scripts/check_vaxrank_candidates.py b/scripts/check_vaxrank_candidates.py index 05342b7..3419adc 100644 --- a/scripts/check_vaxrank_candidates.py +++ b/scripts/check_vaxrank_candidates.py @@ -8,7 +8,10 @@ import pandas as pd from mhctools import Prediction -from topiary import combine_sources, evaluate_scores, join_annotations, parse, rank_candidates, rescore_candidates +from topiary import ( + SelectionPolicy, combine_sources, evaluate_scores, join_annotations, parse, + rank_with_policy, read_selection_policy, write_selection_policy, rescore_candidates, +) from tests.test_candidate_tables import Model, source from vaxrank.candidate_epitope import candidate_epitopes_from_rows from vaxrank.epitope_config import EpitopeConfig @@ -21,7 +24,7 @@ @pytest.mark.parametrize("antigen_kind", ["mutation", "fusion", "splice", "CTA", "ERV", "viral"]) @pytest.mark.parametrize("policy", ["original", "rescored", "rna_overlay"]) -def test_candidate_features_reach_vaxrank_scoring_and_vaccine_construction(antigen_kind, policy): +def test_candidate_features_reach_vaxrank_scoring_and_vaccine_construction(antigen_kind, policy, tmp_path): proteins = ["MAAASIINFEKL", "MAAAGILGFVFTL"] identity = dict(protein_sequence=proteins, event_id=["event-1", "event-2"]) combined = combine_sources({ @@ -39,8 +42,16 @@ def test_candidate_features_reach_vaxrank_scoring_and_vaccine_construction(antig combined = join_annotations(combined, annotations, on=keys, prefix="rna", provenance={"source": "rna_only", "unit": "TPM"}) expression = "rna_transcript_expression / affinity.value" - ranked = rank_candidates(combined, expression, duplicates="best") - selected_ids = set(ranked.source_observation_id) + saved_policy = SelectionPolicy( + name="example-" + policy, score_by=expression, + filter_by="n_rna_alt >= 5", duplicates="best", + ) + policy_path = tmp_path / "selection.json" + write_selection_policy(saved_policy, policy_path) + restored_policy = read_selection_policy(policy_path) + assert restored_policy.sha256 == saved_policy.sha256 + ranked = rank_with_policy(combined, restored_policy) + selected_ids = set(ranked.df.source_observation_id) frame = combined.long_df[combined.long_df.source_observation_id.isin(selected_ids)].copy() # Vaxrank's current public scoring interface names its provenance key # prediction_id. Retain originals in the combined result; this is a @@ -59,7 +70,8 @@ def test_candidate_features_reach_vaxrank_scoring_and_vaccine_construction(antig overlaps_targetable=True, patient_alleles=[row.allele], )) epitopes = candidate_epitopes_from_rows(rows) - cfg = EpitopeConfig(score_expr=expression, filter_expr="n_rna_alt >= 5", min_epitope_score=0.) + cfg = EpitopeConfig(score_expr=restored_policy.score_by, + filter_expr=restored_policy.filter_by, min_epitope_score=0.) scored = attach_per_allele_scores(epitopes, cfg, topiary_df=frame) expected = dict(zip(frame.source_observation_id, evaluate_scores(frame, parse(expression)))) vaccines = [] diff --git a/tests/test_consumer_workflows.py b/tests/test_consumer_workflows.py index d915b94..330cdf0 100644 --- a/tests/test_consumer_workflows.py +++ b/tests/test_consumer_workflows.py @@ -56,6 +56,75 @@ from .test_twin_conformance import DSL_MEASUREMENT_TWINS +@pytest.mark.parametrize("form", ["long", "wide"]) +def test_saved_profiles_replay_exact_scores_and_change_selection(tmp_path, form): + from topiary import ( + SelectionPolicy, combine_sources, rank_with_policy, + read_selection_policy, write_selection_policy, + ) + from .test_candidate_tables import source + from .test_twin_conformance import DELIMITED_IO_TWINS + + boundary = 0.12345678901234567 + annotations = [boundary, np.nextafter(boundary, np.inf)] + binding = source(values=(50., 100.), review_score=annotations) + processing = source( + values=(0.1, 0.9), score=[0.1, 0.9], review_score=annotations, + kind="proteasome_cleavage", prediction_method_name="cleavage_fixture", allele="", + ) + combined = combine_sources({"input": TopiaryResult(pd.concat( + [binding.df, processing.df], ignore_index=True))}, sample_name="patient") + before = combined.df.copy(deep=True) + # Synthetic policies demonstrate behavior, not a calibrated scientific recipe. + profiles = [ + SelectionPolicy("example-v1", "1 / affinity.value"), + SelectionPolicy("example-v2", "peptide_view(proteasome_cleavage.score)"), + SelectionPolicy("example-v3", "1 / affinity.value", filter_by=f"review_score <= {boundary!r}"), + ] + paths = [] + for profile in profiles: + path = tmp_path / (profile.name + ".json") + write_selection_policy(profile, path) + paths.append(path) + expected = [rank_with_policy(combined, profile) for profile in profiles] + assert [r.df.peptide.tolist() for r in expected] == [ + ["SIINFEKL", "GILGFVFTL"], ["GILGFVFTL", "SIINFEKL"], ["SIINFEKL"], + ] + pd.testing.assert_frame_equal(combined.df, before) + assert "selection_policy" not in combined.extra + columns = ["candidate_id", "candidate_score", "candidate_rank", "ranking_status", "candidate_observations"] + for suffix, writer, method, reader in DELIMITED_IO_TWINS: + evidence_path = tmp_path / ("evidence." + suffix) + writer(combined if form == "long" else combined.to_wide(), evidence_path) + restored = reader(evidence_path) + for path, original in zip(paths, expected): + profile = read_selection_policy(path) + ranked = rank_with_policy(restored, profile) + pd.testing.assert_frame_equal(ranked.df[columns], original.df[columns], check_exact=True) + output_path = tmp_path / (profile.name + "." + suffix) + method(ranked if form == "long" else ranked.to_wide(), output_path) + saved = reader(output_path).to_long() + # Wide IO records the models actually present in this narrower + # view; the full policy and original-source provenance survive. + for key in ("selection_policy", "combined_sources"): + assert saved.extra[key] == ranked.extra[key] + assert saved.topiary_version == ranked.topiary_version + assert saved.filter_by_str == profile.filter_by + assert saved.sort_by_str == profile.score_by + record = saved.extra["selection_policy"] + assert record["definition"] == profile.to_dict() + assert record["sha256"] == profile.sha256 + assert record["execution"]["topiary_version"] == ranked.topiary_version + # Loading the definition from the ranked export is sufficient to + # replay against the full evidence; filtered exports cannot recover it. + embedded = SelectionPolicy.from_dict(saved.extra["selection_policy"]["definition"]) + reranked = rank_with_policy(restored, embedded) + pd.testing.assert_frame_equal(reranked.df[columns], original.df[columns], check_exact=True) + actual = saved.df.set_index("candidate_id").candidate_score.sort_index() + wanted = original.df.set_index("candidate_id").candidate_score.sort_index() + pd.testing.assert_series_equal(actual, wanted, check_exact=True) + + def test_sv_interest_api_and_cli_retain_and_rank_the_same_nominations(tmp_path): import json from .test_sv_interest import catalogue, export diff --git a/tests/test_io.py b/tests/test_io.py index b38fcc7..e5241e4 100644 --- a/tests/test_io.py +++ b/tests/test_io.py @@ -658,6 +658,26 @@ def test_legacy_comment_metadata_keeps_builtins_and_custom_values(tmp_path): # --------------------------------------------------------------------------- +@pytest.mark.parametrize("form", ["long", "wide"]) +def test_binary_floats_round_trip_exactly_through_both_writer_doors(tmp_path, form): + from .test_candidate_tables import source + + boundary = 0.12345678901234567 + values = [boundary, np.nextafter(boundary, np.inf)] + result = source(values=values, review_score=values, score=values) + if form == "wide": + result = result.to_wide() + for suffix, writer, method, reader in DELIMITED_IO_TWINS: + for write in (writer, method): + path = tmp_path / ("precise." + suffix) + write(result, path) + restored = reader(path).to_long().df.set_index("peptide") + for column in ("value", "score", "review_score"): + actual = restored.loc[["SIINFEKL", "GILGFVFTL"], column].tolist() + assert [x.hex() for x in actual] == [x.hex() for x in values] + assert [x <= boundary for x in actual] == [True, False] + + class TestReadWriteTSV: def test_long_form_roundtrip(self, tmp_path): df = _sample_long_df() diff --git a/tests/test_selection_policy.py b/tests/test_selection_policy.py new file mode 100644 index 0000000..4f5ce20 --- /dev/null +++ b/tests/test_selection_policy.py @@ -0,0 +1,234 @@ +"""Saved policy definitions replay the public candidate ranker unchanged.""" + +from dataclasses import FrozenInstanceError, replace +from copy import deepcopy +import json + +import numpy as np +import pandas as pd +import pytest +import yaml + +from topiary import ( + SelectionPolicy, combine_sources, read_selection_policy, write_selection_policy, + resolve_selection_policy, rank_with_policy, +) +from .test_candidate_tables import source +from .test_twin_conformance import SELECTION_POLICY_TWINS, POLICY_MAPPING_TWINS + + +def test_definition_is_immutable_and_file_preserves_every_setting(tmp_path): + methods = {"affinity": "original"} + versions = {("affinity", "original"): "1"} + strata = ["source_label", "candidate_mhc_class"] + profile = SelectionPolicy( + "openvax-v1", "1 / affinity.value", filter_by="n_rna_alt >= 5", + ascending=True, duplicates="worst", strata=strata, + default_methods=methods, default_versions=versions, + ) + before = profile.to_dict() + methods.clear() + versions.clear() + strata.clear() + assert profile.to_dict() == before + with pytest.raises(FrozenInstanceError): + profile.score_by = "0" + with pytest.raises(TypeError): + profile.default_methods["affinity"] = "other" + path = tmp_path / "openvax-v1.json" + write_selection_policy(profile, path) + restored = read_selection_policy(path) + assert restored == profile + assert restored.sha256 == profile.sha256 + assert json.loads(path.read_text()) == before + with pytest.raises(FileExistsError): + write_selection_policy(replace(profile, score_by="affinity.value"), path) + assert read_selection_policy(path) == profile + assert replace(profile, score_by="affinity.value").sha256 != profile.sha256 + + +@pytest.mark.parametrize("settings", [ + {"name": ""}, {"score_by": ""}, {"score_by": 1}, {"score_by": "(??"}, + {"filter_by": False}, {"ascending": "false"}, {"duplicates": []}, + {"strata": "source_label"}, {"strata": ["x", "x"]}, {"strata": [None]}, + {"default_methods": []}, {"default_methods": {"affinity": ""}}, + {"default_versions": {"affinity": "1"}}, +]) +def test_invalid_definition_rejected(settings): + with pytest.raises((ValueError, SyntaxError)): + SelectionPolicy(**dict({"name": "test-v1", "score_by": "affinity.value"}, **settings)) + + +@pytest.mark.parametrize("change", [ + {"schema_version": 2}, {"schema_version": True}, {"unexpected": "setting"}, + {"default_versions": {}}, + {"default_versions": [{"kind": "affinity", "method": "original"}]}, + {"default_versions": [{"kind": "affinity", "method": "original", "version": "1"}] * 2}, +]) +def test_unknown_or_ambiguous_schema_rejected(change): + definition = SelectionPolicy("test-v1", "affinity.value").to_dict() + with pytest.raises(ValueError): + SelectionPolicy.from_dict(dict(definition, **change)) + del definition["duplicates"] + with pytest.raises(ValueError): + SelectionPolicy.from_dict(definition) + + +def test_duplicate_json_keys_rejected(tmp_path): + path = tmp_path / "ambiguous.json" + path.write_text('{"name": "first", "name": "second"}') + with pytest.raises(ValueError, match="Duplicate"): + read_selection_policy(path) + + +def test_yaml_composition_resolves_once_and_preserves_null_vs_inheritance(tmp_path): + base = yaml.safe_load(''' +name: openvax-v1 +score_by: 1 / affinity.value +filter_by: n_rna_alt >= 10 +strata: [source_label, candidate_mhc_class] +default_methods: + affinity: original + proteasome_cleavage: cleavage_fixture +default_versions: + - {kind: affinity, method: original, version: '1'} +''') + override = yaml.safe_load(''' +score_by: affinity.value +ascending: true +strata: [] +default_methods: + affinity: alternative +default_versions: [] +''') + # Match Vaxrank's authoring rule, which belongs to the consumer. Topiary + # receives the composed subtree, never its YAML files or merge algorithm. + composed = deepcopy(base) + for key, value in override.items(): + if isinstance(value, dict) and isinstance(composed.get(key), dict): + composed[key].update(value) + else: + composed[key] = value + policy = resolve_selection_policy(composed) + assert policy.name == "openvax-v1" + assert policy.filter_by == base["filter_by"] # omitted override inherits + assert policy.strata == () # lists replace, never concatenate + assert policy.default_versions == {} + assert dict(policy.default_methods) == {"affinity": "alternative", "proteasome_cleavage": "cleavage_fixture"} + assert policy.sha256 != resolve_selection_policy(base).sha256 + changed_filter = resolve_selection_policy(dict(composed, filter_by="n_rna_alt >= 1")) + assert changed_filter.filter_by == "n_rna_alt >= 1" # replace, never AND + no_filter = resolve_selection_policy(dict(composed, filter_by=None)) + assert no_filter.filter_by is None + evidence = combine_sources({"one": source()}, sample_name="patient") + assert len(rank_with_policy(evidence, policy).df) == 1 + assert len(rank_with_policy(evidence, no_filter).df) == 2 + path = tmp_path / "policy.json" + write_selection_policy(policy, path) + for decode in (json.loads, yaml.safe_load): + definition = decode(path.read_text()) + for resolve in POLICY_MAPPING_TWINS: + restored = resolve(definition) + assert restored.to_dict() == policy.to_dict() + assert restored.sha256 == read_selection_policy(path).sha256 + pd.testing.assert_frame_equal(rank_with_policy(evidence, restored).df, + rank_with_policy(evidence, policy).df) + + +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)) + assert SelectionPolicy("new-defaults", "affinity.value").ascending is True + for resolve in POLICY_MAPPING_TWINS: + assert resolve(saved).to_dict() == saved + + +def test_derivation_and_execution_provenance_are_distinct_and_detached(tmp_path): + from topiary import __version__ + from .test_twin_conformance import DELIMITED_IO_TWINS + + original = source(predictor_version=None) + original.topiary_version = "historical" + evidence = combine_sources({"one": original}, sample_name="patient") + policy = SelectionPolicy("openvax-v1", "affinity.value") + provenance = {"base_sha256": "base-digest", "overrides": [{"file": "overrides.yaml", "sha256": "file-digest"}]} + ranked = rank_with_policy(evidence, policy, provenance=provenance) + record = deepcopy(ranked.extra["selection_policy"]) + provenance["overrides"].clear() + assert record["provenance"]["overrides"] + assert record["sha256"] == policy.sha256 + assert record["definition"] == policy.to_dict() + assert record["execution"]["topiary_version"] == __version__ + assert record["execution"]["prediction_inventory"][0]["predictor_version"] is None + assert "execution" not in policy.to_dict() + for suffix, writer, method, reader in DELIMITED_IO_TWINS: + path = tmp_path / ("ranked." + suffix) + writer(ranked, path) + assert reader(path).extra["selection_policy"] == record + assert "selection_policy" not in evidence.extra + + +@pytest.mark.parametrize("configuration,field", [ + ({"score_by": "1"}, "name"), ({"name": "test"}, "score_by"), + ({"name": "test", "score_by": "1", "typo": 1}, "typo"), + ({"name": "test", "score_by": "??"}, "score_by"), + ({"name": "test", "score_by": "1", "filter_by": "??"}, "filter_by"), +]) +def test_authoring_errors_identify_the_field(configuration, field): + with pytest.raises(ValueError, match=field): + resolve_selection_policy(configuration) + + +@pytest.mark.parametrize("settings", [ + {}, {"ascending": True}, {"filter_by": "n_rna_alt >= 10"}, + {"filter_by": "n_rna_alt > 100"}, {"duplicates": "best"}, + {"duplicates": "worst"}, {"strata": ()}, + {"strata": ("source_label", "candidate_mhc_class")}, + {"default_methods": {"affinity": "original"}, + "default_versions": {("affinity", "original"): "1"}}, +]) +@pytest.mark.parametrize("case", ["ordinary", "missing", "conflicting", "empty"]) +def test_profile_and_direct_ranker_agree(case, settings): + inputs = {"one": source(values=(50., np.nan) if case == "missing" else (50., 500.))} + if case == "conflicting": + inputs["two"] = source(values=(60., 600.)) + if case == "empty": + inputs = {} + evidence = combine_sources(inputs, sample_name="patient") + original = evidence.df.copy(deep=True) + profile = SelectionPolicy("test-v1", "affinity.value", **settings) + direct, replay = SELECTION_POLICY_TWINS + calls = (lambda: direct(evidence, profile.score_by, **settings), + lambda: replay(evidence, profile).df) + outputs, errors = [], [] + for call in calls: + try: + outputs.append(call()) + errors.append(None) + except ValueError as error: + errors.append((type(error), str(error))) + assert errors[0] == errors[1] + if outputs: + pd.testing.assert_frame_equal(*outputs) + pd.testing.assert_frame_equal(evidence.df, original) + + +def test_profile_model_and_version_selections_change_results(): + from topiary import TopiaryResult + + rows = pd.concat([ + source(values=(50., 500.)).df, + source(values=(900., 10.), predictor_version="2").df, + source(values=(800., 20.), prediction_method_name="alternative").df, + ], ignore_index=True) + evidence = combine_sources({"one": TopiaryResult(rows)}, sample_name="patient") + direct, replay = SELECTION_POLICY_TWINS + orders = [] + for method, version in (("original", "1"), ("original", "2"), ("alternative", "1")): + settings = dict(ascending=True, default_methods={"affinity": method}, + default_versions={("affinity", method): version}) + profile = SelectionPolicy("selection", "affinity.value", **settings) + result = replay(evidence, profile) + pd.testing.assert_frame_equal(result.df, direct(evidence, profile.score_by, **settings)) + orders.append(result.df.peptide.tolist()) + assert orders == [["SIINFEKL", "GILGFVFTL"], ["GILGFVFTL", "SIINFEKL"], ["GILGFVFTL", "SIINFEKL"]] diff --git a/tests/test_twin_conformance.py b/tests/test_twin_conformance.py index f2b426b..3f3253f 100644 --- a/tests/test_twin_conformance.py +++ b/tests/test_twin_conformance.py @@ -47,7 +47,8 @@ describe_isovar_result, fragment_from_isovar_result, TopiaryResult, to_tsv, to_csv, read_tsv, read_csv, read_isovar_hypotheses, - combine_sources, rank_candidates, evaluate_scores, + combine_sources, rank_candidates, rank_with_policy, evaluate_scores, + SelectionPolicy, resolve_selection_policy, Affinity, Column, apply_filter, apply_sort, ) from topiary.io_isovar import _check_isovar @@ -139,6 +140,14 @@ def _sv_support_record(support): # lower-level DSL scorer that downstream consumers already call. CANDIDATE_SCORE_TWINS = (rank_candidates, evaluate_scores) +# Named policy replay must preserve the existing candidate ranker's decisions. +# Both receive the same settings in test_selection_policy.py. +SELECTION_POLICY_TWINS = (rank_candidates, rank_with_policy) + +# Complete definitions resolve identically whether decoded from a saved file +# or encountered in an already-composed authoring configuration. +POLICY_MAPPING_TWINS = (SelectionPolicy.from_dict, resolve_selection_policy) + # 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 99a3049..03bd7f0 100644 --- a/topiary/__init__.py +++ b/topiary/__init__.py @@ -3,6 +3,10 @@ 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 .selection_policy import ( + SelectionPolicy, resolve_selection_policy, read_selection_policy, + write_selection_policy, rank_with_policy, +) from .reconciliation import ( reconcile_evidence, evidence_views, normalize_rna_observation, union_rna_observations, ) @@ -191,7 +195,7 @@ encode_amino_acids, ) -__version__ = "5.88.1" +__version__ = "5.89.0" __all__ = [ "normalize_isovar_rna_support", @@ -212,6 +216,11 @@ "union_rna_observations", "protein_evidence_view", "rank_candidates", + "SelectionPolicy", + "resolve_selection_policy", + "read_selection_policy", + "write_selection_policy", + "rank_with_policy", "rescore_candidates", "osteosarc_fixture_paths", "normalize_python_types", diff --git a/topiary/io.py b/topiary/io.py index ab75205..e509348 100644 --- a/topiary/io.py +++ b/topiary/io.py @@ -416,7 +416,7 @@ def _read_delimited(path, sep, tag=None): for column in _FLANK_COLUMNS if column in columns}) version_types = {key: value for key, value in version_types.items() if key not in converters} df = pd.read_csv(StringIO(data_text), sep=sep, dtype=version_types, - converters=converters) + converters=converters, float_precision="round_trip") # Record source (tag overrides filename). source_label = tag if tag is not None else path.name diff --git a/topiary/selection_policy.py b/topiary/selection_policy.py new file mode 100644 index 0000000..df4ea33 --- /dev/null +++ b/topiary/selection_policy.py @@ -0,0 +1,347 @@ +"""Portable selection policy definitions over Topiary's existing DSL.""" + +from collections.abc import Mapping +from copy import deepcopy +from dataclasses import dataclass +import hashlib +import json +from pathlib import Path +from types import MappingProxyType + +from .ranking import as_dsl_node + + +_POLICY_FIELDS = {"schema_version", "name", "score_by", "filter_by", "ascending", + "duplicates", "strata", "default_methods", "default_versions"} + + +def _text(value, label): + if not isinstance(value, str) or not value.strip(): + raise ValueError(f"{label} must be a nonempty string") + return value + + +@dataclass(frozen=True) +class SelectionPolicy: + """A named filter and scoring policy for combined source candidates. + + Parameters + ---------- + name : str + User-assigned policy identifier, for example ``openvax-v1``. Give a + changed policy a new name; the content digest also distinguishes + different definitions carrying the same name. + score_by : str + Numeric Topiary DSL expression. There is no implicit scientific + recipe: callers must state the expression they want to preserve. + filter_by : str, optional + Boolean Topiary DSL expression; None applies no additional filter. + Existing ranker semantics apply, including exclusion on unknown + filter results and retention of missing scores as unranked candidates. + ascending : bool + False ranks larger scores first; True ranks smaller scores first. + duplicates : {'error', 'best', 'worst'} + Policy for conflicting observations of a candidate. Default error + requires an explicit choice instead of silently choosing evidence. + strata : sequence of str + Independent ranking partitions, defaulting to MHC class. An empty + sequence explicitly requests a ranking across classes. + default_methods : mapping of str to str, optional + Explicit kind-to-method selections. None leaves model resolution to + the existing DSL; no model is selected from installed software. + 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. + + Notes + ----- + Expressions are validated when the policy is created. Referenced input + columns and model selections are resolved by the existing ranker when + the policy is applied. Sequences and mappings are copied and made immutable, + so editing the caller's + configuration cannot change a saved policy. Policies do not run predictors. + """ + + name: str + score_by: str + filter_by: str | None = None + ascending: bool = False + duplicates: str = "error" + strata: tuple[str, ...] = ("candidate_mhc_class",) + default_methods: Mapping | None = None + default_versions: Mapping | None = None + + def __post_init__(self): + _text(self.name, "name") + 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)) + except (ValueError, SyntaxError) as error: + raise ValueError(f"{label}: {error}") from error + 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"}: + raise ValueError("duplicates must be 'error', 'best', or 'worst'") + if isinstance(self.strata, str) or not isinstance(self.strata, (list, tuple)): + raise ValueError("strata must be a sequence of distinct column names") + strata = tuple(_text(value, "stratum") for value in self.strata) + if len(set(strata)) != len(strata): + raise ValueError("strata must name distinct columns") + object.__setattr__(self, "strata", strata) + for label in ("default_methods", "default_versions"): + supplied = getattr(self, label) + if supplied is None: + continue + if not isinstance(supplied, Mapping): + raise ValueError(f"{label} must be a mapping or None") + copied = dict(supplied) + for key, value in copied.items(): + if label == "default_versions": + if not isinstance(key, tuple) or len(key) != 2: + raise ValueError("default_versions keys must be (kind, method) pairs") + for part in key: + _text(part, "version selection kind/method") + else: + _text(key, "method selection kind") + _text(value, label) + object.__setattr__(self, label, MappingProxyType(copied)) + + def to_dict(self): + """Return an independent, JSON-compatible schema-version-1 definition. + + Explicit None selections remain None; empty mappings remain empty. + Every setting is written, including defaults, so a future constructor + default cannot alter the meaning of the saved definition. + + Returns + ------- + dict + 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, + 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())), + default_versions=None if versions is None else [ + dict(kind=kind, method=method, version=version) + for (kind, method), version in sorted(versions.items())], + ) + + @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 + raise instead of silently taking the last record. + + Parameters + ---------- + definition : mapping + Complete output of ``to_dict()``, including explicit defaults. + Empty, partial or extra-field definitions raise ValueError. + + Returns + ------- + SelectionPolicy + Validated immutable policy, independent of the supplied mapping. + """ + if not isinstance(definition, Mapping): + raise ValueError("SelectionPolicy definition must be a mapping") + missing, unknown = _POLICY_FIELDS - set(definition), set(definition) - _POLICY_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"] + versions = values["default_versions"] + if versions is not None: + if not isinstance(versions, list): + raise ValueError("Saved default_versions must be a list of kind/method/version records") + decoded = {} + for record in versions: + if not isinstance(record, Mapping) or set(record) != {"kind", "method", "version"}: + raise ValueError("Each version selection must contain kind, method, and version") + key = (_text(record["kind"], "kind"), _text(record["method"], "method")) + if key in decoded: + raise ValueError(f"Duplicate version selection for {key!r}") + decoded[key] = record["version"] + values["default_versions"] = decoded + return cls(**values) + + @property + def sha256(self): + """SHA-256 of the complete canonical definition, including its name.""" + canonical = json.dumps(self.to_dict(), sort_keys=True, separators=(",", ":"), allow_nan=False) + return hashlib.sha256(canonical.encode("utf-8")).hexdigest() + + +def resolve_selection_policy(configuration): + """Resolve an already-composed authoring mapping into a complete policy. + + Parameters + ---------- + configuration : mapping + JSON/YAML-compatible settings. ``name`` and ``score_by`` are required; + other fields use the current constructor defaults if omitted. Version + selections use the same list of records as :meth:`SelectionPolicy.to_dict`. + Unknown fields and unsupported schema versions raise ValueError. + + Returns + ------- + SelectionPolicy + Immutable effective policy with every default materialized. Persist + ``to_dict()`` and reload it using ``from_dict()``; do not resolve a + historical partial configuration against newer defaults. + + Notes + ----- + Compose authoring overrides *before* this call. An omitted filter can + inherit a base value during that composition; an explicit null replaces + it with no filter. This function does not merge files, AND expressions, + concatenate lists, resolve models from input, or interpret Vaxrank settings. + Consumers retain their derivation/override records separately from the + effective definition and digest. + """ + if not isinstance(configuration, Mapping): + raise ValueError("SelectionPolicy configuration must be a mapping") + unknown = set(configuration) - _POLICY_FIELDS + 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.update(configuration) + return SelectionPolicy.from_dict(defaults) + + +def _unique_json_keys(pairs): + result = {} + for key, value in pairs: + if key in result: + raise ValueError(f"Duplicate selection policy JSON key: {key!r}") + result[key] = value + return result + + +def read_selection_policy(path): + """Read a named selection policy from a UTF-8 JSON file. + + The file must contain a complete versioned definition produced by + :func:`write_selection_policy`. Malformed JSON, duplicate keys, unsupported + schema versions, and invalid expressions raise ValueError. + + Parameters + ---------- + path : str or path-like + Path containing a complete saved definition, not a partial override. + + Returns + ------- + SelectionPolicy + Validated, immutable definition. An empty file raises ValueError. + """ + definition = json.loads(Path(path).read_text(encoding="utf-8"), object_pairs_hook=_unique_json_keys) + return SelectionPolicy.from_dict(definition) + + +def write_selection_policy(policy, path): + """Save a SelectionPolicy as a new UTF-8 JSON file. + + Existing files raise FileExistsError: save a changed policy under a new + version/name and path instead of overwriting the historical definition. + This stores configuration only; observations belong in the result file. + + Parameters + ---------- + policy : SelectionPolicy + Complete effective definition to preserve. + path : str or path-like + New file to create. Its parent directory must exist. + + Returns + ------- + None + The definition is written with all defaults explicitly present. + """ + if not isinstance(policy, SelectionPolicy): + raise TypeError("policy must be a SelectionPolicy") + text = json.dumps(policy.to_dict(), indent=2, sort_keys=True, allow_nan=False) + "\n" + with Path(path).open("x", encoding="utf-8") as stream: + stream.write(text) + + +def rank_with_policy(result, policy, *, provenance=None): + """Apply a saved policy to combined source candidates without prediction. + + Parameters + ---------- + result : TopiaryResult + Combined source evidence, as accepted by :func:`rank_candidates`. + It remains unchanged. Re-rank this full evidence to compare policies; + a previously filtered ranking cannot restore excluded observations. + policy : SelectionPolicy + Exact filter, score, selection, and ranking settings to apply. + provenance : mapping, optional + JSON-compatible authoring/override provenance, for example ordered + config file hashes and a base policy digest. It is copied into output + metadata, outside the effective policy digest. No provenance is inferred + from a surviving policy name. Unknown provenance remains None. + + Returns + ------- + TopiaryResult + The existing ranker's representative rows and candidate scores/ranks, + with source metadata, the full policy, its SHA-256, and the Topiary + execution version. Missing scores remain unranked. Empty inputs follow + the same validation and output rules as :func:`rank_candidates`. + CSV/TSV exports preserve the definition in ``extra['selection_policy']``. + This does not add the eligibility/reason view tracked in issue #366. + """ + from . import __version__ + from .candidates import rank_candidates + from .result import TopiaryResult + from .ranking import EvalContext, known_versions + + if not isinstance(policy, SelectionPolicy): + raise TypeError("policy must be a SelectionPolicy") + if not isinstance(result, TopiaryResult): + raise TypeError("result must be a TopiaryResult from combine_sources") + 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)) + methods = None if policy.default_methods is None else dict(policy.default_methods) + versions = None if policy.default_versions is None else dict(policy.default_versions) + ranked = rank_candidates( + result, policy.score_by, filter_by=policy.filter_by, ascending=policy.ascending, + duplicates=policy.duplicates, strata=policy.strata, + default_methods=methods, default_versions=versions, + ) + extra = deepcopy(result.extra) + inventory_columns = [c for c in ("source_label", "kind", "prediction_method_name", + "predictor_version", "prediction_run_name") if c in result.long_df] + inventory = result.long_df[inventory_columns].drop_duplicates().copy() + if "predictor_version" in inventory: + inventory["predictor_version"] = inventory.predictor_version.where(known_versions(inventory.predictor_version)) + inventory = inventory.astype(object).where(inventory.notna(), None) + extra["selection_policy"] = { + "definition": policy.to_dict(), "sha256": policy.sha256, + "provenance": derivation, + "execution": { + "topiary_version": __version__, "operation": "rank_candidates", + "group_keys": list(EvalContext(result.long_df).group_keys), + "input_topiary_version": result.topiary_version, + "input_sources": list(result.sources), "input_models": dict(result.models), + "prediction_inventory": inventory.to_dict(orient="records"), + "kind_support": deepcopy(result._kind_support()), + }, + } + return TopiaryResult( + ranked, topiary_version=__version__, form="long", sources=list(result.sources), + models=dict(result.models), filter_by_str=policy.filter_by, + sort_by_str=policy.score_by, extra=extra, + )