Add waterfall-progression regression test (incremental bucket flags)

bedrock/transform/__tests__/test_waterfall_progression.py

@@ -0,0 +1,105 @@
+"""Waterfall-progression regression: incremental bucket flags reproduce the release steps.
+
+The v0→v0.3 release waterfall's bedrock-side steps are, in bucket-flag
+vocabulary (the ``v03_waterfall_*`` configs):
+
+    v0 snapshot baseline
+      → + use_cornerstone_ghg_model            (G1a: "GHG model allocation")
+      → + implement_waste_disaggregation       (G1b: "Waste disagg.")
+      → + apply_io_year_adjustments + margins  (G2:  "IO year adjustments")
+      → usa_ghg_data_year: 2023 → 2024         (G3:  "US data update")
+
+Each test derives one step's q-weighted average total emission factor
+
+    wavg N = Σᵢ Nᵢ qᵢ / Σᵢ qᵢ,   N = 1ᵀ B L
+
+in a fresh subprocess (so ``functools.cache`` state cannot leak between
+configs), each step weighted by its own run's gross-output vector q — the
+release waterfall convention, where every bar is a true quantity of that
+state.
+
+The per-sector N vectors are grounded in the ``N_and_diffs`` tabs of the
+pinned diagnostics sheets in
+``bedrock.utils.validation.analysis.release_v0_v03_ceda_groups`` — the
+sheets the v0.3 assessment was built from. The expected levels below were
+pinned from a run whose per-sector N matched those sheets, with the same
+weighted average recomputed from the sheet N columns agreeing to well
+within the tolerance.
+
+If the pre-built ``GHG_national_Cornerstone_{year}`` FBS parquets are
+re-uploaded with new vintages, these levels move by design — re-pin the
+expected values consciously, exactly like a snapshot bump.
+"""
+
+from __future__ import annotations
+
+import json
+import re
+import subprocess
+import sys
+
+import pytest
+
+# q-weighted average N (kgCO2e per USD gross output, model_base_year dollars).
+# Pinned 2026-07-31 from live derivations cross-checked per sector against the
+# pinned sheets' ``N_new`` columns; see release_v0_v03_ceda_groups for sheet IDs.
+EXPECTED_WEIGHTED_AVG_N = {
+    'v0_baseline': 0.2563185,
+    'v03_waterfall_ceda_g1a_schema_ghg': 0.2655134,
+    'v03_waterfall_ceda_g1b_waste_disagg': 0.2543695,
+    'v03_waterfall_g2_methods': 0.2398356,
+    'v03_waterfall_g3_data': 0.2416736,
+    # FINAL is the full v0.3 methodology; it must telescope to the last step.
+    'v03_waterfall_final': 0.2416736,
+}
+
+# Absolute tolerance in kgCO2e/USD. Steps reproduce the pinned levels to
+# <1e-6; 1e-4 headroom absorbs float/order noise without masking any change
+# big enough to move a waterfall bar (~1e-2 between steps).
+ATOL_KG_PER_USD = 1e-4
+
+_STEP_TIMEOUT_S = 3600
+
+
+def _weighted_avg_n(arg: str) -> float:
+    """Run the weighted-average-N computation for one step in a fresh interpreter."""
+    proc = subprocess.run(
+        [sys.executable, '-m', 'bedrock.utils.validation.waterfall_progression', arg],
+        capture_output=True,
+        text=True,
+        timeout=_STEP_TIMEOUT_S,
+        check=False,
+    )
+    assert proc.returncode == 0, (
+        f'waterfall step {arg!r} failed (rc={proc.returncode}):\n'
+        f'stdout tail: {proc.stdout[-2000:]}\nstderr tail: {proc.stderr[-2000:]}'
+    )
+    match = re.search(r'\{"weighted_avg_n_kg_per_usd":\s*([-0-9.eE]+)\}', proc.stdout)
+    assert match, f'no JSON result on stdout for {arg!r}: {proc.stdout[-2000:]}'
+    return float(json.loads(match.group(0))['weighted_avg_n_kg_per_usd'])
+
+
+@pytest.mark.eeio_integration
+def test_v0_baseline_weighted_avg_n() -> None:
+    level = _weighted_avg_n('--v0-baseline')
+    assert level == pytest.approx(
+        EXPECTED_WEIGHTED_AVG_N['v0_baseline'], abs=ATOL_KG_PER_USD
+    )
+
+
+@pytest.mark.eeio_integration
+@pytest.mark.parametrize(
+    'config_name',
+    [
+        'v03_waterfall_ceda_g1a_schema_ghg',
+        'v03_waterfall_ceda_g1b_waste_disagg',
+        'v03_waterfall_g2_methods',
+        'v03_waterfall_g3_data',
+        'v03_waterfall_final',
+    ],
+)
+def test_waterfall_step_weighted_avg_n(config_name: str) -> None:
+    level = _weighted_avg_n(config_name)
+    assert level == pytest.approx(
+        EXPECTED_WEIGHTED_AVG_N[config_name], abs=ATOL_KG_PER_USD
+    )

