| |
| """Comprehensive wrap-up analysis: address weaknesses, add rigor. |
| |
| No retraining — uses existing fitted results + re-analyzes data. |
| |
| 1. Fair comparison: analytical vs DeepPTR on SAME 300 genes |
| 2. Bootstrap CIs on half-life correlations |
| 3. Validate PT-specific genes against eCLIP RBP targets |
| 4. Examine sci-fate tautology honestly |
| 5. Sparsity analysis: gamma quality vs unspliced detection rate |
| 6. CI coverage breakdown: where does the posterior fail? |
| 7. ARE/NMD enrichment of PT-specific genes |
| 8. Honest limitations table |
| |
| All results saved to output/wrapup/. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import os |
| os.environ["OMP_NUM_THREADS"] = "4" |
| os.environ["MKL_NUM_THREADS"] = "4" |
| os.environ["OPENBLAS_NUM_THREADS"] = "4" |
|
|
| import json |
| import sys |
| from pathlib import Path |
|
|
| import matplotlib |
| matplotlib.use("Agg") |
| import matplotlib.pyplot as plt |
| import numpy as np |
| import pandas as pd |
| from scipy import stats |
|
|
| import torch |
| torch.set_num_threads(4) |
|
|
| sys.path.insert(0, str(Path(__file__).parent)) |
| from _common import set_figure_style |
|
|
| import scptr |
|
|
| OUTPUT_DIR = Path(__file__).parent.parent / "output" / "wrapup" |
| DATA_DIR = Path(scptr.benchmark.__file__).parent / "data" |
|
|
|
|
| def save_fig(fig, name, subdir="figures"): |
| if fig is None: |
| return |
| out_dir = OUTPUT_DIR / subdir |
| out_dir.mkdir(parents=True, exist_ok=True) |
| path = out_dir / f"{name}.png" |
| fig.savefig(path, dpi=150, bbox_inches="tight") |
| plt.close(fig) |
| print(f" Saved: {path}") |
|
|
|
|
| def ensure_dirs(): |
| for sub in ("figures", "results"): |
| (OUTPUT_DIR / sub).mkdir(parents=True, exist_ok=True) |
|
|
|
|
| def select_top_genes(adata, n_top=300): |
| from scipy.sparse import issparse |
| u = adata.layers["unspliced"] |
| if issparse(u): |
| u = np.asarray(u.todense()) |
| u = np.asarray(u, dtype=np.float32) |
| score = u.sum(axis=0) * (u > 0).mean(axis=0) |
| top_idx = np.sort(np.argsort(score)[::-1][:n_top]) |
| return adata.var_names[top_idx].tolist() |
|
|
|
|
| def prepare_analytical(adata_loader): |
| adata = adata_loader() |
| scptr.pp.filter_genes(adata) |
| scptr.pp.normalize_layers(adata) |
| scptr.pp.neighbors(adata, n_neighbors=30) |
| scptr.pp.smooth_layers(adata) |
| scptr.tl.estimate_beta(adata) |
| scptr.tl.estimate_gamma(adata) |
| return adata |
|
|
|
|
| |
| |
| |
|
|
| def analysis_fair_comparison(adata_an, dataset_name, top_genes): |
| """Compare half-life correlation using analytical gamma on the SAME 300 genes.""" |
| print(f"\n{'=' * 60}") |
| print(f"1. FAIR COMPARISON: Same 300 genes ({dataset_name})") |
| print("=" * 60) |
|
|
| |
| prev_file = Path(__file__).parent.parent / "output" / "deep_benchmark" / "results" / f"{dataset_name}_benchmark.json" |
| if prev_file.exists(): |
| with open(prev_file) as f: |
| prev = json.load(f) |
| else: |
| prev = {} |
|
|
| |
| hl_mouse = scptr.datasets.herzog2017_halflives() |
| hl_human = scptr.datasets.schofield2018_halflives() |
|
|
| gamma_all = np.median(adata_an.layers["gamma"], axis=0) |
|
|
| |
| gene_mask = np.isin(adata_an.var_names, top_genes) |
| gamma_300 = gamma_all.copy() |
| gamma_300[~gene_mask] = 0 |
|
|
| results = {} |
| for ref_name, hl_df in [("mouse", hl_mouse), ("human", hl_human)]: |
| |
| corr_full = scptr.benchmark.correlate_with_halflives(adata_an, hl_df) |
|
|
| |
| adata_300 = adata_an[:, top_genes].copy() |
| |
| an_300_idx = [list(adata_an.var_names).index(g) for g in top_genes if g in adata_an.var_names] |
| adata_300.layers["gamma"] = adata_an.layers["gamma"][:, an_300_idx] |
| corr_300 = scptr.benchmark.correlate_with_halflives(adata_300, hl_df) |
|
|
| |
| hl_key = "mouse_herzog" if ref_name == "mouse" else "human_schofield" |
| dp_r = prev.get("halflife", {}).get(hl_key, {}).get("deepptr", {}).get("spearman_r", np.nan) |
| dp_n = prev.get("halflife", {}).get(hl_key, {}).get("deepptr", {}).get("n_genes", 0) |
|
|
| results[ref_name] = { |
| "analytical_all": {"r": corr_full["spearman_r"], "n": corr_full["n_genes"]}, |
| "analytical_300": {"r": corr_300["spearman_r"], "n": corr_300["n_genes"]}, |
| "deepptr_300": {"r": dp_r, "n": dp_n}, |
| } |
|
|
| print(f"\n {ref_name}:") |
| print(f" Analytical (all {adata_an.n_vars} genes): r={corr_full['spearman_r']:.4f} (n={corr_full['n_genes']})") |
| print(f" Analytical (same 300 genes): r={corr_300['spearman_r']:.4f} (n={corr_300['n_genes']})") |
| print(f" DeepPTR (same 300 genes): r={dp_r:.4f} (n={dp_n})") |
|
|
| return results |
|
|
|
|
| |
| |
| |
|
|
| def analysis_bootstrap_ci(adata_an, dataset_name, n_boot=1000): |
| """Bootstrap CIs on half-life correlations.""" |
| print(f"\n{'=' * 60}") |
| print(f"2. BOOTSTRAP CIs ({dataset_name})") |
| print("=" * 60) |
|
|
| hl_human = scptr.datasets.schofield2018_halflives() |
| hl_s = hl_human.set_index("gene_symbol")["half_life_hours"] |
|
|
| gamma_med = np.median(adata_an.layers["gamma"], axis=0) |
| gamma_s = pd.Series(gamma_med, index=adata_an.var_names) |
|
|
| |
| gamma_upper = {g.upper(): g for g in gamma_s.index} |
| hl_upper = {g.upper(): g for g in hl_s.index if isinstance(g, str)} |
| shared_upper = set(gamma_upper.keys()) & set(hl_upper.keys()) |
|
|
| g_vals = np.array([gamma_s[gamma_upper[u]] for u in shared_upper], dtype=float) |
| h_vals = np.array([hl_s[hl_upper[u]] for u in shared_upper], dtype=float) |
|
|
| valid = np.isfinite(g_vals) & np.isfinite(h_vals) & (g_vals > 0) & (h_vals > 0) |
| g_vals, h_vals = g_vals[valid], h_vals[valid] |
| n = len(g_vals) |
|
|
| |
| sp_r, _ = stats.spearmanr(g_vals, h_vals) |
|
|
| |
| rng = np.random.RandomState(42) |
| boot_rs = np.zeros(n_boot) |
| for i in range(n_boot): |
| idx = rng.choice(n, size=n, replace=True) |
| boot_rs[i], _ = stats.spearmanr(g_vals[idx], h_vals[idx]) |
|
|
| ci_lo, ci_hi = np.percentile(boot_rs, [2.5, 97.5]) |
| se = np.std(boot_rs) |
|
|
| print(f" Spearman r = {sp_r:.4f} (n={n})") |
| print(f" 95% CI: [{ci_lo:.4f}, {ci_hi:.4f}]") |
| print(f" Bootstrap SE: {se:.4f}") |
|
|
| result = { |
| "spearman_r": float(sp_r), |
| "n_genes": n, |
| "ci_95_lo": float(ci_lo), |
| "ci_95_hi": float(ci_hi), |
| "bootstrap_se": float(se), |
| } |
| return result |
|
|
|
|
| |
| |
| |
|
|
| def analysis_eclip_validation(dataset_name): |
| """Check if PT-specific genes are enriched for eCLIP RBP targets.""" |
| print(f"\n{'=' * 60}") |
| print(f"3. eCLIP VALIDATION ({dataset_name})") |
| print("=" * 60) |
|
|
| |
| adv_file = Path(__file__).parent.parent / "output" / "deep_advantages" / "results" / f"{dataset_name}_advantages.json" |
| if not adv_file.exists(): |
| print(" [SKIP] No advantage results found") |
| return None |
|
|
| with open(adv_file) as f: |
| adv = json.load(f) |
|
|
| pt_genes = adv.get("disentanglement", {}).get("pt_specific_genes", []) |
| if not pt_genes: |
| print(" [SKIP] No PT-specific genes") |
| return None |
|
|
| |
| eclip = pd.read_csv(DATA_DIR / "eclip_targets.csv") |
| eclip_targets = set(eclip["target_gene"].str.upper()) |
| eclip_by_rbp = eclip.groupby("rbp")["target_gene"].apply(lambda x: set(x.str.upper())).to_dict() |
|
|
| |
| pt_upper = set(g.upper() for g in pt_genes) |
|
|
| |
| |
| pt_de_genes = [g["gene"] for g in adv.get("disentanglement", {}).get("top_pt_de_genes", [])] |
| all_genes_upper = pt_upper | set(g.upper() for g in pt_de_genes) |
|
|
| |
| |
| pt_in_eclip = pt_upper & eclip_targets |
| frac_pt = len(pt_in_eclip) / max(len(pt_upper), 1) |
|
|
| print(f" PT-specific genes: {len(pt_genes)}") |
| print(f" In eCLIP database: {len(pt_in_eclip)} ({frac_pt*100:.0f}%)") |
| if pt_in_eclip: |
| print(f" Validated genes: {sorted(pt_in_eclip)[:20]}") |
|
|
| |
| rbp_hits = {} |
| for rbp, targets in eclip_by_rbp.items(): |
| overlap = pt_upper & targets |
| if overlap: |
| rbp_hits[rbp] = sorted(overlap) |
|
|
| if rbp_hits: |
| print(f"\n RBPs targeting PT-specific genes:") |
| for rbp in sorted(rbp_hits, key=lambda x: len(rbp_hits[x]), reverse=True)[:10]: |
| print(f" {rbp}: {len(rbp_hits[rbp])} targets — {rbp_hits[rbp][:5]}") |
|
|
| |
| |
| result = { |
| "n_pt_genes": len(pt_genes), |
| "n_in_eclip": len(pt_in_eclip), |
| "frac_in_eclip": frac_pt, |
| "validated_genes": sorted(pt_in_eclip), |
| "rbp_hits": {k: v for k, v in sorted(rbp_hits.items(), key=lambda x: len(x[1]), reverse=True)[:15]}, |
| } |
|
|
| return result |
|
|
|
|
| |
| |
| |
|
|
| def analysis_scifate_tautology(): |
| """Honestly examine the sci-fate tautology concern. |
| |
| gamma ∝ beta * Mu / Ms ∝ new / old (approximately) |
| ground truth = new / old |
| |
| How much of the r=0.99 is structural vs learned? |
| """ |
| print(f"\n{'=' * 60}") |
| print("4. SCI-FATE TAUTOLOGY ANALYSIS") |
| print("=" * 60) |
|
|
| import gzip |
| from scipy.io import mmread |
| from scipy.sparse import csc_matrix |
|
|
| CACHE_DIR = Path.home() / ".cache" / "scptr" / "scifate" |
| if not CACHE_DIR.exists(): |
| print(" [SKIP] sci-fate data not cached") |
| return None |
|
|
| |
| cell_ann = pd.read_csv(CACHE_DIR / "GSM3770930_A549_cell_annotate.txt.gz", compression="gzip") |
| gene_ann = pd.read_csv(CACHE_DIR / "GSM3770930_A549_gene_annotate.txt.gz", compression="gzip") |
|
|
| with gzip.open(CACHE_DIR / "GSM3770930_A549_gene_count.txt.gz", "rb") as f: |
| total_mat = csc_matrix(mmread(f)).T |
| with gzip.open(CACHE_DIR / "GSM3770930_A549_gene_count_newly_synthesised.txt.gz", "rb") as f: |
| new_mat = csc_matrix(mmread(f)).T |
|
|
| total = np.asarray(total_mat.todense()) |
| new = np.asarray(new_mat.todense()) |
| old = total - new |
|
|
| mean_new = new.mean(axis=0) |
| mean_old = old.mean(axis=0) |
| mean_total = total.mean(axis=0) |
|
|
| reliable = (mean_total >= 0.5) & (mean_old > 0.1) |
| gt_ratio = np.full(total.shape[1], np.nan) |
| gt_ratio[reliable] = mean_new[reliable] / mean_old[reliable] |
|
|
| |
| |
| |
| |
| |
|
|
| |
| trivial_ratio = np.full(total.shape[1], np.nan) |
| trivial_ratio[reliable] = mean_new[reliable] / mean_old[reliable] |
|
|
| |
| import anndata as ad |
| keep = mean_total >= 0.5 |
| if "gene_type" in gene_ann.columns: |
| is_pc = gene_ann["gene_type"] == "protein_coding" |
| keep = keep & is_pc.values |
|
|
| gene_ann_indexed = gene_ann.set_index("gene_id") |
| adata = ad.AnnData( |
| X=total[:, keep].astype(np.float32), |
| obs=cell_ann.set_index("sample"), |
| var=gene_ann_indexed.iloc[keep].copy(), |
| ) |
| adata.layers["unspliced"] = new[:, keep].astype(np.float32) |
| adata.layers["spliced"] = old[:, keep].astype(np.float32) |
| adata.var_names = adata.var["gene_short_name"].values |
| adata.var_names_make_unique() |
|
|
| scptr.pp.filter_genes(adata, min_unspliced_counts=1, min_unspliced_cells=1) |
| scptr.pp.normalize_layers(adata) |
| scptr.pp.neighbors(adata, n_neighbors=30) |
| scptr.pp.smooth_layers(adata) |
| scptr.tl.estimate_beta(adata) |
| scptr.tl.estimate_gamma(adata) |
|
|
| gamma_med = np.median(adata.layers["gamma"], axis=0) |
| beta_vals = adata.var["beta"].values |
|
|
| |
| gamma_s = pd.Series(gamma_med, index=adata.var_names) |
| beta_s = pd.Series(beta_vals, index=adata.var_names) |
|
|
| |
| gene_names_raw = gene_ann["gene_short_name"].values |
| gt_dict = {} |
| for i, gn in enumerate(gene_names_raw): |
| if isinstance(gn, str) and reliable[i] and gn not in gt_dict: |
| gt_dict[gn] = gt_ratio[i] |
| gt_s = pd.Series(gt_dict) |
|
|
| shared = gamma_s.index.intersection(gt_s.dropna().index) |
| g = gamma_s[shared].values.astype(float) |
| t = gt_s[shared].values.astype(float) |
| b = beta_s[shared].values.astype(float) |
|
|
| valid = np.isfinite(g) & np.isfinite(t) & (g > 0) & (t > 0) & np.isfinite(b) |
| g, t, b = g[valid], t[valid], b[valid] |
|
|
| |
| r_gamma_gt, _ = stats.spearmanr(g, t) |
| r_trivial, _ = stats.spearmanr(t, t) |
|
|
| |
| |
| gamma_over_beta = g / (b + 1e-8) |
| r_residual, _ = stats.spearmanr(gamma_over_beta, t) |
|
|
| |
| beta_cv = np.std(b) / np.mean(b) |
|
|
| |
| r_beta_gamma, _ = stats.spearmanr(b, g) |
|
|
| print(f" n genes: {len(g)}") |
| print(f" gamma vs ground truth: r = {r_gamma_gt:.4f}") |
| print(f" gamma/beta vs GT: r = {r_residual:.4f}") |
| print(f" beta CV: {beta_cv:.4f}") |
| print(f" beta vs gamma: r = {r_beta_gamma:.4f}") |
| print(f"\n Interpretation:") |
| print(f" gamma = beta * (Mu/Ms) ≈ beta * (new/old) = beta * GT") |
| print(f" Since beta CV = {beta_cv:.2f}, beta adds {'modest' if beta_cv < 0.5 else 'substantial'} variation") |
| print(f" After dividing out beta, residual r = {r_residual:.4f}") |
| print(f" → The r={r_gamma_gt:.3f} correlation is {'largely' if r_residual > 0.95 else 'partially'} " |
| f"tautological") |
|
|
| |
| |
| |
| raw_ratio = np.median(new[:, keep], axis=0) / np.clip(np.median(old[:, keep], axis=0), 1e-8, None) |
| raw_s = pd.Series(raw_ratio, index=adata.var_names[:len(raw_ratio)]) |
| shared2 = raw_s.index.intersection(gt_s.dropna().index) |
| r_raw_vals = raw_s[shared2].values.astype(float) |
| t_raw_vals = gt_s[shared2].values.astype(float) |
| v2 = np.isfinite(r_raw_vals) & np.isfinite(t_raw_vals) & (r_raw_vals > 0) & (t_raw_vals > 0) |
| if v2.sum() > 3: |
| r_raw, _ = stats.spearmanr(r_raw_vals[v2], t_raw_vals[v2]) |
| print(f"\n Raw median(new)/median(old) vs GT: r = {r_raw:.4f} (n={v2.sum()})") |
| print(f" scPTR pipeline adds: Δr = {r_gamma_gt - r_raw:.4f}") |
| else: |
| r_raw = np.nan |
|
|
| result = { |
| "r_gamma_gt": float(r_gamma_gt), |
| "r_gamma_over_beta_gt": float(r_residual), |
| "r_raw_ratio_gt": float(r_raw) if not np.isnan(r_raw) else None, |
| "beta_cv": float(beta_cv), |
| "r_beta_gamma": float(r_beta_gamma), |
| "n_genes": len(g), |
| "tautology_severity": "high" if r_residual > 0.98 else "moderate" if r_residual > 0.90 else "low", |
| } |
|
|
| return result |
|
|
|
|
| |
| |
| |
|
|
| def analysis_sparsity(adata_an, dataset_name): |
| """Does gamma quality depend on unspliced detection rate?""" |
| print(f"\n{'=' * 60}") |
| print(f"5. SPARSITY ANALYSIS ({dataset_name})") |
| print("=" * 60) |
|
|
| from scipy.sparse import issparse |
|
|
| u = adata_an.layers["unspliced"] |
| if issparse(u): |
| u = np.asarray(u.todense()) |
| u = np.asarray(u) |
|
|
| |
| frac_detected = (u > 0).mean(axis=0) |
|
|
| gamma_med = np.median(adata_an.layers["gamma"], axis=0) |
|
|
| |
| hl_human = scptr.datasets.schofield2018_halflives() |
| hl_s = hl_human.set_index("gene_symbol")["half_life_hours"] |
|
|
| gamma_upper = {g.upper(): i for i, g in enumerate(adata_an.var_names)} |
| hl_upper = {g.upper(): g for g in hl_s.index if isinstance(g, str)} |
| shared = set(gamma_upper.keys()) & set(hl_upper.keys()) |
|
|
| g_idx = np.array([gamma_upper[u] for u in shared]) |
| h_vals = np.array([hl_s[hl_upper[u]] for u in shared], dtype=float) |
| g_vals = gamma_med[g_idx] |
| det_vals = frac_detected[g_idx] |
|
|
| valid = np.isfinite(g_vals) & np.isfinite(h_vals) & (g_vals > 0) & (h_vals > 0) |
| g_vals, h_vals, det_vals = g_vals[valid], h_vals[valid], det_vals[valid] |
|
|
| |
| quartiles = np.percentile(det_vals, [25, 50, 75]) |
| bins = [ |
| ("Q1 (lowest)", det_vals <= quartiles[0]), |
| ("Q2", (det_vals > quartiles[0]) & (det_vals <= quartiles[1])), |
| ("Q3", (det_vals > quartiles[1]) & (det_vals <= quartiles[2])), |
| ("Q4 (highest)", det_vals > quartiles[2]), |
| ] |
|
|
| records = [] |
| print(f"\n Half-life correlation by unspliced detection rate:") |
| for label, mask in bins: |
| if mask.sum() < 10: |
| continue |
| sp_r, _ = stats.spearmanr(g_vals[mask], h_vals[mask]) |
| records.append({ |
| "quartile": label, |
| "n_genes": int(mask.sum()), |
| "spearman_r": float(sp_r), |
| "median_detection": float(np.median(det_vals[mask])), |
| }) |
| print(f" {label}: r={sp_r:.4f} (n={mask.sum()}, median det={np.median(det_vals[mask]):.2f})") |
|
|
| |
| r_det, p_det = stats.spearmanr(det_vals, np.abs(g_vals)) |
| print(f"\n Detection rate vs |gamma|: r={r_det:.4f} (p={p_det:.2e})") |
|
|
| |
| fig, axes = plt.subplots(1, 2, figsize=(12, 5)) |
|
|
| ax = axes[0] |
| for rec in records: |
| ax.bar(rec["quartile"], abs(rec["spearman_r"]), color="steelblue", alpha=0.7) |
| ax.set_ylabel("|Spearman r| with half-life") |
| ax.set_title(f"{dataset_name}: Half-life r by detection rate") |
| ax.set_xticklabels([r["quartile"] for r in records], rotation=30, ha="right") |
|
|
| ax = axes[1] |
| ax.scatter(det_vals, g_vals, alpha=0.1, s=3, c="steelblue") |
| ax.set_xlabel("Unspliced detection rate") |
| ax.set_ylabel("Median gamma") |
| ax.set_title(f"Detection rate vs gamma (r={r_det:.3f})") |
|
|
| fig.tight_layout() |
| save_fig(fig, f"{dataset_name}_sparsity") |
|
|
| return {"stratified": records, "detection_gamma_r": float(r_det)} |
|
|
|
|
| |
| |
| |
|
|
| def analysis_ci_breakdown(): |
| """Examine where DeepPTR CI coverage fails on synthetic data.""" |
| print(f"\n{'=' * 60}") |
| print("6. CI COVERAGE BREAKDOWN (synthetic)") |
| print("=" * 60) |
|
|
| from scptr.deep.synthetic import generate_kinetic_data |
|
|
| adata, truth = generate_kinetic_data(n_cells=1500, n_genes=100, seed=0) |
|
|
| torch.set_num_threads(4) |
| model, history = scptr.deep.fit_deepptr( |
| adata, d_T=8, d_PT=8, d_hidden=48, n_enc_layers=2, |
| batch_size=256, max_epochs=150, kl_warmup_epochs=20, |
| patience=15, n_posterior_samples=30, |
| device="cpu", seed=0, verbose=False, |
| ) |
|
|
| gamma_true = truth["gamma"] |
| gamma_mean = adata.layers["gamma"] |
| gamma_var = adata.layers["gamma_var"] |
|
|
| z = 1.96 |
| std = np.sqrt(np.clip(gamma_var, 1e-10, None)) |
| lower = gamma_mean - z * std |
| upper = gamma_mean + z * std |
| inside = (gamma_true >= lower) & (gamma_true <= upper) |
|
|
| overall_coverage = float(inside.mean()) |
| print(f" Overall 95% CI coverage: {overall_coverage:.4f} (target: 0.95)") |
|
|
| |
| per_gene_cov = inside.mean(axis=0) |
| |
| per_cell_cov = inside.mean(axis=1) |
|
|
| |
| |
| gene_gamma_std = gamma_true.std(axis=0) |
| r_cov_std, _ = stats.spearmanr(per_gene_cov, gene_gamma_std) |
| print(f" Per-gene coverage vs true gamma std: r={r_cov_std:.4f}") |
|
|
| |
| gene_gamma_mean = gamma_true.mean(axis=0) |
| r_cov_mean, _ = stats.spearmanr(per_gene_cov, gene_gamma_mean) |
| print(f" Per-gene coverage vs true gamma mean: r={r_cov_mean:.4f}") |
|
|
| |
| error = gamma_mean - gamma_true |
| relative_error = np.abs(error) / (gamma_true + 1e-8) |
| mean_rel_error = np.median(relative_error) |
| mean_ci_width = np.median(2 * z * std) |
| mean_true_range = np.median(np.ptp(gamma_true, axis=0)) |
|
|
| print(f"\n Diagnosis:") |
| print(f" Median relative error: {mean_rel_error:.4f}") |
| print(f" Median 95% CI width: {mean_ci_width:.4f}") |
| print(f" Median true range: {mean_true_range:.4f}") |
| print(f" → CI width / true range = {mean_ci_width / max(mean_true_range, 1e-8):.4f}") |
| print(f" → {'Overconfident (CI too narrow)' if overall_coverage < 0.5 else 'Moderate calibration'}") |
|
|
| result = { |
| "overall_coverage": overall_coverage, |
| "target_coverage": 0.95, |
| "per_gene_cov_vs_std_r": float(r_cov_std), |
| "per_gene_cov_vs_mean_r": float(r_cov_mean), |
| "median_relative_error": float(mean_rel_error), |
| "median_ci_width": float(mean_ci_width), |
| "median_true_range": float(mean_true_range), |
| "diagnosis": "overconfident" if overall_coverage < 0.5 else "moderate", |
| } |
|
|
| return result |
|
|
|
|
| |
| |
| |
|
|
| def analysis_pt_gene_enrichment(): |
| """Are PT-specific genes enriched for ARE or NMD targets?""" |
| print(f"\n{'=' * 60}") |
| print("7. ARE/NMD ENRICHMENT OF PT-SPECIFIC GENES") |
| print("=" * 60) |
|
|
| are_genes = set() |
| with open(DATA_DIR / "are_genes.txt") as f: |
| for line in f: |
| are_genes.add(line.strip().upper()) |
|
|
| nmd_genes = set() |
| with open(DATA_DIR / "nmd_genes.txt") as f: |
| for line in f: |
| nmd_genes.add(line.strip().upper()) |
|
|
| results = {} |
| for dataset_name in ("pancreas", "dentate_gyrus"): |
| adv_file = Path(__file__).parent.parent / "output" / "deep_advantages" / "results" / f"{dataset_name}_advantages.json" |
| if not adv_file.exists(): |
| continue |
|
|
| with open(adv_file) as f: |
| adv = json.load(f) |
|
|
| pt_genes = adv.get("disentanglement", {}).get("pt_specific_genes", []) |
| pt_upper = set(g.upper() for g in pt_genes) |
|
|
| are_overlap = pt_upper & are_genes |
| nmd_overlap = pt_upper & nmd_genes |
|
|
| print(f"\n {dataset_name}: {len(pt_genes)} PT-specific genes") |
| print(f" ARE overlap: {len(are_overlap)} ({len(are_overlap)/max(len(pt_upper),1)*100:.0f}%)") |
| if are_overlap: |
| print(f" {sorted(are_overlap)}") |
| print(f" NMD overlap: {len(nmd_overlap)} ({len(nmd_overlap)/max(len(pt_upper),1)*100:.0f}%)") |
| if nmd_overlap: |
| print(f" {sorted(nmd_overlap)}") |
|
|
| results[dataset_name] = { |
| "n_pt_genes": len(pt_genes), |
| "are_overlap": sorted(are_overlap), |
| "nmd_overlap": sorted(nmd_overlap), |
| } |
|
|
| return results |
|
|
|
|
| |
| |
| |
|
|
| def print_limitations(): |
| print(f"\n{'=' * 60}") |
| print("8. HONEST LIMITATIONS") |
| print("=" * 60) |
|
|
| limitations = [ |
| ("Steady-state assumption", "Violated in actively differentiating cells; dynamic mode requires velocity (circular)"), |
| ("Smoothing pre-processing", "Neighbor averaging collapses per-cell variation before gamma estimation"), |
| ("Beta estimation", "Upper-quantile regression is crude; beta errors propagate directly into gamma"), |
| ("Half-life correlations", "r=-0.35 to -0.40 explains ~15% of variance; modest biological signal"), |
| ("sci-fate tautology", "gamma ∝ new/old ≈ ground truth; high correlation is partially structural"), |
| ("DeepPTR CI coverage", "27% for 95% CI; posterior is severely overconfident (amortized VI gap)"), |
| ("Gene subset", "DeepPTR evaluated on 300 genes for CPU tractability; not full genome"), |
| ("No method comparison", "No benchmarking against velVI, DeepVelo, scVI, or other deep methods"), |
| ("Single seed", "No error bars; results may vary across random initializations"), |
| ("PT-specific genes", "No external perturbation validation; could be technical artifacts"), |
| ("Scalability", "Tested on 3K-7K cells; untested on modern 100K+ cell atlases"), |
| ] |
|
|
| for name, desc in limitations: |
| print(f" {name:<25} {desc}") |
|
|
| return limitations |
|
|
|
|
| |
| |
| |
|
|
| def main(): |
| set_figure_style() |
| ensure_dirs() |
|
|
| all_results = {} |
|
|
| |
| datasets = [ |
| ("pancreas", scptr.datasets.pancreas, "clusters"), |
| ("dentate_gyrus", scptr.datasets.dentate_gyrus, "clusters"), |
| ] |
|
|
| for name, loader, cluster_key in datasets: |
| print(f"\n{'#' * 60}") |
| print(f"# {name.upper()}") |
| print(f"{'#' * 60}") |
|
|
| adata_an = prepare_analytical(loader) |
| top_genes = select_top_genes(adata_an, n_top=300) |
| ds_results = {} |
|
|
| |
| ds_results["fair_comparison"] = analysis_fair_comparison(adata_an, name, top_genes) |
|
|
| |
| ds_results["bootstrap_ci"] = analysis_bootstrap_ci(adata_an, name) |
|
|
| |
| ds_results["eclip_validation"] = analysis_eclip_validation(name) |
|
|
| |
| ds_results["sparsity"] = analysis_sparsity(adata_an, name) |
|
|
| all_results[name] = ds_results |
|
|
| |
| all_results["scifate_tautology"] = analysis_scifate_tautology() |
|
|
| |
| all_results["ci_breakdown"] = analysis_ci_breakdown() |
|
|
| |
| all_results["pt_enrichment"] = analysis_pt_gene_enrichment() |
|
|
| |
| limitations = print_limitations() |
| all_results["limitations"] = [{"name": n, "description": d} for n, d in limitations] |
|
|
| |
| with open(OUTPUT_DIR / "results" / "wrapup_results.json", "w") as f: |
| json.dump(all_results, f, indent=2, default=str) |
|
|
| print(f"\n{'=' * 60}") |
| print("WRAP-UP COMPLETE") |
| print("=" * 60) |
| print(f"Results saved to: {OUTPUT_DIR}") |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|