LNG_UB_LIGASE_UNIT: a ubiquitination does not deplete its own ligase (deltasignal specs/089), default on

PIPELINE_SETTINGS.md

@@ -74,6 +74,7 @@ what NOT to touch. Overriding is for A/B benchmarking only.
 | `LNG_CAP_POOLS` | `1` | past the variant cap, one OR pool per set instead of an all-required bundle (deltasignal specs/045) |
 | `LNG_RECYCLE_POOLS` | `1` | write a modified protein's return to the complex it starts in as a pool with several resting states, plus a structural `recycle` edge from the free copy to the starting copy in `boundary_edges.csv` (p130/p107 in Mitotic G1; deltasignal specs/081) |
 | `LNG_RELEASE_POOLS` | `1` | a protein bound, modified and released by a complex, whose released copy carries on (beta-catenin and the destruction complex), as a free/bound pool whose released copy reads the free state; writes `released_twin` and `recycle` edges (deltasignal specs/086) |
+| `LNG_UB_LIGASE_UNIT` | `1` | a ubiquitination does not deplete its own ligase: a leaf of the catalyst's curated `activeUnit`, or of a complex released unchanged, that leaves the reaction unmodified is not the substrate (CKS1B under SCF(Skp2); deltasignal specs/089) |
 | `LNG_CATALYST_BUNDLE` | `0` | legacy; subsumed by `LNG_COMPLEX_AS_NODE` |
 | `LNG_HANDOFF_EDGES` | `0` | precedingEvent hand-off bridges (net-negative, off) |
 

src/env_flags.py

@@ -34,6 +34,7 @@
     "LNG_SET_EXPAND": True,
     "LNG_SET_MEMBERS_OR": False,
     "LNG_SET_POOL": True,
+    "LNG_UB_LIGASE_UNIT": True,       # default since deltasignal specs/089 (tuning +22, experimental +3, held-out -2 traced)
     "LNG_VARIANT_NODES": True,   # default since deltasignal specs/046 (vn7: held-out +130, experimental -10 ns)
 }
 

src/logic_network_generator.py

@@ -1444,6 +1444,63 @@ def base_of(e: Dict[str, Any]) -> str:
                 f"{sum(len(set(v)) for v in pooled.values())} catalyst variants")
 
 
