Blame for biochime_regulon_core.py

#!/usr/bin/env python3

from __future__ import annotations

import csv
from dataclasses import dataclass
from typing import Dict, Iterable, List, Sequence, Tuple

import numpy as np
import torch
import torch.nn as nn


# -----------------------------------------------------------------------------
# Regulatory prior
# -----------------------------------------------------------------------------
@dataclass
class TFRegulon:
    tf: str
    targets: np.ndarray          # column indices in current expression matrix
    weights: np.ndarray          # signed CollecTRI edge weights
    n_total_targets: int
    n_present_targets: int
    coverage: float
    programs: Tuple[str, ...]
    tf_weight: float
    reproducibility: float
    perturbed_cl13: bool


class RegulonScorer:
    """Simple, dependency-light signed TF activity scorer.

    Activity_r(cell) = sum_g w_rg * z_g(cell) / sqrt(sum_g w_rg^2)

    where z_g is standardized using retained TRAINING cells only.

    This transparent scoring method is the signed target-gene prior
    used in the TRACE implementation; it is not TF transcript abundance.
    """

    def __init__(
        self,
        prior_csv: str,
        var_names: Sequence[str],
        min_targets: int = 5,
        min_coverage: float = 0.15,
        candidate_source: str = "taiji",
        max_tfs: int = 0,
        edge_weight_mode: str = "weight",
    ):
        self.prior_csv = prior_csv
        self.var_names = np.asarray([str(x) for x in var_names])
        self.lookup = {g.upper(): i for i, g in enumerate(self.var_names)}
        self.min_targets = int(min_targets)
        self.min_coverage = float(min_coverage)
        self.candidate_source = candidate_source
        self.max_tfs = int(max_tfs)
        self.edge_weight_mode = str(edge_weight_mode).lower()
        if self.edge_weight_mode not in {"weight", "sign", "sign_resources"}:
            raise ValueError("edge_weight_mode must be one of: weight, sign, sign_resources")

        rows = []
        with open(prior_csv, newline="") as f:
            reader = csv.DictReader(f)
            required = {"program_id", "tf", "target_gene", "weight", "provenance"}
            if not required.issubset(set(reader.fieldnames or [])):
                raise ValueError(
                    f"{prior_csv} must contain at least {sorted(required)}; "
                    f"found {reader.fieldnames}"
                )
            for r in reader:
                rows.append(r)
        if not rows:
            raise ValueError(f"No rows found in {prior_csv}")

        # Candidate TF metadata comes from Taiji state-association rows.  This
        # selects TFs biologically linked to the T-cell programs without imposing
        # a state order on the trajectory.
        meta: Dict[str, dict] = {}
        for r in rows:
            if (r.get("provenance") or "").strip().lower() != "taiji_pagerank":
                continue
            tf = (r.get("tf") or "").strip()
            if not tf:
                continue
            m = meta.setdefault(tf, {
                "programs": set(), "tf_weight": [], "reproducibility": [], "perturbed": False
            })
            m["programs"].add((r.get("program_id") or "").strip())
            try:
                m["tf_weight"].append(abs(float(r.get("tf_weight") or 1.0)))
            except Exception:
                pass
            try:
                val = float(r.get("reproducibility") or "nan")
                if np.isfinite(val):
                    m["reproducibility"].append(val)
            except Exception:
                pass
            m["perturbed"] = m["perturbed"] or str(r.get("perturbed_cl13", "")).lower() in {"true", "1", "yes"}

        # Collect signed TF-target edges.  The source file repeats some edges in
        # multiple program blocks; aggregate duplicates by mean edge weight so a
        # TF-target relation is counted once.
        edge_vals: Dict[Tuple[str, str], List[float]] = {}
        total_targets: Dict[str, set] = {}
        for r in rows:
            if (r.get("provenance") or "").strip().lower() != "collectri":
                continue
            tf = (r.get("tf") or "").strip()
            tg = (r.get("target_gene") or "").strip()
            if not tf or not tg:
                continue
            if candidate_source == "taiji" and tf not in meta:
                continue
            try:
                if self.edge_weight_mode == "weight":
                    # Legacy BioCHIME behavior.  In the assembled T-cell prior this
                    # column may include state-specific scaling, so the paper
                    # configuration uses state-independent sign mode.
                    w = float(r.get("weight") or 0.0)
                else:
                    sg = float(r.get("sign") or 0.0)
                    if sg == 0.0:
                        raw = float(r.get("weight") or 0.0)
                        sg = float(np.sign(raw))
                    if self.edge_weight_mode == "sign":
                        w = sg
                    else:
                        nr = max(float(r.get("n_resources") or 1.0), 1.0)
                        w = sg * np.sqrt(nr)
            except Exception:
                continue
            if not np.isfinite(w) or abs(w) < 1e-15:
                continue
            edge_vals.setdefault((tf, tg), []).append(w)
            total_targets.setdefault(tf, set()).add(tg)

        # Rank candidates only if requested.  We use reproducibility first, then
        # Taiji weight.  This ranking is based on external prior metadata, not on
        # the held-out expression data.
        all_tfs = sorted(total_targets)
        if max_tfs > 0:
            def rank_key(tf: str):
                m = meta.get(tf, {})
                rep = np.nanmean(m.get("reproducibility", [np.nan])) if m.get("reproducibility") else -np.inf
                tw = np.nanmax(m.get("tf_weight", [0.0])) if m.get("tf_weight") else 0.0
                if not np.isfinite(rep):
                    rep = -1e9
                return (rep, tw)
            all_tfs = sorted(all_tfs, key=rank_key, reverse=True)[:max_tfs]

        regulons: List[TFRegulon] = []
        for tf in all_tfs:
            tg_all = sorted(total_targets[tf])
            present_idx, present_w = [], []
            for tg in tg_all:
                j = self.lookup.get(tg.upper())
                if j is None:
                    continue
                vals = edge_vals.get((tf, tg), [])
                if not vals:
                    continue
                present_idx.append(j)
                present_w.append(float(np.mean(vals)))
            nt = len(tg_all)
            npres = len(present_idx)
            cov = npres / max(nt, 1)
            if npres < min_targets or cov < min_coverage:
                continue
            m = meta.get(tf, {})
            tw = float(np.nanmax(m.get("tf_weight", [1.0]))) if m.get("tf_weight") else 1.0
            reps = m.get("reproducibility", [])
            rep = float(np.nanmax(reps)) if reps else float("nan")
            regulons.append(TFRegulon(
                tf=tf,
                targets=np.asarray(present_idx, dtype=int),
                weights=np.asarray(present_w, dtype=np.float64),
                n_total_targets=nt,
                n_present_targets=npres,
                coverage=cov,
                programs=tuple(sorted(x for x in m.get("programs", set()) if x)),
                tf_weight=tw,
                reproducibility=rep,
                perturbed_cl13=bool(m.get("perturbed", False)),
            ))

        if not regulons:
            raise ValueError(
                "No TF regulons passed coverage filters. Try --min_targets 3 "
                "and/or --min_coverage 0.05, or rebuild the expression matrix "
                "with more genes / forced inclusion of prior target genes."
            )
        self.regulons = regulons
        self.tf_names = [r.tf for r in regulons]
        self.gene_mu = None
        self.gene_sd = None

    def fit_gene_scaler(self, X_train: np.ndarray):
        X_train = np.asarray(X_train, dtype=np.float64)
        self.gene_mu = X_train.mean(0)
        self.gene_sd = X_train.std(0)
        self.gene_sd[self.gene_sd < 1e-8] = 1.0
        return self

    def score(self, X: np.ndarray) -> np.ndarray:
        if self.gene_mu is None or self.gene_sd is None:
            raise RuntimeError("Call fit_gene_scaler(X_train) before score(X)")
        X = np.asarray(X, dtype=np.float64)
        Zg = (X - self.gene_mu) / self.gene_sd
        out = np.empty((len(X), len(self.regulons)), dtype=np.float32)
        for j, r in enumerate(self.regulons):
            w = r.weights
            denom = np.sqrt(np.sum(w * w)) + 1e-12
            out[:, j] = (Zg[:, r.targets] @ w / denom).astype(np.float32)
        return out

    def coverage_rows(self) -> List[dict]:
        rows = []
        for r in self.regulons:
            rows.append({
                "tf": r.tf,
                "n_targets_total": r.n_total_targets,
                "n_targets_present": r.n_present_targets,
                "coverage": r.coverage,
                "programs": ";".join(r.programs),
                "tf_weight": r.tf_weight,
                "reproducibility": r.reproducibility,
                "perturbed_cl13": r.perturbed_cl13,
            })
        return rows