bedrock/transform/eeio/cornerstone_year_scaling.py

@@ -188,3 +188,18 @@ def scale_cornerstone_q(
         )
 
     return q_scaled
+
+
+def scale_cornerstone_B(
+    B: pd.DataFrame,
+    target_year: USA_SUMMARY_MUT_YEARS,
+    original_year: USA_SUMMARY_MUT_YEARS,
+) -> pd.DataFrame:
+    """Scale B columns using summary q ratios (legacy pre-IO-adjustments footing)."""
+    ratio = (
+        derive_summary_q_usa(original_year) / derive_summary_q_usa(target_year)
+    ).fillna(1.0)
+    return ta.cast(
+        pd.DataFrame,
+        _apply_summary_ratio_to_sectors(ratio, B, axis='columns'),
+    )

bedrock/transform/eeio/derived_cornerstone.py

@@ -65,6 +65,7 @@
 )
 from bedrock.transform.eeio.cornerstone_year_scaling import (
     scale_cornerstone_A,
+    scale_cornerstone_B,
     scale_cornerstone_q,
 )
 from bedrock.transform.eeio.derived_2017 import (
@@ -82,6 +83,7 @@
     get_cornerstone_industry_price_ratio,
     inflate_cornerstone_A_matrix_with_commodity_pi,
     inflate_cornerstone_A_matrix_with_industry_pi,
+    inflate_cornerstone_B_matrix_with_industry_pi,
     inflate_cornerstone_q_or_y_with_commodity_pi,
     inflate_cornerstone_q_or_y_with_industry_pi,
     inflate_cornerstone_V_with_industry_pi,
@@ -690,8 +692,27 @@ def derive_cornerstone_B_via_vnorm() -> pd.DataFrame:
 
 @functools.cache
 def derive_cornerstone_B_non_finetuned() -> pd.DataFrame:
-    """Year-scaled + inflated B, derived self-contained from CEDA v7 → cornerstone."""
-    return derive_cornerstone_B_via_vnorm()
+    """Year-scaled + inflated B, derived self-contained from CEDA v7 → cornerstone.
+
+    With the IO-year adjustments on (``use_ghg_year_x_in_B``), B is already on
+    the GHG-year footing and stays on vnorm only. On the legacy footing, B is
+    scaled 2017 → ``usa_io_data_year`` with summary q ratios and then inflated
+    to ``model_base_year`` with the industry PI. (This keying was inadvertently
+    moved to an experimental flag in #482, silently changing the legacy
+    footing; restored here — guarded by test_waterfall_progression.)
+    """
+    cfg = get_usa_config()
+    if cfg.use_ghg_year_x_in_B:
+        return derive_cornerstone_B_via_vnorm()
+    return inflate_cornerstone_B_matrix_with_industry_pi(
+        scale_cornerstone_B(
+            B=derive_cornerstone_B_via_vnorm(),
+            original_year=cfg.usa_detail_original_year,
+            target_year=cfg.usa_io_data_year,
+        ),
+        original_year=cfg.usa_io_data_year,
+        target_year=cfg.model_base_year,
+    )
 
 
 @functools.cache

bedrock/utils/economic/inflation_helpers_cornerstone.py

@@ -225,6 +225,14 @@ def inflate_cornerstone_q_or_y_with_industry_pi(
     return q_or_y * price_ratio.reindex(q_or_y.index, fill_value=1.0)
 
 
+def inflate_cornerstone_B_matrix_with_industry_pi(
+    B: pd.DataFrame, original_year: int, target_year: int
+) -> pd.DataFrame:
+    """Inflate B's monetary denominators via the industry PI (legacy footing)."""
+    price_ratio = get_cornerstone_industry_price_ratio(target_year, original_year)
+    return B * price_ratio.reindex(B.columns, fill_value=1.0).values
+
+
 def inflate_cornerstone_V_with_industry_pi(
     V: pd.DataFrame,
     *,

bedrock/utils/validation/__tests__/test_diagnostics_helpers.py

@@ -58,7 +58,7 @@ def test_skip_default_b_inflation_path(self) -> None:
         )
         ok, reason = d_n_new_inflated_eligibility(cfg)
         assert ok is False
-        assert 'no denominator inflation adjustment' in reason
+        assert 'double-apply' in reason
 
     def test_skip_legacy_non_cornerstone(self) -> None:
         cfg = USAConfig(

bedrock/utils/validation/diagnostics_helpers.py

@@ -152,8 +152,9 @@ def d_n_new_inflated_eligibility(cfg: USAConfig) -> tuple[bool, str]:
         )
     return (
         False,
-        'B is not built with deflate_x_to_detail_io_year_for_B or '
-        'use_E_data_year_for_x_in_B; no denominator inflation adjustment applies',
+        'derive_cornerstone_B_non_finetuned already applies '
+        'inflate_cornerstone_B_matrix_with_industry_pi to model_base_year; '
+        'skipping second denominator inflation pass (would double-apply)',
     )
 
 

bedrock/utils/validation/waterfall_progression.py

@@ -0,0 +1,116 @@
+"""q-weighted average N for one config — waterfall regression helper.
+
+The v0→v0.3 release waterfall's bedrock-side steps toggle the bucket flags
+incrementally (GHG model allocation → waste disaggregation → IO year
+adjustments → US data update). Each step's level is the q-weighted average
+total emission factor
+
+    wavg N = Σᵢ Nᵢ qᵢ / Σᵢ qᵢ,   N = 1ᵀ M,  M = B L
+
+with each step weighted by its own run's gross-output vector q (the release
+waterfall convention: every bar is a true quantity of that state). This
+module computes that level for a single config in a fresh process, so
+successive steps cannot leak ``functools.cache`` state into each other.
+
+CLI (used by ``test_waterfall_progression`` via subprocess):
+
+    uv run python -m bedrock.utils.validation.waterfall_progression <config_name>
+    uv run python -m bedrock.utils.validation.waterfall_progression --v0-baseline
+
+Prints a JSON object ``{"weighted_avg_n_kg_per_usd": <float>}`` (kgCO2e per
+USD of gross output, in that run's ``model_base_year`` dollars) on stdout.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+
+import pandas as pd
+
+
+def _weighted_avg_n(
+    B: pd.DataFrame,
+    Adom: pd.DataFrame,
+    Aimp: pd.DataFrame,
+    q: 'pd.Series[float]',
+) -> float:
+    """Σ(N·q)/Σq with N = 1ᵀ B L, L = (I − (Adom+Aimp))⁻¹."""
+    from bedrock.utils.math.formulas import (  # noqa: PLC0415
+        compute_L_matrix,
+        compute_M_matrix,
+        compute_n,
+    )
+
+    N = compute_n(M=compute_M_matrix(B=B, L=compute_L_matrix(A=Adom + Aimp)))
+    q_aligned = q.reindex(N.index).astype(float)
+    return float((N * q_aligned).sum() / q_aligned.sum())
+
+
+def weighted_avg_n_for_config(config_name: str) -> float:
+    """q-weighted average N (kgCO2e/USD) for *config_name*, derived live."""
+    import bedrock.utils.config.common as common  # noqa: PLC0415
+    from bedrock.utils.config.usa_config import (  # noqa: PLC0415
+        reset_usa_config,
+        set_global_usa_config,
+    )
+
+    common.download_fba_on_api_error = True
+    # This runs in a fresh interpreter, but the parent test process exports
+    # USA_CONFIG_FILE into the inherited environment; clear it so the step
+    # uses exactly the requested config.
+    reset_usa_config(should_reset_env_var=True)
+    set_global_usa_config(config_name)
+
+    # Late-binding imports — depend on the global config.
+    from bedrock.transform.eeio.derived import (  # noqa: PLC0415
+        derive_Aq_usa,
+        derive_B_usa_non_finetuned,
+    )
+
+    aq = derive_Aq_usa()
+    return _weighted_avg_n(
+        B=derive_B_usa_non_finetuned(),
+        Adom=aq.Adom,
+        Aimp=aq.Aimp,
+        q=aq.scaled_q,
+    )
+
+
+def weighted_avg_n_v0_baseline() -> float:
+    """q-weighted average N recomputed from the frozen CEDA v0 snapshots."""
+    from bedrock.utils.snapshots.loader import load_snapshot  # noqa: PLC0415
+
+    q_raw = load_snapshot('scaled_q_USA', 'v0')
+    q: pd.Series = q_raw.iloc[:, 0] if isinstance(q_raw, pd.DataFrame) else q_raw
+    return _weighted_avg_n(
+        B=load_snapshot('B_USA_non_finetuned', 'v0'),
+        Adom=load_snapshot('Adom_USA', 'v0'),
+        Aimp=load_snapshot('Aimp_USA', 'v0'),
+        q=q.astype(float),
+    )
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    group = parser.add_mutually_exclusive_group(required=True)
+    group.add_argument('config_name', nargs='?', default=None)
+    group.add_argument(
+        '--v0-baseline',
+        action='store_true',
+        help='weighted average N from the frozen CEDA v0 snapshots instead of a live config',
+    )
+    args = parser.parse_args(argv)
+
+    level = (
+        weighted_avg_n_v0_baseline()
+        if args.v0_baseline
+        else weighted_avg_n_for_config(args.config_name)
+    )
+    json.dump({'weighted_avg_n_kg_per_usd': level}, sys.stdout)
+    return 0
+
+
+if __name__ == '__main__':
+    raise SystemExit(main())