scCS.pairwise ============= .. py:module:: scCS.pairwise .. autoapi-nested-parse:: pairwise.py — PairScorer: pairwise condition comparison for scCS. Extends the single-condition SingleScorer to handle exactly 2 experimental conditions (e.g., treatment vs. control, mutant vs. wild-type). Key design principle: shared embedding --------------------------------------- All conditions are embedded in a SINGLE shared star layout built on the pooled data. This is critical — if each condition had its own embedding, the arm angles would differ and CS values would not be comparable across conditions. Architecture ------------ PairScorer Wraps SingleScorer. Pools both conditions for embedding, then scores each condition separately using cell masks on the shared embedding. Tier 1 — Core pairwise API score_all_conditions() : dict[condition -> CommitmentScoreResult] Tier 2 — Statistical comparison compare_conditions() : permutation test on per-cell fate affinity compute_delta_CS() : ΔCS = nCS_A − nCS_B with bootstrap CI plot_affinity_distributions() : violin/box plots of per-cell affinities Tier 3 — Advanced fit_mixed_model() : linear mixed-effects model on per-cell fate affinity scores via statsmodels MixedLM trajectory_shift() : KS test + Wasserstein distance on pseudotime distributions per fate arm plot_trajectory_shift() : visualization of pseudotime distributions Usage ----- >>> pscorer = scCS.PairScorer( ... adata, ... root='17', ... branches=['homeostatic', 'activated'], ... condition_obs_key='treatment', ... obs_key='leiden', ... ) >>> pscorer.build_embedding(ordering_metric='pseudotime') >>> pscorer.fit() >>> results = pscorer.score_all_conditions() >>> delta = pscorer.compute_delta_CS('control', 'treated') >>> stats = pscorer.compare_conditions(results) >>> pscorer.plot_affinity_distributions(results) >>> shift = pscorer.trajectory_shift(results) >>> pscorer.plot_trajectory_shift(shift) Classes ------- .. autoapisummary:: scCS.pairwise.PairScorer Module Contents --------------- .. py:class:: PairScorer(adata, root: str, branches: List[str], condition_obs_key: str, obs_key: str = 'leiden', n_angle_bins: int = 36, sector_method: Literal['centroid', 'equal'] = 'centroid', copy: bool = False) RNA velocity commitment scorer for pairwise (2-condition) experiments. Builds a SHARED star embedding on the pooled data from both conditions, then scores each condition separately. This ensures arm geometry is identical across conditions, making CS values directly comparable. :param adata: Full single-cell dataset containing both conditions. :type adata: AnnData :param root: Label of the progenitor/root cluster in adata.obs[obs_key]. :type root: str :param branches: Labels of the k terminal fate clusters. :type branches: list of str :param condition_obs_key: Column in adata.obs with condition labels (e.g., 'treatment'). Must contain exactly 2 unique values. :type condition_obs_key: str :param obs_key: Column in adata.obs with cluster labels. Default: 'leiden'. :type obs_key: str :param n_angle_bins: Number of angular bins. Default: 36. :type n_angle_bins: int :param sector_method: Sector definition strategy. :type sector_method: {'centroid', 'equal'} :param copy: Work on a copy of adata. :type copy: bool :raises ValueError: If condition_obs_key has fewer or more than 2 unique values. For 3+ conditions, use MultiScorer instead. .. rubric:: Examples >>> pscorer = PairScorer( ... adata, ... root='17', ... branches=['homeostatic', 'activated'], ... condition_obs_key='treatment', ... obs_key='leiden', ... ) >>> pscorer.build_embedding(ordering_metric='pseudotime') >>> pscorer.fit() >>> results = pscorer.score_all_conditions() >>> delta = pscorer.compute_delta_CS('control', 'treated') >>> stats = pscorer.compare_conditions(results) .. py:attribute:: adata .. py:attribute:: root :value: '' .. py:attribute:: branches .. py:attribute:: condition_obs_key .. py:attribute:: obs_key :value: 'leiden' .. py:attribute:: n_angle_bins :value: 36 .. py:attribute:: sector_method :value: 'centroid' .. py:attribute:: conditions .. py:method:: build_embedding(ordering_metric: Union[str, numpy.ndarray] = 'pseudotime', invert_ordering: bool = False, scale_ordering: bool = False, arm_scale: float = 10.0, jitter: float = 0.3, seed: int = 42, arm_norm: str = 'global', verbose: bool = True) -> PairScorer Build the shared star embedding on pooled data from both conditions. The embedding is built on ALL cells (both conditions pooled), ensuring that arm geometry is identical across conditions. :param ordering_metric: See SingleScorer.build_embedding(). :type ordering_metric: str or np.ndarray :param invert_ordering: :type invert_ordering: bool :param scale_ordering: :type scale_ordering: bool :param arm_scale: :type arm_scale: float :param jitter: :type jitter: float :param seed: :type seed: int :param verbose: :type verbose: bool :rtype: self .. py:method:: refit_pseudotime(scale_01: bool = True, arm_scale: float = 10.0, jitter: float = 0.3, seed: int = 42, arm_norm: str = 'global', verbose: bool = True) -> PairScorer Rebuild the shared embedding using subset-local pseudotime. See SingleScorer.refit_pseudotime(). .. py:method:: fit(verbose: bool = True) -> PairScorer Fit the shared FateMap and project velocity. Must be called after build_embedding(). :rtype: self .. py:method:: score_all_conditions(cell_level: bool = True, k_nn: Optional[int] = None, n_bootstrap: int = 0, bootstrap_ci: float = 0.95, verbose: bool = True) -> Dict[str, scCS.scores.CommitmentScoreResult] Compute commitment scores separately for each condition. Uses the shared embedding and FateMap. Each condition's cells are masked from the shared adata_sub, so arm geometry is identical. :param cell_level: Compute per-cell fate affinity scores. :type cell_level: bool :param k_nn: NN-smoothed entropy neighbors. :type k_nn: int, optional :param n_bootstrap: Bootstrap replicates for CI. 0 = disabled. :type n_bootstrap: int :param bootstrap_ci: CI level for bootstrap. :type bootstrap_ci: float :param verbose: :type verbose: bool :returns: **dict** :rtype: condition_label -> CommitmentScoreResult .. py:method:: compute_delta_CS(condition_a: str, condition_b: str, n_bootstrap: int = 500, ci: float = 0.95, seed: int = 42, verbose: bool = True) -> Dict Compute ΔCS = nCS_A − nCS_B with bootstrap confidence intervals. For each pair of fates (i, j), computes the difference in normalized commitment score between condition A and condition B, with a bootstrap CI obtained by resampling cells within each condition. :param condition_a: Condition labels (must be in self.conditions). :type condition_a: str :param condition_b: Condition labels (must be in self.conditions). :type condition_b: str :param n_bootstrap: Number of bootstrap replicates. Default 500. :type n_bootstrap: int :param ci: Confidence interval level. Default 0.95. :type ci: float :param seed: :type seed: int :param verbose: :type verbose: bool :returns: 'delta_nCS' : np.ndarray (k, k) — nCS_A − nCS_B 'ci_low' : np.ndarray (k, k) — lower CI bound on delta 'ci_high' : np.ndarray (k, k) — upper CI bound on delta 'nCS_A' : np.ndarray (k, k) — nCS for condition A 'nCS_B' : np.ndarray (k, k) — nCS for condition B 'fate_names' : list of str 'condition_a': str 'condition_b': str 'n_bootstrap': int 'ci_level' : float :rtype: dict with keys .. py:method:: compare_conditions(results: Dict[str, scCS.scores.CommitmentScoreResult], test: Literal['permutation', 'kruskal'] = 'permutation', n_permutations: int = 1000, pval_threshold: float = 0.05, seed: int = 42, verbose: bool = True) -> pandas.DataFrame Statistical comparison of per-cell fate affinity scores across conditions. For PairScorer (k=2 conditions), the default test is a permutation test: shuffle condition labels, recompute mean per-cell affinity difference, and get an empirical null distribution. :param results: Output of score_all_conditions() with cell_level=True. :type results: dict :param test: Statistical test to use. Default: 'permutation' (recommended for k=2). :type test: {'permutation', 'kruskal'} :param n_permutations: Number of permutations for the permutation test. Default 1000. :type n_permutations: int :param pval_threshold: Significance threshold. Default 0.05. :type pval_threshold: float :param seed: :type seed: int :param verbose: :type verbose: bool :returns: fate, test, statistic, pval, pval_adj, significant [+ comparison column for pairwise tests] :rtype: pd.DataFrame with columns .. py:method:: plot_affinity_distributions(results: Dict[str, scCS.scores.CommitmentScoreResult], plot_type: Literal['violin', 'box', 'strip'] = 'violin', color_map: Optional[Dict[str, str]] = None, figsize: Optional[Tuple[float, float]] = None, title: Optional[str] = None, save_path: Optional[str] = None) -> matplotlib.figure.Figure Violin/box plots of per-cell fate affinity scores by condition. One panel per fate, showing the distribution of per-cell affinity scores split by condition. :param results: Output of score_all_conditions(cell_level=True). :type results: dict :param plot_type: :type plot_type: {'violin', 'box', 'strip'} :param color_map: condition_label -> hex color. :type color_map: dict, optional :param figsize: :type figsize: tuple, optional :param title: :type title: str, optional :param save_path: :type save_path: str, optional :returns: **fig** :rtype: matplotlib Figure .. py:method:: fit_mixed_model(results: Dict[str, scCS.scores.CommitmentScoreResult], replicate_key: Optional[str] = None, ref_condition: Optional[str] = None, verbose: bool = True) -> pandas.DataFrame Linear mixed-effects model on per-cell fate affinity scores. Models per-cell fate affinity as a function of condition (fixed effect) with optional sample/replicate as a random effect. Model (per fate j): affinity_ij ~ condition_i + (1 | sample_id_i) Uses statsmodels MixedLM. :param results: Output of score_all_conditions(cell_level=True). :type results: dict :param replicate_key: Column in adata_sub.obs with sample/replicate IDs. :type replicate_key: str, optional :param ref_condition: Reference condition for the fixed effect. :type ref_condition: str, optional :param verbose: :type verbose: bool :returns: fate, condition, coef, std_err, z_score, pval, pval_adj, ci_low, ci_high, significant :rtype: pd.DataFrame with columns .. py:method:: trajectory_shift(results: Dict[str, scCS.scores.CommitmentScoreResult], pseudotime_key: str = 'sccs_pseudotime', n_bootstrap: int = 500, seed: int = 42, verbose: bool = True) -> pandas.DataFrame Test whether pseudotime distributions differ across conditions per fate arm. For each fate arm, computes: - Kolmogorov-Smirnov (KS) statistic and p-value - Wasserstein distance (Earth Mover's Distance) - Bootstrap CI on the Wasserstein distance :param results: Output of score_all_conditions(). :type results: dict :param pseudotime_key: Column in adata_sub.obs with pseudotime values. :type pseudotime_key: str :param n_bootstrap: Bootstrap replicates for Wasserstein CI. Default 500. :type n_bootstrap: int :param seed: :type seed: int :param verbose: :type verbose: bool :returns: fate, comparison, ks_stat, ks_pval, wasserstein, wasserstein_ci_low, wasserstein_ci_high, mean_pt_A, mean_pt_B, delta_mean_pt, significant :rtype: pd.DataFrame with columns .. py:method:: plot_trajectory_shift(shift_df: pandas.DataFrame, pseudotime_key: str = 'sccs_pseudotime', color_map: Optional[Dict[str, str]] = None, figsize: Optional[Tuple[float, float]] = None, title: Optional[str] = None, save_path: Optional[str] = None) -> matplotlib.figure.Figure Visualize pseudotime distributions per condition per fate arm. Produces a grid of KDE plots: one row per fate arm. Overlaid KDEs show how pseudotime distributions shift between conditions. :param shift_df: Output of trajectory_shift(). :type shift_df: pd.DataFrame :param pseudotime_key: :type pseudotime_key: str :param color_map: :type color_map: dict, optional :param figsize: :type figsize: tuple, optional :param title: :type title: str, optional :param save_path: :type save_path: str, optional :returns: **fig** :rtype: matplotlib Figure .. py:method:: transfer_labels(results: Dict[str, scCS.scores.CommitmentScoreResult], prefix: str = 'cs_') -> None Transfer per-cell commitment scores to the full adata for all conditions. Calls SingleScorer.transfer_labels() for each condition's result, writing condition-specific columns to adata.obs. :param results: Output of score_all_conditions(cell_level=True). :type results: dict :param prefix: Column prefix. Default: 'cs_'. :type prefix: str .. py:method:: plot_star(result: scCS.scores.CommitmentScoreResult, **kwargs) Radial star embedding plot. .. py:method:: plot_star_grid(results: Dict[str, scCS.scores.CommitmentScoreResult], color_map: Optional[Dict[str, str]] = None, figsize_per_panel: Tuple[float, float] = (6, 6), save_path: Optional[str] = None) -> matplotlib.figure.Figure Side-by-side star embedding plots, one per condition. All panels share the same arm geometry and color scale. :param results: :type results: dict :param color_map: :type color_map: dict, optional :param figsize_per_panel: :type figsize_per_panel: tuple :param save_path: :type save_path: str, optional :returns: **fig** :rtype: matplotlib Figure .. py:method:: plot_rose_grid(results: Dict[str, scCS.scores.CommitmentScoreResult], color_map: Optional[Dict[str, str]] = None, figsize_per_panel: Tuple[float, float] = (5, 5), title: Optional[str] = None, save_path: Optional[str] = None) -> matplotlib.figure.Figure Grid of polar rose plots — one per condition. All panels share the same radial scale, making magnitudes directly comparable across conditions. :param results: :type results: dict :param color_map: :type color_map: dict, optional :param figsize_per_panel: :type figsize_per_panel: tuple :param title: :type title: str, optional :param save_path: :type save_path: str, optional :returns: **fig** :rtype: matplotlib Figure .. py:method:: plot_delta_cs_heatmap(delta_result: dict, **kwargs) -> matplotlib.figure.Figure Heatmap of ΔCS = nCS_A − nCS_B with CI annotation. :param delta_result: Output of compute_delta_CS(). :type delta_result: dict :param \*\*kwargs: Passed to :func:`scCS.plot.plot_delta_cs_heatmap`. :returns: **fig** :rtype: matplotlib Figure .. py:method:: plot_compare_conditions_bar(results: Dict[str, scCS.scores.CommitmentScoreResult], **kwargs) -> matplotlib.figure.Figure Grouped bar chart of nCS per condition. :param results: Output of score_all_conditions(). :type results: dict :param \*\*kwargs: Passed to :func:`scCS.plot.plot_compare_conditions_bar`. :returns: **fig** :rtype: matplotlib Figure .. py:method:: plot_commitment_vector_radar(results: Dict[str, scCS.scores.CommitmentScoreResult], **kwargs) -> matplotlib.figure.Figure Radar / spider chart of commitment vectors per condition. :param results: Output of score_all_conditions(). :type results: dict :param \*\*kwargs: Passed to :func:`scCS.plot.plot_commitment_vector_radar`. :returns: **fig** :rtype: matplotlib Figure .. py:property:: scorer :type: Optional[scCS.single.SingleScorer] The internal SingleScorer used for embedding and scoring. .. py:property:: adata_sub The embedding subset (from the internal SingleScorer). .. py:property:: is_fitted :type: bool