def standardize_activity(A_train: np.ndarray, A: np.ndarray | None = None):
    A_train = np.asarray(A_train, dtype=np.float64)
    mu = A_train.mean(0)
    sd = A_train.std(0)
    sd[sd < 1e-8] = 1.0
    if A is None:
        return ((A_train - mu) / sd).astype(np.float32), mu.astype(np.float32), sd.astype(np.float32)
    return ((np.asarray(A) - mu) / sd).astype(np.float32), mu.astype(np.float32), sd.astype(np.float32)


def fit_ridge_map(A: np.ndarray, Z: np.ndarray, lam: float = 1.0) -> Tuple[np.ndarray, np.ndarray]:
    """Fit Z ~= A @ W + b; derivatives map as dZ/dt ~= dA/dt @ W."""
    A = np.asarray(A, dtype=np.float64)
    Z = np.asarray(Z, dtype=np.float64)
    am, zm = A.mean(0), Z.mean(0)
    Ac, Zc = A - am, Z - zm
    G = Ac.T @ Ac + float(lam) * np.eye(A.shape[1])
    W = np.linalg.solve(G, Ac.T @ Zc)
    b = zm - am @ W
    return W.astype(np.float32), b.astype(np.float32)


# -----------------------------------------------------------------------------
#Hermite helpers
# -----------------------------------------------------------------------------
def pchip_slopes(x: np.ndarray, Y: np.ndarray) -> np.ndarray:
    x = np.asarray(x, dtype=float)
    K = len(x)
    h = np.diff(x)
    d = (Y[1:] - Y[:-1]) / h[:, None, None]
    m = np.zeros_like(Y)
    if K == 2:
        m[0] = d[0]
        m[1] = d[0]
        return m
    for k in range(1, K - 1):
        d0, d1 = d[k - 1], d[k]
        h0, h1 = h[k - 1], h[k]
        w0, w1 = 2 * h1 + h0, h1 + 2 * h0
        mono = (d0 * d1) > 0
        with np.errstate(divide="ignore", invalid="ignore"):
            mk = (w0 + w1) / (w0 / d0 + w1 / d1)
        m[k] = np.where(mono, mk, 0.0)

    def endpoint(h0, h1, d0, d1):
        me = ((2 * h0 + h1) * d0 - h0 * d1) / (h0 + h1)
        me = np.where(np.sign(me) != np.sign(d0), 0.0, me)
        me = np.where((np.sign(d0) != np.sign(d1)) & (np.abs(me) > 3 * np.abs(d0)), 3 * d0, me)
        return me

    m[0] = endpoint(h[0], h[1], d[0], d[1])
    m[-1] = endpoint(h[-1], h[-2], d[-1], d[-2])
    return m


