Spaces:
Running
Running
File size: 6,175 Bytes
df4d5b4 | 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 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 | """Statistical validation helpers for benchmark and comparison studies.
These routines are designed for performance reporting rather than biological
inference. They expose sample sizes, uncertainty, effect sizes, sensitivity,
cross-validation summaries and failure modes without silently inventing exact
p-values when an exact distribution is unavailable.
"""
from __future__ import annotations
import math
import random
from collections import Counter
from typing import Callable, Iterable
from app.engines.stats_engine import bootstrap_ci, cohens_d
def _mean(values: Iterable[float]) -> float:
vals = [float(v) for v in values]
return sum(vals) / len(vals) if vals else 0.0
def bootstrap_metric(
values: list[float],
confidence: float = 0.95,
iterations: int = 5000,
seed: int = 0,
) -> dict:
"""Bootstrap a scalar benchmark metric with deterministic resampling."""
result = bootstrap_ci(values, confidence=confidence, n_boot=iterations, seed=seed)
return {
**result,
"reporting_role": "performance_uncertainty",
"interpretation_boundary": "This interval quantifies resampling uncertainty for the supplied observations only.",
}
def paired_effect_size(reference: list[float], candidate: list[float]) -> dict:
n = min(len(reference), len(candidate))
if n == 0:
return {"n": 0, "mean_difference": None, "cohens_d": None}
ref = [float(x) for x in reference[:n]]
cand = [float(x) for x in candidate[:n]]
diffs = [c - r for r, c in zip(ref, cand)]
return {
"n": n,
"mean_reference": _mean(ref),
"mean_candidate": _mean(cand),
"mean_difference": _mean(diffs),
"cohens_d_unpaired_approximation": cohens_d(cand, ref),
"direction": "candidate_minus_reference",
}
def sensitivity_analysis(
parameter_values: list[float],
metric_values: list[float],
baseline_index: int = 0,
) -> dict:
"""Summarise how a reported metric changes across a declared parameter sweep."""
n = min(len(parameter_values), len(metric_values))
if n == 0:
return {"n": 0, "rows": [], "max_absolute_change": None}
params = [float(x) for x in parameter_values[:n]]
metrics = [float(x) for x in metric_values[:n]]
baseline_index = min(max(int(baseline_index), 0), n - 1)
baseline = metrics[baseline_index]
rows = []
for p, m in zip(params, metrics):
rows.append({
"parameter": p,
"metric": m,
"absolute_change_from_baseline": m - baseline,
"relative_change_from_baseline": None if baseline == 0 else (m - baseline) / abs(baseline),
})
return {
"n": n,
"baseline_index": baseline_index,
"baseline_parameter": params[baseline_index],
"baseline_metric": baseline,
"rows": rows,
"max_absolute_change": max(abs(r["absolute_change_from_baseline"]) for r in rows),
"interpretation_boundary": "Sensitivity describes the evaluated parameter grid; it does not prove robustness outside that grid.",
}
def kfold_indices(n: int, folds: int = 5, seed: int = 0) -> list[dict]:
n = max(0, int(n))
folds = max(2, min(int(folds), n)) if n >= 2 else 0
if folds == 0:
return []
indices = list(range(n))
random.Random(seed).shuffle(indices)
buckets = [indices[i::folds] for i in range(folds)]
out = []
for i, test in enumerate(buckets):
test_set = set(test)
train = [j for j in indices if j not in test_set]
out.append({"fold": i + 1, "train_indices": train, "test_indices": test})
return out
def cross_validation_summary(scores: list[float], folds: int | None = None) -> dict:
vals = [float(x) for x in scores]
if not vals:
return {"folds": 0, "scores": [], "mean": None, "sd": None, "min": None, "max": None}
mean = _mean(vals)
sd = math.sqrt(sum((x - mean) ** 2 for x in vals) / (len(vals) - 1)) if len(vals) > 1 else 0.0
return {
"folds": int(folds or len(vals)),
"scores": vals,
"mean": mean,
"sd": sd,
"min": min(vals),
"max": max(vals),
"range": max(vals) - min(vals),
}
def failure_analysis(cases: list[dict]) -> dict:
"""Aggregate benchmark failures without hiding them from the final report."""
total = len(cases)
failed = [case for case in cases if not bool(case.get("passed", case.get("pass", False)))]
reasons = Counter(str(case.get("failure_reason") or case.get("reason") or "unspecified") for case in failed)
domains = Counter(str(case.get("domain") or "unspecified") for case in failed)
return {
"total_cases": total,
"failed_cases": len(failed),
"failure_rate": (len(failed) / total) if total else 0.0,
"failure_reasons": dict(reasons.most_common()),
"failure_domains": dict(domains.most_common()),
"failed_case_ids": [str(c.get("id") or c.get("benchmark_id") or i) for i, c in enumerate(failed)],
"policy": "Failed cases remain visible and are not excluded from aggregate reporting unless a predeclared exclusion rule exists.",
}
def comparison_report(
candidate: list[float],
reference: list[float],
confidence: float = 0.95,
iterations: int = 5000,
seed: int = 0,
) -> dict:
"""One object suitable for a benchmark figure/report panel."""
n = min(len(candidate), len(reference))
cand = [float(x) for x in candidate[:n]]
ref = [float(x) for x in reference[:n]]
diffs = [c - r for c, r in zip(cand, ref)]
return {
"n": n,
"candidate": bootstrap_metric(cand, confidence, iterations, seed),
"reference": bootstrap_metric(ref, confidence, iterations, seed + 1),
"difference": bootstrap_metric(diffs, confidence, iterations, seed + 2),
"effect_size": paired_effect_size(ref, cand),
"statistical_significance": {
"p_value": None,
"status": "not_computed",
"reason": "Use a predeclared validated paired inferential test appropriate to the benchmark design before reporting significance."
},
}
|