Skills › Scientific Agent Skills › market-research-reports › audit_claim_citations.py
audit_claim_citations.py
326 lines · skills/market-research-reports/scripts/audit_claim_citations.py at 5f2bc03
1#!/usr/bin/env python3
2"""Audit claim-to-source mappings in local CSV ledgers."""
4from __future__ import annotations
6import argparse
7from pathlib import Path
8from typing import Any
10from _common import (
11 MAX_ROWS,
12 ValidationError,
13 error_exit,
14 parse_currency,
15 parse_iso_date,
16 parse_year,
17 read_csv_records,
18 read_json,
19 require_identifier,
20 require_list,
21 require_object,
22 require_text,
23 require_unique_identifiers,
24 split_ids,
25 write_json_report,
26)
28CLAIM_FIELDS = (
29 "claim_id",
30 "claim_text",
31 "claim_type",
32 "source_ids",
33 "calculation_id",
34 "assumption_ids",
35 "location",
36 "as_of_date",
37 "geography",
38 "currency",
39 "base_year",
40 "price_basis",
41 "measure_type",
42 "unit",
43 "taxonomy",
44 "taxonomy_version",
45 "revision_status",
46 "confidence",
47 "notes",
48)
49CLAIM_TYPES = {
50 "quantitative_fact",
51 "quantitative_estimate",
52 "calculation",
53 "forecast",
54 "qualitative_fact",
55 "opinion",
56 "recommendation",
57}
58QUANTITATIVE_TYPES = {
59 "quantitative_fact",
60 "quantitative_estimate",
61 "calculation",
62 "forecast",
63}
64EVIDENCE_REQUIRED_TYPES = CLAIM_TYPES - {"opinion", "recommendation"}
65REVISION_STATUSES = {
66 "preliminary",
67 "revised",
68 "final",
69 "current",
70 "vintage",
71 "unknown",
72 "not-applicable",
73}
74CONFIDENCE_LEVELS = {"high", "medium", "low", "not-assessed"}
75MEASURE_TYPES = {
76 "stock",
77 "flow",
78 "index",
79 "share",
80 "count",
81 "rate",
82 "price",
83 "not-applicable",
84}
87def _load_sources(path: str | Path) -> list[dict[str, Any]]:
88 suffix = Path(path).suffix.lower()
89 if suffix == ".csv":
90 return read_csv_records(
91 path,
92 required_fields=("source_id", "publication_date", "retrieval_date"),
93 )
94 if suffix == ".json":
95 payload = read_json(path)
96 raw = (
97 require_list(
98 payload.get("sources"), "sources", minimum=1, maximum=MAX_ROWS
99 )
100 if isinstance(payload, dict)
101 else require_list(payload, "sources", minimum=1, maximum=MAX_ROWS)
102 )
103 return [
104 require_object(record, f"sources[{index}]")
105 for index, record in enumerate(raw)
106 ]
107 raise ValidationError("source ledger must be .csv or .json")
110def _source_index(records: list[dict[str, Any]]) -> dict[str, dict[str, Any]]:
111 index: dict[str, dict[str, Any]] = {}
112 for position, record in enumerate(records):
113 context = f"sources[{position}]"
114 source_id = require_identifier(record.get("source_id"), f"{context}.source_id")
115 if source_id in index:
116 raise ValidationError(f"duplicate source_id: {source_id}")
117 publication = require_text(
118 record.get("publication_date"), f"{context}.publication_date", maximum=10
119 )
120 if publication != "not-stated":
121 if len(publication) == 4:
122 parse_year(publication, f"{context}.publication_date")
123 else:
124 parse_iso_date(publication, f"{context}.publication_date")
125 parse_iso_date(record.get("retrieval_date"), f"{context}.retrieval_date")
126 index[source_id] = record
127 return index
130def audit(
131 claims: list[dict[str, str]], sources: list[dict[str, Any]]
132) -> dict[str, Any]:
133 source_index = _source_index(sources)
134 errors: list[str] = []
135 warnings: list[str] = []
136 claim_ids: list[str] = []
137 cited_source_ids: set[str] = set()
138 type_counts: dict[str, int] = {}
139 uncited_claims: list[str] = []
141 for row_number, claim in enumerate(claims, start=2):
142 context = f"claim row {row_number}"
143 try:
144 claim_id = require_identifier(claim["claim_id"], f"{context}.claim_id")
145 claim_ids.append(claim_id)
146 claim_text = require_text(claim["claim_text"], f"{context}.claim_text")
147 if any(marker in claim_text.upper() for marker in ("[TBD]", "[TODO]")):
148 warnings.append(f"{claim_id}: claim text contains a placeholder")
150 claim_type = require_text(
151 claim["claim_type"], f"{context}.claim_type", maximum=30
152 )
153 if claim_type not in CLAIM_TYPES:
154 raise ValidationError(
155 f"{context}.claim_type must be one of "
156 f"{', '.join(sorted(CLAIM_TYPES))}"
157 )
158 type_counts[claim_type] = type_counts.get(claim_type, 0) + 1
160 source_ids = split_ids(
161 claim["source_ids"], f"{context}.source_ids", allow_empty=True
162 )
163 cited_source_ids.update(source_ids)
164 if claim_type in EVIDENCE_REQUIRED_TYPES and not source_ids:
165 uncited_claims.append(claim_id)
166 raise ValidationError(
167 f"{context}: {claim_type} requires at least one source_id"
168 )
169 missing_sources = sorted(set(source_ids) - set(source_index))
170 if missing_sources:
171 raise ValidationError(
172 f"{context}: source IDs are absent from the ledger: "
173 f"{', '.join(missing_sources)}"
174 )
176 calculation_id = require_text(
177 claim["calculation_id"],
178 f"{context}.calculation_id",
179 allow_empty=True,
180 maximum=96,
181 )
182 if calculation_id:
183 require_identifier(calculation_id, f"{context}.calculation_id")
184 assumption_ids = split_ids(
185 claim["assumption_ids"],
186 f"{context}.assumption_ids",
187 allow_empty=True,
188 )
189 if claim_type in {"calculation", "forecast"} and not calculation_id:
190 raise ValidationError(
191 f"{context}: {claim_type} requires calculation_id"
192 )
193 if claim_type == "forecast" and not assumption_ids:
194 raise ValidationError(
195 f"{context}: forecast requires assumption_ids"
196 )
198 require_text(claim["location"], f"{context}.location")
199 parse_iso_date(claim["as_of_date"], f"{context}.as_of_date")
200 require_text(claim["geography"], f"{context}.geography")
201 currency = parse_currency(
202 claim["currency"], f"{context}.currency", allow_empty=True
203 )
204 base_year = require_text(
205 claim["base_year"],
206 f"{context}.base_year",
207 allow_empty=True,
208 maximum=4,
209 )
210 if base_year:
211 parse_year(base_year, f"{context}.base_year")
212 price_basis = require_text(
213 claim["price_basis"], f"{context}.price_basis", maximum=20
214 )
215 if price_basis not in {
216 "nominal",
217 "real",
218 "current",
219 "constant",
220 "not-applicable",
221 }:
222 raise ValidationError(
223 f"{context}.price_basis has an unsupported value"
224 )
225 if currency and (not base_year or price_basis == "not-applicable"):
226 raise ValidationError(
227 f"{context}: monetary claims require base_year and price_basis"
228 )
230 measure_type = require_text(
231 claim["measure_type"], f"{context}.measure_type", maximum=20
232 )
233 if measure_type not in MEASURE_TYPES:
234 raise ValidationError(
235 f"{context}.measure_type has an unsupported value"
236 )
237 unit = require_text(claim["unit"], f"{context}.unit")
238 if claim_type in QUANTITATIVE_TYPES and (
239 measure_type == "not-applicable" or unit == "not-applicable"
240 ):
241 raise ValidationError(
242 f"{context}: quantitative claims require an explicit unit "
243 "and measure_type"
244 )
246 require_text(
247 claim["taxonomy"], f"{context}.taxonomy", allow_empty=True
248 )
249 require_text(
250 claim["taxonomy_version"],
251 f"{context}.taxonomy_version",
252 allow_empty=True,
253 )
254 revision_status = require_text(
255 claim["revision_status"],
256 f"{context}.revision_status",
257 maximum=20,
258 )
259 if revision_status not in REVISION_STATUSES:
260 raise ValidationError(
261 f"{context}.revision_status has an unsupported value"
262 )
263 confidence = require_text(
264 claim["confidence"], f"{context}.confidence", maximum=20
265 )
266 if confidence not in CONFIDENCE_LEVELS:
267 raise ValidationError(
268 f"{context}.confidence must be one of "
269 f"{', '.join(sorted(CONFIDENCE_LEVELS))}"
270 )
271 require_text(claim["notes"], f"{context}.notes", allow_empty=True)
272 except ValidationError as exc:
273 errors.append(str(exc))
275 try:
276 require_unique_identifiers(claim_ids, "claim ledger")
277 except ValidationError as exc:
278 errors.append(str(exc))
280 unused_sources = sorted(set(source_index) - cited_source_ids)
281 if unused_sources:
282 warnings.append(
283 f"{len(unused_sources)} source ledger entries are not mapped to a claim"
284 )
285 return {
286 "valid": not errors,
287 "claim_count": len(claims),
288 "source_count": len(source_index),
289 "claim_type_counts": dict(sorted(type_counts.items())),
290 "cited_source_count": len(cited_source_ids),
291 "uncited_claim_ids": sorted(set(uncited_claims)),
292 "unused_source_ids": unused_sources,
293 "warnings": warnings,
294 "errors": errors,
295 }
298def build_parser() -> argparse.ArgumentParser:
299 parser = argparse.ArgumentParser(
300 description=(
301 "Audit a strict local claims CSV against a local source ledger. "
302 "Every evidence claim must map to explicit source IDs."
303 )
304 )
305 parser.add_argument("claims", help="Local .csv claims ledger")
306 parser.add_argument("sources", help="Local .csv or .json source ledger")
307 parser.add_argument("--output", help="Optional local .json audit report")
308 parser.add_argument(
309 "--force", action="store_true", help="Replace an existing --output file"
310 )
311 return parser
314def main() -> int:
315 args = build_parser().parse_args()
316 try:
317 claims = read_csv_records(args.claims, required_fields=CLAIM_FIELDS)
318 report = audit(claims, _load_sources(args.sources))
319 write_json_report(report, args.output, force=args.force)
320 return 0 if report["valid"] else 1
321 except ValidationError as exc:
322 return error_exit(exc)
325if __name__ == "__main__":
326 raise SystemExit(main())