diff --git a/packages/microcosm-build/src/microcosm/build/country_spec.py b/packages/microcosm-build/src/microcosm/build/country_spec.py index c288d18ab3..894f554c40 100644 --- a/packages/microcosm-build/src/microcosm/build/country_spec.py +++ b/packages/microcosm-build/src/microcosm/build/country_spec.py @@ -131,6 +131,7 @@ "input_mass_parity", "ledger_compile_parity", "macro_realism", + "measurement_prefit", "nonconstant_columns", "nonnegative_columns", "parity", diff --git a/packages/microcosm-build/src/microcosm/build/reference_intervals.py b/packages/microcosm-build/src/microcosm/build/reference_intervals.py index 250be3fd64..76170f396e 100644 --- a/packages/microcosm-build/src/microcosm/build/reference_intervals.py +++ b/packages/microcosm-build/src/microcosm/build/reference_intervals.py @@ -46,6 +46,7 @@ import numpy as np +from microcosm.frame.concepts import SOURCE_HOUSEHOLD_ID_COLUMN from microcosm.graph.canonical import canonical_json __all__ = [ @@ -55,6 +56,7 @@ "IntervalPreparationError", "IntervalTable", "alternative_closure", + "band_support", "boundary_diagnostics", "classify_records", "frame_boundary_diagnostics", @@ -62,6 +64,8 @@ "interval_indicators", "interval_input_fingerprint", "interval_table", + "kish_ess", + "record_source_households", "scale_interval_document", "scale_interval_table", "source_interval_table", @@ -127,6 +131,11 @@ _DIAGNOSTIC_FIELDS = frozenset( {"boundary_window_cents", "alternative_positive_band_closure"} ) +#: Optional diagnostics fields. ``support_household_entity`` names the +#: household entity whose ``source_household_id`` the per-band support +#: diagnostics group contributions by; a table without it reports no support. +_OPTIONAL_DIAGNOSTIC_FIELDS = frozenset({"support_household_entity"}) +SUPPORT_SCHEMA_VERSION = 1 class IntervalPreparationError(ValueError): @@ -159,6 +168,7 @@ class IntervalTable: cuts: tuple[Fraction, ...] scaled: bool declaration: Mapping[str, object] + support_household_entity: str | None = None def interval_document_sha256(document: object) -> str: @@ -318,8 +328,14 @@ def interval_table(value: object) -> IntervalTable: if "description" in row: _text(row["description"], "Interval table description") diagnostics = _closed( - row["diagnostics"], _DIAGNOSTIC_FIELDS, frozenset(), "interval diagnostics" + row["diagnostics"], + _DIAGNOSTIC_FIELDS, + _OPTIONAL_DIAGNOSTIC_FIELDS, + "interval diagnostics", ) + support_household = diagnostics.get("support_household_entity") + if support_household is not None: + _text(support_household, "Interval support household entity") window = diagnostics["boundary_window_cents"] if type(window) is not int or window < 0: raise ValueError("boundary_window_cents must be a nonnegative integer.") @@ -427,6 +443,7 @@ def interval_table(value: object) -> IntervalTable: cuts=tuple(cuts), scaled=scaled, declaration=copy.deepcopy(row), + support_household_entity=support_household, ) if scaled: _validate_scaling(table, row) @@ -762,8 +779,15 @@ def boundary_diagnostics( *, weight_record_ids: Sequence[object], weight_kind: str, + source_households: Sequence[object] | None = None, ) -> dict[str, object]: - """Fixed-weight boundary sensitivity, computed with exact weight arithmetic.""" + """Fixed-weight boundary sensitivity, computed with exact weight arithmetic. + + ``source_households`` gives each record's source household, aligned with + ``record_ids``. A table that declares ``support_household_entity`` needs + it and gains per-band support (:func:`band_support`); a table that does + not declare it refuses it, so its report is unchanged. + """ order = _record_order(record_ids) if _record_order(weight_record_ids) != order: @@ -774,6 +798,11 @@ def boundary_diagnostics( amounts = list(amounts) if len(amounts) != len(order): raise ValueError("Diagnostic amounts must be record-aligned.") + if (table.support_household_entity is None) != (source_households is None): + raise ValueError( + f"Interval table {table.id!r} support diagnostics need source " + "households exactly when the table declares support_household_entity." + ) alternative = alternative_closure(table.closure) floored = _validated(table, amounts, universe) zero = Fraction(0) @@ -901,8 +930,19 @@ def floored_by_destination(closure): row["alternative"] = measures(other[index]) row["delta"] = delta(index) bands.append(row) + if source_households is not None: + support = band_support(table, floored, values.tolist(), source_households) + declared_bands = {band["id"]: band for band in table.declaration["bands"]} + for row in bands: + row["support"] = support[row["band"]] + band = declared_bands[row["band"]] + if "count_reference" in band: + row["references"] = { + "count": band["count_reference"], + "amount": band["amount_reference"], + } scaling = table.declaration.get("scaling") - return { + report = { "schema_version": 1, "kind": "interval_boundary_diagnostics", "table": table.id, @@ -952,6 +992,146 @@ def floored_by_destination(closure): "publisher_reclassification": PUBLISHER_RECLASSIFICATION, "limitation": LIMITATION, } + if source_households is not None: + report["support"] = { + "schema_version": SUPPORT_SCHEMA_VERSION, + "household_entity": table.support_household_entity, + "source_household_column": SOURCE_HOUSEHOLD_ID_COLUMN, + "grouping": "source_household", + } + return report + + +# --------------------------------------------------------------------------- +# Per-band support (G8-r2c hunk D) +# --------------------------------------------------------------------------- + + +def kish_ess(contributions: Sequence[Fraction]) -> Fraction: + """Kish effective sample size ``(sum c)^2 / sum c^2``, exactly; 0 when empty.""" + + total = sum(contributions, Fraction(0)) + squares = sum((value * value for value in contributions), Fraction(0)) + return Fraction(0) if squares == 0 else total * total / squares + + +def _support_measure(groups: Mapping[object, Fraction]) -> dict[str, object]: + contributions = [value for value in groups.values() if value != 0] + ess = kish_ess(contributions) + return { + "source_households": len(contributions), + "kish_ess": _rational(ess), + } + + +def band_support( + table: IntervalTable, + floored: Sequence[int | None], + weights: Sequence[float], + source_households: Sequence[object], +) -> dict[str, dict[str, object]]: + """Distinct source households and source-household Kish ESS per band. + + Each included record contributes ``w`` to its band's count and ``w * z`` + to its band's amount, where ``w`` is its weight as the diagnostics read + it (the frame's weights on the table's entity, the weights entering + calibration) and ``z`` is its floored amount in cents: an amount + contribution's weight is ``w * z``, the record's share of the band's + weighted amount. The + contributions are summed per source household first, so donor clones of + one household add no support, and only then enter the Kish ratio. A + household counts as support only where its summed contribution is + nonzero. Counts and amounts are reported separately. + """ + + households = list(source_households) + if len(households) != len(floored) or len(weights) != len(floored): + raise ValueError("Support diagnostics need record-aligned source households.") + if any( + isinstance(item, (bool, np.bool_)) or not isinstance(item, (str, Integral)) + for item in households + ): + raise ValueError("Source household identities are strings or integers.") + counts = {band: {} for band in table.band_ids} + amounts = {band: {} for band in table.band_ids} + for weight, z, household in zip(weights, floored, households, strict=True): + if z is None: + continue + position = _position(table, z, table.closure) + if position == BELOW_FIRST_CUT: + continue + band = table.band_ids[position] + key = household if isinstance(household, str) else int(household) + exact = Fraction(weight) + counts[band][key] = counts[band].get(key, Fraction(0)) + exact + amounts[band][key] = amounts[band].get(key, Fraction(0)) + exact * z + return { + band: { + "count": _support_measure(counts[band]), + "amount": _support_measure(amounts[band]), + } + for band in table.band_ids + } + + +def record_source_households(frame, entity: str, household_entity: str) -> list: + """Each ``entity`` record's source household, through household membership. + + The frame declares membership only through the person table: each group + entity has a person column ``EntitySchema.membership_column(group)`` + (``microcosm/frame/schema.py``), which ``Frame._group_positions`` reads + (``microcosm/frame/bundle.py``). A person reads its household column + directly. A group unit reads the households of its members, as + ``Frame._aggregate_group_to_group`` does when it places one group onto + another, and like it requires the unit to nest in one household: a unit + with no members, or members in several households, has no household + membership and refuses by name. So two units in one household map to + one source household and count once. The household table supplies + ``source_household_id``. + """ + + schema = frame.schema + refusal = ( + f"Interval support: entity {entity!r} has no household membership in " + f"{household_entity!r}" + ) + if household_entity not in schema.group_entities: + raise ValueError(f"{refusal}; the frame declares no such household entity.") + households = frame.table(household_entity) + if SOURCE_HOUSEHOLD_ID_COLUMN not in households: + raise ValueError( + f"Interval support needs {household_entity}." + f"{SOURCE_HOUSEHOLD_ID_COLUMN} on the frame." + ) + source = dict( + zip( + households[schema.id_column(household_entity)].tolist(), + households[SOURCE_HOUSEHOLD_ID_COLUMN].tolist(), + strict=True, + ) + ) + if entity == household_entity: + return [source[item] for item in households[schema.id_column(entity)]] + person = frame.table(schema.person_entity) + membership = person[schema.membership_column(household_entity)].tolist() + if entity == schema.person_entity: + return [source[item] for item in membership] + if entity not in schema.group_entities: + raise ValueError(f"{refusal}; it is not a declared entity.") + unit_households: dict[object, set] = {} + for unit, household in zip( + person[schema.membership_column(entity)].tolist(), membership, strict=True + ): + unit_households.setdefault(unit, set()).add(household) + result = [] + for unit in frame.table(entity)[schema.id_column(entity)].tolist(): + found = unit_households.get(unit, set()) + if len(found) != 1: + raise ValueError( + f"{refusal}; unit {unit!r} has members in {len(found)} households." + ) + result.append(source[next(iter(found))]) + return result def frame_boundary_diagnostics(table: IntervalTable, frame) -> dict[str, object]: @@ -973,4 +1153,9 @@ def frame_boundary_diagnostics(table: IntervalTable, frame) -> dict[str, object] weights.values, weight_record_ids=ids, weight_kind=weights.kind.value, + source_households=None + if table.support_household_entity is None + else record_source_households( + frame, table.entity, table.support_household_entity + ), ) diff --git a/packages/microcosm-build/src/microcosm/build/reference_profiles.py b/packages/microcosm-build/src/microcosm/build/reference_profiles.py index 5354a795bc..e42ef39769 100644 --- a/packages/microcosm-build/src/microcosm/build/reference_profiles.py +++ b/packages/microcosm-build/src/microcosm/build/reference_profiles.py @@ -69,9 +69,44 @@ def _selection_identity(receipt: Mapping) -> dict: identity["transformed_names"] = receipt["transformed_names"] if "interval_tables" in receipt: identity["interval_tables_sha256"] = receipt["interval_tables_sha256"] + if "measurement_proofs" in receipt: + identity["measurement_proofs"] = receipt["measurement_proofs"] return identity +def _measurement_proofs(proofs: Mapping | None) -> dict: + """Validate composition's proof map. + + Each proof names its measurement and, for every calibration node, the + one producer node upstream of it. + """ + if proofs is None: + return {} + proofs = _mapping(proofs, "measurement proofs") + for binding, proof in proofs.items(): + proof = _mapping(proof, "measurement proof") + producers = proof.get("producers") + if ( + not isinstance(binding, str) + or not binding + or proof.get("measurement_id") != binding + or not isinstance(producers, Mapping) + or not producers + or any( + not isinstance(key, str) + or not key + or not isinstance(value, str) + or not value + for key, value in producers.items() + ) + ): + raise ValueError( + "A measurement proof names its measurement and, for each " + "calibration node, one producer node." + ) + return _copy(proofs) + + def validate_reference_factor_evidence(receipt: Mapping) -> dict: """Bind selected declarations and receipts without claiming feed authenticity.""" from microcosm.build.reference_factors import validate_factor_context @@ -143,6 +178,20 @@ def _explicit_names(transform: Mapping, field: str) -> list[str]: return names +def _member_order(value: object) -> bytes: + return canonical_json(value) + + +def _canonical_pins(pins: object) -> object: + """Order a summed row's member lists, so a pair compares as member sets.""" + if not isinstance(pins, Mapping): + return pins + return { + key: sorted(value, key=_member_order) if isinstance(value, list) else value + for key, value in pins.items() + } + + def _companion_key(row: Mapping, *, scope: bool = False) -> bytes: selector = _mapping(row.get("ledger_selector"), "companion selector") excluded = {"aggregate_fact_key", "source_record_id"} @@ -152,7 +201,9 @@ def _companion_key(row: Mapping, *, scope: bool = False) -> bytes: excluded.add("dimension_values") payload = { "selector": { - key: value for key, value in selector.items() if key not in excluded + key: _canonical_pins(value) if key == "dimension_values" else value + for key, value in selector.items() + if key not in excluded }, "entity": row.get("entity"), "period": row.get("period"), @@ -163,6 +214,161 @@ def _companion_key(row: Mapping, *, scope: bool = False) -> bytes: return canonical_json(payload) +def _summed_members(row: Mapping) -> frozenset | None: + """The member cells of a summed row, or ``None`` for a singleton row. + + A summed row names no ``aggregate_fact_key``. Its selector pins at least + one dimension to a list of members, and its ``expected_member_count`` + equals the number of cells those lists span (a scalar pin spans one). + The ledger sums exactly those cells (``ledger_targets.py`` + ``_resolve_linear_combination_reference_facts`` for money and + ``_resolve_sum_reference_facts`` for counts), so a declared count that + disagrees with the lists would sum a different member set than the + reviewer read. + """ + selector = _mapping(row.get("ledger_selector"), "summed selector") + pins = selector.get("dimension_values") + listed = isinstance(pins, Mapping) and any( + isinstance(value, list) for value in pins.values() + ) + if "aggregate_fact_key" in selector: + if listed or "expected_member_count" in row: + raise ValueError( + f"Reference {row.get('name')!r} names a singleton fact key and " + "a member set; a summed row names members only." + ) + return None + if not listed: + return None + cells = [()] + for key in sorted(pins): + value = pins[key] + members = value if isinstance(value, list) else [value] + ordered = sorted(members, key=_member_order) + if any( + _member_order(item) == _member_order(previous) + for item, previous in zip(ordered[1:], ordered, strict=False) + ) or any(isinstance(item, (Mapping, list)) for item in members): + raise ValueError( + f"Reference {row.get('name')!r} lists dimension {key!r} members " + "that repeat or are not scalars." + ) + cells = [ + (*cell, (key, _member_order(item).decode())) + for cell in cells + for item in members + ] + count = row.get("expected_member_count") + if type(count) is not int or count != len(cells): + raise ValueError( + f"Reference {row.get('name')!r} lists {len(cells)} members but " + f"declares expected_member_count {count!r}." + ) + return frozenset(cells) + + +def _row_cells(row: Mapping) -> frozenset: + """The dimension cells one row selects, singleton or summed.""" + members = _summed_members(row) + if members is not None: + return members + pins = _mapping(row.get("ledger_selector"), "selector").get("dimension_values") + if not isinstance(pins, Mapping): + return frozenset() + return frozenset( + {tuple((key, _member_order(pins[key]).decode()) for key in sorted(pins))} + ) + + +def _validate_summed_rows(money: list, counts: list, by_name: Mapping) -> None: + """Pair summed money and count rows on one member set, without overlap. + + A summed money row is an ``identity`` row the resolver turns into one + weighted operand. Its count companion keeps ``sum`` semantics over the + same members: the ledger's ``identity`` path resolves a single fact, so + it cannot sum a member set. Within one selector scope no two rows of one + role may share a member, because the summed facts would then count twice. + """ + pairs = {_companion_key(by_name[name]): name for name in counts} + summed = set() + for name in money: + money_row = by_name[name] + members = _summed_members(money_row) + count_name = pairs.get(_companion_key(money_row)) + count_members = ( + None if count_name is None else _summed_members(by_name[count_name]) + ) + if members is None and count_members is None: + continue + # The companion key holds the canonical member lists, so a count row + # pairs with this money row only over the same member set. + if count_name is None: + raise ValueError( + f"Summed money reference {name!r} and its count companion " + "select different member sets: no count row sums its members." + ) + summed.update({name, count_name}) + count_row = by_name[count_name] + if count_row.get("value_operation") != "sum" or count_row.get( + "value_operands", [] + ): + raise ValueError( + f"Summed count reference {count_name!r} must sum its members " + "with value_operation 'sum'." + ) + for role in (money, counts): + if not summed & set(role): + continue + seen: dict[tuple, str] = {} + for name in role: + scope = _companion_key(by_name[name], scope=True) + for cell in _row_cells(by_name[name]): + other = seen.setdefault((scope, cell), name) + if other != name and summed & {name, other}: + raise ValueError( + f"References {other!r} and {name!r} share a summed member." + ) + + +def _money_operand(row: Mapping, weight: float) -> dict: + """The one weighted operand that scales a money row by its factor.""" + selector = _mapping(row.get("ledger_selector"), "money selector") + members = _summed_members(row) + if members is not None: + return { + "dimension_values": _copy(selector["dimension_values"]), + "expected_member_count": row["expected_member_count"], + "weight": weight, + } + key = selector.get("aggregate_fact_key") + if not isinstance(key, str) or not key: + raise ValueError( + "Money scaling needs a singleton source fact key or a summed member set." + ) + return {"aggregate_fact_key": key, "expected_member_count": 1, "weight": weight} + + +def _binary64_weight(rational: Mapping) -> dict: + """State how far the binary64 operand weight sits from the exact factor. + + Interval cuts stay exact rationals, while the ledger multiplies each + member value by a float weight. The record gives the float, the exact + factor and the exact relative rounding as a reduced fraction. + """ + from fractions import Fraction + + exact = Fraction(rational["numerator"], rational["denominator"]) + weight = rational["numerator"] / rational["denominator"] + relative = (Fraction(weight) - exact) / exact + return { + "weight": weight, + "exact_numerator": rational["numerator"], + "exact_denominator": rational["denominator"], + "binary64_relative_rounding": f"{relative.numerator}/{relative.denominator}", + "rounding": "round_half_even_binary64", + } + + def _validate_companions(transform: Mapping, by_name: Mapping) -> tuple[list, list]: money = _explicit_names(transform, "reference_names") counts = _explicit_names(transform, "count_reference_names") @@ -187,6 +393,7 @@ def _validate_companions(transform: Mapping, by_name: Mapping) -> tuple[list, li } if complete != set(names): raise ValueError("Money and count rosters are incomplete companions.") + _validate_summed_rows(money, counts, by_name) money_keys = [_companion_key(by_name[name]) for name in money] count_keys = [_companion_key(by_name[name]) for name in counts] if len(set(money_keys)) != len(money) or set(money_keys) != set(count_keys): @@ -224,6 +431,15 @@ def _factor_companions(chosen: Mapping, by_name: Mapping) -> dict: "value_operation", "identity" ), "value_operands": row.get("value_operands", []), + **( + { + "expected_member_count": row[ + "expected_member_count" + ] + } + if "expected_member_count" in row + else {} + ), "metadata": { "reference_value_kind": row["metadata"][ "reference_value_kind" @@ -419,7 +635,17 @@ def _validate_interval_resolution( raise ValueError("Effective interval table scaling identity mismatch.") -def _satisfy_factor_prerequisite(row: dict, dimension: str, binding: str) -> list: +def _satisfy_factor_prerequisite( + row: dict, dimension: str, binding: str, *, proven: bool = False +) -> list: + """Satisfy the profile half, and the measurement half only on a proof. + + ``proven`` is composition's result (``compose.measurement_proofs``): it + is true only when exactly one node upstream of every calibration node + declares the measurement and writes its interval columns. A proven row + becomes active; any other row stays parked with the measurement named in + ``reference_prerequisites_remaining``. + """ metadata = row.setdefault("metadata", {}) if metadata.get("activation_status") != REQUIRES_PREREQUISITES: raise ValueError( @@ -446,10 +672,15 @@ def _satisfy_factor_prerequisite(row: dict, dimension: str, binding: str) -> lis raise ValueError( "Reference prerequisites must name the profile and measurement." ) - remaining = [ - item for item in prerequisites if item != f"reference_profile:{dimension}" - ] + satisfied = {f"reference_profile:{dimension}"} + if proven: + satisfied.add(f"measurement:{binding}") + remaining = [item for item in prerequisites if item not in satisfied] metadata["reference_prerequisites_remaining"] = canonical_json(remaining).decode() + if not remaining: + metadata["activation_status"] = "active" + metadata.pop("activation_reason", None) + metadata.pop("parking_reason", None) return remaining @@ -460,6 +691,7 @@ def resolve_reference_profiles( interval_document: Mapping | None = None, *, factor_context: Mapping | None = None, + measurement_proofs: Mapping | None = None, ) -> tuple[dict, dict | None, dict]: """Return effective references, effective intervals and a bound receipt. @@ -470,7 +702,10 @@ def resolve_reference_profiles( scaled by the same selected factor as their money companions; without named tables the interval document passes through unchanged. Resource headers are bound into the base identity; effective references use - the compiler's closed document shape. + the compiler's closed document shape. ``measurement_proofs`` is + composition's proof result (``compose.measurement_proofs``), keyed by + measurement id: only a proven binding satisfies ``measurement:``, and + the proofs a selection uses enter its identity. """ base = _copy(_mapping(base_reference_document, "base references")) declarations = _copy(_mapping(profile_declarations, "profile declarations")) @@ -528,6 +763,17 @@ def resolve_reference_profiles( if not isinstance(factor_id, str) or not factor_id: raise ValueError("Money scaling needs a declared factor id.") factor_ids.add(factor_id) + proofs = _measurement_proofs(measurement_proofs) + bindings = { + transform.get("required_measurement_binding") + for choice in chosen.values() + for transform in choice["declaration"].get("transformations", []) + if isinstance(transform, Mapping) + and transform.get("operation") == SCALE_MONEY_AND_INTERVALS + } + used_proofs = { + binding: proofs[binding] for binding in sorted(bindings & set(proofs)) + } selected_factors = {} if factor_ids: from microcosm.build.reference_factors import validate_factor_context @@ -579,6 +825,12 @@ def resolve_reference_profiles( interval_tables=interval_records, interval_tables_sha256=reference_document_sha256(interval_records), ) + receipt["factor_weight_rounding"] = { + factor_id: _binary64_weight(entry["operational_factor"]) + for factor_id, entry in sorted(selected_factors.items()) + } + if used_proofs: + receipt["measurement_proofs"] = _copy(used_proofs) selection_hash = reference_document_sha256(_selection_identity(receipt)) interval_hash = None if intervals is None else reference_document_sha256(intervals) transformed, policies = {}, [] @@ -642,7 +894,9 @@ def resolve_reference_profiles( raise ValueError( "Factor and reference source/target periods are incompatible." ) - remaining = _satisfy_factor_prerequisite(row, dimension, binding) + remaining = _satisfy_factor_prerequisite( + row, dimension, binding, proven=binding in used_proofs + ) if name in money: if row.get( "value_operation", "identity" @@ -650,22 +904,8 @@ def resolve_reference_profiles( raise ValueError( "Money scaling requires an untransformed source reference." ) - selector = _mapping( - row.get("ledger_selector"), "money selector" - ) - key = selector.get("aggregate_fact_key") - if not isinstance(key, str) or not key: - raise ValueError( - "Money scaling needs a singleton source fact key." - ) row["value_operation"] = "linear_combination" - row["value_operands"] = [ - { - "aggregate_fact_key": key, - "expected_member_count": 1, - "weight": weight, - } - ] + row["value_operands"] = [_money_operand(row, weight)] row_metadata = row.setdefault("metadata", {}) row_metadata.update(metadata) row_metadata["reference_selection_sha256"] = selection_hash @@ -830,6 +1070,12 @@ def validate_reference_resolution( ) != resolution.get("effective_intervals_sha256"): raise ValueError("Effective interval document hash mismatch.") _validate_interval_resolution(resolution, factors, interval_document) + if factors and resolution.get("factor_weight_rounding") != { + factor_id: _binary64_weight(entry["operational_factor"]) + for factor_id, entry in sorted(factors.items()) + }: + raise ValueError("Factor weight rounding record mismatch.") + _measurement_proofs(resolution.get("measurement_proofs")) factor_bindings = _factor_reference_bindings(resolution) transformed_names = resolution.get("transformed_names") if not isinstance(transformed_names, list) or not all( @@ -906,15 +1152,23 @@ def validate_reference_resolution( or action.get("target_period") != declaration["target_period"] ): raise ValueError("Effective reference factor identity mismatch.") + proven = metadata.get("required_measurement_binding") in resolution.get( + "measurement_proofs", {} + ) + if (metadata.get("activation_status") == "active") is not proven: + raise ValueError( + "A factor companion is active exactly when composition " + "proved its measurement." + ) if role == "money": expected = [ - { - "aggregate_fact_key": source.get("ledger_selector", {}).get( - "aggregate_fact_key" - ), - "expected_member_count": 1, - "weight": rational["numerator"] / rational["denominator"], - } + _money_operand( + { + **source, + "name": name, + }, + rational["numerator"] / rational["denominator"], + ) ] if ( row.get("value_operation") != "linear_combination" diff --git a/packages/microcosm-build/src/microcosm/build/transport/compose.py b/packages/microcosm-build/src/microcosm/build/transport/compose.py index 0fef19eef3..e85c658fde 100644 --- a/packages/microcosm-build/src/microcosm/build/transport/compose.py +++ b/packages/microcosm-build/src/microcosm/build/transport/compose.py @@ -460,6 +460,319 @@ def _raw_transport_resource(spec: Mapping[str, object], name: str) -> Mapping: return _mapping(value, f"{stem}.json") +def _measurement_tables(spec: Mapping[str, object]) -> dict[str, list[Mapping]]: + """Each money-scaling measurement binding with the interval tables it scales. + + Every declared profile counts, selected or not, so a proof never depends + on which profile a scenario selects. + """ + declarations = _raw_transport_resource(spec, "reference_profiles") + interval = declarations.get("interval_resource") + tables = {} + if interval is not None and _has_resource(spec, interval): + tables = { + table.get("id"): table + for table in _raw_transport_resource(spec, interval).get("tables", ()) + if isinstance(table, Mapping) + } + found: dict[str, list[Mapping]] = {} + dimensions = declarations.get("dimensions", {}) + for dimension in dimensions.values() if isinstance(dimensions, Mapping) else (): + profiles = ( + dimension.get("profiles", {}) if isinstance(dimension, Mapping) else {} + ) + for profile in profiles.values() if isinstance(profiles, Mapping) else (): + for transform in ( + profile.get("transformations", ()) + if isinstance(profile, Mapping) + else () + ): + if ( + not isinstance(transform, Mapping) + or transform.get("operation") != "scale_money_and_intervals" + ): + continue + binding = transform.get("required_measurement_binding") + if not isinstance(binding, str) or not binding: + continue + listed = found.setdefault(binding, []) + for table_id in transform.get("interval_table_ids", ()): + table = tables.get(table_id) + if table is not None and table not in listed: + listed.append(table) + return found + + +MEASUREMENT_PARAM = "measurement_id" + + +def _measurement_declaration(row: Mapping) -> object: + """The ``measurement_id`` a node declares in its params, or ``None``. + + A producer declares the id as a literal string param, as + ``measure.combine@1`` reads ``context.params["measurement_id"]``. + """ + params = row.get("params") + return params.get(MEASUREMENT_PARAM) if isinstance(params, Mapping) else None + + +def _declared_calibration_predecessors( + rows: Mapping[str, Mapping], +) -> tuple[dict[str, set[str]], dict[str, set[tuple[str, str, str]]]]: + """Declared predecessors and reads, by the S5 ancestry guard's rule. + + A node follows its population version (or, if structural, its base and + every member of the base), each artifact producer it names, and each + writer :func:`_precalibration_declared_reads` resolves for its reads. + """ + predecessors = {name: set() for name in rows} + reads = _precalibration_declared_reads(rows) + for name, row in rows.items(): + base = row.get("base") + if isinstance(base, str): + predecessors[name].add(base) + predecessors[name].update( + item["id"] for item in rows.values() if item.get("population") == base + ) + elif isinstance(row.get("population"), str): + predecessors[name].add(row["population"]) + predecessors[name].update( + edge["producer"] for edge in row.get("artifact_inputs", ()) + ) + predecessors[name].update(producer for _, _, producer in reads[name]) + return predecessors, reads + + +def _calibration_roots(rows: Mapping[str, Mapping], document: Mapping) -> set[str]: + roots = { + name + for name, row in rows.items() + if isinstance(row.get("kernel"), str) and row["kernel"].startswith("calibrate.") + } + endpoint = (document.get("endpoints") or {}).get("calibration") + if isinstance(endpoint, str) and endpoint in rows: + roots.add(endpoint) + return roots + + +def _measurement_refusal( + binding: str, + root: str, + found: Sequence[str], + declared: Sequence[str], +) -> str: + return ( + f"{len(found)} nodes declaring measurement_id {binding!r} " + f"sit upstream of calibration node {root!r} " + f"(declared by {list(declared)}); exactly one must" + ) + + +def _second_writer_refusal( + binding: str, + root: str, + producer: str, + writers: Mapping[tuple[str, str], Sequence[str]], +) -> str | None: + """Why a measured column has a writer other than its producer, or ``None``. + + R2C-15: the producer is the only ancestor of each calibration node that + writes each measured column, wherever another writer sits and whether + or not it rewrites. An EXPAND writes the cells it declares. + """ + for (entity, column), names in sorted(writers.items(), key=str): + if list(names) != [producer]: + return ( + f"measurement {binding!r}: calibration node {root!r} has " + f"writers {list(names)} of {entity}.{column} upstream; " + f"only its producer {producer!r} may write it" + ) + return None + + +def _declared_writer(row: Mapping, entity: object, column: object) -> bool: + """Whether a declared node may write a cell, by output or as an EXPAND. + + An EXPAND whose cells only composition can tell writes every column, as + :func:`_precalibration_declared_reads` reads it. + """ + cells = _precalibration_cells(row) + return ( + (entity, column) in _owned_columns(row) + or cells is None + or (entity, column) in cells + ) + + +def measurement_proofs( + spec: Mapping[str, object], +) -> tuple[dict[str, dict], dict[str, str]]: + """Prove each money-scaling measurement's producer from the declaration. + + G8-r2c ruling R2C-1: composition satisfies ``measurement:`` only when, + for every calibration node, exactly one node upstream of it declares + ``params.measurement_id: `` (a string) and writes the income and + universe columns of every interval table the measurement's profiles + scale, on that table's entity. R2C-15 adds that the producer is the only + node upstream of that calibration node which writes each of those + columns. "Upstream" is ancestry in the declared + graph under the S5 guard's rule (:func:`_declared_calibration_predecessors`); + composition then checks the same producers against the compiled graph's + exact predecessors (:func:`_check_measurement_proofs`). Returns the + proofs by measurement id and, for each measurement without one, the + reason it stays parked. + """ + if not _has_resource(spec, "reference_profiles"): + return {}, {} + wanted = _measurement_tables(spec) + if not wanted: + return {}, {} + if _has_resource(spec, GRAPH_RESOURCE): + document = _raw_transport_resource(spec, GRAPH_RESOURCE) + rows = _precalibration_rows(document) + else: + document, rows = {}, {} + roots = _calibration_roots(rows, document) + predecessors, _ = _declared_calibration_predecessors(rows) + upstream = {root: _ancestor_closure(predecessors, {root}) for root in sorted(roots)} + malformed = sorted( + name + for name, row in rows.items() + if _measurement_declaration(row) is not None + and not _is_text(_measurement_declaration(row)) + ) + proofs: dict[str, dict] = {} + refusals: dict[str, str] = {} + for binding, tables in sorted(wanted.items()): + required = sorted( + { + (table.get("entity"), table.get(field)) + for table in tables + for field in ("income_column", "universe_column") + }, + key=str, + ) + declared = sorted( + name + for name, row in rows.items() + if _measurement_declaration(row) == binding + ) + producers, reason = {}, None + if malformed: + reason = f"nodes {malformed} declare a measurement_id that is not a string" + elif not tables: + reason = "its profiles scale no declared interval table" + elif not upstream: + reason = "the graph declares no calibration node" + for root, closure in upstream.items(): + if reason is not None: + break + found = [name for name in declared if name in closure] + if len(found) != 1: + reason = _measurement_refusal(binding, root, found, declared) + break + (producer,) = found + missing = [ + f"{entity}.{column}" + for entity, column in required + if (entity, column) not in _owned_columns(rows[producer]) + ] + if missing: + reason = ( + f"producer {producer!r} does not write {missing} on the " + "interval table's entity" + ) + break + reason = _second_writer_refusal( + binding, + root, + producer, + { + (entity, column): sorted( + name + for name in closure + if _declared_writer(rows[name], entity, column) + ) + for entity, column in required + }, + ) + if reason is not None: + break + producers[root] = producer + if reason is None: + proofs[binding] = { + "measurement_id": binding, + "producers": producers, + "columns": [f"{entity}.{column}" for entity, column in required], + } + else: + refusals[binding] = reason + return proofs, refusals + + +def _check_measurement_proofs( + spec: Mapping[str, object], graph: Graph, compiled: CompiledGraph +) -> None: + """Refuse a declared proof the compiled graph's ancestry contradicts. + + For each proved measurement and each calibration node of the composed + graph, the compiled ancestors of that node must hold exactly one node + whose params declare the measurement, and it must be the proof's + producer for that node. That producer must also be the only compiled + ancestor whose outputs or EXPAND cells hold each of the proof's columns + (R2C-15). + """ + proofs, _ = measurement_proofs(spec) + if not proofs: + return + roots = {node.id for node in graph.nodes if node.kernel.startswith("calibrate.")} + endpoint = transport_endpoints(spec).get("calibration") + if isinstance(endpoint, str): + roots.add(endpoint) + declared = { + node.id: node.params.get(MEASUREMENT_PARAM) + for node in graph.nodes + if isinstance(node.params, Mapping) and MEASUREMENT_PARAM in node.params + } + by_id = {node.id: node for node in graph.nodes} + _, expand_cells = _precalibration_expansions(graph) + for binding, proof in sorted(proofs.items()): + named = sorted(name for name, value in declared.items() if value == binding) + for root in sorted(roots): + closure = _ancestor_closure(compiled.predecessors, {root}) + found = [name for name in named if name in closure] + if found != [proof["producers"].get(root)]: + raise ValueError( + f"Measurement {binding!r}: the compiled graph contradicts " + "the declared proof; " + + _measurement_refusal(binding, root, found, named) + + f", and the declaration proved {proof['producers'].get(root)!r}." + ) + columns = [tuple(name.split(".", 1)) for name in proof["columns"]] + refusal = _second_writer_refusal( + binding, + root, + proof["producers"][root], + { + (entity, column): sorted( + name + for name in closure + if (entity, column) in expand_cells[name] + or any( + (output.entity, output.column) == (entity, column) + for output in by_id[name].outputs + ) + ) + for entity, column in columns + }, + ) + if refusal is not None: + raise ValueError( + f"Measurement {binding!r}: the compiled graph contradicts " + f"the declared proof; {refusal}." + ) + + def reference_profile_view( spec: Mapping[str, object], selection: Mapping | None = None ): @@ -488,6 +801,7 @@ def reference_profile_view( factor_context=None if factor_context_resource is None else _raw_transport_resource(spec, factor_context_resource), + measurement_proofs=measurement_proofs(spec)[0], ) if _has_resource(spec, CHRONICLE_FEED_RESOURCE): feed_pin = _raw_transport_resource(spec, CHRONICLE_FEED_RESOURCE) @@ -644,8 +958,10 @@ def _reference_activation_gaps( resource: str, exempt: Mapping | None = None, selected: Mapping[str, Sequence[str]] | None = None, + refusals: Mapping[str, str] | None = None, ) -> Iterator[str]: # The absent-resource path preserves the legacy activation walk and messages. + # A row parked on an unproved measurement names it and the proof's reason. exemptions = ( set() if exempt is None @@ -665,7 +981,24 @@ def _reference_activation_gaps( reference.get("name") not in exemptions and metadata.get("activation_status", "active") != "active" ): - yield f"placeholder reference {resource}.{reference.get('name', '?')}" + gap = f"placeholder reference {resource}.{reference.get('name', '?')}" + try: + remaining = json.loads( + metadata.get("reference_prerequisites_remaining", "[]") + ) + except (TypeError, ValueError): + remaining = [] + measurements = [ + item.removeprefix("measurement:") + for item in remaining + if isinstance(item, str) and item.startswith("measurement:") + ] + if measurements: + gap += "; awaits " + ", ".join( + f"measurement:{item} ({(refusals or {}).get(item, 'no proof')})" + for item in measurements + ) + yield gap def _reference_selections(spec: Mapping[str, object]) -> dict[str, dict]: @@ -1112,6 +1445,7 @@ def validate_transport_activation(spec: Mapping[str, object]) -> None: missing = [] # Resolve first so activation never inspects an executable profile's raw rows. reference_profile_views(spec) + _, measurement_refusals = measurement_proofs(spec) reference_selections = _reference_selections(spec) selection_closure = ( _reference_selection_closure(spec) @@ -1211,6 +1545,7 @@ def validate_transport_activation(spec: Mapping[str, object]) -> None: resource, reference_selections.get(resource), selection_closure.get(resource), + measurement_refusals, ) ) for (resource, identity), document in selected_reference_views.items(): @@ -1221,6 +1556,7 @@ def validate_transport_activation(spec: Mapping[str, object]) -> None: resource, reference_selections.get(resource), selection_closure.get(resource), + measurement_refusals, ) ) receipt = ( @@ -3021,21 +3357,7 @@ def _validate_precalibration_ancestry( "name a node." ) roots.add(endpoints["calibration"]) - predecessors = {name: set() for name in rows} - reads = _precalibration_declared_reads(rows) - for name, row in rows.items(): - base = row.get("base") - if isinstance(base, str): - predecessors[name].add(base) - predecessors[name].update( - item["id"] for item in rows.values() if item.get("population") == base - ) - elif isinstance(row.get("population"), str): - predecessors[name].add(row["population"]) - predecessors[name].update( - edge["producer"] for edge in row.get("artifact_inputs", ()) - ) - predecessors[name].update(producer for _, _, producer in reads[name]) + predecessors, reads = _declared_calibration_predecessors(rows) declarations = rows @@ -4041,6 +4363,7 @@ def compose_transport_graph( ) compiled = compile_graph(skeleton) validate_transport_calibration_ancestry(spec, graph=skeleton, config=config) + _check_measurement_proofs(spec, skeleton, compiled) # Money ownership is checked after each graph pass of the ancestry guard. _check_single_money_normalization(spec, skeleton, compiled) _check_money_bridge_owners(skeleton, compiled) diff --git a/packages/microcosm-build/src/microcosm/build/transport/gate_bindings.py b/packages/microcosm-build/src/microcosm/build/transport/gate_bindings.py index 5e594de1f2..91b08a24f8 100644 --- a/packages/microcosm-build/src/microcosm/build/transport/gate_bindings.py +++ b/packages/microcosm-build/src/microcosm/build/transport/gate_bindings.py @@ -43,6 +43,10 @@ ``gates.json``. - ``weight_ess`` and ``weight_ratio``: the frame's one weighted entity. +``measurement_prefit`` reads ``problem`` and the battery's reserved +``graph_evidence`` map: the problem's interval diagnostics with their +per-band support, and its matrix at the weights entering calibration. + ``release_input_coverage`` reads ``donor_artifact_receipt`` (``transport.donor_receipt@1``) and ``axiom_input_closure`` (``transport.input_closure@1``) with the battery's reserved @@ -1629,6 +1633,206 @@ def _transport_weights_audit( return _with_failures(result, failures=failures, details=details) +# --------------------------------------------------------------------------- +# Pre-fit measurement acceptance (G8-r2c hunk D, ruling R2C-3) +# --------------------------------------------------------------------------- + + +def _ratio_bounds(name: str, value: object) -> tuple[Fraction, Fraction]: + if isinstance(value, str | bytes) or not isinstance(value, Sequence): + raise TypeError(f"{name} must be a [lower, upper] pair.") + # Exactly two items, counted without a numeric literal: this module + # holds no threshold literal (test_holds_no_threshold_literal). + *rest, last = value or (None,) + if len(rest) != 1: + raise ValueError(f"{name} must be a [lower, upper] pair.") + lower, upper = (Fraction(_number(name, item)) for item in (*rest, last)) + if not 0 <= lower <= upper: + raise ValueError(f"{name} must satisfy 0 <= lower <= upper.") + return lower, upper + + +def _inside(model: Fraction, reference: Fraction, bounds) -> bool: + lower, upper = bounds + return reference > 0 and lower * reference <= model <= upper * reference + + +def _ratio_text(model: Fraction, reference: Fraction) -> str: + return "undefined" if reference == 0 else f"{float(model / reference):.6g}" + + +def _measurement_prefit( + *, + problem: object, + graph_evidence: object, + tables: Sequence[str], + roster_total_bounds: Sequence[float], + band_bounds: Sequence[float], + maximum_discrepant_bands: int, + minimum_source_households: int, + minimum_kish_ess: float, + required_evidence: Sequence[str], +) -> GateResult: + """Pre-fit acceptance of a measurement's interval targets (ruling R2C-3). + + Each comparator is the weighted model total at the weights entering + calibration (the problem's ``A @ w0``) against the reference the matrix + uses (its right-hand side), for every band's count and amount rows of + each declared interval table. The gate refuses on four checks: + + - roster totals: the summed count, or the summed amount, outside + ``roster_total_bounds`` of the summed references; + - concentrated discrepancy: more than ``maximum_discrepant_bands`` bands + whose count or amount falls outside ``band_bounds`` (a band counts + once if either fails); + - support: any band with fewer than ``minimum_source_households`` + distinct source households, or a source-household Kish ESS below + ``minimum_kish_ess``, checked for count and amount contributions + separately (``reference_intervals.band_support``); + - evidence: no interval diagnostics or per-band support for a declared + table, a band whose references did not enter the matrix, or a + ``required_evidence`` alias the battery node does not wire. + + Every threshold comes from ``gates.json``; none has a default. + """ + + declared_tables = _strings("tables", tables) + totals_bounds = _ratio_bounds("roster_total_bounds", roster_total_bounds) + per_band = _ratio_bounds("band_bounds", band_bounds) + if isinstance(maximum_discrepant_bands, bool) or not isinstance( + maximum_discrepant_bands, int + ): + raise TypeError("maximum_discrepant_bands must be an integer.") + if isinstance(minimum_source_households, bool) or not isinstance( + minimum_source_households, int + ): + raise TypeError("minimum_source_households must be an integer.") + minimum_ess = Fraction(_number("minimum_kish_ess", minimum_kish_ess)) + evidence = _strings("required_evidence", required_evidence) + failures: list[str] = [] + wired = graph_evidence if isinstance(graph_evidence, Mapping) else {} + for alias in evidence: + if alias not in wired: + failures.append(f"evidence: the run lacks required evidence {alias!r}.") + report = problem.bindings.get("interval_diagnostics") + reports = ( + {row.get("table"): row for row in report.get("tables", ())} + if isinstance(report, Mapping) + else {} + ) + names = [str(row.get("reference")) for row in problem.target_metadata] + rows = {name: index for index, name in enumerate(names)} + problem_rows = problem.problem + estimates = problem_rows.estimates(problem_rows.initial_weights.values) + targets = problem_rows.target_vector + band_details: list[dict[str, object]] = [] + sums = {"count": [Fraction(0), Fraction(0)], "amount": [Fraction(0), Fraction(0)]} + discrepant: list[str] = [] + for table_id in declared_tables: + table = reports.get(table_id) + if table is None: + failures.append( + f"evidence: the problem has no interval diagnostics for {table_id!r}." + ) + continue + if not isinstance(table.get("support"), Mapping): + failures.append( + f"evidence: interval table {table_id!r} reports no per-band support." + ) + continue + for band in table.get("bands", ()): + label = f"{table_id}/{band.get('band')}" + references = band.get("references") + support = band.get("support") + if not isinstance(references, Mapping) or not isinstance(support, Mapping): + failures.append( + f"evidence: band {label} names no count and amount references." + ) + continue + detail: dict[str, object] = {"band": label} + band_fails = False + for role in ("count", "amount"): + name = references.get(role) + if name not in rows: + failures.append( + f"evidence: band {label} {role} reference {name!r} did not " + "enter the matrix." + ) + band_fails = None + continue + model = Fraction(float(estimates[rows[name]])) + reference = Fraction(float(targets[rows[name]])) + sums[role][0] += model + sums[role][1] += reference + inside = _inside(model, reference, per_band) + if not inside and band_fails is not None: + band_fails = True + measured = support.get(role, {}) + households = measured.get("source_households") + ess = measured.get("kish_ess") + if not isinstance(households, int) or not isinstance(ess, Mapping): + failures.append(f"evidence: band {label} has no {role} support.") + else: + value = Fraction(ess["numerator"], ess["denominator"]) + if households < minimum_source_households: + failures.append( + f"support: band {label} {role} has {households} source " + f"households, below {minimum_source_households}." + ) + if value < minimum_ess: + failures.append( + f"support: band {label} {role} has source-household Kish " + f"ESS {float(value):.6g}, below {float(minimum_ess):.6g}." + ) + detail[role] = { + "reference": name, + "model": float(model), + "comparator": float(reference), + "ratio": _ratio_text(model, reference), + "inside_band_bounds": inside, + "source_households": households, + "kish_ess": None + if not isinstance(ess, Mapping) + else float(Fraction(ess["numerator"], ess["denominator"])), + } + if band_fails: + discrepant.append(label) + band_details.append(detail) + for role in ("count", "amount"): + model, reference = sums[role] + if band_details and not _inside(model, reference, totals_bounds): + failures.append( + f"roster totals: the summed {role} is " + f"{_ratio_text(model, reference)} of its comparator, outside " + f"[{float(totals_bounds[0]):.6g}, {float(totals_bounds[1]):.6g}]." + ) + if len(discrepant) > maximum_discrepant_bands: + failures.append( + f"concentrated discrepancy: {len(discrepant)} bands fall outside " + f"[{float(per_band[0]):.6g}, {float(per_band[1]):.6g}] " + f"({discrepant}); at most {maximum_discrepant_bands} may." + ) + return GateResult( + name="measurement_prefit", + passed=not failures, + failures=tuple(failures), + details={ + "tables": list(declared_tables), + "bands": band_details, + "roster_totals": { + role: { + "model": float(sums[role][0]), + "comparator": float(sums[role][1]), + "ratio": _ratio_text(*sums[role]), + } + for role in ("count", "amount") + }, + "discrepant_bands": discrepant, + "required_evidence": list(evidence), + }, + ) + + # --------------------------------------------------------------------------- # Evidence payloads (the report's evidence_sha256 lines) # --------------------------------------------------------------------------- @@ -1693,6 +1897,12 @@ def _weights_audit_evidence( } +def _measurement_prefit_evidence( + context: EvidenceContext, parameters: Mapping[str, Any] +) -> object: + return {"problem_sha256": context.artifacts[PROBLEM].sha256} + + def _cell_diagnostics_evidence( context: EvidenceContext, parameters: Mapping[str, Any] ) -> object: @@ -1831,6 +2041,15 @@ def _parameters(gate: object) -> frozenset[str]: frame_argument="frame", evidence=_support_evidence, ), + "measurement_prefit": FunctionBinding( + name="measurement_prefit", + gate=_measurement_prefit, + parameter_keys=_parameters(_measurement_prefit), + artifact_arguments=MappingProxyType( + {PROBLEM: PROBLEM, GRAPH_EVIDENCE: GRAPH_EVIDENCE} + ), + evidence=_measurement_prefit_evidence, + ), "release_input_coverage": FunctionBinding( name="release_input_coverage", gate=_release_input_coverage, diff --git a/packages/microcosm-build/tests/engine_free/shared/test_reference_interval_support.py b/packages/microcosm-build/tests/engine_free/shared/test_reference_interval_support.py new file mode 100644 index 0000000000..07d00af6d5 --- /dev/null +++ b/packages/microcosm-build/tests/engine_free/shared/test_reference_interval_support.py @@ -0,0 +1,386 @@ +"""Per-band source-household support in interval diagnostics (G8-r2c hunk D). + +A table that declares ``diagnostics.support_household_entity`` reports, per +band, the distinct source households and the source-household Kish ESS of +its count and amount contributions. Contributions are summed per source +household first, so donor clones add no support. A table without the field +parses, classifies and reports exactly as before. +""" + +from __future__ import annotations + +import copy +from fractions import Fraction + +import numpy as np +import pandas as pd +import pytest +from hypothesis import HealthCheck, given, settings +from hypothesis import strategies as st + +from microcosm.build.reference_intervals import ( + band_support, + boundary_diagnostics, + classify_records, + frame_boundary_diagnostics, + interval_indicators, + interval_table, + kish_ess, +) +from microcosm.frame import EntitySchema, Frame, WeightKind, Weights +from test_support.microcosm_build.reference_intervals import interval_table_fixture + +_SETTINGS = settings( + max_examples=60, + database=None, + deadline=None, + suppress_health_check=[HealthCheck.too_slow], +) + + +def _declaration(entity="person", support="household", **kwargs): + declaration = interval_table_fixture(entity=entity, companions=True, **kwargs) + if support is not None: + declaration["diagnostics"]["support_household_entity"] = support + return declaration + + +def _rational(value): + return Fraction(value["numerator"], value["denominator"]) + + +def test_a_table_without_support_fields_parses_and_reports_as_before(): + legacy = _declaration(support=None) + supported = _declaration() + old, new = interval_table(legacy), interval_table(supported) + assert old.support_household_entity is None + assert new.support_household_entity == "household" + amounts = [0, 50_000, 100_000, 100_001, 2_000_000, -5] + universe = ["included"] * 5 + ["excluded"] + assert classify_records(old, amounts, universe) == classify_records( + new, amounts, universe + ) + weights = [1.0, 2.0, 0.5, 4.0, 3.0, 1.0] + ids = list(range(6)) + report = boundary_diagnostics( + old, + ids, + amounts, + universe, + weights, + weight_record_ids=ids, + weight_kind="design", + ) + assert "support" not in report + assert all( + "support" not in row and "references" not in row for row in report["bands"] + ) + assert set(report) == { + "schema_version", + "kind", + "table", + "entity", + "currency", + "input_unit", + "positive_band_closure", + "alternative_positive_band_closure", + "nil_band", + "negative_income_policy", + "boundary_window_cents", + "weight_kind", + "records", + "included_records", + "digests", + "cuts", + "bands", + "below_first_cut", + "floored_negative_income", + "publisher_reclassification", + "limitation", + } + with pytest.raises(ValueError, match="support_household_entity"): + boundary_diagnostics( + old, + ids, + amounts, + universe, + weights, + weight_record_ids=ids, + weight_kind="design", + source_households=ids, + ) + supported_report = boundary_diagnostics( + new, + ids, + amounts, + universe, + weights, + weight_record_ids=ids, + weight_kind="design", + source_households=["h0", "h0", "h1", "h2", "h3", "h4"], + ) + for before, after in zip(report["bands"], supported_report["bands"], strict=True): + assert {key: after[key] for key in before} == before + with pytest.raises(ValueError, match="support_household_entity"): + boundary_diagnostics( + new, ids, amounts, universe, weights, weight_record_ids=ids, weight_kind="d" + ) + + +def _tax_unit_frame( + household_of_unit=(0, 0, 1), + sources=("s0", "s1"), + weights=(4.0, 6.0), + incomes=(50_000, 60_000, 70_000), +): + persons = [] + for unit, household in enumerate(household_of_unit): + persons.append((len(persons), household, unit)) + return Frame( + { + "person": pd.DataFrame( + { + "person_id": [row[0] for row in persons], + "person_household_id": [row[1] for row in persons], + "person_tax_unit_id": [row[2] for row in persons], + } + ), + "household": pd.DataFrame( + { + "household_id": list(range(len(sources))), + "source_household_id": list(sources), + } + ), + "tax_unit": pd.DataFrame( + { + "tax_unit_id": list(range(len(household_of_unit))), + "unit_income": list(incomes), + "unit_in": ["included"] * len(household_of_unit), + } + ), + }, + EntitySchema(group_entities=("household", "tax_unit")), + {"household": Weights(np.asarray(weights), WeightKind.DESIGN)}, + ) + + +def _tax_unit_table(support="household"): + declaration = _declaration(entity="tax_unit", support=support) + declaration.update(income_column="unit_income", universe_column="unit_in") + return interval_table(declaration) + + +def test_two_tax_units_in_one_household_count_once_with_source_household_ess(): + report = frame_boundary_diagnostics(_tax_unit_table(), _tax_unit_frame()) + bands = {row["band"]: row for row in report["bands"]} + support = bands["b0"]["support"] + # Units 0 and 1 share household 0 (weight 4 each); unit 2 is household 1 + # (weight 6). Count contributions per source household: 8 and 6. + assert support["count"]["source_households"] == 2 + assert _rational(support["count"]["kish_ess"]) == Fraction(14 * 14, 64 + 36) + amount = [Fraction(4 * 50_000 + 4 * 60_000), Fraction(6 * 70_000)] + assert _rational(support["amount"]["kish_ess"]) == kish_ess(amount) + assert bands["b0"]["references"] == { + "count": "toy_count_b0", + "amount": "toy_money_b0", + } + assert report["support"]["household_entity"] == "household" + + +def test_a_zero_weight_household_is_not_counted(): + """The hand case plus household 2 at weight 0: support is unchanged.""" + frame = _tax_unit_frame( + household_of_unit=(0, 0, 1, 2), + sources=("s0", "s1", "s2"), + weights=(4.0, 6.0, 0.0), + incomes=(50_000, 60_000, 70_000, 70_000), + ) + report = frame_boundary_diagnostics(_tax_unit_table(), frame) + support = {row["band"]: row for row in report["bands"]}["b0"]["support"] + assert support["count"]["source_households"] == 2 + assert _rational(support["count"]["kish_ess"]) == Fraction(14 * 14, 64 + 36) + assert support["amount"]["source_households"] == 2 + amount = [Fraction(4 * 50_000 + 4 * 60_000), Fraction(6 * 70_000)] + assert _rational(support["amount"]["kish_ess"]) == kish_ess(amount) + + +def test_clone_households_share_one_source_household(): + frame = _tax_unit_frame(sources=("s0", "s0")) + report = frame_boundary_diagnostics(_tax_unit_table(), frame) + support = {row["band"]: row for row in report["bands"]}["b0"]["support"] + assert support["count"]["source_households"] == 1 + assert _rational(support["count"]["kish_ess"]) == 1 + + +@pytest.mark.parametrize( + ("frame", "table", "message"), + [ + ( + _tax_unit_frame(household_of_unit=(0, 1, 1)), + None, + None, + ), + (None, "missing_household", "no household membership in 'missing_household'"), + ], +) +def test_an_entity_without_household_membership_refuses_by_name(frame, table, message): + if frame is not None: + # A unit whose members sit in two households has no one household. + persons = frame.table("person").copy() + persons.loc[len(persons)] = [3, 1, 0] + frame = Frame( + { + "person": persons, + "household": frame.table("household"), + "tax_unit": frame.table("tax_unit"), + }, + frame.schema, + # Explicit unit weights, so the frame derives none from members + # that span two households. + { + "household": Weights(np.asarray([4.0, 6.0]), WeightKind.DESIGN), + "tax_unit": Weights(np.asarray([4.0, 6.0, 6.0]), WeightKind.DESIGN), + }, + ) + with pytest.raises( + ValueError, + match="entity 'tax_unit' has no household membership in 'household'; " + "unit 0 has members in 2 households", + ): + frame_boundary_diagnostics(_tax_unit_table(), frame) + return + with pytest.raises(ValueError, match=message): + frame_boundary_diagnostics(_tax_unit_table(table), _tax_unit_frame()) + + +def test_support_needs_the_source_household_column(): + frame = _tax_unit_frame() + stripped = Frame( + { + "person": frame.table("person"), + "household": frame.table("household").drop(columns="source_household_id"), + "tax_unit": frame.table("tax_unit"), + }, + frame.schema, + {"household": Weights(np.asarray([4.0, 6.0]), WeightKind.DESIGN)}, + ) + with pytest.raises(ValueError, match="household.source_household_id"): + frame_boundary_diagnostics(_tax_unit_table(), stripped) + + +_records = st.lists( + st.tuples( + st.integers(min_value=0, max_value=6), + st.integers(min_value=1, max_value=64), + st.integers(min_value=0, max_value=3_000_000), + ), + min_size=1, + max_size=40, +) + + +@_SETTINGS +@given(records=_records, clones=st.integers(min_value=2, max_value=4)) +def test_ess_is_at_most_distinct_households_and_clones_leave_it_unchanged( + records, clones +): + """Records draw from 7 source households over 3 positive bands and a nil + band, so the strategy reaches shared households, clones and empty bands. + """ + table = interval_table(_declaration()) + floored = [amount for _, _, amount in records] + # Each original record carries clones x w; each clone carries w exactly. + weights = [float(weight * clones) for _, weight, _ in records] + sources = [f"s{source}" for source, _, _ in records] + support = band_support(table, floored, weights, sources) + for band in support.values(): + for role in ("count", "amount"): + measured = band[role] + assert _rational(measured["kish_ess"]) <= measured["source_households"] + if measured["source_households"] == 0: + assert _rational(measured["kish_ess"]) == 0 + split = [ + (source, float(weight), amount) + for source, (_, weight, amount) in zip(sources, records, strict=True) + for _ in range(clones) + ] + cloned = band_support( + table, + [row[2] for row in split], + [row[1] for row in split], + [row[0] for row in split], + ) + assert cloned == support + + +def test_an_empty_band_reports_no_support(): + table = interval_table(_declaration()) + support = band_support(table, [50_000], [2.0], ["s0"]) + assert support["b0"]["count"] == { + "source_households": 1, + "kish_ess": {"numerator": 1, "denominator": 1}, + } + for band in ("nil", "b1", "b2"): + for role in ("count", "amount"): + assert support[band][role] == { + "source_households": 0, + "kish_ess": {"numerator": 0, "denominator": 1}, + } + + +@_SETTINGS +@given(records=_records) +def test_weighted_band_counts_equal_weighted_indicator_sums(records): + table = interval_table(_declaration()) + amounts = [amount for _, _, amount in records] + weights = [float(weight) / 8 for _, weight, _ in records] + universe = ["included"] * len(records) + ids = list(range(len(records))) + report = boundary_diagnostics( + table, + ids, + amounts, + universe, + weights, + weight_record_ids=ids, + weight_kind="design", + source_households=[f"s{source}" for source, _, _ in records], + ) + indicators = interval_indicators(table, amounts, universe) + for band, indicator in zip(table.band_ids, table.indicators, strict=True): + row = next(item for item in report["bands"] if item["band"] == band) + expected = sum( + ( + Fraction(weight) + for weight, flag in zip(weights, indicators[indicator], strict=True) + if flag + ), + Fraction(0), + ) + assert _rational(row["declared"]["weighted_count"]) == expected + households = { + source + for (source, _, _), flag in zip(records, indicators[indicator], strict=True) + if flag + } + assert row["support"]["count"]["source_households"] == len(households) + + +def test_kish_ess_of_equal_contributions_is_their_number(): + assert kish_ess([Fraction(3)] * 7) == 7 + assert kish_ess([]) == 0 + assert kish_ess([Fraction(1), Fraction(3)]) == Fraction(16, 10) + + +def test_scaled_tables_keep_the_support_field(): + from microcosm.build.reference_intervals import scale_interval_table + + declaration = copy.deepcopy(_declaration()) + scaled = scale_interval_table( + declaration, + {"numerator": 11, "denominator": 10}, + factor_id="f", + factor_declaration_sha256="a" * 64, + factor_receipt_sha256="b" * 64, + ) + assert interval_table(scaled).support_household_entity == "household" diff --git a/packages/microcosm-build/tests/engine_free/shared/test_reference_summed_bands.py b/packages/microcosm-build/tests/engine_free/shared/test_reference_summed_bands.py new file mode 100644 index 0000000000..9302b0bb85 --- /dev/null +++ b/packages/microcosm-build/tests/engine_free/shared/test_reference_summed_bands.py @@ -0,0 +1,353 @@ +"""Summed-band money rows in ``scale_money_and_intervals`` (G8-r2c hunk A). + +A coarse money row sums several published bands: the resolver turns its +``identity`` row into one weighted operand over the member set, and its count +companion keeps ``sum`` semantics over the same members. Compilation runs the +real ledger, so these tests check the arithmetic the matrix receives. +""" + +from __future__ import annotations + +import copy +import json +import math +from fractions import Fraction + +import pytest +from hypothesis import HealthCheck, given, settings +from hypothesis import strategies as st + +from microcosm.build.country_spec import _validate_target_references +from microcosm.build.ledger_targets import compile_ledger_target_references +from microcosm.build.reference_profiles import ( + reference_document_sha256, + resolve_reference_profiles, + validate_reference_resolution, +) +from test_support.microcosm_build.reference_intervals import ( + SUMMED_MEASUREMENT, + band_fact, + proof_for, + summed_profile_fixture, + summed_row, + with_target_labels, +) + +TEN = [f"n{index:02d}" for index in range(10)] + + +def _facts(labels, *, count=lambda i: 100 + i, money=lambda i: 1_000 * (i + 1)): + facts = [] + for index, label in enumerate(labels): + facts.append(band_fact("count", label, count(index))) + facts.append(band_fact("money", label, money(index))) + return facts + + +def _resolve(inputs, proofs=None, selection=None): + base, profiles, context, intervals = inputs + return resolve_reference_profiles( + base, + profiles, + selection or {}, + intervals, + factor_context=context, + measurement_proofs=proofs, + ) + + +def _compile(effective, facts): + references = _validate_target_references(effective, country=effective["country"]) + registry = compile_ledger_target_references( + facts, references, country=effective["country"] + ) + return {spec.name: spec for spec in registry.specs} + + +def _rows(document): + return {row["name"]: row for row in document["target_references"]} + + +def test_ten_member_pair_resolves_to_one_operand_with_count_ten(): + inputs = summed_profile_fixture([TEN]) + effective, _, receipt = _resolve(inputs, proof_for()) + rows = _rows(effective) + money, count = rows["toy_money_b0"], rows["toy_count_b0"] + assert money["value_operation"] == "linear_combination" + assert money["value_operands"] == [ + { + "dimension_values": {"band": TEN}, + "expected_member_count": 10, + "weight": 1.1, + } + ] + assert count["value_operation"] == "sum" + assert count["expected_member_count"] == 10 + assert "value_operands" not in count or count["value_operands"] == [] + for row in (money, count): + assert row["metadata"]["activation_status"] == "active" + assert json.loads(row["metadata"]["reference_prerequisites_remaining"]) == [] + assert "parking_reason" not in row["metadata"] + assert receipt["measurement_proofs"] == proof_for() + assert validate_reference_resolution(effective, receipt) == receipt + + +def test_summed_money_operand_resolves_amount_facts_only(): + """Counts and amounts share one record set and one band dimension.""" + inputs = summed_profile_fixture([TEN]) + effective, _, _ = _resolve(inputs, proof_for()) + facts = _facts(TEN) + specs = _compile(effective, facts) + money_keys = json.loads(specs["toy_money_b0"].metadata["ledger_member_fact_keys"]) + count_keys = json.loads(specs["toy_count_b0"].metadata["ledger_member_fact_keys"]) + assert sorted(money_keys) == sorted( + f"ledger.aggregate_fact.v2:toy-money-{label}" for label in TEN + ) + assert sorted(count_keys) == sorted( + f"ledger.aggregate_fact.v2:toy-count-{label}" for label in TEN + ) + assert specs["toy_count_b0"].value == sum(100 + i for i in range(10)) + assert math.isclose( + specs["toy_money_b0"].value, + 1.1 * sum(1_000 * (i + 1) for i in range(10)), + rel_tol=1e-15, + ) + + +def test_without_a_proof_rows_stay_parked_and_compilation_refuses(): + inputs = summed_profile_fixture([TEN]) + effective, _, receipt = _resolve(inputs) + for row in effective["target_references"]: + assert row["metadata"]["activation_status"] == "requires_prerequisites" + assert json.loads(row["metadata"]["reference_prerequisites_remaining"]) == [ + f"measurement:{SUMMED_MEASUREMENT}" + ] + assert "measurement_proofs" not in receipt + assert validate_reference_resolution(effective, receipt) == receipt + with pytest.raises(ValueError, match="non-executable placeholder"): + _compile(effective, _facts(TEN)) + + +def test_a_proof_for_another_measurement_does_not_satisfy_this_one(): + inputs = summed_profile_fixture([TEN]) + other = { + "other": { + "measurement_id": "other", + "producers": {"toy.calibrate": "toy.other"}, + } + } + effective, _, receipt = _resolve(inputs, other) + assert { + row["metadata"]["activation_status"] for row in effective["target_references"] + } == {"requires_prerequisites"} + assert "measurement_proofs" not in receipt + + +def test_an_active_row_without_its_proof_in_the_receipt_refuses(): + inputs = summed_profile_fixture([TEN]) + effective, _, receipt = _resolve(inputs, proof_for()) + parked, _, parked_receipt = _resolve(inputs) + with pytest.raises(ValueError): + validate_reference_resolution(effective, parked_receipt) + forged = copy.deepcopy(parked) + for row in forged["target_references"]: + row["metadata"]["activation_status"] = "active" + with pytest.raises(ValueError): + validate_reference_resolution(forged, parked_receipt) + assert validate_reference_resolution(effective, receipt) == receipt + + +def _mutate(mutation): + groups = [TEN[:5], TEN[5:]] + base, profiles, context, intervals = summed_profile_fixture(groups) + rows = _rows(base) + if mutation == "money_member_count": + rows["toy_money_b0"]["expected_member_count"] = 4 + elif mutation == "count_member_count": + rows["toy_count_b0"]["expected_member_count"] = 6 + elif mutation == "non_identity_money": + rows["toy_money_b0"]["value_operation"] = "sum" + elif mutation == "mismatched_members": + rows["toy_count_b0"]["ledger_selector"]["dimension_values"]["band"] = [ + *TEN[:4], + "elsewhere", + ] + elif mutation == "overlapping_members": + for kind in ("count", "money"): + selector = rows[f"toy_{kind}_b1"]["ledger_selector"] + selector["dimension_values"]["band"] = [TEN[4], *TEN[6:]] + elif mutation == "identity_count": + rows["toy_count_b0"]["value_operation"] = "identity" + elif mutation == "fact_key_and_members": + rows["toy_money_b0"]["ledger_selector"]["aggregate_fact_key"] = "toy-money-x" + elif mutation == "repeated_member": + for kind in ("count", "money"): + selector = rows[f"toy_{kind}_b0"]["ledger_selector"] + selector["dimension_values"]["band"] = [*TEN[:4], TEN[0]] + elif mutation in ("mapping_member", "list_member"): + member = {"band": TEN[4]} if mutation == "mapping_member" else [TEN[4]] + for kind in ("count", "money"): + selector = rows[f"toy_{kind}_b0"]["ledger_selector"] + selector["dimension_values"]["band"] = [*TEN[:4], member] + else: # pragma: no cover - parametrization guard + raise AssertionError(mutation) + return base, profiles, context, intervals + + +@pytest.mark.parametrize( + ("mutation", "message"), + [ + ("money_member_count", "lists 5 members but declares expected_member_count 4"), + ("count_member_count", "lists 5 members but declares expected_member_count 6"), + ("non_identity_money", "untransformed source reference"), + ("mismatched_members", "different member sets"), + ("overlapping_members", "share a summed member"), + ("identity_count", "value_operation 'sum'"), + ("fact_key_and_members", "singleton fact key and a member set"), + ("repeated_member", "repeat or are not scalars"), + ("mapping_member", "repeat or are not scalars"), + ("list_member", "repeat or are not scalars"), + ], +) +def test_summed_rows_refuse_each_inconsistency(mutation, message): + with pytest.raises(ValueError, match=message): + _resolve(_mutate(mutation), proof_for()) + + +def test_receipt_states_the_binary64_rounding_of_the_weight(): + inputs = summed_profile_fixture([TEN]) + _, _, receipt = _resolve(inputs, proof_for()) + record = receipt["factor_weight_rounding"]["toy_central"] + exact = Fraction(record["exact_numerator"], record["exact_denominator"]) + assert record["weight"] == 1.1 + assert Fraction(record["binary64_relative_rounding"]) == ( + (Fraction(1.1) - exact) / exact + ) + assert abs(Fraction(record["binary64_relative_rounding"])) <= Fraction(1, 2**53) + + +@pytest.mark.parametrize( + "field", ["weight", "exact_numerator", "binary64_relative_rounding"] +) +def test_a_tampered_rounding_record_refuses(field): + """Only the rounding recomputation sees this: the receipt hash is redone.""" + effective, _, receipt = _resolve(summed_profile_fixture([TEN]), proof_for()) + tampered = copy.deepcopy(receipt) + record = tampered["factor_weight_rounding"]["toy_central"] + record[field] = { + "weight": 1.2, + "exact_numerator": record["exact_numerator"] + 1, + "binary64_relative_rounding": "0", + }[field] + del tampered["receipt_sha256"] + tampered["receipt_sha256"] = reference_document_sha256(tampered) + with pytest.raises(ValueError, match="Factor weight rounding record mismatch"): + validate_reference_resolution(effective, tampered) + assert validate_reference_resolution(effective, receipt) == receipt + + +def _partition(draw_sizes, labels): + groups, start = [], 0 + for size in draw_sizes: + groups.append(labels[start : start + size]) + start += size + return groups + + +@given( + sizes=st.lists(st.integers(min_value=1, max_value=6), min_size=1, max_size=5), + values=st.lists( + st.integers(min_value=0, max_value=10**9), min_size=30, max_size=30 + ), + counts=st.lists( + st.integers(min_value=0, max_value=10**6), min_size=30, max_size=30 + ), + numerator=st.integers(min_value=500_000, max_value=2_000_000), +) +@settings( + max_examples=25, + database=None, + deadline=None, + suppress_health_check=[HealthCheck.too_slow], +) +def test_compiled_coarse_rows_equal_factor_times_member_sums( + sizes, values, counts, numerator +): + labels = [f"m{index:02d}" for index in range(sum(sizes))] + groups = _partition(sizes, labels) + anchor = f"{numerator // 10_000}.{numerator % 10_000:04d}" + inputs = summed_profile_fixture(groups, anchors={"central": anchor}) + effective, _, receipt = _resolve(inputs, proof_for()) + rational = receipt["selected_factors"]["toy_central"]["operational_factor"] + factor = Fraction(rational["numerator"], rational["denominator"]) + assert factor == Fraction(numerator, 10**6) + facts = _facts(labels, count=lambda i: counts[i], money=lambda i: values[i]) + specs = _compile(effective, facts) + for band, members in enumerate(groups): + indices = [labels.index(label) for label in members] + assert specs[f"toy_count_b{band}"].value == sum(counts[i] for i in indices) + exact = factor * sum(values[i] for i in indices) + assert math.isclose( + specs[f"toy_money_b{band}"].value, float(exact), rel_tol=1e-12, abs_tol=1e-6 + ) + + +def _singleton_inputs(labels, anchors=None): + """Today's path: one singleton count/money pair per native band.""" + base, profiles, context, intervals = summed_profile_fixture( + [labels], anchors=anchors + ) + rows = [] + for label in labels: + for kind in ("count", "money"): + row = summed_row(kind, f"single_{kind}_{label}", [label], indicator="i") + row.pop("expected_member_count") + row["filter"] = f"single_{label}" + selector = row["ledger_selector"] + selector["dimension_values"] = {"band": label} + selector["aggregate_fact_key"] = ( + f"ledger.aggregate_fact.v2:toy-{kind}-{label}" + ) + if kind == "count": + row["value_operation"] = "identity" + rows.append(row) + base["target_references"] = rows + for row in rows: + row["metadata"]["contract_target_id"] = row["name"] + with_target_labels(base) + transform = profiles["dimensions"]["index"]["profiles"]["central"][ + "transformations" + ][0] + for profile in profiles["dimensions"]["index"]["profiles"].values(): + (transform,) = profile["transformations"] + transform["reference_names"] = [ + row["name"] for row in rows if row["name"].startswith("single_money") + ] + transform["count_reference_names"] = [ + row["name"] for row in rows if row["name"].startswith("single_count") + ] + transform["interval_table_ids"] = [] + return base, profiles, context, None + + +@pytest.mark.parametrize("anchor", ["105", "110", "117.1109"]) +def test_summed_path_equals_the_sum_of_singleton_rows_aged_today(anchor): + facts = _facts(TEN, money=lambda i: 123_457 * (i + 3) + 11) + summed, _, _ = _resolve( + summed_profile_fixture([TEN], anchors={"central": anchor}), proof_for() + ) + singles, _, _ = _resolve(_singleton_inputs(TEN, {"central": anchor}), proof_for()) + summed_specs = _compile(summed, facts) + single_specs = _compile(singles, facts) + assert summed_specs["toy_count_b0"].value == sum( + single_specs[f"single_count_{label}"].value for label in TEN + ) + assert math.isclose( + summed_specs["toy_money_b0"].value, + sum(single_specs[f"single_money_{label}"].value for label in TEN), + rel_tol=1e-13, + ) + for label in TEN: + operand = _rows(singles)[f"single_money_{label}"]["value_operands"][0] + assert set(operand) == {"aggregate_fact_key", "expected_member_count", "weight"} + assert operand["expected_member_count"] == 1 diff --git a/packages/microcosm-build/tests/engine_free/shared/test_transport_composed_graph.py b/packages/microcosm-build/tests/engine_free/shared/test_transport_composed_graph.py index f597aa1335..addd69d4bf 100644 --- a/packages/microcosm-build/tests/engine_free/shared/test_transport_composed_graph.py +++ b/packages/microcosm-build/tests/engine_free/shared/test_transport_composed_graph.py @@ -56,7 +56,6 @@ from hypothesis import HealthCheck, example, given, settings from hypothesis import strategies as st -from microcosm.build.reference_profiles import reference_document_sha256 from microcosm.build.transport import cli as cli_module from microcosm.build.transport import compose as compose_module from microcosm.build.transport import target_kernels as target_kernels_module @@ -114,6 +113,7 @@ from test_support.microcosm_build.chronicle_feed import write_pinned_artifact from test_support.microcosm_build.money_normalization import graph_declaration from test_support.microcosm_build.reference_factor_transport import ( + REVIEWED_MEASUREMENT_PROOF, factor_transport_fixture, ) from test_support.microcosm_build.reference_intervals import ( @@ -3207,19 +3207,21 @@ def test_cli_require_replays_a_warm_store_after_selected_factors_expire( ) profile_view = compose_module.reference_profile_view + proofs = compose_module.measurement_proofs + def measured_view(spec, selection=None): # As in factor_transport_fixture(activate=True), supply a synthetic - # reviewed measurement closure. The real resolver still selects and - # binds the factors; composition must attach its rehashed receipt. - # Copies, so the resolver's own return values are never edited. - references, intervals, receipt = profile_view(spec, selection) - references, receipt = copy.deepcopy(references), copy.deepcopy(receipt) - for row in references["target_references"]: - row["metadata"]["activation_status"] = "active" - receipt["effective_references_sha256"] = reference_document_sha256(references) - receipt.pop("receipt_sha256") - receipt["receipt_sha256"] = reference_document_sha256(receipt) - return references, intervals, receipt + # proof of the reviewed measurement, only while the view resolves. + # The real resolver still selects and binds the factors and activates + # the proved rows; composition must attach its receipt. + compose_module.measurement_proofs = lambda spec: ( + REVIEWED_MEASUREMENT_PROOF, + {}, + ) + try: + return profile_view(spec, selection) + finally: + compose_module.measurement_proofs = proofs monkeypatch.setattr(compose_module, "reference_profile_view", measured_view) declaration = spec["resources"]["transport_graph"] diff --git a/packages/microcosm-build/tests/engine_free/shared/test_transport_gate_bindings.py b/packages/microcosm-build/tests/engine_free/shared/test_transport_gate_bindings.py index 1a81c11dfd..234a11bef4 100644 --- a/packages/microcosm-build/tests/engine_free/shared/test_transport_gate_bindings.py +++ b/packages/microcosm-build/tests/engine_free/shared/test_transport_gate_bindings.py @@ -602,7 +602,7 @@ def test_extends_the_default_registry_with_the_bound_gates(self) -> None: | set(BOUND) | set(PENDING) | set(EVIDENCE) - | {"cell_support", "cell_support_target"} + | {"cell_support", "cell_support_target", "measurement_prefit"} ) for name in DEFAULT_REGISTRY: if name not in BOUND: diff --git a/packages/microcosm-build/tests/engine_free/shared/test_transport_measurement_prefit.py b/packages/microcosm-build/tests/engine_free/shared/test_transport_measurement_prefit.py new file mode 100644 index 0000000000..e9536a56ea --- /dev/null +++ b/packages/microcosm-build/tests/engine_free/shared/test_transport_measurement_prefit.py @@ -0,0 +1,300 @@ +"""The ``measurement_prefit`` gate (G8-r2c hunk D, ruling R2C-3). + +The gate reads the problem's interval diagnostics with their per-band +support and the problem's matrix at the weights entering calibration. Every +threshold comes from the ``gates.json`` entry. These tests drive each of the +four checks to just inside and just outside its threshold, on each side. +""" + +from __future__ import annotations + +import copy +import json +from types import SimpleNamespace + +import numpy as np +import pytest + +from microcosm.build.country_spec import ALLOWED_GATE_FUNCTIONS, GatesManifest +from microcosm.build.transport.gate_bindings import ( + TRANSPORT_GATE_REGISTRY, + _measurement_prefit, + required_gate_parameters, +) +from test_support.microcosm_build.reference_intervals import ( + R2C3_PREFIT_THRESHOLDS, + TOY_PREFIT_EVIDENCE, + prefit_gate_entry, +) + +BANDS = 12 +TABLE = "toy_coarse" +_MANIFEST_HEADER = { + "version": 2, + "country": "toy", + "policy": "Toy gates for the pre-fit rule.", + "phases": ["terminal"], +} + + +#: Non-unit weights entering calibration. The matrix gives the declared +#: model values exactly at these weights, and other values at unit weights. +INITIAL_WEIGHTS = (2.0, 4.0) + + +class _Problem: + """The slice of ``OrderedProblem`` the gate reads. + + ``estimates`` multiplies a two-column matrix by the weights it receives. + Its columns are model/4 and model/8, so A·w₀ equals the model exactly at + ``INITIAL_WEIGHTS`` (powers of two divide exactly) and equals 3/8 of it + at unit weights. + """ + + def __init__(self, model, reference, report): + self.target_metadata = tuple({"reference": name} for name in model) + names = list(model) + values = np.asarray([model[name] for name in names], dtype=float) + self._matrix = np.column_stack([values / 4, values / 8]) + self.problem = SimpleNamespace( + initial_weights=SimpleNamespace(values=np.asarray(INITIAL_WEIGHTS)), + target_vector=np.asarray([reference[name] for name in names], dtype=float), + estimates=lambda weights: self._matrix @ np.asarray(weights, dtype=float), + ) + self.bindings = {"interval_diagnostics": report} + + +def _support(households=40, ess=(40, 1)): + measure = { + "source_households": households, + "kish_ess": {"numerator": ess[0], "denominator": ess[1]}, + } + return {"count": dict(measure), "amount": copy.deepcopy(measure)} + + +def _case(ratios=None, *, support=None): + """Twelve bands; ``ratios[band] = (count ratio, amount ratio)``.""" + ratios = ratios or {} + model, reference, bands = {}, {}, [] + for band in range(BANDS): + count_ratio, amount_ratio = ratios.get(band, (1, 1)) + count, amount = f"count_b{band}", f"money_b{band}" + reference[count], reference[amount] = 1_000, 1_000_000 + model[count] = 1_000 * count_ratio + model[amount] = 1_000_000 * amount_ratio + bands.append( + { + "band": f"b{band}", + "references": {"count": count, "amount": amount}, + "support": (support or {}).get(band, _support()), + } + ) + report = { + "tables": [{"table": TABLE, "support": {"schema_version": 1}, "bands": bands}] + } + return _Problem(model, reference, report) + + +def _run(problem, *, evidence=None, **overrides): + parameters = prefit_gate_entry(tables=(TABLE,))["parameters"] + parameters.update(overrides) + wired = ( + {alias: {} for alias in TOY_PREFIT_EVIDENCE} if evidence is None else evidence + ) + return _measurement_prefit(problem=problem, graph_evidence=wired, **parameters) + + +def test_r2c3_thresholds_are_pinned(): + """The NZ session reuses this pin; a change needs a hub ruling first.""" + assert R2C3_PREFIT_THRESHOLDS == { + "roster_total_bounds": [0.75, 1.25], + "band_bounds": [0.5, 2.0], + "maximum_discrepant_bands": 2, + "minimum_source_households": 30, + "minimum_kish_ess": 30, + } + + +def test_a_balanced_well_supported_roster_passes(): + result = _run(_case()) + assert result.passed, result.failures + assert result.details["roster_totals"]["count"]["ratio"] == "1" + + +def test_the_model_side_is_the_matrix_at_the_initial_weights(): + """R2C-3's A·w₀: the gate evaluates the matrix at w₀, not at unit weights.""" + problem = _case({0: (0.75, 1)}) + weights = problem.problem.initial_weights.values + assert not np.all(weights == 1) + assert problem.problem.estimates(np.ones_like(weights))[0] != 750 + result = _run(problem) + assert result.passed, result.failures + assert result.details["bands"][0]["count"]["model"] == 750 + + +def _uniform(ratio, role): + return { + band: (ratio, 1) if role == "count" else (1, ratio) for band in range(BANDS) + } + + +@pytest.mark.parametrize("role", ["count", "amount"]) +@pytest.mark.parametrize( + ("ratio", "passes"), + [(0.75, True), (0.7499, False), (1.25, True), (1.2501, False)], +) +def test_roster_totals_refuse_outside_75_to_125_percent(role, ratio, passes): + result = _run(_case(_uniform(ratio, role))) + assert result.passed is passes, result.failures + if not passes: + assert any(f"summed {role}" in line for line in result.failures) + + +def _on(role, ratio): + return (ratio, 1) if role == "count" else (1, ratio) + + +@pytest.mark.parametrize(("discrepant", "passes"), [(2, True), (3, False), (4, False)]) +@pytest.mark.parametrize("low", [True, False]) +@pytest.mark.parametrize("role", ["count", "amount"]) +def test_three_discrepant_bands_refuse_and_two_pass(discrepant, passes, low, role): + # Pair each discrepant band with an offsetting one so the totals hold. + # With role "amount", every discrepant band fails on its amount only. + ratios = {} + for band in range(discrepant): + ratios[band] = _on(role, 0.49 if low else 2.01) + ratios[BANDS - 1 - band] = _on(role, 1.51 if low else 0.5) + result = _run(_case(ratios)) + assert len(result.details["discrepant_bands"]) == discrepant + assert result.passed is passes, result.failures + if not passes: + assert any( + line.startswith(f"concentrated discrepancy: {discrepant} bands") + for line in result.failures + ) + + +@pytest.mark.parametrize( + ("ratio", "inside"), [(0.5, True), (0.4999, False), (2.0, True), (2.0001, False)] +) +def test_band_bounds_are_inclusive_at_50_and_200_percent(ratio, inside): + ratios = {0: (ratio, 1), 1: (ratio, 1), 2: (ratio, 1)} + result = _run(_case(ratios), roster_total_bounds=[0, 10]) + assert result.passed is inside, result.failures + + +def test_a_band_failing_count_and_amount_counts_once(): + ratios = {0: (0.1, 0.1), 1: (0.1, 0.1)} + result = _run(_case(ratios), roster_total_bounds=[0, 10]) + assert result.details["discrepant_bands"] == [f"{TABLE}/b0", f"{TABLE}/b1"] + assert result.passed, result.failures + + +@pytest.mark.parametrize("role", ["count", "amount"]) +@pytest.mark.parametrize( + ("households", "ess", "refusals"), + [ + (30, (30, 1), []), + (29, (29, 1), ["29 source households", "Kish ESS 29"]), + (30, (2999, 100), ["Kish ESS 29.99"]), + (60, (30, 1), []), + # Separately: the household floor alone refuses, whatever the ESS. + (29, (30, 1), ["29 source households"]), + ], +) +def test_support_refuses_below_30_households_or_ess_separately( + role, households, ess, refusals +): + support = _support() + support[role] = { + "source_households": households, + "kish_ess": {"numerator": ess[0], "denominator": ess[1]}, + } + result = _run(_case(support={4: support})) + assert result.passed is (not refusals), result.failures + assert len(result.failures) == len(refusals) + for line, reason in zip(result.failures, refusals, strict=True): + assert line.startswith(f"support: band {TABLE}/b4 {role}") + assert reason in line + + +def test_missing_evidence_refuses_each_way(): + missing_alias = _run(_case(), evidence={"native_band_report": {}}) + assert not missing_alias.passed + assert any("omission_register" in line for line in missing_alias.failures) + + no_report = _case() + no_report.bindings = {} + result = _run(no_report) + assert not result.passed + assert any("no interval diagnostics" in line for line in result.failures) + + no_support = _case() + del no_support.bindings["interval_diagnostics"]["tables"][0]["support"] + assert any("no per-band support" in line for line in _run(no_support).failures) + + not_in_matrix = _case() + not_in_matrix.bindings["interval_diagnostics"]["tables"][0]["bands"][3][ + "references" + ]["count"] = "elsewhere" + assert any( + "did not enter the matrix" in line for line in _run(not_in_matrix).failures + ) + + other_table = _run(_case(), tables=["another_table"]) + assert any("another_table" in line for line in other_table.failures) + + +def test_the_binding_takes_every_threshold_from_gates_json(): + binding = TRANSPORT_GATE_REGISTRY["measurement_prefit"] + assert required_gate_parameters(binding) == frozenset( + { + "tables", + "roster_total_bounds", + "band_bounds", + "maximum_discrepant_bands", + "minimum_source_households", + "minimum_kish_ess", + "required_evidence", + } + ) + assert set(binding.artifact_arguments) == {"problem", "graph_evidence"} + + +def test_a_country_without_the_entry_gets_no_gate(): + manifest = GatesManifest.from_mapping( + { + **_MANIFEST_HEADER, + "gates": [ + { + "id": "coverage", + "gate": "target_profile_coverage", + "phase": "terminal", + "criticality": "release_blocking", + "parameters": {"required_families": ["toy"]}, + } + ], + }, + country="toy", + ) + assert [entry.gate for entry in manifest.gates] == ["target_profile_coverage"] + + +def test_a_gates_json_entry_naming_the_gate_loads(): + """Ruling R2C-12: the loader admits ``measurement_prefit``.""" + assert "measurement_prefit" in ALLOWED_GATE_FUNCTIONS + entry = prefit_gate_entry() + manifest = GatesManifest.from_mapping( + {**_MANIFEST_HEADER, "gates": [entry]}, country="toy" + ) + (loaded,) = manifest.gates + assert loaded.gate == "measurement_prefit" + assert loaded.criticality == "release_blocking" + # The loader freezes lists into tuples; the values are the entry's. + assert json.loads(json.dumps(dict(loaded.parameters))) == entry["parameters"] + + +@pytest.mark.parametrize("bounds", [[], [0.75], [0.75, 1, 1.25], "ab", [1.25, 0.75]]) +def test_ratio_bounds_must_be_an_ordered_pair(bounds): + with pytest.raises((TypeError, ValueError), match="roster_total_bounds"): + _run(_case(), roster_total_bounds=bounds) diff --git a/packages/microcosm-build/tests/engine_free/shared/test_transport_measurement_proof.py b/packages/microcosm-build/tests/engine_free/shared/test_transport_measurement_proof.py new file mode 100644 index 0000000000..de485f7674 --- /dev/null +++ b/packages/microcosm-build/tests/engine_free/shared/test_transport_measurement_proof.py @@ -0,0 +1,845 @@ +"""Composition's proof of a measurement producer (G8-r2c ruling R2C-1). + +``compose.measurement_proofs`` satisfies ``measurement:`` only when, for +every calibration node, exactly one node upstream of it declares +``params.measurement_id: `` and writes the scaled interval table's income and +universe columns on that table's entity. Each mutant below breaks one clause +and must leave the rows parked, with the activation gap naming the +measurement. +""" + +from __future__ import annotations + +import copy +import json + +import pytest +from hypothesis import HealthCheck, example, given, settings +from hypothesis import strategies as st + +from microcosm.build.transport.compose import ( + _reference_activation_gaps, + compose_transport_graph, + measurement_proofs, + reference_profile_view, +) +from test_support.microcosm_build.reference_intervals import ( + INCOME_COLUMN, + SUMMED_MEASUREMENT, + UNIVERSE_COLUMN, + summed_profile_fixture, +) + +GROUPS = [[f"n{index}" for index in range(5)], [f"n{index}" for index in range(5, 9)]] + + +def _graph(entity="person", *, producer_entity=None, measured=True): + """CREATE, the producer, preparation, the problem and calibration. + + Preparation reads the income and universe columns on the producer's + entity, so the producer always sits upstream of calibration. + """ + producer_entity = entity if producer_entity is None else producer_entity + measure = { + "id": "toy.measure", + "kernel": "measure.combine@1", + "population": "toy.create", + "inputs": [{"entity": producer_entity, "columns": ["toy_wages"]}], + "outputs": [ + {"entity": producer_entity, "column": INCOME_COLUMN, "dtype": "int64"}, + {"entity": producer_entity, "column": UNIVERSE_COLUMN, "dtype": "str"}, + ], + } + if measured: + measure["params"] = {"measurement_id": SUMMED_MEASUREMENT} + return { + "schema_version": 1, + "country": "toy", + "nodes": [ + { + "id": "toy.create", + "kernel": "transport.create@1", + "structural": "create", + }, + measure, + { + "id": "toy.prepare", + "kernel": "calibration.prepare@1", + "population": "toy.create", + "inputs": [ + { + "entity": producer_entity, + "columns": [INCOME_COLUMN, UNIVERSE_COLUMN], + } + ], + "outputs": [ + {"entity": entity, "column": "toy_income_band_b0", "dtype": "bool"} + ], + }, + { + "id": "toy.problem", + "kernel": "targets.problem@1", + "population": "toy.create", + "artifact_inputs": [{"name": "prepared", "producer": "toy.prepare"}], + }, + { + "id": "toy.calibrate", + "kernel": "calibrate.ordered_adam@1", + "population": "toy.create", + "artifact_inputs": [{"name": "problem", "producer": "toy.problem"}], + }, + ], + "endpoints": {"calibration": "toy.calibrate"}, + } + + +def _spec(graph, *, entity="person"): + base, profiles, context, intervals = summed_profile_fixture(GROUPS, entity=entity) + return { + "country": "toy", + "resources": { + "reference_profiles": profiles, + "target_references": base, + "factor_context": context, + "profile_intervals": intervals, + "transport_graph": graph, + }, + } + + +def _nodes(graph): + return {row["id"]: row for row in graph["nodes"]} + + +def test_one_upstream_producer_proves_and_activates_the_rows(): + spec = _spec(_graph()) + proofs, refusals = measurement_proofs(spec) + assert refusals == {} + assert proofs == { + SUMMED_MEASUREMENT: { + "measurement_id": SUMMED_MEASUREMENT, + "producers": {"toy.calibrate": "toy.measure"}, + "columns": [f"person.{INCOME_COLUMN}", f"person.{UNIVERSE_COLUMN}"], + } + } + references, _, receipt = reference_profile_view(spec) + assert receipt["measurement_proofs"] == proofs + assert { + row["metadata"]["activation_status"] for row in references["target_references"] + } == {"active"} + assert list(_reference_activation_gaps(references, "target_references")) == [] + + +def test_a_third_entity_table_passes_the_proof(): + """A table on a non-household, non-person entity, as ``tax_unit`` is.""" + spec = _spec(_graph("tax_unit"), entity="tax_unit") + proofs, refusals = measurement_proofs(spec) + assert refusals == {} + assert proofs[SUMMED_MEASUREMENT]["columns"] == [ + f"tax_unit.{INCOME_COLUMN}", + f"tax_unit.{UNIVERSE_COLUMN}", + ] + + +def _no_producer(graph): + del _nodes(graph)["toy.measure"]["params"] + + +def _after_calibration(graph): + # The producer reads the calibration artifact, so nothing upstream of + # calibration can read what it writes; preparation reads a CREATE column. + nodes = _nodes(graph) + nodes["toy.measure"]["artifact_inputs"] = [ + {"name": "fit", "producer": "toy.calibrate"} + ] + nodes["toy.prepare"]["inputs"] = [{"entity": "person", "columns": ["toy_wages"]}] + + +def _two_producers(graph): + second = copy.deepcopy(_nodes(graph)["toy.measure"]) + second["id"] = "toy.measure.again" + second["outputs"] = [ + {"entity": "person", "column": "toy_other_income", "dtype": "int64"} + ] + graph["nodes"].insert(2, second) + _nodes(graph)["toy.prepare"]["inputs"][0]["columns"].append("toy_other_income") + + +def _wrong_column(graph): + # The producer writes another income column; CREATE supplies the one + # the table names. + nodes = _nodes(graph) + nodes["toy.measure"]["outputs"][0]["column"] = "toy_other_cents" + + +def _income_without_universe(graph): + nodes = _nodes(graph) + nodes["toy.measure"]["outputs"] = nodes["toy.measure"]["outputs"][:1] + graph["nodes"].insert( + 2, + { + "id": "toy.universe", + "kernel": "measure.combine@1", + "population": "toy.create", + "outputs": [ + {"entity": "person", "column": UNIVERSE_COLUMN, "dtype": "str"} + ], + }, + ) + + +def _filtered(graph): + """Open a ``filter`` version on CREATE and move calibration into it.""" + nodes = _nodes(graph) + position = graph["nodes"].index(nodes["toy.prepare"]) + graph["nodes"].insert( + position, + { + "id": "toy.filter", + "kernel": "transport.boundary@1", + "structural": "filter", + "base": "toy.create", + }, + ) + for name in ("toy.prepare", "toy.problem", "toy.calibrate"): + nodes[name]["population"] = "toy.filter" + return position + 1 + + +def _rewriter(name, column, dtype, *, population="toy.filter", rewrite=True): + output = {"entity": "person", "column": column, "dtype": dtype} + if rewrite: + output["rewrite"] = True + return { + "id": name, + "kernel": "toy.rewrite@1", + "population": population, + "inputs": [{"entity": "person", "columns": [column]}], + "outputs": [output], + } + + +def _later_version_rewrite(graph): + """The review's item 2c: an undeclared later-version rewrite of the income.""" + position = _filtered(graph) + graph["nodes"].insert(position, _rewriter("toy.overwrite", INCOME_COLUMN, "int64")) + + +def _writer_before_the_producer(graph): + """An undeclared CREATE-version writer; the producer rewrites it later.""" + _filtered(graph) + nodes = _nodes(graph) + nodes["toy.measure"]["population"] = "toy.filter" + nodes["toy.measure"]["outputs"][0]["rewrite"] = True + graph["nodes"].remove(nodes["toy.measure"]) + graph["nodes"].insert( + graph["nodes"].index(nodes["toy.prepare"]), nodes["toy.measure"] + ) + graph["nodes"].insert( + 1, + _rewriter( + "toy.early", + INCOME_COLUMN, + "int64", + population="toy.create", + rewrite=False, + ), + ) + graph["nodes"][1]["inputs"] = [{"entity": "person", "columns": ["toy_wages"]}] + + +def _universe_rewrite(graph): + position = _filtered(graph) + graph["nodes"].insert(position, _rewriter("toy.overwrite", UNIVERSE_COLUMN, "str")) + + +def _expand_writer(graph, cells=None): + """An EXPAND version that may write the income, with calibration in it.""" + nodes = _nodes(graph) + graph["nodes"].insert( + graph["nodes"].index(nodes["toy.prepare"]), + { + "id": "toy.expand", + "kernel": "transport.boundary@1", + "structural": "expand", + "base": "toy.create", + "params": { + "expand_cells": ( + [["person", INCOME_COLUMN, "int64"]] if cells is None else cells + ) + }, + }, + ) + for name in ("toy.prepare", "toy.problem", "toy.calibrate"): + nodes[name]["population"] = "toy.expand" + + +def _expand_with_resource_cells(graph): + """An EXPAND whose cells only composition can tell writes every column.""" + _expand_writer(graph, {"resource": "toy_cells"}) + + +@pytest.mark.parametrize( + ("mutant", "graph", "entity", "reason"), + [ + ( + "expand_cells", + _expand_writer, + "person", + f"writers ['toy.expand', 'toy.measure'] of person.{INCOME_COLUMN} upstream", + ), + ( + "expand_resource_cells", + _expand_with_resource_cells, + "person", + f"writers ['toy.expand', 'toy.measure'] of person.{INCOME_COLUMN} upstream", + ), + ( + "later_version_rewrite", + _later_version_rewrite, + "person", + "writers ['toy.measure', 'toy.overwrite'] of person." + f"{INCOME_COLUMN} upstream", + ), + ( + "writer_before_the_producer", + _writer_before_the_producer, + "person", + f"writers ['toy.early', 'toy.measure'] of person.{INCOME_COLUMN} upstream", + ), + ( + "universe_rewrite", + _universe_rewrite, + "person", + "writers ['toy.measure', 'toy.overwrite'] of person." + f"{UNIVERSE_COLUMN} upstream", + ), + ("no_producer", _no_producer, "person", "0 nodes declaring"), + ("after_calibration", _after_calibration, "person", "0 nodes declaring"), + ("two_producers", _two_producers, "person", "2 nodes declaring"), + ("wrong_column", _wrong_column, "person", "does not write"), + ( + "income_without_universe", + _income_without_universe, + "person", + "does not write", + ), + ("wrong_entity", None, "person", "does not write"), + ("wrong_entity_tax_unit", None, "tax_unit", "does not write"), + ], +) +def test_each_proof_mutant_leaves_rows_parked_and_names_the_measurement( + mutant, graph, entity, reason +): + if graph is None: + document = _graph(entity, producer_entity="household") + else: + document = _graph(entity) + graph(document) + spec = _spec(document, entity=entity) + proofs, refusals = measurement_proofs(spec) + assert proofs == {} + assert reason in refusals[SUMMED_MEASUREMENT] + references, _, receipt = reference_profile_view(spec) + assert "measurement_proofs" not in receipt + rows = references["target_references"] + assert {row["metadata"]["activation_status"] for row in rows} == { + "requires_prerequisites" + } + gaps = list( + _reference_activation_gaps(references, "target_references", refusals=refusals) + ) + assert len(gaps) == len(rows) + assert all(f"measurement:{SUMMED_MEASUREMENT} (" in gap for gap in gaps) + assert all(reason in gap for gap in gaps) + + +def test_every_calibration_node_needs_its_own_upstream_producer(): + graph = _graph() + graph["nodes"].append( + { + "id": "toy.calibrate.variant", + "kernel": "calibrate.ordered_adam@1", + "population": "toy.create", + } + ) + proofs, refusals = measurement_proofs(_spec(graph)) + assert proofs == {} + assert "toy.calibrate.variant" in refusals[SUMMED_MEASUREMENT] + + +def test_a_spec_without_profiles_or_scaling_proves_nothing(): + assert measurement_proofs({"country": "toy", "resources": {}}) == ({}, {}) + spec = _spec(_graph()) + for profile in spec["resources"]["reference_profiles"]["dimensions"]["index"][ + "profiles" + ].values(): + profile["transformations"] = [] + assert measurement_proofs(spec) == ({}, {}) + + +def test_proof_result_is_bound_into_the_selection_identity(): + proved, _, proved_receipt = reference_profile_view(_spec(_graph())) + document = _graph() + _no_producer(document) + parked, _, parked_receipt = reference_profile_view(_spec(document)) + assert ( + proved_receipt["reference_selection_sha256"] + != parked_receipt["reference_selection_sha256"] + ) + assert ( + json.loads( + proved["target_references"][0]["metadata"][ + "reference_prerequisites_remaining" + ] + ) + == [] + ) + assert parked != proved + + +def _compiled_toy(*, second_producer=False, second_writer=False, expand_cells=None): + """A compiled graph: CREATE, one or two producers, preparation, calibration. + + With ``second_writer``, a ``filter`` version holds an undeclared rewrite + of the income column, and preparation and calibration live in it. With + ``expand_cells``, preparation and calibration live in an EXPAND version + that declares those cells. + """ + from microcosm.graph import ( + Graph, + Node, + Owned, + Slice, + SourceRef, + StructuralDelta, + compile_graph, + ) + + def producer(name, column): + return Node( + id=name, + kernel="measure.combine@1", + population="toy.create", + params={"measurement_id": SUMMED_MEASUREMENT}, + outputs=(Owned("person", column, "int64"),), + ) + + nodes = [ + Node( + id="toy.create", + kernel="transport.create@1", + structural=StructuralDelta.CREATE, + sources=("toy_donor",), + outputs=(Owned("person", "person_id", "int64"),), + ), + producer("toy.measure", INCOME_COLUMN), + ] + read = [INCOME_COLUMN] + if second_producer: + nodes.append(producer("toy.measure.again", "toy_other_cents")) + read.append("toy_other_cents") + population = "toy.create" + if second_writer: + population = "toy.filter" + nodes += [ + Node( + id="toy.filter", + kernel="transport.boundary@1", + structural=StructuralDelta.FILTER, + base="toy.create", + ), + Node( + id="toy.overwrite", + kernel="toy.rewrite@1", + population="toy.filter", + inputs=(Slice("person", (INCOME_COLUMN,)),), + outputs=(Owned("person", INCOME_COLUMN, "int64", rewrite=True),), + ), + ] + if expand_cells is not None: + population = "toy.expand" + nodes.append( + Node( + id="toy.expand", + kernel="transport.boundary@1", + structural=StructuralDelta.EXPAND, + base="toy.create", + params={"expand_cells": expand_cells}, + ) + ) + nodes += [ + Node( + id="toy.prepare", + kernel="calibration.prepare@1", + population=population, + inputs=(Slice("person", tuple(read)),), + outputs=(Owned("person", "toy_income_band_b0", "bool"),), + ), + Node( + id="toy.calibrate", + kernel="calibrate.ordered_adam@1", + population=population, + inputs=(Slice("person", ("toy_income_band_b0",)),), + ), + ] + graph = Graph( + country="toy", + sources=(SourceRef("toy_donor", "toy-codec-v1"),), + nodes=tuple(nodes), + ) + return graph, compile_graph(graph) + + +def _forced_proof(monkeypatch, producer, columns=(f"person.{INCOME_COLUMN}",)): + from microcosm.build.transport import compose as compose_module + + proof = { + SUMMED_MEASUREMENT: { + "measurement_id": SUMMED_MEASUREMENT, + "producers": {"toy.calibrate": producer}, + "columns": list(columns), + } + } + monkeypatch.setattr(compose_module, "measurement_proofs", lambda spec: (proof, {})) + monkeypatch.setattr( + compose_module, + "transport_endpoints", + lambda spec: {"calibration": "toy.calibrate"}, + ) + return compose_module._check_measurement_proofs + + +def test_the_compiled_graph_confirms_the_declared_producer(monkeypatch): + check = _forced_proof(monkeypatch, "toy.measure") + graph, compiled = _compiled_toy() + check({}, graph, compiled) + + +@pytest.mark.parametrize( + ("producer", "second", "found"), + [("toy.elsewhere", False, "1 nodes"), ("toy.measure", True, "2 nodes")], +) +def test_composition_refuses_a_proof_the_compiled_graph_contradicts( + monkeypatch, producer, second, found +): + check = _forced_proof(monkeypatch, producer) + graph, compiled = _compiled_toy(second_producer=second) + with pytest.raises(ValueError, match=f"contradicts the declared proof; {found}"): + check({}, graph, compiled) + + +def test_the_compiled_check_alone_refuses_a_second_writer(monkeypatch): + """The declared check never sees this graph: a forced proof drives it.""" + check = _forced_proof(monkeypatch, "toy.measure") + graph, compiled = _compiled_toy(second_writer=True) + with pytest.raises(ValueError) as refused: + check({}, graph, compiled) + assert str(refused.value) == ( + f"Measurement {SUMMED_MEASUREMENT!r}: the compiled graph contradicts the " + f"declared proof; measurement {SUMMED_MEASUREMENT!r}: calibration node " + "'toy.calibrate' has writers ['toy.measure', 'toy.overwrite'] of " + f"person.{INCOME_COLUMN} upstream; only its producer 'toy.measure' may " + "write it." + ) + + +def test_the_compiled_check_alone_counts_an_expand_as_a_writer(monkeypatch): + check = _forced_proof(monkeypatch, "toy.measure") + graph, compiled = _compiled_toy(expand_cells=(("person", INCOME_COLUMN, "int64"),)) + with pytest.raises(ValueError, match=r"writers \['toy.expand', 'toy.measure'\]"): + check({}, graph, compiled) + + +def test_the_compiled_check_passes_an_expand_of_another_column(monkeypatch): + check = _forced_proof(monkeypatch, "toy.measure") + graph, compiled = _compiled_toy(expand_cells=(("person", "toy_other", "int64"),)) + check({}, graph, compiled) + + +def test_a_non_string_measurement_id_param_proves_nothing(): + document = _graph() + _nodes(document)["toy.measure"]["params"] = { + "measurement_id": {"resource": "factor_context", "path": [], "encoding": "json"} + } + proofs, refusals = measurement_proofs(_spec(document)) + assert proofs == {} + assert "not a string" in refusals[SUMMED_MEASUREMENT] + + +# --------------------------------------------------------------------------- +# R2C-15: a measurement has one writer +# --------------------------------------------------------------------------- + + +def test_a_later_version_rewrite_names_the_measurement_node_column_and_writers(): + """The review's item 2c refuses in the declared check, naming every fact.""" + document = _graph() + _later_version_rewrite(document) + proofs, refusals = measurement_proofs(_spec(document)) + assert proofs == {} + assert refusals == { + SUMMED_MEASUREMENT: ( + f"measurement {SUMMED_MEASUREMENT!r}: calibration node " + "'toy.calibrate' has writers ['toy.measure', 'toy.overwrite'] of " + f"person.{INCOME_COLUMN} upstream; only its producer 'toy.measure' " + "may write it" + ) + } + + +def test_a_second_writer_refuses_without_rewrite(): + """A later writer refuses whether or not its output says ``rewrite``.""" + document = _graph() + position = _filtered(document) + document["nodes"].insert( + position, _rewriter("toy.overwrite", INCOME_COLUMN, "int64", rewrite=False) + ) + proofs, refusals = measurement_proofs(_spec(document)) + assert proofs == {} + assert "['toy.measure', 'toy.overwrite']" in refusals[SUMMED_MEASUREMENT] + + +def test_a_same_named_column_on_another_entity_is_not_a_second_writer(): + document = _graph() + position = _filtered(document) + other = _rewriter("toy.household", INCOME_COLUMN, "int64", rewrite=False) + other["inputs"] = [{"entity": "person", "columns": ["toy_wages"]}] + other["outputs"][0]["entity"] = "household" + document["nodes"].insert(position, other) + _nodes(document)["toy.prepare"]["inputs"].append( + {"entity": "household", "columns": [INCOME_COLUMN]} + ) + proofs, refusals = measurement_proofs(_spec(document)) + assert refusals == {} + assert proofs[SUMMED_MEASUREMENT]["producers"] == {"toy.calibrate": "toy.measure"} + + +def test_a_writer_outside_the_calibration_closure_is_not_a_second_writer(): + """A later version that reads the fit rewrites the column after calibration.""" + document = _graph() + document["nodes"] += [ + { + "id": "toy.after", + "kernel": "transport.boundary@1", + "structural": "filter", + "base": "toy.create", + "artifact_inputs": [{"name": "fit", "producer": "toy.calibrate"}], + }, + _rewriter("toy.overwrite", INCOME_COLUMN, "int64", population="toy.after"), + ] + proofs, refusals = measurement_proofs(_spec(document)) + assert refusals == {} + assert proofs[SUMMED_MEASUREMENT]["producers"] == {"toy.calibrate": "toy.measure"} + + +@given( + cells=st.lists( + st.sampled_from([INCOME_COLUMN, UNIVERSE_COLUMN, "toy_other"]), unique=True + ) +) +@settings( + max_examples=30, + database=None, + deadline=None, + suppress_health_check=[HealthCheck.too_slow], +) +def test_an_expand_proves_exactly_when_its_cells_avoid_the_measured_columns(cells): + document = _graph() + _expand_writer(document, [["person", column, "int64"] for column in cells]) + proofs, refusals = measurement_proofs(_spec(document)) + proved = not {INCOME_COLUMN, UNIVERSE_COLUMN} & set(cells) + assert (SUMMED_MEASUREMENT in proofs) is proved + assert (SUMMED_MEASUREMENT in refusals) is not proved + + +def _chain(draws): + """CREATE, then nodes chained by artifacts, all upstream of calibration. + + Each draw is (declares the id, writes the income, writes the universe). + """ + nodes = [ + {"id": "toy.create", "kernel": "transport.create@1", "structural": "create"} + ] + previous = None + for index, (declares, income, universe) in enumerate(draws): + row = { + "id": f"toy.n{index}", + "kernel": "toy.chain@1", + "population": "toy.create", + "outputs": [ + {"entity": "person", "column": column, "dtype": dtype} + for column, dtype, wanted in ( + (INCOME_COLUMN, "int64", income), + (UNIVERSE_COLUMN, "str", universe), + (f"toy_step_{index}", "bool", True), + ) + if wanted + ], + } + if declares: + row["params"] = {"measurement_id": SUMMED_MEASUREMENT} + if previous is not None: + row["inputs"] = [{"entity": "person", "columns": [previous]}] + previous = f"toy_step_{index}" + nodes.append(row) + nodes.append( + { + "id": "toy.calibrate", + "kernel": "calibrate.ordered_adam@1", + "population": "toy.create", + "inputs": [{"entity": "person", "columns": [previous]}], + } + ) + return { + "schema_version": 1, + "country": "toy", + "nodes": nodes, + "endpoints": {"calibration": "toy.calibrate"}, + } + + +@given( + draws=st.lists( + st.tuples(st.booleans(), st.booleans(), st.booleans()), + min_size=1, + max_size=4, + ) +) +@settings( + max_examples=60, + database=None, + deadline=None, + suppress_health_check=[HealthCheck.too_slow], +) +# Random draws rarely reach these: one declarer alone, then with a second +# writer after it and before it. +@example(draws=[(True, True, True)]) +@example(draws=[(True, True, True), (False, True, False)]) +@example(draws=[(False, False, True), (True, True, True)]) +def test_a_chain_proves_exactly_when_one_declarer_is_the_only_writer(draws): + proofs, refusals = measurement_proofs(_spec(_chain(draws))) + declarers = [index for index, draw in enumerate(draws) if draw[0]] + income = [index for index, draw in enumerate(draws) if draw[1]] + universe = [index for index, draw in enumerate(draws) if draw[2]] + proved = len(declarers) == 1 and income == universe == declarers + assert (SUMMED_MEASUREMENT in proofs) is proved + assert (SUMMED_MEASUREMENT in refusals) is not proved + if proved: + assert proofs[SUMMED_MEASUREMENT]["producers"] == { + "toy.calibrate": f"toy.n{declarers[0]}" + } + + +def test_a_calibration_endpoint_without_a_calibrate_kernel_is_a_root(): + """Only the endpoint clause of the roots finds this calibration node.""" + document = _graph() + _nodes(document)["toy.calibrate"]["kernel"] = "toy.fit@1" + proofs, refusals = measurement_proofs(_spec(document)) + assert refusals == {} + assert proofs[SUMMED_MEASUREMENT]["producers"] == {"toy.calibrate": "toy.measure"} + + +@pytest.fixture(scope="module") +def composed_toy(tmp_path_factory): + from test_support.microcosm_build.transport_composed import ( + make_composed_fixture, + ) + + return make_composed_fixture(tmp_path_factory.mktemp("composed")) + + +def _composed_spec(composed_toy, *, overwrite, expand=False): + """The composed toy with summed rows and a producer upstream of calibration. + + ``transport.currency@1`` stands in for the producer's kernel, because + composition checks the skeleton vocabulary and runs no kernel. + """ + spec = copy.deepcopy(composed_toy.spec) + base, profiles, context, intervals = summed_profile_fixture(GROUPS) + references = spec["resources"]["target_references"] + references["target_references"] += base["target_references"] + for key, value in base["hierarchy"].items(): + references["hierarchy"].setdefault(key, {}).update(value) + spec["resources"].update( + reference_profiles=profiles, + factor_context=context, + profile_intervals=intervals, + ) + graph = spec["resources"]["transport_graph"] + nodes = _nodes(graph) + calibration = graph["endpoints"]["calibration"] + (create,) = ( + row["id"] for row in graph["nodes"] if row.get("structural") == "create" + ) + opened = nodes[calibration]["base"] + graph["nodes"].insert( + 1, + { + "id": "toy.measure", + "kernel": "transport.currency@1", + "population": create, + "inputs": [{"entity": "person", "columns": ["age"]}], + "outputs": [ + {"entity": "person", "column": INCOME_COLUMN, "dtype": "Int64"}, + {"entity": "person", "column": UNIVERSE_COLUMN, "dtype": "string"}, + ], + "params": {"measurement_id": SUMMED_MEASUREMENT}, + }, + ) + if overwrite: + graph["nodes"].insert( + graph["nodes"].index(nodes[opened]) + 1, + { + **_rewriter("toy.overwrite", INCOME_COLUMN, "Int64", population=opened), + "kernel": "transport.currency@1", + }, + ) + if expand: + graph["nodes"].insert( + graph["nodes"].index(nodes[calibration]), + { + "id": "toy.expand", + "kernel": "transport.boundary@1", + "structural": "expand", + "base": opened, + "params": {"expand_cells": [["person", INCOME_COLUMN, "Int64"]]}, + }, + ) + nodes[calibration]["base"] = "toy.expand" + nodes[calibration].setdefault("inputs", []).append( + {"entity": "person", "columns": [INCOME_COLUMN, UNIVERSE_COLUMN]} + ) + return spec, calibration + + +def test_composition_refuses_a_later_version_rewrite_of_a_measured_column( + composed_toy, +): + spec, calibration = _composed_spec(composed_toy, overwrite=False) + assert measurement_proofs(spec)[0][SUMMED_MEASUREMENT]["producers"] == { + calibration: "toy.measure" + } + compose_transport_graph(spec, composed_toy.config) + spec, calibration = _composed_spec(composed_toy, overwrite=True) + with pytest.raises(ValueError) as refused: + compose_transport_graph(spec, composed_toy.config) + message = str(refused.value) + assert message.startswith("Transport spec is not activated:") + assert f"awaits measurement:{SUMMED_MEASUREMENT}" in message + assert ( + f"calibration node {calibration!r} has writers ['toy.measure', " + f"'toy.overwrite'] of person.{INCOME_COLUMN} upstream" + ) in message + + +def test_composition_refuses_an_expand_of_a_measured_column(composed_toy): + spec, calibration = _composed_spec(composed_toy, overwrite=False, expand=True) + with pytest.raises(ValueError) as refused: + compose_transport_graph(spec, composed_toy.config) + message = str(refused.value) + assert message.startswith("Transport spec is not activated:") + assert ( + f"calibration node {calibration!r} has writers ['toy.expand', " + f"'toy.measure'] of person.{INCOME_COLUMN} upstream" + ) in message diff --git a/test_support/microcosm_build/reference_factor_transport.py b/test_support/microcosm_build/reference_factor_transport.py index 1fe2f52bb4..ada34c86d4 100644 --- a/test_support/microcosm_build/reference_factor_transport.py +++ b/test_support/microcosm_build/reference_factor_transport.py @@ -9,13 +9,21 @@ ) from microcosm.build.reference_factors import resolve_reference_factor from microcosm.build.reference_profiles import ( - reference_document_sha256, resolve_reference_profiles, ) from test_support.microcosm_build.chronicle_feed import write_pinned_artifact from test_support.microcosm_build.reference_factors import factor_fixture from test_support.microcosm_build.transport_graph import reference_document +#: The proof composition returns for one ``reviewed_measurement`` producer. +REVIEWED_MEASUREMENT_PROOF = { + "reviewed_measurement": { + "measurement_id": "reviewed_measurement", + "producers": {"toy.calibrate": "toy.measure"}, + "columns": [], + } +} + def factor_transport_fixture( tmp_path, *, forged=False, activate=False, manifest_commit=False @@ -146,17 +154,17 @@ def factor_transport_fixture( base["hierarchy"]["target_labels"] = { row["name"]: f"Aged toy {row['name']}" for row in references } + # A test-only measurement proof (what composition's measurement_proofs + # returns for one producer upstream of calibration) lets the compiler test + # its own replay and arithmetic independently of this lane's parked + # country declarations. document, _, resolution = resolve_reference_profiles( - base, profiles, {}, factor_context=context + base, + profiles, + {}, + factor_context=context, + measurement_proofs=REVIEWED_MEASUREMENT_PROOF if activate else None, ) - if activate: - # Test-only future measurement closure lets the compiler test its own replay - # and arithmetic independently of this lane's parked country declarations. - for row in document["target_references"]: - row["metadata"]["activation_status"] = "active" - resolution["effective_references_sha256"] = reference_document_sha256(document) - resolution.pop("receipt_sha256") - resolution["receipt_sha256"] = reference_document_sha256(resolution) spec = { "country": "xx", "resources": { diff --git a/test_support/microcosm_build/reference_intervals.py b/test_support/microcosm_build/reference_intervals.py index ad7d8bf489..259f1a19dd 100644 --- a/test_support/microcosm_build/reference_intervals.py +++ b/test_support/microcosm_build/reference_intervals.py @@ -91,8 +91,10 @@ def interval_bindings(table, *, entity=None): } -def _factor_entry(profile): - declaration, facts, identity = factor_fixture(anchor_level=PROFILE_ANCHORS[profile]) +def _factor_entry(profile, anchor=None): + declaration, facts, identity = factor_fixture( + anchor_level=PROFILE_ANCHORS[profile] if anchor is None else anchor + ) declaration["id"] = f"toy_{profile}" factor, receipt = resolve_reference_factor(declaration, facts, identity) return identity, { @@ -188,3 +190,234 @@ def interval_profile_fixture(): base = reference_document(references) intervals = {"schema_version": 1, "tables": [table]} return base, profiles, context, intervals + + +# --------------------------------------------------------------------------- +# Summed-band money rows (G8-r2c hunk A) +# --------------------------------------------------------------------------- + +SUMMED_MEASUREMENT = "reviewed_measurement" + + +def band_fact(kind, label, value, *, period=2001, record_set="toy.bands"): + """One Ledger consumer fact for a native band's count or amount. + + Count and amount facts share one record set and one band dimension, as + a published table that reports both measures over the same classes does. + """ + key = f"toy-{kind}-{label}" + return { + "aggregate_fact_key": f"ledger.aggregate_fact.v2:{key}", + "semantic_fact_key": f"ledger.semantic_fact.v2:{key}", + "legacy_fact_key": f"ledger.fact.v1:{key}", + "label": f"Toy {kind} in band {label}", + "lineage": { + "source_record_id": f"toy.bands.{kind}.{label}", + "source_cell_keys": [f"ledger.source_cell.v1:{key}"], + "source_row_keys": [], + }, + "value": value, + "period": {"type": "year", "value": period}, + "geography": {"level": "country", "id": "XT", "name": "Toy", "vintage": "v1"}, + "entity": {"name": "person", "role": "person"}, + "observed_measure": { + "source_name": "toy_office", + "source_table": "Toy bands", + "source_measure_id": kind, + "source_concept": f"toy.{kind}", + "unit": "people" if kind == "count" else "xts", + }, + "aggregation": {"method": "sum"}, + "source": { + "source_name": "toy_office", + "source_table": "Toy bands", + "source_file": "toy.csv", + "url": "https://example.invalid/toy.csv", + "vintage": "v1", + }, + "dimensions": {"band": label}, + "dimension_labels": {"band": "Band"}, + "dimension_value_labels": {"band": {label: f"Band {label}"}}, + "universe_constraints": {"domain": "toy"}, + "layout": { + "record_set_id": record_set, + "groupby_dimension": "band", + "groupby_value_id": label, + "measure_id": kind, + }, + } + + +def summed_row(kind, name, members, *, indicator, entity="person"): + """A parked coarse row over ``members``: ``sum`` counts, ``identity`` money.""" + row = { + "name": name, + "entity": entity, + "measure": "people" if kind == "count" else "toy_income", + "filter": indicator, + "period": 2002, + "family": "toy", + "category_id": "toy.population", + "period_match_policy": "latest_not_after", + "expected_member_count": len(members), + "ledger_selector": { + "source_name": "toy_office", + "source_measure_id": kind, + "source_concept": f"toy.{kind}", + "entity_name": "person", + "period_type": "year", + "period_value": 2001, + "dimensions": ["band"], + "dimension_values": {"band": list(members)}, + }, + "metadata": { + "contract_target_id": name, + "reference_value_kind": kind, + "activation_status": "requires_prerequisites", + "required_reference_profile": "index", + "required_measurement_binding": SUMMED_MEASUREMENT, + "reference_prerequisites": json.dumps( + ["reference_profile:index", f"measurement:{SUMMED_MEASUREMENT}"] + ), + "parking_reason": "Reviewed measurement is required.", + }, + } + if kind == "count": + row["value_operation"] = "sum" + return row + + +def summed_profile_fixture( + groups, *, entity="person", table_id="toy_coarse", anchors=None +): + """Coarse references, profiles, factor context and a coarse interval table. + + ``groups`` lists the native member labels of each coarse band, in band + order; band ``b`` sums ``groups[i]``. The table has no nil band, so + every coarse band is positive and names a summed count/money pair. + """ + table = interval_table_fixture( + table_id=table_id, + entity=entity, + cuts=tuple(100_000 * (index + 1) for index in range(len(groups) - 1)), + nil=False, + companions=True, + ) + references = [] + for band, members in zip(table["bands"], groups, strict=True): + for kind in ("count", "money"): + references.append( + summed_row( + kind, + f"toy_{kind}_{band['id']}", + members, + indicator=band["indicator"], + entity=entity, + ) + ) + identity, factors = None, [] + for profile in PROFILE_ANCHORS: + identity, entry = _factor_entry(profile, (anchors or {}).get(profile)) + factors.append(entry) + context = {"schema_version": 1, "feed_identity": identity, "factors": factors} + transformation = { + "operation": "scale_money_and_intervals", + "reference_names": [ + row["name"] for row in references if row["name"].startswith("toy_money") + ], + "count_reference_names": [ + row["name"] for row in references if row["name"].startswith("toy_count") + ], + "interval_table_ids": [table["id"]], + "required_measurement_binding": SUMMED_MEASUREMENT, + } + profiles = { + "schema_version": 1, + "reference_resource": "target_references", + "factor_context_resource": "factor_context", + "interval_resource": "profile_intervals", + "dimensions": { + "index": { + "default": "central", + "profiles": { + profile: { + "transformations": [ + {**deepcopy(transformation), "factor_id": f"toy_{profile}"} + ] + } + for profile in PROFILE_ANCHORS + }, + } + }, + } + base = with_target_labels(reference_document(references)) + intervals = {"schema_version": 1, "tables": [table]} + return base, profiles, context, intervals + + +def with_target_labels(document): + """Give every row the Microcosm label a summed or scaled target needs. + + The ledger refuses a multi-fact or transformed target without an explicit + target label (``ledger_targets.py`` ``_target_label``). + """ + document["hierarchy"]["target_labels"] = { + row["metadata"]["contract_target_id"]: f"Toy target {row['name']}" + for row in document["target_references"] + } + return document + + +def proof_for(producer="toy.measure", calibration="toy.calibrate"): + """The proof composition returns for one producer upstream of calibration.""" + return { + SUMMED_MEASUREMENT: { + "measurement_id": SUMMED_MEASUREMENT, + "producers": {calibration: producer}, + "columns": [f"person.{INCOME_COLUMN}", f"person.{UNIVERSE_COLUMN}"], + } + } + + +# --------------------------------------------------------------------------- +# Pre-fit acceptance (G8-r2c hunk D, ruling R2C-3) +# --------------------------------------------------------------------------- + +#: R2C-3's frozen pre-fit thresholds, as a ``measurement_prefit`` entry +#: declares them. ``maximum_discrepant_bands`` is 2 because three or more +#: discrepant bands refuse. A change needs a new hub ruling first. +R2C3_PREFIT_THRESHOLDS = { + "roster_total_bounds": [0.75, 1.25], + "band_bounds": [0.5, 2.0], + "maximum_discrepant_bands": 2, + "minimum_source_households": 30, + "minimum_kish_ess": 30, +} + +#: Evidence aliases the toy battery wires; a country names its own. +TOY_PREFIT_EVIDENCE = [ + "native_band_report", + "omission_register", + "scenario_vsi", + "scenario_vex", + "scenario_vpt", + "identity_receipts", +] + + +def prefit_gate_entry(*, tables=("toy_coarse",), evidence=None, **overrides): + """A fixture ``gates.json`` entry carrying R2C-3's numbers.""" + return { + "id": "measurement_prefit_toy", + "gate": "measurement_prefit", + "phase": "terminal", + "criticality": "release_blocking", + "parameters": { + **deepcopy(R2C3_PREFIT_THRESHOLDS), + "tables": list(tables), + "required_evidence": list( + TOY_PREFIT_EVIDENCE if evidence is None else evidence + ), + **overrides, + }, + }