Skills › Scientific Agent Skills › market-research-reports › check_unit_consistency.py
check_unit_consistency.py
221 lines · skills/market-research-reports/scripts/check_unit_consistency.py at 5f2bc03
1#!/usr/bin/env python3
2"""Check units, currency, base year, taxonomy, and denominator consistency."""
4from __future__ import annotations
6import argparse
7import re
8from typing import Any
10from _common import (
11 ValidationError,
12 error_exit,
13 parse_currency,
14 parse_iso_date,
15 parse_number,
16 parse_year,
17 read_csv_records,
18 require_identifier,
19 require_text,
20 require_unique_identifiers,
21 write_json_report,
22)
24REQUIRED_FIELDS = (
25 "record_id",
26 "comparison_group",
27 "value",
28 "unit",
29 "currency",
30 "base_year",
31 "price_basis",
32 "measure_type",
33 "geography",
34 "period",
35 "taxonomy",
36 "taxonomy_version",
37 "denominator_id",
38 "source_id",
39)
40PRICE_BASES = {
41 "nominal",
42 "real",
43 "current",
44 "constant",
45 "chained",
46 "not-applicable",
47}
48MEASURE_TYPES = {"stock", "flow", "index", "share", "count", "rate", "price"}
49PERIOD_RE = re.compile(
50 r"^(?:\d{4}|\d{4}-Q[1-4]|\d{4}-M(?:0[1-9]|1[0-2])|\d{4}-\d{2}-\d{2})$"
51)
52CONSISTENCY_FIELDS = (
53 "unit",
54 "currency",
55 "base_year",
56 "price_basis",
57 "measure_type",
58 "geography",
59 "taxonomy",
60 "taxonomy_version",
61 "denominator_id",
62)
65def check(rows: list[dict[str, str]]) -> dict[str, Any]:
66 errors: list[str] = []
67 record_ids: list[str] = []
68 groups: dict[str, dict[str, set[str]]] = {}
70 for row_number, row in enumerate(rows, start=2):
71 context = f"row {row_number}"
72 try:
73 record_id = require_identifier(row["record_id"], f"{context}.record_id")
74 record_ids.append(record_id)
75 group = require_identifier(
76 row["comparison_group"], f"{context}.comparison_group"
77 )
78 parse_number(
79 row["value"],
80 f"{context}.value",
81 minimum=-1e18,
82 maximum=1e18,
83 )
84 unit = require_text(row["unit"], f"{context}.unit", maximum=80)
85 currency = parse_currency(
86 row["currency"], f"{context}.currency", allow_empty=True
87 )
88 base_year = require_text(
89 row["base_year"],
90 f"{context}.base_year",
91 allow_empty=True,
92 maximum=4,
93 )
94 if base_year:
95 parse_year(base_year, f"{context}.base_year")
96 price_basis = require_text(
97 row["price_basis"], f"{context}.price_basis", maximum=20
98 )
99 if price_basis not in PRICE_BASES:
100 raise ValidationError(
101 f"{context}.price_basis must be one of "
102 f"{', '.join(sorted(PRICE_BASES))}"
103 )
104 if currency and (not base_year or price_basis == "not-applicable"):
105 raise ValidationError(
106 f"{context}: monetary rows require base_year and price_basis"
107 )
108 if not currency and price_basis != "not-applicable":
109 raise ValidationError(
110 f"{context}: non-monetary rows must use price_basis "
111 "'not-applicable'"
112 )
114 measure_type = require_text(
115 row["measure_type"], f"{context}.measure_type", maximum=20
116 )
117 if measure_type not in MEASURE_TYPES:
118 raise ValidationError(
119 f"{context}.measure_type must be one of "
120 f"{', '.join(sorted(MEASURE_TYPES))}"
121 )
122 geography = require_text(
123 row["geography"], f"{context}.geography", maximum=200
124 )
125 period = require_text(row["period"], f"{context}.period", maximum=10)
126 if not PERIOD_RE.fullmatch(period):
127 raise ValidationError(
128 f"{context}.period must be YYYY, YYYY-Qn, YYYY-Mnn, "
129 "or YYYY-MM-DD"
130 )
131 parse_year(period[:4], f"{context}.period year")
132 if len(period) == 10:
133 parse_iso_date(period, f"{context}.period")
134 taxonomy = require_text(
135 row["taxonomy"], f"{context}.taxonomy", allow_empty=True
136 )
137 taxonomy_version = require_text(
138 row["taxonomy_version"],
139 f"{context}.taxonomy_version",
140 allow_empty=True,
141 )
142 denominator_id = require_identifier(
143 row["denominator_id"], f"{context}.denominator_id"
144 )
145 require_identifier(row["source_id"], f"{context}.source_id")
147 normalized = {
148 "unit": unit,
149 "currency": currency,
150 "base_year": base_year,
151 "price_basis": price_basis,
152 "measure_type": measure_type,
153 "geography": geography,
154 "taxonomy": taxonomy,
155 "taxonomy_version": taxonomy_version,
156 "denominator_id": denominator_id,
157 }
158 group_values = groups.setdefault(
159 group, {field: set() for field in CONSISTENCY_FIELDS}
160 )
161 for field, value in normalized.items():
162 group_values[field].add(value)
163 except ValidationError as exc:
164 errors.append(str(exc))
166 try:
167 require_unique_identifiers(record_ids, "record IDs")
168 except ValidationError as exc:
169 errors.append(str(exc))
171 mismatches: dict[str, dict[str, list[str]]] = {}
172 for group, values in sorted(groups.items()):
173 group_mismatches = {
174 field: sorted(observed)
175 for field, observed in values.items()
176 if len(observed) > 1
177 }
178 if group_mismatches:
179 mismatches[group] = group_mismatches
180 errors.append(
181 f"comparison group {group!r} mixes: "
182 + ", ".join(sorted(group_mismatches))
183 )
185 return {
186 "valid": not errors,
187 "record_count": len(rows),
188 "comparison_group_count": len(groups),
189 "mismatches": mismatches,
190 "errors": errors,
191 }
194def build_parser() -> argparse.ArgumentParser:
195 parser = argparse.ArgumentParser(
196 description=(
197 "Check a local CSV for internally comparable units, currencies, base "
198 "years, price bases, measure types, geographies, taxonomies, and "
199 "denominators."
200 )
201 )
202 parser.add_argument("input", help="Local .csv consistency input")
203 parser.add_argument("--output", help="Optional local .json report")
204 parser.add_argument(
205 "--force", action="store_true", help="Replace an existing --output file"
206 )
207 return parser
210def main() -> int:
211 args = build_parser().parse_args()
212 try:
213 report = check(read_csv_records(args.input, required_fields=REQUIRED_FIELDS))
214 write_json_report(report, args.output, force=args.force)
215 return 0 if report["valid"] else 1
216 except ValidationError as exc:
217 return error_exit(exc)
220if __name__ == "__main__":
221 raise SystemExit(main())