feat(publish): integrate Phi purchaser-price adjustment

bedrock/publish/README.md

@@ -19,21 +19,25 @@ shareable file formats, mirroring the shape of `useeior`'s
   configs via `bedrock.publish.emission_factors`. Emits a long-form CO2e
   table (`CornerstoneSupplyChainGHG_CO2e_USD<year>.csv`) with three
   supply-chain factor columns (without margins, margins, with margins).
-  Purchaser-price adjustment uses industry price ratios from
-  `model_base_year` to `--dollar_year`. Phi and margin factors are
-  placeholders (identity Phi, zero margins) until real valuation matrices
-  land.
+  Purchaser-price adjustment applies PRO:PUR (Phi) from
+  `derive_phi_cornerstone_usa()` and rebases denominators from
+  `model_base_year` to `--dollar_year` via commodity price indices.
+  Margin SEF column remains zero until `N_margin` is wired.
+  CLI: `--purchaser_price` / `--no-purchaser_price` (default on).
 - **Supply-chain factors (R repo)**: not ported. Upstream counterpart:
   [cornerstone-data/supply-chain-factors](https://github.com/cornerstone-data/supply-chain-factors).
 - **Import-emissions matrices** (`A_m`, `M_m`, `N_m`): require real
   import emission factors (`B_imp`), which bedrock does not yet
   produce. Registered in `excel/writer.py` as `lambda: None`
   placeholders; sheets are omitted via the "skip if NULL" rule until
   `B_imp` lands.
-- **Useeior-only valuation matrices** (`Rho`, `Phi`, `Tau`) and
-  long-form metadata (`demands`, `SectorCrosswalk`): registered as
-  placeholders. Each carries an inline `TODO` pointing at the design
-  call or missing derivation.
+- **`Phi` sheet**: emitted when `useeio_margins` or
+  `cornerstone_industry_avg_margins` is active (`get_Phi()` in
+  `model_objects.py`). Values are at **`model_base_year` only** (one
+  column per config, e.g. 2017 for `useeio_phoebe_23`); not a full
+  year panel like useeior's `Phi` matrix yet.
+- **Useeior-only valuation matrices** (`Rho`, `Tau`) and long-form
+  metadata (`demands`, `SectorCrosswalk`): registered as placeholders.
 
 ## Known divergence from useeior (B units)
 
@@ -165,3 +169,13 @@ factor. Commodity codes are emitted with a `/US` suffix.
 
 Shared cached getters live in `bedrock/publish/model_objects.py` (also
 used by the XLSX publisher).
+
+### Phi / SEF validation (what is compared)
+
+| Artifact | Automated test | Year / basis |
+|----------|----------------|--------------|
+| `Phi` sheet vs pinned phoebe USEEIO workbook | `test_published_phi_matches_useeio_workbook` (`eeio_integration`) | Workbook **`Phi` column `model_base_year`** only (2017 on phoebe pin) |
+| SEF CSV wiring (Phi × CPI on `N`) | `test_sef_phi_wiring` (`eeio_integration`) | Export at `--dollar_year` (test uses 2024); **Phi stays at `model_base_year` (2017)** |
+| SEF vs [supply-chain-factors](https://github.com/cornerstone-data/supply-chain-factors) or Zenodo NAICS publish | None in bedrock | Not in scope for Phi PR (#449) |
+
+Default `uv run pytest` excludes `eeio_integration`; run `-m eeio_integration` for workbook Phi parity and SEF wiring tests.

bedrock/publish/__tests__/test_sef_phi_wiring.py

@@ -0,0 +1,56 @@
+"""Tests that emission-factor publish applies Phi and dollar-year rebasing."""
+
+from __future__ import annotations
+
+import numpy as np
+import pytest
+
+from bedrock.publish.__tests__._helpers import setup_config, teardown
+from bedrock.publish.emission_factors.table import (
+    COL_WITHOUT,
+    build_emission_factor_table,
+    finalize_cornerstone_ef_table,
+)
+from bedrock.publish.model_objects import get_N
+from bedrock.transform.iot.derive_PRO_to_PUR_ratio import phi_for_sectors
+from bedrock.utils.config.usa_config import get_usa_config
+from bedrock.utils.economic.inflation_helpers_cornerstone import (
+    get_vnorm_adjusted_commodity_price_ratio,
+)
+from bedrock.utils.emissions.characterization import GREENHOUSE_GASES_INDICATOR
+
+
+@pytest.mark.eeio_integration
+def test_sef_applies_phi_and_dollar_year() -> None:
+    """without_margins / (N_producer / cpi) equals phi per sector."""
+    setup_config('useeio_phoebe_23')
+    try:
+        dollar_year = 2024
+        cfg = get_usa_config()
+        table = finalize_cornerstone_ef_table(
+            build_emission_factor_table(
+                dollar_year=dollar_year,
+                purchaser_price=True,
+            )
+        )
+        n_producer = get_N().loc[GREENHOUSE_GASES_INDICATOR].astype(float)
+        pi = get_vnorm_adjusted_commodity_price_ratio(cfg.model_base_year, dollar_year)
+        n_producer_cpi = n_producer / pi.reindex(n_producer.index, fill_value=1.0)
+        phi = phi_for_sectors(n_producer.index)
+
+        for _, row in table.iterrows():
+            code = str(row['Cornerstone Commodity Code']).removesuffix('/US')
+            without = float(row[COL_WITHOUT])
+            if code not in n_producer_cpi.index:
+                continue
+            denom = float(n_producer_cpi[code])
+            if denom == 0.0:
+                continue
+            np.testing.assert_allclose(
+                without / denom,
+                float(phi[code]),
+                rtol=1e-9,
+                err_msg=f'sector {code}',
+            )
+    finally:
+        teardown()

bedrock/publish/__tests__/test_sef_vs_useeio_baseline.py

@@ -0,0 +1,102 @@
+"""eeio_integration: published Phi vs pinned phoebe USEEIO workbook."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import numpy as np
+import pandas as pd
+import pytest
+
+import bedrock.utils.config.common as common
+from bedrock.publish.__tests__._helpers import clear_all_caches, teardown
+from bedrock.publish.model_objects import PUBLISH_LOCATION, get_Phi
+from bedrock.utils.config.usa_config import (
+    get_usa_config,
+    reset_usa_config,
+    set_global_usa_config,
+)
+from bedrock.utils.validation.useeio_excel_baseline import (
+    _local_cache_path,
+    ensure_useeio_xlsx_local,
+    load_useeio_baseline_pin_overrides,
+)
+
+_PIN_JSON = (
+    Path(__file__).resolve().parents[2]
+    / 'utils'
+    / 'snapshots'
+    / 'useeio_baseline_pin.json'
+)
+
+_PHI_RTOL = 0.01
+
+
+def _setup_phoebe_with_useeio_pin() -> dict[str, str]:
+    pin = load_useeio_baseline_pin_overrides(str(_PIN_JSON))
+    overrides: dict[str, object] = {
+        **pin,
+        'diagnostics_baseline_source': 'gcs_useeio_xlsx',
+    }
+    clear_all_caches()
+    reset_usa_config(should_reset_env_var=True)
+    set_global_usa_config(
+        'useeio_phoebe_23',
+        diagnostics_cli_overrides=overrides,
+    )
+    common.download_fba_on_api_error = True
+    return pin
+
+
+def _strip_loc_suffix(values: pd.Index) -> pd.Index:
+    suffix = f'/{PUBLISH_LOCATION}'
+    return pd.Index(
+        [str(v)[: -len(suffix)] if str(v).endswith(suffix) else str(v) for v in values],
+        name=values.name,
+    )
+
+
+def _load_workbook_phi(xlsx_path: str, year: int) -> pd.Series:
+    raw = pd.read_excel(xlsx_path, sheet_name='Phi', header=None, engine='openpyxl')
+    headers = (
+        raw.iloc[0, 1:].astype(str).str.strip().str.replace(r'\.0$', '', regex=True)
+    )
+    sectors = raw.iloc[1:, 0].astype(str).str.strip()
+    values = raw.iloc[1:, 1:].copy()
+    values.columns = pd.Index(headers)
+    values.index = pd.Index(sectors)
+    year_str = str(year)
+    phi = values[year_str].astype(float)
+    phi.index = pd.Index(
+        [s[:-3] if s.endswith('/US') else s for s in phi.index], name='sector'
+    )
+    return phi.dropna()
+
+
+@pytest.mark.eeio_integration
+def test_published_phi_matches_useeio_workbook() -> None:
+    try:
+        pin = _setup_phoebe_with_useeio_pin()
+        cfg = get_usa_config()
+        xlsx = _local_cache_path(pin['useeio_baseline_xlsx_gs_uri'])
+        ensure_useeio_xlsx_local(
+            pin['useeio_baseline_xlsx_gs_uri'],
+            pin['useeio_baseline_xlsx_sha256'],
+            xlsx,
+        )
+        bedrock_phi_df = get_Phi()
+        assert bedrock_phi_df is not None
+        bedrock_phi = bedrock_phi_df[str(cfg.model_base_year)].astype(float)
+        bedrock_phi.index = _strip_loc_suffix(bedrock_phi.index)
+
+        ref_phi = _load_workbook_phi(xlsx, cfg.model_base_year)
+        common_sectors = bedrock_phi.index.intersection(ref_phi.index)
+        assert len(common_sectors) > 100
+        np.testing.assert_allclose(
+            bedrock_phi.reindex(common_sectors).astype(float),
+            ref_phi.reindex(common_sectors).astype(float),
+            rtol=_PHI_RTOL,
+            atol=1e-12,
+        )
+    finally:
+        teardown()

bedrock/publish/cache_reset.py

@@ -4,6 +4,10 @@
 
 from collections.abc import Callable
 
+from bedrock.extract.iot.io_2017 import (
+    load_2017_margins_after_redef_usa,
+    load_2017_margins_before_redef_usa,
+)
 from bedrock.publish.model_objects import clear_publish_caches
 from bedrock.transform.eeio.cornerstone_disagg_pipeline import (
     cornerstone_sector_disagg_active,
@@ -55,6 +59,8 @@
     derive_cornerstone_Aq_scaled,
     derive_cornerstone_B_non_finetuned,
     derive_cornerstone_y_nab,
+    load_2017_margins_before_redef_usa,
+    load_2017_margins_after_redef_usa,
 ]
 
 

bedrock/publish/emission_factors/cli.py

@@ -50,11 +50,17 @@ def _default_output_dir(config_name: str) -> str:
     default=False,
     help='Also write M_pur and N_pur CSVs under matrices/.',
 )
+@click.option(
+    '--purchaser_price/--no-purchaser_price',
+    default=True,
+    help='Apply PRO:PUR (Phi) purchaser-price adjustment (default: on).',
+)
 def publish(
     config_name: str,
     dollar_year: int,
     output_dir: str | None,
     write_matrices: bool,
+    purchaser_price: bool,
 ) -> None:
     from bedrock.publish.emission_factors.writer import write_emission_factors
 
@@ -68,6 +74,7 @@ def publish(
         local_dir,
         config_name=config_name,
         dollar_year=dollar_year,
+        purchaser_price=purchaser_price,
         write_matrices=write_matrices,
     )
 

bedrock/publish/emission_factors/placeholders.py

@@ -4,17 +4,13 @@
 
 import pandas as pd
 
+from bedrock.transform.iot.derive_PRO_to_PUR_ratio import phi_for_sectors
 from bedrock.utils.config.usa_config import get_usa_config
 from bedrock.utils.economic.inflation_helpers_cornerstone import (
     get_vnorm_adjusted_commodity_price_ratio,
 )
 
 
-def placeholder_phi(sector_index: pd.Index) -> pd.Series[float]:
-    """PRO:PUR price ratio; identity until real Phi is wired."""
-    return pd.Series(1.0, index=sector_index, dtype=float)
-
-
 def placeholder_margin_ef(without_margins: pd.Series) -> pd.Series[float]:
     """Per-sector margin supply-chain factors; zeros until N_margin is wired."""
     return pd.Series(0.0, index=without_margins.index, dtype=float)
@@ -39,8 +35,7 @@ def adjust_publish_matrix(
         out = out.div(price_ratio_for_columns.values, axis=1)
 
     if purchaser_price:
-        phi = placeholder_phi(out.columns)
-        aligned_phi = phi.reindex(out.columns, fill_value=1.0)
-        out = out.mul(aligned_phi.values, axis=1)
+        phi = phi_for_sectors(out.columns)
+        out = out.mul(phi.reindex(out.columns, fill_value=1.0).values, axis=1)
 
     return out

bedrock/publish/emission_factors/table.py

@@ -36,8 +36,9 @@ def _greenhouse_gases_row(n: pd.DataFrame) -> pd.Series:
     return cast(pd.Series, row)
 
 
-def _unit_label(dollar_year: int) -> str:
-    return f'kg CO2e / {dollar_year} USD, purchaser price'
+def _unit_label(dollar_year: int, *, purchaser_price: bool) -> str:
+    price_type = 'purchaser price' if purchaser_price else 'producer price'
+    return f'kg CO2e / {dollar_year} USD, {price_type}'
 
 
 def _commodity_base_code(code: str) -> str:
@@ -55,12 +56,14 @@ def _is_excluded_commodity(code: str) -> bool:
     return False
 
 
-def build_emission_factor_table(*, dollar_year: int) -> pd.DataFrame:
-    """Long-form CO2e supply-chain factors at purchaser price in ``dollar_year``."""
+def build_emission_factor_table(
+    *, dollar_year: int, purchaser_price: bool = True
+) -> pd.DataFrame:
+    """Long-form CO2e supply-chain factors in ``dollar_year`` USD."""
     n_pur = adjust_publish_matrix(
         get_N(),
         dollar_year=dollar_year,
-        purchaser_price=True,
+        purchaser_price=purchaser_price,
     )
     without = _greenhouse_gases_row(n_pur)
     margins = placeholder_margin_ef(without)
@@ -73,7 +76,7 @@ def build_emission_factor_table(*, dollar_year: int) -> pd.DataFrame:
                 COL_CODE: code,
                 COL_NAME: COMMODITY_DESC.get(code, ''),
                 COL_GHG: GHG_LABEL,
-                COL_UNIT: _unit_label(dollar_year),
+                COL_UNIT: _unit_label(dollar_year, purchaser_price=purchaser_price),
                 COL_WITHOUT: float(without[code]),
                 COL_MARGINS: float(margins[code]),
                 COL_WITH: float(with_margins[code]),
@@ -95,16 +98,18 @@ def finalize_cornerstone_ef_table(table: pd.DataFrame) -> pd.DataFrame:
     return out.reset_index(drop=True)
 
 
-def build_purchaser_matrices(*, dollar_year: int) -> tuple[pd.DataFrame, pd.DataFrame]:
-    """Return M and N at purchaser price in ``dollar_year`` (raw sector labels)."""
+def build_purchaser_matrices(
+    *, dollar_year: int, purchaser_price: bool = True
+) -> tuple[pd.DataFrame, pd.DataFrame]:
+    """Return M and N adjusted to ``dollar_year`` (raw sector labels)."""
     m_pur = adjust_publish_matrix(
         get_M(),
         dollar_year=dollar_year,
-        purchaser_price=True,
+        purchaser_price=purchaser_price,
     )
     n_pur = adjust_publish_matrix(
         get_N(),
         dollar_year=dollar_year,
-        purchaser_price=True,
+        purchaser_price=purchaser_price,
     )
     return m_pur, n_pur

bedrock/publish/emission_factors/writer.py

@@ -20,14 +20,18 @@ def write_emission_factors(
     *,
     config_name: str,
     dollar_year: int,
+    purchaser_price: bool = True,
     write_matrices: bool = False,
 ) -> dict[str, str]:
     """Write CO2e SEF CSV (and optional M/N purchaser matrices) under ``output_dir``."""
     require_cornerstone_config()
     os.makedirs(output_dir, exist_ok=True)
 
     table = finalize_cornerstone_ef_table(
-        build_emission_factor_table(dollar_year=dollar_year)
+        build_emission_factor_table(
+            dollar_year=dollar_year,
+            purchaser_price=purchaser_price,
+        )
     )
     co2e_name = f'CornerstoneSupplyChainGHG_CO2e_USD{dollar_year}.csv'
     co2e_path = os.path.join(output_dir, co2e_name)
@@ -44,7 +48,10 @@ def write_emission_factors(
     if write_matrices:
         matrices_dir = os.path.join(output_dir, 'matrices')
         os.makedirs(matrices_dir, exist_ok=True)
-        m_pur, n_pur = build_purchaser_matrices(dollar_year=dollar_year)
+        m_pur, n_pur = build_purchaser_matrices(
+            dollar_year=dollar_year,
+            purchaser_price=purchaser_price,
+        )
         m_path = os.path.join(matrices_dir, f'M_pur_{dollar_year}.csv')
         n_path = os.path.join(matrices_dir, f'N_pur_{dollar_year}.csv')
         apply_loc_suffix(m_pur).to_csv(m_path)

bedrock/publish/excel/writer.py

@@ -83,6 +83,7 @@
     get_Mdom,
     get_N,
     get_Ndom,
+    get_Phi,
     get_q,
     get_U,
     get_Udom,
@@ -294,10 +295,8 @@ def _build_matrix_registry(config_name: str) -> list[SheetSpec]:
         SheetSpec('N', get_N),
         SheetSpec('N_d', get_Ndom),
         SheetSpec('N_m', lambda: None),  # requires B_imp
-        # Rho, Phi, Tau are useeior valuation-adjustment matrices with no
-        # direct bedrock analogue. Leave as TODO until a design call is made.
         SheetSpec('Rho', lambda: None),
-        SheetSpec('Phi', lambda: None),
+        SheetSpec('Phi', get_Phi),
         SheetSpec('Tau', lambda: None),
         # --- outputs (useeior writes these after the matrices block) ---
         SheetSpec('q', get_q),