如何追踪Pandas面向对象代码库的数据依赖，获取方法所需DataFrame列？

你目前的删列捕获KeyError的方案存在三个明显缺陷：

• 性能差：N个列需要把所有方法重复运行N次，列数多的情况下耗时会成倍增长
• 易漏依赖：如果方法存在分支逻辑，测试用例没走到的分支即使缺少列也不会抛错，导致依赖统计不全
• 副作用风险：如果方法包含修改数据、写外部存储等副作用，重复运行多次会产生非预期结果

下面提供三种更优雅的实现思路，可根据你的代码特性选择：

方案1：AST静态代码解析（无需运行代码） #

直接扫描类方法的源代码，通过抽象语法树提取列访问逻辑和方法调用关系，递归合并得到每个方法的全量依赖，适合列名均为静态硬编码的场景。

import ast
import inspect
from collections import defaultdict

class DFColumnDependencyAnalyzer:
    def __init__(self, target_class):
        self.target_class = target_class
        # 存储每个方法直接依赖的列、直接调用的其他方法
        self.raw_deps = defaultdict(lambda: {"cols": set(), "calls": set()})
        self.final_deps = defaultdict(set)
    
    def scan_single_method(self, method_name):
        method = getattr(self.target_class, method_name)
        src = inspect.getsource(method)
        tree = ast.parse(src)
        for node in ast.walk(tree):
            # 匹配self.df['列名']形式的列访问
            if isinstance(node, ast.Subscript) and isinstance(node.value, ast.Attribute):
                if (node.value.attr == 'df' 
                    and isinstance(node.value.value, ast.Name) 
                    and node.value.value.id == 'self'
                    and isinstance(node.slice, ast.Constant)):
                    self.raw_deps[method_name]["cols"].add(node.slice.value)
            # 匹配self.方法名()形式的内部方法调用
            if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute):
                if (isinstance(node.func.value, ast.Name) 
                    and node.func.value.id == 'self'):
                    self.raw_deps[method_name]["calls"].add(node.func.attr)
    
    def resolve_full_dependency(self, method_name, visited=None):
        if visited is None:
            visited = set()
        # 避免循环调用导致的死递归
        if method_name in visited:
            return set()
        visited.add(method_name)
        if method_name in self.final_deps:
            return self.final_deps[method_name]
        # 合并自身直接依赖 + 所有调用方法的依赖
        total_cols = set(self.raw_deps[method_name]["cols"])
        for called_method in self.raw_deps[method_name]["calls"]:
            total_cols.update(self.resolve_full_dependency(called_method, visited.copy()))
        self.final_deps[method_name] = total_cols
        return total_cols

# 使用示例
analyzer = DFColumnDependencyAnalyzer(MyObject)
# 第一步：扫描所有目标方法的原始依赖
for method_name in MyObject.calculations:
    analyzer.scan_single_method(method_name)
# 第二步：递归解析全量依赖
for method_name in MyObject.calculations:
    analyzer.resolve_full_dependency(method_name)
# 转成目标输出格式
result = {k: list(v) for k,v in analyzer.final_deps.items()}

优势：运行速度极快，不会触发业务方法的副作用，只要列访问写法规范准确率可达100%。
注意：如果你的列名是动态拼接生成的（比如self.df[f"{prefix}_value"]），该方案无法识别动态列名，需选择下方的动态方案。

方案2：DataFrame代理埋点（运行时采集访问日志） #

给Pandas DataFrame写一个代理类，重写列访问方法记录所有被访问的列名，每个方法只需运行一次即可拿到全量依赖，支持动态列名、任意嵌套调用的场景。

import pandas as pd
from collections import defaultdict

class TraceableDF(pd.DataFrame):
    _metadata = ["accessed_cols"]
    def __init__(self, *args, **kwargs):
        self.accessed_cols = set()
        super().__init__(*args, **kwargs)
    
    def __getitem__(self, key):
        # 记录所有被访问的列名
        if isinstance(key, str):
            self.accessed_cols.add(key)
        elif isinstance(key, (list, pd.Index)):
            for k in key:
                if isinstance(k, str):
                    self.accessed_cols.add(k)
        return super().__getitem__(key)

# 采集逻辑
dependency_dict = {}
# 准备包含所有列的测试df，尽量覆盖所有方法分支逻辑
full_df = pd.read_csv("your_data.csv")
for method_name in MyObject.calculations:
    trace_df = TraceableDF(full_df.copy())
    obj = MyObject(trace_df)
    # 运行目标方法
    getattr(obj, method_name)()
    # 保存该方法访问过的所有列
    dependency_dict[method_name] = list(trace_df.accessed_cols)

优势：完全不需要修改原有业务代码，支持动态列名，所有方法只需运行一次，性能远高于删列抛错的方案。
注意：测试用的df需要尽量覆盖所有方法的分支逻辑，否则未走到的分支的依赖列会漏统计。

方案3：静态+动态混合方案 #

如果你的代码同时存在静态硬编码列和动态生成列，且分支逻辑较多，可以结合两种方案：先用AST解析拿到所有静态依赖，再用代理DF运行一次补全动态列和分支漏的依赖，准确率最高。

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