如何简化Pandas DataFrame多投影系统关联的重复代码？

我正在开发一个函数，允许用户导入一组Over/Under（上下盘）数据列表，并将其与一系列投影系统数据关联，以此对比球员的某项数据（比如安打Over/Under值与6个不同系统的安打预测值）。当前函数通过if分支处理用户选择的三种输入参数（安打hits、本垒打home runs、打点rbis），但每个分支内的关联、字段筛选、平均值计算等步骤大量重复。我希望代码能根据初始参数选择自动执行上述步骤（不同参数对应投影系统中的不同字段，比如选择RBI时对应rbi_pecota_50而非h_pecota_50），避免重复编写逻辑，求可行的实现方案。

当前代码片段 #

if investment_type_sub.lower() == 'hits':
        
        # 选择用于关联投影数据的ID列
        ou_hits_all_projections = ou_hitting_ids[['BPID','IDFANGRAPHS', 'DAVENPORTID', 'Player', 'Hits_OU']]

        ou_hits_all_projections = ou_hits_all_projections.query('Hits_OU > 0')

        # 关联PECOTA 50分位数据
        ou_hits_all_projections = pd.merge(ou_hits_all_projections, pecota_hitting_50, how = 'left', left_on='BPID', right_on='bpid_PECOTA_50')
        ou_hits_all_projections = ou_hits_all_projections[['BPID','IDFANGRAPHS', 'DAVENPORTID', 'Player', 'Hits_OU','h_PECOTA_50']]

        # 关联PECOTA 99分位数据
        ou_hits_all_projections = pd.merge(ou_hits_all_projections, pecota_hitting_99, how = 'left', left_on='BPID', right_on='bpid_PECOTA_99')
        ou_hits_all_projections = ou_hits_all_projections[['BPID','IDFANGRAPHS', 'DAVENPORTID', 'Player', 'Hits_OU','h_PECOTA_50', 'h_PECOTA_99']]

        # 关联ZIPs数据
        ou_hits_all_projections = pd.merge(ou_hits_all_projections, zips_hitting, how = 'left', left_on='IDFANGRAPHS', right_on='PlayerId_ZIPs')
        ou_hits_all_projections = ou_hits_all_projections[['BPID','IDFANGRAPHS', 'DAVENPORTID', 'Player', 'Hits_OU', 'h_PECOTA_50','h_PECOTA_99', 'H_ZIPs']]

        # 关联Steamer数据
        ou_hits_all_projections = pd.merge(ou_hits_all_projections, steamer_hitting, how = 'left', left_on='IDFANGRAPHS', right_on='PlayerId_steamer')
        ou_hits_all_projections = ou_hits_all_projections[['BPID','IDFANGRAPHS', 'DAVENPORTID', 'Player', 'Hits_OU', 'h_PECOTA_50','h_PECOTA_99', 'H_ZIPs', 'H_steamer']]

        # 关联The Bat数据
        ou_hits_all_projections = pd.merge(ou_hits_all_projections, the_bat_hitting, how = 'left', left_on='IDFANGRAPHS', right_on='PlayerId_the_bat')
        ou_hits_all_projections = ou_hits_all_projections[['BPID','IDFANGRAPHS', 'DAVENPORTID', 'Player', 'Hits_OU', 'h_PECOTA_50','h_PECOTA_99', 'H_ZIPs', 'H_steamer', 'H_the_bat']]

        # 关联Davenport数据
        ou_hits_all_projections = pd.merge(ou_hits_all_projections, davenport_hitting, how = 'left', left_on='DAVENPORTID', right_on='HOWEID_davenport')
        ou_hits_all_projections = ou_hits_all_projections[['BPID','IDFANGRAPHS', 'DAVENPORTID', 'Player', 'Hits_OU', 'h_PECOTA_50','h_PECOTA_99', 'H_ZIPs', 'H_steamer', 'H_the_bat', 'H_davenport']]

        # 计算所有投影系统的平均值
        ou_hits_all_projections['hit_avg'] = np.where((ou_hits_all_projections.Hits_OU > 0),
                                             ((ou_hits_all_projections.h_PECOTA_50+
                                               ou_hits_all_projections.h_PECOTA_99+
                                               ou_hits_all_projections.H_ZIPs
                                              +ou_hits_all_projections.H_steamer + ou_hits_all_projections.H_the_bat +
                                               ou_hits_all_projections.H_davenport))/
                                          ((len(ou_hits_all_projections.columns)-5)), 0)
        
        # 计算平均值与OU值的百分比差异
        ou_hits_all_projections['hit_avg_diff'] = (ou_hits_all_projections['Hits_OU']-ou_hits_all_projections['hit_avg'])/(ou_hits_all_projections['Hits_OU'])
        
        return ou_hits_all_projections.sort_values('hit_avg_diff', ascending=False).reset_index(drop=True)

核心思路是用配置字典映射不同参数对应的字段规则，然后把重复的逻辑（数据初始化、关联投影、计算统计值）封装成通用流程，通过循环或函数调用完成，避免重复代码。

1. 定义参数映射配置 #

先创建一个字典，把每个参数（hits/home runs/rbis）对应的所有相关字段统一管理：

