多行列标题下ADB MRIO数据库国家与部门聚合技术问询

明确聚合需求 #

1. 国家维度（行/列通用） #

• 合并为ROW：BRA、CAN、MEX、原ROW
• 合并为ROA：AUS、TAP、KAZ、SRI、PAK、FIJ、BRU、BHU、KGZ、MLD
• 保留原样：PRC、JPN、KOR、BAN、CAM、HKG、INO、IND、LAO、MAL、MON、NEP、PHI、SIN、THA、VIE、US、GER
• 剩余所有国家合并为EU

2. 部门维度 #

• 保留部门：C4、C12、C13、C14、C15
• 其余部门统一合并为OTH

3. F1-F5需求项 #

• 完全遵循上述国家聚合规则执行合并

实操实现与优化 #

MRIO数据一般是国家+部门的多层级索引结构，用Pandas的groupby结合映射字典是最高效的方式，以下是验证和优化方案：

第一步：构建映射字典 #

先把所有聚合规则转换成字典，避免后续重复逻辑：

import pandas as pd
import numpy as np

# 1. 国家聚合映射
country_map = {}
# 填充ROW组
row_grp = ['BRA', 'CAN', 'MEX', 'ROW']
for c in row_grp:
    country_map[c] = 'ROW'
# 填充ROA组
roa_grp = ['AUS', 'TAP', 'KAZ', 'SRI', 'PAK', 'FIJ', 'BRU', 'BHU', 'KGZ', 'MLD']
for c in roa_grp:
    country_map[c] = 'ROA'
# 保留原样的国家
keep_grp = ['PRC', 'JPN', 'KOR', 'BAN', 'CAM', 'HKG', 'INO', 'IND', 'LAO', 'MAL', 'MON', 'NEP', 'PHI', 'SIN', 'THA', 'VIE', 'US', 'GER']
for c in keep_grp:
    country_map[c] = c
# 剩余国家自动映射为EU（如果有原始国家列表，直接提前定义更稳妥）
all_countries = pd.Index(your_mrio_data.index.get_level_values(0)).unique()
for c in all_countries:
    if c not in country_map:
        country_map[c] = 'EU'

# 2. 部门聚合映射
sector_map = {s: s if s in ['C4', 'C12', 'C13', 'C14', 'C15'] else 'OTH' 
              for s in pd.Index(your_mrio_data.index.get_level_values(1)).unique()}

第二步：执行行列聚合 #

直接替换原索引为聚合后的新层级，再分组求和：

# 生成新的行索引（聚合后的国家+部门）
new_rows = pd.MultiIndex.from_tuples(
    [(country_map[c], sector_map[s]) for c, s in your_mrio_data.index],
    names=['Agg_Country', 'Agg_Sector']
)
# 生成新的列索引
new_cols = pd.MultiIndex.from_tuples(
    [(country_map[c], sector_map[s]) for c, s in your_mrio_data.columns],
    names=['Agg_Country', 'Agg_Sector']
)

# 替换索引后分组求和，完成聚合
aggregated_mrio = your_mrio_data.set_axis(new_rows, axis=0)\
                                .set_axis(new_cols, axis=1)\
                                .groupby(level=['Agg_Country', 'Agg_Sector'], axis=0).sum()\
                                .groupby(level=['Agg_Country', 'Agg_Sector'], axis=1).sum()

第三步：F1-F5项的聚合 #

如果F1-F5是单独的数据集（比如索引为国家的单列/多列），直接复用国家映射：

# 假设f_data是F1-F5的DataFrame，索引为国家代码
f_agg = f_data.assign(Agg_Country=f_data.index.map(country_map))\
              .groupby('Agg_Country').sum()

验证要点 #

你需要重点检查以下几点，确保聚合正确：

1. 分组数量：国家组应为21个（ROW+ROA+18个保留国+EU），部门组应为6个（5个保留部门+OTH）
2. 数值一致性：随机抽取原始数据中某组的总和（比如BRA所有部门的数值之和），与聚合后ROW组对应部门的数值对比，应完全一致
3. 索引正确性：检查聚合后的MultiIndex层级是否正确，无错误映射或缺失项
4. 空值处理：如果原始数据有缺失，确认聚合时是否正确处理（比如用sum(min_count=1)避免无数据时返回0）

优化方向 #

如果你的代码已经实现了核心逻辑，可以从以下几点优化：

1. 预定义所有映射：不要依赖原始数据推导剩余国家，提前把63个国家的映射规则写全，避免遗漏
2. 用分类类型加速：把聚合后的国家/部门列转为pd.Categorical类型，能大幅提升groupby的执行速度
3. 避免循环：用字典推导式替代手动循环填充映射字典，代码更简洁高效
4. 批量处理：如果有多个MRIO表需要聚合，把映射和聚合逻辑封装成函数，复用性更强

内容的提问来源于stack exchange，提问作者annely