Spaces:
Sleeping
Sleeping
| """Conservative, deterministic evidence classification and trial-planning statistics. | |
| Request SESOI units are SD units for SMD, original outcome units for MD, and | |
| *natural-log* units for logOR/logRR/logHR. Raw OR/RR/HR request scales also work, | |
| but their SESOI must be a multiplicative margin > 1 (e.g. OR=1.25 maps to | |
| log(1.25), with equivalence bounds [1/1.25, 1.25]). Study ratio estimates and | |
| interval endpoints are always log-transformed before analysis. | |
| References: | |
| - https://www.cochrane.org/authors/handbooks-and-manuals/handbook/current/chapter-06 | |
| (log scales, CI-to-SE, and estimate-plus-exact-Wald-p reconstruction). | |
| - https://www.statsmodels.org/stable/generated/statsmodels.stats.meta_analysis.combine_effects.html | |
| (DerSimonian-Laird, method_re="chi2"). We explicitly clamp negative tau². | |
| - https://doi.org/10.1002/pst.175 (assurance as expected power). | |
| - Cochrane Handbook 6.4-6.5 and Hedges (1981): effect sizes from reported arm-level | |
| means/SDs or event counts, with the small-sample correction for standardized differences. | |
| - Egger et al. (1997), BMJ 315:629 (regression test for funnel-plot asymmetry), applied | |
| only to pools of at least ten studies as Cochrane Handbook 13.3.5.3 recommends. | |
| Assurance here is two-sided statistical significance, including effects in the | |
| unfavorable direction. It is not a probability of clinically meaningful benefit. | |
| All intervals and power calculations use a normal approximation. | |
| """ | |
| from __future__ import annotations | |
| import calendar | |
| import math | |
| import re | |
| import warnings as python_warnings | |
| from collections import defaultdict | |
| from datetime import UTC, date, datetime | |
| from statistics import NormalDist | |
| from typing import Any | |
| import numpy as np | |
| from scipy.stats import beta as beta_dist | |
| from scipy.stats import t as student_t | |
| from statsmodels.stats.meta_analysis import combine_effects | |
| _NORMAL = NormalDist() | |
| _Z95 = _NORMAL.inv_cdf(0.975) | |
| _MDE_FACTOR = _Z95 + _NORMAL.inv_cdf(0.8) | |
| _RATIO_TYPES = {"OR": "logOR", "RR": "logRR", "HR": "logHR"} | |
| _MAX_REQUIRED_N = 10_000_000 | |
| # Scales computable from arm-level summaries, in the order used when none was requested. | |
| DERIVED_SCALES = ("SMD", "logOR", "MD", "logRR") | |
| # Why a study is inconclusive. Only ``wide_interval`` describes the study itself (it | |
| # could not tell no effect from a meaningful one); the others describe what we could read. | |
| INCONCLUSIVE_REASONS = ( | |
| "wide_interval", "mixed_result", "no_result", "other_scale", "no_threshold", "review" | |
| ) | |
| _MAX_PLAUSIBLE_SMD = 5.0 | |
| # Funnel-plot asymmetry tests have little power below ten studies (Cochrane 13.3.5.3). | |
| EGGER_MIN_STUDIES = 10 | |
| # Conventional screening level for asymmetry tests, which are underpowered at 0.05. | |
| EGGER_ALPHA = 0.10 | |
| # Pseudo-observations one verdict adds to the Beta posterior, by how its numbers were read. | |
| # A quoted CI is one observation; a text-only claim is half of one. | |
| PURSUIT_TIER_WEIGHTS = {"numeric": 1.0, "derived": 1.0, "reconstructed": 0.75, "text_only": 0.5} | |
| # Share of the informative weight on the minority side above which the record is contested. | |
| PURSUIT_CONFLICT_SHARE = 0.3 | |
| def _number(value: Any) -> float | None: | |
| if value is None or isinstance(value, bool): | |
| return None | |
| try: | |
| result = float(value) | |
| except (TypeError, ValueError, OverflowError): | |
| return None | |
| return result if math.isfinite(result) else None | |
| def _text(value: Any) -> str: | |
| return " ".join(str(value or "").casefold().split()) | |
| def _effect_type(value: Any) -> str | None: | |
| cleaned = re.sub(r"[\s_()-]", "", str(value or "")).upper() | |
| aliases = { | |
| "SMD": "SMD", | |
| "STANDARDIZEDMEANDIFFERENCE": "SMD", | |
| "STANDARDISEDMEANDIFFERENCE": "SMD", | |
| "COHENSD": "SMD", | |
| "HEDGESG": "SMD", | |
| "MD": "MD", | |
| "MEANDIFFERENCE": "MD", | |
| "DIFFERENCEINMEANS": "MD", | |
| "OR": "OR", | |
| "ODDSRATIO": "OR", | |
| "RR": "RR", | |
| "RISKRATIO": "RR", | |
| "RELATIVERISK": "RR", | |
| "HR": "HR", | |
| "HAZARDRATIO": "HR", | |
| "LOGOR": "logOR", | |
| "LOGODDSRATIO": "logOR", | |
| "LNOR": "logOR", | |
| "LOGRR": "logRR", | |
| "LOGRISKRATIO": "logRR", | |
| "LNRR": "logRR", | |
| "LOGHR": "logHR", | |
| "LOGHAZARDRATIO": "logHR", | |
| "LNHR": "logHR", | |
| } | |
| if cleaned in aliases: | |
| return aliases[cleaned] | |
| # Extracted scale names are free text ("mean difference in IOP (mmHg)"). Log scales | |
| # and the standardized difference are tested first so they are not read as plain ones. | |
| for name in ("LOGODDSRATIO", "LOGRISKRATIO", "LOGHAZARDRATIO", "STANDARDIZEDMEANDIFFERENCE", | |
| "STANDARDISEDMEANDIFFERENCE", "HAZARDRATIO", "ODDSRATIO", "RISKRATIO", | |
| "RELATIVERISK", "MEANDIFFERENCE", "DIFFERENCEINMEANS"): | |
| if name in cleaned: | |
| return aliases[name] | |
| return None | |
| def _analysis_type(value: Any) -> str | None: | |
| kind = _effect_type(value) | |
| return _RATIO_TYPES.get(kind, kind) | |
| def _margin(sesoi: Any, effect_type: str) -> float | None: | |
| margin = _number(sesoi) | |
| if margin is None or margin <= 0 or _effect_type(effect_type) is None: | |
| return None | |
| if _effect_type(effect_type) in _RATIO_TYPES: | |
| return math.log(margin) if margin > 1 else None | |
| return margin | |
| def _completion_date(value: Any) -> date | None: | |
| """Use the latest possible day for partial dates, avoiding early overdue labels.""" | |
| if isinstance(value, datetime): | |
| return value.date() | |
| if isinstance(value, date): | |
| return value | |
| try: | |
| text = str(value or "").strip() | |
| if re.fullmatch(r"\d{4}", text): | |
| return date(int(text), 12, 31) | |
| if re.fullmatch(r"\d{4}-\d{2}", text): | |
| year, month = map(int, text.split("-")) | |
| return date(year, month, calendar.monthrange(year, month)[1]) | |
| return date.fromisoformat(text[:10]) | |
| except (ValueError, TypeError): | |
| return None | |
| def _more_than_twelve_months_ago(value: Any, today: date) -> bool: | |
| completed = _completion_date(value) | |
| if completed is None or completed.year >= 9999: | |
| return False | |
| anniversary = completed.replace( | |
| year=completed.year + 1, | |
| day=min(completed.day, calendar.monthrange(completed.year + 1, completed.month)[1]), | |
| ) | |
| return today > anniversary | |
| def _is_registry(study: dict) -> bool: | |
| return study.get("source") in {"ctgov", "merged"} or bool(study.get("nct_ids")) | |
| def _unreported(study: dict, today: date) -> bool: | |
| # Unknown reporting status cannot be treated as confirmed absence of results. | |
| return ( | |
| _is_registry(study) | |
| and str(study.get("overall_status", "")).upper() == "COMPLETED" | |
| and study.get("has_results") is False | |
| and study.get("has_linked_publication") is False | |
| and not study.get("result_pmids") | |
| and study.get("source") not in {"merged", "openalex"} | |
| and _more_than_twelve_months_ago(study.get("primary_completion_date"), today) | |
| ) | |
| def derived_field(scale: str, name: str) -> str: | |
| return f"derived_{scale.lower()}_{name}" | |
| def _count(value: Any) -> int | None: | |
| number = _number(value) | |
| return int(number) if number is not None and number >= 0 and number.is_integer() else None | |
| def derive_effects(study: dict) -> dict[str, dict]: | |
| """Effect sizes computed from arm-level summaries, as intervention minus comparator. | |
| Nothing is imputed: a continuous effect needs both means, both SDs and both arm | |
| sizes; a binary effect needs both event counts and both arm sizes. Hazard ratios | |
| cannot be recovered from summaries and are never derived. | |
| """ | |
| effects: dict[str, dict] = {} | |
| n1, n2 = _count(study.get("n_intervention")), _count(study.get("n_comparator")) | |
| if n1 is None or n2 is None or n1 < 2 or n2 < 2: | |
| return effects | |
| m1, m2 = _number(study.get("mean_intervention")), _number(study.get("mean_comparator")) | |
| sd1, sd2 = _number(study.get("sd_intervention")), _number(study.get("sd_comparator")) | |
| if None not in (m1, m2, sd1, sd2) and sd1 > 0 and sd2 > 0: | |
| difference = m1 - m2 | |
| pooled_sd = math.sqrt(((n1 - 1) * sd1 * sd1 + (n2 - 1) * sd2 * sd2) / (n1 + n2 - 2)) | |
| d = difference / pooled_sd | |
| # Arms several pooled SDs apart almost always mean a standard error was reported | |
| # as an SD, so neither the SMD nor the SE of the mean difference is usable. | |
| if abs(d) <= _MAX_PLAUSIBLE_SMD: | |
| correction = 1 - 3 / (4 * (n1 + n2 - 2) - 1) | |
| effects["MD"] = { | |
| "estimate": difference, | |
| "se": math.sqrt(sd1 * sd1 / n1 + sd2 * sd2 / n2), | |
| "note": "Mean difference and SE computed from reported arm means, SDs and sizes.", | |
| } | |
| effects["SMD"] = { | |
| "estimate": correction * d, | |
| "se": correction * math.sqrt((n1 + n2) / (n1 * n2) + d * d / (2 * (n1 + n2))), | |
| "note": "Hedges' g computed from reported arm means, SDs and sizes (pooled SD, " | |
| "small-sample correction).", | |
| } | |
| e1, e2 = _count(study.get("events_intervention")), _count(study.get("events_comparator")) | |
| if e1 is not None and e2 is not None and e1 <= n1 and e2 <= n2: | |
| cells = [float(e1), float(n1 - e1), float(e2), float(n2 - e2)] | |
| # With no events, or only events, in both arms the ratio carries no information. | |
| if not (e1 == e2 == 0 or (e1 == n1 and e2 == n2)): | |
| note = "computed from reported arm event counts and sizes" | |
| if 0 in cells: | |
| cells = [cell + 0.5 for cell in cells] | |
| note += " with a 0.5 continuity correction for an empty cell" | |
| a, b, c, d_ = cells | |
| effects["logOR"] = { | |
| "estimate": math.log(a * d_ / (b * c)), | |
| "se": math.sqrt(1 / a + 1 / b + 1 / c + 1 / d_), | |
| "note": f"Log odds ratio {note}.", | |
| } | |
| variance = 1 / a - 1 / (a + b) + 1 / c - 1 / (c + d_) | |
| if variance > 0: | |
| effects["logRR"] = { | |
| "estimate": math.log((a / (a + b)) / (c / (c + d_))), | |
| "se": math.sqrt(variance), | |
| "note": f"Log risk ratio {note}.", | |
| } | |
| return { | |
| scale: effect | |
| for scale, effect in effects.items() | |
| if math.isfinite(effect["estimate"]) and math.isfinite(effect["se"]) and effect["se"] > 0 | |
| } | |
| def _numeric_evidence(study: dict, requested: str | None = None) -> dict: | |
| """Reported numbers first; arm-level derivation only fills a gap on the requested scale. | |
| A reported interval on the requested scale is never replaced, so registry | |
| primary-outcome analyses stay authoritative. | |
| """ | |
| result = _reported_evidence(study) | |
| derived = derive_effects(study) | |
| for scale in DERIVED_SCALES: | |
| effect = derived.get(scale) | |
| for name in ("estimate", "ci_low", "ci_high"): | |
| result[derived_field(scale, name)] = None | |
| if effect: | |
| result[derived_field(scale, "estimate")] = effect["estimate"] | |
| result[derived_field(scale, "ci_low")] = effect["estimate"] - _Z95 * effect["se"] | |
| result[derived_field(scale, "ci_high")] = effect["estimate"] + _Z95 * effect["se"] | |
| target = _analysis_type(requested) | |
| reported = result["analysis_ci"] is not None | |
| if reported and (target is None or result["analysis_effect_type"] == target): | |
| return result | |
| scale = target if target in derived else None | |
| if scale is None and not reported: | |
| scale = next((name for name in DERIVED_SCALES if name in derived), None) | |
| if scale is None: | |
| return result | |
| effect = derived[scale] | |
| low, high = effect["estimate"] - _Z95 * effect["se"], effect["estimate"] + _Z95 * effect["se"] | |
| result.update( | |
| analysis_effect_type=scale, | |
| analysis_estimate=effect["estimate"], | |
| analysis_se=effect["se"], | |
| analysis_ci=[low, high], | |
| analysis_ci_low=low, | |
| analysis_ci_high=high, | |
| evidence_tier="derived", | |
| mde=_MDE_FACTOR * effect["se"], | |
| numeric_notes=[*result["numeric_notes"], effect["note"]], | |
| ) | |
| return result | |
| def stored_analysis(study: dict) -> dict: | |
| """Scale-neutral analysis fields for the index: reported numbers, else the first derived. | |
| Buckets are written on the default scale, but a query may use any scale. Storing the | |
| reported analysis (rather than whatever the default scale selected) lets the | |
| aggregation script find it, with ``derived_*`` fields covering every other scale. | |
| """ | |
| evidence = _numeric_evidence(study) | |
| keys = ("analysis_effect_type", "analysis_estimate", "analysis_se", "analysis_ci", | |
| "analysis_ci_low", "analysis_ci_high", "evidence_tier", "mde", "numeric_notes") | |
| return {key: evidence[key] for key in keys} | |
| def _reported_evidence(study: dict) -> dict: | |
| kind = _effect_type(study.get("effect_type")) | |
| result = { | |
| "analysis_effect_type": _analysis_type(kind), | |
| "analysis_estimate": None, | |
| "analysis_se": None, | |
| "analysis_ci": None, | |
| "analysis_ci_low": None, | |
| "analysis_ci_high": None, | |
| "evidence_tier": "text_only", | |
| "mde": None, | |
| "numeric_notes": [], | |
| } | |
| if kind is None: | |
| if any(study.get(key) is not None for key in ("estimate", "ci_low", "ci_high")): | |
| result["numeric_notes"].append("Effect scale is missing or unsupported.") | |
| return result | |
| estimate = _number(study.get("estimate")) | |
| low, high = _number(study.get("ci_low")), _number(study.get("ci_high")) | |
| if kind in _RATIO_TYPES: | |
| if any(value is not None and value <= 0 for value in (estimate, low, high)): | |
| result["numeric_notes"].append( | |
| "Ratio estimates and confidence bounds must be positive." | |
| ) | |
| return result | |
| estimate, low, high = ( | |
| math.log(value) if value is not None else None for value in (estimate, low, high) | |
| ) | |
| result["numeric_notes"].append("Ratio estimate and CI transformed to natural-log units.") | |
| result["analysis_estimate"] = estimate | |
| se = None | |
| tier = "numeric" | |
| sides = str(study.get("ci_sides", "TWO_SIDED")).upper().replace("-", "_") | |
| if sides in {"ONE_SIDED", "ONESIDED", "1"}: | |
| result["numeric_notes"].append( | |
| "One-sided inference cannot establish a two-sided 95% interval." | |
| ) | |
| return result | |
| if low is not None and high is not None: | |
| if low >= high or (estimate is not None and not low <= estimate <= high): | |
| result["numeric_notes"].append( | |
| "Invalid or inconsistent confidence interval was excluded." | |
| ) | |
| return result | |
| level = _number(study.get("ci_level", study.get("confidence_level", 0.95))) | |
| if level is not None and level > 1: | |
| level /= 100 | |
| if level is None or not 0 < level < 1: | |
| result["numeric_notes"].append("Invalid confidence level; interval was excluded.") | |
| return result | |
| z = _NORMAL.inv_cdf((1 + level) / 2) | |
| if z <= 0 or not math.isfinite(z): | |
| result["numeric_notes"].append( | |
| "Confidence level is outside usable numerical precision." | |
| ) | |
| return result | |
| se = (high - low) / (2 * z) | |
| if not math.isclose(level, 0.95): | |
| if estimate is None: | |
| result["numeric_notes"].append( | |
| "A non-95% CI without an estimate cannot be converted." | |
| ) | |
| return result | |
| low, high = estimate - _Z95 * se, estimate + _Z95 * se | |
| tier = "reconstructed" | |
| result["numeric_notes"].append("95% CI reconstructed under a normal approximation.") | |
| elif estimate is not None: | |
| # A reported SE is usable on a difference/log scale. A raw-ratio SE is | |
| # not a log-scale SE and is deliberately not silently reinterpreted. | |
| reported_se = _number(study.get("se", study.get("standard_error"))) | |
| if reported_se is not None and reported_se > 0 and kind not in _RATIO_TYPES: | |
| se = reported_se | |
| result["numeric_notes"].append( | |
| "95% normal CI calculated from reported estimate and SE." | |
| ) | |
| else: | |
| p = _number(study.get("p_value")) | |
| modifier = _text(study.get("p_value_operator", study.get("p_value_modifier"))) | |
| exact = study.get("p_value_is_exact") is not False and modifier in {"", "="} | |
| if p is not None and 0 < p < 1 and p / 2 > 0 and estimate != 0 and exact: | |
| # inv_cdf(p/2), unlike inv_cdf(1-p/2), remains finite for tiny p. | |
| z = -_NORMAL.inv_cdf(p / 2) | |
| se = abs(estimate) / z | |
| tier = "reconstructed" | |
| result["numeric_notes"].append( | |
| "SE reconstructed from estimate and exact p assuming a two-sided normal Wald test." | |
| ) | |
| if se is not None: | |
| low, high = estimate - _Z95 * se, estimate + _Z95 * se | |
| if se is None or not math.isfinite(se) or se <= 0 or not 0 < se * se < math.inf: | |
| if study.get("p_value") is not None: | |
| result["numeric_notes"].append( | |
| "P-value and sample size alone do not identify an effect or standard error." | |
| ) | |
| return result | |
| if low is None or high is None or not all(math.isfinite(x) for x in (low, high)): | |
| return result | |
| result.update( | |
| analysis_se=se, | |
| analysis_ci=[low, high], | |
| analysis_ci_low=low, | |
| analysis_ci_high=high, | |
| evidence_tier=tier, | |
| mde=_MDE_FACTOR * se, | |
| ) | |
| return result | |
| def assign_bucket( | |
| study: dict, | |
| sesoi: float = 0.2, | |
| effect_type: str = "SMD", | |
| today: date | None = None, | |
| ) -> dict: | |
| """Return classification metadata without changing the input study document. | |
| Text-only null claims never establish equivalence: they are ``reported_null``, | |
| never ``credible_null``. A text-only positive label is likewise a provisional | |
| reported effect, with magnitude explicitly unverified. | |
| Methodological failures override numerical findings. Reviews are discovery | |
| sources, never independent patient samples in a pool. | |
| """ | |
| result = _numeric_evidence(study, effect_type) | |
| result.update( | |
| bucket="inconclusive", rationale="", significant_but_trivial=False, | |
| inconclusive_reason=None, | |
| ) | |
| status = str(study.get("overall_status", "")).upper() | |
| failures = [] | |
| if study.get("is_retracted") is True: | |
| failures.append("Publication is retracted") | |
| if status in {"TERMINATED", "WITHDRAWN", "SUSPENDED"}: | |
| reason = str(study.get("why_stopped") or "reason not provided") | |
| failures.append(f"Registry status {status.lower()}: {reason}") | |
| actual, planned = ( | |
| _number(study.get("enrollment_actual")), | |
| _number(study.get("enrollment_planned")), | |
| ) | |
| if actual is not None and planned is not None and 0 <= actual < planned / 2: | |
| failures.append("Actual enrollment was less than half of planned enrollment") | |
| if study.get("has_control") is False and study.get("is_review") is not True: | |
| failures.append("No control arm was identified") | |
| if failures: | |
| result.update(bucket="failed", rationale="; ".join(failures) + ".") | |
| return result | |
| if study.get("is_review") is True: | |
| result["rationale"] = "Review used for discovery; not an independent primary study." | |
| result["inconclusive_reason"] = "review" | |
| return result | |
| if _unreported(study, today or datetime.now(UTC).date()): | |
| result.update( | |
| bucket="unreported", | |
| rationale="Registered trial completed more than 12 calendar months ago, with no posted " | |
| "results or linked result publication in the available metadata.", | |
| ) | |
| return result | |
| interval = result["analysis_ci"] | |
| if interval is None: | |
| # A quoted statement read from the report outranks the index-time phrase lexicon, | |
| # but a between-group result needs a comparison group: once a paper has been read | |
| # and no control arm was quoted (a case report, a single-arm series), neither its | |
| # stated result nor its wording earns a verdict. | |
| comparative = study.get("has_control") is True or _is_registry(study) | |
| read = _text(study.get("extraction_status")) == "verified" | |
| stated = _text(study.get("reported_result")) if comparative else "" | |
| label = stated or ("" if read and not comparative else _text(study.get("result_label"))) | |
| origin = "The report states" if stated else "Abstract classifier reports" | |
| if label == "positive": | |
| result.update( | |
| bucket="effect", | |
| rationale=f"{origin} an effect; statistical significance and " | |
| "meaningful magnitude are unverified without compatible numerical evidence.", | |
| ) | |
| elif label == "null": | |
| result.update( | |
| bucket="reported_null", | |
| rationale=f"{origin} no significant difference. Without a compatible " | |
| "numerical CI, this cannot establish absence of a meaningful effect.", | |
| ) | |
| elif label == "mixed": | |
| result["inconclusive_reason"] = "mixed_result" | |
| result["rationale"] = ( | |
| f"{origin} conflicting primary results, and no compatible numerical CI " | |
| "is available to weigh them." | |
| ) | |
| else: | |
| result["inconclusive_reason"] = "no_result" | |
| result["rationale"] = ( | |
| "No comparative result could be read: no usable numbers and no stated outcome." | |
| ) | |
| return result | |
| margin = _margin(sesoi, effect_type) | |
| if margin is None: | |
| result["inconclusive_reason"] = "no_threshold" | |
| result["rationale"] = ( | |
| "A positive SESOI on a supported effect scale is required; raw ratio margins must exceed 1." | |
| ) | |
| return result | |
| if result["analysis_effect_type"] != _analysis_type(effect_type): | |
| result["inconclusive_reason"] = "other_scale" | |
| result["rationale"] = ( | |
| f"Study uses {result['analysis_effect_type']} but the SESOI uses " | |
| f"{_analysis_type(effect_type) or effect_type}; meaningfulness cannot be compared across scales." | |
| ) | |
| return result | |
| low, high = interval | |
| excludes_zero = low > 0 or high < 0 | |
| estimate = result["analysis_estimate"] | |
| if low > -margin and high < margin: | |
| result.update( | |
| bucket="credible_null", | |
| significant_but_trivial=excludes_zero, | |
| rationale="The entire 95% CI lies strictly inside the SESOI equivalence bounds." | |
| + ( | |
| " The effect is statistically significant but smaller than the meaningful threshold." | |
| if excludes_zero | |
| else "" | |
| ), | |
| ) | |
| elif excludes_zero and estimate is not None and abs(estimate) >= margin: | |
| result.update( | |
| bucket="effect", | |
| rationale="95% CI excludes the null and the point estimate reaches the SESOI; " | |
| "the CI need not exclude effects smaller than the SESOI.", | |
| ) | |
| else: | |
| result["inconclusive_reason"] = "wide_interval" | |
| result["rationale"] = ( | |
| "The 95% CI does not establish equivalence and the evidence does not establish " | |
| "a statistically significant effect whose estimate reaches the SESOI." | |
| ) | |
| return result | |
| def _unit(study: dict, kind: str) -> str: | |
| unit = _text(study.get("outcome_unit")) | |
| if kind == "SMD" and unit in {"", "sd", "sds", "standard deviation", "standard deviations"}: | |
| return "SD" | |
| return unit or ( | |
| {"logOR": "log odds ratio", "logRR": "log risk ratio", "logHR": "log hazard ratio"}.get( | |
| kind, "" | |
| ) | |
| ) | |
| def _study_id(study: dict, position: int) -> str: | |
| return str(study.get("id") or study.get("study_id") or f"study-{position + 1}") | |
| def _egger(effects: np.ndarray, ses: np.ndarray) -> dict | None: | |
| """Egger's regression test for funnel-plot asymmetry, or None when it cannot be run. | |
| The standard normal deviate ``estimate / se`` is regressed on precision ``1 / se`` | |
| by ordinary least squares; the intercept is the asymmetry, zero under symmetry. | |
| Asymmetry has many causes besides publication bias (heterogeneity, small-study | |
| effects, chance), so the result is reported, never used to adjust an estimate. | |
| """ | |
| k = len(effects) | |
| if k < EGGER_MIN_STUDIES: | |
| return None | |
| precision = 1.0 / ses | |
| deviate = effects / ses | |
| center = float(precision.mean()) | |
| spread = float(np.sum((precision - center) ** 2)) | |
| # Studies of near-identical precision give a funnel with no vertical extent. | |
| if not math.isfinite(spread) or spread <= 0: | |
| return None | |
| slope = float(np.sum((precision - center) * (deviate - deviate.mean())) / spread) | |
| intercept = float(deviate.mean() - slope * center) | |
| degrees = k - 2 | |
| residual_variance = float(np.sum((deviate - intercept - slope * precision) ** 2)) / degrees | |
| se = math.sqrt(residual_variance * (1 / k + center * center / spread)) | |
| if not math.isfinite(intercept) or not math.isfinite(se) or se <= 0: | |
| return None | |
| statistic = intercept / se | |
| critical = float(student_t.ppf(0.975, degrees)) | |
| p_value = float(2 * student_t.sf(abs(statistic), degrees)) | |
| return { | |
| "intercept": intercept, | |
| "se": se, | |
| "ci": [intercept - critical * se, intercept + critical * se], | |
| "t": statistic, | |
| "df": degrees, | |
| "pValue": p_value, | |
| "slope": slope, | |
| "asymmetric": p_value < EGGER_ALPHA, | |
| "alpha": EGGER_ALPHA, | |
| "method": "Egger regression of the standard normal deviate on precision (OLS)", | |
| } | |
| def _pursuit( | |
| classifications: list[dict], pools: list[dict], margin: float | None, kind: str | None | |
| ) -> dict: | |
| """Posterior probability that a real effect exists, from a Beta(1, 1) prior. | |
| Every verdict that answers the yes/no question moves the posterior: ``effect`` is a | |
| success, ``credible_null`` and ``reported_null`` are failures, weighted by how the | |
| numbers were read (``PURSUIT_TIER_WEIGHTS``). Inconclusive, failed and unreported | |
| records carry no information about the answer and leave the prior alone. Studies | |
| are treated as exchangeable, so contradicting results narrow the interval around | |
| one half rather than widening it; the ``conflict`` share and the ``contested`` state | |
| report that explicitly instead of hiding it in a confident-looking 50%. | |
| With a numeric pool, ``pMeaningful`` is the probability under the predictive | |
| Normal(pooled mean, tau² + SE²) that the true effect reaches the SESOI in either | |
| direction, and ``pFavours`` the probability that it is positive as coded. | |
| """ | |
| successes = failures = 0.0 | |
| counted = {"effect": 0, "credible_null": 0, "reported_null": 0} | |
| uninformative = 0 | |
| for verdict in classifications: | |
| weight = PURSUIT_TIER_WEIGHTS.get(verdict["evidence_tier"], 0.5) | |
| bucket = verdict["bucket"] | |
| if bucket == "effect": | |
| successes += weight | |
| elif bucket in {"credible_null", "reported_null"}: | |
| failures += weight | |
| else: | |
| uninformative += 1 | |
| continue | |
| counted[bucket] += 1 | |
| alpha, beta = 1.0 + successes, 1.0 + failures | |
| informative = successes + failures | |
| low, high = (float(value) for value in beta_dist.ppf([0.025, 0.975], alpha, beta)) | |
| conflict = min(successes, failures) / informative if informative else 0.0 | |
| if informative == 0: | |
| state = "unknown" | |
| elif conflict >= PURSUIT_CONFLICT_SHARE and informative >= 3: | |
| state = "contested" | |
| elif low <= 0.5 <= high: | |
| state = "open" | |
| else: | |
| state = "favours_effect" if low > 0.5 else "favours_null" | |
| # Chance the record's current lean is wrong: twice the smaller tail around even odds. | |
| # 1 with no evidence or a perfect split, near 0 once one side dominates. | |
| below = float(beta_dist.cdf(0.5, alpha, beta)) | |
| result = { | |
| "prior": [1.0, 1.0], | |
| "posterior": [alpha, beta], | |
| "pEffect": alpha / (alpha + beta), | |
| "pOpen": 2 * min(below, 1 - below), | |
| "ci": [low, high], | |
| "successes": successes, | |
| "failures": failures, | |
| "counted": counted, | |
| "uninformative": uninformative, | |
| "conflict": conflict, | |
| "state": state, | |
| "pMeaningful": None, | |
| "pFavours": None, | |
| "poolStudyIds": [], | |
| "method": ( | |
| "Beta(1, 1) prior updated with tier-weighted verdicts; 95% equal-tailed credible " | |
| "interval. Pool probabilities integrate Normal(pooled mean, tau² + SE²)." | |
| ), | |
| } | |
| matching = [pool for pool in pools if pool["effectType"] == kind] | |
| if margin is not None and len(matching) == 1: | |
| pool = matching[0] | |
| spread = math.sqrt(pool["tau2"] + pool["se"] ** 2) | |
| if spread > 0: | |
| result.update( | |
| pMeaningful=_NORMAL.cdf((-margin - pool["estimate"]) / spread) | |
| + (1 - _NORMAL.cdf((margin - pool["estimate"]) / spread)), | |
| pFavours=1 - _NORMAL.cdf(-pool["estimate"] / spread), | |
| poolStudyIds=pool["studyIds"], | |
| ) | |
| return result | |
| # Below this chance a new study of the same design is not recommended. | |
| PURSUE_DEPRIORITIZE = 0.25 | |
| # From here the record is open enough to repeat the design as asked. | |
| PURSUE_AS_PLANNED = 0.5 | |
| def _decide( | |
| pursuit: dict, planned_se: float | None, margin: float | None, alpha: float | None, drawer: dict | |
| ) -> None: | |
| """Turn the posterior into the chance a new study should be run, and a recommendation. | |
| ``pPursue = pOpen x power``: the chance the record's lean is wrong times the chance the | |
| planned design would detect the SESOI if it is real (two-sided test at ``alpha`` with | |
| the planned SE). Without a computable design the power factor is left out and said | |
| so. A contested record can score high, but the recommendation is then to change the | |
| design, since repeating it adds to the disagreement rather than resolving it. | |
| """ | |
| power = None | |
| if planned_se is not None and margin is not None and alpha is not None and planned_se > 0: | |
| z = -_NORMAL.inv_cdf(alpha / 2) | |
| power = (1 - _NORMAL.cdf(z - margin / planned_se)) + _NORMAL.cdf(-z - margin / planned_se) | |
| p_pursue = pursuit["pOpen"] * (power if power is not None else 1.0) | |
| state = pursuit["state"] | |
| if state == "contested": | |
| recommendation = "pursue_with_changes" | |
| elif p_pursue >= PURSUE_AS_PLANNED: | |
| recommendation = "pursue" | |
| elif p_pursue >= PURSUE_DEPRIORITIZE: | |
| recommendation = "pursue_with_changes" | |
| else: | |
| recommendation = "deprioritize" | |
| reasons = [] | |
| if state == "unknown": | |
| reasons.append( | |
| "No read study answered the question either way; a new study would be the first answer." | |
| ) | |
| elif state == "contested": | |
| reasons.append( | |
| f"Studies disagree ({pursuit['conflict']:.0%} of the evidence weight on the minority " | |
| "side); a study that separates the conditions under which the effect appears is worth " | |
| "more than another undifferentiated trial." | |
| ) | |
| elif state == "open": | |
| reasons.append( | |
| f"The record leans one way ({pursuit['pEffect']:.0%} chance of a real effect) but the " | |
| "credible interval still spans even odds." | |
| ) | |
| else: | |
| side = "a real effect" if state == "favours_effect" else "no effect" | |
| reasons.append( | |
| f"The record already favors {side}: only a {pursuit['pOpen']:.0%} chance that lean " | |
| "is wrong." | |
| ) | |
| if power is not None: | |
| reasons.append( | |
| f"The planned design has {power:.0%} power to detect the SESOI, so it would answer the " | |
| "question if run." | |
| if power >= 0.8 | |
| else f"The planned design has only {power:.0%} power to detect the SESOI; it would likely " | |
| "end inconclusive, which lowers the chance it is worth running as planned." | |
| ) | |
| else: | |
| reasons.append( | |
| "No planned sample size or SESOI margin was usable, so power is not factored in." | |
| ) | |
| if drawer.get("share") and drawer["share"] >= 0.3: | |
| reasons.append( | |
| f"{drawer['share']:.0%} of eligible completed trials never reported; the record may " | |
| "understate null results, so the lean above may be too optimistic." | |
| ) | |
| pursuit.update(power=power, pPursue=p_pursue, recommendation=recommendation, reasons=reasons) | |
| def _pool(group: list[tuple[dict, dict, str]], key: tuple) -> dict: | |
| effects = np.array([row[1]["analysis_estimate"] for row in group], dtype=float) | |
| ses = np.array([row[1]["analysis_se"] for row in group], dtype=float) | |
| variances = np.square(ses) | |
| # statsmodels DL can emit invalid interim random-effect weights for negative | |
| # tau²; use its estimator, then recompute all weights with tau² clamped at 0. | |
| with python_warnings.catch_warnings(), np.errstate(all="ignore"): | |
| python_warnings.simplefilter("ignore", RuntimeWarning) | |
| combined = combine_effects(effects, variances, method_re="chi2") | |
| tau2 = max(0.0, float(combined.tau2)) | |
| weights = 1 / (variances + tau2) | |
| estimate = float(np.sum(weights * effects) / np.sum(weights)) | |
| se = float(math.sqrt(1 / np.sum(weights))) | |
| kind, outcome, unit, intervention, comparator, direction = key | |
| return { | |
| "effectType": kind, | |
| "outcome": group[0][0].get("outcome") or outcome, | |
| "unit": unit, | |
| "k": len(group), | |
| "estimate": estimate, | |
| "se": se, | |
| "ci": [estimate - _Z95 * se, estimate + _Z95 * se], | |
| "tau2": tau2, | |
| "studyIds": [row[2] for row in group], | |
| "intervention": intervention, | |
| "comparator": comparator, | |
| "direction": direction or "as reported", | |
| "method": "DerSimonian-Laird (nonnegative tau²), normal 95% CI", | |
| "egger": _egger(effects, ses), | |
| } | |
| def _variance_factor(kind: str | None, plan: dict, warnings: list[str]) -> float | None: | |
| if kind == "SMD": | |
| return 4.0 | |
| if kind == "MD": | |
| sd = _number(plan.get("outcomeSd")) | |
| if sd is not None and sd > 0: | |
| return 4 * sd * sd | |
| warnings.append( | |
| "Mean-difference planning needs a positive outcomeSd in the outcome's units." | |
| ) | |
| elif kind in {"logOR", "logRR"}: | |
| risk = _number(plan.get("baselineRisk")) | |
| if risk is not None and 0 < risk < 1: | |
| return 4 / (risk * (1 - risk)) if kind == "logOR" else 4 * (1 - risk) / risk | |
| warnings.append("Binary ratio planning needs baselineRisk strictly between 0 and 1.") | |
| elif kind == "logHR": | |
| warnings.append( | |
| "Hazard-ratio planning needs expected events and follow-up; total N is insufficient." | |
| ) | |
| else: | |
| warnings.append("Planning is unavailable for this effect scale.") | |
| return None | |
| def _assurance(estimate: float, prior_variance: float, se: float, alpha: float) -> float: | |
| """Exact integral of normal-approximation two-sided power over a normal prior.""" | |
| threshold = -_NORMAL.inv_cdf(alpha / 2) * se | |
| predictive_sd = math.sqrt(prior_variance + se * se) | |
| probability = _NORMAL.cdf((-threshold - estimate) / predictive_sd) | |
| probability += _NORMAL.cdf((estimate - threshold) / predictive_sd) | |
| return min(1.0, max(0.0, probability)) | |
| def _required_n(pool: dict, factor: float, alpha: float, target: float = 0.8) -> int | None: | |
| prior_variance = pool["tau2"] + pool["se"] ** 2 | |
| def enough(n: int) -> bool: | |
| return _assurance(pool["estimate"], prior_variance, math.sqrt(factor / n), alpha) >= target | |
| high = 4 | |
| while high < _MAX_REQUIRED_N and not enough(high): | |
| high = min(2 * high, _MAX_REQUIRED_N) | |
| if not enough(high): | |
| return None | |
| # Search numbers of participants per arm, so the returned total N is even. | |
| low_arm, high_arm = 2, high // 2 | |
| while low_arm < high_arm: | |
| middle = (low_arm + high_arm) // 2 | |
| if enough(2 * middle): | |
| high_arm = middle | |
| else: | |
| low_arm = middle + 1 | |
| return 2 * low_arm | |
| def _file_drawer(studies: list[dict], supplied: dict | None) -> dict: | |
| if supplied is not None: | |
| completed, unreported = ( | |
| _number(supplied.get("completed")), | |
| _number(supplied.get("unreported")), | |
| ) | |
| if ( | |
| completed is not None | |
| and unreported is not None | |
| and completed >= 0 | |
| and 0 <= unreported <= completed | |
| and completed.is_integer() | |
| and unreported.is_integer() | |
| ): | |
| result = { | |
| "completed": int(completed), | |
| "unreported": int(unreported), | |
| "share": unreported / completed if completed else None, | |
| } | |
| overdue = _number(supplied.get("overdue")) | |
| if overdue is not None and overdue.is_integer() and 0 <= overdue <= unreported: | |
| result["overdue"] = int(overdue) | |
| return result | |
| completed_studies = [ | |
| study | |
| for study in studies | |
| if _is_registry(study) and str(study.get("overall_status", "")).upper() == "COMPLETED" | |
| ] | |
| overdue = sum(_unreported(study, datetime.now(UTC).date()) for study in completed_studies) | |
| unreported = sum( | |
| _unreported({**study, "primary_completion_date": "1900-01-01"}, date(2000, 1, 1)) | |
| for study in completed_studies | |
| ) | |
| return { | |
| "completed": len(completed_studies), | |
| "unreported": unreported, | |
| "overdue": overdue, | |
| "share": unreported / len(completed_studies) if completed_studies else None, | |
| } | |
| def analyze_studies( | |
| studies: list[dict], | |
| plan: dict, | |
| file_drawer: dict | None = None, | |
| outcome_groups: dict[str, str] | None = None, | |
| ) -> dict: | |
| """Analyze retrieved studies; caller supplies corpus-wide file-drawer counts. | |
| ``outcome_groups`` maps study id to a shared outcome label for studies a language | |
| model judged comparable (same construct, same kind of comparison). It replaces only | |
| the exact-text match on outcome and comparison wording; scale, units for mean | |
| differences and every number are still checked and computed here, and such pools | |
| are marked so the judgment can be reviewed. | |
| Pools require >=3 independent primary studies with the same normalized effect | |
| type, outcome, outcome unit, intervention/comparator orientation and explicit | |
| direction metadata (when supplied). Missing outcomes, and MD units, preclude | |
| pooling. Population differences remain heterogeneity, not hidden conversions. | |
| When multiple pools match the request, no single assurance is selected. | |
| """ | |
| assumptions = [ | |
| "Study CIs are 95% unless a confidence level is supplied; SEs and power use normal approximations.", | |
| "SESOI is in SD units (SMD), outcome units (MD), or natural-log units (logOR/logRR/logHR).", | |
| ( | |
| "Pooling requires at least three independent primary studies with matching outcome, scale, " | |
| "units, comparison and reported direction; semantic equivalence is not inferred." | |
| ), | |
| ( | |
| "Assurance is expected probability of two-sided statistical significance, including harm; " | |
| "it is not the probability of meaningful benefit." | |
| ), | |
| ( | |
| "The predictive true effect is Normal(pooled mean, tau² + pooled SE²); " | |
| "heterogeneity and pooled uncertainty are estimated from retrieved studies." | |
| ), | |
| ( | |
| "Planned N is total enrollment split equally between two independent arms; " | |
| "no allowance is made for attrition, clustering or multiplicity." | |
| ), | |
| ( | |
| "EV uses the supplied values' units; V_null is the value assigned to any nonsignificant " | |
| "result, which may still be inconclusive." | |
| ), | |
| ] | |
| warnings: list[str] = [] | |
| labels: dict[str, str] = {} | |
| groups: dict[tuple, list] = defaultdict(list) | |
| seen: set[str] = set() | |
| unique_studies = [] | |
| classifications: list[dict] = [] | |
| ordered_studies = sorted(studies, key=lambda row: row.get("source") not in {"ctgov", "merged"}) | |
| for position, study in enumerate(ordered_studies): | |
| sid = _study_id(study, position) | |
| aliases = {f"id:{sid}"} | |
| fields = ( | |
| ("nct_ids",) if study.get("source") in {"ctgov", "merged"} else ("nct_ids", "pmids") | |
| ) | |
| for field in fields: | |
| values = study.get(field) or [] | |
| if isinstance(values, str): | |
| values = [values] | |
| aliases.update(f"{field}:{value}" for value in values) | |
| if aliases & seen: | |
| seen.update(aliases) | |
| warnings.append("Duplicate study identifiers were counted once.") | |
| continue | |
| seen.update(aliases) | |
| unique_studies.append(study) | |
| classification = assign_bucket(study, plan.get("sesoi", 0.2), plan.get("effectType", "SMD")) | |
| if study.get("is_review"): | |
| continue | |
| classifications.append(classification) | |
| if classification["bucket"] in {"failed", "unreported"}: | |
| continue | |
| if classification["analysis_se"] is None or classification["analysis_estimate"] is None: | |
| continue | |
| kind = classification["analysis_effect_type"] | |
| outcome = _text(study.get("outcome")) | |
| derived = classification["evidence_tier"] == "derived" | |
| # A derived SMD is in SD units whatever instrument the arms were measured on. | |
| unit = "SD" if derived and kind == "SMD" else _unit(study, kind) | |
| if not outcome or (kind == "MD" and not unit): | |
| warnings.append( | |
| "Numeric studies with unknown outcomes or missing MD units were not pooled." | |
| ) | |
| continue | |
| if classification["evidence_tier"] == "reconstructed": | |
| warnings.append( | |
| "Some pooled evidence uses reconstructed normal-approximation uncertainty." | |
| ) | |
| if derived: | |
| warnings.append( | |
| "Some pooled evidence uses effect sizes computed from reported arm-level summaries." | |
| ) | |
| direction = _text(study.get("effect_direction") or study.get("outcome_direction")) | |
| comparison = (_text(study.get("intervention")), _text(study.get("comparator"))) | |
| label = (outcome_groups or {}).get(sid) | |
| if label: | |
| # Different instruments for one construct share SD units, never raw units. | |
| if kind == "SMD": | |
| unit = "SD" | |
| outcome, comparison = f"group:{_text(label)}", ("", "") | |
| labels[outcome] = label | |
| if not direction: | |
| assumptions.append( | |
| "Within each comparison, effects retain their reported signs; " | |
| "consistent outcome coding is assumed where direction metadata is absent." | |
| ) | |
| key = (kind, outcome, unit, *comparison, direction) | |
| groups[key].append((study, classification, sid)) | |
| pools = [] | |
| for key, group in groups.items(): | |
| name = labels.get(key[1], key[1]) | |
| if len(group) >= 3: | |
| pool = _pool(group, key) | |
| if key[1] in labels: | |
| pool.update(outcome=name, grouping="model") | |
| warnings.append( | |
| f"Studies pooled under “{name}” were judged comparable by a language model " | |
| "from their outcome and comparison wording; check the grouped studies." | |
| ) | |
| pools.append(pool) | |
| else: | |
| warnings.append( | |
| f"Only {len(group)} compatible numeric studies for {name} ({key[0]}); " | |
| "at least 3 are required for pooling." | |
| ) | |
| pools.sort(key=lambda pool: (-pool["k"], pool["outcome"], pool["effectType"])) | |
| if not pools: | |
| warnings.append( | |
| "No eligible meta-analysis pool; assurance and expected value are unavailable." | |
| ) | |
| if any(pool["egger"] for pool in pools): | |
| assumptions.append( | |
| "Egger's test regresses each study's estimate/SE on its precision 1/SE; a nonzero " | |
| "intercept is funnel asymmetry, which publication bias, small-study effects or " | |
| "heterogeneity can all produce. No pooled estimate is adjusted for it." | |
| ) | |
| for pool in pools: | |
| egger = pool["egger"] | |
| if egger is None: | |
| warnings.append( | |
| f"Funnel asymmetry was not tested for {pool['outcome']} ({pool['effectType']}): " | |
| f"Egger's test needs at least {EGGER_MIN_STUDIES} studies of differing precision " | |
| f"and this pool has {pool['k']}." | |
| ) | |
| elif egger["asymmetric"]: | |
| warnings.append( | |
| f"Egger's test indicates funnel asymmetry for {pool['outcome']} " | |
| f"({pool['effectType']}): intercept {egger['intercept']:.2f} " | |
| f"(p = {egger['pValue']:.3f}); small studies report systematically different " | |
| "effects, so the pooled estimate may reflect selective reporting." | |
| ) | |
| drawer = _file_drawer(unique_studies, file_drawer) | |
| if file_drawer is None: | |
| warnings.append( | |
| "File-drawer counts cover supplied studies only, not the full search corpus." | |
| ) | |
| assumptions.append( | |
| "File drawer counts completed registered trials with no posted or linked results " | |
| "at any age, divided by all completed registered trials; overdue separately counts " | |
| "those more than 12 calendar months after completion. Registry coverage is incomplete " | |
| "and does not prove nonpublication." | |
| ) | |
| if drawer["unreported"]: | |
| warnings.append( | |
| "Unreported completed trials are absent from the numeric pool; publication bias " | |
| "may affect pooled estimates and assurance." | |
| ) | |
| pursuit = _pursuit( | |
| classifications, | |
| pools, | |
| _margin(plan.get("sesoi", 0.2), plan.get("effectType", "SMD")), | |
| _analysis_type(plan.get("effectType", "SMD")), | |
| ) | |
| if pursuit["state"] == "contested": | |
| warnings.append( | |
| f"The record is contested: {pursuit['conflict']:.0%} of the informative evidence " | |
| "weight sits on the minority side. A posterior near one half here means the studies " | |
| "disagree, not that the question is half-settled; look for moderators before pooling." | |
| ) | |
| assumptions.append( | |
| "The chance a real effect exists starts from a Beta(1, 1) prior (50/50) and counts each " | |
| "effect verdict as a success and each credible or reported null as a failure, weighted " | |
| "1 for quoted numbers, 0.75 for reconstructed uncertainty and 0.5 for text-only claims. " | |
| "Inconclusive, stopped and unreported records do not move it. Studies are assumed " | |
| "exchangeable, so it is a summary of the record, not a probability of clinical benefit." | |
| ) | |
| result = { | |
| "pools": pools, | |
| "pursuit": pursuit, | |
| "assurance": None, | |
| "requiredN": None, | |
| "expectedValue": None, | |
| "plannedMde": None, | |
| "plannedNullBucket": None, | |
| "plannedNullCi": None, | |
| "fileDrawer": drawer, | |
| "assumptions": assumptions, | |
| "warnings": warnings, | |
| "assuranceDefinition": "Probability of two-sided statistical significance (either direction)", | |
| "requiredNTarget": 0.8, | |
| "selectedPoolStudyIds": [], | |
| } | |
| kind = _analysis_type(plan.get("effectType", "SMD")) | |
| factor = _variance_factor(kind, plan, warnings) | |
| alpha = _number(plan.get("alpha", 0.05)) | |
| n = _number(plan.get("plannedN", 200)) | |
| if alpha is None or not 0 < alpha < 1 or alpha / 2 == 0: | |
| warnings.append("Alpha must lie strictly between 0 and 1.") | |
| factor = None | |
| if n is None or n < 4 or not n.is_integer(): | |
| warnings.append("Planned N must be an integer total sample size of at least 4.") | |
| factor = None | |
| if factor is not None and (not math.isfinite(factor) or factor <= 0): | |
| warnings.append("Design variance is outside usable numerical precision.") | |
| factor = None | |
| planned_se = None | |
| if factor is not None: | |
| if kind == "SMD": | |
| assumptions.append( | |
| "SMD planning uses SE = sqrt(1/n_intervention + 1/n_control), " | |
| "assuming known common outcome SD; this is a large-sample approximation." | |
| ) | |
| elif kind == "MD": | |
| assumptions.append( | |
| "MD planning assumes the supplied outcomeSd is known and common to both arms." | |
| ) | |
| elif kind in {"logOR", "logRR"}: | |
| assumptions.append( | |
| "Binary ratio planning uses a fixed delta-method variance at the supplied " | |
| "baseline risk in both arms; it is a local approximation near no effect " | |
| "and may be inaccurate for rare events or large effects." | |
| ) | |
| # For odd total N, explicitly use the actual nearly equal arm sizes. | |
| arm1 = int(n) // 2 | |
| arm2 = int(n) - arm1 | |
| planned_se = math.sqrt(factor / 4 * (1 / arm1 + 1 / arm2)) | |
| z = -_NORMAL.inv_cdf(alpha / 2) | |
| result["plannedMde"] = (z + _NORMAL.inv_cdf(0.8)) * planned_se | |
| margin = _margin(plan.get("sesoi", 0.2), plan.get("effectType", "SMD")) | |
| if margin is not None: | |
| # Equivalence classification always uses 95% CIs, regardless of test alpha. | |
| half_width = _Z95 * planned_se | |
| result["plannedNullCi"] = [-half_width, half_width] | |
| result["plannedNullBucket"] = "credible_null" if half_width < margin else "inconclusive" | |
| candidates = [pool for pool in pools if pool["effectType"] == kind] | |
| if plan.get("outcome"): | |
| candidates = [ | |
| pool for pool in candidates if _text(pool["outcome"]) == _text(plan["outcome"]) | |
| ] | |
| if plan.get("outcomeUnit"): | |
| target_unit = _unit({"outcome_unit": plan["outcomeUnit"]}, kind) | |
| candidates = [pool for pool in candidates if pool["unit"] == target_unit] | |
| if len(candidates) == 1: | |
| pool = candidates[0] | |
| assurance = _assurance( | |
| pool["estimate"], pool["tau2"] + pool["se"] ** 2, planned_se, alpha | |
| ) | |
| result.update( | |
| assurance=assurance, | |
| requiredN=_required_n(pool, factor, alpha), | |
| selectedPoolStudyIds=pool["studyIds"], | |
| ) | |
| if result["requiredN"] is None: | |
| warnings.append( | |
| f"80% assurance was not reached by the search cap of {_MAX_REQUIRED_N:,} total participants." | |
| ) | |
| values = [ | |
| _number(plan.get(key, default)) | |
| for key, default in (("valueSuccess", 100), ("valueNull", 20), ("studyCost", 30)) | |
| ] | |
| if all(value is not None for value in values): | |
| result["expectedValue"] = ( | |
| assurance * values[0] + (1 - assurance) * values[1] - values[2] | |
| ) | |
| else: | |
| warnings.append( | |
| "Expected value needs finite valueSuccess, valueNull and studyCost." | |
| ) | |
| elif len(candidates) > 1: | |
| warnings.append( | |
| "Multiple compatible pools match the plan; specify one outcome/unit/comparison " | |
| "before interpreting a single assurance or EV." | |
| ) | |
| elif pools: | |
| warnings.append("No pooled outcome and effect scale match the planned study.") | |
| _decide( | |
| pursuit, | |
| planned_se, | |
| _margin(plan.get("sesoi", 0.2), plan.get("effectType", "SMD")), | |
| alpha if factor is not None else None, | |
| drawer, | |
| ) | |
| assumptions.append( | |
| "The chance you should pursue the study is the chance the record's lean is wrong " | |
| "(twice the smaller posterior tail around even odds) times the planned design's power " | |
| "to detect the SESOI. It weighs whether a new study would change the answer, not the " | |
| "value of the answer; the expected value below carries the supplied values." | |
| ) | |
| result["assumptions"] = list(dict.fromkeys(assumptions)) | |
| result["warnings"] = list(dict.fromkeys(warnings)) | |
| return result | |