param_config = {
    'hits': {
        'ou_col': 'Hits_OU',
        'projection_fields': [
            ('pecota_hitting_50', 'BPID', 'bpid_PECOTA_50', 'h_PECOTA_50'),
            ('pecota_hitting_99', 'BPID', 'bpid_PECOTA_99', 'h_PECOTA_99'),
            ('zips_hitting', 'IDFANGRAPHS', 'PlayerId_ZIPs', 'H_ZIPs'),
            ('steamer_hitting', 'IDFANGRAPHS', 'PlayerId_steamer', 'H_steamer'),
            ('the_bat_hitting', 'IDFANGRAPHS', 'PlayerId_the_bat', 'H_the_bat'),
            ('davenport_hitting', 'DAVENPORTID', 'HOWEID_davenport', 'H_davenport')
        ],
        'avg_col': 'hit_avg',
        'diff_col': 'hit_avg_diff'
    },
    'home runs': {
        'ou_col': 'HR_OU',
        'projection_fields': [
            ('pecota_hitting_50', 'BPID', 'bpid_PECOTA_50', 'hr_PECOTA_50'),
            ('pecota_hitting_99', 'BPID', 'bpid_PECOTA_99', 'hr_PECOTA_99'),
            ('zips_hitting', 'IDFANGRAPHS', 'PlayerId_ZIPs', 'HR_ZIPs'),
            ('steamer_hitting', 'IDFANGRAPHS', 'PlayerId_steamer', 'HR_steamer'),
            ('the_bat_hitting', 'IDFANGRAPHS', 'PlayerId_the_bat', 'HR_the_bat'),
            ('davenport_hitting', 'DAVENPORTID', 'HOWEID_davenport', 'HR_davenport')
        ],
        'avg_col': 'hr_avg',
        'diff_col': 'hr_avg_diff'
    },
    'rbis': {
        'ou_col': 'RBI_OU',
        'projection_fields': [
            ('pecota_hitting_50', 'BPID', 'bpid_PECOTA_50', 'rbi_PECOTA_50'),
            ('pecota_hitting_99', 'BPID', 'bpid_PECOTA_99', 'rbi_PECOTA_99'),
            ('zips_hitting', 'IDFANGRAPHS', 'PlayerId_ZIPs', 'RBI_ZIPs'),
            ('steamer_hitting', 'IDFANGRAPHS', 'PlayerId_steamer', 'RBI_steamer'),
            ('the_bat_hitting', 'IDFANGRAPHS', 'PlayerId_the_bat', 'RBI_the_bat'),
            ('davenport_hitting', 'DAVENPORTID', 'HOWEID_davenport', 'RBI_davenport')
        ],
        'avg_col': 'rbi_avg',
        'diff_col': 'rbi_avg_diff'
    }
}

注：需根据实际数据集调整字段名，比如本垒打对应的PECOTA字段为hr_PECOTA_50、打点为rbi_PECOTA_50，确保与你的数据匹配。

2. 封装通用处理函数 #

把重复逻辑写成函数，根据输入参数从配置字典取规则，自动执行所有步骤：

def process_ou_data(investment_type_sub, ou_hitting_ids, **projection_dfs):
    # 统一参数小写，匹配配置
    param = investment_type_sub.lower()
    config = param_config.get(param)
    if not config:
        raise ValueError(f"不支持的参数类型：{investment_type_sub}")
    
    # 初始化数据：选择ID列和对应的OU列
    base_cols = ['BPID','IDFANGRAPHS', 'DAVENPORTID', 'Player', config['ou_col']]
    df = ou_hitting_ids[base_cols].query(f"{config['ou_col']} > 0")
    
    # 循环关联所有投影系统数据
    current_cols = base_cols.copy()
    for df_name, left_on, right_on, proj_col in config['projection_fields']:
        # 获取投影数据集（从传入的关键字参数中取）
        proj_df = projection_dfs[df_name]
        df = pd.merge(df, proj_df, how='left', left_on=left_on, right_on=right_on)
        # 更新要保留的列
        current_cols.append(proj_col)
        df = df[current_cols]
    
    # 计算投影平均值：自动筛选出所有投影字段求和
    proj_cols = [col for _,_,_,col in config['projection_fields']]
    df[config['avg_col']] = np.where(df[config['ou_col']] > 0,
                                     df[proj_cols].sum(axis=1) / len(proj_cols),
                                     0)
    
    # 计算百分比差异
    df[config['diff_col']] = (df[config['ou_col']] - df[config['avg_col']]) / df[config['ou_col']]
    
    # 返回排序后的结果
    return df.sort_values(config['diff_col'], ascending=False).reset_index(drop=True)

3. 调用函数 #

处理不同参数时，只需调用该函数即可，无需重复写if分支：

# 处理安打数据
result_hits = process_ou_data('hits', ou_hitting_ids,
                              pecota_hitting_50=pecota_hitting_50,
                              pecota_hitting_99=pecota_hitting_99,
                              zips_hitting=zips_hitting,
                              steamer_hitting=steamer_hitting,
                              the_bat_hitting=the_bat_hitting,
                              davenport_hitting=davenport_hitting)

# 处理本垒打数据
result_hr = process_ou_data('home runs', ou_hitting_ids,
                            pecota_hitting_50=pecota_hitting_50,
                            pecota_hitting_99=pecota_hitting_99,
                            zips_hitting=zips_hitting,
                            steamer_hitting=steamer_hitting,
                            the_bat_hitting=the_bat_hitting,
                            davenport_hitting=davenport_hitting)

# 处理打点数据
result_rbi = process_ou_data('rbis', ou_hitting_ids,
                             pecota_hitting_50=pecota_hitting_50,
                             pecota_hitting_99=pecota_hitting_99,
                             zips_hitting=zips_hitting,
                             steamer_hitting=steamer_hitting,
                             the_bat_hitting=the_bat_hitting,
                             davenport_hitting=davenport_hitting)

优化点说明 #

• 配置驱动：新增参数或修改字段时，只需更新param_config字典，无需改动核心逻辑代码。
• 自动列管理：通过current_cols动态维护要保留的列，避免硬编码列名列表。
• 平均值计算优化：用df[proj_cols].sum(axis=1)代替手动相加，更简洁且不易出错。
• 扩展性强：后续新增投影系统时，只需在对应参数的projection_fields中添加一行规则即可。

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