feat(fusion): add core multi-modal fuse() with per-disease attribution

src/fusion/engine.py

@@ -0,0 +1,105 @@
+"""Multi-modal fusion engine — combines MRI, EEG, and clinical signals into
+per-disease confidence with full attribution.
+"""
+from __future__ import annotations
+
+import math
+from typing import Callable
+
+from src.core.logger import get_logger
+from src.fusion import clinical as clinical_signals
+from src.fusion import weights as weight_registry
+from src.fusion.modality import signal_for_disease
+from src.fusion.types import (
+    ClinicalScores,
+    DiseaseScore,
+    FusionInput,
+    FusionOutput,
+    ModalityContribution,
+    ModalityPrediction,
+)
+
+logger = get_logger(__name__)
+
+_LOGIT_SCALE = 4.0  # tuned so a single saturated modality maps to ~0.88
+
+
+# Clinical-test name -> (signal_fn, attribute_on_ClinicalScores)
+_CLINICAL_FNS: dict[str, tuple[Callable[[float], float], str]] = {
+    "clinical_mmse":  (clinical_signals.mmse_to_signal,  "mmse"),
+    "clinical_moca":  (clinical_signals.moca_to_signal,  "moca"),
+    "clinical_updrs": (clinical_signals.updrs_to_signal, "updrs"),
+    "clinical_gait":  (clinical_signals.gait_to_signal,  "gait_speed_m_s"),
+    "clinical_age":   (clinical_signals.age_to_signal,   "age_years"),
+}
+
+
+def fuse(inp: FusionInput) -> FusionOutput:
+    """Combine all available modalities into a per-disease confidence."""
+    missing: list[str] = []
+    if inp.mri is None:
+        missing.append("mri")
+    if inp.eeg is None:
+        missing.append("eeg")
+
+    diseases: list[DiseaseScore] = []
+    for disease in weight_registry.available_diseases():
+        diseases.append(_score_one_disease(disease, inp))
+
+    top = max(diseases, key=lambda d: d.probability).disease
+    return FusionOutput(diseases=diseases, top_disease=top, missing_inputs=missing)
+
+
+def _score_one_disease(disease: str, inp: FusionInput) -> DiseaseScore:
+    weights = weight_registry.get_weights(disease)
+    contributions: list[ModalityContribution] = []
+
+    for modality_key, weight in weights.items():
+        signal = _signal_for_modality(modality_key, disease, inp.mri, inp.eeg, inp.clinical)
+        if signal is None:
+            continue
+        contributions.append(ModalityContribution(
+            modality=modality_key,
+            weight=weight,
+            signal=signal,
+            delta_logit=weight * signal,
+        ))
+
+    if not contributions:
+        logger.info("no signals available for disease=%s; returning baseline 0.5", disease)
+        return DiseaseScore(disease=disease, probability=0.5, contributions=[])
+
+    logit = sum(c.delta_logit for c in contributions)
+    probability = _sigmoid(_LOGIT_SCALE * logit)
+    return DiseaseScore(
+        disease=disease,
+        probability=probability,
+        contributions=contributions,
+    )
+
+
+def _signal_for_modality(
+    modality_key: str,
+    disease: str,
+    mri: ModalityPrediction | None,
+    eeg: ModalityPrediction | None,
+    clinical: ClinicalScores,
+) -> float | None:
+    if modality_key == "mri":
+        return signal_for_disease(mri, disease) if mri is not None else None
+    if modality_key == "eeg":
+        return signal_for_disease(eeg, disease) if eeg is not None else None
+    if modality_key in _CLINICAL_FNS:
+        fn, attr = _CLINICAL_FNS[modality_key]
+        value = getattr(clinical, attr, None)
+        return fn(value) if value is not None else None
+    logger.warning("unknown modality key in weights table: %s", modality_key)
+    return None
+
+
+def _sigmoid(x: float) -> float:
+    if x >= 0:
+        z = math.exp(-x)
+        return 1.0 / (1.0 + z)
+    z = math.exp(x)
+    return z / (1.0 + z)

tests/fusion/test_engine.py

@@ -0,0 +1,107 @@
+"""Tests for src.fusion.engine.fuse — the core multi-modal combiner."""
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+import pytest
+
+from src.fusion import engine
+from src.fusion.types import (
+    ClinicalScores,
+    FusionInput,
+    ModalityClassProb,
+    ModalityPrediction,
+)
+
+
+def _mri(prob_alz: float, prob_pd: float = 0.0) -> ModalityPrediction:
+    p_other = max(0.0, 1.0 - prob_alz - prob_pd)
+    items = [
+        ModalityClassProb(label_text="control", probability=p_other),
+        ModalityClassProb(label_text="alzheimers", probability=prob_alz),
+        ModalityClassProb(label_text="parkinsons", probability=prob_pd),
+    ]
+    top = max(items, key=lambda p: p.probability)
+    return ModalityPrediction(
+        label_text=top.label_text,
+        label=[p.label_text for p in items].index(top.label_text),
+        confidence=top.probability,
+        probabilities=items,
+    )
+
+
+class TestFuse:
+    def test_empty_input_returns_baseline_with_missing_listed(self) -> None:
+        out = engine.fuse(FusionInput())
+        assert {d.disease for d in out.diseases} >= {"alzheimers", "parkinsons", "other"}
+        for ds in out.diseases:
+            assert ds.probability == pytest.approx(0.5, abs=1e-6)
+            assert ds.contributions == []
+        assert "mri" in out.missing_inputs
+        assert "eeg" in out.missing_inputs
+
+    def test_mri_only_alzheimers_high(self) -> None:
+        inp = FusionInput(mri=_mri(prob_alz=0.9))
+        out = engine.fuse(inp)
+        alz = next(d for d in out.diseases if d.disease == "alzheimers")
+        assert alz.probability > 0.7
+        assert any(c.modality == "mri" for c in alz.contributions)
+        assert out.top_disease == "alzheimers"
+
+    def test_mri_eeg_agreement_boosts_above_either_alone(self) -> None:
+        only_mri = engine.fuse(FusionInput(mri=_mri(prob_alz=0.8)))
+        only_eeg = engine.fuse(FusionInput(eeg=_mri(prob_alz=0.8)))
+        both = engine.fuse(FusionInput(
+            mri=_mri(prob_alz=0.8), eeg=_mri(prob_alz=0.8),
+        ))
+
+        def alz(out: Any) -> float:
+            return next(d for d in out.diseases if d.disease == "alzheimers").probability
+
+        assert alz(both) > alz(only_mri)
+        assert alz(both) > alz(only_eeg)
+
+    def test_clinical_only_low_mmse_raises_alzheimers(self) -> None:
+        out = engine.fuse(FusionInput(clinical=ClinicalScores(mmse=10.0)))
+        alz = next(d for d in out.diseases if d.disease == "alzheimers")
+        assert alz.probability > 0.55
+        assert any(c.modality == "clinical_mmse" for c in alz.contributions)
+
+    def test_disagreement_moderates_confidence(self) -> None:
+        out = engine.fuse(FusionInput(
+            mri=_mri(prob_alz=0.85),
+            clinical=ClinicalScores(mmse=30.0),
+        ))
+        alz = next(d for d in out.diseases if d.disease == "alzheimers")
+        assert 0.5 < alz.probability < 0.78
+
+    def test_unknown_clinical_field_is_ignored_safely(self) -> None:
+        out = engine.fuse(FusionInput(clinical=ClinicalScores(age_years=80.0)))
+        assert out.top_disease in {"alzheimers", "parkinsons", "other"}
+
+    def test_engine_does_not_depend_on_bbb(self) -> None:
+        # Independence regression: fusion must not couple to BBB. A patient
+        # with only MRI/EEG/clinical data must produce a valid output even
+        # though no BBB module is involved.
+        import inspect
+        import src.fusion.engine as engine_mod
+        import src.fusion.weights as weights_mod
+        assert "bbb" not in inspect.getsource(engine_mod).lower()
+        for disease in weights_mod.available_diseases():
+            for key in weights_mod.get_weights(disease):
+                assert "bbb" not in key.lower(), (disease, key)
+
+    def test_warning_logged_when_disease_has_no_signals(
+        self, caplog: pytest.LogCaptureFixture
+    ) -> None:
+        # 'other' disease with no MRI/EEG inputs -> no signals available.
+        engine.logger.addHandler(caplog.handler)
+        caplog.handler.setLevel(logging.INFO)
+        try:
+            out = engine.fuse(FusionInput(clinical=ClinicalScores(mmse=10.0)))
+        finally:
+            engine.logger.removeHandler(caplog.handler)
+        other = next(d for d in out.diseases if d.disease == "other")
+        assert other.probability == pytest.approx(0.5, abs=1e-6)
+        assert other.contributions == []