Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions .gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -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
46 changes: 46 additions & 0 deletions shepherd_utils/statistical_significance_qualifier.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,46 @@
"""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).
"""

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,
}

# 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
82 changes: 82 additions & 0 deletions tests/unit/aragorn/test_aragorn_score_ranker.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"]
134 changes: 134 additions & 0 deletions tests/unit/test_arax_rank_ranker.py
Original file line number Diff line number Diff line change
@@ -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
80 changes: 80 additions & 0 deletions tests/unit/test_statistical_significance_qualifier.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,80 @@
"""Tests for shepherd_utils.statistical_significance_qualifier (shepherd#134)."""

from shepherd_utils.statistical_significance_qualifier import (
SIGNIFICANCE_BAND_SCORES,
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_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
Loading