Report each QRF output's largest single record in the US tail-concentration evidence

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`

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.

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),
+    }

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"),
             )
         )