Spaces:
Sleeping
Sleeping
File size: 3,710 Bytes
0fff343 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 | """H1 verification — known-answer check, NOT blind discovery.
Composes the DSL operators over the NAMED expression matrix (real gene symbols)
with hand-picked MMR and immune gene panels, and reports the four numbers the
H1 program should produce on the usable MSI-H-vs-MSS CRC cohort:
1. medians of `mmr_score` and `immune_score` split by MSI status — MSI-H should
sit LOW on the MMR score and HIGH on the immune score;
2. `Effect(mmr_score, immune_score, adjust={stage, age})` — the adjusted partial
correlation vs the unadjusted pearson on the same rows, and the n used;
3. `Fit(state=(mmr_score, immune_score)) → MSI-H` — held-out AUROC and
balanced accuracy.
This module is biology-aware on purpose. The blind-discovery path in
[`engine`](../engine) will instead consume the airgapped view and is forbidden
from importing these gene lists.
"""
from __future__ import annotations
from dataclasses import dataclass
import pandas as pd
from dsl import Cohort, Effect, EffectResult, Fit, FitResult, Load, Reduce, Select
MMR_GENES: list[str] = ["MLH1", "MSH2", "MSH6", "PMS2"]
IMMUNE_GENES: list[str] = ["CD8A", "GZMA", "PRF1"]
LABEL_COL = "msi_status"
POSITIVE_LABEL = "MSI-H"
NEGATIVE_LABEL = "MSS"
def usable_msi_cohort(cohort: Cohort) -> Cohort:
"""Restrict to MSI-H vs MSS samples with non-missing stage and age."""
lab = cohort.labels
clin = cohort.clinical
mask = (
lab[LABEL_COL].isin([POSITIVE_LABEL, NEGATIVE_LABEL])
& (clin["stage"] != "NA")
& clin["stage"].notna()
& clin["age"].notna()
)
return cohort.restrict(mask)
@dataclass
class H1Result:
n_used: int
n_msi_h: int
n_mss: int
mmr_score: pd.Series
immune_score: pd.Series
msi_status: pd.Series
mmr_median_msi_h: float
mmr_median_mss: float
immune_median_msi_h: float
immune_median_mss: float
effect: EffectResult
fit: FitResult
@property
def mmr_separates_correct_direction(self) -> bool:
return self.mmr_median_msi_h < self.mmr_median_mss
@property
def immune_separates_correct_direction(self) -> bool:
return self.immune_median_msi_h > self.immune_median_mss
def run_h1(cohort: Cohort | None = None) -> H1Result:
"""Run the full H1 program and return the verification numbers."""
if cohort is None:
cohort = Load("processed")
cohort = usable_msi_cohort(cohort)
M = cohort.expression
mmr_score = Reduce(Select(M, MMR_GENES), agg="mean")
immune_score = Reduce(Select(M, IMMUNE_GENES), agg="mean")
status = cohort.labels[LABEL_COL]
is_msi_h = status == POSITIVE_LABEL
mmr_medians = mmr_score.groupby(status).median()
immune_medians = immune_score.groupby(status).median()
adjust = pd.DataFrame({
"stage": cohort.clinical["stage"].astype("category"),
"age": cohort.clinical["age"].astype(float),
})
effect = Effect(mmr_score, immune_score, adjust)
state = pd.DataFrame({"mmr_score": mmr_score, "immune_score": immune_score})
fit = Fit(state, is_msi_h.astype(int).values)
return H1Result(
n_used=len(cohort.sample_ids),
n_msi_h=int(is_msi_h.sum()),
n_mss=int((status == NEGATIVE_LABEL).sum()),
mmr_score=mmr_score,
immune_score=immune_score,
msi_status=status,
mmr_median_msi_h=float(mmr_medians.get(POSITIVE_LABEL, float("nan"))),
mmr_median_mss=float(mmr_medians.get(NEGATIVE_LABEL, float("nan"))),
immune_median_msi_h=float(immune_medians.get(POSITIVE_LABEL, float("nan"))),
immune_median_mss=float(immune_medians.get(NEGATIVE_LABEL, float("nan"))),
effect=effect,
fit=fit,
)
|