如何在Python代码库中利用命名规范避免重复代码且不使用eval/exec

我在Python中有多个DataFrame：

[df1, df2, df3, df4, ...]

还有大量对应处理它们的函数：

[clean_df1, clean_df2, clean_df3, ...]

[format_df1, format_df2, format_df3, ...]

这导致代码重复度很高，我想写出类似这样的代码：

final_data = {}
for i in [1,2,3,4,5,...]:
    final_data[f'df{i}_final'] = eval(f'df{i}.pipe(clean_df{i}).pipe(format_df{i})')

但用eval总觉得不妥，而且对LSP（语言服务器协议）不友好……

有没有Python特性可以利用这种命名规范来避免重复代码，同时不用eval/exec或代码生成这类方式？

补充说明1 #

我目前的实现是用几个字典：

data_registry = {'df1': df1, 'df2': df2, ...}
cleaning_registry = {'df1': clean_df1, 'df2': clean_df2, ...}
format_registry = {'df1': format_df1, 'df2': format_df2, ...}

但维护这些字典很容易出错（比如忘加条目或者拼写错误）。我核心的需求是自动生成这类字典或类似结构，不用频繁手动编辑。

补充说明2 #

针对定义细节的询问，补充背景：这些是特征工程 pipeline 里的特征——每个特征对应三个函数：

• 创建该特征DataFrame的函数
• 归一化该特征的函数（归一化逻辑是特征专属的）
• 将未归一化特征格式化为人类可读值的函数（同样是特征专属的）

我用字典而不是列表，是因为按索引匹配更容易出错——比如两个列表里的对应函数没对齐位置。

这些函数的大致结构如下：

def normalize_feature_i() -> pl.Expr:
    return pl.max_horizontal(
        pl.col('feature_i').list.get(0).pipe(_default_descending, -0.03),
        pl.min_horizontal(
            pl.col('feature_i').list.get(1).pipe(_default_descending, -0.03),
            pl.col('feature_i').list.get(2).pipe(_default_descending, -0.03)
        )
    ).alias('feature_i')

def fmt_feature_i(col: pl.Expr) -> pl.Expr:
    def _fmt_feature_i(x):
        val = min(x)
        i = '' if x[0]==val else '*' if x[1]==val else '**' if x[2]==val else ''
        return f'{val:.0%}{i}'
    return col.map_elements(_fmt_feature_i)

在代码库的不同部分，有时需要对所有特征应用格式化操作，有时需要应用归一化操作……

我曾考虑过用class Feature()来整合这些逻辑，但感觉很混乱——机器学习代码不该掺和可视化相关的代码。

方案1：利用全局命名空间自动注册 #

遍历当前模块的全局命名空间，根据命名规则自动提取对应的DataFrame和函数，生成注册表：

import inspect

# 获取当前模块的全局命名空间
current_module = inspect.currentframe().f_globals

# 自动生成data_registry：提取所有以'df'开头的变量
data_registry = {name: obj for name, obj in current_module.items() if name.startswith('df')}

# 自动生成cleaning_registry：提取所有以'clean_df'开头的函数
cleaning_registry = {name.replace('clean_', ''): obj for name, obj in current_module.items() if name.startswith('clean_df')}

# 自动生成format_registry：提取所有以'format_df'开头的函数
format_registry = {name.replace('format_', ''): obj for name, obj in current_module.items() if name.startswith('format_df')}

每次新增dfN、clean_dfN或format_dfN时，注册表会自动更新，无需手动维护。

方案2：用装饰器自动注册函数 #

如果函数是自定义的，用装饰器自动将函数注册到对应字典，避免手动添加：

cleaning_registry = {}
format_registry = {}

def register_cleaner(df_name):
    def decorator(func):
        cleaning_registry[df_name] = func
        return func
    return decorator

def register_formatter(df_name):
    def decorator(func):
        format_registry[df_name] = func
        return func
    return decorator

# 使用示例
@register_cleaner('df1')
def clean_df1(df):
    # 清洗逻辑
    return df

@register_formatter('df1')
def format_df1(df):
    # 格式化逻辑
    return df

DataFrame的注册可以结合方案1的自动提取，或者同样用装饰器（如果是函数生成的DataFrame）。

方案3：用数据类整合特征逻辑 #

用轻量的数据类封装单个特征的所有逻辑，既保证关联性，又避免逻辑耦合：

from dataclasses import dataclass
import polars as pl

@dataclass
class Feature:
    name: str
    create_func: callable
    normalize_func: callable
    fmt_func: callable

# 定义单个特征
def create_feature_i():
    # 创建DataFrame逻辑
    return pl.DataFrame({'feature_i': [[0.1, 0.2, 0.3]]})

def normalize_feature_i() -> pl.Expr:
    return pl.max_horizontal(
        pl.col('feature_i').list.get(0).pipe(_default_descending, -0.03),
        pl.min_horizontal(
            pl.col('feature_i').list.get(1).pipe(_default_descending, -0.03),
            pl.col('feature_i').list.get(2).pipe(_default_descending, -0.03)
        )
    ).alias('feature_i')

def fmt_feature_i(col: pl.Expr) -> pl.Expr:
    def _fmt_feature_i(x):
        val = min(x)
        i = '' if x[0]==val else '*' if x[1]==val else '**' if x[2]==val else ''
        return f'{val:.0%}{i}'
    return col.map_elements(_fmt_feature_i)

# 注册所有特征
features = [
    Feature(
        name='df1',
        create_func=create_feature_i,
        normalize_func=normalize_feature_i,
        fmt_func=fmt_feature_i
    ),
    # 其他特征...
]

# 批量处理示例
final_data = {}
for feat in features:
    df = feat.create_func()
    normalized_df = df.with_columns(feat.normalize_func())
    formatted_df = normalized_df.with_columns(feat.fmt_func(pl.col(feat.name)))
    final_data[f'{feat.name}_final'] = formatted_df

这种方式把单个特征的所有逻辑绑定在一起，避免了命名匹配错误，同时结构清晰，机器学习和格式化逻辑仅在数据类中关联，没有深度耦合。

方案4：利用globals()直接匹配（谨慎使用） #

如果命名规则严格一致，直接在循环中通过globals()获取变量和函数，无需eval：

final_data = {}
for i in range(1, 6):  # 假设处理df1到df5
    df_name = f'df{i}'
    clean_func_name = f'clean_{df_name}'
    format_func_name = f'format_{df_name}'
    
    # 从全局命名空间获取对象
    df = globals().get(df_name)
    clean_func = globals().get(clean_func_name)
    format_func = globals().get(format_func_name)
    
    if all([df, clean_func, format_func]):
        final_data[f'{df_name}_final'] = df.pipe(clean_func).pipe(format_func)

这种方式无需维护字典，但依赖严格的命名规范，且如果变量名不存在会返回None，需要加判断避免报错。

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