def project_shape_safe_slopes(x: np.ndarray, Y: np.ndarray, M: np.ndarray) -> np.ndarray:
    """Conservative coordinate-wise projection that prevents PCHIP-like overshoot."""
    x = np.asarray(x, dtype=float)
    h = np.diff(x)
    d = (Y[1:] - Y[:-1]) / h[:, None, None]
    out = np.asarray(M).copy()
    K = len(x)
    if K == 2:
        sec = d[0]
        for k in [0, 1]:
            ok = np.sign(out[k]) == np.sign(sec)
            out[k] = np.where(ok, out[k], 0.0)
            out[k] = np.sign(sec) * np.minimum(np.abs(out[k]), 3.0 * np.abs(sec))
        return out

    for k, sec in [(0, d[0]), (K - 1, d[-1])]:
        ok = np.sign(out[k]) == np.sign(sec)
        out[k] = np.where(ok, out[k], 0.0)
        out[k] = np.sign(sec) * np.minimum(np.abs(out[k]), 3.0 * np.abs(sec))

    for k in range(1, K - 1):
        dl, dr = d[k - 1], d[k]
        mono = (dl * dr) > 0
        sgn = np.sign(dl)
        cap = 3.0 * np.minimum(np.abs(dl), np.abs(dr))
        cand = np.where(np.sign(out[k]) == sgn, out[k], 0.0)
        cand = sgn * np.minimum(np.abs(cand), cap)
        out[k] = np.where(mono, cand, 0.0)
    return out


