TCR annotation API + PRISM (#157/#158)
A uniform, pluggable way to add optional, always-defined per-clone
columns to any TCR table, plus the PRISM selection method.
Pgen / Ppost — robust, never-zero (#157)
The OLGA/SONIA runtime path returned Pgen = 0 on ~3.5–5% of α junctions
(CDR3s not ending in the conserved F/W — an anchor parse failure), and
Ppost inherited it, mis-ranking exactly the rare clones of interest. The
data-driven backends in seqprob are finite for every
valid junction.
Two roles per chain:
- Pgen — model fit on an OLGA-generated (synthetic) repertoire =
generation probability.
- Ppost (default) — model fit on an observed repertoire =
post-selection publicness.
log_q = log Ppost − log Pgen is the
data-driven selection factor (Q = Ppost/Pgen).
from tcrsift.annotate_tcrs import add_pgen_ppost
clones = add_pgen_ppost(clones) # pgen_/ppost_/log_q_ {alpha,beta}
Shipped k-mer defaults: Pgen (α+β, OLGA-generated) and Ppost (β, observed).
No observed α reference is bundled → α-Ppost falls back to α-Pgen (finite,
not circular) with a logged note until one is supplied. TCRpeg is the
higher-accuracy backend (backend="tcrpeg", needs pip install
tcrsift[tcrpeg]).
GEX signature scores, z-scored per sample/donor
antigen_response_score (TNFRSF9, MKI67) and naive_score (TCF7, LEF1,
CCR7, SELL, IL7R, CD27, CD28) from per-cell expression, z-scored within
each sample or donor group (#144/#145) before averaging per clone:
from tcrsift.annotate_tcrs import add_gex_signature_scores
clones = add_gex_signature_scores(clones, per_cell, group_col="sample")
per_cell has gex.<SYMBOL> columns + a clone column + the group column.
PRISM — Percentile-Rank In-Silico Multi-criterion
Rank every clone by the mean of percentile ranks of: low ppost_alpha,
low ppost_beta, high antigen_response_score, low naive_score; take the
top-K. Pilot result: ~33% / 45% condition-private clones vs 6% / 17% for
frequency selection.
from tcrsift.annotate_tcrs import annotate_tcrs, select_prism
annotated = annotate_tcrs(clones, per_cell=per_cell, group_col="sample",
backend="kmer")
picks = select_prism(annotated, k=20, group_col="enrichment_method")
Per-dimension weights and K are tunable; PRISM composes with the
in-silico filter layer (insilico_filter).
annotate_tcrs
Uniform per-clone TCR annotation API + PRISM selection (#158).
Adds optional, always-defined columns to any table of TCRs — generated
sequences, clonotypes.csv, filtered/selected clones — and a named
selection method, PRISM, built on them. Every annotation is
independently toggleable; the sequence-probability columns never return
0/NaN for a valid junction (#157), unlike the OLGA/SONIA runtime path.
Annotations
- Pgen / Ppost per chain (
pgen_alpha/pgen_beta,
ppost_alpha/ppost_beta) via :mod:tcrsift.seqprob. Ppost
(observed-repertoire background) is the default publicness measure;
log_q_<chain> = log Ppost − log Pgen is the data-driven selection
factor.
- GEX signature scores (
antigen_response_score, naive_score)
from per-cell expression, z-scored within each sample or donor group
(#144/#145), then averaged per clone.
PRISM (Percentile-Rank In-Silico Multi-criterion)
Rank every clone by the mean of percentile ranks of:
low ppost_alpha, low ppost_beta, high antigen_response_score,
low naive_score — then take the top-K. On the B1-2/B1-3 pilot PRISM
selected ~33% / 45% condition-private clones vs 6% / 17% for frequency.
annotate_tcrs
annotate_tcrs(df: DataFrame, *, methods: list[str] | None = None, per_cell: DataFrame | None = None, backend: str = 'kmer', chains: tuple[str, ...] = ('alpha', 'beta'), clone_col: str = 'CDR3ab', group_col: str | None = None, gex_prefix: str = 'gex') -> pd.DataFrame
Add the requested annotation columns to a TCR table (#158).
methods subset of {"pgen", "ppost", "promiscuity", "gex_signatures",
"prism"} (default all that are computable). per_cell (with
{gex_prefix}. columns + clone_col + group_col) is required for
gex_signatures/prism. Returns a copy with the columns added.
Source code in tcrsift/annotate_tcrs.py
| def annotate_tcrs(
df: pd.DataFrame,
*,
methods: list[str] | None = None,
per_cell: pd.DataFrame | None = None,
backend: str = "kmer",
chains: tuple[str, ...] = ("alpha", "beta"),
clone_col: str = "CDR3ab",
group_col: str | None = None,
gex_prefix: str = "gex",
) -> pd.DataFrame:
"""Add the requested annotation columns to a TCR table (#158).
``methods`` subset of ``{"pgen", "ppost", "promiscuity", "gex_signatures",
"prism"}`` (default all that are computable). ``per_cell`` (with
``{gex_prefix}.`` columns + ``clone_col`` + ``group_col``) is required for
``gex_signatures``/``prism``. Returns a copy with the columns added.
"""
methods = methods or ["pgen", "ppost", "promiscuity", "gex_signatures", "prism"]
out = df.copy()
if "pgen" in methods or "ppost" in methods:
out = add_pgen_ppost(
out, chains=chains, backend=backend,
with_pgen=("pgen" in methods or "prism" in methods),
with_ppost=("ppost" in methods or "prism" in methods),
)
# α–β pairing promiscuity (#148): a no-op when the CDR3 columns are absent,
# so it's safe to include by default alongside the sequence-axis features.
if "promiscuity" in methods:
out = add_pairing_promiscuity(out)
if ("gex_signatures" in methods or "prism" in methods) and per_cell is not None:
out = add_gex_signature_scores(
out, per_cell, clone_col=clone_col, group_col=group_col,
gex_prefix=gex_prefix,
)
if "prism" in methods:
have = [p.score for p in PRISM_DEFAULT_PREDICATES if p.score in out.columns]
if len(have) == len(PRISM_DEFAULT_PREDICATES):
out = prism_score(out, group_col=group_col)
else:
logger.warning(
"annotate_tcrs: PRISM needs %s; have only %s — skipping PRISM",
[p.score for p in PRISM_DEFAULT_PREDICATES], have,
)
return out
|
add_pgen_ppost
add_pgen_ppost(df: DataFrame, *, chains: tuple[str, ...] = ('alpha', 'beta'), backend: str = 'kmer', cdr3_cols: dict[str, str] | None = None, with_pgen: bool = True, with_ppost: bool = True, with_q: bool = True, auto_train: bool = True) -> pd.DataFrame
Add pgen_<chain> / ppost_<chain> / log_q_<chain> columns.
Ppost (observed-repertoire background) is the publicness default; Pgen is
the OLGA-generated background; log_q is their difference (the
selection factor). cdr3_cols maps chain → CDR3 column (default
CDR3_<chain>). Chains whose CDR3 column is absent are skipped.
Returns a copy.
Source code in tcrsift/annotate_tcrs.py
| def add_pgen_ppost(
df: pd.DataFrame,
*,
chains: tuple[str, ...] = ("alpha", "beta"),
backend: str = "kmer",
cdr3_cols: dict[str, str] | None = None,
with_pgen: bool = True,
with_ppost: bool = True,
with_q: bool = True,
auto_train: bool = True,
) -> pd.DataFrame:
"""Add ``pgen_<chain>`` / ``ppost_<chain>`` / ``log_q_<chain>`` columns.
Ppost (observed-repertoire background) is the publicness default; Pgen is
the OLGA-generated background; ``log_q`` is their difference (the
selection factor). ``cdr3_cols`` maps chain → CDR3 column (default
``CDR3_<chain>``). Chains whose CDR3 column is absent are skipped.
Returns a copy.
"""
from .pgen_models import ensure_model
out = df.copy()
cdr3_cols = cdr3_cols or {}
def _resolve(chain: str, role: str):
"""(model, estimator) honestly, or (None, None) if unavailable.
Pgen may fall back across *estimators* (tcrpeg→k-mer) — still genuine
Pgen. Ppost is role-pure: never substituted by Pgen, so an absent
Ppost stays unavailable (→ NaN) rather than masquerading as Pgen.
"""
try:
return ensure_model(chain, backend=backend, role=role,
auto_train=auto_train), backend
except (ImportError, ValueError) as exc: # Pgen can't be (re)trained
if role == "pgen":
logger.warning("%s; using shipped k-mer Pgen for %s", exc, chain)
from .seqprob import load_background_model
return load_background_model(chain, "kmer", "pgen"), "kmer"
return None, None
except FileNotFoundError as exc:
if role == "pgen" and backend != "kmer":
from .seqprob import load_background_model
try:
return load_background_model(chain, "kmer", "pgen"), "kmer"
except FileNotFoundError:
pass
logger.warning("add_pgen_ppost: %s %s unavailable for %s (%s) → NaN",
backend, role, chain, exc)
return None, None
for chain in chains:
cdr3_col = cdr3_cols.get(chain, f"CDR3_{chain}")
if cdr3_col not in out.columns:
logger.info("add_pgen_ppost: no %r column; skipping %s",
cdr3_col, chain)
continue
pgen_est = ppost_est = None
if with_pgen:
model, pgen_est = _resolve(chain, "pgen")
out[f"pgen_{chain}"] = (
seqprob.score_log_prob(out, chain=chain, cdr3_col=cdr3_col,
model=model, out_col=f"pgen_{chain}")
if model is not None else np.nan
)
if with_ppost:
model, ppost_est = _resolve(chain, "ppost")
out[f"ppost_{chain}"] = (
seqprob.score_log_prob(out, chain=chain, cdr3_col=cdr3_col,
model=model, out_col=f"ppost_{chain}")
if model is not None else np.nan
)
# Q = log Ppost − log Pgen is only meaningful within one estimator;
# don't emit a cross-estimator ratio. (Rank on Ppost, not Q — Q alone
# is the weaker signal; publicness lives in the observed frequency.)
if with_q and with_pgen and with_ppost:
if pgen_est is not None and pgen_est == ppost_est:
out[f"log_q_{chain}"] = out[f"ppost_{chain}"] - out[f"pgen_{chain}"]
else:
out[f"log_q_{chain}"] = np.nan
return out
|
add_gex_signature_scores
add_gex_signature_scores(df: DataFrame, per_cell: DataFrame, *, signatures: dict[str, object] | None = None, clone_col: str = 'CDR3ab', group_col: str | None = None, gex_prefix: str = 'gex') -> pd.DataFrame
Join per-clone, group-z-scored signature scores onto df.
signatures maps output column → signature (name or object); defaults
to {"antigen_response_score": "AcuteActivation", "naive_score":
"DifferentiatedNaive"} analogues. group_col is the sample/donor
column in per_cell to z-score within. Returns a copy of df with
the score columns joined on clone_col.
Source code in tcrsift/annotate_tcrs.py
| def add_gex_signature_scores(
df: pd.DataFrame,
per_cell: pd.DataFrame,
*,
signatures: dict[str, object] | None = None,
clone_col: str = "CDR3ab",
group_col: str | None = None,
gex_prefix: str = "gex",
) -> pd.DataFrame:
"""Join per-clone, group-z-scored signature scores onto ``df``.
``signatures`` maps output column → signature (name or object); defaults
to ``{"antigen_response_score": "AcuteActivation", "naive_score":
"DifferentiatedNaive"}`` analogues. ``group_col`` is the sample/donor
column in ``per_cell`` to z-score within. Returns a copy of ``df`` with
the score columns joined on ``clone_col``.
"""
if signatures is None:
signatures = {
"antigen_response_score": "AcuteActivation",
"naive_score": naive_signature(),
}
out = df.copy()
for out_col, sig in signatures.items():
per_clone = score_gex_signature_per_clone(
per_cell, sig, clone_col=clone_col, group_col=group_col,
gex_prefix=gex_prefix, out_col=out_col,
)
out[out_col] = out[clone_col].map(per_clone)
return out
|
score_gex_signature_per_clone
score_gex_signature_per_clone(per_cell: DataFrame, signature, *, clone_col: str = 'CDR3ab', group_col: str | None = None, gex_prefix: str = 'gex', log1p: bool = True, out_col: str | None = None) -> pd.Series
Per-clone GEX signature score, z-scored within sample/donor groups.
per_cell is a per-cell frame with {gex_prefix}.<SYMBOL> columns,
a clone_col and (optionally) a group_col (sample or donor). Each
signature gene is z-scored across cells within each group (so a
high-baseline sample doesn't dominate, #144/#145), averaged across the
signature's genes per cell, then averaged across each clone's cells.
signature is a :class:tcrsift.signature_methods.Signature or a name
in its registry. Returns a per-clone Series indexed by clone id.
Source code in tcrsift/annotate_tcrs.py
| def score_gex_signature_per_clone(
per_cell: pd.DataFrame,
signature,
*,
clone_col: str = "CDR3ab",
group_col: str | None = None,
gex_prefix: str = "gex",
log1p: bool = True,
out_col: str | None = None,
) -> pd.Series:
"""Per-clone GEX signature score, z-scored within sample/donor groups.
``per_cell`` is a per-cell frame with ``{gex_prefix}.<SYMBOL>`` columns,
a ``clone_col`` and (optionally) a ``group_col`` (sample or donor). Each
signature gene is z-scored across cells **within each group** (so a
high-baseline sample doesn't dominate, #144/#145), averaged across the
signature's genes per cell, then averaged across each clone's cells.
``signature`` is a :class:`tcrsift.signature_methods.Signature` or a name
in its registry. Returns a per-clone Series indexed by clone id.
"""
from .signature_methods import SIGNATURES, score_signature
sig = SIGNATURES[signature] if isinstance(signature, str) else signature
cols = {g: f"{gex_prefix}.{g}" for g in sig.all_genes}
present = {g: c for g, c in cols.items() if c in per_cell.columns}
if not present:
raise ValueError(
f"score_gex_signature_per_clone: none of {sig.name}'s genes "
f"({list(sig.all_genes)}) found as {gex_prefix}.<gene> columns"
)
expr = per_cell[list(present.values())].copy()
expr.columns = list(present.keys()) # rename to bare symbols
groups = per_cell[group_col] if group_col and group_col in per_cell.columns else None
per_cell_score = score_signature(
expr, sig, combine="zscore", log1p=log1p, groups=groups,
on_missing="ignore",
)
tmp = pd.DataFrame({"clone": per_cell[clone_col].values,
"score": per_cell_score.values})
per_clone = tmp.groupby("clone", observed=True)["score"].mean()
per_clone.name = out_col or f"{sig.name}_score"
return per_clone
|
naive_signature
The naïve/stem signature used for naive_score (#141/#145).
The down-pole of the Differentiated contrast: TCF7, LEF1, CCR7, SELL,
IL7R, CD27, CD28.
Source code in tcrsift/annotate_tcrs.py
| def naive_signature():
"""The naïve/stem signature used for ``naive_score`` (#141/#145).
The down-pole of the Differentiated contrast: TCF7, LEF1, CCR7, SELL,
IL7R, CD27, CD28.
"""
from .signature_methods import Signature
return Signature(
"Naive",
("TCF7", "LEF1", "CCR7", "SELL", "IL7R", "CD27", "CD28"),
panel="broad",
description="Naïve/stem-memory program (#141/#145) for naive_score.",
)
|
prism_score
prism_score(df: DataFrame, *, predicates: list[FilterPredicate] | None = None, weights: list[float] | None = None, group_col: str | None = None, score_col: str = 'prism_score', rank_col: str = 'prism_rank') -> pd.DataFrame
Compute the PRISM score: (weighted) mean of per-dimension percentile ranks.
Each predicate contributes a lower-is-better percentile rank in [0, 1]
(0 = best); PRISM averages them (optionally weighted) so a low score =
a strong multi-criterion candidate. rank_col is the 1-based ordering
(1 = best). When group_col is given, percentile ranks are computed
within each group (e.g. per assay condition). Missing any dimension →
NaN PRISM score (ranked last) — a clone we can't fully score isn't picked.
Returns a copy with the two columns added.
Delegates the composite to :func:insilico_filter.average_percentile_rank
(the single row-wise PRISM engine) so this and the selection-path PRISM
can't diverge.
Note: the ppost_* dimensions are strongly anti-correlated with CDR3
length (longer CDR3 -> lower Pgen/Ppost, ~0.9 |corr| for an order-2 k-mer),
so those PRISM axes partly encode length, not just rarity. This is inherent
to any generation-probability score; interpret accordingly.
Raises if a predicate column is missing OR entirely NaN — PRISM on an
unpopulated score (e.g. ppost never computed) would silently rank nothing.
Source code in tcrsift/annotate_tcrs.py
| def prism_score(
df: pd.DataFrame,
*,
predicates: list[FilterPredicate] | None = None,
weights: list[float] | None = None,
group_col: str | None = None,
score_col: str = "prism_score",
rank_col: str = "prism_rank",
) -> pd.DataFrame:
"""Compute the PRISM score: (weighted) mean of per-dimension percentile ranks.
Each predicate contributes a lower-is-better percentile rank in [0, 1]
(0 = best); PRISM averages them (optionally weighted) so a low score =
a strong multi-criterion candidate. ``rank_col`` is the 1-based ordering
(1 = best). When ``group_col`` is given, percentile ranks are computed
within each group (e.g. per assay condition). Missing any dimension →
NaN PRISM score (ranked last) — a clone we can't fully score isn't picked.
Returns a copy with the two columns added.
Delegates the composite to :func:`insilico_filter.average_percentile_rank`
(the single row-wise PRISM engine) so this and the selection-path PRISM
can't diverge.
Note: the ``ppost_*`` dimensions are strongly anti-correlated with CDR3
length (longer CDR3 -> lower Pgen/Ppost, ~0.9 |corr| for an order-2 k-mer),
so those PRISM axes partly encode length, not just rarity. This is inherent
to any generation-probability score; interpret accordingly.
Raises if a predicate column is missing OR entirely NaN — PRISM on an
unpopulated score (e.g. ppost never computed) would silently rank nothing.
"""
predicates = predicates or PRISM_DEFAULT_PREDICATES
missing = [p.score for p in predicates if p.score not in df.columns]
if missing:
raise ValueError(f"prism_score: missing predicate columns {missing}")
all_nan = [p.score for p in predicates if not df[p.score].notna().any()]
if all_nan:
raise ValueError(
f"prism_score: predicate column(s) {all_nan} are entirely NaN — "
"every clone is unscorable so PRISM would rank nothing. Populate "
"these scores (e.g. add_pgen_ppost / add_gex_signature_scores) first."
)
out = df.copy()
scores = average_percentile_rank(
out, predicates, group_col=group_col, weights=weights, require_complete=True,
)
out[score_col] = scores
out[rank_col] = scores.rank(method="min", ascending=True, na_option="bottom")
return out
|
select_prism
select_prism(df: DataFrame, *, k: int, group_col: str | None = None, score_col: str = 'prism_score', selected_col: str = 'prism_selected', **prism_kwargs) -> pd.DataFrame
Compute PRISM and flag the top-k clones (per group if given).
Returns a copy with the PRISM columns plus a boolean selected_col.
Source code in tcrsift/annotate_tcrs.py
| def select_prism(
df: pd.DataFrame,
*,
k: int,
group_col: str | None = None,
score_col: str = "prism_score",
selected_col: str = "prism_selected",
**prism_kwargs,
) -> pd.DataFrame:
"""Compute PRISM and flag the top-``k`` clones (per group if given).
Returns a copy with the PRISM columns plus a boolean ``selected_col``.
"""
out = prism_score(df, group_col=group_col, score_col=score_col,
**prism_kwargs)
if group_col is not None and group_col in out.columns:
rank = out.groupby(group_col, observed=True)[score_col].rank(
method="first", ascending=True, na_option="bottom",
)
else:
rank = out[score_col].rank(method="first", ascending=True,
na_option="bottom")
out[selected_col] = out[score_col].notna() & (rank <= k)
return out
|