diff --git a/bedrock/analysis/margins/compare_margin_approaches.py b/bedrock/analysis/margins/compare_margin_approaches.py deleted file mode 100644 index 178a0b17..00000000 --- a/bedrock/analysis/margins/compare_margin_approaches.py +++ /dev/null @@ -1,226 +0,0 @@ -"""Compare commodity-level PRO:PUR ratios across margin model approaches and years. - -Scenarios (each at ``derive_margins_cornerstone_usa_at_year(year)`` unless noted): - - useeio — ``useeio_phoebe_23`` with ``useeio_margins`` (Rho PRO inflation) - cornerstone — ``2025_usa_cornerstone_v0_2`` with industry-avg margins - (V-norm commodity PI on PRO) - ceda — ``derive_phi_ceda_usa`` mapped to Cornerstone commodities (IO year) - -Usage:: - - uv run python -m bedrock.analysis.margins.compare_margin_approaches - -Outputs: - output/plots/margin_approach_comparison_.png - output/margin_approach_comparison.csv -""" - -from __future__ import annotations - -import os - -import matplotlib.pyplot as plt -import numpy as np -import pandas as pd - -OUT = os.path.join(os.path.dirname(__file__), 'output') -PLOTS = os.path.join(OUT, 'plots') -os.makedirs(PLOTS, exist_ok=True) - -from bedrock.transform.iot.derive_PRO_to_PUR_ratio import ( # noqa: E402 - derive_margins_cornerstone_usa_at_year, - derive_phi_ceda_usa, -) -from bedrock.utils.config.config_controllers import temp_usa_config # noqa: E402 -from bedrock.utils.config.usa_config import get_usa_config # noqa: E402 -from bedrock.utils.taxonomy.mappings.bea_v2017_sector__cornerstone_commodity import ( # noqa: E402 - load_bea_v2017_sector_commodity_to_cornerstone_commodity, -) -from bedrock.utils.taxonomy.usa_taxonomy_correspondence_helpers import ( # noqa: E402 - load_ceda_v7_commodity__cornerstone_commodity_correspondence, -) - -_PANEL_YEAR = 2024 -_CORNERSTONE_SCENARIOS = ('useeio', 'cornerstone', 'ceda') -_COLORS = { - 'useeio': 'tab:orange', - 'cornerstone': 'tab:green', - 'ceda': 'tab:blue', -} - -_CACHE_MODULES = ( - 'bedrock.extract.iot.io_2017', - 'bedrock.transform.iot.derive_PRO_to_PUR_ratio', -) - - -def _ratio_from_margins(margins: pd.DataFrame) -> pd.Series: - return (margins["Producers' Value"] / margins["Purchasers' Value"]).replace( - [np.inf, -np.inf, np.nan], 1.0 - ) - - -def _comparison_years() -> tuple[int, ...]: - with temp_usa_config('useeio_phoebe_23', cache_bearing_modules=_CACHE_MODULES): - base_year = int(get_usa_config().usa_base_io_data_year) - years: list[int] = [base_year] - if _PANEL_YEAR not in years: - years.append(_PANEL_YEAR) - return tuple(years) - - -def _ceda_ratios_on_cornerstone() -> pd.Series: - _ceda_corresp = load_ceda_v7_commodity__cornerstone_commodity_correspondence() - _ceda_corresp_norm = _ceda_corresp.div(_ceda_corresp.sum(axis=1), axis=0).fillna( - 0.0 - ) - # Enable CEDA Phi filters on legacy-default footing (test_usa_config). - with temp_usa_config( - 'test_usa_config', - cache_bearing_modules=_CACHE_MODULES, - ceda_margins=True, - ): - ratio_ceda_by_sector = derive_phi_ceda_usa() - return _ceda_corresp_norm @ ratio_ceda_by_sector - - -def _ratios_by_scenario(year: int, *, ratio_ceda: pd.Series) -> dict[str, pd.Series]: - print(f'Computing margin ratios for {year}...') - with temp_usa_config('useeio_phoebe_23', cache_bearing_modules=_CACHE_MODULES): - ratio_useeio = _ratio_from_margins(derive_margins_cornerstone_usa_at_year(year)) - with temp_usa_config( - '2025_usa_cornerstone_v0_3', cache_bearing_modules=_CACHE_MODULES - ): - ratio_cornerstone = _ratio_from_margins( - derive_margins_cornerstone_usa_at_year(year) - ) - return { - 'useeio': ratio_useeio, - 'cornerstone': ratio_cornerstone, - 'ceda': ratio_ceda, - } - - -def _plot_year(long_df: pd.DataFrame, active_sectors: list[str], year: int) -> str: - plot_df = long_df[long_df['year'] == year] - n_scenarios = len(_CORNERSTONE_SCENARIOS) - n_sectors = len(active_sectors) - group_width = 0.8 - violin_width = group_width / n_scenarios - - fig, ax = plt.subplots(figsize=(max(10, n_sectors * 2.5), 6)) - - for g_idx, sector in enumerate(active_sectors): - for s_idx, scenario in enumerate(_CORNERSTONE_SCENARIOS): - data = plot_df.loc[ - (plot_df['sector'] == sector) & (plot_df['scenario'] == scenario), - 'ratio', - ].dropna() - if len(data) < 2: - y_vals = ( - data.to_numpy(dtype=float) - if len(data) - else np.array([1.0], dtype=float) - ) - ax.scatter( - [g_idx + (s_idx - (n_scenarios - 1) / 2) * violin_width], - y_vals, - color=_COLORS[scenario], - s=20, - zorder=3, - ) - continue - pos = g_idx + (s_idx - (n_scenarios - 1) / 2) * violin_width - parts = ax.violinplot( - data, - positions=[pos], - widths=violin_width * 0.9, - showmedians=True, - showextrema=False, - ) - for pc in parts['bodies']: # type: ignore[attr-defined] - pc.set_facecolor(_COLORS[scenario]) - pc.set_alpha(0.6) - parts['cmedians'].set_color(_COLORS[scenario]) - parts['cmedians'].set_linewidth(1.5) - - ax.axhline( - 1.0, color='black', linewidth=0.8, linestyle='--', alpha=0.5, label='ratio = 1' - ) - ax.set_xticks(range(n_sectors)) - ax.set_xticklabels(active_sectors, rotation=45, ha='right', fontsize=9) - ax.set_ylabel('PRO:PUR ratio') - ax.set_title( - f'PRO:PUR ratio by BEA sector and margin approach ({year} USD margins)\n' - '(Cornerstone commodities; CEDA column is IO-year only)' - ) - ax.grid(True, axis='y', linestyle=':', alpha=0.4) - - legend_handles = [ - plt.Rectangle((0, 0), 1, 1, fc=_COLORS[s], alpha=0.6, label=s) - for s in _CORNERSTONE_SCENARIOS - ] - ax.legend(handles=legend_handles, loc='upper right', fontsize=8) - - fig.tight_layout() - plot_path = os.path.join(PLOTS, f'margin_approach_comparison_{year}.png') - fig.savefig(plot_path, dpi=150) - plt.close(fig) - return plot_path - - -def main() -> None: - years = _comparison_years() - ratio_ceda = _ceda_ratios_on_cornerstone() - ratio_columns: dict[str, pd.Series] = {} - for year in years: - for scenario, series in _ratios_by_scenario( - year, ratio_ceda=ratio_ceda - ).items(): - ratio_columns[f'{scenario}_{year}'] = series - - ratio_df = pd.DataFrame(ratio_columns) - ratio_df.index.name = 'commodity' - - sector_map = load_bea_v2017_sector_commodity_to_cornerstone_commodity() - rows: list[dict[str, object]] = [] - for year in years: - for sector, commodities in sector_map.items(): - for commodity in commodities: - if commodity not in ratio_df.index: - continue - for scenario in _CORNERSTONE_SCENARIOS: - col = f'{scenario}_{year}' - if col not in ratio_df.columns: - continue - rows.append( - { - 'year': year, - 'sector': sector, - 'commodity': commodity, - 'scenario': scenario, - 'ratio': ratio_df.loc[commodity, col], - } - ) - long_df = pd.DataFrame(rows) - - non_unity = long_df.groupby('sector')['ratio'].apply( - lambda s: (s - 1.0).abs().max() > 1e-6 - ) - active_sectors = non_unity[non_unity].index.tolist() - print( - f'\n{len(active_sectors)} of {len(sector_map)} sectors have non-unity ratios.' - ) - - for year in years: - plot_path = _plot_year(long_df, active_sectors, year) - print(f'Plot saved to: {plot_path}') - - csv_path = os.path.join(OUT, 'margin_approach_comparison.csv') - ratio_df.to_csv(csv_path) - print(f'Cornerstone table saved to: {csv_path}') - - -if __name__ == '__main__': - main() diff --git a/bedrock/analysis/margins/compare_phi_to_reference.py b/bedrock/analysis/margins/compare_phi_to_reference.py deleted file mode 100644 index ad31b318..00000000 --- a/bedrock/analysis/margins/compare_phi_to_reference.py +++ /dev/null @@ -1,250 +0,0 @@ -"""Compare model-computed Phi (PRO:PUR ratio) against external reference data. - -USEEIO: ``derive_phi_cornerstone_usa_at_year`` vs the Phi sheet in the pinned - USEEIO Excel workbook at ``usa_base_io_data_year`` and 2024. -CEDA: ``derive_phi_ceda_usa`` vs the 'Purchaser - producer conversion' sheet - in the CEDA 2025 Excel workbook (IO-year only; no year panel). - -Usage:: - - uv run python -m bedrock.analysis.margins.compare_phi_to_reference - -Outputs: - output/plots/phi_comparison.png - output/phi_comparison_useeio_.csv - output/phi_comparison_ceda.csv -""" - -from __future__ import annotations - -import os -import re - -import matplotlib.pyplot as plt -import numpy as np -import pandas as pd - -OUT = os.path.join(os.path.dirname(__file__), 'output') -PLOTS = os.path.join(OUT, 'plots') -os.makedirs(PLOTS, exist_ok=True) - -from bedrock.transform.iot.derive_PRO_to_PUR_ratio import ( # noqa: E402 - derive_phi_ceda_usa, - derive_phi_cornerstone_usa_at_year, -) -from bedrock.utils.config.config_controllers import temp_usa_config # noqa: E402 -from bedrock.utils.config.usa_config import get_usa_config # noqa: E402 -from bedrock.utils.io.gcp import download_gcs_file_if_not_exists # noqa: E402 -from bedrock.utils.snapshots.loader import useeio_baseline_local_dir # noqa: E402 -from bedrock.utils.validation.useeio_excel_baseline import ( # noqa: E402 - ensure_useeio_xlsx_local, - load_useeio_baseline_pin_overrides, -) - -_PIN_JSON = os.path.join( - os.path.dirname(__file__), - '..', - '..', - 'utils', - 'snapshots', - 'useeio_baseline_pin.json', -) -_CEDA_GS_URI = ( - 'gs://cornerstone-default/snapshots/CEDA_2025/CEDA 2025 (updated 2025-11-12).xlsx' -) -_GCS_PREFIX = 'gs://cornerstone-default/' -_USEEIO_PANEL_YEAR = 2024 - -_CACHE_MODULES = ( - 'bedrock.extract.iot.io_2017', - 'bedrock.transform.iot.derive_PRO_to_PUR_ratio', -) - - -def _xlsx_local_path(gs_uri: str) -> str: - """Local cache path for any xlsx under gs://cornerstone-default/.""" - safe = re.sub(r'[^a-zA-Z0-9_.-]+', '_', gs_uri.removeprefix(_GCS_PREFIX)) - path = os.path.join(useeio_baseline_local_dir(), safe) - return path if path.lower().endswith('.xlsx') else path + '.xlsx' - - -def _ensure_ceda_xlsx_local() -> str: - local = _xlsx_local_path(_CEDA_GS_URI) - if not os.path.isfile(local): - rest = _CEDA_GS_URI.removeprefix(_GCS_PREFIX).strip('/') - parts = rest.split('/') - download_gcs_file_if_not_exists(parts[-1], '/'.join(parts[:-1]), local) - return local - - -def _load_useeio_phi_reference(local_path: str, year: int) -> pd.Series: - """Phi sheet: row 1 = year headers, col A = sector codes ({code}/US).""" - raw = pd.read_excel(local_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) - if year_str not in values.columns: - available = values.columns.tolist() - raise ValueError(f'Year {year} not in USEEIO Phi sheet; available: {available}') - 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() - - -def _load_ceda_phi_reference(local_path: str) -> pd.Series: - """'Purchaser - producer conversion': headers in row 5 (B onward), data in row 6.""" - raw = pd.read_excel( - local_path, - sheet_name='Purchaser - producer conversion', - header=None, - engine='openpyxl', - ) - headers = raw.iloc[4, 1:].astype(str).str.strip() - data_row = raw.iloc[5, 1:].copy() - phi = pd.Series( - pd.to_numeric(pd.Series(data_row), errors='coerce').values, - index=pd.Index(headers), - name='phi_reference', - ) - phi.index.name = 'sector' - return phi.dropna() - - -def _compute_useeio_phi_model(year: int) -> pd.Series: - """Phi from the publish margins path at *year* USD.""" - phi = derive_phi_cornerstone_usa_at_year(year).astype(float) - phi.index.name = 'sector' - return phi - - -def _scatter_comparison( - ax: plt.Axes, model: pd.Series, ref: pd.Series, title: str -) -> pd.DataFrame: - """Scatter model vs reference; return aligned comparison DataFrame.""" - common = model.index.intersection(ref.index) - x = ref.reindex(common).astype(float) - y = model.reindex(common).astype(float) - mask = x.notna() & y.notna() - x, y = x[mask], y[mask] - - ax.scatter(x, y, s=8, alpha=0.5, color='steelblue', linewidths=0) - lo = min(float(x.min()), float(y.min()), 0.0) - 0.02 - hi = max(float(x.max()), float(y.max()), 1.0) + 0.02 - ax.plot([lo, hi], [lo, hi], 'k--', linewidth=0.8, label='1:1') - ax.set_xlim(lo, hi) - ax.set_ylim(lo, hi) - ax.set_xlabel('Reference Phi (external)') - ax.set_ylabel('Model Phi (bedrock)') - ax.set_title(title) - ax.legend(fontsize=7) - if len(x) > 1: - corr = float(x.corr(y)) - mae = float((y - x).abs().mean()) - med_rel = float(((y - x) / x.replace(0.0, np.nan)).abs().median()) - ax.text( - 0.04, - 0.92, - f'n={len(x)} r={corr:.3f} MAE={mae:.4f} med|rel|={med_rel:.3f}', - transform=ax.transAxes, - fontsize=7.5, - ) - - diff = y - x - return pd.DataFrame( - {'phi_model': y, 'phi_reference': x, 'diff': diff, 'abs_diff': diff.abs()} - ).reindex(common) - - -def _useeio_comparison_years() -> tuple[int, ...]: - with temp_usa_config('useeio_phoebe_23', cache_bearing_modules=_CACHE_MODULES): - base_year = int(get_usa_config().usa_base_io_data_year) - years: list[int] = [base_year] - if _USEEIO_PANEL_YEAR not in years: - years.append(_USEEIO_PANEL_YEAR) - return tuple(years) - - -def main() -> None: - useeio_years = _useeio_comparison_years() - useeio_model_by_year: dict[int, pd.Series] = {} - print('Computing USEEIO model Phi...') - with temp_usa_config('useeio_phoebe_23', cache_bearing_modules=_CACHE_MODULES): - for year in useeio_years: - print(f' year {year}...') - useeio_model_by_year[year] = _compute_useeio_phi_model(year) - - print('Loading USEEIO reference Phi...') - pin = load_useeio_baseline_pin_overrides(_PIN_JSON) - useeio_gs_uri = pin['useeio_baseline_xlsx_gs_uri'] - useeio_local = _xlsx_local_path(useeio_gs_uri) - ensure_useeio_xlsx_local( - useeio_gs_uri, pin['useeio_baseline_xlsx_sha256'], useeio_local - ) - useeio_ref_by_year = { - year: _load_useeio_phi_reference(useeio_local, year) for year in useeio_years - } - - print('Computing CEDA model Phi...') - # Enable CEDA Phi filters on legacy-default footing (test_usa_config). - with temp_usa_config( - 'test_usa_config', - cache_bearing_modules=_CACHE_MODULES, - ceda_margins=True, - ): - phi_ceda_model = derive_phi_ceda_usa() - - print('Loading CEDA reference Phi...') - ceda_local = _ensure_ceda_xlsx_local() - phi_ceda_ref = _load_ceda_phi_reference(ceda_local) - - n_axes = len(useeio_years) + 1 - fig, axes = plt.subplots(1, n_axes, figsize=(5 * n_axes, 5)) - if n_axes == 1: - axes = [axes] - - useeio_tables: dict[int, pd.DataFrame] = {} - for ax, year in zip(axes, useeio_years, strict=False): - model = useeio_model_by_year[year] - ref = useeio_ref_by_year[year] - print( - f'\nUSEEIO {year}: {len(model.dropna())} model sectors, ' - f'{len(ref)} reference sectors' - ) - useeio_tables[year] = _scatter_comparison( - ax, model, ref, f'USEEIO Phi ({year})' - ) - - print( - f'\nCEDA: {len(phi_ceda_model)} model sectors, ' - f'{len(phi_ceda_ref)} reference sectors' - ) - df_ceda = _scatter_comparison( - axes[-1], phi_ceda_model, phi_ceda_ref, 'CEDA Phi (IO year)' - ) - - fig.suptitle('PRO:PUR Phi — model (bedrock) vs external reference', fontsize=11) - fig.tight_layout() - plot_path = os.path.join(PLOTS, 'phi_comparison.png') - fig.savefig(plot_path, dpi=150) - plt.close(fig) - print(f'\nPlot saved to: {plot_path}') - - for year, table in useeio_tables.items(): - path = os.path.join(OUT, f'phi_comparison_useeio_{year}.csv') - table.to_csv(path) - print(f'USEEIO CSV ({year}): {path}') - - ceda_csv = os.path.join(OUT, 'phi_comparison_ceda.csv') - df_ceda.to_csv(ceda_csv) - print(f'CEDA CSV: {ceda_csv}') - - -if __name__ == '__main__': - main() diff --git a/bedrock/analysis/margins/compare_sef_margins_sources.py b/bedrock/analysis/margins/compare_sef_margins_sources.py index c7bbd563..73863cf2 100644 --- a/bedrock/analysis/margins/compare_sef_margins_sources.py +++ b/bedrock/analysis/margins/compare_sef_margins_sources.py @@ -14,7 +14,7 @@ uv run python -m bedrock.analysis.margins.compare_sef_margins_sources Optional: - --phoebe-sef-csv PATH + --phoebe-sef-csv PATH (required; pinned SEF from a pre-retirement phoebe run) --v0-3-sef-csv PATH --zenodo-xlsx PATH (defaults to cached download under ``bedrock/utils/snapshots/data/zenodo_sef_v1.4.0/``) @@ -63,7 +63,6 @@ COL_WITH = 'Supply Chain Emission Factors with Margins' SEF_VALUE_COLS: tuple[str, ...] = (COL_WITHOUT, COL_MARGINS, COL_WITH) -_PHOEBE_CONFIG = 'useeio_phoebe_23' _V0_3_CONFIG = '2025_usa_cornerstone_v0_3' @@ -246,10 +245,12 @@ def main() -> int: zenodo = load_zenodo_sef_by_reference_code(zenodo_path) if args.phoebe_sef_csv is None: - logger.info('publishing %s at dollar_year=%d', _PHOEBE_CONFIG, args.dollar_year) - phoebe_path = publish_sef(_PHOEBE_CONFIG, args.dollar_year) - else: - phoebe_path = args.phoebe_sef_csv + parser.error( + '--phoebe-sef-csv is required: the useeio_phoebe_23 config was ' + 'retired with the USEEIO-recreation flags; use a pinned phoebe SEF ' + 'CSV from a prior publish run.' + ) + phoebe_path = args.phoebe_sef_csv logger.info('phoebe SEF: %s', phoebe_path) phoebe = load_bedrock_sef(phoebe_path) diff --git a/bedrock/transform/allocation/derived.py b/bedrock/transform/allocation/derived.py index d587552a..3c7f6bc2 100644 --- a/bedrock/transform/allocation/derived.py +++ b/bedrock/transform/allocation/derived.py @@ -25,9 +25,6 @@ logger = logging.getLogger(__name__) -# USEEIO does not distinguish fossil from non-fossil CH4 -_USEEIO_WORKBOOK_CH4_GWP = 27.9 - def _build_mapping_with_allocations( mapping: pd.DataFrame, *, use_output_weights: bool @@ -72,10 +69,6 @@ def _build_mapping_with_allocations( return mapping2[['Activity', 'Sector', 'Allocation']].reset_index(drop=True) -def _should_use_output_weighted_mapping() -> bool: - return bool(get_usa_config().use_ghg_national_2023_m2) - - def _apply_electricity_disagg_cornerstone_mapping( mapping: pd.DataFrame, ) -> pd.DataFrame: @@ -106,50 +99,6 @@ def _apply_cornerstone_waste_overrides(mapping: pd.DataFrame) -> pd.DataFrame: ).drop_duplicates() -def _build_naics_to_bea_weighted_mapping() -> pd.DataFrame: - """Build NAICS->BEA mapping weighted by gross output for GHG year. - - When waste disaggregation is enabled, start from the NAICS->BEA crosswalk and - override waste NAICS rows with the Cornerstone waste disaggregation. - """ - cw = load_crosswalk('NAICS_to_BEA_Crosswalk_2017') - mapping = cw.rename( - columns={ - 'NAICS_2017_Code': 'Sector', - 'BEA_2017_Detail_Code': 'Activity', - } - )[['Sector', 'Activity']] - mapping = mapping.dropna().drop_duplicates().astype('string') - if get_usa_config().implement_waste_disaggregation: - mapping = _apply_cornerstone_waste_overrides(mapping) - - cfg = get_usa_config() - go = derive_gross_output( - target_year=cfg.usa_ghg_data_year, - iot_before_or_after_redefinition=cfg.iot_before_or_after_redefinition, - ) - mapping['Output'] = mapping['Activity'].map(go).fillna(0.0) - - group_sum = mapping.groupby('Sector')['Output'].transform('sum') - group_size = mapping.groupby('Sector')['Sector'].transform('size') - bad_one_to_many = (group_size > 1) & (group_sum <= 0) - if bad_one_to_many.any(): - bad_sectors = sorted(mapping.loc[bad_one_to_many, 'Sector'].dropna().unique()) - raise ValueError( - 'Missing/zero gross output for one-to-many weighted NAICS->BEA sectors: ' - f'{bad_sectors[:20]}' - ) - - mapping['Allocation'] = 0.0 - valid_weight = group_sum > 0 - mapping.loc[valid_weight, 'Allocation'] = ( - mapping.loc[valid_weight, 'Output'] / group_sum.loc[valid_weight] - ) - mapping.loc[~valid_weight, 'Allocation'] = 1.0 / group_size.loc[~valid_weight] - - return mapping[['Sector', 'Activity', 'Allocation']].reset_index(drop=True) - - def derive_E_usa() -> pd.DataFrame: return load_E_from_flowsa() @@ -157,61 +106,47 @@ def derive_E_usa() -> pd.DataFrame: def map_fbs_sectors_to_model_schema(fbs: pd.DataFrame) -> pd.DataFrame: """Map FBS NAICS sectors into the active model schema. - Behavior differs by path: - - Weighted path (`use_ghg_national_2023_m2`): preserve original NAICS code - levels (e.g., `53`, `531`, `531110`) and apply weighted NAICS->BEA mapping. - This avoids pre-collapsing aggregates to a single NAICS_6 code. - - Non-weighted paths: expand mixed-digit NAICS to NAICS_6 with a 1:1 - first-match helper mapping, then map into Cornerstone/CEDA activities. + Expands mixed-digit NAICS to NAICS_6 with a 1:1 first-match helper + mapping, then maps into Cornerstone/CEDA activities. """ - use_output_weights = _should_use_output_weighted_mapping() - # For weighted NAICS->BEA mapping (m2 path), preserve the original NAICS - # code level (e.g., 531) so allocation uses all matching crosswalk rows. - # Pre-collapsing to one NAICS_6 (keep='first') can bias allocations. - if use_output_weights: - fbs2 = fbs.copy() - fbs2['NAICS_6'] = fbs2['SectorProducedBy'] - mapping = _build_naics_to_bea_weighted_mapping() - else: - # Prepare NAICS:NAICS_6 expansion used for non-weighted mapping flows. - cw = load_crosswalk('NAICS_2017_Crosswalk') - cols_to_stack = ['NAICS_3', 'NAICS_4', 'NAICS_5'] - cw_stack = ( - cw.astype({c: 'string' for c in cols_to_stack + ['NAICS_6']}) - .melt( - id_vars='NAICS_6', - value_vars=cols_to_stack, - var_name='level', - value_name='NAICS', - ) - .dropna(subset=['NAICS_6', 'NAICS'])[['NAICS', 'NAICS_6']] - .drop_duplicates(subset='NAICS', keep='first') - .reset_index(drop=True) - ) - fbs2 = fbs.merge( - cw_stack, - how='left', - left_on='SectorProducedBy', - right_on='NAICS', - validate='m:1', + # Prepare NAICS:NAICS_6 expansion used for non-weighted mapping flows. + cw = load_crosswalk('NAICS_2017_Crosswalk') + cols_to_stack = ['NAICS_3', 'NAICS_4', 'NAICS_5'] + cw_stack = ( + cw.astype({c: 'string' for c in cols_to_stack + ['NAICS_6']}) + .melt( + id_vars='NAICS_6', + value_vars=cols_to_stack, + var_name='level', + value_name='NAICS', ) - fbs2['NAICS_6'] = fbs2['NAICS_6'].fillna(fbs2['SectorProducedBy']) + .dropna(subset=['NAICS_6', 'NAICS'])[['NAICS', 'NAICS_6']] + .drop_duplicates(subset='NAICS', keep='first') + .reset_index(drop=True) + ) + fbs2 = fbs.merge( + cw_stack, + how='left', + left_on='SectorProducedBy', + right_on='NAICS', + validate='m:1', + ) + fbs2['NAICS_6'] = fbs2['NAICS_6'].fillna(fbs2['SectorProducedBy']) - if get_usa_config().use_cornerstone_2026_model_schema: - mapping = _build_mapping_with_allocations( - get_activitytosector_mapping('Cornerstone_2025'), - use_output_weights=False, - ) - if get_usa_config().implement_electricity_disaggregation: - mapping = _apply_electricity_disagg_cornerstone_mapping(mapping) - else: - mapping = _build_mapping_with_allocations( - get_activitytosector_mapping('CEDA_2025'), - use_output_weights=False, - ) + if get_usa_config().use_cornerstone_2026_model_schema: + mapping = _build_mapping_with_allocations( + get_activitytosector_mapping('Cornerstone_2025'), + use_output_weights=False, + ) + if get_usa_config().implement_electricity_disaggregation: + mapping = _apply_electricity_disagg_cornerstone_mapping(mapping) + else: + mapping = _build_mapping_with_allocations( + get_activitytosector_mapping('CEDA_2025'), + use_output_weights=False, + ) - pre_total = float(fbs2['FlowAmount'].sum()) fbs2 = ( fbs2.merge( mapping[['Activity', 'Sector', 'Allocation']], @@ -230,18 +165,7 @@ def map_fbs_sectors_to_model_schema(fbs: pd.DataFrame) -> pd.DataFrame: ) # Re-assign SectorProducedBy and aggregate using existing functions. - fbs3 = pd.DataFrame(FlowBySector(fbs2).aggregate_flowby()) - - if use_output_weights: - post_total = float(fbs3['FlowAmount'].sum()) - rel_diff = abs(post_total - pre_total) / abs(pre_total) if pre_total else 0.0 - if rel_diff > 0.005: - raise ValueError( - 'FlowAmount conservation failed in weighted NAICS->BEA mapping ' - f'(pre={pre_total}, post={post_total}, rel_diff={rel_diff:.6f})' - ) - - return fbs3 + return pd.DataFrame(FlowBySector(fbs2).aggregate_flowby()) _EGRID_FBS_METHOD = 'GHG_national_Cornerstone_2023_egrid' @@ -326,8 +250,6 @@ def load_E_from_flowsa() -> pd.DataFrame: FBS parquet from GCS. Which inventory/attribution vintages that carries (EPA GHGI vs UMD GHGIA, MECS survey year) is defined per year by the method files in ``bedrock/transform/ghg/``. - - use_ghg_national_2023_m2 → GHG_national_2023_m2 via getFlowBySector - (USEEIO workbook parity). - otherwise → GHG_national_CEDA_{year}, the flowsa implementation of the legacy CEDA allocation methodology (method files exist for 2023 only). """ @@ -343,11 +265,6 @@ def load_E_from_flowsa() -> pd.DataFrame: # `transform/output_data/` (GHG_national_Cornerstone_) directly # so the year-Y diagnostics get year-Y GHG data. fbs = _load_cornerstone_ghg_fbs_from_gcs(year) - elif usa.use_ghg_national_2023_m2: - # For m2, explicitly attempt remote FBS download before generation. - fbs = getFlowBySector( - methodname='GHG_national_2023_m2', download_FBS_if_missing=True - ) else: if year != 2023: raise ValueError( @@ -404,10 +321,6 @@ def load_E_from_flowsa() -> pd.DataFrame: # Convert values to CO2e ghg_mapping: dict[str, float] = {k: v for k, v in GWP100_AR6_CEDA.items()} - if usa.use_ghg_national_2023_m2: - # Keep m2 diagnostics aligned with USEEIO workbook characterization. - ghg_mapping['CH4_fossil'] = _USEEIO_WORKBOOK_CH4_GWP - ghg_mapping['CH4_non_fossil'] = _USEEIO_WORKBOOK_CH4_GWP ghg_mapping['HFCs'] = 1 # should already be in CO2e ghg_mapping['PFCs'] = 1 # should already be in CO2e fbs['CO2e'] = fbs['FlowAmount'] * fbs['Flowable'].map(ghg_mapping) diff --git a/bedrock/transform/iot/__tests__/test_phi_helpers.py b/bedrock/transform/iot/__tests__/test_phi_helpers.py index ad6e629c..bacde4fc 100644 --- a/bedrock/transform/iot/__tests__/test_phi_helpers.py +++ b/bedrock/transform/iot/__tests__/test_phi_helpers.py @@ -15,10 +15,6 @@ class TestMarginsPhiActive: - def test_active_when_useeio_margins(self) -> None: - cfg = USAConfig(useeio_margins=True, cornerstone_industry_avg_margins=False) - assert margins_phi_active(cfg) is True - def test_active_when_cornerstone_margins(self) -> None: cfg = USAConfig(cornerstone_industry_avg_margins=True) assert margins_phi_active(cfg) is True diff --git a/bedrock/transform/iot/derive_PRO_to_PUR_ratio.py b/bedrock/transform/iot/derive_PRO_to_PUR_ratio.py index 0c03920f..3bf93938 100644 --- a/bedrock/transform/iot/derive_PRO_to_PUR_ratio.py +++ b/bedrock/transform/iot/derive_PRO_to_PUR_ratio.py @@ -19,7 +19,6 @@ import pandas as pd from bedrock.extract.iot.io_2017 import load_2017_margins_usa -from bedrock.transform.eeio.derived_2017_helpers import EXPANDED_SECTORS_2012_TO_2017 from bedrock.transform.iot.derived_gross_industry_output import ( available_gross_output_years, ) @@ -32,7 +31,6 @@ from bedrock.utils.taxonomy.bea.v2017_final_demand import USA_2017_FINAL_DEMAND_CODES from bedrock.utils.taxonomy.usa_taxonomy_correspondence_helpers import ( USA_2017_COMMODITY_INDEX, - load_usa_2017_commodity__ceda_v7_correspondence, load_usa_2017_commodity__cornerstone_commodity_correspondence, ) @@ -56,10 +54,6 @@ class MarginsFilters: # Exclude all final demand destinations — the CEDA pipeline only wants # industry-to-industry margin flows. -_ceda_margins_filters: MarginsFilters = MarginsFilters( - exclude_industry_codes=frozenset(USA_2017_FINAL_DEMAND_CODES) -) - # All BEA 2017 detailed commodity codes whose code begins with "4" # (wholesale trade, retail trade, transportation, and warehousing sectors). _COMMODITY_CODES_STARTING_WITH_4: frozenset[str] = frozenset( @@ -105,11 +99,6 @@ class MarginsFilters: # # Matches exported ``useeior`` ``model$Margins``: keep Import ``F05000`` rows because # R ``purchaser_removal`` uses ``%in%`` on a length-3 vector; drop only Export and # change-in-inventories industries; scrap/RoW commodities only (no ``4*``). -_useeio_margins_filters: MarginsFilters = MarginsFilters( - exclude_commodity_codes=frozenset({'S00401', 'S00402', 'S00300', 'S00900'}), - exclude_industry_codes=frozenset({'F04000', 'F03000'}), -) - # Cornerstone filters # Exclude BEA bookkeeping commodities that are removed from model as well as scrap # S00300 Noncomparable imports, S00900 Rest of the world adjustment, S00401 Scrap @@ -131,19 +120,13 @@ class MarginsFilters: def _get_active_margins_filters() -> MarginsFilters: - """Return the active filter set based on config flags. + """Margins filters for the active config. - ``useeio_margins`` takes precedence; otherwise ``cornerstone_industry_avg_margins`` - controls the Cornerstone path, then ``ceda_margins`` the - CEDA path. Returns an empty ``MarginsFilters`` (no-op) when no flag is set. + ``cornerstone_industry_avg_margins`` selects the Cornerstone filter set; + otherwise no filtering applies. """ - cfg = get_usa_config() - if cfg.useeio_margins: - return _useeio_margins_filters - if cfg.cornerstone_industry_avg_margins: + if get_usa_config().cornerstone_industry_avg_margins: return _cornerstone_industry_avg_margins_filters - if cfg.ceda_margins: - return _ceda_margins_filters return MarginsFilters() @@ -168,35 +151,28 @@ def _apply_margins_filter(df: pd.DataFrame, filters: MarginsFilters) -> pd.DataF def _margin_negatives_treatment( df: pd.DataFrame, - abs_negative_producers_value: bool = False, abs_negative_margin_columns: bool = False, ) -> pd.DataFrame: """Flip negative margin values to positive in-place. ``abs_negative_margin_columns`` (triggered by ``cornerstone_industry_avg_margins`` config - flag) flips negatives across all four margin columns and takes precedence. - ``abs_negative_producers_value`` flips only ``Producers' Value``. + flag) flips negatives across all four margin columns. """ if abs_negative_margin_columns: for col in _MARGIN_VALUE_COLUMNS: mask = df[col] < 0 df.loc[mask, col] = df.loc[mask, col].abs() - elif abs_negative_producers_value: - mask = df["Producers' Value"] < 0 - df.loc[mask, "Producers' Value"] = df.loc[mask, "Producers' Value"].abs() return df def _margins_by_commodity( filters: MarginsFilters, - abs_negative_producers_value: bool = False, abs_negative_margin_columns: bool = False, ) -> pd.DataFrame: """Load raw margins, apply ``filters``, and sum to per-commodity totals.""" df = _apply_margins_filter(load_2017_margins_usa(), filters) df = _margin_negatives_treatment( df, - abs_negative_producers_value=abs_negative_producers_value, abs_negative_margin_columns=abs_negative_margin_columns, ) result = ( @@ -216,44 +192,6 @@ def _margins_by_commodity( return result -def derive_2017_margins_ceda_usa() -> pd.DataFrame: - """ - Margins aggregated to CEDA v7 sector taxonomy, summed over all industries. - Applies ``_ceda_margins_filters`` when ``USAConfig.ceda_margins`` is set. - - Returns a DataFrame indexed by CEDA v7 sectors with columns: - ``Producers' Value``, ``Transportation``, ``Wholesale``, ``Retail``, - ``Purchasers' Value``. Unit is USD. - """ - corresp = load_usa_2017_commodity__ceda_v7_correspondence() - corresp.columns.names = ['commodity'] - filters = ( - _ceda_margins_filters if get_usa_config().ceda_margins else MarginsFilters() - ) - margin = corresp @ _margins_by_commodity(filters) - # Expanded sectors share value equally from the aggregated 2012 sector. - margin.loc[EXPANDED_SECTORS_2012_TO_2017, :] *= 1 / len( - EXPANDED_SECTORS_2012_TO_2017 - ) - return margin - - -def derive_phi_ceda_usa() -> pd.Series[float]: - """ - Derive the Phi ratio to convert EF from producer to purchaser price for each CEDA v7 sector. - Formula: purchaser price = producer price + margin - Since original EF is in kgCO2e/USD_producer, Phi here is calculated as - (output_producer / (output_producer + margin)). - """ - margin = derive_2017_margins_ceda_usa() - phi = margin["Producers' Value"] / margin["Purchasers' Value"] - avg_mask = (phi > 0) & (phi <= 1) - avg = phi[avg_mask].mean() - in_range_mask = (phi > 0) & (phi < 1) - phi[~in_range_mask] = avg - return phi - - def _inflate_margin_trade_components( df: pd.DataFrame, *, original_year: int, target_year: int ) -> pd.DataFrame: @@ -269,7 +207,7 @@ def _inflate_margin_trade_components( def _inflate_margins_to_year(df: pd.DataFrame, *, target_year: int) -> pd.DataFrame: """Inflate margin components from ``usa_base_io_data_year`` to *target_year*.""" cfg = get_usa_config() - if not (cfg.useeio_margins or cfg.cornerstone_industry_avg_margins): + if not cfg.cornerstone_industry_avg_margins: return df original_year = cfg.usa_base_io_data_year if original_year == target_year: @@ -288,9 +226,7 @@ def derive_margins_cornerstone_usa_at_year(target_year: int) -> pd.DataFrame: Margins aggregated to Cornerstone commodity taxonomy, summed over all industries. Margin components inflate from ``usa_base_io_data_year`` to *target_year* when - a margins methodology flag is active. PRO inflation follows the useeior ``Rho`` - path when ``useeio_margins`` is set; otherwise the V-norm commodity PI path - (``cornerstone_industry_avg_margins``). + ``cornerstone_industry_avg_margins`` is active (V-norm commodity PI path). Returns a DataFrame indexed by Cornerstone ``COMMODITIES`` with columns: ``Producers' Value``, ``Transportation``, ``Wholesale``, ``Retail``, @@ -300,7 +236,6 @@ def derive_margins_cornerstone_usa_at_year(target_year: int) -> pd.DataFrame: corresp = load_usa_2017_commodity__cornerstone_commodity_correspondence() df = corresp @ _margins_by_commodity( _get_active_margins_filters(), - abs_negative_producers_value=cfg.useeio_margins, abs_negative_margin_columns=cfg.cornerstone_industry_avg_margins, ) df = _inflate_margins_to_year(df, target_year=target_year) @@ -369,7 +304,7 @@ def derive_phi_cornerstone_usa_panel(years: tuple[int, ...]) -> pd.DataFrame: def margins_phi_active(cfg: USAConfig | None = None) -> bool: """Return whether margins-based Phi should be applied for *cfg*.""" c = cfg or get_usa_config() - return bool(c.useeio_margins or c.cornerstone_industry_avg_margins) + return bool(c.cornerstone_industry_avg_margins) def phi_for_sectors( diff --git a/bedrock/utils/config/__tests__/test_usa_config.py b/bedrock/utils/config/__tests__/test_usa_config.py index 41048807..267574ef 100644 --- a/bedrock/utils/config/__tests__/test_usa_config.py +++ b/bedrock/utils/config/__tests__/test_usa_config.py @@ -118,49 +118,6 @@ def test_config_via_environment_variable() -> None: assert usa_config.usa_ghg_data_year == 2023 -def test_phoebe_year_vector_accepts_2017_io_fields() -> None: - cfg = USAConfig.model_validate( - {'model_base_year': 2017, 'usa_io_data_year': 2017}, - strict=True, - ) - assert cfg.model_base_year == 2017 - assert cfg.usa_io_data_year == 2017 - - -def test_disallow_cornerstone_ghg_model_with_m2_flag() -> None: - with pytest.raises( - ValueError, - match='use_cornerstone_ghg_model and use_ghg_national_2023_m2 ', - ): - USAConfig.model_validate( - {'use_cornerstone_ghg_model': True, 'use_ghg_national_2023_m2': True}, - strict=True, - ) - - -def test_allow_m2_flag_when_cornerstone_ghg_model_is_false() -> None: - cfg = USAConfig.model_validate( - { - 'use_cornerstone_ghg_model': False, - 'use_ghg_national_2023_m2': True, - 'use_useeio_schema': True, - }, - strict=True, - ) - assert cfg.use_ghg_national_2023_m2 is True - - -def test_disallow_m2_without_useeio_schema() -> None: - with pytest.raises( - ValueError, - match='use_ghg_national_2023_m2 requires use_useeio_schema to be true', - ): - USAConfig.model_validate( - {'use_ghg_national_2023_m2': True, 'use_useeio_schema': False}, - strict=True, - ) - - def test_disallow_deflate_x_without_use_e_for_x_in_b() -> None: with pytest.raises( ValueError, @@ -175,19 +132,6 @@ def test_disallow_deflate_x_without_use_e_for_x_in_b() -> None: ) -@pytest.mark.parametrize( - 'flags', - [ - {'useeio_margins': True, 'ceda_margins': True}, - {'useeio_margins': True, 'cornerstone_industry_avg_margins': True}, - {'ceda_margins': True, 'cornerstone_industry_avg_margins': True}, - ], -) -def test_disallow_multiple_margins_flags(flags: dict[str, bool]) -> None: - with pytest.raises(ValueError, match='At most one margins flag may be true'): - USAConfig.model_validate(flags, strict=True) - - def test_electricity_disagg_config_parsing() -> None: config = _load_usa_config_from_file_name( 'test_usa_config_waste_disagg_electricity.yaml' diff --git a/bedrock/utils/config/configs/useeio_phoebe_23.yaml b/bedrock/utils/config/configs/useeio_phoebe_23.yaml deleted file mode 100644 index d2ae2a7e..00000000 --- a/bedrock/utils/config/configs/useeio_phoebe_23.yaml +++ /dev/null @@ -1,33 +0,0 @@ -# USEEIO Phoebe anchor (all five alternate toggles enabled). -# Year vector: IO/economic = 2017, GHG inventory = 2023. - -##### -# Model base settings -##### -model_base_year: 2017 -iot_before_or_after_redefinition: before - -##### -# Data selection -##### -usa_io_data_year: 2017 -usa_ghg_data_year: 2023 - -##### -# Methodology selection -##### -use_cornerstone_2026_model_schema: true -use_E_data_year_for_x_in_B: true -implement_waste_disaggregation: true - -# USEEIO-style A: keep 2017 base A when scaling A,q (PR #229 / Issue #182). -scale_a_matrix_with_useeio_method: true - -use_useeio_schema: true -deflate_x_to_detail_io_year_for_B: true -use_ghg_national_2023_m2: true - -# Must remain false whenever use_ghg_national_2023_m2 is true. - -useeio_margins: True -cornerstone_industry_avg_margins: False \ No newline at end of file diff --git a/bedrock/utils/config/configs/v03_waterfall_g2_methods.yaml b/bedrock/utils/config/configs/v03_waterfall_g2_methods.yaml index c72272fb..16a27f2d 100644 --- a/bedrock/utils/config/configs/v03_waterfall_g2_methods.yaml +++ b/bedrock/utils/config/configs/v03_waterfall_g2_methods.yaml @@ -12,5 +12,4 @@ apply_io_year_adjustments: true implement_waste_disaggregation: true scale_a_matrix_with_useeio_method: false -useeio_margins: false cornerstone_industry_avg_margins: true diff --git a/bedrock/utils/config/configs/v03_waterfall_useeio_g1_schema_ghg.yaml b/bedrock/utils/config/configs/v03_waterfall_useeio_g1_schema_ghg.yaml deleted file mode 100644 index 5314e2b9..00000000 --- a/bedrock/utils/config/configs/v03_waterfall_useeio_g1_schema_ghg.yaml +++ /dev/null @@ -1,21 +0,0 @@ -# v03_waterfall USEEIO G1: USEEIO-like A/margins + Cornerstone schema/GHG @ IO@2024 producer. -# Cumulative wholesale USEEIO group anchor (pinned → Cornerstone GHG + schema). - -model_base_year: 2024 -price_type: producer -iot_before_or_after_redefinition: before - -usa_ghg_data_year: 2023 - -use_cornerstone_2026_model_schema: true -use_E_data_year_for_x_in_B: true -implement_waste_disaggregation: true - -scale_a_matrix_with_useeio_method: true -use_useeio_schema: false -deflate_x_to_detail_io_year_for_B: true -use_ghg_national_2023_m2: false -use_cornerstone_ghg_model: true - -useeio_margins: true -cornerstone_industry_avg_margins: false diff --git a/bedrock/utils/config/usa_config.py b/bedrock/utils/config/usa_config.py index a788d2b2..d94a0372 100644 --- a/bedrock/utils/config/usa_config.py +++ b/bedrock/utils/config/usa_config.py @@ -62,7 +62,6 @@ class USAConfig(BaseModel): ##### ### Schema/Taxonomy selection use_cornerstone_2026_model_schema: bool = False # DRI: mo.li - use_useeio_schema: bool = False ### IO Methodology selection # "IO year adjustments" bucket: CEDA A/q scaling to usa_io_data_year with # summary dollar-year rebase, bedrock-derived industry inflation factors, @@ -91,14 +90,11 @@ class USAConfig(BaseModel): implement_electricity_disaggregation: bool = False # DRI: jorge.vendries implement_electricity_mixed_units: bool = False # DRI: jorge.vendries scale_a_matrix_with_useeio_method: bool = False # DRI: mo.li - ceda_margins: bool = False # DRI: WesIngwersen - useeio_margins: bool = False # DRI: WesIngwersen cornerstone_industry_avg_margins: bool = False # DRI: WesIngwersen ### GHG Methodology selection # "GHG model allocation" bucket: Cornerstone GHG FBS (pre-built parquet at # usa_ghg_data_year) vs the legacy CEDA-methodology FBS (2023 only). use_cornerstone_ghg_model: bool = False - use_ghg_national_2023_m2: bool = False ##### # Diagnostics baseline (parquet snapshots vs USEEIO Excel on GCS) @@ -156,37 +152,8 @@ def _validate_deflate_x_requires_use_e_for_x_in_b(self) -> USAConfig: ) return self - @model_validator(mode='after') - def _validate_margins_mutual_exclusivity(self) -> USAConfig: - active = [ - name - for name, val in [ - ('useeio_margins', self.useeio_margins), - ('ceda_margins', self.ceda_margins), - ( - 'cornerstone_industry_avg_margins', - self.cornerstone_industry_avg_margins, - ), - ] - if val - ] - if len(active) > 1: - raise ValueError( - f'At most one margins flag may be true; got: {", ".join(active)}' - ) - return self - @model_validator(mode='after') def _validate_ghg_flag_compatibility(self) -> USAConfig: - if self.use_cornerstone_ghg_model and self.use_ghg_national_2023_m2: - raise ValueError( - 'use_cornerstone_ghg_model and use_ghg_national_2023_m2 ' - 'cannot both be true' - ) - if self.use_ghg_national_2023_m2 and not self.use_useeio_schema: - raise ValueError( - 'use_ghg_national_2023_m2 requires use_useeio_schema to be true' - ) if ( self.implement_electricity_reallocation and not self.implement_waste_disaggregation diff --git a/bedrock/utils/economic/inflation_helpers_cornerstone.py b/bedrock/utils/economic/inflation_helpers_cornerstone.py index 4550b34d..7cebccc8 100644 --- a/bedrock/utils/economic/inflation_helpers_cornerstone.py +++ b/bedrock/utils/economic/inflation_helpers_cornerstone.py @@ -21,9 +21,6 @@ from bedrock.utils.economic.inflation_helpers_ceda import ( obtain_inflation_factors_from_reference_data, ) -from bedrock.utils.economic.inflation_helpers_useeio import ( - obtain_useeior_detail_industry_cpi_levels, -) from bedrock.utils.math.formulas import ( compute_commodity_mix_matrix, compute_Vnorm_matrix, @@ -83,13 +80,10 @@ def _cornerstone_to_ceda_v7_parent() -> dict[str, str]: def _industry_price_index_levels() -> pd.DataFrame: """Wide sector × year PI levels for cornerstone industry-ratio math. - ``useeio_margins``: USEEIOR v1.8.0 ``Detail_CPI_IO_17sch`` (GCS snapshot). ``apply_io_year_adjustments``: bedrock-derived industry PI. Otherwise: bedrock BEA parquet (``BEA_PriceIndex``). """ cfg = get_usa_config() - if cfg.useeio_margins: - return obtain_useeior_detail_industry_cpi_levels() if cfg.apply_io_year_adjustments: return derive_industry_price_index() return obtain_inflation_factors_from_reference_data() @@ -146,7 +140,7 @@ def get_rho_inflation_ratio(original_year: int, target_year: int) -> pd.Series[f (``calculateProducerbyPurchaserPriceRatio``). Implemented on the same 1:1 sector index as ``get_cornerstone_industry_price_ratio``. - This is the ``useeio_margins`` branch of ``get_price_index_ratio``; kept + Kept for the Excel ``Rho`` panel; as a named primitive for sector-level index math and tests. """ if original_year == target_year: @@ -188,7 +182,6 @@ def derive_price_index_panel(years: tuple[int, ...]) -> pd.DataFrame: def get_price_index_ratio(original_year: int, target_year: int) -> pd.Series[float]: """Cross-year price-index ratio for the active config construction. - ``useeio_margins``: sector 1:1 ``PI[original] / PI[target]`` (Rho approach). ``cornerstone_industry_avg_margins``: V-norm commodity ``PI[target] / PI[original]``. @@ -198,8 +191,6 @@ def get_price_index_ratio(original_year: int, target_year: int) -> pd.Series[flo which always uses the useeior sector 1:1 convention. """ cfg = get_usa_config() - if cfg.useeio_margins: - return get_rho_inflation_ratio(original_year, target_year) if cfg.cornerstone_industry_avg_margins: return get_vnorm_adjusted_commodity_price_ratio(original_year, target_year) return pd.Series( @@ -685,7 +676,6 @@ def _cornerstone_indexed_industry_pi(year: int) -> pd.Series[float]: _get_summary_industry_price_index, get_summary_industry_price_ratio, derive_cornerstone_q_and_vnorm_for_year, - obtain_useeior_detail_industry_cpi_levels, ) diff --git a/bedrock/utils/economic/inflation_helpers_useeio.py b/bedrock/utils/economic/inflation_helpers_useeio.py deleted file mode 100644 index 51d8eebf..00000000 --- a/bedrock/utils/economic/inflation_helpers_useeio.py +++ /dev/null @@ -1,84 +0,0 @@ -"""USEEIOR industry CPI levels — separate from bedrock BEA / CEDA parquet. - -Loaded only when ``USAConfig.useeio_margins`` is true. - -Upstream (cornerstone-data/useeior) ------------------------------------- -- Object: ``data/Detail_CPI_IO_17sch.rda`` → ``MultiYearIndustryCPI`` in Detail / - 2017-schema ``loadIOData``. -- Bedrock pin: tag ``v1.8.0`` (released 2025-11-11). -- Last change to ``Detail_CPI_IO_17sch.rda`` on ``v1.8.0``: **2025-09-26**, commit - ``4007dad`` (*update GO, CPI, and Value added from annual update #7*). - -Bedrock packaging ------------------ -- GCS: ``extract/input-data/USEEIOR_v180_IndustryCPI/useeior_v1.8.0_Detail_CPI_IO_17sch.csv`` -- Local cache: ``bedrock/extract/input_data/USEEIOR_v180_IndustryCPI/`` (gitignored CSV) - -Cornerstone configs (``useeio_margins: false``) keep ``BEA_PriceIndex`` parquet. -""" - -from __future__ import annotations - -import functools -import os - -import pandas as pd - -from bedrock.utils.io.gcp import download_gcs_file_if_not_exists -from bedrock.utils.io.gcp_paths import gcs_extract_input_path -from bedrock.utils.io.local_extract_input_data import local_extract_input_dir - -# Distinct from ``BEA_PriceIndex`` (bedrock parquet) and ``apply_io_year_adjustments``. -USEEIOR_INDUSTRY_CPI_GCS_SOURCE = 'USEEIOR_v180_IndustryCPI' -USEEIOR_DETAIL_CPI_IO_17SCH_FILENAME = 'useeior_v1.8.0_Detail_CPI_IO_17sch.csv' -USEEIOR_DETAIL_CPI_IO_17SCH_GCS_PATH = gcs_extract_input_path( - USEEIOR_INDUSTRY_CPI_GCS_SOURCE -) -USEEIOR_DETAIL_CPI_TAG = 'v1.8.0' -USEEIOR_DETAIL_CPI_UPSTREAM_PATH = 'data/Detail_CPI_IO_17sch.rda' -USEEIOR_DETAIL_CPI_LAST_UPSTREAM_CHANGE = '2025-09-26' # useeior 4007dad @ v1.8.0 - - -def _normalize_useeior_sector_index(index: pd.Index) -> pd.Index: - return pd.Index( - index.astype(str).str.replace('/US', '', regex=False).str.strip(), - name=index.name, - ) - - -@functools.cache -def obtain_useeior_detail_industry_cpi_levels() -> pd.DataFrame: - """Sector × year industry CPI levels from pinned USEEIOR v1.8.0 export. - - Returns a wide ``DataFrame``: index = BEA 2017 detail sector codes, - columns = int years, values = chain-type price index (2017 = 100). - """ - local_path = os.path.join( - local_extract_input_dir(USEEIOR_INDUSTRY_CPI_GCS_SOURCE), - USEEIOR_DETAIL_CPI_IO_17SCH_FILENAME, - ) - download_gcs_file_if_not_exists( - USEEIOR_DETAIL_CPI_IO_17SCH_FILENAME, - USEEIOR_DETAIL_CPI_IO_17SCH_GCS_PATH, - local_path, - ) - if not os.path.isfile(local_path): - raise FileNotFoundError( - f'USEEIOR industry CPI not found at {local_path!r}. ' - f'Expected GCS object gs://cornerstone-default/' - f'{USEEIOR_DETAIL_CPI_IO_17SCH_GCS_PATH}/' - f'{USEEIOR_DETAIL_CPI_IO_17SCH_FILENAME} or a local copy under ' - f'bedrock/extract/input_data/{USEEIOR_INDUSTRY_CPI_GCS_SOURCE}/. ' - f'Upstream: useeior {USEEIOR_DETAIL_CPI_UPSTREAM_PATH} @ ' - f'tag {USEEIOR_DETAIL_CPI_TAG} (last changed ' - f'{USEEIOR_DETAIL_CPI_LAST_UPSTREAM_CHANGE}).' - ) - - raw = pd.read_csv(local_path, index_col=0) - raw.index = _normalize_useeior_sector_index(raw.index) - year_cols: dict[int, pd.Series] = {} - for col in raw.columns: - year_cols[int(str(col).strip())] = pd.to_numeric(raw[col], errors='coerce') - out = pd.DataFrame(year_cols).sort_index(axis=1) - return out.astype(float)