From c4fd3d66887a7a834631c7bb78cc9f00bf1aa67a Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Tue, 13 Oct 2026 14:11:55 -0400 Subject: [PATCH 1/3] Report each QRF output's largest single record in the US tail evidence The release's tail-concentration gate tests the top 100 records' share of a column's weighted absolute mass (fails above 0.75). One record can carry a fifth of a column and pass: in the #1251 rebuild one tax unit of weight 644.7 drew a $218.9M Schedule E line-32 value and carried 21.7% of partnership_income at initial weights, while the column's top-100 share was 0.399. microcosm.build.gates.single_record_concentration measures, for one column, the largest single record's share of weighted absolute mass, that record's weight, the top-k share and the carrier count, on the gate's own population. It returns measurements only: no verdict, no threshold. The US release tool's tail surface now carries it for every numeric QRF output, dense and thin ones included, as surface.single_record_concentration in qrf_tail_concentration.json. The gate's details are unchanged. The dry run's rows carry the share and the weight at base weights, and the post-export probe compares the block with a release that recorded it. Tests, docs and the changelog entry follow on this branch. Co-Authored-By: Claude Opus 5.5 --- .../src/microcosm/build/gates.py | 97 +++++++++++++++++++ .../build/us_runtime/release_gate_dry_run.py | 48 ++++++--- tools/build_us_fiscal_refresh_release.py | 29 ++++-- tools/probe_us_post_export.py | 46 ++++++++- 4 files changed, 195 insertions(+), 25 deletions(-) diff --git a/packages/microcosm-build/src/microcosm/build/gates.py b/packages/microcosm-build/src/microcosm/build/gates.py index abc494a99a..13c06c0336 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,99 @@ 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 ``partnership_income`` at initial weights, while the + column's top-100 share was 0.399 against the 0.75 threshold. + + 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/tools/build_us_fiscal_refresh_release.py b/tools/build_us_fiscal_refresh_release.py index be88543d84..052878c9a9 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,12 @@ 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. """ qrf_outputs = sorted(_qrf_imputed_source_outputs()) values: dict[str, np.ndarray] = {} @@ -7965,6 +7972,7 @@ 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]] = {} entity_weights: dict[str, np.ndarray] = {} for column in qrf_outputs: try: @@ -7981,13 +7989,18 @@ def _qrf_tail_concentration_gate( nonzero_share = float((array != 0.0).mean()) if array.size else 0.0 nonzero_shares[column] = nonzero_share nonzero_records[column] = int((np.isfinite(array) & (array != 0.0)).sum()) - if nonzero_share > US_QRF_SPARSE_NONZERO_SHARE_MAX: - dense.append(column) - continue if entity not in entity_weights: entity_weights[entity] = np.asarray( export_frame.resolve_weights(entity).values, dtype=np.float64 ) + single_record[column] = single_record_concentration( + array, + entity_weights[entity], + top_k=US_QRF_TAIL_CONCENTRATION_TOP_K, + ) + if nonzero_share > US_QRF_SPARSE_NONZERO_SHARE_MAX: + dense.append(column) + continue values[column] = array weights[column] = entity_weights[entity] gate = tail_concentration_gate( @@ -8007,6 +8020,10 @@ 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, + }, } return gate, surface @@ -8108,9 +8125,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) From a6644d62b2f7b6d4e2afbd91212e8f25dd82ac1d Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Tue, 13 Oct 2026 14:39:29 -0400 Subject: [PATCH 2/3] Test the single-record diagnostic's invariants and keep the gate path unchanged single_record_concentration gets Hypothesis properties for the share chain (0 < top-1 <= top-k <= 1), the pigeonhole floors, scale-freedom in the weights (exact for powers of two), sign and value-scale invariance, totality on any float64 input (zero or non-finite mass is reported as None, never divided by), order-freedom, monotonicity, and two differentials: against tail_concentration_gate's share and counts, and against exact rational arithmetic. Example tests pin the #1251 line-32 shape, ties, and two float cases: the gate's own top-k sum can pass the total by one rounding step (the measurement caps at 1), and a column with at most top_k carriers reports exactly 1. All ten mutants of the function are killed. The release tool once again resolves weights only where main did for sparse outputs. A dense output whose entity's weights do not resolve is listed under single_record_concentration.unmeasured with the reason, so the reporting-only measurement cannot fail a run the gate would pass. A property checks the gate's verdict, failures and details equal the shared gate on the sparse columns alone for random geometries. The dry run's verdicts, lines and details are unchanged by any values in the block, or by its absence; the probe compares the block with a release that has it. Co-Authored-By: Claude Opus 5.5 --- .../test_single_record_concentration.py | 441 ++++++++++++++++++ .../us/test_us_fiscal_refresh_builder.py | 300 ++++++++++++ .../us/test_us_post_export_probe.py | 94 ++++ .../us/test_us_release_gate_dry_run.py | 139 ++++++ tools/build_us_fiscal_refresh_release.py | 34 +- 5 files changed, 1002 insertions(+), 6 deletions(-) create mode 100644 packages/microcosm-build/tests/engine_free/shared/test_single_record_concentration.py 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..5a0a5c473a --- /dev/null +++ b/packages/microcosm-build/tests/engine_free/shared/test_single_record_concentration.py @@ -0,0 +1,441 @@ +"""``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.39, + near the rebuild's 0.399 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.6, 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.35 < result["top_k_share"] < 0.45 + 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 052878c9a9..d4251ba651 100644 --- a/tools/build_us_fiscal_refresh_release.py +++ b/tools/build_us_fiscal_refresh_release.py @@ -7962,7 +7962,10 @@ def _qrf_tail_concentration_gate( 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. + 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] = {} @@ -7973,7 +7976,9 @@ def _qrf_tail_concentration_gate( 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) @@ -7989,20 +7994,36 @@ def _qrf_tail_concentration_gate( nonzero_share = float((array != 0.0).mean()) if array.size else 0.0 nonzero_shares[column] = nonzero_share 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( export_frame.resolve_weights(entity).values, dtype=np.float64 ) + 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, ) - if nonzero_share > US_QRF_SPARSE_NONZERO_SHARE_MAX: - dense.append(column) - continue - values[column] = array - weights[column] = entity_weights[entity] gate = tail_concentration_gate( values, weights, @@ -8023,6 +8044,7 @@ def _qrf_tail_concentration_gate( "single_record_concentration": { "top_k": US_QRF_TAIL_CONCENTRATION_TOP_K, "columns": single_record, + "unmeasured": unmeasured, }, } return gate, surface From 206ce717df3c35c012afee4285bd9fae6af381bb Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Tue, 13 Oct 2026 14:43:16 -0400 Subject: [PATCH 3/3] Document the single-record diagnostic and pair each #1251 share with its weights README and a changelog entry describe surface.single_record_concentration: reported for every numeric QRF output, never graded, no share for a column without mass. The motivating figures are now paired with the weights they were measured at: in the #1251 rebuild the weight-644.7 unit carried 21.7% of line 32's weighted absolute mass at initial weights while the top 100 tax units carried 44.9%; at the certified Route A final weights the two were 4.4% and 39.9%. The first commit's message paired 21.7% with 0.399. The worked example is re-tuned to the initial-weight top-100 share. Co-Authored-By: Claude Opus 5.5 --- README.md | 12 ++++++++++++ changelog.d/us-qrf-tail-single-record-share.added.md | 1 + .../microcosm-build/src/microcosm/build/gates.py | 5 +++-- .../shared/test_single_record_concentration.py | 11 ++++++----- 4 files changed, 22 insertions(+), 7 deletions(-) create mode 100644 changelog.d/us-qrf-tail-single-record-share.added.md 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 13c06c0336..ae6a3890f0 100644 --- a/packages/microcosm-build/src/microcosm/build/gates.py +++ b/packages/microcosm-build/src/microcosm/build/gates.py @@ -3360,8 +3360,9 @@ def single_record_concentration( 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 ``partnership_income`` at initial weights, while the - column's top-100 share was 0.399 against the 0.75 threshold. + 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 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 index 5a0a5c473a..25048a21d6 100644 --- 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 @@ -293,12 +293,13 @@ 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.39, - near the rebuild's 0.399 and well under the gate's 0.75: the gate - passes, and the single-record share is what shows the draw. + 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.6, size=14_597) + 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. @@ -318,7 +319,7 @@ def test_the_1251_line_32_draw(self) -> None: 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.35 < result["top_k_share"] < 0.45 + 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: