diff --git a/src/openamp_foundry/evidence/candidate_family_clustering.py b/src/openamp_foundry/evidence/candidate_family_clustering.py new file mode 100644 index 00000000..955aa4bf --- /dev/null +++ b/src/openamp_foundry/evidence/candidate_family_clustering.py @@ -0,0 +1,198 @@ +from __future__ import annotations +from dataclasses import dataclass, field +from typing import List, Optional + + +VALID_NOVELTY_EVIDENCE_TYPES = frozenset({ + "blast_search_no_hits", # BLAST against known AMP databases returned no hits + "hmm_profile_no_match", # HMM profile search found no match to known families + "structural_divergence", # structural prediction differs from known AMP classes + "charge_length_outside_known", # charge/length distribution outside all known families + "expert_assessment", # domain expert assessed family as novel + "insufficient_evidence", # not enough evidence to claim novelty +}) + +VALID_CLUSTERING_METHODS = frozenset({ + "sequence_identity", # pairwise sequence identity threshold + "edit_distance", # Levenshtein distance threshold + "kmer_similarity", # k-mer profile similarity + "embedding_distance", # learned embedding distance (e.g. ESM-2) + "manual_grouping", # manually assigned by researcher +}) + +VALID_NOVELTY_VERDICTS = frozenset({ + "novel_family", # sufficient evidence of novelty + "variant_of_known", # candidates are variants of a known AMP family + "insufficient_evidence", # not enough data to assess novelty + "disputed", # novelty contested by multiple evidence sources +}) + +MIN_FAMILY_SIZE = 2 +MAX_SIMILARITY_THRESHOLD = 1.0 +MIN_SIMILARITY_THRESHOLD = 0.0 + + +@dataclass +class CandidateFamilyClustering: + cfc_id: str + family_name: str + candidate_ids: List[str] + clustering_method: str + similarity_threshold: float + novelty_evidence_types: List[str] + novelty_verdict: str + n_candidates: int + n_toy_excluded: int + known_family_hits: List[str] + dry_lab_only: bool + limitations: str + notes: str = "" + min_pairwise_similarity: Optional[float] = None + max_pairwise_similarity: Optional[float] = None + + +@dataclass +class CFCValidationResult: + valid: bool + violations: List[str] = field(default_factory=list) + + +def validate_candidate_family_clustering(cfc: CandidateFamilyClustering) -> CFCValidationResult: + violations = [] + + if not cfc.cfc_id.startswith("CFC-"): + violations.append("cfc_id must start with 'CFC-'") + + if not cfc.family_name or len(cfc.family_name.strip()) < 3: + violations.append("family_name must be at least 3 characters") + + if not cfc.candidate_ids: + violations.append("candidate_ids must not be empty") + + for cid in cfc.candidate_ids: + if cid.startswith("TOY-"): + violations.append(f"TOY- candidate_id '{cid}' is not allowed in real CFC- records") + + if cfc.clustering_method not in VALID_CLUSTERING_METHODS: + violations.append( + f"clustering_method '{cfc.clustering_method}' must be one of {sorted(VALID_CLUSTERING_METHODS)}" + ) + + if not (MIN_SIMILARITY_THRESHOLD <= cfc.similarity_threshold <= MAX_SIMILARITY_THRESHOLD): + violations.append( + f"similarity_threshold {cfc.similarity_threshold} must be in [0.0, 1.0]" + ) + + for net in cfc.novelty_evidence_types: + if net not in VALID_NOVELTY_EVIDENCE_TYPES: + violations.append( + f"novelty_evidence_type '{net}' must be one of {sorted(VALID_NOVELTY_EVIDENCE_TYPES)}" + ) + + if not cfc.novelty_evidence_types: + violations.append("at least one novelty_evidence_type is required") + + if cfc.novelty_verdict not in VALID_NOVELTY_VERDICTS: + violations.append( + f"novelty_verdict '{cfc.novelty_verdict}' must be one of {sorted(VALID_NOVELTY_VERDICTS)}" + ) + + if cfc.n_candidates < MIN_FAMILY_SIZE: + violations.append(f"n_candidates must be >= {MIN_FAMILY_SIZE} (minimum family size)") + + if cfc.n_candidates != len(cfc.candidate_ids): + violations.append( + f"n_candidates ({cfc.n_candidates}) must equal len(candidate_ids) ({len(cfc.candidate_ids)})" + ) + + if cfc.n_toy_excluded < 0: + violations.append("n_toy_excluded must be >= 0") + + # novel_family requires at least one hard evidence type (not insufficient_evidence or expert_assessment alone) + if cfc.novelty_verdict == "novel_family": + hard_evidence = {"blast_search_no_hits", "hmm_profile_no_match", "structural_divergence", "charge_length_outside_known"} + if not any(net in hard_evidence for net in cfc.novelty_evidence_types): + violations.append( + "novelty_verdict='novel_family' requires at least one hard evidence type " + "(blast_search_no_hits, hmm_profile_no_match, structural_divergence, or charge_length_outside_known)" + ) + + # variant_of_known requires known_family_hits + if cfc.novelty_verdict == "variant_of_known" and not cfc.known_family_hits: + violations.append( + "novelty_verdict='variant_of_known' requires at least one entry in known_family_hits" + ) + + if not cfc.dry_lab_only: + violations.append("dry_lab_only must be True for CFC- records (clustering is a computational result)") + + if not cfc.limitations or len(cfc.limitations.strip()) < 10: + violations.append("limitations must be a non-empty string (at least 10 characters)") + + # pairwise similarity consistency + if cfc.min_pairwise_similarity is not None and cfc.max_pairwise_similarity is not None: + if cfc.min_pairwise_similarity > cfc.max_pairwise_similarity: + violations.append( + f"min_pairwise_similarity ({cfc.min_pairwise_similarity}) cannot exceed max_pairwise_similarity ({cfc.max_pairwise_similarity})" + ) + + return CFCValidationResult(valid=len(violations) == 0, violations=violations) + + +def build_candidate_family_clustering( + cfc_id: str, + family_name: str, + candidate_ids: List[str], + clustering_method: str, + similarity_threshold: float, + novelty_evidence_types: List[str], + novelty_verdict: str, + n_toy_excluded: int, + known_family_hits: List[str], + limitations: str, + notes: str = "", + min_pairwise_similarity: Optional[float] = None, + max_pairwise_similarity: Optional[float] = None, +) -> CandidateFamilyClustering: + cfc = CandidateFamilyClustering( + cfc_id=cfc_id, + family_name=family_name, + candidate_ids=candidate_ids, + clustering_method=clustering_method, + similarity_threshold=similarity_threshold, + novelty_evidence_types=novelty_evidence_types, + novelty_verdict=novelty_verdict, + n_candidates=len(candidate_ids), + n_toy_excluded=n_toy_excluded, + known_family_hits=known_family_hits, + dry_lab_only=True, + limitations=limitations, + notes=notes, + min_pairwise_similarity=min_pairwise_similarity, + max_pairwise_similarity=max_pairwise_similarity, + ) + result = validate_candidate_family_clustering(cfc) + if not result.valid: + raise ValueError(f"Invalid CFC: {result.violations}") + return cfc + + +def format_candidate_family_clustering(cfc: CandidateFamilyClustering) -> str: + lines = [ + f"Candidate Family Clustering — {cfc.cfc_id}", + f"Family: {cfc.family_name}", + f"Novelty Verdict: {cfc.novelty_verdict}", + f"Clustering Method: {cfc.clustering_method} | Similarity Threshold: {cfc.similarity_threshold:.3f}", + f"Candidates: {cfc.n_candidates} ({', '.join(cfc.candidate_ids[:3])}{'...' if len(cfc.candidate_ids) > 3 else ''})", + f"TOY Excluded: {cfc.n_toy_excluded}", + f"Novelty Evidence: {', '.join(cfc.novelty_evidence_types)}", + ] + if cfc.known_family_hits: + lines.append(f"Known Family Hits: {', '.join(cfc.known_family_hits)}") + if cfc.min_pairwise_similarity is not None and cfc.max_pairwise_similarity is not None: + lines.append(f"Pairwise Similarity Range: [{cfc.min_pairwise_similarity:.3f}, {cfc.max_pairwise_similarity:.3f}]") + lines.append(f"Limitations: {cfc.limitations}") + if cfc.notes: + lines.append(f"Notes: {cfc.notes}") + lines.append("dry_lab_only: True (computational clustering result)") + return "\n".join(lines) diff --git a/tests/evidence/test_candidate_family_clustering.py b/tests/evidence/test_candidate_family_clustering.py new file mode 100644 index 00000000..e2a8d06f --- /dev/null +++ b/tests/evidence/test_candidate_family_clustering.py @@ -0,0 +1,438 @@ +import pytest +from src.openamp_foundry.evidence.candidate_family_clustering import ( + VALID_NOVELTY_EVIDENCE_TYPES, + VALID_CLUSTERING_METHODS, + VALID_NOVELTY_VERDICTS, + MIN_FAMILY_SIZE, + CandidateFamilyClustering, + CFCValidationResult, + validate_candidate_family_clustering, + build_candidate_family_clustering, + format_candidate_family_clustering, +) + + +def _valid_cfc(**kwargs): + defaults = dict( + cfc_id="CFC-0001", + family_name="Alpha-Helix Family 1", + candidate_ids=["AMPF-0001", "AMPF-0002", "AMPF-0003"], + clustering_method="sequence_identity", + similarity_threshold=0.70, + novelty_evidence_types=["blast_search_no_hits"], + novelty_verdict="novel_family", + n_candidates=3, + n_toy_excluded=0, + known_family_hits=[], + dry_lab_only=True, + limitations="Single clustering method; structural clustering not performed.", + notes="", + min_pairwise_similarity=0.65, + max_pairwise_similarity=0.82, + ) + defaults.update(kwargs) + return CandidateFamilyClustering(**defaults) + + +# ── Constants ────────────────────────────────────────────────────────────────── + +class TestConstants: + def test_novelty_evidence_types_is_frozenset(self): + assert isinstance(VALID_NOVELTY_EVIDENCE_TYPES, frozenset) + + def test_novelty_evidence_types_count(self): + assert len(VALID_NOVELTY_EVIDENCE_TYPES) == 6 + + def test_blast_search_no_hits_in_evidence_types(self): + assert "blast_search_no_hits" in VALID_NOVELTY_EVIDENCE_TYPES + + def test_hmm_profile_no_match_in_evidence_types(self): + assert "hmm_profile_no_match" in VALID_NOVELTY_EVIDENCE_TYPES + + def test_insufficient_evidence_in_evidence_types(self): + assert "insufficient_evidence" in VALID_NOVELTY_EVIDENCE_TYPES + + def test_clustering_methods_is_frozenset(self): + assert isinstance(VALID_CLUSTERING_METHODS, frozenset) + + def test_clustering_methods_count(self): + assert len(VALID_CLUSTERING_METHODS) == 5 + + def test_sequence_identity_in_clustering_methods(self): + assert "sequence_identity" in VALID_CLUSTERING_METHODS + + def test_embedding_distance_in_clustering_methods(self): + assert "embedding_distance" in VALID_CLUSTERING_METHODS + + def test_novelty_verdicts_is_frozenset(self): + assert isinstance(VALID_NOVELTY_VERDICTS, frozenset) + + def test_novelty_verdicts_count(self): + assert len(VALID_NOVELTY_VERDICTS) == 4 + + def test_novel_family_in_verdicts(self): + assert "novel_family" in VALID_NOVELTY_VERDICTS + + def test_variant_of_known_in_verdicts(self): + assert "variant_of_known" in VALID_NOVELTY_VERDICTS + + def test_min_family_size_is_2(self): + assert MIN_FAMILY_SIZE == 2 + + +# ── Dataclass ────────────────────────────────────────────────────────────────── + +class TestCandidateFamilyClusteringDataclass: + def test_instantiation(self): + c = _valid_cfc() + assert c.cfc_id == "CFC-0001" + + def test_dry_lab_only_true(self): + c = _valid_cfc() + assert c.dry_lab_only is True + + def test_candidate_ids_list(self): + c = _valid_cfc() + assert len(c.candidate_ids) == 3 + + def test_notes_default_empty(self): + c = _valid_cfc(notes="") + assert c.notes == "" + + def test_optional_pairwise_similarity(self): + c = _valid_cfc(min_pairwise_similarity=None, max_pairwise_similarity=None) + assert c.min_pairwise_similarity is None + + def test_known_family_hits_empty(self): + c = _valid_cfc() + assert c.known_family_hits == [] + + +# ── Validation ───────────────────────────────────────────────────────────────── + +class TestValidateCandidateFamilyClustering: + def test_valid_cfc_passes(self): + c = _valid_cfc() + result = validate_candidate_family_clustering(c) + assert result.valid + + def test_invalid_cfc_id_prefix(self): + c = _valid_cfc(cfc_id="BAD-0001") + result = validate_candidate_family_clustering(c) + assert not result.valid + assert any("cfc_id" in v for v in result.violations) + + def test_short_family_name_blocked(self): + c = _valid_cfc(family_name="AB") + result = validate_candidate_family_clustering(c) + assert not result.valid + + def test_empty_candidate_ids_blocked(self): + c = _valid_cfc(candidate_ids=[], n_candidates=0) + result = validate_candidate_family_clustering(c) + assert not result.valid + + def test_toy_candidate_id_blocked(self): + c = _valid_cfc(candidate_ids=["TOY-0001", "AMPF-0002", "AMPF-0003"]) + result = validate_candidate_family_clustering(c) + assert not result.valid + assert any("TOY-" in v for v in result.violations) + + def test_invalid_clustering_method_blocked(self): + c = _valid_cfc(clustering_method="magic_clustering") + result = validate_candidate_family_clustering(c) + assert not result.valid + + def test_similarity_threshold_above_1_blocked(self): + c = _valid_cfc(similarity_threshold=1.1) + result = validate_candidate_family_clustering(c) + assert not result.valid + + def test_similarity_threshold_negative_blocked(self): + c = _valid_cfc(similarity_threshold=-0.1) + result = validate_candidate_family_clustering(c) + assert not result.valid + + def test_invalid_novelty_evidence_type_blocked(self): + c = _valid_cfc(novelty_evidence_types=["magic_evidence"]) + result = validate_candidate_family_clustering(c) + assert not result.valid + + def test_empty_novelty_evidence_types_blocked(self): + c = _valid_cfc(novelty_evidence_types=[]) + result = validate_candidate_family_clustering(c) + assert not result.valid + + def test_invalid_novelty_verdict_blocked(self): + c = _valid_cfc(novelty_verdict="probably_novel") + result = validate_candidate_family_clustering(c) + assert not result.valid + + def test_n_candidates_below_min_blocked(self): + c = _valid_cfc(candidate_ids=["AMPF-0001"], n_candidates=1) + result = validate_candidate_family_clustering(c) + assert not result.valid + assert any("n_candidates" in v for v in result.violations) + + def test_n_candidates_mismatch_blocked(self): + c = _valid_cfc(n_candidates=5) + result = validate_candidate_family_clustering(c) + assert not result.valid + assert any("n_candidates" in v for v in result.violations) + + def test_n_toy_excluded_negative_blocked(self): + c = _valid_cfc(n_toy_excluded=-1) + result = validate_candidate_family_clustering(c) + assert not result.valid + + def test_novel_family_requires_hard_evidence(self): + c = _valid_cfc(novelty_verdict="novel_family", novelty_evidence_types=["expert_assessment"]) + result = validate_candidate_family_clustering(c) + assert not result.valid + assert any("hard evidence" in v for v in result.violations) + + def test_variant_of_known_requires_known_hits(self): + c = _valid_cfc(novelty_verdict="variant_of_known", known_family_hits=[]) + result = validate_candidate_family_clustering(c) + assert not result.valid + assert any("known_family_hits" in v for v in result.violations) + + def test_dry_lab_only_false_blocked(self): + c = _valid_cfc(dry_lab_only=False) + result = validate_candidate_family_clustering(c) + assert not result.valid + assert any("dry_lab_only" in v for v in result.violations) + + def test_empty_limitations_blocked(self): + c = _valid_cfc(limitations="") + result = validate_candidate_family_clustering(c) + assert not result.valid + + def test_min_pairwise_above_max_blocked(self): + c = _valid_cfc(min_pairwise_similarity=0.9, max_pairwise_similarity=0.5) + result = validate_candidate_family_clustering(c) + assert not result.valid + assert any("min_pairwise_similarity" in v for v in result.violations) + + def test_variant_of_known_with_hits_valid(self): + c = _valid_cfc( + novelty_verdict="variant_of_known", + novelty_evidence_types=["blast_search_no_hits"], + known_family_hits=["defensin_family_1"], + ) + result = validate_candidate_family_clustering(c) + assert result.valid + + def test_insufficient_evidence_verdict_valid(self): + c = _valid_cfc( + novelty_verdict="insufficient_evidence", + novelty_evidence_types=["insufficient_evidence"], + ) + result = validate_candidate_family_clustering(c) + assert result.valid + + def test_multiple_novelty_evidence_types_valid(self): + c = _valid_cfc(novelty_evidence_types=["blast_search_no_hits", "hmm_profile_no_match"]) + result = validate_candidate_family_clustering(c) + assert result.valid + + def test_similarity_threshold_boundary_0(self): + c = _valid_cfc(similarity_threshold=0.0) + result = validate_candidate_family_clustering(c) + assert result.valid + + def test_similarity_threshold_boundary_1(self): + c = _valid_cfc(similarity_threshold=1.0) + result = validate_candidate_family_clustering(c) + assert result.valid + + def test_no_pairwise_similarity_valid(self): + c = _valid_cfc(min_pairwise_similarity=None, max_pairwise_similarity=None) + result = validate_candidate_family_clustering(c) + assert result.valid + + def test_large_family_valid(self): + ids = [f"AMPF-{i:04d}" for i in range(1, 21)] + c = _valid_cfc(candidate_ids=ids, n_candidates=20) + result = validate_candidate_family_clustering(c) + assert result.valid + + +# ── Build ────────────────────────────────────────────────────────────────────── + +class TestBuildCandidateFamilyClustering: + def test_build_valid(self): + c = build_candidate_family_clustering( + cfc_id="CFC-0001", + family_name="Alpha-Helix Family 1", + candidate_ids=["AMPF-0001", "AMPF-0002", "AMPF-0003"], + clustering_method="sequence_identity", + similarity_threshold=0.70, + novelty_evidence_types=["blast_search_no_hits"], + novelty_verdict="novel_family", + n_toy_excluded=0, + known_family_hits=[], + limitations="Single clustering method; structural clustering not performed.", + ) + assert c.cfc_id == "CFC-0001" + assert c.dry_lab_only is True + assert c.n_candidates == 3 + + def test_build_sets_n_candidates_from_ids(self): + c = build_candidate_family_clustering( + cfc_id="CFC-0002", + family_name="Beta-Sheet Cluster", + candidate_ids=["AMPF-0010", "AMPF-0011"], + clustering_method="kmer_similarity", + similarity_threshold=0.80, + novelty_evidence_types=["hmm_profile_no_match"], + novelty_verdict="novel_family", + n_toy_excluded=2, + known_family_hits=[], + limitations="Limited to gram-positive bacteria; not tested on gram-negative.", + ) + assert c.n_candidates == 2 + + def test_build_enforces_dry_lab_only_true(self): + c = build_candidate_family_clustering( + cfc_id="CFC-0003", + family_name="Test Family XYZ", + candidate_ids=["AMPF-0020", "AMPF-0021"], + clustering_method="edit_distance", + similarity_threshold=0.60, + novelty_evidence_types=["structural_divergence"], + novelty_verdict="novel_family", + n_toy_excluded=0, + known_family_hits=[], + limitations="Structural prediction only; not experimentally validated.", + ) + assert c.dry_lab_only is True + + def test_build_raises_on_toy_id(self): + with pytest.raises(ValueError): + build_candidate_family_clustering( + cfc_id="CFC-0004", + family_name="Toy Family", + candidate_ids=["TOY-0001", "TOY-0002"], + clustering_method="sequence_identity", + similarity_threshold=0.70, + novelty_evidence_types=["blast_search_no_hits"], + novelty_verdict="novel_family", + n_toy_excluded=0, + known_family_hits=[], + limitations="This should fail because TOY- IDs are used.", + ) + + def test_build_raises_on_no_hard_evidence(self): + with pytest.raises(ValueError): + build_candidate_family_clustering( + cfc_id="CFC-0005", + family_name="Alleged Novel Family", + candidate_ids=["AMPF-0050", "AMPF-0051"], + clustering_method="manual_grouping", + similarity_threshold=0.50, + novelty_evidence_types=["expert_assessment"], + novelty_verdict="novel_family", + n_toy_excluded=0, + known_family_hits=[], + limitations="Expert opinion only; no database searches performed.", + ) + + def test_build_variant_of_known(self): + c = build_candidate_family_clustering( + cfc_id="CFC-0006", + family_name="Defensin Variant Cluster", + candidate_ids=["AMPF-0060", "AMPF-0061", "AMPF-0062"], + clustering_method="sequence_identity", + similarity_threshold=0.65, + novelty_evidence_types=["blast_search_no_hits"], + novelty_verdict="variant_of_known", + n_toy_excluded=1, + known_family_hits=["alpha_defensin_family"], + limitations="BLAST search only; HMM not performed.", + ) + assert c.novelty_verdict == "variant_of_known" + assert "alpha_defensin_family" in c.known_family_hits + + +# ── Format ───────────────────────────────────────────────────────────────────── + +class TestFormatCandidateFamilyClustering: + def _build(self, **kwargs): + defaults = dict( + cfc_id="CFC-0001", + family_name="Alpha-Helix Family 1", + candidate_ids=["AMPF-0001", "AMPF-0002", "AMPF-0003"], + clustering_method="sequence_identity", + similarity_threshold=0.70, + novelty_evidence_types=["blast_search_no_hits"], + novelty_verdict="novel_family", + n_toy_excluded=0, + known_family_hits=[], + limitations="Single clustering method; structural clustering not performed.", + ) + defaults.update(kwargs) + return build_candidate_family_clustering(**defaults) + + def test_format_returns_string(self): + c = self._build() + assert isinstance(format_candidate_family_clustering(c), str) + + def test_format_contains_cfc_id(self): + c = self._build() + assert "CFC-0001" in format_candidate_family_clustering(c) + + def test_format_contains_family_name(self): + c = self._build() + assert "Alpha-Helix Family 1" in format_candidate_family_clustering(c) + + def test_format_contains_novelty_verdict(self): + c = self._build() + assert "novel_family" in format_candidate_family_clustering(c) + + def test_format_contains_clustering_method(self): + c = self._build() + assert "sequence_identity" in format_candidate_family_clustering(c) + + def test_format_contains_similarity_threshold(self): + c = self._build() + assert "0.700" in format_candidate_family_clustering(c) + + def test_format_contains_candidate_count(self): + c = self._build() + assert "3" in format_candidate_family_clustering(c) + + def test_format_contains_novelty_evidence(self): + c = self._build() + assert "blast_search_no_hits" in format_candidate_family_clustering(c) + + def test_format_contains_dry_lab_only_true(self): + c = self._build() + assert "dry_lab_only: True" in format_candidate_family_clustering(c) + + def test_format_contains_limitations(self): + c = self._build() + assert "Single clustering method" in format_candidate_family_clustering(c) + + def test_format_contains_pairwise_range(self): + c = self._build(min_pairwise_similarity=0.65, max_pairwise_similarity=0.82) + assert "0.650" in format_candidate_family_clustering(c) + assert "0.820" in format_candidate_family_clustering(c) + + def test_format_contains_known_hits(self): + c = self._build( + novelty_verdict="variant_of_known", + known_family_hits=["defensin_family_1"], + ) + assert "defensin_family_1" in format_candidate_family_clustering(c) + + def test_format_multiline(self): + c = self._build() + lines = format_candidate_family_clustering(c).split("\n") + assert len(lines) >= 7 + + def test_format_truncates_long_candidate_list(self): + ids = [f"AMPF-{i:04d}" for i in range(1, 11)] + c = self._build(candidate_ids=ids) + out = format_candidate_family_clustering(c) + assert "..." in out diff --git a/tests/test_test_count_regression.py b/tests/test_test_count_regression.py index c93d9846..5dae853f 100644 --- a/tests/test_test_count_regression.py +++ b/tests/test_test_count_regression.py @@ -3,7 +3,7 @@ import subprocess import math -BASELINE = 8900 +BASELINE = 8963 def test_test_count_regression():