def hermite_eval(x: np.ndarray, Y: np.ndarray, M: np.ndarray, t: np.ndarray):
    x = np.asarray(x, dtype=float)
    t = np.asarray(t, dtype=float)
    K, B, _ = Y.shape
    idx = np.clip(np.searchsorted(x, t, side="right") - 1, 0, K - 2)
    xi, xi1 = x[idx], x[idx + 1]
    hi = xi1 - xi
    s = ((t - xi) / hi)[:, None]
    hb = hi[:, None]
    ar = np.arange(B)
    yi, yi1 = Y[idx, ar], Y[idx + 1, ar]
    mi, mi1 = M[idx, ar], M[idx + 1, ar]
    mu = ((2*s**3 - 3*s**2 + 1)*yi + (s**3 - 2*s**2 + s)*hb*mi
          + (-2*s**3 + 3*s**2)*yi1 + (s**3 - s**2)*hb*mi1)
    mup = ((6*s**2 - 6*s)*yi + (3*s**2 - 4*s + 1)*hb*mi
           + (-6*s**2 + 6*s)*yi1 + (3*s**2 - 2*s)*hb*mi1) / hb
    return mu, mup


def align_joint(
    clouds_z: Sequence[np.ndarray],
    clouds_a: Sequence[np.ndarray],
    rng: np.random.Generator,
    batch: int = 128,
):
    """Use the existing Euclidean OT plan, carrying TF activities with cells."""
    B = min(batch, min(len(c) for c in clouds_z))
    i0 = rng.choice(len(clouds_z[0]), B, replace=False)
    z = [clouds_z[0][i0]]
    a = [clouds_a[0][i0]]
    for k in range(1, len(clouds_z)):
        cand = rng.choice(len(clouds_z[k]), B, replace=False)
        zc, ac = clouds_z[k][cand], clouds_a[k][cand]
        import benchmark_table as BT
        P = BT.ot_plan(z[-1], zc)
        Pn = P / (P.sum(1, keepdims=True) + 1e-12)
        jj = np.array([rng.choice(B, p=Pn[i]) for i in range(B)])
        z.append(zc[jj])
        a.append(ac[jj])
    return np.stack(z, 0), np.stack(a, 0)


def _match_bio_vector_norm(m_geom: np.ndarray, m_bio: np.ndarray, clip=(0.25, 4.0)) -> np.ndarray:
    """Make biology contribute direction rather than arbitrary numerical scale."""
    ng = np.linalg.norm(m_geom, axis=-1, keepdims=True)
    nb = np.linalg.norm(m_bio, axis=-1, keepdims=True)
    scale = ng / (nb + 1e-12)
    scale = np.clip(scale, clip[0], clip[1])
    scale = np.where((ng > 1e-8) & (nb > 1e-8), scale, 0.0)
    return m_bio * scale


@dataclass
class BioFitInfo:
    rho: float
    ridge_W: np.ndarray
    ridge_b: np.ndarray


