Skills › Scientific Agent Skills › peer-review › audit_statistics_reproducibility.py
audit_statistics_reproducibility.py
305 lines · skills/peer-review/scripts/audit_statistics_reproducibility.py at 5f2bc03
1#!/usr/bin/env python3
2"""Audit a structured statistics and reproducibility checklist locally."""
4from __future__ import annotations
6import argparse
7from collections import Counter
8from typing import Any
10from _common import (
11 ValidationError,
12 error_exit,
13 issue,
14 read_json,
15 require_bool,
16 require_enum,
17 require_exact_keys,
18 require_identifier,
19 require_identifier_list,
20 require_list,
21 require_object,
22 require_text,
23 require_text_list,
24 require_unique,
25 write_json_report,
26)
28CORE_ITEM_IDS = (
29 "question.estimand_alignment",
30 "design.unit_and_independence",
31 "design.sample_size_precision",
32 "design.allocation_randomization",
33 "design.blinding",
34 "data.inclusion_exclusion",
35 "data.missing_data",
36 "data.outliers_transformations",
37 "analysis.prespecification",
38 "analysis.method_design_alignment",
39 "analysis.assumptions_diagnostics",
40 "analysis.multiplicity",
41 "analysis.clustering_repeated_measures",
42 "results.effect_sizes_uncertainty",
43 "results.denominators_flow",
44 "results.complete_outcomes_harms",
45 "reproducibility.data_materials_access",
46 "reproducibility.code_environment_parameters",
47 "reproducibility.provenance_versions",
48 "ethics.approval_consent_governance",
49 "interpretation.claim_evidence_causality",
50 "integrity.deviations_selective_reporting",
51)
52CATEGORIES = {
53 "question",
54 "design",
55 "data",
56 "analysis",
57 "results",
58 "reproducibility",
59 "ethics",
60 "interpretation",
61 "integrity",
62}
63APPLICABILITY = {"applicable", "not_applicable"}
64STATUSES = {
65 "verified_present",
66 "partly_documented",
67 "missing",
68 "not_assessed",
69 "not_applicable",
70}
71SPECIALIST_TRIGGER_IDS = {
72 "question.estimand_alignment",
73 "data.missing_data",
74 "analysis.method_design_alignment",
75 "analysis.assumptions_diagnostics",
76 "analysis.multiplicity",
77 "analysis.clustering_repeated_measures",
78 "results.effect_sizes_uncertainty",
79 "interpretation.claim_evidence_causality",
80}
83def load_checklist(payload: Any) -> dict[str, Any]:
84 root = require_object(payload, "checklist")
85 require_exact_keys(
86 root,
87 required={
88 "schema_version",
89 "checklist_id",
90 "study_design",
91 "specialist_review",
92 "items",
93 },
94 context="checklist",
95 )
96 schema_version = require_enum(
97 root["schema_version"], {"2.0"}, "checklist.schema_version"
98 )
99 checklist_id = require_identifier(root["checklist_id"], "checklist.checklist_id")
100 study_design = require_identifier(root["study_design"], "checklist.study_design")
102 specialist = require_object(
103 root["specialist_review"], "checklist.specialist_review"
104 )
105 require_exact_keys(
106 specialist,
107 required={"needed", "areas", "requested"},
108 context="checklist.specialist_review",
109 )
110 specialist_review = {
111 "needed": require_enum(
112 specialist["needed"],
113 {"yes", "no", "undetermined"},
114 "checklist.specialist_review.needed",
115 ),
116 "areas": require_identifier_list(
117 specialist["areas"], "checklist.specialist_review.areas"
118 ),
119 "requested": require_bool(
120 specialist["requested"], "checklist.specialist_review.requested"
121 ),
122 }
124 raw_items = require_list(root["items"], "checklist.items", minimum=1, maximum=200)
125 items: list[dict[str, Any]] = []
126 item_ids: list[str] = []
127 for index, raw_item in enumerate(raw_items):
128 context = f"checklist.items[{index}]"
129 item = require_object(raw_item, context)
130 require_exact_keys(
131 item,
132 required={
133 "id",
134 "category",
135 "applicability",
136 "status",
137 "evidence_locations",
138 "note",
139 "requested_action",
140 },
141 context=context,
142 )
143 item_id = require_identifier(item["id"], f"{context}.id")
144 item_ids.append(item_id)
145 items.append(
146 {
147 "id": item_id,
148 "category": require_enum(
149 item["category"], CATEGORIES, f"{context}.category"
150 ),
151 "applicability": require_enum(
152 item["applicability"], APPLICABILITY, f"{context}.applicability"
153 ),
154 "status": require_enum(
155 item["status"], STATUSES, f"{context}.status"
156 ),
157 "evidence_locations": require_text_list(
158 item["evidence_locations"],
159 f"{context}.evidence_locations",
160 maximum=50,
161 ),
162 "note": require_text(
163 item["note"],
164 f"{context}.note",
165 allow_empty=True,
166 maximum=2_000,
167 ),
168 "requested_action": require_text(
169 item["requested_action"],
170 f"{context}.requested_action",
171 allow_empty=True,
172 maximum=2_000,
173 ),
174 }
175 )
176 require_unique(item_ids, "checklist.items")
177 return {
178 "schema_version": schema_version,
179 "checklist_id": checklist_id,
180 "study_design": study_design,
181 "specialist_review": specialist_review,
182 "items": items,
183 }
186def audit(checklist: dict[str, Any]) -> dict[str, Any]:
187 errors: list[dict[str, str]] = []
188 warnings: list[dict[str, str]] = []
189 items_by_id = {item["id"]: item for item in checklist["items"]}
190 missing_core = sorted(set(CORE_ITEM_IDS) - set(items_by_id))
191 if missing_core:
192 errors.extend(issue("CORE_ITEM_MISSING", item_id) for item_id in missing_core)
194 status_counts = Counter()
195 gaps_by_category: dict[str, list[str]] = {}
196 action_missing: list[str] = []
197 specialist_triggers: list[str] = []
199 for item in checklist["items"]:
200 item_id = item["id"]
201 status = item["status"]
202 status_counts[status] += 1
203 if item["applicability"] == "not_applicable":
204 if status != "not_applicable":
205 errors.append(issue("APPLICABILITY_STATUS_MISMATCH", item_id))
206 if not item["note"]:
207 errors.append(issue("NOT_APPLICABLE_RATIONALE_REQUIRED", item_id))
208 continue
209 if status == "not_applicable":
210 errors.append(issue("APPLICABILITY_STATUS_MISMATCH", item_id))
211 continue
212 if status == "verified_present" and not item["evidence_locations"]:
213 errors.append(issue("EVIDENCE_LOCATION_REQUIRED", item_id))
214 if status == "partly_documented":
215 if not item["evidence_locations"]:
216 errors.append(issue("PARTIAL_ITEM_NEEDS_EVIDENCE_LOCATION", item_id))
217 if not item["requested_action"]:
218 errors.append(issue("PARTIAL_ITEM_NEEDS_REQUESTED_ACTION", item_id))
219 if status in {"partly_documented", "missing", "not_assessed"}:
220 gaps_by_category.setdefault(item["category"], []).append(item_id)
221 if not item["requested_action"]:
222 action_missing.append(item_id)
223 warnings.append(issue("REQUESTED_ACTION_MISSING", item_id))
224 if item_id in SPECIALIST_TRIGGER_IDS:
225 specialist_triggers.append(item_id)
227 specialist = checklist["specialist_review"]
228 specialist_recommended = bool(specialist_triggers)
229 if specialist["needed"] == "yes" and not specialist["requested"]:
230 warnings.append(
231 issue("SPECIALIST_REVIEW_NOT_REQUESTED", "specialist_review.requested")
232 )
233 if specialist["needed"] == "undetermined":
234 warnings.append(
235 issue("SPECIALIST_REVIEW_UNDETERMINED", "specialist_review.needed")
236 )
237 if specialist_recommended and specialist["needed"] == "no":
238 warnings.append(
239 issue("SPECIALIST_REVIEW_DECLARATION_RECHECK", "specialist_review.needed")
240 )
242 return {
243 "schema_version": "2.0",
244 "checklist_id": checklist["checklist_id"],
245 "valid": not errors,
246 "status": (
247 "INVALID_CHECKLIST"
248 if errors
249 else "VALID_WITH_REVIEW_GAPS"
250 if gaps_by_category
251 else "VALID_NO_RECORDED_GAPS"
252 ),
253 "errors": errors,
254 "warnings": warnings,
255 "study_design": checklist["study_design"],
256 "item_count": len(checklist["items"]),
257 "status_counts": dict(sorted(status_counts.items())),
258 "gap_item_ids_by_category": {
259 category: sorted(item_ids)
260 for category, item_ids in sorted(gaps_by_category.items())
261 },
262 "item_ids_missing_requested_action": sorted(action_missing),
263 "specialist_review": {
264 "declared_needed": specialist["needed"],
265 "declared_areas": specialist["areas"],
266 "declared_requested": specialist["requested"],
267 "trigger_item_ids": sorted(specialist_triggers),
268 "recheck_recommended": specialist_recommended,
269 },
270 "notice": (
271 "This is a structured completeness and consistency audit, not a "
272 "statistical reanalysis, reproducibility claim, quality score, or "
273 "publication recommendation. A qualified specialist must evaluate "
274 "methods outside the reviewer's competence."
275 ),
276 }
279def build_parser() -> argparse.ArgumentParser:
280 parser = argparse.ArgumentParser(
281 description=(
282 "Audit a local statistics/reproducibility checklist and emit only "
283 "item identifiers and counts."
284 )
285 )
286 parser.add_argument("checklist", help="Local checklist JSON")
287 parser.add_argument("-o", "--output", help="Optional local JSON report")
288 parser.add_argument(
289 "--force", action="store_true", help="Replace an existing output file"
290 )
291 return parser
294def main() -> int:
295 args = build_parser().parse_args()
296 try:
297 report = audit(load_checklist(read_json(args.checklist)))
298 write_json_report(report, args.output, force=args.force)
299 return 0 if report["valid"] else 1
300 except ValidationError as exc:
301 return error_exit(exc)
304if __name__ == "__main__":
305 raise SystemExit(main())