Python如何精简CSV统计场景的大量重复if-elif-else分支

读取CSV数据运行主循环时，通过num_2 += 1做轮次计数，每轮需要累加当前轮次数据后计算中位数、标准差统计值。由于需要处理约50个对象，现有实现需要编写近50条if-elif-else分支，需要精简代码、提升执行效率。原有代码如下：

if num_2 == 1:
    #mag
    test1_flux.append(mag_flux)
    median1_flux = np.median(test1_flux)
    test1_ap.append(mag_ap)
    median1_ap = np.median(test1_ap)
    #err
    test1_flux_err.append(mag_flux)
    std1_flux = np.std(test1_flux_err)
    test1_ap_err.append(mag_flux)
    std1_ap = np.std(test1_ap_err)
elif num_2 == 2:
    test2_flux.append(mag_flux)
    median2_flux = np.median(test2_flux)
    test2_ap.append(mag_ap)
    median2_ap = np.median(test2_ap)
    #err
    test2_flux_err.append(mag_flux)
    std2_flux = np.std(test2_flux_err)
    test2_ap_err.append(mag_flux)
    std2_ap = np.std(test2_ap_err)
elif num_2 == 3:
    test3_flux.append(mag_flux)
    median3_flux = np.median(test3_flux)
    test3_ap.append(mag_ap)
    median3_ap = np.median(test3_ap)
    #err
    test3_flux_err.append(mag_flux)
    std3_flux = np.std(test3_flux_err)
    test3_ap_err.append(mag_flux)
    std3_ap = np.std(test3_ap_err)
elif num_2 == 4:
    test4_flux.append(mag_flux)
    median4_flux = np.median(test4_flux)
    test4_ap.append(mag_ap)
    median4_ap = np.median(test4_ap)
    #err
    test4_flux_err.append(mag_flux)
    std4_flux = np.std(test4_flux_err)
    test4_ap_err.append(mag_flux)
    std4_ap = np.std(test4_ap_err)
elif num_2 == 5:
    test5_flux.append(mag_flux)
    median5_flux = np.median(test5_flux)
    test5_ap.append(mag_ap)
    median5_ap = np.median(test5_ap)
    #err
    test5_flux_err.append(mag_flux)
    std5_flux = np.std(test5_flux_err)
    test5_ap_err.append(mag_flux)
    std5_ap = np.std(test5_ap_err)
elif num_2 == 6:
    test6_flux.append(mag_flux)
    median6_flux = np.median(test6_flux)
    test6_ap.append(mag_ap)
    median6_ap = np.median(test6_ap)
    #err
    test6_flux_err.append(mag_flux)
    std6_flux = np.std(test6_flux_err)
    test6_ap_err.append(mag_flux)
    std6_ap = np.std(test6_ap_err)
elif num_2 == 7:
    test7_flux.append(mag_flux)
    median7_flux = np.median(test7_flux)
    test7_ap.append(mag_ap)
    median7_ap = np.median(test7_ap)
    #err
    test7_flux_err.append(mag_flux)
    std7_flux = np.std(test7_flux_err)
    test7_ap_err.append(mag_flux)
    std7_ap = np.std(test7_ap_err)
else:
    pass

核心思路是用结构化容器替代硬编码分支，彻底删除所有条件判断，后续扩展处理对象数量时无需新增分支代码。

具体实现 #

代码中test1_flux、test2_flux这类按序号命名的变量，本质是同结构的分组数据，完全可以统一存入列表/字典容器，直接用num_2作为索引定位当前处理的分组，不需要逐一枚举判断。

• 循环开始前提前初始化分组容器，num_2从1开始计数，因此容器多预留1个索引位对齐序号，避免偏移计算：

import numpy as np

# 配置需要处理的总对象数量，后续调整数量只需要改这个值
MAX_TARGET_NUM = 50
# 初始化分组容器，索引0留空，1~MAX_TARGET_NUM 对应num_2=1到MAX_TARGET_NUM的轮次
test_groups = [
    {
        "flux": [],
        "ap": [],
        "flux_err": [],
        "ap_err": [],
        "median_flux": None,
        "median_ap": None,
        "std_flux": None,
        "std_ap": None
    } for _ in range(MAX_TARGET_NUM + 1)
]

• 把原有所有if-elif分支替换为以下逻辑，直接通过索引定位当前分组操作：

current = test_groups[num_2]
# 处理mag相关统计
current["flux"].append(mag_flux)
current["median_flux"] = np.median(current["flux"])
current["ap"].append(mag_ap)
current["median_ap"] = np.median(current["ap"])
# 处理err相关统计
# 注意：原代码err分支下append传入的均为mag_flux，若为笔误请替换为对应的误差变量
current["flux_err"].append(mag_flux)
current["std_flux"] = np.std(current["flux_err"])
current["ap_err"].append(mag_flux)
current["std_ap"] = np.std(current["ap_err"])

后续需要获取第n组的统计结果时，直接通过test_groups[n]["median_flux"]、test_groups[n]["std_ap"]取值即可，和原有逐个定义变量的访问效果完全一致。

额外效率优化建议 #

现有逻辑每追加一个数据点就全量遍历数组计算中位数、标准差，单组数据量较大时会产生大量冗余计算：

• 标准差可以通过增量维护数据总和、平方和、计数的方式计算，无需每次遍历全量数组
• 如果不需要每轮循环都输出中位数结果，可以等单组所有数据收集完成后再一次性计算中位数，能大幅减少重复计算开销

• 核心逻辑从近百行分支判断压缩到10行以内，可读性大幅提升
• 后续扩展处理对象数量仅需修改MAX_TARGET_NUM常量，无需新增任何分支代码
• 省去了多分支条件判断的开销，执行效率高于长串if-elif写法
• 从结构上避免了分支遗漏、变量名拼写错误这类低级bug

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