Skills › Scientific Agent Skills › hypothesis-generation › generate_preregistration_scaffold.py
generate_preregistration_scaffold.py
383 lines · skills/hypothesis-generation/scripts/generate_preregistration_scaffold.py at 5f2bc03
1#!/usr/bin/env python3
2"""Generate a deterministic local Markdown preregistration scaffold."""
4from __future__ import annotations
6import argparse
7import html
8import re
9from pathlib import Path
10from typing import Any
12from _common import (
13 ValidationError,
14 error_exit,
15 read_json,
16 read_markdown,
17 write_markdown,
18)
19from validate_hypothesis_schema import load_hypothesis_record, validate_record
21TEMPLATE_PATH = (
22 Path(__file__).resolve().parents[1]
23 / "assets"
24 / "preregistration_scaffold_template.md"
25)
26PLACEHOLDER_RE = re.compile(r"\{\{[A-Z0-9_]+\}\}")
27MARKDOWN_SPECIAL_RE = re.compile(r"([\\`*_[\]{}#+|])")
30def _safe_text(value: Any) -> str:
31 """Render bounded input as inert one-line Markdown text."""
32 text = " ".join(str(value).split())
33 text = html.escape(text, quote=False)
34 return MARKDOWN_SPECIAL_RE.sub(r"\\\1", text)
37def _bullet(label: str, value: Any) -> str:
38 return f"- **{_safe_text(label)}:** {_safe_text(value)}"
41def _render_ethics(record: dict[str, Any]) -> str:
42 ethics = record["ethics_and_feasibility"]
43 lines = [
44 _bullet(field.replace("_", " "), ethics[field])
45 for field in (
46 "human_subjects_gate",
47 "animal_research_gate",
48 "biosafety_gate",
49 "dual_use_gate",
50 "regulatory_gate",
51 "data_governance_gate",
52 "feasibility_status",
53 )
54 ]
55 reviews = ethics["required_reviews"] or ["None declared"]
56 blocks = ethics["unresolved_blocks"] or ["None declared"]
57 lines.append(_bullet("required reviews", "; ".join(reviews)))
58 lines.append(_bullet("unresolved blocks", "; ".join(blocks)))
59 return "\n".join(lines)
62def _render_evidence(record: dict[str, Any]) -> str:
63 evidence = record["evidence"]
64 lines = [
65 _bullet("search boundary ID", evidence["search_boundary_id"]),
66 _bullet("evidence ledger path", evidence["ledger_path"]),
67 _bullet("declared source IDs", "; ".join(evidence["source_ids"])),
68 _bullet("evidence limitations", "; ".join(evidence["evidence_limitations"])),
69 ]
70 return "\n".join(lines)
73def _render_observation(record: dict[str, Any]) -> str:
74 observation = record["observation"]
75 return "\n".join(
76 [
77 _bullet("statement", observation["statement"]),
78 _bullet("provenance", observation["provenance"]),
79 _bullet("source IDs", "; ".join(observation["source_ids"])),
80 _bullet("uncertainties", "; ".join(observation["uncertainties"])),
81 ]
82 )
85def _render_question(record: dict[str, Any]) -> str:
86 question = record["research_question"]
87 return "\n".join(
88 _bullet(field.replace("_", " "), question[field])
89 for field in (
90 "statement",
91 "framework",
92 "question_type",
93 "population_or_system",
94 "intervention_or_exposure",
95 "comparator",
96 "outcome",
97 "timeframe",
98 )
99 )
102def _render_hypotheses(record: dict[str, Any]) -> str:
103 sections: list[str] = []
104 for hypothesis in record["hypotheses"]:
105 sections.extend(
106 [
107 f"### {_safe_text(hypothesis['hypothesis_id'])}",
108 "",
109 _bullet("status", hypothesis["status"]),
110 _bullet("candidate statement", hypothesis["statement"]),
111 _bullet("proposed mechanism", hypothesis["mechanism"]),
112 _bullet("rival IDs", "; ".join(hypothesis["rival_hypothesis_ids"])),
113 _bullet("assumptions", "; ".join(hypothesis["assumptions"])),
114 _bullet(
115 "boundary conditions", "; ".join(hypothesis["boundary_conditions"])
116 ),
117 _bullet("uncertainties", "; ".join(hypothesis["uncertainties"])),
118 _bullet("source IDs", "; ".join(hypothesis["source_ids"])),
119 "",
120 ]
121 )
122 return "\n".join(sections).rstrip()
125def _render_estimands(record: dict[str, Any]) -> str:
126 if not record["causal_estimands"]:
127 return "- Not applicable to the declared non-causal question."
128 sections: list[str] = []
129 for estimand in record["causal_estimands"]:
130 sections.extend(
131 [
132 f"### {_safe_text(estimand['estimand_id'])}",
133 "",
134 _bullet(
135 "linked hypothesis IDs",
136 "; ".join(estimand["linked_hypothesis_ids"]),
137 ),
138 _bullet("population", estimand["population"]),
139 _bullet(
140 "intervention or exposure", estimand["intervention_or_exposure"]
141 ),
142 _bullet("comparator", estimand["comparator"]),
143 _bullet("outcome", estimand["outcome"]),
144 _bullet("time horizon", estimand["time_horizon"]),
145 _bullet("population summary", estimand["population_summary"]),
146 _bullet(
147 "intercurrent-event strategy",
148 estimand["intercurrent_event_strategy"],
149 ),
150 _bullet(
151 "identification assumptions",
152 "; ".join(estimand["identification_assumptions"]),
153 ),
154 "",
155 ]
156 )
157 return "\n".join(sections).rstrip()
160def _render_predictions(record: dict[str, Any]) -> str:
161 sections: list[str] = []
162 for prediction in record["predictions"]:
163 sections.extend(
164 [
165 f"### {_safe_text(prediction['prediction_id'])} "
166 f"({_safe_text(prediction['hypothesis_id'])})",
167 "",
168 _bullet("statement", prediction["statement"]),
169 _bullet("conditions", prediction["conditions"]),
170 _bullet("observable", prediction["observable"]),
171 _bullet("expected pattern", prediction["expected_pattern"]),
172 _bullet("falsifier", prediction["falsifier"]),
173 _bullet(
174 "rival hypothesis IDs",
175 "; ".join(prediction["rival_hypothesis_ids"]),
176 ),
177 _bullet("measurement IDs", "; ".join(prediction["measurement_ids"])),
178 _bullet("analysis IDs", "; ".join(prediction["analysis_ids"])),
179 "",
180 ]
181 )
182 return "\n".join(sections).rstrip()
185def _render_nulls_and_controls(record: dict[str, Any]) -> str:
186 sections = ["### Null hypotheses", ""]
187 for null in record["null_hypotheses"]:
188 sections.extend(
189 [
190 _bullet(
191 null["null_id"],
192 f"{null['statement']} | Rule: "
193 f"{null['rejection_or_compatibility_rule']}",
194 ),
195 "",
196 ]
197 )
198 sections.extend(["### Negative controls", ""])
199 for control in record["negative_controls"]:
200 sections.extend(
201 [
202 _bullet(
203 control["control_id"],
204 f"{control['control_type']} | Rationale: {control['rationale']} "
205 f"| Expected: {control['expected_result']} | Failure: "
206 f"{control['failure_implication']}",
207 ),
208 "",
209 ]
210 )
211 return "\n".join(sections).rstrip()
214def _render_operationalizations(record: dict[str, Any]) -> str:
215 sections: list[str] = []
216 for measurement in record["operationalizations"]:
217 sections.extend(
218 [
219 f"### {_safe_text(measurement['measurement_id'])}",
220 "",
221 _bullet("construct", measurement["construct"]),
222 _bullet("variable and role", f"{measurement['variable']} / {measurement['role']}"),
223 _bullet(
224 "operational definition", measurement["operational_definition"]
225 ),
226 _bullet(
227 "instrument or method", measurement["instrument_or_method"]
228 ),
229 _bullet("unit", measurement["unit"]),
230 _bullet("timing", measurement["timing"]),
231 _bullet("population or system", measurement["population_or_system"]),
232 _bullet(
233 "validity evidence source IDs",
234 "; ".join(measurement["validity_evidence_source_ids"]),
235 ),
236 _bullet("reliability plan", measurement["reliability_plan"]),
237 _bullet("missingness plan", measurement["missingness_plan"]),
238 _bullet(
239 "blinding or masking", measurement["blinding_or_masking"]
240 ),
241 _bullet("threshold rationale", measurement["threshold_rationale"]),
242 "",
243 ]
244 )
245 return "\n".join(sections).rstrip()
248def _render_analyses(record: dict[str, Any]) -> str:
249 sections: list[str] = []
250 for analysis in record["analysis_plan"]["analyses"]:
251 sections.extend(
252 [
253 f"### {_safe_text(analysis['analysis_id'])}",
254 "",
255 _bullet("mode", analysis["exploratory_or_confirmatory"]),
256 _bullet("prediction IDs", "; ".join(analysis["prediction_ids"])),
257 _bullet("estimand IDs", "; ".join(analysis["estimand_ids"]) or "None"),
258 _bullet("analysis population", analysis["analysis_population"]),
259 _bullet("method", analysis["method"]),
260 _bullet(
261 "effect or summary measure",
262 analysis["effect_or_summary_measure"],
263 ),
264 _bullet("uncertainty method", analysis["uncertainty_method"]),
265 _bullet("missing-data plan", analysis["missing_data_plan"]),
266 _bullet("multiplicity plan", analysis["multiplicity_plan"]),
267 _bullet(
268 "sensitivity analyses",
269 "; ".join(analysis["sensitivity_analyses"]),
270 ),
271 _bullet("decision rule", analysis["decision_rule"]),
272 "",
273 ]
274 )
275 return "\n".join(sections).rstrip()
278def _render_ai_use(record: dict[str, Any]) -> str:
279 ai_use = record["ai_use"]
280 return "\n".join(
281 [
282 _bullet("AI used", str(ai_use["used"]).lower()),
283 _bullet(
284 "sensitive or unpublished data sent externally",
285 str(
286 ai_use["sensitive_or_unpublished_data_sent_externally"]
287 ).lower(),
288 ),
289 _bullet(
290 "local policy checked", str(ai_use["local_policy_checked"]).lower()
291 ),
292 _bullet(
293 "citation verification required",
294 str(ai_use["citation_verification_required"]).lower(),
295 ),
296 _bullet(
297 "human accountable", str(ai_use["human_accountable"]).lower()
298 ),
299 _bullet("diversity mitigation", ai_use["diversity_mitigation"]),
300 ]
301 )
304def generate(
305 record: dict[str, Any], template_path: Path = TEMPLATE_PATH
306) -> str:
307 """Render all candidates into a local draft without ranking or selection."""
308 validation = validate_record(record)
309 if not validation["valid"]:
310 codes = sorted({item["code"] for item in validation["errors"]})
311 raise ValidationError(
312 "hypothesis record is invalid; resolve these controls first: "
313 + ", ".join(codes)
314 )
315 if validation["unresolved_gate_fields"]:
316 raise ValidationError(
317 "hypothesis record has unresolved safety or feasibility gates: "
318 + ", ".join(validation["unresolved_gate_fields"])
319 )
321 template = read_markdown(template_path)
322 replacements = {
323 "{{PROJECT_ID}}": _safe_text(record["project_id"]),
324 "{{HUMAN_OWNER}}": _safe_text(record["human_owner"]),
325 "{{RECORD_STATUS}}": _safe_text(record["status"]),
326 "{{UPDATED_ON}}": _safe_text(record["updated_on"]),
327 "{{ETHICS_AND_FEASIBILITY}}": _render_ethics(record),
328 "{{EVIDENCE_BOUNDARY}}": _render_evidence(record),
329 "{{OBSERVATION}}": _render_observation(record),
330 "{{RESEARCH_QUESTION}}": _render_question(record),
331 "{{HYPOTHESES}}": _render_hypotheses(record),
332 "{{ESTIMANDS}}": _render_estimands(record),
333 "{{PREDICTIONS}}": _render_predictions(record),
334 "{{NULLS_AND_CONTROLS}}": _render_nulls_and_controls(record),
335 "{{OPERATIONALIZATIONS}}": _render_operationalizations(record),
336 "{{ANALYSES}}": _render_analyses(record),
337 "{{AI_USE}}": _render_ai_use(record),
338 "{{DEVIATION_PLAN}}": _safe_text(
339 record["analysis_plan"]["deviation_reporting"]
340 ),
341 }
342 rendered = template
343 for placeholder, value in replacements.items():
344 rendered = rendered.replace(placeholder, value)
345 unresolved = sorted(set(PLACEHOLDER_RE.findall(rendered)))
346 if unresolved:
347 raise ValidationError(
348 "scaffold template contains unresolved placeholders: "
349 + ", ".join(unresolved)
350 )
351 return rendered.rstrip() + "\n"
354def build_parser() -> argparse.ArgumentParser:
355 parser = argparse.ArgumentParser(
356 description=(
357 "Generate a private local Markdown preregistration scaffold from a "
358 "valid hypothesis record; no upload or registration occurs."
359 )
360 )
361 parser.add_argument("record", help="Local hypothesis record JSON")
362 parser.add_argument("-o", "--output", required=True, help="Output Markdown path")
363 parser.add_argument(
364 "--force", action="store_true", help="Replace an existing output file"
365 )
366 return parser
369def main() -> int:
370 args = build_parser().parse_args()
371 try:
372 record = load_hypothesis_record(read_json(args.record))
373 destination = write_markdown(
374 generate(record), args.output, force=args.force
375 )
376 print(f"Created local preregistration scaffold: {destination}")
377 return 0
378 except ValidationError as exc:
379 return error_exit(exc)
382if __name__ == "__main__":
383 raise SystemExit(main())