+def _ub_substrates(in_leaves: List[Tuple[Optional[str], Optional[List[str]]]],
+                   output_genes: Set[str], output_leaves: Set[str],
+                   ligase_leaves: Set[str]) -> Tuple[Set[str], Set[str]]:
+    """A ubiquitination's substrate: the input leaves whose gene also appears
+    in the output (the modified form), as (genes, leaf stIds).
+
+    deltasignal specs/089: a leaf of the ligase (``ligase_leaves``) is not the
+    substrate when its own stId is also an output leaf, i.e. it leaves the
+    reaction unmodified. Without this, SCF(Skp2):Cks1 drained its own free
+    CKS1B. A ligase that is itself modified (autoubiquitination: Ub-BRAP has
+    another stId) stays a substrate, and so do the substrate's partners
+    (EGFR and GRB2 under CBL, specs/078)."""
+    subst_stids: Set[str] = set()
+    subst_genes: Set[str] = set()
+    if not output_genes:
+        return subst_genes, subst_stids
+    for leaf_stid, leaf_genes in in_leaves:
+        if not leaf_stid or not leaf_genes:
+            continue
+        if leaf_stid in ligase_leaves and leaf_stid in output_leaves:
+            continue
+        common = output_genes & set(leaf_genes)
+        if common:
+            subst_stids.add(leaf_stid)
+            subst_genes.update(common)
+    return subst_genes, subst_stids
+
+
+def _ub_ligase_leaves(rstids: List[str]) -> Dict[str, Set[str]]:
+    """The ligase's leaf stIds in each ubiquitination (deltasignal specs/089):
+    the leaves of its catalyst activity's curated ``activeUnit``; or, where no
+    activeUnit is curated, the leaves of an output entity that is also a direct
+    component of an input complex (the ligase released unchanged, as
+    CUL1:SKP1:SKP2:CKS1B from R-HSA-1363331)."""
+    from src.neo4j_connector import get_graph
+    leaf = ("OPTIONAL MATCH (x)-[:hasComponent|hasMember|hasCandidate*0..3]->(l:PhysicalEntity) "
+            "RETURN id AS rxn, l.stId AS leaf_stid")
+    au_rows = get_graph().run(
+        "UNWIND $ids AS id "
+        "MATCH (:ReactionLikeEvent {stId: id})-[:catalystActivity]->(:CatalystActivity)-[:activeUnit]->(x) "
+        + leaf, ids=rstids).data()
+    out: Dict[str, Set[str]] = {}
+    for r in au_rows:
+        if r["leaf_stid"]:
+            out.setdefault(r["rxn"], set()).add(r["leaf_stid"])
+    rest = [r for r in rstids if r not in out]
+    if rest:
+        rel_rows = get_graph().run(
+            "UNWIND $ids AS id "
+            "MATCH (rx:ReactionLikeEvent {stId: id})-[:output]->(x)<-[:hasComponent]-(:Complex)<-[:input]-(rx) "
+            + leaf, ids=rest).data()
+        for r in rel_rows:
+            if r["leaf_stid"]:
+                out.setdefault(r["rxn"], set()).add(r["leaf_stid"])
+    return out
+
+
 def _emit_substrate_depletion_edges(
     pathway_logic_network_data: List[Dict[str, Any]],
     reactome_id_to_uuid: Dict[str, str],
@@ -1556,32 +1613,30 @@ def _emit_substrate_depletion_edges(
                 "WHERE NOT o.stId IN $ub_stids "
                 "OPTIONAL MATCH (o)-[:hasComponent|hasMember|hasCandidate*0..3]->(leaf:PhysicalEntity) "
                 "OPTIONAL MATCH (leaf)-[:referenceEntity]->(re:ReferenceEntity) "
-                "RETURN id AS rxn, re.geneName AS genes",
+                "RETURN id AS rxn, leaf.stId AS leaf_stid, re.geneName AS genes",
                 ids=ubiq_rstids,
                 ub_stids=list(_UBIQUITIN_STIDS),
             ).data() if ubiq_rstids else []
+            # deltasignal specs/089: the ligase's own leaves, which a
+            # ubiquitination carries through rather than modifies.
+            ligase_by_rxn = _ub_ligase_leaves(ubiq_rstids) \
+                if ubiq_rstids and env_flag("LNG_UB_LIGASE_UNIT") else {}
             in_by_rxn: Dict[str, List[Tuple[Optional[str], Optional[List[str]]]]] = {}
             for r in in_rows:
                 in_by_rxn.setdefault(r["rxn"], []).append((r["leaf_stid"], r["genes"]))
             out_by_rxn: Dict[str, Set[str]] = {}
+            out_leaves_by_rxn: Dict[str, Set[str]] = {}
             for r in out_rows:
+                if r.get("leaf_stid"):
+                    out_leaves_by_rxn.setdefault(r["rxn"], set()).add(r["leaf_stid"])
                 g = r["genes"]
                 if g:
                     for gn in g:
                         out_by_rxn.setdefault(r["rxn"], set()).add(gn)
             for rxn in ubiq_rstids:
-                output_genes = out_by_rxn.get(rxn, set())
-                if not output_genes: continue
-                # The substrate: input-leaf proteins whose gene also appears
-                # in the output (modified form). Collect their leaf stids.
-                subst_stids: Set[str] = set()
-                subst_genes: Set[str] = set()
-                for leaf_stid, leaf_genes in in_by_rxn.get(rxn, []):
-                    if not leaf_stid or not leaf_genes: continue
-                    common = output_genes & set(leaf_genes)
-                    if common:
-                        subst_stids.add(leaf_stid)
-                        subst_genes.update(common)
+                subst_genes, subst_stids = _ub_substrates(
+                    in_by_rxn.get(rxn, []), out_by_rxn.get(rxn, set()),
+                    out_leaves_by_rxn.get(rxn, set()), ligase_by_rxn.get(rxn, set()))
                 if subst_stids:
                     ubiquitin_subst_by_rstid[rxn] = (subst_genes, subst_stids)
         except Exception as exc:

tests/test_ub_ligase_unit.py

@@ -0,0 +1,108 @@
+"""deltasignal specs/089: a ubiquitination does not deplete its own ligase
+(LNG_UB_LIGASE_UNIT)."""
+from typing import Any, Dict
+
+import pandas as pd
+import pytest
+
+import src.logic_network_generator as lng
+import src.neo4j_connector as nc
+
+UB = "R-HSA-113595"
+RX = "R-HSA-1363331"          # SCF(Skp2) ubiquitinates p130 and is released
+CKS1B, RBL2, UB_RBL2 = "R-HSA-CKS1B", "R-HSA-RBL2", "R-HSA-UB-RBL2"
+
+
+def _e(a, b, et, pn="pos", ao="and"):
+    return {"source_id": a, "target_id": b, "pos_neg": pn, "and_or": ao, "edge_type": et,
+            "stoichiometry": 1}
+
+
+# --- the substrate rule on its own ------------------------------------------
+
+def test_a_ligase_leaf_carried_unchanged_is_not_the_substrate():
+    genes, stids = lng._ub_substrates([(CKS1B, ["CKS1B"]), (RBL2, ["RBL2"])],
+                                      {"CKS1B", "RBL2"}, {CKS1B, UB_RBL2}, {CKS1B})
+    assert stids == {RBL2} and genes == {"RBL2"}
+
+
+def test_without_ligase_leaves_every_shared_gene_is_the_substrate():
+    _, stids = lng._ub_substrates([(CKS1B, ["CKS1B"]), (RBL2, ["RBL2"])],
+                                  {"CKS1B", "RBL2"}, {CKS1B, UB_RBL2}, set())
+    assert stids == {CKS1B, RBL2}
+
+
+def test_an_autoubiquitinated_ligase_stays_the_substrate():
+    # BRAP autoubiquitinates: BRAP is the active unit, but its output form is
+    # Ub-BRAP, another stId, so it is modified and still depleted.
+    _, stids = lng._ub_substrates([("R-HSA-BRAP", ["BRAP"])], {"BRAP"},
+                                  {"R-HSA-UB-BRAP"}, {"R-HSA-BRAP"})
+    assert stids == {"R-HSA-BRAP"}
+
+
+def test_the_substrates_partners_are_kept():
+    # CBL ubiquitinates EGFR: CBL (active unit) is dropped, the receptor complex
+    # partners are still depleted (specs/078 lost the CBL class by dropping them).
+    ins = [("R-HSA-CBL", ["CBL"]), ("R-HSA-EGFR", ["EGFR"]), ("R-HSA-GRB2", ["GRB2"])]
+    _, stids = lng._ub_substrates(ins, {"CBL", "EGFR", "GRB2"},
+                                  {"R-HSA-CBL", "R-HSA-EGFR", "R-HSA-GRB2"}, {"R-HSA-CBL"})
+    assert stids == {"R-HSA-EGFR", "R-HSA-GRB2"}
+
+
+# --- the whole pass, against a stub graph ------------------------------------
+
+class _Result:
+    def __init__(self, rows): self.rows = rows
+    def data(self): return self.rows
+
+
+class _Graph:
+    """Answers the ubiquitin pass's queries for one reaction: catalyst C
+    (p-RBL2:SCF(Skp2):Cks1) + Ub -> Ub-p-RBL2 + the released ligase, no activeUnit."""
+    def run(self, q: str, **kw: Any) -> _Result:
+        if "activeUnit" in q:
+            return _Result([])
+        if "<-[:hasComponent]-(:Complex)<-[:input]-" in q:
+            return _Result([{"rxn": RX, "leaf_stid": CKS1B}])
+        if "-[:output]->" in q:
+            return _Result([{"rxn": RX, "leaf_stid": CKS1B, "genes": ["CKS1B"]},
+                            {"rxn": RX, "leaf_stid": UB_RBL2, "genes": ["RBL2"]}])
+        if "-[:input]->(inp" in q:
+            return _Result([{"rxn": RX, "leaf_stid": CKS1B, "genes": ["CKS1B"]},
+                            {"rxn": RX, "leaf_stid": RBL2, "genes": ["RBL2"]}])
+        if "ub.stId IN $ub_stids" in q:
+            return _Result([{"rxn": RX}])
+        raise AssertionError(q)
+
+
+@pytest.fixture
+def pass_(monkeypatch):
+    monkeypatch.setattr(nc, "get_graph", lambda: _Graph())
+    monkeypatch.setattr(lng, "_cofactor_stids", lambda: frozenset())
+    monkeypatch.setenv("LNG_VARIANT_NODES", "0")
+
+    def run() -> set:
+        data = [_e("u-c", "rx", "catalyst"), _e("u-c", "rx", "input"), _e("rx", "u-out", "output"),
+                _e("u-cks", "x1", "input"), _e("u-rbl2", "x2", "input")]
+        r2u: Dict[str, str] = {"u-c": "R-HSA-1363326", "u-out": UB_RBL2,
+                               "u-cks": CKS1B, "u-rbl2": RBL2}
+        cmap = pd.DataFrame([{"reaction_uuid": "rx", "reaction_id": RX}])
+        lng._emit_substrate_depletion_edges(data, r2u, cmap)
+        return {(e["source_id"], e["target_id"]) for e in data if e["edge_type"] == "depletion"}
+    return run
+
+
+def test_off_the_ligase_drains_its_own_subunit(pass_, monkeypatch):
+    monkeypatch.setenv("LNG_UB_LIGASE_UNIT", "0")
+    assert pass_() == {("u-c", "u-cks"), ("u-c", "u-rbl2")}
+
+
+def test_on_only_the_substrate_is_drained(pass_, monkeypatch):
+    monkeypatch.setenv("LNG_UB_LIGASE_UNIT", "1")
+    assert pass_() == {("u-c", "u-rbl2")}
+
+
+def test_the_rule_is_the_default(pass_, monkeypatch):
+    monkeypatch.delenv("LNG_UB_LIGASE_UNIT", raising=False)
+    assert lng.env_flag("LNG_UB_LIGASE_UNIT") is True
+    assert pass_() == {("u-c", "u-rbl2")}