diff --git a/README.md b/README.md index 8fad299f47..d3a2af13e8 100644 --- a/README.md +++ b/README.md @@ -398,6 +398,18 @@ reference and the stored-input verdict (`stored_inputs`), so a waiver ships with the release it waives, and a `gate_evidence` block that says of each verdict whether it is bound, skipped by flag or never evaluated. +The QRF tail gate fails a sparse column whose top 100 records carry more than +75% of its weighted absolute mass. That test cannot see one record carrying a +fifth of a column: in the #1251 rebuild one tax unit of weight 644.7 carried +21.7% of line 32's weighted absolute mass at initial weights, while the top 100 +tax units carried 44.9%. +`qrf_tail_concentration.json` therefore also reports, for every numeric QRF +output, dense and thin ones included, the largest single record's share of +the weighted absolute mass and that record's weight, beside the top-100 share +(`surface.single_record_concentration`). These numbers are reported only and +never change the gate's verdict. A column with no mass reports no share. The +dry run reports the same two numbers at base weights. + A US release may not store a column that looks like a policyengine-us variable (lowercase snake_case) unless the engine it is certified against defines that variable or `microcosm.data.stored_inputs.US_STORED_NON_VARIABLE_COLUMNS` diff --git a/changelog.d/us-qrf-tail-single-record-share.added.md b/changelog.d/us-qrf-tail-single-record-share.added.md new file mode 100644 index 0000000000..8c2578fccb --- /dev/null +++ b/changelog.d/us-qrf-tail-single-record-share.added.md @@ -0,0 +1 @@ +The US release's QRF tail evidence now reports each QRF output's largest single record. The tail gate tests the top 100 records' share of a column's weighted absolute mass, so one record can carry a fifth of a column and pass: in the #1251 rebuild one tax unit of weight 644.7 carried 21.7% of line 32's weighted absolute mass at initial weights, while the top 100 tax units carried 44.9%. `microcosm.build.gates.single_record_concentration` measures a column's top-1 share, the weight of that record, the top-k share and the carrier count, on the gate's own population. `qrf_tail_concentration.json` records it for every numeric QRF output, dense and thin ones included, under `surface.single_record_concentration`. It is reported only: it adds no threshold, and the gate's verdict and details are unchanged. A column with no mass, or a mass beyond the float range, reports no share. The release dry run carries the share and the weight on each row at base weights, and the post-export probe compares the block with a release that recorded it. diff --git a/packages/microcosm-build/src/microcosm/build/gates.py b/packages/microcosm-build/src/microcosm/build/gates.py index abc494a99a..ae6a3890f0 100644 --- a/packages/microcosm-build/src/microcosm/build/gates.py +++ b/packages/microcosm-build/src/microcosm/build/gates.py @@ -78,6 +78,7 @@ "source_coverage_gate", "source_stage_input_coverage_gate", "tail_concentration_gate", + "single_record_concentration", "target_fit_gate", "target_profile_coverage_gate", "TargetCoverageRequirement", @@ -3345,3 +3346,100 @@ def tail_concentration_gate( "dormant_exclusions": dormant_exclusions, }, ) + + +def single_record_concentration( + values: Iterable[float], + weights: Iterable[float], + *, + top_k: int = 100, +) -> dict[str, float | int | None]: + """One column's largest single record, as a share of its weighted mass. + + The measurement :func:`tail_concentration_gate` does not make. That gate + asks whether the ``top_k`` records together carry too much of a column, so + one record can carry a fifth of it and pass. In the microcosm#1251 rebuild + one tax unit of weight 644.7 drew a $218.9M Schedule E line-32 value and + carried 21.7% of the column's weighted absolute mass at initial weights, + while the top 100 tax units carried 44.9%, under the 0.75 threshold. At + the certified Route A final weights the two shares were 4.4% and 39.9%. + + Reporting only: this returns measurements, with no verdict and no + threshold. After microcosm#1259's draw bound, the worst single-draw share + in a 12-seed sweep was still 33% for collectibles gains, so a threshold + needs a justification per column. + + The population is the gate's: records whose value and weight are finite + and whose weighted absolute mass ``|value| * weight`` is positive. No + column is skipped for being thin, so a column with at most ``top_k`` + carriers reports a ``top_k_share`` of exactly 1. + + Args: + values: The column's per-record values. A signed column is measured + on absolute mass. + weights: Per-record weights, aligned with ``values``. + top_k: How many of the largest records ``top_k_share`` covers + (default 100, the gate's tail). + + Returns: + ``carriers``, the number of records with positive mass, and three + measurements: + + * ``top_1_share``: the largest record's share of the total mass. + * ``top_1_weight``: that record's weight. Among records tied for the + largest mass it is the largest weight, so record order never + changes it. + * ``top_k_share``: the ``top_k`` largest records' share. On a column + the gate checks, it equals the gate's ``top_share``. + + The three are ``None`` when the column has no share to report: its + total mass is zero (no carriers) or not finite. Otherwise + ``0 < top_1_share <= top_k_share <= 1``. Scaling every weight by one + positive factor scales ``top_1_weight`` by it and leaves both shares + unchanged, up to float rounding. + + Raises: + ValueError: If ``top_k`` is below 1 or the values and weights do not + align. + """ + if top_k < 1: + raise ValueError(f"top_k must be at least 1, got {top_k!r}.") + value_vector = np.asarray(values, dtype=np.float64).reshape(-1) + weight_vector = np.asarray(weights, dtype=np.float64).reshape(-1) + if value_vector.shape[0] != weight_vector.shape[0]: + raise ValueError( + "values and weights must align, got " + f"{value_vector.shape[0]} values vs {weight_vector.shape[0]} weights." + ) + finite = np.isfinite(value_vector) & np.isfinite(weight_vector) + carrier_weights = weight_vector[finite] + # A product or a sum past the float64 range is reported as a column with + # no share, never as a warning or a NaN. + with np.errstate(over="ignore"): + mass = np.abs(value_vector[finite]) * carrier_weights + carrying = mass > 0.0 + mass = mass[carrying] + carriers = int(mass.size) + total = float(mass.sum()) + if not carriers or not math.isfinite(total): + return { + "carriers": carriers, + "top_1_share": None, + "top_1_weight": None, + "top_k_share": None, + } + top = float(mass.max()) + # With at most top_k carriers the tail is the whole column. Above + # that, the tail is summed as the gate sums it; a sum in another + # order can pass the total by a rounding step, hence the cap. + tail = ( + total + if carriers <= top_k + else float(np.partition(mass, -top_k)[-top_k:].sum()) + ) + return { + "carriers": carriers, + "top_1_share": top / total, + "top_1_weight": float(carrier_weights[carrying][mass == top].max()), + "top_k_share": min(tail / total, 1.0), + } diff --git a/packages/microcosm-build/src/microcosm/build/us_runtime/release_gate_dry_run.py b/packages/microcosm-build/src/microcosm/build/us_runtime/release_gate_dry_run.py index 439bfd38e8..b501aafdf0 100644 --- a/packages/microcosm-build/src/microcosm/build/us_runtime/release_gate_dry_run.py +++ b/packages/microcosm-build/src/microcosm/build/us_runtime/release_gate_dry_run.py @@ -94,7 +94,7 @@ import math from collections.abc import Callable, Iterable, Mapping, Sequence -from dataclasses import dataclass +from dataclasses import dataclass, replace from typing import Any, Literal, get_args from microcosm.build.us_runtime.release_gate_preflight import PreflightReport @@ -485,6 +485,9 @@ class TailColumnDryRun: top_share: float | None = None nonzero_share: float | None = None carriers: int | None = None + # Reported only: no class or verdict reads the largest record's share. + top_1_share: float | None = None + top_1_weight: float | None = None @property def status(self) -> PreflightStatus: @@ -514,6 +517,8 @@ def to_row(self) -> dict[str, Any]: "top_share_at_base_weights": self.top_share, "nonzero_share": self.nonzero_share, "carriers_at_base_weights": self.carriers, + "top_1_share_at_base_weights": self.top_1_share, + "top_1_weight_at_base_weights": self.top_1_weight, } @@ -749,7 +754,10 @@ def qrf_tail_register_check( qrf_outputs: ``_qrf_imputed_source_outputs()``. register: The loaded per-run register (column -> reason). surface: The surface dict from ``_qrf_tail_concentration_gate``, which - must carry ``nonzero_shares`` and ``nonzero_records``. + must carry ``nonzero_shares`` and ``nonzero_records``. Its + ``single_record_concentration`` block, when present, gives each + row the largest single record's share and weight at base weights. + They are reported only; no class or verdict reads them. gate_details: That call's gate details. release_lines_at_base_weights: The lines the release would append at these weights. @@ -787,6 +795,9 @@ def qrf_tail_register_check( str(k): int(v) for k, v in dict(gate_details["thin_columns"]).items() } top_share = {str(k): float(v) for k, v in gate_details["top_share"].items()} + single_record = dict( + dict(surface.get("single_record_concentration", {})).get("columns", {}) + ) rows: list[TailColumnDryRun] = [] owners: dict[str, str] = {} @@ -797,21 +808,26 @@ def qrf_tail_register_check( owner = evidence_owner(column) if evidence_owner is not None else None if owner is not None: owners[column] = owner + measured = dict(single_record.get(column, {})) rows.append( - classify_tail_column( - column, - base_class=base_class, - in_register=column in register, - allow_concentration=allow_concentration, - top_share=top_share.get(column), - nonzero_share=nonzero_shares.get(column), - carriers=carriers, - support_fixed=support_fixed, - margins=margins, - max_top_share=float(gate["max_top_share"]), - sparse_nonzero_share_max=float(gate["sparse_nonzero_share_max"]), - min_nonzero_records=int(gate["min_nonzero_records"]), - evidence_owned=owner is not None, + replace( + classify_tail_column( + column, + base_class=base_class, + in_register=column in register, + allow_concentration=allow_concentration, + top_share=top_share.get(column), + nonzero_share=nonzero_shares.get(column), + carriers=carriers, + support_fixed=support_fixed, + margins=margins, + max_top_share=float(gate["max_top_share"]), + sparse_nonzero_share_max=float(gate["sparse_nonzero_share_max"]), + min_nonzero_records=int(gate["min_nonzero_records"]), + evidence_owned=owner is not None, + ), + top_1_share=measured.get("top_1_share"), + top_1_weight=measured.get("top_1_weight"), ) ) diff --git a/packages/microcosm-build/tests/engine_free/shared/test_single_record_concentration.py b/packages/microcosm-build/tests/engine_free/shared/test_single_record_concentration.py new file mode 100644 index 0000000000..25048a21d6 --- /dev/null +++ b/packages/microcosm-build/tests/engine_free/shared/test_single_record_concentration.py @@ -0,0 +1,442 @@ +"""``single_record_concentration``: one column's largest record, as a share. + +Invariants pinned here, each for every input the strategies reach: + +1. **Share chain.** A measured column has + ``0 < top_1_share <= top_k_share <= 1``. With ``top_k = 1`` the two shares + are the same number, and with at most ``top_k`` carriers ``top_k_share`` is + exactly 1. +2. **Pigeonhole floor.** The largest of ``n`` positive masses is at least + their mean, so ``top_1_share >= 1 / carriers``, and + ``top_k_share >= min(top_k, carriers) / carriers``. +3. **Scale-free in the weights.** Multiplying every weight by one positive + factor leaves the carrier count and both shares unchanged and multiplies + ``top_1_weight`` by the factor: exactly for a power of two, to float + rounding otherwise. Scaling or negating every value changes nothing either, + since a signed column is measured on absolute mass. +4. **Total.** Any float64 values and weights (NaN, infinities, zeros, + negative weights, products past the float64 range) give a report that is + valid JSON without NaN. A column with no carriers, or with a total mass + that is not finite, reports its three measurements as ``None``: zero mass + is reported, never divided by. +5. **Order-free.** Permuting the records leaves ``carriers`` and + ``top_1_weight`` unchanged and the shares unchanged to float rounding. + Records tied for the largest mass report the largest of their weights. +6. **Monotone.** Raising the largest record's value never lowers + ``top_1_share``; raising ``top_k`` never lowers ``top_k_share``. +7. **Differential, against the gate.** On every column + ``tail_concentration_gate`` checks, ``top_k_share`` is the gate's + ``top_share`` bit for bit and ``carriers`` its carrier count; on a thin + column ``carriers`` is the gate's thin count. +8. **Differential, against exact arithmetic.** Both shares equal the ratios of + the same float masses summed as exact rationals, to float rounding. +""" + +from __future__ import annotations + +import json +import math +from fractions import Fraction + +import numpy as np +import pytest +from hypothesis import assume, given, settings +from hypothesis import strategies as st + +from microcosm.build.gates import ( + single_record_concentration, + tail_concentration_gate, +) + +PROPERTY = settings(max_examples=300, deadline=None, database=None) +_ROUNDING = 1e-12 + +# Dollar-like magnitudes: no product or sum leaves the float64 range, and no +# positive mass rounds to zero. The totality property drops these bounds. +_VALUE = st.one_of( + st.just(0.0), + st.floats(min_value=1e-3, max_value=1e12), + st.floats(min_value=-1e12, max_value=-1e-3), +) +_WEIGHT = st.one_of(st.just(0.0), st.floats(min_value=1e-3, max_value=1e6)) +_RECORDS = st.lists(st.tuples(_VALUE, _WEIGHT), min_size=0, max_size=40) +_TOP_K = st.integers(min_value=1, max_value=12) +_ANY_FLOAT = st.floats(allow_nan=True, allow_infinity=True) +_ANY_RECORDS = st.lists(st.tuples(_ANY_FLOAT, _ANY_FLOAT), min_size=0, max_size=40) +_MEASUREMENTS = ("top_1_share", "top_1_weight", "top_k_share") + + +def _split(records) -> tuple[list[float], list[float]]: + return [value for value, _ in records], [weight for _, weight in records] + + +def _measure(records, top_k: int = 100) -> dict[str, float | int | None]: + values, weights = _split(records) + return single_record_concentration(values, weights, top_k=top_k) + + +def _carrying(records) -> list[tuple[float, float]]: + return [(value, weight) for value, weight in records if abs(value) * weight > 0.0] + + +@PROPERTY +@given(records=_RECORDS, top_k=_TOP_K) +def test_share_chain(records, top_k) -> None: + result = _measure(records, top_k) + carriers = len(_carrying(records)) + + assert result["carriers"] == carriers + if not carriers: + assert all(result[key] is None for key in _MEASUREMENTS) + return + assert 0.0 < result["top_1_share"] <= result["top_k_share"] <= 1.0 + assert result["top_1_weight"] > 0.0 + if carriers <= top_k: + assert result["top_k_share"] == 1.0 + # The pigeonhole floors, to the rounding of one float sum. + assert result["top_1_share"] >= (1.0 / carriers) * (1.0 - _ROUNDING) + assert result["top_k_share"] >= (min(top_k, carriers) / carriers) * ( + 1.0 - _ROUNDING + ) + + +@PROPERTY +@given(records=_RECORDS) +def test_top_k_of_one_is_the_top_1_share(records) -> None: + result = _measure(records, top_k=1) + + assert result["top_k_share"] == result["top_1_share"] + + +@PROPERTY +@given(records=_RECORDS, top_k=_TOP_K, exponent=st.integers(-20, 20)) +def test_scaling_weights_by_a_power_of_two_is_exact(records, top_k, exponent) -> None: + factor = math.ldexp(1.0, exponent) + values, weights = _split(records) + base = single_record_concentration(values, weights, top_k=top_k) + + scaled = single_record_concentration( + values, [weight * factor for weight in weights], top_k=top_k + ) + + assert scaled["carriers"] == base["carriers"] + assert scaled["top_1_share"] == base["top_1_share"] + assert scaled["top_k_share"] == base["top_k_share"] + if base["top_1_weight"] is None: + assert scaled["top_1_weight"] is None + else: + assert scaled["top_1_weight"] == base["top_1_weight"] * factor + + +@PROPERTY +@given( + records=_RECORDS, + top_k=_TOP_K, + factor=st.floats(min_value=1e-3, max_value=1e3), +) +def test_shares_are_scale_free_in_the_weights(records, top_k, factor) -> None: + values, weights = _split(records) + base = single_record_concentration(values, weights, top_k=top_k) + + scaled = single_record_concentration( + values, [weight * factor for weight in weights], top_k=top_k + ) + + assert scaled["carriers"] == base["carriers"] + for key in ("top_1_share", "top_k_share"): + if base[key] is None: + assert scaled[key] is None + else: + assert scaled[key] == pytest.approx(base[key], rel=1e-9, abs=0.0) + + +@PROPERTY +@given( + records=_RECORDS, + top_k=_TOP_K, + exponent=st.integers(-10, 10), + negate=st.booleans(), +) +def test_scaling_or_negating_values_changes_nothing( + records, top_k, exponent, negate +) -> None: + factor = math.ldexp(-1.0 if negate else 1.0, exponent) + values, weights = _split(records) + + assert single_record_concentration( + [value * factor for value in values], weights, top_k=top_k + ) == single_record_concentration(values, weights, top_k=top_k) + + +@PROPERTY +@given(records=_ANY_RECORDS, top_k=_TOP_K) +def test_any_floats_give_a_finite_report(records, top_k) -> None: + result = _measure(records, top_k) + + # allow_nan=False refuses NaN and both infinities. + json.dumps(result, allow_nan=False) + assert set(result) == {"carriers", *_MEASUREMENTS} + assert isinstance(result["carriers"], int) and result["carriers"] >= 0 + measured = [result[key] is not None for key in _MEASUREMENTS] + assert all(measured) or not any(measured) + if not result["carriers"]: + assert not any(measured) + if all(measured): + assert 0.0 < result["top_1_share"] <= result["top_k_share"] <= 1.0 + assert result["top_1_weight"] > 0.0 + + +@PROPERTY +@given(records=_RECORDS, top_k=_TOP_K, seed=st.integers(0, 2**32 - 1)) +def test_record_order_does_not_matter(records, top_k, seed) -> None: + order = np.random.default_rng(seed).permutation(len(records)) + base = _measure(records, top_k) + + shuffled = _measure([records[index] for index in order], top_k) + + assert shuffled["carriers"] == base["carriers"] + assert shuffled["top_1_weight"] == base["top_1_weight"] + for key in ("top_1_share", "top_k_share"): + if base[key] is None: + assert shuffled[key] is None + else: + assert shuffled[key] == pytest.approx(base[key], rel=_ROUNDING, abs=0.0) + + +@PROPERTY +@given( + records=_RECORDS, + top_k=_TOP_K, + lift=st.floats(min_value=1.0, max_value=1e3), +) +def test_a_larger_top_record_never_lowers_the_top_1_share(records, top_k, lift) -> None: + carrying = _carrying(records) + assume(carrying) + top = max(range(len(records)), key=lambda i: abs(records[i][0]) * records[i][1]) + lifted = list(records) + lifted[top] = (records[top][0] * lift, records[top][1]) + + assert _measure(lifted, top_k)["top_1_share"] >= _measure(records, top_k)[ + "top_1_share" + ] * (1.0 - _ROUNDING) + + +@PROPERTY +@given(records=_RECORDS, top_k=_TOP_K) +def test_a_larger_top_k_never_lowers_the_top_k_share(records, top_k) -> None: + assume(_carrying(records)) + + assert _measure(records, top_k + 1)["top_k_share"] >= _measure(records, top_k)[ + "top_k_share" + ] * (1.0 - _ROUNDING) + + +@PROPERTY +@given( + columns=st.dictionaries( + st.sampled_from(["a", "b", "c", "d"]), _RECORDS, min_size=1, max_size=4 + ), + top_k=st.integers(min_value=1, max_value=6), + extra=st.integers(min_value=1, max_value=6), +) +def test_agrees_with_the_tail_concentration_gate(columns, top_k, extra) -> None: + """The gate's own share and counts, on every column it checks or calls thin.""" + values = {column: _split(records)[0] for column, records in columns.items()} + weights = {column: _split(records)[1] for column, records in columns.items()} + + gate = tail_concentration_gate( + values, weights, top_k=top_k, min_nonzero_records=top_k + extra + ) + + checked = gate.details["top_share"] + thin = gate.details["thin_columns"] + assert set(checked) | set(thin) == set(columns) + for column in columns: + result = single_record_concentration( + values[column], weights[column], top_k=top_k + ) + if column in thin: + assert result["carriers"] == thin[column] + continue + assert result["carriers"] == gate.details["carrier_counts"][column] + # The gate sums its tail and its total in different orders, so its + # share can pass 1 by a rounding step; the measurement caps it. + assert result["top_k_share"] == min(checked[column], 1.0) + + +@PROPERTY +@given(records=_RECORDS, top_k=_TOP_K) +def test_agrees_with_exact_rational_arithmetic(records, top_k) -> None: + masses = sorted( + (Fraction(abs(value) * weight) for value, weight in _carrying(records)), + reverse=True, + ) + assume(masses) + total = sum(masses) + result = _measure(records, top_k) + + assert result["top_1_share"] == pytest.approx( + float(masses[0] / total), rel=_ROUNDING, abs=0.0 + ) + assert result["top_k_share"] == pytest.approx( + float(sum(masses[:top_k]) / total), rel=_ROUNDING, abs=0.0 + ) + top_weights = [ + weight + for value, weight in _carrying(records) + if Fraction(abs(value) * weight) == masses[0] + ] + assert result["top_1_weight"] == max(top_weights) + + +class TestExamples: + def test_the_1251_line_32_draw(self) -> None: + """One weight-644.7 tax unit draws $218.9M: $141B, 21.7% of the column. + + The other 14,597 carriers are spread so the top 100 hold about 0.45, + as in the rebuild at initial weights (0.449) and well under the gate's + 0.75: the gate passes, and the single-record share is what shows the + draw. + """ + rng = np.random.default_rng(1251) + body_values = rng.lognormal(mean=10.0, sigma=1.8, size=14_597) + body_weights = rng.uniform(100.0, 900.0, size=14_597) + draw_mass = 218.9e6 * 644.7 + # Scale the body so the draw is 21.7% of the column's mass. + body_values *= (draw_mass * (1 - 0.217) / 0.217) / float( + body_values @ body_weights + ) + values = np.concatenate([body_values, [218.9e6]]) + weights = np.concatenate([body_weights, [644.7]]) + + result = single_record_concentration(values, weights) + gate = tail_concentration_gate( + {"partnership_income": values}, {"partnership_income": weights} + ) + + assert gate.passed + assert result["carriers"] == 14_598 + assert result["top_1_share"] == pytest.approx(0.217, abs=1e-9) + assert result["top_1_weight"] == 644.7 + assert result["top_k_share"] == gate.details["top_share"]["partnership_income"] + assert 0.40 < result["top_k_share"] < 0.50 + assert result["top_1_share"] < result["top_k_share"] < 0.75 + + def test_a_signed_column_is_measured_on_absolute_mass(self) -> None: + result = single_record_concentration([-60.0, 30.0, 10.0], [1.0, 1.0, 1.0]) + + assert result == { + "carriers": 3, + "top_1_share": 0.6, + "top_1_weight": 1.0, + "top_k_share": 1.0, + } + + def test_the_weight_decides_the_largest_record(self) -> None: + # The smaller value on the heavier record carries more mass. + result = single_record_concentration([100.0, 30.0], [1.0, 10.0]) + + assert result["top_1_share"] == pytest.approx(300.0 / 400.0) + assert result["top_1_weight"] == 10.0 + + def test_tied_records_report_the_largest_weight(self) -> None: + # Masses 6, 6 and 6 from weights 3, 2 and 6, in any order. + for values, weights in ( + ([2.0, 3.0, 1.0], [3.0, 2.0, 6.0]), + ([1.0, 2.0, 3.0], [6.0, 3.0, 2.0]), + ): + result = single_record_concentration(values, weights) + + assert result["top_1_share"] == pytest.approx(1.0 / 3.0) + assert result["top_1_weight"] == 6.0 + + def test_top_k_share_counts_only_the_top_k(self) -> None: + values = np.concatenate([np.full(4, 10.0), np.full(6, 1.0)]) + + result = single_record_concentration(values, np.ones(10), top_k=4) + + assert result["top_1_share"] == pytest.approx(10.0 / 46.0) + assert result["top_k_share"] == pytest.approx(40.0 / 46.0) + + @pytest.mark.parametrize( + ("values", "weights"), + [ + ([], []), + ([0.0, 0.0, 0.0], [1.0, 2.0, 3.0]), + ([5.0, 7.0], [0.0, 0.0]), + ([float("nan"), float("inf")], [1.0, 1.0]), + ([5.0], [float("nan")]), + ([5.0, 7.0], [-1.0, -2.0]), + ], + ) + def test_a_zero_mass_column_is_reported_not_divided(self, values, weights) -> None: + assert single_record_concentration(values, weights) == { + "carriers": 0, + "top_1_share": None, + "top_1_weight": None, + "top_k_share": None, + } + + @pytest.mark.parametrize( + ("values", "weights"), + [ + # One product past the float64 range. + ([1e200, 1.0], [1e200, 1.0]), + # Finite products whose sum is past it. + ([1.5e308, 1.5e308], [1.0, 1.0]), + ], + ) + def test_a_mass_past_the_float_range_is_reported_without_a_share( + self, values, weights + ) -> None: + with np.errstate(over="raise"): + result = single_record_concentration(values, weights) + + assert result == { + "carriers": 2, + "top_1_share": None, + "top_1_weight": None, + "top_k_share": None, + } + + def test_non_finite_records_are_dropped_like_the_gate_drops_them(self) -> None: + result = single_record_concentration( + [float("nan"), float("inf"), 3.0, 1.0], [1.0, 1.0, 2.0, float("inf")] + ) + + assert result == { + "carriers": 1, + "top_1_share": 1.0, + "top_1_weight": 2.0, + "top_k_share": 1.0, + } + + def test_a_rounding_step_never_lifts_the_top_k_share_past_1(self) -> None: + """The gate sums its tail in partition order and its total in array + order; here the tail comes out one rounding step above the total.""" + values = [2.0, 3.0, 2.0, 1.0, 1e16, 7.0, 1e15, 1.0, 0.5] + weights = [1.0] * len(values) + gate = tail_concentration_gate( + {"column": values}, {"column": weights}, top_k=8, min_nonzero_records=9 + ) + + result = single_record_concentration(values, weights, top_k=8) + + assert gate.details["top_share"]["column"] > 1.0 + assert result["top_k_share"] == 1.0 + + def test_top_k_share_is_exactly_1_when_every_carrier_is_in_the_top_k( + self, + ) -> None: + # Summed in partition order these masses give 0.9999999999999998. + values = [2.0, 1e15, 4.5e15, 1e16, 1.0, 1e15, 4.5e15] + + result = single_record_concentration(values, [1.0] * 7, top_k=7) + + assert result["top_k_share"] == 1.0 + + def test_invalid_arguments_are_rejected(self) -> None: + with pytest.raises(ValueError, match="top_k"): + single_record_concentration([1.0], [1.0], top_k=0) + with pytest.raises(ValueError, match="align"): + single_record_concentration([1.0, 2.0], [1.0]) diff --git a/packages/microcosm-build/tests/engine_free/us/test_us_fiscal_refresh_builder.py b/packages/microcosm-build/tests/engine_free/us/test_us_fiscal_refresh_builder.py index a1d494147c..42227dbe78 100644 --- a/packages/microcosm-build/tests/engine_free/us/test_us_fiscal_refresh_builder.py +++ b/packages/microcosm-build/tests/engine_free/us/test_us_fiscal_refresh_builder.py @@ -11,10 +11,13 @@ from types import SimpleNamespace import h5py +import hypothesis import numpy as np import pandas as pd import pytest +from hypothesis import strategies as st +from microcosm.build.gates import single_record_concentration, tail_concentration_gate from microcosm.build.us_runtime import h5_io as builder_h5_io from microcosm.calibrate import ( CalibrationHierarchy, @@ -15099,6 +15102,303 @@ def test_qrf_tail_concentration_gate_passes_dispersed_sparse_mass() -> None: assert gate.passed +# --- Single-record concentration: reported beside the gate, never graded --- + +#: The gate's details keys before the single-record diagnostic. The +#: diagnostic lives in the surface, so the details keep exactly these. +_TAIL_GATE_DETAIL_KEYS = { + "columns_checked", + "top_k", + "max_top_share", + "min_nonzero_records", + "top_share", + "carrier_counts", + "thin_columns", + "reviewed_exclusions", + "stale_exclusions", + "dormant_exclusions", +} + + +def _qrf_person_frame( + builder, + columns: dict[str, np.ndarray], + household_weights: np.ndarray, + *, + household_columns: dict[str, np.ndarray] | None = None, + tax_unit_weights: np.ndarray | None = None, +) -> Frame: + n = int(household_weights.size) + ids = np.arange(1, n + 1, dtype="int64") + person = pd.DataFrame( + { + "person_id": ids, + "person_household_id": ids, + "person_tax_unit_id": ids, + "person_spm_unit_id": ids, + "person_family_id": ids, + "person_marital_unit_id": ids, + **columns, + } + ) + weights = { + "household": builder.Weights( + values=np.asarray(household_weights, dtype=np.float64), + kind=WeightKind.DESIGN, + ) + } + if tax_unit_weights is not None: + weights["tax_unit"] = builder.Weights( + values=np.asarray(tax_unit_weights, dtype=np.float64), + kind=WeightKind.DESIGN, + ) + return Frame( + { + "person": person, + "household": pd.DataFrame( + {"household_id": ids, **(household_columns or {})} + ), + "tax_unit": pd.DataFrame({"tax_unit_id": ids}), + "spm_unit": pd.DataFrame({"spm_unit_id": ids}), + "family": pd.DataFrame({"family_id": ids}), + "marital_unit": pd.DataFrame({"marital_unit_id": ids}), + }, + builder.US_SCHEMA, + weights, + ) + + +def _single_record_fixture(builder) -> tuple[Frame, np.ndarray]: + """One record carries 57% of a checked column whose gate passes. + + ``non_sch_d_capital_gains``: 500 carriers of 12,000 records. Record 0 has + weight 2 and value $1M, so mass $2M of $3,486,521 (57.4%); the top 100 + carry 65.8%, under the 0.75 threshold. ``taxable_interest_income`` is + dense, ``casualty_loss`` thin (10 carriers), ``estate_income`` all zero, + ``business_is_sstb`` boolean, and ``short_term_capital_gains`` absent. + """ + n = 12_000 + weights = np.ones(n) + weights[0] = 2.0 + non_sch_d = np.zeros(n) + non_sch_d[:500] = 2_979.0 + non_sch_d[0] = 1_000_000.0 + interest = np.zeros(n) + interest[: n // 2] = 1_000.0 + interest[1] = 50_000.0 + casualty = np.zeros(n) + casualty[100:110] = 500.0 + frame = _qrf_person_frame( + builder, + { + "non_sch_d_capital_gains": non_sch_d, + "taxable_interest_income": interest, + "casualty_loss": casualty, + "estate_income": np.zeros(n), + "business_is_sstb": np.zeros(n, dtype=bool), + }, + weights, + ) + return frame, weights + + +def test_qrf_tail_surface_reports_the_largest_single_record() -> None: + builder = _load_builder_module() + frame, weights = _single_record_fixture(builder) + + gate, surface = builder._qrf_tail_concentration_gate(frame) + + # The gate is unchanged: it passes, and its details are the gate's own. + assert gate.passed + assert set(gate.details) == _TAIL_GATE_DETAIL_KEYS + sparse = ["casualty_loss", "estate_income", "non_sch_d_capital_gains"] + direct = tail_concentration_gate( + {column: frame.table("person")[column].to_numpy() for column in sparse}, + {column: weights for column in sparse}, + top_k=builder.US_QRF_TAIL_CONCENTRATION_TOP_K, + max_top_share=builder.US_QRF_TAIL_CONCENTRATION_MAX_TOP_SHARE, + min_nonzero_records=builder.US_QRF_TAIL_CONCENTRATION_MIN_NONZERO_RECORDS, + ) + assert gate.details == direct.details + assert surface["checked_sparse_columns"] == sparse + + block = surface["single_record_concentration"] + assert block["top_k"] == 100 + assert block["unmeasured"] == {} + # Every numeric output is measured, dense and thin ones included. + assert ( + set(block["columns"]) + == set(surface["nonzero_shares"]) + == { + *sparse, + "taxable_interest_income", + } + ) + columns = block["columns"] + assert columns["non_sch_d_capital_gains"] == { + "carriers": 500, + "top_1_share": pytest.approx(2_000_000.0 / 3_486_521.0), + "top_1_weight": 2.0, + "top_k_share": gate.details["top_share"]["non_sch_d_capital_gains"], + } + assert gate.details["top_share"]["non_sch_d_capital_gains"] == pytest.approx( + 2_294_921.0 / 3_486_521.0 + ) + assert columns["taxable_interest_income"] == { + "carriers": 6_000, + # Record 0 (weight 2) carries $2,000 here: 5,998 x $1,000 + $2,000 + # + $50,000, and the top 100 are $50,000 + $2,000 + 98 x $1,000. + "top_1_share": pytest.approx(50_000.0 / 6_050_000.0), + "top_1_weight": 1.0, + "top_k_share": pytest.approx(150_000.0 / 6_050_000.0), + } + assert columns["casualty_loss"] == { + "carriers": 10, + "top_1_share": pytest.approx(0.1), + "top_1_weight": 1.0, + "top_k_share": 1.0, + } + # A zero-mass column is reported, not divided by. + assert columns["estate_income"] == { + "carriers": 0, + "top_1_share": None, + "top_1_weight": None, + "top_k_share": None, + } + + +def test_qrf_tail_evidence_file_carries_the_single_record_block(tmp_path) -> None: + builder = _load_builder_module() + frame, _ = _single_record_fixture(builder) + _, surface = builder._qrf_tail_concentration_gate(frame) + failures: list[str] = [] + + _record_qrf_tail( + builder, tmp_path, frame, register=None, allow=False, failures=failures + ) + + assert failures == [] + + def refuse(constant: str) -> None: + raise AssertionError(f"non-finite {constant} in the evidence file") + + payload = json.loads( + (tmp_path / "qrf_tail_concentration.json").read_text(), + parse_constant=refuse, + ) + assert payload["surface"]["single_record_concentration"] == json.loads( + json.dumps(surface["single_record_concentration"]) + ) + assert set(payload["tail_concentration"]["details"]) == _TAIL_GATE_DETAIL_KEYS + + +def test_a_dense_output_with_unresolvable_weights_is_unmeasured_not_fatal() -> None: + """Weighted on two group entities, person weights do not resolve. The gate + never resolves weights for a dense output, so neither may its measurement + fail the run; a sparse person output still raises, as it always did.""" + builder = _load_builder_module() + n = 12_000 + interest = np.zeros(n) + interest[: n // 2] = 1_000.0 + auto_loans = np.zeros(n) + auto_loans[:500] = 4_000.0 + frame = _qrf_person_frame( + builder, + {"taxable_interest_income": interest}, + np.ones(n), + household_columns={"auto_loan_balance": auto_loans}, + tax_unit_weights=np.ones(n), + ) + + gate, surface = builder._qrf_tail_concentration_gate(frame) + + assert gate.passed + assert surface["dense_columns"] == ["taxable_interest_income"] + assert surface["checked_sparse_columns"] == ["auto_loan_balance"] + block = surface["single_record_concentration"] + assert set(block["columns"]) == {"auto_loan_balance"} + assert block["columns"]["auto_loan_balance"]["top_1_share"] == pytest.approx( + 1 / 500 + ) + assert set(block["unmeasured"]) == {"taxable_interest_income"} + assert block["unmeasured"]["taxable_interest_income"].startswith( + "ValueError: Cannot resolve person-level weights" + ) + + sparse_person = np.zeros(n) + sparse_person[:500] = 2_979.0 + broken = _qrf_person_frame( + builder, + {"taxable_interest_income": interest, "casualty_loss": sparse_person}, + np.ones(n), + tax_unit_weights=np.ones(n), + ) + with pytest.raises(ValueError, match="Cannot resolve person-level weights"): + builder._qrf_tail_concentration_gate(broken) + + +def test_qrf_tail_single_record_measurement_never_moves_the_gate() -> None: + """For any column geometry the strategies reach: the gate's details equal + the shared gate run on the sparse columns alone; every numeric output is + measured exactly once, with the shared measurement at the frame's + weights; and a checked column's top-k share is the gate's own.""" + builder = _load_builder_module() + n = 12_000 + max_share = builder.US_QRF_SPARSE_NONZERO_SHARE_MAX + columns = ("non_sch_d_capital_gains", "casualty_loss", "estate_income") + + @hypothesis.settings(max_examples=40, deadline=None, database=None) + @hypothesis.given( + carriers=st.tuples(*(st.integers(0, 1_200) for _ in columns)), + sigma=st.floats(min_value=0.1, max_value=4.0), + seed=st.integers(0, 2**32 - 1), + ) + def check(carriers, sigma, seed) -> None: + rng = np.random.default_rng(seed) + weights = rng.uniform(0.5, 900.0, size=n) + values: dict[str, np.ndarray] = {} + for column, count in zip(columns, carriers, strict=True): + array = np.zeros(n) + chosen = rng.choice(n, size=count, replace=False) + array[chosen] = rng.lognormal(8.0, sigma, size=count) * rng.choice( + [-1.0, 1.0], size=count + ) + values[column] = array + frame = _qrf_person_frame(builder, values, weights) + + gate, surface = builder._qrf_tail_concentration_gate(frame) + + sparse = { + column: array + for column, array in values.items() + if (array != 0.0).mean() <= max_share + } + direct = tail_concentration_gate( + sparse, + {column: weights for column in sparse}, + top_k=builder.US_QRF_TAIL_CONCENTRATION_TOP_K, + max_top_share=builder.US_QRF_TAIL_CONCENTRATION_MAX_TOP_SHARE, + min_nonzero_records=(builder.US_QRF_TAIL_CONCENTRATION_MIN_NONZERO_RECORDS), + ) + assert gate.passed == direct.passed + assert gate.failures == direct.failures + assert gate.details == direct.details + block = surface["single_record_concentration"] + assert block["unmeasured"] == {} + assert set(block["columns"]) == set(columns) + for column, array in values.items(): + measured = block["columns"][column] + assert measured == single_record_concentration(array, weights) + if column in gate.details["top_share"]: + assert measured["top_k_share"] == min( + gate.details["top_share"][column], 1.0 + ) + if measured["carriers"]: + assert 0.0 < measured["top_1_share"] <= measured["top_k_share"] <= 1.0 + + check() + + def test_allow_qrf_tail_concentration_flag_parses(monkeypatch) -> None: builder = _load_builder_module() diff --git a/packages/microcosm-build/tests/engine_free/us/test_us_post_export_probe.py b/packages/microcosm-build/tests/engine_free/us/test_us_post_export_probe.py index 81d866f646..89a2c3daa7 100644 --- a/packages/microcosm-build/tests/engine_free/us/test_us_post_export_probe.py +++ b/packages/microcosm-build/tests/engine_free/us/test_us_post_export_probe.py @@ -780,3 +780,97 @@ def test_a_malformed_reference_smoke_file_costs_only_the_comparison( assert "malformed" in smoke["reference_error"] assert len(smoke["probes"]) == len(fixture_engine_probes()) assert all("reference" not in row for row in smoke["probes"]) + + +def _tail_result(single_record=None, *, passed=True, share=0.4): + surface = {"checked_sparse_columns": ["partnership_income"]} + if single_record is not None: + surface["single_record_concentration"] = { + "top_k": 100, + "columns": single_record, + "unmeasured": {}, + } + return { + "passed": passed, + "surface": surface, + "details": {"top_share": {"partnership_income": share}}, + } + + +def _tail_reference(single_record=None, **options): + result = _tail_result(single_record, **options) + return { + "surface": result["surface"], + "tail_concentration": { + "passed": result["passed"], + "details": result["details"], + }, + } + + +_LINE_32 = { + "carriers": 14_598, + "top_1_share": 0.217, + "top_1_weight": 644.7, + "top_k_share": 0.399, +} +_ZERO_MASS = { + "carriers": 0, + "top_1_share": None, + "top_1_weight": None, + "top_k_share": None, +} + + +def test_the_probe_compares_single_record_measurements(probe_tool) -> None: + measured = {"partnership_income": _LINE_32, "estate_income": _ZERO_MASS} + + same = probe_tool.compare_qrf_tail( + _tail_result(measured), _tail_reference(measured) + ) + + assert same == { + "differences": [], + "columns_compared": 1, + "single_record_columns_compared": 2, + } + + moved = probe_tool.compare_qrf_tail( + _tail_result( + {**measured, "partnership_income": {**_LINE_32, "top_1_weight": 1.08}} + ), + _tail_reference(measured), + ) + + assert moved["differences"] == ["partnership_income: top_1_weight 1.08 vs 644.7"] + + lost = probe_tool.compare_qrf_tail( + _tail_result({**measured, "estate_income": {**_ZERO_MASS, "top_1_share": 0.5}}), + _tail_reference(measured), + ) + + assert lost["differences"] == ["estate_income: top_1_share 0.5 vs None"] + + narrower = probe_tool.compare_qrf_tail( + _tail_result({"partnership_income": _LINE_32}), _tail_reference(measured) + ) + + assert narrower["differences"] == [ + "single-record columns differ: only probe [], only release ['estate_income']" + ] + assert narrower["single_record_columns_compared"] == 1 + + +def test_a_release_without_single_record_measurements_compares_none( + probe_tool, +) -> None: + """A release built before its tail surface carried the block (Route A).""" + comparison = probe_tool.compare_qrf_tail( + _tail_result({"partnership_income": _LINE_32}), _tail_reference() + ) + + assert comparison == { + "differences": [], + "columns_compared": 1, + "single_record_columns_compared": 0, + } diff --git a/packages/microcosm-build/tests/engine_free/us/test_us_release_gate_dry_run.py b/packages/microcosm-build/tests/engine_free/us/test_us_release_gate_dry_run.py index c0c8a2c27c..87c28c5cd5 100644 --- a/packages/microcosm-build/tests/engine_free/us/test_us_release_gate_dry_run.py +++ b/packages/microcosm-build/tests/engine_free/us/test_us_release_gate_dry_run.py @@ -807,6 +807,145 @@ def check(removed, stale, unused, allow): check() +_SINGLE_RECORD_ROW_KEYS = ( + "top_1_share_at_base_weights", + "top_1_weight_at_base_weights", +) + + +def _register_check(builder, surface, gate, register, *, allow, margins): + mismatch = builder._qrf_tail_register_mismatch(register, gate) + gate_lines, register_lines = builder._qrf_tail_gate_lines( + gate, mismatch, allow_concentration=allow + ) + return dry.qrf_tail_register_check( + qrf_outputs=sorted(builder._qrf_imputed_source_outputs()), + register=register, + surface=surface, + gate_details=gate.details, + release_lines_at_base_weights=[*gate_lines, *register_lines], + support_fixed=True, + margins=margins, + allow_concentration=allow, + register_source={}, + ) + + +def test_rows_carry_the_single_record_measurements(builder, tail_frame) -> None: + register = {column: "reviewed" for column in _OVER} + gate, surface = builder._qrf_tail_concentration_gate( + tail_frame, reviewed_exclusions=register + ) + + check = _register_check( + builder, surface, gate, register, allow=False, margins=dry.DryRunMargins() + ) + + measured = surface["single_record_concentration"]["columns"] + rows = {row["column"]: row for row in check.rows} + for column, row in rows.items(): + if column in measured: + assert ( + row["top_1_share_at_base_weights"] == (measured[column]["top_1_share"]) + ) + assert ( + row["top_1_weight_at_base_weights"] + == (measured[column]["top_1_weight"]) + ) + else: + assert row["top_1_share_at_base_weights"] is None + assert row["top_1_weight_at_base_weights"] is None + # 100 records of $594,484 and 450 of $2,979, all of weight 1. + assert rows["estate_income"]["top_1_share_at_base_weights"] == pytest.approx( + 594_484.0 / (100 * 594_484.0 + 450 * 2_979.0) + ) + assert set(_BOOLEAN) | set(_ABSENT) <= { + column for column in rows if column not in measured + } + + +def test_single_record_measurements_never_move_a_dry_run_verdict( + builder, tail_frame +) -> None: + """Reported only: any values in the block, or no block at all (a release + built before it existed), give the same verdicts, lines and details; the + rows differ only in the two reported fields.""" + + hypothesis, st = _hypothesis() + gate_base, surface = builder._qrf_tail_concentration_gate(tail_frame) + columns = sorted(surface["single_record_concentration"]["columns"]) + share = st.one_of(st.none(), st.floats(min_value=0.0, max_value=1.0)) + weight = st.one_of(st.none(), st.floats(min_value=0.0, max_value=1e6)) + + @hypothesis.settings(max_examples=60, deadline=None) + @hypothesis.given( + replacement=st.one_of( + st.none(), + st.fixed_dictionaries( + { + column: st.fixed_dictionaries( + { + "carriers": st.integers(0, 10_000), + "top_1_share": share, + "top_1_weight": weight, + "top_k_share": share, + } + ) + for column in columns + } + ), + ), + register=st.sets(st.sampled_from((*_OVER, *_UNDER, *_THIN))), + allow=st.booleans(), + exact=st.booleans(), + ) + def check(replacement, register, allow, exact): + entries = {column: "reviewed" for column in register} + gate, original = builder._qrf_tail_concentration_gate( + tail_frame, reviewed_exclusions=entries + ) + changed = dict(original) + if replacement is None: + del changed["single_record_concentration"] + else: + changed["single_record_concentration"] = { + "top_k": 100, + "columns": replacement, + "unmeasured": {}, + } + margins = dry.DryRunMargins.exact() if exact else dry.DryRunMargins() + + before = _register_check( + builder, original, gate, entries, allow=allow, margins=margins + ) + after = _register_check( + builder, changed, gate, entries, allow=allow, margins=margins + ) + + assert after.status == before.status + assert after.summary == before.summary + assert after.failures == before.failures + assert after.at_risks == before.at_risks + assert after.details == before.details + + def without(rows): + return [ + {k: v for k, v in row.items() if k not in _SINGLE_RECORD_ROW_KEYS} + for row in rows + ] + + assert without(after.rows) == without(before.rows) + if replacement is None: + assert all( + row[key] is None + for row in after.rows + for key in _SINGLE_RECORD_ROW_KEYS + ) + + assert gate_base.details["top_share"] + check() + + def test_a_surface_that_misplaces_an_output_is_refused() -> None: with pytest.raises(ValueError, match="exactly once"): dry.base_tail_classes( diff --git a/tools/build_us_fiscal_refresh_release.py b/tools/build_us_fiscal_refresh_release.py index be88543d84..d4251ba651 100644 --- a/tools/build_us_fiscal_refresh_release.py +++ b/tools/build_us_fiscal_refresh_release.py @@ -62,6 +62,7 @@ nonconstant_columns_gate, observed_value_counts, parity_gate, + single_record_concentration, tail_concentration_gate, target_fit_gate, target_profile_coverage_gate, @@ -7956,6 +7957,15 @@ def _qrf_tail_concentration_gate( absent, or non-numeric, and each numeric output's record nonzero share and finite nonzero record count, the numbers behind the dense/sparse split) for the release artifact and the ``--dry-run-gates-report`` certainty model. + + The surface also measures every numeric output, dense and thin ones + included, with :func:`microcosm.build.gates.single_record_concentration`: + the largest single record's share of the column's weighted absolute mass + and that record's weight, next to the top-k share. It is reported only and + never changes the gate's verdict. A dense output whose entity's weights do + not resolve is listed under ``unmeasured`` with the reason; the gate never + resolves weights for dense outputs, so the measurement cannot raise where + the gate would not. """ qrf_outputs = sorted(_qrf_imputed_source_outputs()) values: dict[str, np.ndarray] = {} @@ -7965,7 +7975,10 @@ def _qrf_tail_concentration_gate( non_numeric: list[str] = [] nonzero_shares: dict[str, float] = {} nonzero_records: dict[str, int] = {} + single_record: dict[str, dict[str, float | int | None]] = {} + unmeasured: dict[str, str] = {} entity_weights: dict[str, np.ndarray] = {} + weight_errors: dict[str, str] = {} for column in qrf_outputs: try: entity = export_frame.column_entity(column) @@ -7983,6 +7996,22 @@ def _qrf_tail_concentration_gate( nonzero_records[column] = int((np.isfinite(array) & (array != 0.0)).sum()) if nonzero_share > US_QRF_SPARSE_NONZERO_SHARE_MAX: dense.append(column) + if entity not in entity_weights and entity not in weight_errors: + try: + entity_weights[entity] = np.asarray( + export_frame.resolve_weights(entity).values, + dtype=np.float64, + ) + except ValueError as error: + weight_errors[entity] = f"{type(error).__name__}: {error}" + if entity in weight_errors: + unmeasured[column] = weight_errors[entity] + else: + single_record[column] = single_record_concentration( + array, + entity_weights[entity], + top_k=US_QRF_TAIL_CONCENTRATION_TOP_K, + ) continue if entity not in entity_weights: entity_weights[entity] = np.asarray( @@ -7990,6 +8019,11 @@ def _qrf_tail_concentration_gate( ) values[column] = array weights[column] = entity_weights[entity] + single_record[column] = single_record_concentration( + array, + entity_weights[entity], + top_k=US_QRF_TAIL_CONCENTRATION_TOP_K, + ) gate = tail_concentration_gate( values, weights, @@ -8007,6 +8041,11 @@ def _qrf_tail_concentration_gate( "sparse_nonzero_share_max": US_QRF_SPARSE_NONZERO_SHARE_MAX, "nonzero_shares": nonzero_shares, "nonzero_records": nonzero_records, + "single_record_concentration": { + "top_k": US_QRF_TAIL_CONCENTRATION_TOP_K, + "columns": single_record, + "unmeasured": unmeasured, + }, } return gate, surface @@ -8108,9 +8147,9 @@ def _record_qrf_tail_concentration_gate( flag says) to ``terminal_gate_failures``, so the batched pre-export raise refuses the run with the #568 weight sidecar on disk. Whenever the gate evaluated, ``qrf_tail_concentration.json`` is written with the per-column - shares, carrier counts, and the mismatch. Only a genuine evaluation crash - escapes: it propagates on an otherwise-clean run and becomes one unowned - failure line under earlier failures. + shares, carrier counts, single-record shares, and the mismatch. Only a + genuine evaluation crash escapes: it propagates on an otherwise-clean run + and becomes one unowned failure line under earlier failures. Returns the register-mismatch failure lines (empty when the register matches) so --evidence-release can refuse them. diff --git a/tools/probe_us_post_export.py b/tools/probe_us_post_export.py index 746b14cb2d..69e4ae7afe 100644 --- a/tools/probe_us_post_export.py +++ b/tools/probe_us_post_export.py @@ -960,7 +960,10 @@ def compare_qrf_tail( ) -> dict[str, Any]: """The probe's source-export QRF tail result against a release's ``qrf_tail_concentration.json`` (same frame columns and weights, so the - shares must agree to rounding).""" + shares must agree to rounding). + + The single-record measurements are compared when the release recorded + them; a release built before its surface carried them compares none.""" differences: list[str] = [] tail = reference["tail_concentration"] if bool(tail["passed"]) != bool(result["passed"]): @@ -984,7 +987,39 @@ def compare_qrf_tail( continue if not math.isclose(float(left), float(right), rel_tol=1e-9, abs_tol=1e-12): differences.append(f"{column}: top share {left!r} vs {right!r}") - return {"differences": differences, "columns_compared": len(shares)} + measured = _single_record_columns(result["surface"]) + recorded = _single_record_columns(reference["surface"]) + if recorded is None or measured is None: + measured, recorded = {}, {} + elif sorted(measured) != sorted(recorded): + differences.append( + "single-record columns differ: only probe " + f"{sorted(set(measured) - set(recorded))}, only release " + f"{sorted(set(recorded) - set(measured))}" + ) + compared = sorted(set(measured) & set(recorded)) + for column in compared: + for field in ("carriers", "top_1_share", "top_1_weight", "top_k_share"): + left, right = measured[column].get(field), recorded[column].get(field) + if left is None or right is None: + if left != right: + differences.append(f"{column}: {field} {left} vs {right}") + continue + if not math.isclose(float(left), float(right), rel_tol=1e-9, abs_tol=1e-12): + differences.append(f"{column}: {field} {left!r} vs {right!r}") + return { + "differences": differences, + "columns_compared": len(shares), + "single_record_columns_compared": len(compared), + } + + +def _single_record_columns( + surface: Mapping[str, Any], +) -> Mapping[str, Mapping[str, Any]] | None: + """A tail surface's per-column single-record measurements, if it has them.""" + block = surface.get("single_record_concentration") + return None if block is None else block["columns"] def compare_post_export_plans( @@ -1824,7 +1859,12 @@ def evaluate(frame) -> dict[str, Any]: "pass" if not comparison["differences"] else "fail", AUTHORITATIVE, "; ".join(comparison["differences"]) - or "same verdict, columns and top-k shares", + or ( + "same verdict, columns, top-k shares and single-record " + "measurements" + if comparison["single_record_columns_compared"] + else "same verdict, columns and top-k shares" + ), "fidelity of this probe to the release", ) _write_json(self.out_dir / "qrf_tail_concentration.json", record)