Spaces:
Running
Running
| """ | |
| Blood-Brain Omics Benchmark API | |
| FastAPI + DuckDB backend serving benchmark results. | |
| Queries Parquet files remotely via DuckDB httpfs — no data downloaded to disk. | |
| DuckDB reads only the needed columns/row groups per query via HTTP range requests. | |
| Data source modes (in priority order): | |
| 1. Remote: HF Dataset repo via httpfs (default, zero RAM footprint) | |
| 2. Local: files in BENCHMARK_DATA_DIR (for local dev) | |
| Endpoints: | |
| GET /api/v1/registry - Full registry metadata | |
| GET /api/v1/maxn/heatmap - Blood x brain heatmap data | |
| GET /api/v1/maxn/detail - Per-module results for one combination | |
| GET /api/v1/maxn/panel_summary - One panel across all outcomes (mean/min/max) | |
| GET /api/v1/maxn/outcome_summary - One outcome across all panels (mean/min/max) | |
| GET /api/v1/maxn/target_comparison - Per-panel performance for a specific target | |
| GET /api/v1/h2h/blood - Blood H2H comparison | |
| GET /api/v1/h2h/blood/summary - H2H win counts | |
| GET /api/v1/h2h/brain - Brain H2H comparison | |
| GET /api/v1/h2h/models - Model H2H comparison (future) | |
| GET /api/v1/temporal - Temporal decay results | |
| GET /api/v1/features - Feature importance for a target | |
| GET /api/v1/features/cross - Cross-target feature importance | |
| GET /api/v1/health - Health check (DB row counts) | |
| GET /api/v1/ready - Readiness probe (SELECT 1) | |
| """ | |
| import asyncio | |
| import json | |
| import logging | |
| import os | |
| import time | |
| import traceback | |
| from typing import Optional | |
| import duckdb | |
| from fastapi import FastAPI, HTTPException, Query, Request | |
| from fastapi.middleware.cors import CORSMiddleware | |
| from fastapi.responses import JSONResponse, Response | |
| logging.basicConfig( | |
| level=logging.INFO, | |
| format="%(asctime)s %(levelname)s %(name)s: %(message)s", | |
| ) | |
| logger = logging.getLogger("blood_brain_api") | |
| # ============================================================================= | |
| # Configuration | |
| # ============================================================================= | |
| # Local data directory. Holds parquets + registry. On the HF Space the | |
| # start.sh entrypoint downloads these from the HF bucket | |
| # (stasaking/blood-brain-benchmark) into /app/data before launching the | |
| # API. In dev this points at webapp/data with the files already on disk. | |
| DATA_DIR = os.environ.get( | |
| "BENCHMARK_DATA_DIR", | |
| os.path.join(os.path.dirname(__file__), "..", "data") | |
| ) | |
| # DuckDB resource settings — tuned for HF Spaces CPU Basic (2 GB RAM, 2 vCPU). | |
| # Override via environment variables when running on larger hardware. | |
| DUCKDB_MEMORY_LIMIT = os.environ.get("DUCKDB_MEMORY_LIMIT", "512MB") | |
| DUCKDB_THREADS = int(os.environ.get("DUCKDB_THREADS", "1")) | |
| DUCKDB_TEMP_DIR = os.environ.get("DUCKDB_TEMP_DIR", "/tmp/duckdb_tmp") | |
| # Concurrency cap — bound in-flight queries to avoid OOM/process crash on | |
| # CPU Basic. Excess requests get 429 and the frontend can retry. | |
| MAX_INFLIGHT = int(os.environ.get("API_MAX_INFLIGHT", "8")) | |
| # Response cache TTL (seconds) — absorbs repeat queries from rapid clicks. | |
| CACHE_TTL = float(os.environ.get("API_CACHE_TTL", "60")) | |
| CACHE_MAX_ENTRIES = int(os.environ.get("API_CACHE_MAX_ENTRIES", "256")) | |
| app = FastAPI( | |
| title="Blood-Brain Omics Benchmark API", | |
| version="1.0.0", | |
| description="Interactive exploration of blood omics → brain phenotype predictions", | |
| ) | |
| app.add_middleware( | |
| CORSMiddleware, | |
| allow_origins=["*"], | |
| allow_methods=["*"], | |
| allow_headers=["*"], | |
| ) | |
| # ============================================================================= | |
| # Concurrency cap + response cache | |
| # ============================================================================= | |
| # Bound the number of in-flight requests so that bursts cannot OOM/crash the | |
| # DuckDB process on CPU Basic. Excess requests fast-fail with 429 instead of | |
| # wedging the worker. | |
| _inflight_sem = asyncio.Semaphore(MAX_INFLIGHT) | |
| # Tiny TTL cache keyed on (path, query_string). Frontend rapid clicks | |
| # (e.g. switching panels) repeatedly hit the same URLs — caching them | |
| # absorbs the load before it ever reaches DuckDB. | |
| _cache: dict[str, tuple[float, int, bytes, str]] = {} | |
| _CACHEABLE_PATH_PREFIXES = ( | |
| "/api/v1/registry", | |
| "/api/v1/maxn/", | |
| "/api/v1/h2h/", | |
| "/api/v1/temporal", | |
| "/api/v1/features", | |
| ) | |
| def _cache_key(request: Request) -> Optional[str]: | |
| if request.method != "GET": | |
| return None | |
| path = request.url.path | |
| if not any(path.startswith(p) for p in _CACHEABLE_PATH_PREFIXES): | |
| return None | |
| return f"{path}?{request.url.query}" | |
| def _cache_get(key: str) -> Optional[tuple[int, bytes, str]]: | |
| entry = _cache.get(key) | |
| if entry is None: | |
| return None | |
| expires, status, body, ctype = entry | |
| if expires < time.monotonic(): | |
| _cache.pop(key, None) | |
| return None | |
| return status, body, ctype | |
| def _cache_put(key: str, status: int, body: bytes, ctype: str) -> None: | |
| if status != 200: | |
| return | |
| if len(_cache) >= CACHE_MAX_ENTRIES: | |
| # Cheap eviction: drop the oldest-expiring entry. | |
| oldest = min(_cache.items(), key=lambda kv: kv[1][0])[0] | |
| _cache.pop(oldest, None) | |
| _cache[key] = (time.monotonic() + CACHE_TTL, status, body, ctype) | |
| async def cache_and_throttle(request: Request, call_next): | |
| key = _cache_key(request) | |
| if key is not None: | |
| hit = _cache_get(key) | |
| if hit is not None: | |
| status, body, ctype = hit | |
| return Response( | |
| content=body, status_code=status, media_type=ctype, | |
| headers={"x-cache": "hit"}, | |
| ) | |
| # Bound concurrency. If saturated, fail fast with 429 rather than queue | |
| # indefinitely (queueing under sustained load is what crashes the worker). | |
| try: | |
| await asyncio.wait_for(_inflight_sem.acquire(), timeout=0.05) | |
| except asyncio.TimeoutError: | |
| return JSONResponse( | |
| status_code=429, | |
| content={"error": "busy", "detail": "Server busy, please retry."}, | |
| headers={"retry-after": "1"}, | |
| ) | |
| try: | |
| response = await call_next(request) | |
| finally: | |
| _inflight_sem.release() | |
| if key is not None and response.status_code == 200: | |
| # Buffer the streaming body so we can both cache and return it. | |
| body_chunks = [chunk async for chunk in response.body_iterator] | |
| body = b"".join(body_chunks) | |
| ctype = response.headers.get("content-type", "application/json") | |
| _cache_put(key, response.status_code, body, ctype) | |
| return Response( | |
| content=body, status_code=response.status_code, media_type=ctype, | |
| headers={"x-cache": "miss"}, | |
| ) | |
| return response | |
| # ============================================================================= | |
| # Middleware: request logging + unhandled-exception handler | |
| # ============================================================================= | |
| async def log_requests(request: Request, call_next): | |
| start = time.monotonic() | |
| try: | |
| response = await call_next(request) | |
| except Exception: | |
| elapsed = (time.monotonic() - start) * 1000 | |
| logger.error( | |
| "UNHANDLED %s %s — %.1f ms\n%s", | |
| request.method, request.url.path, elapsed, | |
| traceback.format_exc(), | |
| ) | |
| return JSONResponse( | |
| status_code=500, | |
| content={"error": "internal_server_error", "detail": "An unexpected error occurred."}, | |
| ) | |
| elapsed = (time.monotonic() - start) * 1000 | |
| logger.info( | |
| "%s %s %d %.1f ms", | |
| request.method, request.url.path, response.status_code, elapsed, | |
| ) | |
| return response | |
| # ============================================================================= | |
| # Startup: configure DuckDB with httpfs views or local tables | |
| # ============================================================================= | |
| db = None | |
| registry = None | |
| def startup(): | |
| global db, registry | |
| # Ensure DuckDB temp directory exists before connecting. | |
| os.makedirs(DUCKDB_TEMP_DIR, exist_ok=True) | |
| # Initialize DuckDB with resource limits suitable for CPU Basic. | |
| db = duckdb.connect(":memory:") | |
| db.execute(f"SET memory_limit='{DUCKDB_MEMORY_LIMIT}';") | |
| db.execute(f"SET threads={DUCKDB_THREADS};") | |
| db.execute(f"SET temp_directory='{DUCKDB_TEMP_DIR}';") | |
| db.execute("SET preserve_insertion_order=false;") | |
| results_src = os.path.join(DATA_DIR, "benchmark_results.parquet") | |
| features_src = os.path.join(DATA_DIR, "feature_importance.parquet") | |
| logger.info( | |
| "DuckDB settings — memory_limit=%s threads=%d temp_dir=%s data_dir=%s", | |
| DUCKDB_MEMORY_LIMIT, DUCKDB_THREADS, DUCKDB_TEMP_DIR, DATA_DIR, | |
| ) | |
| # Load both tables into memory for reliable, fast queries. | |
| # Total ~100MB in RAM — well within cpu-basic 2GB limit. | |
| try: | |
| db.execute(f""" | |
| CREATE TABLE results AS | |
| SELECT * FROM read_parquet('{results_src}') | |
| """) | |
| except Exception as exc: | |
| logger.error("STARTUP FAILED: could not load results table: %s", exc) | |
| raise RuntimeError(f"Failed to load results table: {exc}") from exc | |
| try: | |
| db.execute(f""" | |
| CREATE TABLE features AS | |
| SELECT * FROM read_parquet('{features_src}') | |
| """) | |
| except Exception as exc: | |
| logger.error("STARTUP FAILED: could not load features table: %s", exc) | |
| raise RuntimeError(f"Failed to load features table: {exc}") from exc | |
| n_results = db.execute("SELECT COUNT(*) FROM results").fetchone()[0] | |
| n_features = db.execute("SELECT COUNT(*) FROM features").fetchone()[0] | |
| logger.info("Tables loaded: %d results, %d features", n_results, n_features) | |
| # Load registry JSON from local data dir. | |
| registry_local = os.path.join(DATA_DIR, "benchmark_registry.json") | |
| try: | |
| with open(registry_local) as f: | |
| registry = json.load(f) | |
| except Exception as e: | |
| logger.warning("Could not load registry from %s: %s", registry_local, e) | |
| registry = {"project": {}, "blood_platforms": {}, | |
| "brain_targets": {}, "phases": {}, "models": {}} | |
| def shutdown(): | |
| global db | |
| if db is not None: | |
| try: | |
| db.close() | |
| logger.info("DuckDB connection closed.") | |
| except Exception as exc: | |
| logger.warning("Error closing DuckDB connection: %s", exc) | |
| db = None | |
| # ============================================================================= | |
| # Helpers | |
| # ============================================================================= | |
| VALID_METRICS = {"r2", "pearson", "mse", "mae", "accuracy", "f1", | |
| "roc_auc", "balanced_accuracy", "pr_auc"} | |
| # Metrics where lower is better (used by every endpoint that compares values). | |
| ASCENDING_METRICS = {"mse", "mae"} | |
| VALID_BASELINES = {"none", "covariates"} | |
| def metric_expr(metric: str, baseline: str, alias: str = "value") -> tuple[str, str]: | |
| """Build a SQL fragment for the metric value, optionally as lift over | |
| the maxn_covonly baseline. Returns (select_expr, extra_join_clause). | |
| Used by maxn-family endpoints. ``baseline='none'`` returns the raw | |
| metric column from the main alias 'r'; ``baseline='covariates'`` joins | |
| a per-(blood, brain, target) maxn_covonly row and returns the | |
| difference (predictors+covariates) − (covariates only). | |
| """ | |
| if baseline == "covariates": | |
| join = ( | |
| " LEFT JOIN results c" | |
| " ON c.phase = 'maxn_covonly'" | |
| " AND c.blood_platform = r.blood_platform" | |
| " AND c.brain_target = r.brain_target" | |
| " AND c.target = r.target" | |
| " AND c.model = r.model" | |
| ) | |
| return f"r.{metric} - c.{metric} AS {alias}", join | |
| return f"r.{metric} AS {alias}", "" | |
| def q(sql: str, params=None): | |
| """Run a query on an isolated DuckDB cursor. | |
| DuckDB Connection objects are NOT safe for concurrent use across | |
| requests — sharing the global ``db`` connection between overlapping | |
| FastAPI requests causes corrupted result state and hung responses. | |
| Cursors created via ``db.cursor()`` are cheap, isolated, and safe | |
| for concurrent queries against the same in-memory database. | |
| """ | |
| cur = db.cursor() | |
| return cur.execute(sql, params) if params is not None else cur.execute(sql) | |
| def validate_metric(metric: str) -> str: | |
| if metric not in VALID_METRICS: | |
| raise HTTPException(400, f"Invalid metric: {metric}. " | |
| f"Must be one of {VALID_METRICS}") | |
| return metric | |
| def covariate_phase(base_phase: str, covariates: str) -> str: | |
| """Map covariates param to actual phase name.""" | |
| if covariates == "none": | |
| return base_phase | |
| elif covariates == "only": | |
| return f"{base_phase}_covonly" | |
| elif covariates == "included": | |
| return f"{base_phase}_withcov" | |
| return base_phase | |
| # ============================================================================= | |
| # Endpoints | |
| # ============================================================================= | |
| def get_registry(): | |
| """Full registry metadata.""" | |
| return registry | |
| def maxn_heatmap( | |
| metric: str = Query("r2", description="Metric to aggregate"), | |
| covariates: str = Query("none", enum=["none", "only", "included"]), | |
| baseline: str = Query("none", enum=["none", "covariates"]), | |
| model: str = Query("TabPFN"), | |
| ): | |
| """ | |
| Blood x brain heatmap data. | |
| Returns median metric across modules for each combination. | |
| If baseline=covariates, the value is metric_lift = withcov - covonly. | |
| """ | |
| validate_metric(metric) | |
| phase = covariate_phase("maxn", covariates) | |
| val_expr, join_clause = metric_expr(metric, baseline, alias="m") | |
| rows = q(f""" | |
| SELECT r.blood_platform, r.brain_target, | |
| MEDIAN({val_expr.replace(' AS m','')}) as value, | |
| MEDIAN(r.n_samples) as n_samples, | |
| COUNT(*) as n_targets | |
| FROM results r{join_clause} | |
| WHERE r.phase = ? AND r.model = ? | |
| GROUP BY r.blood_platform, r.brain_target | |
| ORDER BY r.blood_platform, r.brain_target | |
| """, [phase, model]).fetchall() | |
| if not rows: | |
| return {"rows": [], "cols": [], "values": [], "n_samples": [], | |
| "metric": metric, "covariates": covariates} | |
| # Build matrix | |
| blood_set = sorted(set(r[0] for r in rows)) | |
| brain_set = sorted(set(r[1] for r in rows)) | |
| values = [[None] * len(brain_set) for _ in range(len(blood_set))] | |
| n_samples = [[None] * len(brain_set) for _ in range(len(blood_set))] | |
| blood_idx = {b: i for i, b in enumerate(blood_set)} | |
| brain_idx = {b: i for i, b in enumerate(brain_set)} | |
| for blood, brain, val, ns, nt in rows: | |
| i = blood_idx[blood] | |
| j = brain_idx[brain] | |
| values[i][j] = round(val, 4) if val is not None else None | |
| n_samples[i][j] = int(ns) if ns is not None else None | |
| return { | |
| "rows": blood_set, | |
| "cols": brain_set, | |
| "values": values, | |
| "n_samples": n_samples, | |
| "metric": metric, | |
| "covariates": covariates, | |
| "baseline": baseline, | |
| "model": model, | |
| } | |
| def maxn_detail( | |
| blood: str = Query(..., description="Blood platform name"), | |
| brain: str = Query(..., description="Brain target name"), | |
| covariates: str = Query("none", enum=["none", "only", "included"]), | |
| model: str = Query("TabPFN"), | |
| ): | |
| """Per-module results for one blood x brain combination.""" | |
| phase = covariate_phase("maxn", covariates) | |
| rows = q(""" | |
| SELECT target, task_type, n_samples, r2, pearson, mse, mae, | |
| accuracy, f1, roc_auc, balanced_accuracy, pr_auc | |
| FROM results | |
| WHERE phase = ? AND blood_platform = ? AND brain_target = ? | |
| AND model = ? | |
| ORDER BY COALESCE(r2, balanced_accuracy, accuracy) DESC | |
| """, [phase, blood, brain, model]).fetchall() | |
| return { | |
| "blood": blood, | |
| "brain": brain, | |
| "covariates": covariates, | |
| "model": model, | |
| "targets": [ | |
| {"target": r[0], "task_type": r[1], "n_samples": r[2], | |
| "r2": r[3], "pearson": r[4], "mse": r[5], "mae": r[6], | |
| "accuracy": r[7], "f1": r[8], | |
| "roc_auc": r[9], "balanced_accuracy": r[10], | |
| "pr_auc": r[11]} | |
| for r in rows | |
| ], | |
| } | |
| def h2h_blood( | |
| pair: str = Query(..., description="Platform pair, e.g. SomaScan_vs_TMT"), | |
| brain: str = Query(..., description="Brain target name"), | |
| metric: str = Query("r2"), | |
| covariates: str = Query("none", enum=["none", "included"]), | |
| model: str = Query("TabPFN"), | |
| ): | |
| """Per-module comparison for a blood platform pair on one brain target.""" | |
| validate_metric(metric) | |
| phase = "h2h_blood_withcov" if covariates == "included" else "h2h_blood" | |
| rows = q(f""" | |
| SELECT target, blood_platform, {metric}, n_samples | |
| FROM results | |
| WHERE phase = ? AND h2h_pair = ? AND brain_target = ? AND model = ? | |
| ORDER BY target | |
| """, [phase, pair, brain, model]).fetchall() | |
| # Pivot: target -> {platform_a: val, platform_b: val, n_samples: max(n)} | |
| platforms = sorted(set(r[1] for r in rows)) | |
| targets: dict[str, dict] = {} | |
| for target, platform, val, n_samples in rows: | |
| bucket = targets.setdefault(target, {"_n": 0}) | |
| bucket[platform] = round(val, 4) if val is not None else None | |
| if n_samples and n_samples > bucket["_n"]: | |
| bucket["_n"] = n_samples | |
| return { | |
| "pair": pair, | |
| "brain": brain, | |
| "platforms": platforms, | |
| "metric": metric, | |
| "targets": [ | |
| {"target": t, "n_samples": vals.pop("_n"), **vals} | |
| for t, vals in sorted(targets.items()) | |
| ], | |
| } | |
| def h2h_blood_summary( | |
| metric: str = Query("r2"), | |
| covariates: str = Query("none", enum=["none", "included"]), | |
| model: str = Query("TabPFN"), | |
| brain: Optional[str] = Query(None, description="Restrict win counts to a single brain target"), | |
| ): | |
| """Win counts across blood H2H pairs. | |
| If ``brain`` is provided, wins are counted only against that brain | |
| target (e.g. ``neuropathology``). Otherwise wins are counted across | |
| every brain target in the H2H phase. | |
| """ | |
| validate_metric(metric) | |
| phase = "h2h_blood_withcov" if covariates == "included" else "h2h_blood" | |
| sql = f""" | |
| SELECT h2h_pair, brain_target, target, blood_platform, {metric} | |
| FROM results | |
| WHERE phase = ? AND model = ? AND h2h_pair IS NOT NULL | |
| """ | |
| params = [phase, model] | |
| if brain is not None: | |
| sql += " AND brain_target = ?" | |
| params.append(brain) | |
| sql += " ORDER BY h2h_pair, brain_target, target" | |
| rows = q(sql, params).fetchall() | |
| # Count wins per pair | |
| pair_wins = {} | |
| current = None | |
| buffer = {} | |
| for pair, row_brain, target, platform, val in rows: | |
| key = (pair, row_brain, target) | |
| if key != current: | |
| if current and len(buffer) == 2: | |
| pair_key = current[0] | |
| if pair_key not in pair_wins: | |
| pair_wins[pair_key] = {} | |
| platforms = list(buffer.keys()) | |
| v0, v1 = buffer[platforms[0]], buffer[platforms[1]] | |
| if v0 is not None and v1 is not None: | |
| asc = metric in ASCENDING_METRICS | |
| winner = platforms[0] if (v0 < v1 if asc else v0 > v1) else platforms[1] | |
| pair_wins[pair_key][winner] = pair_wins[pair_key].get(winner, 0) + 1 | |
| current = key | |
| buffer = {} | |
| buffer[platform] = val | |
| # Process last group | |
| if current and len(buffer) == 2: | |
| pair_key = current[0] | |
| if pair_key not in pair_wins: | |
| pair_wins[pair_key] = {} | |
| platforms = list(buffer.keys()) | |
| v0, v1 = buffer[platforms[0]], buffer[platforms[1]] | |
| if v0 is not None and v1 is not None: | |
| asc = metric in ASCENDING_METRICS | |
| winner = platforms[0] if (v0 < v1 if asc else v0 > v1) else platforms[1] | |
| pair_wins[pair_key][winner] = pair_wins[pair_key].get(winner, 0) + 1 | |
| # Note: Covariates_vs_X pairs come from the real h2h_blood_withcov phase | |
| # (visit-matched), where Covariates is a first-class blood platform. | |
| # No synthesis needed. | |
| # Ensure both platforms appear in every pair (with explicit 0 for sweeps). | |
| # Frontend ranking expects two-key win maps, otherwise pairs where one | |
| # platform wins every target get silently dropped from the ranking. | |
| for pair_key, wins in pair_wins.items(): | |
| if "_vs_" in pair_key: | |
| a, b = pair_key.split("_vs_", 1) | |
| wins.setdefault(a, 0) | |
| wins.setdefault(b, 0) | |
| return { | |
| "metric": metric, | |
| "covariates": covariates, | |
| "brain": brain, | |
| "pairs": pair_wins, | |
| } | |
| def h2h_blood_per_outcome( | |
| pair: str = Query(..., description="H2H pair, e.g. ClinLabs_vs_Metabolon"), | |
| metric: str = Query("pearson"), | |
| covariates: str = Query("included", enum=["none", "included"]), | |
| model: str = Query("TabPFN"), | |
| ): | |
| """Per-outcome head-to-head comparison for one blood pair. | |
| Returns one row per brain_target with the mean and best metric value | |
| for each side of the pair, plus per-target win counts. Lets the H2H | |
| page show "for ClinLabs vs Metabolon, here's how each performs on | |
| every outcome group" instead of just the global win count. | |
| """ | |
| validate_metric(metric) | |
| phase = "h2h_blood_withcov" if covariates == "included" else "h2h_blood" | |
| rows = q(f""" | |
| SELECT brain_target, target, blood_platform, {metric} | |
| FROM results | |
| WHERE phase = ? AND model = ? AND h2h_pair = ? | |
| ORDER BY brain_target, target, blood_platform | |
| """, [phase, model, pair]).fetchall() | |
| # Group by (brain_target, target) → {platform: value} | |
| per_target: dict[tuple[str, str], dict[str, Optional[float]]] = {} | |
| platforms_seen: set[str] = set() | |
| for brain_target, target, platform, val in rows: | |
| per_target.setdefault((brain_target, target), {})[platform] = val | |
| platforms_seen.add(platform) | |
| if len(platforms_seen) != 2: | |
| return {"pair": pair, "metric": metric, "covariates": covariates, | |
| "platforms": sorted(platforms_seen), "outcomes": []} | |
| p_a, p_b = sorted(platforms_seen) | |
| asc = metric in ASCENDING_METRICS | |
| # Aggregate per brain_target | |
| by_outcome: dict[str, dict] = {} | |
| for (brain_target, target), vals in per_target.items(): | |
| v_a = vals.get(p_a) | |
| v_b = vals.get(p_b) | |
| if v_a is None or v_b is None: | |
| continue | |
| bucket = by_outcome.setdefault(brain_target, { | |
| "n_targets": 0, "sum_a": 0.0, "sum_b": 0.0, | |
| "wins_a": 0, "wins_b": 0, | |
| "best_a": None, "best_b": None, | |
| }) | |
| bucket["n_targets"] += 1 | |
| bucket["sum_a"] += v_a | |
| bucket["sum_b"] += v_b | |
| if v_a < v_b if asc else v_a > v_b: | |
| bucket["wins_a"] += 1 | |
| else: | |
| bucket["wins_b"] += 1 | |
| if bucket["best_a"] is None or (v_a < bucket["best_a"] if asc else v_a > bucket["best_a"]): | |
| bucket["best_a"] = v_a | |
| if bucket["best_b"] is None or (v_b < bucket["best_b"] if asc else v_b > bucket["best_b"]): | |
| bucket["best_b"] = v_b | |
| outcomes = [] | |
| for brain_target, b in sorted(by_outcome.items()): | |
| n = b["n_targets"] | |
| outcomes.append({ | |
| "brain_target": brain_target, | |
| "n_targets": n, | |
| "mean_a": round(b["sum_a"] / n, 4), | |
| "mean_b": round(b["sum_b"] / n, 4), | |
| "best_a": round(b["best_a"], 4), | |
| "best_b": round(b["best_b"], 4), | |
| "wins_a": b["wins_a"], | |
| "wins_b": b["wins_b"], | |
| }) | |
| return { | |
| "pair": pair, | |
| "metric": metric, | |
| "covariates": covariates, | |
| "platform_a": p_a, | |
| "platform_b": p_b, | |
| "outcomes": outcomes, | |
| } | |
| def h2h_brain( | |
| blood: str = Query(..., description="Blood platform name"), | |
| metric: str = Query("r2"), | |
| covariates: str = Query("none", enum=["none", "included"]), | |
| model: str = Query("TabPFN"), | |
| ): | |
| """Compare brain targets for one blood platform.""" | |
| validate_metric(metric) | |
| phase = "h2h_brain_withcov" if covariates == "included" else "h2h_brain" | |
| rows = q(f""" | |
| SELECT h2h_pair, brain_target, target, {metric} | |
| FROM results | |
| WHERE phase = ? AND blood_platform = ? AND model = ? | |
| ORDER BY h2h_pair, target | |
| """, [phase, blood, model]).fetchall() | |
| # Group by pair | |
| pairs = {} | |
| for pair, brain, target, val in rows: | |
| if pair not in pairs: | |
| pairs[pair] = {} | |
| if target not in pairs[pair]: | |
| pairs[pair][target] = {} | |
| pairs[pair][target][brain] = round(val, 4) if val is not None else None | |
| return { | |
| "blood": blood, | |
| "metric": metric, | |
| "pairs": pairs, | |
| } | |
| def h2h_models( | |
| blood: str = Query(...), | |
| brain: str = Query(...), | |
| metric: str = Query("r2"), | |
| covariates: str = Query("none", enum=["none", "included"]), | |
| ): | |
| """Compare models for one blood x brain combination (future).""" | |
| validate_metric(metric) | |
| phase = covariate_phase("maxn", covariates) | |
| rows = q(f""" | |
| SELECT target, model, {metric} | |
| FROM results | |
| WHERE phase = ? AND blood_platform = ? AND brain_target = ? | |
| ORDER BY target, model | |
| """, [phase, blood, brain]).fetchall() | |
| models = sorted(set(r[1] for r in rows)) | |
| targets = {} | |
| for target, model, val in rows: | |
| if target not in targets: | |
| targets[target] = {} | |
| targets[target][model] = round(val, 4) if val is not None else None | |
| return { | |
| "blood": blood, | |
| "brain": brain, | |
| "models": models, | |
| "metric": metric, | |
| "targets": [ | |
| {"target": t, **vals} | |
| for t, vals in sorted(targets.items()) | |
| ], | |
| } | |
| def temporal( | |
| target: Optional[str] = Query(None, description="Specific target (e.g., gpath)"), | |
| metric: str = Query("r2"), | |
| covariates: str = Query("none", enum=["none", "included"]), | |
| model: str = Query("TabPFN"), | |
| ): | |
| """Temporal decay results: metric across time bins.""" | |
| validate_metric(metric) | |
| phase = "temporal_withcov" if covariates == "included" else "temporal" | |
| if target: | |
| rows = q(f""" | |
| SELECT temporal_bin, brain_target, target, {metric}, n_samples | |
| FROM results | |
| WHERE phase = ? AND model = ? AND target = ? | |
| ORDER BY temporal_bin | |
| """, [phase, model, target]).fetchall() | |
| else: | |
| rows = q(f""" | |
| SELECT temporal_bin, brain_target, target, {metric}, n_samples | |
| FROM results | |
| WHERE phase = ? AND model = ? | |
| ORDER BY temporal_bin, target | |
| """, [phase, model]).fetchall() | |
| results = [ | |
| {"bin": r[0], "brain_target": r[1], "target": r[2], | |
| "value": round(r[3], 4) if r[3] is not None else None, | |
| "n_samples": r[4]} | |
| for r in rows | |
| ] | |
| return { | |
| "metric": metric, | |
| "covariates": covariates, | |
| "results": results, | |
| } | |
| def feature_importance( | |
| blood: str = Query(...), | |
| brain: str = Query(...), | |
| target: str = Query(...), | |
| phase: str = Query("maxn"), | |
| model: str = Query("TabPFN"), | |
| limit: int = Query(30, ge=1, le=100), | |
| ): | |
| """Top features for a specific module/target.""" | |
| rows = q(""" | |
| SELECT feature_name, importance, rank | |
| FROM features | |
| WHERE phase = ? AND blood_platform = ? AND brain_target = ? | |
| AND target = ? AND model = ? | |
| ORDER BY rank | |
| LIMIT ? | |
| """, [phase, blood, brain, target, model, limit]).fetchall() | |
| return { | |
| "blood": blood, | |
| "brain": brain, | |
| "target": target, | |
| "phase": phase, | |
| "features": [ | |
| {"feature": r[0], "importance": round(r[1], 4), "rank": r[2]} | |
| for r in rows | |
| ], | |
| } | |
| def feature_cross_target( | |
| blood: str = Query(...), | |
| brain: str = Query(...), | |
| phase: str = Query("maxn"), | |
| model: str = Query("TabPFN"), | |
| limit: int = Query(30, ge=1, le=100), | |
| ): | |
| """Aggregated feature importance across all modules in a brain target.""" | |
| rows = q(""" | |
| SELECT feature_name, | |
| AVG(importance) as mean_importance, | |
| COUNT(DISTINCT target) as n_targets, | |
| MIN(rank) as best_rank | |
| FROM features | |
| WHERE phase = ? AND blood_platform = ? AND brain_target = ? AND model = ? | |
| GROUP BY feature_name | |
| ORDER BY mean_importance DESC | |
| LIMIT ? | |
| """, [phase, blood, brain, model, limit]).fetchall() | |
| return { | |
| "blood": blood, | |
| "brain": brain, | |
| "phase": phase, | |
| "features": [ | |
| {"feature": r[0], "mean_importance": round(r[1], 4), | |
| "n_targets": r[2], "best_rank": r[3]} | |
| for r in rows | |
| ], | |
| } | |
| def maxn_panel_summary( | |
| blood: str = Query(..., description="Blood platform name"), | |
| metric: str = Query("r2"), | |
| covariates: str = Query("none", enum=["none", "only", "included"]), | |
| baseline: str = Query("none", enum=["none", "covariates"]), | |
| model: str = Query("TabPFN"), | |
| ): | |
| """ | |
| Summary of one predictor panel across all outcome panels. | |
| Returns mean, min, max, n_targets, and median n_samples per outcome. | |
| If baseline=covariates, the metric is the lift over maxn_covonly. | |
| """ | |
| validate_metric(metric) | |
| phase = covariate_phase("maxn", covariates) | |
| val_expr, join_clause = metric_expr(metric, baseline, alias="m") | |
| bare = val_expr.replace(" AS m", "") | |
| rows = q(f""" | |
| SELECT r.brain_target, | |
| AVG({bare}) as mean_val, | |
| MIN({bare}) as min_val, | |
| MAX({bare}) as max_val, | |
| COUNT(*) as n_targets, | |
| MEDIAN(r.n_samples) as n_samples | |
| FROM results r{join_clause} | |
| WHERE r.phase = ? AND r.blood_platform = ? AND r.model = ? | |
| GROUP BY r.brain_target | |
| ORDER BY mean_val DESC | |
| """, [phase, blood, model]).fetchall() | |
| return { | |
| "blood": blood, | |
| "metric": metric, | |
| "covariates": covariates, | |
| "baseline": baseline, | |
| "outcomes": [ | |
| { | |
| "brain_target": r[0], | |
| "mean": round(r[1], 4) if r[1] is not None else None, | |
| "min": round(r[2], 4) if r[2] is not None else None, | |
| "max": round(r[3], 4) if r[3] is not None else None, | |
| "n_targets": r[4], | |
| "n_samples": int(r[5]) if r[5] is not None else None, | |
| } | |
| for r in rows | |
| ], | |
| } | |
| def maxn_outcome_summary( | |
| brain: str = Query(..., description="Brain target name"), | |
| metric: str = Query("r2"), | |
| covariates: str = Query("none", enum=["none", "only", "included"]), | |
| baseline: str = Query("none", enum=["none", "covariates"]), | |
| model: str = Query("TabPFN"), | |
| ): | |
| """ | |
| Summary of one outcome panel across all predictor panels. | |
| Returns mean, min, max, n_targets, and median n_samples per predictor. | |
| If baseline=covariates, the metric is the lift over maxn_covonly. | |
| """ | |
| validate_metric(metric) | |
| phase = covariate_phase("maxn", covariates) | |
| val_expr, join_clause = metric_expr(metric, baseline, alias="m") | |
| bare = val_expr.replace(" AS m", "") | |
| rows = q(f""" | |
| SELECT r.blood_platform, | |
| AVG({bare}) as mean_val, | |
| MIN({bare}) as min_val, | |
| MAX({bare}) as max_val, | |
| COUNT(*) as n_targets, | |
| MEDIAN(r.n_samples) as n_samples | |
| FROM results r{join_clause} | |
| WHERE r.phase = ? AND r.brain_target = ? AND r.model = ? | |
| GROUP BY r.blood_platform | |
| ORDER BY mean_val DESC | |
| """, [phase, brain, model]).fetchall() | |
| return { | |
| "brain": brain, | |
| "metric": metric, | |
| "covariates": covariates, | |
| "baseline": baseline, | |
| "predictors": [ | |
| { | |
| "blood_platform": r[0], | |
| "mean": round(r[1], 4) if r[1] is not None else None, | |
| "min": round(r[2], 4) if r[2] is not None else None, | |
| "max": round(r[3], 4) if r[3] is not None else None, | |
| "n_targets": r[4], | |
| "n_samples": int(r[5]) if r[5] is not None else None, | |
| } | |
| for r in rows | |
| ], | |
| } | |
| def maxn_target_comparison( | |
| brain: str = Query(..., description="Brain target name"), | |
| target: str = Query(..., description="Specific target/module name"), | |
| metric: str = Query("r2"), | |
| covariates: str = Query("none", enum=["none", "only", "included"]), | |
| baseline: str = Query("none", enum=["none", "covariates"]), | |
| model: str = Query("TabPFN"), | |
| ): | |
| """ | |
| Per-predictor performance for a specific target within an outcome panel. | |
| E.g., how each blood platform predicts module m4 of bulkrnaseq_dlpfc. | |
| If baseline=covariates, the value is metric_lift = withcov - covonly. | |
| """ | |
| validate_metric(metric) | |
| phase = covariate_phase("maxn", covariates) | |
| val_expr, join_clause = metric_expr(metric, baseline) | |
| rows = q(f""" | |
| SELECT r.blood_platform, {val_expr}, r.n_samples | |
| FROM results r{join_clause} | |
| WHERE r.phase = ? AND r.brain_target = ? AND r.target = ? AND r.model = ? | |
| ORDER BY value DESC | |
| """, [phase, brain, target, model]).fetchall() | |
| return { | |
| "brain": brain, | |
| "target": target, | |
| "metric": metric, | |
| "covariates": covariates, | |
| "baseline": baseline, | |
| "predictors": [ | |
| { | |
| "blood_platform": r[0], | |
| "value": round(r[1], 4) if r[1] is not None else None, | |
| "n_samples": r[2], | |
| } | |
| for r in rows | |
| ], | |
| } | |
| def health(): | |
| """Health check — returns row counts or degraded status if DB unavailable.""" | |
| if db is None: | |
| return JSONResponse( | |
| status_code=503, | |
| content={"status": "degraded", "detail": "Database not initialized"}, | |
| ) | |
| try: | |
| n_results = q("SELECT COUNT(*) FROM results").fetchone()[0] | |
| n_features = q("SELECT COUNT(*) FROM features").fetchone()[0] | |
| except Exception as exc: | |
| logger.error("Health check query failed: %s", exc) | |
| return JSONResponse( | |
| status_code=503, | |
| content={"status": "degraded", "detail": "Database query failed"}, | |
| ) | |
| return {"status": "ok", "n_results": n_results, "n_features": n_features} | |
| def ready(): | |
| """Readiness probe — verifies DB connectivity with a trivial query.""" | |
| if db is None: | |
| return JSONResponse( | |
| status_code=503, | |
| content={"ready": False, "detail": "Database not initialized"}, | |
| ) | |
| try: | |
| q("SELECT 1").fetchone() | |
| except Exception as exc: | |
| logger.error("Ready check query failed: %s", exc) | |
| return JSONResponse( | |
| status_code=503, | |
| content={"ready": False, "detail": "Database query failed"}, | |
| ) | |
| return {"ready": True} | |