Skills › Scientific Agent Skills › hypothesis-generation › validate_prediction_matrix.py
validate_prediction_matrix.py
286 lines · skills/hypothesis-generation/scripts/validate_prediction_matrix.py at 5f2bc03
1#!/usr/bin/env python3
2"""Validate a prediction/rival-hypothesis CSV without ranking candidates."""
4from __future__ import annotations
6import argparse
7import re
8from collections import Counter
9from typing import Any
11from _common import (
12 ValidationError,
13 error_exit,
14 issue,
15 read_csv_records,
16 read_json,
17 require_identifier,
18 require_text,
19 require_unique,
20 split_identifiers,
21 write_json_report,
22)
23from validate_hypothesis_schema import load_hypothesis_record, validate_record
25FIELDS = (
26 "prediction_id",
27 "hypothesis_id",
28 "rival_hypothesis_ids",
29 "conditions",
30 "observable",
31 "expected_if_focal",
32 "expected_if_rivals",
33 "falsifier",
34 "indeterminate_result",
35 "boundary_conditions",
36 "measurement_ids",
37 "negative_control_ids",
38 "analysis_ids",
39 "uncertainty",
40)
43def load_matrix(raw_path: str) -> list[dict[str, Any]]:
44 rows = read_csv_records(raw_path, fields=FIELDS)
45 parsed: list[dict[str, Any]] = []
46 prediction_ids: list[str] = []
47 for line_number, row in enumerate(rows, start=2):
48 context = f"matrix row {line_number}"
49 prediction_id = require_identifier(
50 row["prediction_id"], f"{context}.prediction_id"
51 )
52 prediction_ids.append(prediction_id)
53 parsed.append(
54 {
55 "prediction_id": prediction_id,
56 "hypothesis_id": require_identifier(
57 row["hypothesis_id"], f"{context}.hypothesis_id"
58 ),
59 "rival_hypothesis_ids": split_identifiers(
60 row["rival_hypothesis_ids"],
61 f"{context}.rival_hypothesis_ids",
62 ),
63 "conditions": require_text(
64 row["conditions"], f"{context}.conditions", minimum=5
65 ),
66 "observable": require_text(
67 row["observable"], f"{context}.observable", minimum=5
68 ),
69 "expected_if_focal": require_text(
70 row["expected_if_focal"],
71 f"{context}.expected_if_focal",
72 minimum=5,
73 ),
74 "expected_if_rivals": require_text(
75 row["expected_if_rivals"],
76 f"{context}.expected_if_rivals",
77 minimum=5,
78 ),
79 "falsifier": require_text(
80 row["falsifier"], f"{context}.falsifier", minimum=10
81 ),
82 "indeterminate_result": require_text(
83 row["indeterminate_result"],
84 f"{context}.indeterminate_result",
85 minimum=10,
86 ),
87 "boundary_conditions": require_text(
88 row["boundary_conditions"],
89 f"{context}.boundary_conditions",
90 minimum=5,
91 ),
92 "measurement_ids": split_identifiers(
93 row["measurement_ids"], f"{context}.measurement_ids"
94 ),
95 "negative_control_ids": split_identifiers(
96 row["negative_control_ids"],
97 f"{context}.negative_control_ids",
98 ),
99 "analysis_ids": split_identifiers(
100 row["analysis_ids"], f"{context}.analysis_ids"
101 ),
102 "uncertainty": require_text(
103 row["uncertainty"], f"{context}.uncertainty", minimum=10
104 ),
105 }
106 )
107 require_unique(prediction_ids, "prediction matrix")
108 return parsed
111def _normalized_expectation(value: str) -> str:
112 return re.sub(r"[^a-z0-9]+", " ", value.casefold()).strip()
115def _unknown(
116 values: list[str], allowed: set[str], code: str, field: str
117) -> list[dict[str, str]]:
118 return [
119 issue(code, f"{field}:{value}")
120 for value in sorted(set(values) - allowed)
121 ]
124def validate_matrix(
125 rows: list[dict[str, Any]], record: dict[str, Any] | None = None
126) -> dict[str, Any]:
127 errors: list[dict[str, str]] = []
128 warnings: list[dict[str, str]] = []
129 hypothesis_counts: Counter[str] = Counter()
131 record_ids: dict[str, set[str]] | None = None
132 if record is not None:
133 record_report = validate_record(record)
134 if not record_report["valid"]:
135 raise ValidationError(
136 "optional hypothesis record must pass schema validation first"
137 )
138 record_ids = {
139 "hypotheses": {
140 item["hypothesis_id"] for item in record["hypotheses"]
141 },
142 "predictions": {
143 item["prediction_id"] for item in record["predictions"]
144 },
145 "measurements": {
146 item["measurement_id"] for item in record["operationalizations"]
147 },
148 "controls": {
149 item["control_id"] for item in record["negative_controls"]
150 },
151 "analyses": {
152 item["analysis_id"] for item in record["analysis_plan"]["analyses"]
153 },
154 }
156 for row in rows:
157 prediction_id = row["prediction_id"]
158 hypothesis_id = row["hypothesis_id"]
159 hypothesis_counts[hypothesis_id] += 1
160 if hypothesis_id in row["rival_hypothesis_ids"]:
161 errors.append(issue("FOCAL_HYPOTHESIS_LISTED_AS_RIVAL", prediction_id))
162 if (
163 _normalized_expectation(row["expected_if_focal"])
164 == _normalized_expectation(row["expected_if_rivals"])
165 ):
166 errors.append(issue("FOCAL_AND_RIVAL_EXPECTATIONS_IDENTICAL", prediction_id))
167 if not row["negative_control_ids"]:
168 warnings.append(issue("NEGATIVE_CONTROL_LINK_MISSING", prediction_id))
170 if record_ids is not None:
171 errors.extend(
172 _unknown(
173 [hypothesis_id],
174 record_ids["hypotheses"],
175 "UNKNOWN_HYPOTHESIS_ID",
176 prediction_id,
177 )
178 )
179 errors.extend(
180 _unknown(
181 row["rival_hypothesis_ids"],
182 record_ids["hypotheses"],
183 "UNKNOWN_RIVAL_HYPOTHESIS_ID",
184 prediction_id,
185 )
186 )
187 errors.extend(
188 _unknown(
189 [prediction_id],
190 record_ids["predictions"],
191 "UNKNOWN_PREDICTION_ID",
192 prediction_id,
193 )
194 )
195 errors.extend(
196 _unknown(
197 row["measurement_ids"],
198 record_ids["measurements"],
199 "UNKNOWN_MEASUREMENT_ID",
200 prediction_id,
201 )
202 )
203 errors.extend(
204 _unknown(
205 row["negative_control_ids"],
206 record_ids["controls"],
207 "UNKNOWN_CONTROL_ID",
208 prediction_id,
209 )
210 )
211 errors.extend(
212 _unknown(
213 row["analysis_ids"],
214 record_ids["analyses"],
215 "UNKNOWN_ANALYSIS_ID",
216 prediction_id,
217 )
218 )
219 record_prediction = next(
220 (
221 item
222 for item in record["predictions"]
223 if item["prediction_id"] == prediction_id
224 ),
225 None,
226 )
227 if (
228 record_prediction is not None
229 and record_prediction["hypothesis_id"] != hypothesis_id
230 ):
231 errors.append(
232 issue("PREDICTION_HYPOTHESIS_MISMATCH", prediction_id)
233 )
235 return {
236 "schema_version": "2.0",
237 "valid": not errors,
238 "status": "INVALID_MATRIX" if errors else "VALID_FOR_HUMAN_REVIEW",
239 "errors": errors,
240 "warnings": warnings,
241 "prediction_count": len(rows),
242 "hypothesis_prediction_counts": dict(sorted(hypothesis_counts.items())),
243 "prediction_ids": sorted(row["prediction_id"] for row in rows),
244 "record_cross_check_performed": record is not None,
245 "notice": (
246 "This report checks CSV structure, identifiers, declared contrasts, "
247 "and optional cross-links. It cannot determine whether a prediction "
248 "is scientifically discriminating, sufficiently precise, or likely, "
249 "and it never ranks or selects a hypothesis."
250 ),
251 }
254def build_parser() -> argparse.ArgumentParser:
255 parser = argparse.ArgumentParser(
256 description=(
257 "Validate a bounded local prediction/rival CSV and emit identifiers "
258 "and rule codes without scientific scoring."
259 )
260 )
261 parser.add_argument("matrix", help="Local prediction/rival matrix CSV")
262 parser.add_argument(
263 "--record", help="Optional local hypothesis record JSON for cross-checks"
264 )
265 parser.add_argument("-o", "--output", help="Optional local JSON report")
266 parser.add_argument(
267 "--force", action="store_true", help="Replace an existing output file"
268 )
269 return parser
272def main() -> int:
273 args = build_parser().parse_args()
274 try:
275 record = (
276 load_hypothesis_record(read_json(args.record)) if args.record else None
277 )
278 report = validate_matrix(load_matrix(args.matrix), record)
279 write_json_report(report, args.output, force=args.force)
280 return 0 if report["valid"] else 1
281 except ValidationError as exc:
282 return error_exit(exc)
285if __name__ == "__main__":
286 raise SystemExit(main())