File size: 5,173 Bytes
98bde72 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 | """Scientific metrics, conservative screening, and sequence-diverse selection."""
from __future__ import annotations
import math
from collections import defaultdict
import numpy as np
from .schema import Measurement, Requirement
def p_affinity(kd_nm: float) -> float:
"""Return minus log10 of a dissociation constant in molar units."""
if kd_nm <= 0:
raise ValueError("Kd must be positive")
return 9.0 - math.log10(kd_nm)
def selectivity(kd_target_nm: float, kd_counter_nm: float) -> float:
"""Log10 selectivity, positive for preferential binding to the target."""
return p_affinity(kd_target_nm) - p_affinity(kd_counter_nm)
def normalized_margin(m: Measurement, r: Requirement, conservative: bool = True) -> float:
if m.endpoint != r.endpoint or m.unit != r.unit or m.assay != r.assay:
raise ValueError("endpoint, unit, or assay mismatch")
value = m.value
if conservative:
bound = m.lower if r.direction == "ge" else m.upper
if bound is None:
raise ValueError("conservative ranking requires an interval")
value = bound
return (value - r.threshold) / r.scale if r.direction == "ge" else (r.threshold - value) / r.scale
def candidate_score(measurements: list[Measurement], requirements: list[Requirement]) -> float:
"""Worst margin over the supplied computational ranking requirements."""
values = []
for r in requirements:
if not r.required:
continue
matched = [m for m in measurements if m.endpoint == r.endpoint]
if len(matched) != 1:
raise ValueError(f"expected one declared estimate for {r.endpoint}")
values.append(normalized_margin(matched[0], r))
if not values:
raise ValueError("no required endpoints")
return min(values)
def identity(a: str, b: str) -> float:
"""One minus normalized Levenshtein distance, including unequal lengths."""
row = list(range(len(b) + 1))
for i, ca in enumerate(a, 1):
nxt = [i]
for j, cb in enumerate(b, 1):
nxt.append(min(nxt[-1] + 1, row[j] + 1, row[j-1] + (ca != cb)))
row = nxt
return 1.0 - row[-1] / max(len(a), len(b), 1)
def diverse_select(sequences: dict[str, str], scores: dict[str, float], k: int,
penalty: float = 0.25, max_identity: float = 0.8) -> list[str]:
"""Greedy quality/diversity selection with an explicit redundancy ceiling."""
if k < 0 or penalty < 0 or not 0 <= max_identity <= 1:
raise ValueError("invalid selection settings")
chosen = []
remaining = set(scores) & set(sequences)
while remaining and len(chosen) < k:
utilities = {}
for cid in remaining:
sim = max((identity(sequences[cid], sequences[c]) for c in chosen), default=0.)
if sim <= max_identity and math.isfinite(scores[cid]):
utilities[cid] = scores[cid] - penalty * sim
if not utilities:
break
best = min(utilities, key=lambda x: (-utilities[x], x))
chosen.append(best)
remaining.remove(best)
return chosen
def conformal_radius(y_true, y_pred, scales, alpha=0.1) -> float:
"""Finite-sample split-conformal radius for normalized absolute residuals.
Calibration rows must be independent of fitting and final selection. Returns
infinity when the requested quantile exceeds the finite calibration sample.
"""
y, p, s = map(lambda x: np.asarray(x, dtype=float), (y_true, y_pred, scales))
if y.shape != p.shape or y.shape != s.shape or y.ndim != 1 or len(y) == 0:
raise ValueError("expected equal nonempty one-dimensional arrays")
if not 0 < alpha < 1 or np.any(s <= 0) or not np.all(np.isfinite([y,p,s])):
raise ValueError("invalid calibration values")
errors = np.abs(y-p)/s
rank = math.ceil((len(errors)+1)*(1-alpha))
return float(np.sort(errors)[rank-1]) if rank <= len(errors) else math.inf
def grouped_bootstrap(rows: list[dict], value: str, group: str, seed=2027, n=2000) -> dict:
"""Equal-weight group mean and percentile interval across independent tasks."""
groups = defaultdict(list)
for r in rows:
groups[r[group]].append(float(r[value]))
x = np.array([np.mean(v) for v in groups.values()])
if len(x) < 2:
raise ValueError("at least two independent groups are required")
rng = np.random.default_rng(seed)
draws = x[rng.integers(len(x), size=(n,len(x)))].mean(axis=1)
return {"mean": float(x.mean()), "lower": float(np.quantile(draws,.025)),
"upper": float(np.quantile(draws,.975)), "groups": len(x)}
def joint_success(measurements: list[Measurement], requirements: list[Requirement]) -> bool:
"""Require every registered endpoint to pass on experimental observations."""
for r in requirements:
if not r.required:
continue
rows = [m for m in measurements if m.endpoint == r.endpoint and m.kind == "experiment"]
if len(rows) != 1 or rows[0].censor != "none":
return False
if normalized_margin(rows[0],r,conservative=False) < 0:
return False
return True
|