diff --git a/.gitignore b/.gitignore index 4fb74c6..e839994 100644 --- a/.gitignore +++ b/.gitignore @@ -176,3 +176,6 @@ cython_debug/ # and can be added to the global gitignore or merged into this file. For a more nuclear # option (not recommended) you can uncomment the following to ignore the entire idea folder. #.idea/ + +# Local planning notes (not part of the shipped code) +PLAN.md diff --git a/compose.yml b/compose.yml index 8847600..5f8ff14 100644 --- a/compose.yml +++ b/compose.yml @@ -160,6 +160,20 @@ services: volumes: - ./logs:/app/logs - ./.env:/app/.env + filter_edges_by_statistical_significance: + container_name: filter_edges_by_statistical_significance + build: + context: . + dockerfile: workers/filter_edges_by_statistical_significance/Dockerfile + restart: unless-stopped + depends_on: + shepherd_db: + condition: service_healthy + shepherd_broker: + condition: service_healthy + volumes: + - ./logs:/app/logs + - ./.env:/app/.env finish_query: container_name: finish_query build: diff --git a/shepherd_utils/shared.py b/shepherd_utils/shared.py index b97f9ff..43dfd03 100644 --- a/shepherd_utils/shared.py +++ b/shepherd_utils/shared.py @@ -935,6 +935,56 @@ def merge_kgraph(og_message, new_message, source, logger: logging.Logger): return og_message +def filter_edges_by_statistical_significance(message, minimum_significance, logger): + """Remove KG edges whose significance band is below `minimum_significance`. + + Qualifier-less edges are kept. Analyses/results that can no longer satisfy the + query graph are pruned so the TRAPI stays valid; downstream filter_kgraph_orphans + removes orphaned nodes. (shepherd#134) + """ + from shepherd_utils.statistical_significance_qualifier import ( + SIGNIFICANCE_ORDINAL, + get_statistical_significance, + ) + + threshold = SIGNIFICANCE_ORDINAL.get(minimum_significance) + if threshold is None: + logger.error(f"Unknown minimum_significance '{minimum_significance}'") + return + msg = message.get("message", {}) + kg_edges = msg.get("knowledge_graph", {}).get("edges", {}) + + # 1. Edges explicitly below threshold (qualifier-less kept). + remove = { + eid + for eid, edge in kg_edges.items() + if (band := get_statistical_significance(edge)) is not None + and (ordinal := SIGNIFICANCE_ORDINAL.get(band)) is not None + and ordinal < threshold + } + if not remove: + return + + # 2. Prune analyses/results that lose a required edge binding. + for result in msg.get("results", []): + for analysis in result.get("analyses", []): + for qedge_key, bindings in analysis.get("edge_bindings", {}).items(): + analysis["edge_bindings"][qedge_key] = [ + b for b in bindings if b.get("id") not in remove + ] + result["analyses"] = [ + a + for a in result.get("analyses", []) + if a.get("edge_bindings") and all(a["edge_bindings"].values()) + ] + msg["results"] = [r for r in msg.get("results", []) if r.get("analyses")] + + # 3. Drop the edges from the KG in place. + for eid in remove: + del kg_edges[eid] + logger.info(f"Removed {len(remove)} edges below '{minimum_significance}'") + + def filter_kgraph_orphans(message, logger: logging.Logger): """Given a result-pruned message, filter out orphaned kgraph nodes and edges.""" try: diff --git a/shepherd_utils/statistical_significance_qualifier.py b/shepherd_utils/statistical_significance_qualifier.py new file mode 100644 index 0000000..59fe4c2 --- /dev/null +++ b/shepherd_utils/statistical_significance_qualifier.py @@ -0,0 +1,56 @@ +"""Shared helpers for biolink:statistical_significance_qualifier (shepherd#134). + +The qualifier (enum StatisticalSignificanceQualifierEnum) is_a statement_qualifier, +a descendant of `qualifier` in biolink-model, so BMT/Retriever route it into TRAPI +edge.qualifiers[]. get_statistical_significance() therefore reads edge['qualifiers'] +ONLY. (ARAX is likewise qualifier-only as of RTXteam/RTX#2859 — its defensive +edge.attributes lookup was removed; add a fallback here only if a KP is found to +send the qualifier as an attribute.) Shared by aragorn_score + arax_rank (ranking) +and the filter_edges_by_statistical_significance worker (filtering). +""" + +from typing import Any, Dict, Optional + +SIGNIFICANCE_QUALIFIER_TYPE_ID = "biolink:statistical_significance_qualifier" + +# Conservative ranking scores per band. TODO: revisit once all KGs populate the +# qualifier (rollout asymmetry: qualifier-bearing edges scored against edges that +# lack it entirely). Mirrors the RTX ARAX_ranker change (RTXteam/RTX#2858). +SIGNIFICANCE_BAND_SCORES: Dict[str, float] = { + "very_strongly_significant": 0.70, + "strongly_significant": 0.55, + "significant": 0.40, + "suggestive": 0.15, + "not_significant": 0.0, +} + +# Ordinal ranking for filtering (remove edges below a threshold). +SIGNIFICANCE_ORDINAL: Dict[str, int] = { + "very_strongly_significant": 4, + "strongly_significant": 3, + "significant": 2, + "suggestive": 1, + "not_significant": 0, +} + +# Source-agnostic trust weight applied to the band score in ranking (conservative; +# matches RTX trust=0.5). NOT routed through aragorn's per-source get_source_weight. +SIGNIFICANCE_SOURCE_WEIGHT: float = 0.5 + + +def _strip_biolink(value: Any) -> Optional[str]: + if isinstance(value, str) and value.startswith("biolink:"): + return value[len("biolink:"):] + return value + + +def get_statistical_significance(edge: Dict[str, Any]) -> Optional[str]: + """Return the bare significance band for a dict-based TRAPI edge, or None. + + Reads edge['qualifiers'] only (it is a biolink qualifier; BMT/Retriever route it + there). Strips any biolink: prefix from the value. + """ + for q in edge.get("qualifiers") or []: + if q.get("qualifier_type_id") == SIGNIFICANCE_QUALIFIER_TYPE_ID: + return _strip_biolink(q.get("qualifier_value")) + return None diff --git a/tests/unit/aragorn/test_aragorn.py b/tests/unit/aragorn/test_aragorn.py index c519a39..9353c68 100644 --- a/tests/unit/aragorn/test_aragorn.py +++ b/tests/unit/aragorn/test_aragorn.py @@ -34,6 +34,7 @@ async def test_aragorn_entrypoint(redis_mock, mocker): "aragorn.lookup", "aragorn.omnicorp", "aragorn.score", + "filter_edges_by_statistical_significance", "sort_results_score", "filter_results_top_n", "filter_kgraph_orphans", diff --git a/tests/unit/aragorn/test_aragorn_examine_query.py b/tests/unit/aragorn/test_aragorn_examine_query.py index 2fa983c..7b56219 100644 --- a/tests/unit/aragorn/test_aragorn_examine_query.py +++ b/tests/unit/aragorn/test_aragorn_examine_query.py @@ -225,6 +225,7 @@ async def test_aragorn_lookup_only_workflow(redis_mock, mocker): "aragorn.lookup", "aragorn.omnicorp", "aragorn.score", + "filter_edges_by_statistical_significance", "sort_results_score", "filter_results_top_n", "filter_kgraph_orphans", diff --git a/tests/unit/aragorn/test_aragorn_score_ranker.py b/tests/unit/aragorn/test_aragorn_score_ranker.py index 5bb28fa..625da45 100644 --- a/tests/unit/aragorn/test_aragorn_score_ranker.py +++ b/tests/unit/aragorn/test_aragorn_score_ranker.py @@ -673,3 +673,85 @@ def test_score_jaccard_like_returns_score_over_one_minus_score(): assert scored["analyses"][0]["score"] == pytest.approx( raw_score / (1 - raw_score) ) + + +# --- statistical significance qualifier (shepherd#134) -------------------- + + +def test_get_edge_values_extracts_significance_qualifier(): + """A qualifier-bearing edge gets a statistical_significance property.""" + edge = { + "subject": "A", + "object": "B", + "predicate": "biolink:related_to", + "sources": [ + {"resource_id": "infores:test", "resource_role": "primary_knowledge_source"} + ], + "qualifiers": [ + { + "qualifier_type_id": "biolink:statistical_significance_qualifier", + "qualifier_value": "very_strongly_significant", + } + ], + } + r = Ranker(_make_msg_with_edge(edge), logger) + vals = r.get_edge_values("e1") + assert "statistical_significance" in vals["infores:test"] + prop = vals["infores:test"]["statistical_significance"] + assert prop["value"] == "very_strongly_significant" + assert prop["weight"] > 0 + assert prop["weight"] == pytest.approx(0.70 * 0.5) + + +def test_get_edge_values_significance_strips_biolink_prefix(): + """biolink:-prefixed qualifier values are stripped.""" + edge = { + "subject": "A", + "object": "B", + "sources": [ + {"resource_id": "infores:test", "resource_role": "primary_knowledge_source"} + ], + "qualifiers": [ + { + "qualifier_type_id": "biolink:statistical_significance_qualifier", + "qualifier_value": "biolink:significant", + } + ], + } + r = Ranker(_make_msg_with_edge(edge), logger) + vals = r.get_edge_values("e1") + assert vals["infores:test"]["statistical_significance"]["value"] == "significant" + + +def test_not_significant_contributes_nothing(): + """Band score 0 -> property omitted -> no admittance contribution (no penalty).""" + edge = { + "subject": "A", + "object": "B", + "sources": [ + {"resource_id": "infores:test", "resource_role": "primary_knowledge_source"} + ], + "qualifiers": [ + { + "qualifier_type_id": "biolink:statistical_significance_qualifier", + "qualifier_value": "not_significant", + } + ], + } + r = Ranker(_make_msg_with_edge(edge), logger) + vals = r.get_edge_values("e1") + assert "statistical_significance" not in vals["infores:test"] + + +def test_qualifierless_edge_has_no_significance_property(): + """Edges without the qualifier get no statistical_significance property.""" + edge = { + "subject": "A", + "object": "B", + "sources": [ + {"resource_id": "infores:test", "resource_role": "primary_knowledge_source"} + ], + } + r = Ranker(_make_msg_with_edge(edge), logger) + vals = r.get_edge_values("e1") + assert "statistical_significance" not in vals["infores:test"] diff --git a/tests/unit/test_arax_rank_ranker.py b/tests/unit/test_arax_rank_ranker.py new file mode 100644 index 0000000..f2bca4d --- /dev/null +++ b/tests/unit/test_arax_rank_ranker.py @@ -0,0 +1,134 @@ +"""Tests for statistical significance qualifier scoring in arax_rank (shepherd#134).""" + +import logging + +from workers.arax_rank.ranker import ARAXRanker + +logger = logging.getLogger(__name__) + + +def _edge(**kw): + return {"subject": "A", "object": "B", "predicate": "biolink:related_to", **kw} + + +def test_significance_additive_boost(): + """A qualifier-bearing edge scores >= the same edge without the qualifier.""" + ranker = ARAXRanker(logger) + base = ranker._calculate_edge_confidence( + "infores:test--A--B", + _edge( + attributes=[ + { + "attribute_type_id": "biolink:pValue", + "original_attribute_name": "pValue", + "value": "0.001", + } + ] + ), + ) + boosted = ranker._calculate_edge_confidence( + "infores:test--A--B", + _edge( + attributes=[ + { + "attribute_type_id": "biolink:pValue", + "original_attribute_name": "pValue", + "value": "0.001", + } + ], + qualifiers=[ + { + "qualifier_type_id": "biolink:statistical_significance_qualifier", + "qualifier_value": "very_strongly_significant", + } + ], + ), + ) + assert boosted >= base # additive qualifier can only help or be neutral + + +def test_significance_score_mapping(): + """Each band maps to band_score * 0.5 trust, appended to the score list.""" + ranker = ARAXRanker(logger) + # Edge with no attributes and no qualifier -> base only + no_qual = ranker._calculate_edge_confidence("infores:test--A--B", _edge()) + # Edge with qualifier only (no attributes) -> base + qualifier boost + with_qual = ranker._calculate_edge_confidence( + "infores:test--A--B", + _edge( + qualifiers=[ + { + "qualifier_type_id": "biolink:statistical_significance_qualifier", + "qualifier_value": "significant", + } + ] + ), + ) + # significant = 0.40 * 0.5 = 0.20 additive boost + assert with_qual > no_qual + + +def test_not_significant_adds_nothing(): + """not_significant (score 0.0) adds no boost.""" + ranker = ARAXRanker(logger) + no_qual = ranker._calculate_edge_confidence("infores:test--A--B", _edge()) + not_sig = ranker._calculate_edge_confidence( + "infores:test--A--B", + _edge( + qualifiers=[ + { + "qualifier_type_id": "biolink:statistical_significance_qualifier", + "qualifier_value": "not_significant", + } + ] + ), + ) + assert not_sig == no_qual + + +def test_biolink_prefix_stripped(): + """biolink:-prefixed qualifier values are handled.""" + ranker = ARAXRanker(logger) + bare = ranker._calculate_edge_confidence( + "infores:test--A--B", + _edge( + qualifiers=[ + { + "qualifier_type_id": "biolink:statistical_significance_qualifier", + "qualifier_value": "strongly_significant", + } + ] + ), + ) + prefixed = ranker._calculate_edge_confidence( + "infores:test--A--B", + _edge( + qualifiers=[ + { + "qualifier_type_id": "biolink:statistical_significance_qualifier", + "qualifier_value": "biolink:strongly_significant", + } + ] + ), + ) + assert bare == prefixed + + +def test_qualifier_works_without_attributes(): + """Qualifier scoring works even for edges with no attributes at all.""" + ranker = ARAXRanker(logger) + # No attributes, no qualifier -> base only (0.5 for infores) + base_only = ranker._calculate_edge_confidence("infores:test--A--B", _edge()) + # No attributes, but has qualifier -> base + boost + with_qual = ranker._calculate_edge_confidence( + "infores:test--A--B", + _edge( + qualifiers=[ + { + "qualifier_type_id": "biolink:statistical_significance_qualifier", + "qualifier_value": "very_strongly_significant", + } + ] + ), + ) + assert with_qual > base_only diff --git a/tests/unit/test_bte.py b/tests/unit/test_bte.py index eb22d86..cd7825e 100644 --- a/tests/unit/test_bte.py +++ b/tests/unit/test_bte.py @@ -154,6 +154,7 @@ async def test_bte_lookup_workflow(redis_mock, mocker): "bte.lookup", "aragorn.omnicorp", "aragorn.score", + "filter_edges_by_statistical_significance", "sort_results_score", "filter_results_top_n", "filter_kgraph_orphans", diff --git a/tests/unit/test_filter_edges_by_statistical_significance.py b/tests/unit/test_filter_edges_by_statistical_significance.py new file mode 100644 index 0000000..1db4897 --- /dev/null +++ b/tests/unit/test_filter_edges_by_statistical_significance.py @@ -0,0 +1,180 @@ +"""Tests for filter_edges_by_statistical_significance (shepherd#134).""" + +import json +import logging + +import pytest + +from shepherd_utils.shared import filter_edges_by_statistical_significance + +logger = logging.getLogger(__name__) + + +def _edge(band): + e = {"subject": "A", "object": "B", "predicate": "biolink:related_to"} + if band: + e["qualifiers"] = [ + { + "qualifier_type_id": "biolink:statistical_significance_qualifier", + "qualifier_value": band, + } + ] + return e + + +def _msg(): + return { + "message": { + "knowledge_graph": { + "nodes": {"A": {}, "B": {}}, + "edges": { + "e_sig": _edge("significant"), + "e_sugg": _edge("suggestive"), + "e_notsig": _edge("not_significant"), + "e_none": _edge(None), + }, + }, + "results": [ + { + "node_bindings": {"n0": [{"id": "A"}], "n1": [{"id": "B"}]}, + "analyses": [{"edge_bindings": {"e0": [{"id": eid}]}}], + } + for eid in ("e_sig", "e_sugg", "e_notsig", "e_none") + ], + } + } + + +def test_ordinal_removal_keeps_qualifierless(): + m = _msg() + filter_edges_by_statistical_significance(m, "significant", logger) + edges = m["message"]["knowledge_graph"]["edges"] + assert "e_sig" in edges and "e_none" in edges # at threshold / no qualifier -> kept + assert "e_sugg" not in edges and "e_notsig" not in edges # below -> removed + + +def test_results_pruned_when_edge_removed(): + m = _msg() + filter_edges_by_statistical_significance(m, "significant", logger) + # results binding only a removed edge are dropped; e_sig/e_none results remain + assert len(m["message"]["results"]) == 2 + + +def test_suggestive_removes_only_not_significant(): + m = _msg() + filter_edges_by_statistical_significance(m, "suggestive", logger) + edges = m["message"]["knowledge_graph"]["edges"] + assert "e_notsig" not in edges + assert "e_sig" in edges and "e_sugg" in edges and "e_none" in edges + + +def test_no_removal_when_no_qualifiers(): + m = { + "message": { + "knowledge_graph": { + "nodes": {"A": {}, "B": {}}, + "edges": {"e1": _edge(None), "e2": _edge(None)}, + }, + "results": [ + { + "node_bindings": {"n0": [{"id": "A"}]}, + "analyses": [{"edge_bindings": {"e0": [{"id": "e1"}]}}], + } + ], + } + } + filter_edges_by_statistical_significance(m, "significant", logger) + assert len(m["message"]["knowledge_graph"]["edges"]) == 2 + assert len(m["message"]["results"]) == 1 + + +def test_unknown_threshold_logs_error_and_returns(): + m = _msg() + filter_edges_by_statistical_significance(m, "bogus", logger) + # Nothing removed + assert len(m["message"]["knowledge_graph"]["edges"]) == 4 + + +def test_very_strongly_significant_removes_all_below(): + m = _msg() + filter_edges_by_statistical_significance(m, "very_strongly_significant", logger) + edges = m["message"]["knowledge_graph"]["edges"] + # Only qualifier-less kept; all banded edges below threshold removed + assert "e_none" in edges + assert "e_sig" not in edges + assert "e_sugg" not in edges + assert "e_notsig" not in edges + + +# --- Worker parameter reading --- + + +@pytest.mark.asyncio +async def test_worker_reads_nested_parameters(redis_mock, mocker): + """The thin worker reads minimum_significance from the nested parameters dict.""" + from workers.filter_edges_by_statistical_significance.worker import do_filter + + msg = _msg() + mocker.patch( + "workers.filter_edges_by_statistical_significance.worker.get_message", + return_value=msg, + ) + mock_save = mocker.patch( + "workers.filter_edges_by_statistical_significance.worker.save_message", + ) + + task = [ + "task_id", + { + "response_id": "resp-1", + "workflow": json.dumps( + [ + { + "id": "filter_edges_by_statistical_significance", + "parameters": {"minimum_significance": "significant"}, + } + ] + ), + }, + ] + + await do_filter(task, logger) + + # With minimum_significance=significant, e_sugg and e_notsig removed + saved_msg = mock_save.call_args[0][1] + edges = saved_msg["message"]["knowledge_graph"]["edges"] + assert "e_sig" in edges and "e_none" in edges + assert "e_sugg" not in edges and "e_notsig" not in edges + + +@pytest.mark.asyncio +async def test_worker_defaults_to_suggestive(redis_mock, mocker): + """Without parameters, the worker defaults to suggestive (removes only not_significant).""" + from workers.filter_edges_by_statistical_significance.worker import do_filter + + msg = _msg() + mocker.patch( + "workers.filter_edges_by_statistical_significance.worker.get_message", + return_value=msg, + ) + mock_save = mocker.patch( + "workers.filter_edges_by_statistical_significance.worker.save_message", + ) + + task = [ + "task_id", + { + "response_id": "resp-1", + "workflow": json.dumps( + [{"id": "filter_edges_by_statistical_significance"}] + ), + }, + ] + + await do_filter(task, logger) + + saved_msg = mock_save.call_args[0][1] + edges = saved_msg["message"]["knowledge_graph"]["edges"] + # Default suggestive: only not_significant removed + assert "e_notsig" not in edges + assert "e_sig" in edges and "e_sugg" in edges and "e_none" in edges diff --git a/tests/unit/test_statistical_significance_qualifier.py b/tests/unit/test_statistical_significance_qualifier.py new file mode 100644 index 0000000..eb9732d --- /dev/null +++ b/tests/unit/test_statistical_significance_qualifier.py @@ -0,0 +1,86 @@ +"""Tests for shepherd_utils.statistical_significance_qualifier (shepherd#134).""" + +from shepherd_utils.statistical_significance_qualifier import ( + SIGNIFICANCE_BAND_SCORES, + SIGNIFICANCE_ORDINAL, + get_statistical_significance, +) + + +def test_band_scores_descending(): + bands = [ + "very_strongly_significant", + "strongly_significant", + "significant", + "suggestive", + "not_significant", + ] + scores = [SIGNIFICANCE_BAND_SCORES[b] for b in bands] + assert scores == sorted(scores, reverse=True) and scores[-1] == 0.0 + + +def test_ordinal_matches_band_order(): + assert SIGNIFICANCE_ORDINAL["very_strongly_significant"] == 4 + assert SIGNIFICANCE_ORDINAL["not_significant"] == 0 + + +def test_lookup_in_qualifiers(): + edge = { + "qualifiers": [ + { + "qualifier_type_id": "biolink:statistical_significance_qualifier", + "qualifier_value": "significant", + } + ] + } + assert get_statistical_significance(edge) == "significant" + + +def test_attributes_are_ignored(): + # Qualifiers-only by design: an attributes-only qualifier is NOT read + # (matches ARAX, qualifier-only as of RTX#2859). + edge = { + "attributes": [ + { + "attribute_type_id": "biolink:statistical_significance_qualifier", + "value": "suggestive", + } + ] + } + assert get_statistical_significance(edge) is None + + +def test_lookup_strips_biolink_prefix(): + edge = { + "qualifiers": [ + { + "qualifier_type_id": "biolink:statistical_significance_qualifier", + "qualifier_value": "biolink:significant", + } + ] + } + assert get_statistical_significance(edge) == "significant" + + +def test_only_qualifiers_read(): + # The band comes from edge['qualifiers']; attributes are not consulted. + edge = { + "qualifiers": [ + { + "qualifier_type_id": "biolink:statistical_significance_qualifier", + "qualifier_value": "significant", + } + ], + "attributes": [ + { + "attribute_type_id": "biolink:statistical_significance_qualifier", + "value": "not_significant", + } + ], + } + assert get_statistical_significance(edge) == "significant" + + +def test_lookup_none_when_absent(): + assert get_statistical_significance({"attributes": []}) is None + assert get_statistical_significance({}) is None diff --git a/workers/aragorn/worker.py b/workers/aragorn/worker.py index 4bdde0d..499f270 100644 --- a/workers/aragorn/worker.py +++ b/workers/aragorn/worker.py @@ -36,6 +36,8 @@ async def aragorn(task, logger: logging.Logger): {"id": "aragorn.lookup"}, {"id": "aragorn.omnicorp"}, {"id": "aragorn.score"}, + {"id": "filter_edges_by_statistical_significance", + "parameters": {"minimum_significance": "suggestive"}}, {"id": "sort_results_score"}, {"id": "filter_results_top_n", "parameters": {"max_results": 500}}, {"id": "filter_kgraph_orphans"}, @@ -54,6 +56,8 @@ async def aragorn(task, logger: logging.Logger): {"id": "aragorn.lookup"}, {"id": "aragorn.omnicorp"}, {"id": "aragorn.score"}, + {"id": "filter_edges_by_statistical_significance", + "parameters": {"minimum_significance": "suggestive"}}, {"id": "sort_results_score"}, {"id": "filter_results_top_n", "parameters": {"max_results": 500}}, {"id": "filter_kgraph_orphans"}, diff --git a/workers/aragorn_score/worker.py b/workers/aragorn_score/worker.py index c4f7c77..e8c83ef 100644 --- a/workers/aragorn_score/worker.py +++ b/workers/aragorn_score/worker.py @@ -17,6 +17,11 @@ from shepherd_utils.otel import setup_tracer from shepherd_utils.process_pool import ProcessPoolManager from shepherd_utils.shared import get_tasks, run_task_lifecycle +from shepherd_utils.statistical_significance_qualifier import ( + SIGNIFICANCE_BAND_SCORES, + SIGNIFICANCE_SOURCE_WEIGHT, + get_statistical_significance, +) # Queue name STREAM = "aragorn.score" @@ -890,6 +895,7 @@ def get_edge_values(self, edge_id): "literature_coocurrence": None, "p_value": None, "affinity": None, + "statistical_significance": None, } # Look through attributes and @@ -998,6 +1004,10 @@ def get_edge_values(self, edge_id): if orig_attr_name == "biolink:tmkp_confidence_score": usable_edge_attr["confidence_score"] = attribute.get("value", 0) + # Qualifier lives in edge["qualifiers"] (BMT/Retriever path); the + # attribute loop above won't see it. + usable_edge_attr["statistical_significance"] = get_statistical_significance(edge) + # At this point we have all of the information extracted from the edge # We have have looked through all attributes and updated # usable_edge_attr. Now we can construct the edge values using these @@ -1091,6 +1101,17 @@ def get_edge_values(self, edge_id): "weight": property_w * source_w, } + if usable_edge_attr["statistical_significance"] is not None: + band = usable_edge_attr["statistical_significance"] + property_w = SIGNIFICANCE_BAND_SCORES.get(band, 0.0) + if property_w > 0: + this_edge_vals[edge_source]["statistical_significance"] = { + "value": band, + "property_weight": property_w, + "source_weight": SIGNIFICANCE_SOURCE_WEIGHT, + "weight": property_w * SIGNIFICANCE_SOURCE_WEIGHT, + } + # Cache it self.edge_values[edge_id] = this_edge_vals return this_edge_vals diff --git a/workers/arax_rank/ranker.py b/workers/arax_rank/ranker.py index 3be4663..0042b16 100644 --- a/workers/arax_rank/ranker.py +++ b/workers/arax_rank/ranker.py @@ -24,6 +24,12 @@ import numpy.typing as npt import scipy.stats +from shepherd_utils.statistical_significance_qualifier import ( + SIGNIFICANCE_BAND_SCORES, + SIGNIFICANCE_SOURCE_WEIGHT, + get_statistical_significance, +) + # Default confidence for manual agent edges (matches ARAX_ranker.py line 24) EDGE_CONFIDENCE_MANUAL_AGENT = 0.90 @@ -294,6 +300,17 @@ def _calculate_edge_confidence(self, edge_key: str, edge: Dict) -> float: if normalized_score > 0: edge_attribute_score_list.append(normalized_score) + # Statistical significance qualifier is carried in edge["qualifiers"] + # (not attributes). Looked up separately (categorical bypass) so the + # enum string never hits the numeric attribute normalizer. Mirrors RTX + # _get_significance_qualifier_value + _significance_trust_weight + # (RTXteam/RTX#2859). + sig_value = get_statistical_significance(edge) + if sig_value is not None: + sig_score = SIGNIFICANCE_BAND_SCORES.get(sig_value, 0.0) + if sig_score > 0: + edge_attribute_score_list.append(sig_score * SIGNIFICANCE_SOURCE_WEIGHT) + # If no attributes scored, return base score (ARAX_ranker.py lines 379-384) if len(edge_attribute_score_list) == 0: return base diff --git a/workers/bte/worker.py b/workers/bte/worker.py index c882263..dcebed6 100644 --- a/workers/bte/worker.py +++ b/workers/bte/worker.py @@ -38,6 +38,8 @@ async def bte(task, logger: logging.Logger): {"id": "bte.lookup"}, {"id": "aragorn.omnicorp"}, {"id": "aragorn.score"}, + {"id": "filter_edges_by_statistical_significance", + "parameters": {"minimum_significance": "suggestive"}}, {"id": "sort_results_score"}, {"id": "filter_results_top_n", "parameters": {"max_results": 500}}, {"id": "filter_kgraph_orphans"}, @@ -47,6 +49,8 @@ async def bte(task, logger: logging.Logger): {"id": "bte.lookup"}, {"id": "aragorn.omnicorp"}, {"id": "aragorn.score"}, + {"id": "filter_edges_by_statistical_significance", + "parameters": {"minimum_significance": "suggestive"}}, {"id": "sort_results_score"}, {"id": "filter_results_top_n", "parameters": {"max_results": 500}}, {"id": "filter_kgraph_orphans"}, diff --git a/workers/filter_edges_by_statistical_significance/Dockerfile b/workers/filter_edges_by_statistical_significance/Dockerfile new file mode 100644 index 0000000..5bf63b2 --- /dev/null +++ b/workers/filter_edges_by_statistical_significance/Dockerfile @@ -0,0 +1,30 @@ +# Use RENCI python base image +FROM ghcr.io/translatorsri/renci-python-image:3.12.13 + +# Add image info +LABEL org.opencontainers.image.source https://github.com/BioPack-team/shepherd + +ENV PYTHONHASHSEED=0 + +# set up requirements +WORKDIR /app + +# make sure all is writeable for the nru USER later on +RUN chmod -R 777 . + +# Install requirements +COPY ./shepherd_utils ./shepherd_utils +COPY ./pyproject.toml . +RUN pip install . + +COPY ./workers/filter_edges_by_statistical_significance/requirements.txt . +RUN pip install -r requirements.txt + +# switch to the non-root user (nru). defined in the base image +USER nru + +# Copy in files +COPY ./workers/filter_edges_by_statistical_significance . + +# Variables that can be overriden +CMD ["python", "worker.py"] diff --git a/workers/filter_edges_by_statistical_significance/__init__.py b/workers/filter_edges_by_statistical_significance/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/workers/filter_edges_by_statistical_significance/requirements.txt b/workers/filter_edges_by_statistical_significance/requirements.txt new file mode 100644 index 0000000..e69de29 diff --git a/workers/filter_edges_by_statistical_significance/worker.py b/workers/filter_edges_by_statistical_significance/worker.py new file mode 100644 index 0000000..c9f1545 --- /dev/null +++ b/workers/filter_edges_by_statistical_significance/worker.py @@ -0,0 +1,64 @@ +"""Filter Edges By Statistical Significance Worker (shepherd#134).""" + +import asyncio +import json +import logging +import uuid + +from shepherd_utils.db import get_message, save_message +from shepherd_utils.otel import setup_tracer +from shepherd_utils.shared import ( + filter_edges_by_statistical_significance, + get_tasks, + run_task_lifecycle, +) + +# Queue name +STREAM = "filter_edges_by_statistical_significance" +GROUP = "consumer" +CONSUMER = str(uuid.uuid4())[:8] +TASK_LIMIT = 10 +DEFAULT_MINIMUM_SIGNIFICANCE = "suggestive" # removes only not_significant by default +tracer = setup_tracer(STREAM) + + +async def do_filter(task, logger: logging.Logger): + """Remove KG edges whose significance band is below the threshold.""" + response_id = task[1]["response_id"] + workflow = json.loads(task[1]["workflow"]) + current_op = workflow[0] or {} + params = current_op.get("parameters") or {} + minimum = ( + params.get("minimum_significance") + or current_op.get("minimum_significance") + or DEFAULT_MINIMUM_SIGNIFICANCE + ) + message = await get_message(response_id, logger) + filter_edges_by_statistical_significance(message, minimum, logger) + await save_message(response_id, message, logger) + + +async def process_task(task, parent_ctx, logger: logging.Logger, limiter): + """Process a given task and ACK in redis.""" + await run_task_lifecycle( + STREAM, GROUP, task, parent_ctx, logger, limiter, do_filter + ) + + +async def poll_for_tasks(): + """On initialization, poll indefinitely for available tasks.""" + while True: + try: + async for task, parent_ctx, logger, limiter in get_tasks( + STREAM, GROUP, CONSUMER, TASK_LIMIT + ): + asyncio.create_task(process_task(task, parent_ctx, logger, limiter)) + except asyncio.CancelledError: + logging.info("Poll loop cancelled, shutting down.") + except Exception as e: + logging.error(f"Error in task polling loop: {e}", exc_info=True) + await asyncio.sleep(5) # back off before retrying + + +if __name__ == "__main__": + asyncio.run(poll_for_tasks())