diff --git a/bedrock/transform/__tests__/test_waterfall_progression.py b/bedrock/transform/__tests__/test_waterfall_progression.py new file mode 100644 index 00000000..e3b8d64e --- /dev/null +++ b/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 + ) diff --git a/bedrock/transform/eeio/cornerstone_year_scaling.py b/bedrock/transform/eeio/cornerstone_year_scaling.py index e6dde263..600b0cb4 100644 --- a/bedrock/transform/eeio/cornerstone_year_scaling.py +++ b/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'), + ) diff --git a/bedrock/transform/eeio/derived_cornerstone.py b/bedrock/transform/eeio/derived_cornerstone.py index 7f4b51a7..e1252525 100644 --- a/bedrock/transform/eeio/derived_cornerstone.py +++ b/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 diff --git a/bedrock/utils/economic/inflation_helpers_cornerstone.py b/bedrock/utils/economic/inflation_helpers_cornerstone.py index 4550b34d..2a6b9709 100644 --- a/bedrock/utils/economic/inflation_helpers_cornerstone.py +++ b/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, *, diff --git a/bedrock/utils/validation/__tests__/test_diagnostics_helpers.py b/bedrock/utils/validation/__tests__/test_diagnostics_helpers.py index e4f231c3..b9f582b8 100644 --- a/bedrock/utils/validation/__tests__/test_diagnostics_helpers.py +++ b/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( diff --git a/bedrock/utils/validation/diagnostics_helpers.py b/bedrock/utils/validation/diagnostics_helpers.py index 1b0915a6..2fd92edf 100644 --- a/bedrock/utils/validation/diagnostics_helpers.py +++ b/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)', ) diff --git a/bedrock/utils/validation/waterfall_progression.py b/bedrock/utils/validation/waterfall_progression.py new file mode 100644 index 00000000..5bc431f8 --- /dev/null +++ b/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 + uv run python -m bedrock.utils.validation.waterfall_progression --v0-baseline + +Prints a JSON object ``{"weighted_avg_n_kg_per_usd": }`` (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())