"""Scan measurements for gene-associated effects.
The scan fits the same guide-level model to each measurement and reports both
within-measurement and across-scan multiplicity corrections. A Simes global-
null p value summarizes each measurement before the across-scan correction;
the default correction is Benjamini-Hochberg.
Effect sizes are coefficients divided by the model's residual standard
deviation. :attr:`ScanResult.effective_n_tests` reports the Li and Ji estimate
for correlated measurements, which is used when
``across_scan_method='bonferroni_effective'``.
"""
from __future__ import annotations
import math
from dataclasses import dataclass, field
from typing import Dict, Iterable, Mapping, Optional, Sequence, Tuple
import numpy as np
from . import schema
from .multiple_testing import adjust_p_values, canonical_method
__all__ = [
"ScanRefused",
"MeasurementEffect",
"ScanResult",
"scan_measurements",
"effective_number_of_tests",
"simes_p_value",
"DEFAULT_BLOCK_COLUMNS",
]
#: What the scan blocks on unless told otherwise.
#:
#: Blocking on screen prevents differences between combined experiments from
#: inflating guide-effect standard errors. A missing or single-valued block is
#: dropped, so a single-screen project uses the same model with one fewer term.
DEFAULT_BLOCK_COLUMNS: Tuple[str, ...] = (schema.SCREEN_KEY,)
#: Minimum wells required to estimate a gene effect. Single-well genes cannot
#: distinguish a phenotype from a well-specific artifact and are excluded
#: before fitting.
MIN_WELLS_PER_GENE = 2
#: Columns that are the DESIGN, not the response.
#:
#: Regressing the guides on the row index is not an analysis, and neither is
#: regressing them on the well's own guide count. Non-numeric columns are
#: excluded automatically; this closed list is for the numeric ones that would
#: otherwise sail through.
_DESIGN_COLUMNS: Tuple[str, ...] = (
'count', 'counts', 'total_count', 'well_count', 'object_count',
'cell_count', 'n', 'n_wells', 'n_cells', 'size',
'fraction', 'gene_fraction', 'grna_fraction',
'index', 'level_0',
)
_IDENTITY_COLUMNS: Tuple[str, ...] = (
schema.SCREEN_KEY, schema.PLATE_KEY, schema.ROW_KEY, schema.COLUMN_KEY,
schema.FIELD_KEY, schema.TIME_KEY, schema.CHANNEL_KEY, schema.SLICE_KEY,
schema.OBJECT_LABEL_KEY, schema.OBJECT_TYPE_KEY,
schema.PRC_KEY, schema.PRCF_KEY, schema.PRCFO_KEY,
'source_database', 'prcfo', 'objectID', 'well', 'condition',
)
[docs]
class ScanRefused(ValueError):
"""Raised when the input cannot support a measurement scan."""
@dataclass(frozen=True)
[docs]
class MeasurementEffect:
"""One scanned measurement: its best gene, and both corrections.
:param measurement: Name of the numeric response column that was scanned.
:param n_wells: Number of wells with finite values used for this
measurement's fit.
:param n_genes: Number of non-baseline gene terms in this measurement's
fitted design.
:param top_gene: Gene term with the largest absolute standardized effect.
:param effect_size: Signed coefficient of ``top_gene`` divided by this
measurement's residual standard deviation.
:param coefficient: Signed ``top_gene`` coefficient in the measurement's
original units.
:param p_value: Raw two-sided P value for ``top_gene``.
:param within_run_q: Smallest adjusted P value across all gene terms for
this measurement; it need not belong to ``top_gene``.
:param within_run_hits: Number of gene terms rejected by the
within-measurement correction at the requested alpha.
:param measurement_p: Simes global-null P value combining this
measurement's raw gene-term P values for the across-scan family.
:param across_scan_q: This measurement's value after correction across all
successfully scanned measurements.
:param survives_within_run: Whether at least one gene term survived the
within-measurement correction.
:param survives_across_scan: Whether this measurement survived the
across-scan correction.
"""
measurement: str
#: Wells the model was fitted on. Not constant across a merged frame --
#: databases from different spaCR versions have different columns filled.
n_wells: int
#: Gene terms in the design (the baseline / control level is not one).
n_genes: int
#: The gene with the largest absolute standardised effect.
top_gene: str
#: That gene's coefficient in units of the residual standard deviation.
#: **This is the primary sort**, and it is signed-magnitude: the table is
#: ranked on ``abs``, and ``coefficient`` keeps the direction.
effect_size: float
#: The same effect in the measurement's own units.
coefficient: float
#: The top gene's raw two-sided P value. Reported, never ranked on.
p_value: float
#: The measurement's best within-run adjusted P, across its own genes.
within_run_q: float
#: How many genes the within-run correction called at ``alpha``.
within_run_hits: int
#: Simes' global-null P for this measurement -- what the across-scan
#: stage was actually given.
measurement_p: float
#: The across-scan adjusted P. Never smaller than ``within_run_q`` under
#: the default methods, which is the whole point of computing it.
across_scan_q: float
survives_within_run: bool
survives_across_scan: bool
@dataclass(frozen=True)
[docs]
class ScanResult:
"""Fitted measurement effects of a scan and the corrections applied.
:param rows: Successfully fitted measurement effects; measurements that
could not be fitted are recorded in ``skipped`` instead.
:param skipped: Mapping from each unscanned measurement name to its refusal
or fitting reason.
:param block_columns: Blocking columns actually retained in the fitted
design after absent and single-valued candidates were removed.
:param gene_column: Input-frame column containing the gene assignment.
:param control_genes: Requested control-gene labels recorded as the
effect-baseline request; absent labels are retained here even though
the design falls back to an observed baseline.
:param within_run_method: Canonical multiplicity method applied across gene
terms separately within each measurement.
:param across_scan_method: Canonical multiplicity method applied across
measurements' Simes global-null P values, including
``"bonferroni_effective"`` for the Li–Ji divisor.
:param alpha: Significance level used for both within-measurement and
across-scan decisions.
:param effective_n_tests: Li–Ji estimate computed from successfully scanned
measurement columns, or ``nan`` when no estimate is available.
:param genes_dropped: Mapping from each non-control gene removed for
insufficient well replication to its observed well count.
"""
rows: Tuple[MeasurementEffect, ...]
#: ``{measurement: why it could not be scanned}``. Named rather than
#: dropped: a measurement missing from the table without explanation reads
#: as a measurement with no effect.
skipped: Mapping[str, str]
block_columns: Tuple[str, ...]
gene_column: str
control_genes: Tuple[str, ...]
within_run_method: str
across_scan_method: str
alpha: float
#: Li & Ji (2005) estimate of how many independent tests the scanned
#: columns amount to. ``nan`` when it could not be estimated.
effective_n_tests: float
#: ``{gene: wells it had}`` for genes dropped as having too few wells to
#: corroborate anything. Named rather than silently filtered: a gene
#: missing from the result with no explanation reads as a gene with no
#: effect, which is the opposite of what happened to it.
genes_dropped: Mapping[str, int] = field(default_factory=dict)
@property
[docs]
def n_measurements_scanned(self) -> int:
"""Return the number of measurements fitted successfully."""
return len(self.rows)
[docs]
def surviving(self) -> Tuple[MeasurementEffect, ...]:
"""Return measurements significant after across-scan correction."""
return tuple(row for row in self.rows if row.survives_across_scan)
[docs]
def frame(self):
"""The result table, ranked by effect size, largest first."""
import pandas as pd
table = pd.DataFrame([vars(row) for row in self.rows])
if table.empty:
return pd.DataFrame(columns=[
'measurement', 'n_wells', 'n_genes', 'top_gene', 'effect_size',
'coefficient', 'p_value', 'within_run_q', 'within_run_hits',
'measurement_p', 'across_scan_q', 'survives_within_run',
'survives_across_scan'])
order = table['effect_size'].abs().sort_values(
ascending=False, kind='stable').index
return table.loc[order].reset_index(drop=True)
[docs]
def describe(self) -> str:
"""Return a text summary of the scan design and its corrections."""
genes = self.rows[0].n_genes if self.rows else 0
lines = [f"{self.n_measurements_scanned} measurement(s) scanned "
f"against {genes} gene term(s)"]
if self.block_columns:
lines.append(" blocked on " + ", ".join(self.block_columns))
lines.append(
f" within-run correction across genes: {self.within_run_method}")
lines.append(
f" across-scan correction across measurements: "
f"{self.across_scan_method}")
if math.isfinite(self.effective_n_tests):
lines.append(
f" effective number of independent tests: "
f"{self.effective_n_tests:.1f} of "
f"{self.n_measurements_scanned} columns (measurements are "
f"correlated, so the column count overstates the family)")
survivors = self.surviving()
lines.append(
f" {len(survivors)} survive the across-scan correction at "
f"alpha={self.alpha:g}; "
f"{sum(1 for row in self.rows if row.survives_within_run)} would "
f"have been reported by a per-measurement run")
if self.skipped:
lines.append(
f" {len(self.skipped)} not scanned: "
+ ", ".join(f"{name} ({why})"
for name, why in sorted(self.skipped.items())))
return "\n".join(lines)
[docs]
def simes_p_value(p_values) -> float:
"""Simes' global-null P value for one family of tests.
``min_i (m * p_(i) / i)`` over the sorted P values -- the probability of
seeing a family this extreme when none of its members is real. Identical
to the minimum Benjamini-Hochberg adjusted P, and valid under positive
dependence among the tests, which gene terms in one design have.
This is the value one measurement contributes to the across-scan stage.
The raw minimum of, for example, 23 gene tests is not a valid P value: it
ignores the within-measurement multiplicity, which the second correction
would then not account for.
:param p_values: raw P values for one family of tests; non-finite values
are ignored.
"""
values = np.asarray(p_values, dtype=float)
values = values[np.isfinite(values)]
if values.size == 0:
return float('nan')
ordered = np.sort(values)
ranks = np.arange(1, ordered.size + 1, dtype=float)
return float(np.min(ordered * ordered.size / ranks))
[docs]
def effective_number_of_tests(matrix) -> float:
"""Li & Ji (2005) effective number of independent tests.
``M_eff = sum_i f(|lambda_i|)`` over the eigenvalues of the correlation
matrix, with ``f(x) = I(x >= 1) + (x - floor(x))``: an eigenvalue of 3
counts as one independent test, an eigenvalue of 0 counts as none.
Correlated measurements are fewer independent tests than columns: twelve
columns that measure four quantities three ways each are not twelve
tests, and a Bonferroni correction over all twelve is conservative. The
estimate is reported beside the column count and is the divisor
``'bonferroni_effective'`` uses.
:param matrix: ``(n_wells, n_measurements)`` array. Columns with no
variance are dropped; NaNs are handled pairwise-complete.
:returns: the estimate, or ``nan`` when it cannot be computed.
"""
data = np.asarray(matrix, dtype=float)
if data.ndim != 2 or data.shape[1] == 0:
return float('nan')
if data.shape[1] == 1:
return 1.0
import pandas as pd
correlation = pd.DataFrame(data).corr().to_numpy()
keep = np.isfinite(np.diag(correlation))
correlation = correlation[np.ix_(keep, keep)]
if correlation.size == 0:
return float('nan')
if correlation.shape[0] == 1:
return 1.0
correlation = np.nan_to_num(correlation, nan=0.0)
eigenvalues = np.abs(np.linalg.eigvalsh(correlation))
total = float(sum((1.0 if value >= 1.0 else 0.0)
+ (value - math.floor(value))
for value in eigenvalues))
return max(1.0, min(total, float(correlation.shape[0])))
def _measurement_columns(frame, gene_column: str,
guide_column: Optional[str]) -> Tuple[str, ...]:
"""Numeric columns that are a response rather than part of the design."""
excluded = {gene_column, guide_column}
excluded.update(_IDENTITY_COLUMNS)
excluded.update(_DESIGN_COLUMNS)
out = []
for name in frame.columns:
if name in excluded or str(name).lower() in excluded:
continue
if not _is_numeric(frame[name]):
continue
out.append(name)
return tuple(out)
def _is_numeric(series) -> bool:
"""Return whether a pandas series has numeric, non-boolean dtype.
:param series: candidate response column.
:returns: ``True`` for numeric dtypes other than Boolean.
"""
import pandas as pd
return bool(pd.api.types.is_numeric_dtype(series)
and not pd.api.types.is_bool_dtype(series))
def _dummy_block(values) -> Tuple[np.ndarray, Tuple[str, ...]]:
"""Treatment-coded dummies for one categorical column, first level out.
A column with one level yields **no** columns rather than a constant one.
That is not a special case to guard: a block that never varies carries no
information, and including it would make the design rank-deficient and the
standard errors meaningless.
"""
levels = sorted({str(value) for value in values})
if len(levels) < 2:
return np.zeros((len(values), 0)), ()
text = np.asarray([str(value) for value in values])
columns = [(text == level).astype(float) for level in levels[1:]]
return np.column_stack(columns), tuple(levels[1:])
def _build_design(frame, gene_column: str, block_columns: Sequence[str],
control_genes: Sequence[str]):
"""Return ``(X, gene_term_names, gene_term_slice, used_blocks)``.
The baseline is the **controls** when the caller names any, because a
screen's effect size is "against the controls" and not "against whichever
gene sorted first" -- and a scan that does not know its controls reports
the control gene itself as the hit whenever the rest of the library moves
together.
"""
n = len(frame)
parts = [np.ones((n, 1))]
used_blocks = []
for column in block_columns:
if column not in frame.columns:
continue
block, levels = _dummy_block(frame[column].to_numpy())
if not levels:
continue
parts.append(block)
used_blocks.append(column)
genes = np.asarray([str(value) for value in frame[gene_column]])
controls = {str(value) for value in control_genes}
levels = sorted(set(genes))
if len(levels) < 2:
raise ScanRefused(
f"{gene_column!r} holds {len(levels)} distinct value(s). With one "
f"gene there is nothing to contrast it against -- the gene term "
f"IS the intercept, and every effect size would be zero.")
terms = [level for level in levels if level not in controls]
if controls and not terms:
raise ScanRefused(
f"every value of {gene_column!r} is a control, so there is no "
f"gene left to measure an effect for.")
if len(terms) == len(levels):
terms = levels[1:]
gene_block = np.column_stack([(genes == term).astype(float)
for term in terms])
start = sum(part.shape[1] for part in parts)
parts.append(gene_block)
design = np.column_stack(parts)
return (design, tuple(terms), slice(start, start + len(terms)),
tuple(used_blocks))
def _fit_columns(design: np.ndarray, responses: np.ndarray,
gene_slice: slice):
"""OLS of every column of ``responses`` on one shared ``design``.
One decomposition for the whole block. The design does not change between
measurements -- only the response does -- so fitting them one at a time
through statsmodels would repeat the same factorisation hundreds of times
for an identical answer.
:returns: ``(betas, standard_errors, residual_sd, df)``, gene terms only.
"""
n, p = design.shape
rank = int(np.linalg.matrix_rank(design))
df = n - rank
if df <= 0:
return None
gram_inverse = np.linalg.pinv(design.T @ design)
coefficients = gram_inverse @ (design.T @ responses)
residuals = responses - design @ coefficients
sigma2 = (residuals ** 2).sum(axis=0) / df
diagonal = np.clip(np.diag(gram_inverse), 0.0, None)
standard_errors = np.sqrt(np.outer(diagonal, sigma2))
return (coefficients[gene_slice], standard_errors[gene_slice],
np.sqrt(sigma2), df)
[docs]
def scan_measurements(frame,
*,
measurements: Optional[Iterable[str]] = None,
gene_column: str = 'gene',
guide_column: Optional[str] = 'grna',
block_columns: Sequence[str] = DEFAULT_BLOCK_COLUMNS,
control_genes: Sequence[str] = (),
within_run_method: str = 'fdr_bh',
across_scan_method: str = 'fdr_bh',
min_wells_per_gene: int = MIN_WELLS_PER_GENE,
alpha: float = 0.05) -> ScanResult:
"""Fit every measurement against the guides and rank them by effect size.
:param frame: one row per **well**, carrying the gene assignment, any
blocking factors, and the measurement columns to scan.
:param measurements: the columns to scan. Default: every numeric column
that is not identity or design (see :data:`_DESIGN_COLUMNS`).
:param gene_column: the independent variable.
:param guide_column: excluded from the scanned columns when present.
:param block_columns: fixed effects fitted alongside the genes and not
reported. Default :data:`DEFAULT_BLOCK_COLUMNS`, the screen key. Absent
or single-valued blocks are dropped.
:param control_genes: the baseline the effects are measured from.
:param within_run_method: correction across the genes of one measurement,
i.e. what a single-measurement analysis would have reported.
:param across_scan_method: correction across the measurements. Any
:mod:`spacr.multiple_testing` method, plus
``'bonferroni_effective'`` -- Bonferroni divided by
:func:`effective_number_of_tests` rather than by the column count.
:param min_wells_per_gene: genes with fewer wells than this are dropped
before fitting, and named in :attr:`ScanResult.genes_dropped`. See
:data:`MIN_WELLS_PER_GENE` -- a gene in one well has nothing
corroborating it, and on real data keeping them made the across-scan
correction fire on 65% of scans over permuted labels instead of 5%.
Pass ``1`` to keep them.
:param alpha: the level both corrections target.
:returns: the :class:`ScanResult`.
:raises ScanRefused: when there is no gene column, fewer than two genes,
or nothing numeric to scan.
"""
import pandas as pd
if gene_column not in frame.columns:
raise ScanRefused(
f"no {gene_column!r} column: a measurement scan with no "
f"independent variable is not an empty result, it is a caller "
f"that has not passed the screen in.")
within_run_method = canonical_method(within_run_method)
effective_bonferroni = (
str(across_scan_method).strip().lower() == 'bonferroni_effective')
if not effective_bonferroni:
across_scan_method = canonical_method(across_scan_method)
else:
across_scan_method = 'bonferroni_effective'
usable = frame[frame[gene_column].notna()].reset_index(drop=True)
genes_dropped: Dict[str, int] = {}
if min_wells_per_gene > 1 and len(usable):
per_gene = usable[gene_column].astype(str).value_counts()
thin = per_gene[per_gene < int(min_wells_per_gene)]
controls = {str(gene) for gene in control_genes}
thin = thin[[gene for gene in thin.index if gene not in controls]]
if len(thin):
genes_dropped = {str(g): int(n) for g, n in thin.items()}
usable = usable[~usable[gene_column].astype(str)
.isin(genes_dropped)].reset_index(drop=True)
kept = usable[gene_column].astype(str).nunique()
if kept < 2:
raise ScanRefused(
f"only {kept} gene(s) have at least {min_wells_per_gene} "
f"wells, so there is nothing to compare. Dropped: "
f"{sorted(genes_dropped)}. A gene in one well has nothing "
f"corroborating it -- its effect is that well's own "
f"deviation, artefacts included. Lower "
f"min_wells_per_gene to keep them, knowing that.")
if measurements is None:
candidates = _measurement_columns(usable, gene_column, guide_column)
else:
candidates = tuple(measurements)
missing = [name for name in candidates if name not in usable.columns]
if missing:
raise ScanRefused(f"no such measurement column(s): {missing}")
if not candidates:
raise ScanRefused(
"no numeric measurement columns to scan. Identity and design "
"columns are excluded on purpose -- regressing the guides on the "
"row index is not an analysis.")
design, gene_terms, gene_slice, used_blocks = _build_design(
usable, gene_column, block_columns, control_genes)
skipped: Dict[str, str] = {}
kept: list = []
for name in candidates:
column = pd.to_numeric(usable[name], errors='coerce').to_numpy(float)
present = np.isfinite(column)
if present.sum() < 3:
skipped[name] = 'too few wells with a value'
continue
if float(np.nanstd(column[present])) <= 0.0:
skipped[name] = 'no variance'
continue
kept.append((name, column, present))
if not kept:
return ScanResult((), skipped, used_blocks, gene_column,
tuple(control_genes), within_run_method,
across_scan_method, float(alpha), float('nan'),
genes_dropped)
groups: Dict[bytes, list] = {}
for name, column, present in kept:
groups.setdefault(present.tobytes(), []).append((name, column, present))
records: list = []
for members in groups.values():
present = members[0][2]
sub_design = design[present]
responses = np.column_stack([column[present] for _n, column, _p in members])
fitted = _fit_columns(sub_design, responses, gene_slice)
if fitted is None:
for name, _column, _present in members:
skipped[name] = 'not enough wells left for the design'
continue
betas, errors, residual_sd, df = fitted
from scipy import stats
for index, (name, column, _present) in enumerate(members):
beta = betas[:, index]
error = errors[:, index]
scale = float(np.nanstd(column[present]))
estimable = residual_sd[index] > 1e-8 * max(scale, 1e-300)
with np.errstate(divide='ignore', invalid='ignore'):
t_values = np.where(error > 0, beta / error, np.nan)
effects = np.where(estimable, beta / residual_sd[index],
np.nan)
p_values = 2.0 * stats.t.sf(np.abs(t_values), df)
p_values = np.where(np.isfinite(t_values), p_values, np.nan)
if not np.isfinite(effects).any():
skipped[name] = 'no estimable gene effect'
continue
top = int(np.nanargmax(np.abs(effects)))
adjusted, rejected = adjust_p_values(
p_values, method=within_run_method, alpha=alpha)
finite = np.isfinite(adjusted)
records.append({
'measurement': name,
'n_wells': int(present.sum()),
'n_genes': len(gene_terms),
'top_gene': gene_terms[top],
'effect_size': float(effects[top]),
'coefficient': float(beta[top]),
'p_value': float(p_values[top]),
'within_run_q': (float(np.min(adjusted[finite]))
if finite.any() else float('nan')),
'within_run_hits': int(rejected.sum()),
'measurement_p': simes_p_value(p_values),
})
if not records:
return ScanResult((), skipped, used_blocks, gene_column,
tuple(control_genes), within_run_method,
across_scan_method, float(alpha), float('nan'),
genes_dropped)
scanned = [record['measurement'] for record in records]
matrix = np.column_stack([
pd.to_numeric(usable[name], errors='coerce').to_numpy(float)
for name in scanned])
m_effective = effective_number_of_tests(matrix)
family = np.asarray([record['measurement_p'] for record in records],
dtype=float)
if effective_bonferroni:
divisor = m_effective if math.isfinite(m_effective) else len(family)
across = np.clip(family * divisor, 0.0, 1.0)
across_rejected = across < alpha
else:
across, across_rejected = adjust_p_values(
family, method=across_scan_method, alpha=alpha)
rows = tuple(
MeasurementEffect(
measurement=record['measurement'],
n_wells=record['n_wells'],
n_genes=record['n_genes'],
top_gene=record['top_gene'],
effect_size=record['effect_size'],
coefficient=record['coefficient'],
p_value=record['p_value'],
within_run_q=record['within_run_q'],
within_run_hits=record['within_run_hits'],
measurement_p=record['measurement_p'],
across_scan_q=float(across[index]),
survives_within_run=bool(record['within_run_hits'] > 0),
survives_across_scan=bool(across_rejected[index]),
)
for index, record in enumerate(records)
)
return ScanResult(rows, skipped, used_blocks, gene_column,
tuple(control_genes), within_run_method,
across_scan_method, float(alpha), float(m_effective),
genes_dropped)