Skills › Scientific Agent Skills › hypothesis-generation › lint_causal_claims.py
lint_causal_claims.py
189 lines · skills/hypothesis-generation/scripts/lint_causal_claims.py at 5f2bc03
1#!/usr/bin/env python3
2"""Lexically lint causal versus associational claims in bounded Markdown."""
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_markdown,
16 write_json_report,
17)
19TAG_RE = re.compile(
20 r"\[(claim|estimand|identification|confounding|selection|collider|"
21 r"reverse-causation):([A-Za-z0-9_.:-]+)\]",
22 re.IGNORECASE,
23)
24CAUSAL_RE = re.compile(
25 r"\b(?:causes?|caused|causal\s+(?:effect|impact)|effect\s+of|"
26 r"leads?\s+to|results?\s+in|increases?|reduces?|prevents?|improves?|"
27 r"worsens?|drives?|mediates?|produces?)\b",
28 re.IGNORECASE,
29)
30ASSOCIATION_RE = re.compile(
31 r"\b(?:associated\s+with|association|correlates?\s+with|correlation|"
32 r"co-var(?:y|ies|ied)\s+with|predicts?|linked\s+to)\b",
33 re.IGNORECASE,
34)
35CLAIM_TYPES = {"causal", "associational", "descriptive", "predictive", "mechanistic"}
36IDENTIFICATION_TYPES = {
37 "randomized",
38 "quasi_experimental",
39 "observational_assumption_dependent",
40 "mechanistic_experiment",
41 "other_assumption_dependent",
42}
43RISK_STATES = {"assessed", "unresolved", "not_applicable"}
44REQUIRED_CAUSAL_TAGS = (
45 "estimand",
46 "identification",
47 "confounding",
48 "selection",
49 "collider",
50 "reverse-causation",
51)
54def _finding(code: str, line_number: int) -> dict[str, Any]:
55 return {"code": code, "line": line_number}
58def lint(markdown: str) -> dict[str, Any]:
59 errors: list[dict[str, Any]] = []
60 warnings: list[dict[str, Any]] = []
61 claim_counts: Counter[str] = Counter()
62 causal_trigger_lines = 0
63 association_trigger_lines = 0
64 in_fence = False
66 for line_number, raw_line in enumerate(markdown.splitlines(), start=1):
67 stripped = raw_line.strip()
68 if stripped.startswith("```") or stripped.startswith("~~~"):
69 in_fence = not in_fence
70 continue
71 if in_fence or not stripped or stripped.startswith("#"):
72 continue
74 matches = [(key.casefold(), value) for key, value in TAG_RE.findall(stripped)]
75 key_counts = Counter(key for key, _ in matches)
76 for key, count in key_counts.items():
77 if count > 1:
78 errors.append(_finding(f"DUPLICATE_{key.upper()}_TAG", line_number))
79 tags = {key: value for key, value in matches}
80 claim_type = tags.get("claim", "").casefold()
81 has_causal_language = CAUSAL_RE.search(stripped) is not None
82 has_association_language = ASSOCIATION_RE.search(stripped) is not None
84 if claim_type:
85 if claim_type not in CLAIM_TYPES:
86 errors.append(_finding("INVALID_CLAIM_TYPE", line_number))
87 else:
88 claim_counts[claim_type] += 1
90 if has_causal_language:
91 causal_trigger_lines += 1
92 if not claim_type:
93 errors.append(_finding("UNMARKED_CAUSAL_LANGUAGE", line_number))
94 elif claim_type not in {"causal", "mechanistic"}:
95 errors.append(
96 _finding("CAUSAL_LANGUAGE_IN_NONCAUSAL_CLAIM", line_number)
97 )
99 if has_association_language:
100 association_trigger_lines += 1
101 if not claim_type:
102 warnings.append(
103 _finding("UNMARKED_ASSOCIATIONAL_LANGUAGE", line_number)
104 )
105 elif claim_type == "causal" and not has_causal_language:
106 warnings.append(
107 _finding("CAUSAL_TAG_WITH_ASSOCIATION_ONLY_LANGUAGE", line_number)
108 )
110 if claim_type == "causal":
111 for required_tag in REQUIRED_CAUSAL_TAGS:
112 if required_tag not in tags:
113 code_tag = required_tag.replace("-", "_").upper()
114 errors.append(
115 _finding(
116 f"CAUSAL_CLAIM_MISSING_{code_tag}_TAG",
117 line_number,
118 )
119 )
120 identification = tags.get("identification", "").casefold()
121 if identification and identification not in IDENTIFICATION_TYPES:
122 errors.append(_finding("INVALID_IDENTIFICATION_TAG", line_number))
123 for risk_tag in (
124 "confounding",
125 "selection",
126 "collider",
127 "reverse-causation",
128 ):
129 state = tags.get(risk_tag, "").casefold()
130 code_tag = risk_tag.replace("-", "_").upper()
131 if state and state not in RISK_STATES:
132 errors.append(
133 _finding(f"INVALID_{code_tag}_STATE", line_number)
134 )
135 if state == "unresolved":
136 warnings.append(
137 _finding(f"UNRESOLVED_{code_tag}_RISK", line_number)
138 )
139 elif claim_type and "estimand" in tags:
140 warnings.append(_finding("ESTIMAND_TAG_ON_NONCAUSAL_CLAIM", line_number))
142 if in_fence:
143 errors.append(issue("UNCLOSED_MARKDOWN_FENCE", "document"))
145 return {
146 "schema_version": "2.0",
147 "valid": not errors,
148 "status": "INVALID_CLAIM_MARKUP" if errors else "VALID_LEXICAL_LINT",
149 "errors": errors,
150 "warnings": warnings,
151 "claim_type_counts": dict(sorted(claim_counts.items())),
152 "causal_trigger_line_count": causal_trigger_lines,
153 "associational_trigger_line_count": association_trigger_lines,
154 "notice": (
155 "This deterministic lexical lint does not determine whether language "
156 "is scientifically causal, whether an estimand is well defined, or "
157 "whether identification assumptions hold. Review every flagged and "
158 "unflagged claim manually."
159 ),
160 }
163def build_parser() -> argparse.ArgumentParser:
164 parser = argparse.ArgumentParser(
165 description=(
166 "Lint bounded local Markdown for causal/associational claim annotations "
167 "and emit only rule codes and line numbers."
168 )
169 )
170 parser.add_argument("document", help="Local Markdown document")
171 parser.add_argument("-o", "--output", help="Optional local JSON report")
172 parser.add_argument(
173 "--force", action="store_true", help="Replace an existing output file"
174 )
175 return parser
178def main() -> int:
179 args = build_parser().parse_args()
180 try:
181 report = lint(read_markdown(args.document))
182 write_json_report(report, args.output, force=args.force)
183 return 0 if report["valid"] else 1
184 except ValidationError as exc:
185 return error_exit(exc)
188if __name__ == "__main__":
189 raise SystemExit(main())