diff --git a/packages/microcosm-build/tests/engine_free/shared/test_calibration_columns.py b/packages/microcosm-build/tests/engine_free/shared/test_calibration_columns.py index 4a94210bc1..943a7b7ada 100644 --- a/packages/microcosm-build/tests/engine_free/shared/test_calibration_columns.py +++ b/packages/microcosm-build/tests/engine_free/shared/test_calibration_columns.py @@ -3,18 +3,26 @@ from __future__ import annotations import copy -import hashlib +import functools import json +import math +from collections.abc import Mapping +from numbers import Integral, Real +from typing import Any import numpy as np import pandas as pd import pytest -from hypothesis import HealthCheck, given, settings +from hypothesis import HealthCheck, example, given, settings from hypothesis import strategies as st from microcosm.build.calibration_columns import prepare_calibration_columns from microcosm.build.calibration_predicates import calibration_predicate_mask from microcosm.build.country_spec import _validate_target_references +from microcosm.build.reference_intervals import ( + interval_indicators, + validate_interval_document, +) from microcosm.build.reference_profiles import resolve_reference_profiles from microcosm.calibrate import Target, TargetSet from microcosm.calibrate.matrix import build_constraint_matrix @@ -720,11 +728,35 @@ def test_receipt_predicate_masks_match_preparation_for_validated_raw_inputs(rows ) -def test_prepare_matches_pinned_legacy_output_digest(): - # Captured from the pre-refactor implementation at 4f333f2c2ee70b6b758586f402d966df7297a2f0, - # run on these NZ bindings, which #1398 extended after the capture. - tables = prepare_calibration_columns(_tables(), _bindings()) - document = { +@functools.cache +def _legacy_prepare(): + """Run 4f333f2c's frozen preparation with today's interval module. + + 2f220f70c moved the predicates into calibration_predicates; the frozen + functions are the oracle for that refactor and for later binding edits. + """ + path = ( + paths_for("microcosm-build").tests + / "fixtures/calibration_columns_legacy.py.txt" + ) + scope = { + "Any": Any, + "Integral": Integral, + "Mapping": Mapping, + "Real": Real, + "interval_indicators": interval_indicators, + "math": math, + "np": np, + "pd": pd, + "validate_interval_document": validate_interval_document, + } + source = "from __future__ import annotations\n" + path.read_text() + exec(compile(source, str(path), "exec"), scope) + return scope["prepare_calibration_columns"] + + +def _table_document(tables): + return { entity: { "columns": list(table), "dtypes": [str(dtype) for dtype in table.dtypes], @@ -733,7 +765,253 @@ def test_prepare_matches_pinned_legacy_output_digest(): } for entity, table in tables.items() } - payload = json.dumps(document, sort_keys=True, separators=(",", ":")).encode() - assert hashlib.sha256(payload).hexdigest() == ( - "1b22f4e11e6e6c901866010d709acadaaad638e0774e80751e0ff23257a2bde7" + + +def _prepared_document(tables): + return json.dumps(_table_document(tables), sort_keys=True, separators=(",", ":")) + + +def _axis(axis): + return [ + type(axis).__qualname__, + repr(axis.dtype), + _typed(list(axis.names)), + _typed(axis.tolist()), + ] + + +def _typed(value): + """Encode ``value`` so that a change of type, content or pandas metadata shows. + + Each scalar carries its type and repr, so ``pd.NA``, ``None``, ``np.int64(7)`` + and the text ``"np.int64(7)"`` all encode differently. A frame keeps each + column's dtype repr (including categories, order flag and string storage), + both axes' class, dtype, names and labels, its ``attrs`` and its + duplicate-label flag. Mapping order doesn't count; sequence order does. State + that a repr hides, such as a MultiIndex level's dtype, stays out of view. + """ + if isinstance(value, pd.DataFrame): + return { + "attrs": _typed(value.attrs), + "columns": _axis(value.columns), + "dtypes": [repr(dtype) for dtype in value.dtypes], + "flags": value.flags.allows_duplicate_labels, + "index": _axis(value.index), + "values": [ + _typed(value.iloc[:, position].tolist()) + for position in range(value.shape[1]) + ], + } + if isinstance(value, Mapping): + items = [[_typed(key), _typed(item)] for key, item in value.items()] + return ["mapping", sorted(items, key=json.dumps)] + if isinstance(value, (list, tuple)): + return [type(value).__qualname__, [_typed(item) for item in value]] + return [f"{type(value).__module__}.{type(value).__qualname__}", repr(value)] + + +def _outcome(prepare, tables, bindings, **options): + """Return one implementation's result and the state of its own input copies. + + Each call gets independent copies of every argument, so neither + implementation can change what the other reads, and the typed snapshot + shows a write an implementation makes to its inputs. + """ + tables, bindings, options = copy.deepcopy((tables, bindings, options)) + try: + prepared = prepare(tables, bindings, **options) + except ValueError as error: + result = ("refused", str(error)) + else: + result = ("prepared", json.dumps(_typed(prepared))) + return result, json.dumps(_typed((tables, bindings, options))) + + +def test_prepare_matches_the_frozen_legacy_on_nz_bindings(): + live = prepare_calibration_columns(_tables(), _bindings()) + legacy = _legacy_prepare()(_tables(), _bindings()) + assert list(live) == list(legacy) == ["person", "household"] + for entity, table in legacy.items(): + pd.testing.assert_frame_equal(live[entity], table, check_exact=True) + assert _prepared_document(live) == _prepared_document(legacy) + + +_NZ_DEFECTS = ( + "negative_age", + "fractional_age", + "boolean_age", + "missing_age", + "unknown_region", + "unknown_household_region", + "unknown_tenure", + "orphan_member", + "empty_household", + "duplicate_person", + "boolean_identity", +) + + +@st.composite +def _nz_raw_tables(draw, defect): + domains = _bindings()["input_domains"] + households = draw(st.integers(min_value=0 if defect is None else 1, max_value=6)) + membership = draw( + st.permutations( + [ + *range(households), + *draw( + st.lists( + st.integers(min_value=0, max_value=max(households - 1, 0)), + min_size=1 if defect == "duplicate_person" else 0, + max_size=10 if households else 0, + ) + ), + ] + ) ) + people = len(membership) + identities = st.one_of( + st.integers(min_value=-(10**6), max_value=10**6), + st.integers(min_value=-(10**6), max_value=10**6).map(np.int64), + st.text(alphabet="abcdefgh0123456789", min_size=1, max_size=4), + ) + household_ids = draw( + st.lists(identities, min_size=households, max_size=households, unique=True) + ) + ages = st.integers(min_value=0, max_value=130) + tables = { + "person": pd.DataFrame( + { + "person_id": draw( + st.lists(identities, min_size=people, max_size=people, unique=True) + ), + "person_household_id": [household_ids[index] for index in membership], + "age": draw( + st.lists( + st.one_of(ages, ages.map(float)), + min_size=people, + max_size=people, + ) + ), + "region_code": draw( + st.lists( + st.sampled_from(domains["person"]["region_code"]["values"]), + min_size=people, + max_size=people, + ) + ), + } + ), + "household": pd.DataFrame( + { + "household_id": household_ids, + "household_region_code": draw( + st.lists( + st.sampled_from( + domains["household"]["household_region_code"]["values"] + ), + min_size=households, + max_size=households, + ) + ), + "tenure": draw( + st.lists( + st.sampled_from(domains["household"]["tenure"]["values"]), + min_size=households, + max_size=households, + ) + ), + } + ), + } + if draw(st.booleans()): + for table in tables.values(): + table.index = draw(st.permutations(range(100, 100 + len(table)))) + if defect is None: + return tables + person, household = tables["person"], tables["household"] + position = draw(st.integers(min_value=0, max_value=people - 1)) + row, next_row = person.index[position], person.index[(position + 1) % people] + if defect == "negative_age": + person["age"] = person["age"].astype(object) + person.at[row, "age"] = -1 + elif defect == "fractional_age": + person["age"] = person["age"].astype(float) + person.at[row, "age"] = 40.5 + elif defect == "boolean_age": + person["age"] = person["age"].astype(object) + person.at[row, "age"] = True + elif defect == "missing_age": + person["age"] = person["age"].astype(float) + person.at[row, "age"] = np.nan + elif defect == "unknown_region": + person.at[row, "region_code"] = "10" + elif defect == "unknown_household_region": + household.at[household.index[0], "household_region_code"] = "1" + elif defect == "unknown_tenure": + household.at[household.index[0], "tenure"] = "owned" + elif defect == "orphan_member": + person["person_household_id"] = person["person_household_id"].astype(object) + person.at[row, "person_household_id"] = "orphan-household" + elif defect == "empty_household": + tables["household"] = pd.concat( + [household, household.iloc[:1].assign(household_id="empty-household")], + ignore_index=True, + ) + elif defect == "duplicate_person": + person["person_id"] = person["person_id"].astype(object) + person.at[row, "person_id"] = person.at[next_row, "person_id"] + elif defect == "boolean_identity": + person["person_id"] = person["person_id"].astype(object) + person.at[row, "person_id"] = False + return tables + + +@st.composite +def _nz_tables_with_categories(draw, defect): + tables = draw(_nz_raw_tables(defect)) + if draw(st.booleans()): + tenure = tables["household"]["tenure"] + categories = draw(st.permutations(sorted(set(tenure.tolist())))) + tables["household"]["tenure"] = tenure.astype( + pd.CategoricalDtype(categories, ordered=draw(st.booleans())) + ) + return tables + + +@pytest.mark.parametrize("defect", [None, *_NZ_DEFECTS]) +@settings(max_examples=15, deadline=None, suppress_health_check=[HealthCheck.too_slow]) +@given(data=st.data()) +def test_prepare_agrees_with_the_frozen_legacy_on_generated_nz_tables(defect, data): + tables = data.draw(_nz_tables_with_categories(defect)) + bindings = _bindings() + live = _outcome(prepare_calibration_columns, tables, bindings) + assert live[0][0] == ("prepared" if defect is None else "refused") + assert live == _outcome(_legacy_prepare(), tables, bindings) + + +@settings(max_examples=60, deadline=None, suppress_health_check=[HealthCheck.too_slow]) +@given( + st.sampled_from(["person", "household"]), + st.lists( + st.tuples( + st.one_of( + st.sampled_from([-1, 0, 1, 99_999, 100_000, 100_001, 1_000_000]), + st.integers(min_value=-(10**7), max_value=10**7), + st.sampled_from([True, 10.0, "10", pd.NA]), + ), + st.sampled_from(["included", "excluded", "unresolved"]), + ), + min_size=1, + max_size=12, + ), +) +# An excluded row may carry a missing amount and still prepare. +@example("person", [(pd.NA, "excluded"), (100_001, "included")]) +@example("household", [(None, "excluded")]) +def test_interval_outputs_agree_with_the_frozen_legacy(entity, rows): + intervals, bindings = _interval_contract(entity=entity) + amounts, statuses = zip(*rows, strict=True) + tables = _interval_tables(pd.array(amounts, dtype=object), list(statuses), entity) + live = _outcome(prepare_calibration_columns, tables, bindings, intervals=intervals) + assert live == _outcome(_legacy_prepare(), tables, bindings, intervals=intervals) diff --git a/packages/microcosm-build/tests/engine_free/shared/test_nz_money_normalization.py b/packages/microcosm-build/tests/engine_free/shared/test_nz_money_normalization.py index b0843c33e7..45296b7557 100644 --- a/packages/microcosm-build/tests/engine_free/shared/test_nz_money_normalization.py +++ b/packages/microcosm-build/tests/engine_free/shared/test_nz_money_normalization.py @@ -9,6 +9,9 @@ import ast import json +import re +from collections.abc import Mapping +from decimal import Decimal, InvalidOperation from fractions import Fraction import pytest @@ -43,6 +46,68 @@ def _declaration(): return validate_money_normalization(_load("money_normalization.json")) +# A whole numeric token: an optional sign, digits and points, then any +# exponent, with no letter, digit or underscore right after it. A sign that +# follows a digit separates two numbers, as in the range 1.650-1.654 or the sum +# 0+1.654, so the number after it stands unsigned. Otherwise the token, +# including its sign, may not follow a letter, digit, underscore, point or +# sign: -1.654 is negative, and v-1.654, v1.654 and a1654e-3b hold no number. +# A run with two points such as 2.1.654 matches, but Decimal rejects it and +# _text_numbers skips it. _text_numbers also drops one trailing point, which +# ends a sentence. +_NUMBER_TOKEN = re.compile( + r"(?:(? dict[str, pd.DataFrame]: + """Return independent tables with the contract's prepared columns. + + All required inputs are validated before any column is produced. Existing + prepared columns are recomputed, so a changed geography or age cannot retain + stale indicators. Missing/unknown inputs never become all-false categories. + An interval output names its entity, an effective table of ``intervals`` + and one of its bands; every record of each named table is validated, and + any refusal raises, before an output exists. + """ + if bindings.get("schema_version") != 1: + raise ValueError("Calibration bindings require schema_version 1.") + required = bindings.get("required_inputs") + domains = bindings.get("input_domains") + columns = bindings.get("columns") + if not isinstance(required, Mapping) or not isinstance(domains, Mapping): + raise ValueError("Calibration bindings need required_inputs and input_domains.") + if not isinstance(columns, (list, tuple)) or not columns: + raise ValueError("Calibration bindings need nonempty columns.") + for entity, names in required.items(): + if entity not in tables: + raise ValueError(f"Calibration preparation needs the {entity!r} table.") + if not isinstance(names, (list, tuple)) or not names: + raise ValueError(f"Calibration required_inputs for {entity!r} are invalid.") + entity_domains = domains.get(entity, {}) + for name in names: + if name not in tables[entity]: + raise ValueError(f"Calibration preparation needs {entity}.{name}.") + domain = entity_domains.get(name) + if not isinstance(domain, Mapping): + raise ValueError(f"Calibration input {entity}.{name} needs a domain.") + _validate_input(tables[entity][name], domain, f"{entity}.{name}") + + interval_columns = _interval_columns(tables, columns, required, intervals) + out = {entity: table.copy(deep=True) for entity, table in tables.items()} + derived = _prepare_roster_counts(out, bindings.get("roster_counts", ()), required) + prepared: dict[str, dict[str, object]] = {} + seen: set[tuple[str, str]] = set() + for entry in columns: + if not isinstance(entry, Mapping): + raise ValueError("Calibration column entries must be objects.") + entity, name = entry.get("entity"), entry.get("name") + if entity not in required or not isinstance(name, str) or not name: + raise ValueError("Calibration column needs a declared entity and name.") + key = (entity, name) + if ( + key in seen + or name in required[entity] + or name in derived.get(entity, ()) + or name.endswith("_id") + ): + raise ValueError( + f"Calibration output {entity}.{name} is repeated or structural." + ) + seen.add(key) + if "interval" in entry: + prepared.setdefault(entity, {})[name] = interval_columns[key] + continue + if "constant" in entry: + value = entry["constant"] + if "predicates" in entry or not _finite_number(value): + raise ValueError(f"Calibration constant {entity}.{name} is invalid.") + prepared.setdefault(entity, {})[name] = value + continue + predicates = entry.get("predicates") + if not isinstance(predicates, (list, tuple)) or not predicates: + raise ValueError(f"Calibration output {entity}.{name} needs predicates.") + mask = np.ones(len(out[entity]), dtype=bool) + for predicate in predicates: + if not isinstance(predicate, Mapping): + raise ValueError("Calibration predicates must be objects.") + column = predicate.get("column") + if ( + column not in {*required[entity], *derived.get(entity, ())} + or "value" not in predicate + ): + raise ValueError( + f"Calibration predicate needs a validated {entity} input." + ) + values = out[entity][column] + operator, value = predicate.get("operator"), predicate["value"] + if operator == "eq": + part = values.eq(value) + elif operator == "ge": + part = values.ge(value) + elif operator == "lt": + part = values.lt(value) + elif operator == "in" and isinstance(value, (list, tuple)) and value: + part = values.isin(value) + else: + raise ValueError(f"Unsupported calibration predicate {operator!r}.") + mask &= part.to_numpy(dtype=bool) + prepared.setdefault(entity, {})[name] = mask + for entity, values in prepared.items(): + original_order = list(out[entity].columns) + computed = pd.DataFrame(values, index=out[entity].index) + out[entity] = pd.concat( + [out[entity].drop(columns=list(values), errors="ignore"), computed], axis=1 + ) + order = original_order + [name for name in values if name not in original_order] + out[entity] = out[entity].loc[:, order] + return out + + +def _interval_columns( + tables: Mapping[str, pd.DataFrame], + columns: object, + required: Mapping[str, Any], + intervals: Mapping[str, Any] | None, +) -> dict[tuple[str, str], np.ndarray]: + """Classify every named table once; refuse before any output is produced.""" + entries = [ + entry for entry in columns if isinstance(entry, Mapping) and "interval" in entry + ] + if not entries: + return {} + if intervals is None: + raise ValueError("Interval outputs need the effective interval document.") + document = validate_interval_document(intervals) + requested: dict[str, list[tuple[str, str, str]]] = {} + for entry in entries: + entity, name, binding = ( + entry.get("entity"), + entry.get("name"), + entry["interval"], + ) + if set(entry) != {"entity", "name", "interval"}: + raise ValueError( + f"Interval output {entity}.{name} declares only its entity, name " + "and interval." + ) + if not isinstance(binding, Mapping) or set(binding) != {"table", "band"}: + raise ValueError("An interval output names exactly its table and band.") + table = document.get(binding["table"]) + if table is None or binding["band"] not in table.band_ids: + raise ValueError( + f"Interval output {entity}.{name} names an undeclared table or band." + ) + if entity not in required or entity != table.entity: + raise ValueError( + f"Interval output {entity}.{name} must name its table's entity." + ) + if name != table.indicators[table.band_ids.index(binding["band"])]: + raise ValueError( + f"Interval output {entity}.{name} must be its band's declared indicator." + ) + if ( + table.declaration["source_period"] != table.declaration["target_period"] + and not table.scaled + ): + raise ValueError( + f"Interval table {table.id!r} has source-period cuts; scale it to its " + "target period before preparing indicators." + ) + requested.setdefault(table.id, []).append((entity, name, binding["band"])) + result: dict[tuple[str, str], np.ndarray] = {} + computed = {} + for table_id in requested: + table = document[table_id] + data = tables.get(table.entity) + if data is None or any( + column not in data + for column in (table.income_column, table.universe_column) + ): + raise ValueError( + f"Interval table {table_id!r} needs {table.entity}." + f"{table.income_column} and {table.entity}.{table.universe_column}." + ) + computed[table_id] = interval_indicators( + table, + data[table.income_column].tolist(), + data[table.universe_column].tolist(), + ) + for table_id, outputs in requested.items(): + for entity, name, _ in outputs: + result[entity, name] = computed[table_id][name] + return result + + +def _finite_number(value: object) -> bool: + return ( + isinstance(value, Real) and not isinstance(value, bool) and math.isfinite(value) + ) + + +def _identity(value: object) -> tuple[type, object]: + if ( + isinstance(value, bool) + or not isinstance(value, (str, Integral)) + or isinstance(value, str) + and not value + ): + raise ValueError( + "Calibration membership identities must be strings or integers." + ) + return (str, value) if isinstance(value, str) else (int, int(value)) + + +def _prepare_roster_counts( + tables: dict[str, pd.DataFrame], entries: object, required: Mapping[str, Any] +) -> dict[str, set[str]]: + if not isinstance(entries, (list, tuple)): + raise ValueError("Calibration roster_counts must be a list.") + derived: dict[str, set[str]] = {} + for entry in entries: + if not isinstance(entry, Mapping): + raise ValueError("Calibration roster count must be an object.") + entity, members, name = ( + entry.get("entity"), + entry.get("member_entity"), + entry.get("name"), + ) + group_id, member_id, membership = ( + entry.get("entity_id"), + entry.get("member_id"), + entry.get("membership"), + ) + if ( + entity not in required + or members not in required + or not isinstance(name, str) + or not name + or name.endswith("_id") + or name in required[entity] + or name in derived.get(entity, ()) + or group_id not in required[entity] + or member_id not in required[members] + or membership not in required[members] + or not isinstance(entry.get("representation"), str) + or not entry["representation"] + ): + raise ValueError( + "Calibration roster count needs declared identities and representation." + ) + groups = [_identity(value) for value in tables[entity][group_id].tolist()] + identities = [_identity(value) for value in tables[members][member_id].tolist()] + if len(set(groups)) != len(groups) or len(set(identities)) != len(identities): + raise ValueError( + "Calibration roster contains duplicate membership identities." + ) + counts = dict.fromkeys(groups, 0) + for value in tables[members][membership].tolist(): + key = _identity(value) + if key not in counts: + raise ValueError("Calibration roster contains orphan membership.") + counts[key] += 1 + if any(count == 0 for count in counts.values()): + raise ValueError("Calibration roster contains empty households or groups.") + tables[entity][name] = [counts[key] for key in groups] + derived.setdefault(entity, set()).add(name) + return derived + + +def _validate_input(values: pd.Series, domain: Mapping[str, Any], name: str) -> None: + if values.isna().any(): + raise ValueError(f"Calibration input {name} contains missing values.") + if domain.get("kind") == "identity": + for value in values.tolist(): + _identity(value) + valid = True + elif domain.get("kind") == "signed_integer": + valid = all( + isinstance(value, Integral) and not isinstance(value, (bool, np.bool_)) + for value in values.tolist() + ) + elif domain.get("kind") == "nonnegative_integer": + valid = all( + _finite_number(value) and value >= 0 and value == int(value) + for value in values.tolist() + ) + elif isinstance(domain.get("values"), (list, tuple)) and domain["values"]: + allowed = domain["values"] + valid = all( + any(type(value) is type(member) and value == member for member in allowed) + for value in values.tolist() + ) + else: + raise ValueError(f"Calibration input {name} has an unsupported domain.") + if not valid: + raise ValueError(f"Calibration input {name} is outside its declared domain.")