def fit_regulon_bio_flow(
    clouds_z_by_day: Dict[float, np.ndarray],
    clouds_a_by_day: Dict[float, np.ndarray],
    dn: Sequence[float],
    dev: str,
    steps: int,
    seed: int,
    rho: float = 1.0,
    batch: int = 128,
    lr: float = 1e-3,
    ridge: float = 1.0,
    ridge_map: tuple[np.ndarray, np.ndarray] | None = None,
    shape_project: bool = True,
    log_every: int = 0,
):
    """Train TRACE with TF-regulon-guided Hermite knot tangents."""
    if not (0.0 <= rho <= 1.0):
        raise ValueError("rho must be in [0,1]")
    torch.manual_seed(seed)
    rng = np.random.default_rng(seed)
    x = np.asarray(sorted(dn), dtype=float)
    zclouds = [clouds_z_by_day[t] for t in x]
    aclouds = [clouds_a_by_day[t] for t in x]
    d = zclouds[0].shape[1]

    Apool = np.vstack(aclouds)
    Zpool = np.vstack(zclouds)
    if ridge_map is None:
        W, b = fit_ridge_map(Apool, Zpool, lam=ridge)
    else:
        # Negative controls such as TF-channel permutation must use the
        # ridge map fitted on the TRUE TF identities.  Re-fitting after a
        # global column permutation would simply permute the rows of W and
        # make the control algebraically equivalent to the unpermuted model.
        W, b = ridge_map
        W = np.asarray(W, dtype=np.float32)
        b = np.asarray(b, dtype=np.float32)
        if W.shape != (Apool.shape[1], Zpool.shape[1]):
            raise ValueError(
                f"ridge_map W has shape {W.shape}, expected "
                f"{(Apool.shape[1], Zpool.shape[1])}"
            )
        if b.shape != (Zpool.shape[1],):
            raise ValueError(
                f"ridge_map b has shape {b.shape}, expected {(Zpool.shape[1],)}"
            )

    import benchmark_table as BT
    field = BT.Field(d).to(dev)
    opt = torch.optim.Adam(field.parameters(), lr=lr)
    ni = len(x) - 1

    for step in range(steps):
        Z, A = align_joint(zclouds, aclouds, rng, batch=batch)
        B = Z.shape[1]
        per = B // ni
        ts = []
        for i in range(ni):
            n = per + (1 if i < B - per * ni else 0)
            ts.append(rng.uniform(x[i], x[i + 1], size=n))
        tq = np.concatenate(ts)[:B]
        rng.shuffle(tq)

        m_geom = pchip_slopes(x, Z)
        if rho > 0:
            m_tf = pchip_slopes(x, A)
            m_bio = np.einsum("kbr,rd->kbd", m_tf, W)
            m_bio = _match_bio_vector_norm(m_geom, m_bio)
            M = (1.0 - rho) * m_geom + rho * m_bio
            if shape_project:
                M = project_shape_safe_slopes(x, Z, M)
        else:
            M = m_geom

        pos, vel = hermite_eval(x, Z, M, tq)
        tt = torch.tensor(tq[:, None], dtype=torch.float32, device=dev)
        pred = field(torch.tensor(pos, dtype=torch.float32, device=dev), tt)
        target = torch.tensor(vel, dtype=torch.float32, device=dev)
        loss = ((pred - target) ** 2).mean()
        opt.zero_grad()
        loss.backward()
        nn.utils.clip_grad_norm_(field.parameters(), 1.0)
        opt.step()
        if log_every and (step + 1) % log_every == 0:
            print(f"[BioCHIME rho={rho:g}] {step+1}/{steps} loss={float(loss):.6f}", flush=True)

    field.eval()
    return field, BioFitInfo(rho=rho, ridge_W=W, ridge_b=b)


def permute_within_day(clouds_a_by_day: Dict[float, np.ndarray], seed: int):
    """Negative control preserving per-day TF distributions but breaking cell correspondence."""
    rng = np.random.default_rng(seed)
    return {t: A[rng.permutation(len(A))].copy() for t, A in clouds_a_by_day.items()}


def permute_tf_channels(clouds_a_by_day: Dict[float, np.ndarray], seed: int):
    """Wrong-biology control: keep values but scramble TF identities globally."""
    rng = np.random.default_rng(seed)
    r = next(iter(clouds_a_by_day.values())).shape[1]
    p = rng.permutation(r)
    return {t: A[:, p].copy() for t, A in clouds_a_by_day.items()}