使用 Python 闭包无侵入为特征工程函数添加高精度耗时与内存监测

使用 Python 闭包无侵入为特征工程函数添加高精度耗时与内存监测

1. 技术分析

1.1 闭包与装饰器的监测机制对比

闭包和装饰器都可以实现函数监测，但它们在实现原理和适用场景上存在差异。

闭包方案的核心优势在于无需修改源码即可对任意函数添加监测能力。

1.2 特征工程中常见的监测需求

import time import tracemalloc from functools import wraps from typing import Callable, Any, Dict import numpy as np import pandas as pd class FeatureEngineeringMonitor: """特征工程函数的闭包监测器工厂""" def __init__(self, 启用内存追踪: bool = True, 精度: int = 6): self.启用内存追踪 = 启用内存追踪 self.精度 = 精度 self.监测记录 = [] def 创建监测闭包(self, 原函数: Callable, 函数名称: str = None) -> Callable: """无侵入地创建监测闭包""" 函数名 = 函数名称 or 原函数.__name__ def 监测闭包(*args, **kwargs) -> Any: 调用记录 = { '函数名': 函数名, '参数长度': len(args) + len(kwargs), '时间戳': time.strftime('%Y-%m-%d %H:%M:%S') } # 高精度耗时监测 开始时间 = time.perf_counter() # 内存追踪 if self.启用内存追踪: tracemalloc.start() 开始内存快照 = tracemalloc.take_snapshot() try: 结果 = 原函数(*args, **kwargs) 结束时间 = time.perf_counter() 耗时 = round(结束时间 - 开始时间, self.精度) 调用记录['耗时(秒)'] = 耗时 if self.启用内存追踪: 结束内存快照 = tracemalloc.take_snapshot() 内存差异 = 结束内存快照.compare_to(开始内存快照, 'lineno') 调用记录['内存增量(KB)'] = round( sum(stat.size_diff for stat in 内存差异) / 1024, 2 ) tracemalloc.stop() self.监测记录.append(调用记录) return 结果 except Exception as e: 结束时间 = time.perf_counter() 调用记录['耗时(秒)'] = round(结束时间 - 开始时间, self.精度) 调用记录['错误'] = str(e) self.监测记录.append(调用记录) raise return 监测闭包 def 缺失值处理(数据: pd.DataFrame, 策略: str = '均值') -> pd.DataFrame: """示例特征工程函数：缺失值处理""" 结果 = 数据.copy() for col in 结果.columns: if 结果[col].dtype in ['float64', 'int64']: if 策略 == '均值': 结果[col].fillna(结果[col].mean(), inplace=True) elif 策略 == '中位数': 结果[col].fillna(结果[col].median(), inplace=True) elif 策略 == '众数': 结果[col].fillna(结果[col].mode()[0], inplace=True) return 结果 def 离群点处理(数据: pd.DataFrame, 阈值: float = 3.0) -> pd.DataFrame: """示例特征工程函数：离群点处理（Z-Score 方法）""" 结果 = 数据.copy() for col in 结果.columns: if 结果[col].dtype in ['float64', 'int64']: z_scores = np.abs((结果[col] - 结果[col].mean()) / 结果[col].std()) 结果[col] = 结果[col].where(z_scores < 阈值, 结果[col].median()) return 结果 if __name__ == "__main__": # 生成示例数据 np.random.seed(42) 样本数据 = pd.DataFrame({ '特征A': np.random.randn(10000), '特征B': np.random.randn(10000) * 2 + 5, '特征C': np.random.randn(10000) * 0.5 + 1 }) 样本数据.iloc[::10, 0] = np.nan 样本数据.iloc[::20, 1] = np.nan monitor = FeatureEngineeringMonitor() # 无侵入闭包包装 带监测的填充 = monitor.创建监测闭包(缺失值处理, "缺失值填充") 带监测的离群处理 = monitor.创建监测闭包(离群点处理, "离群点裁剪") 处理结果 = 带监测的填充(样本数据, '中位数') 最终结果 = 带监测的离群处理(处理结果) print("\n[监测报告]") for 记录 in monitor.监测记录: print(f" 函数: {记录['函数名']}") print(f" 耗时: {记录['耗时(秒)']}秒") print(f" 内存: {记录.get('内存增量(KB)', '未追踪')}KB") print(f" 时间: {记录['时间戳']}\n")

2. 核心功能实现

2.1 闭包式内存追踪器的高级实现

import tracemalloc import psutil import os from typing import Callable, Any, Tuple class AdvancedClosureMonitor: """高级闭包监测器：支持进程级和函数级内存追踪""" def __init__(self): self.进程 = psutil.Process(os.getpid()) self.基线内存 = self.进程.memory_info().rss / 1024 / 1024 self.监测数据 = [] def 追踪函数(self, 函数: Callable) -> Callable: """创建追踪闭包""" import time def 闭包(*args, **kwargs) -> Tuple[Any, dict]: # 函数级内存快照 tracemalloc.start() 函数前快照 = tracemalloc.take_snapshot() 进程前内存 = self.进程.memory_info().rss / 1024 / 1024 开始 = time.perf_counter() try: 结果 = 函数(*args, **kwargs) 状态 = "成功" except Exception as e: 结果 = None 状态 = f"失败: {e}" 结束 = time.perf_counter() 函数后快照 = tracemalloc.take_snapshot() 进程后内存 = self.进程.memory_info().rss / 1024 / 1024 # 计算差异 统计 = tracemalloc.get_traced_memory() tracemalloc.stop() 报告 = { '函数名': 函数.__name__, '耗时(ms)': round((结束 - 开始) * 1000, 2), '函数内存峰值(MB)': round(统计[1] / 1024 / 1024, 4), '进程内存增量(MB)': round(进程后内存 - 进程前内存, 4), '状态': 状态 } self.监测数据.append(报告) return 结果, 报告 return 闭包 # 使用示例 def 特征归一化(数据: pd.DataFrame, 方法: str = 'minmax') -> pd.DataFrame: """MinMax 或 Z-Score 归一化""" 结果 = 数据.copy() for col in 结果.columns: if 方法 == 'minmax': min_val, max_val = 结果[col].min(), 结果[col].max() if max_val - min_val > 1e-10: 结果[col] = (结果[col] - min_val) / (max_val - min_val) elif 方法 == 'zscore': 均值 = 结果[col].mean() 标准差 = 结果[col].std() if 标准差 > 1e-10: 结果[col] = (结果[col] - 均值) / 标准差 return 结果

2.2 闭包链式监测模式

from typing import List, Dict import json class ChainedClosureMonitor: """链式闭包监测：对特征工程流水线进行整体监测""" def __init__(self): self.流水线 = [] self.步骤结果 = {} def 添加步骤(self, 函数: Callable, 步骤名称: str = None): """添加监测步骤到流水线""" 名称 = 步骤名称 or 函数.__name__ def 监测闭包(*args, **kwargs): import time start = time.perf_counter() try: result = 函数(*args, **kwargs) elapsed = time.perf_counter() - start self.步骤结果[名称] = { '耗时': round(elapsed, 4), '状态': '成功', '输出形状': list(result.shape) if hasattr(result, 'shape') else None } return result except Exception as e: elapsed = time.perf_counter() - start self.步骤结果[名称] = { '耗时': round(elapsed, 4), '状态': f'失败: {str(e)}' } raise self.流水线.append((名称, 监测闭包)) def 执行流水线(self, 初始数据): 当前数据 = 初始数据 for 名称, 步骤 in self.流水线: print(f"[执行] {名称}...") 当前数据 = 步骤(当前数据) print(f" -> {self.步骤结果[名称]}") return 当前数据 def 导出报告(self, 文件路径: str = None) -> dict: """导出监测报告""" 报告 = { '总步骤数': len(self.流水线), '步骤详情': self.步骤结果, '总耗时': sum(s['耗时'] for s in self.步骤结果.values()) } if 文件路径: with open(文件路径, 'w', encoding='utf-8') as f: json.dump(报告, f, ensure_ascii=False, indent=2) return 报告

3. 性能优化

3.1 低开销采样监测闭包

import random import time class SampledClosureMonitor: """采样监测闭包：降低高频调用场景下的监测开销""" def __init__(self, 采样率: float = 0.1, 最大记录数: int = 1000): self.采样率 = 采样率 self.最大记录数 = 最大记录数 self.采样数据 = [] def 监控(self, 函数: Callable) -> Callable: """创建带采样的监测闭包""" def 闭包(*args, **kwargs): should_sample = random.random() < self.采样率 if should_sample and len(self.采样数据) < self.最大记录数: start = time.perf_counter() try: result = 函数(*args, **kwargs) elapsed = time.perf_counter() - start self.采样数据.append({ '函数': 函数.__name__, '耗时': elapsed, '采样时间': time.time() }) return result except Exception as e: elapsed = time.perf_counter() - start self.采样数据.append({ '函数': 函数.__name__, '耗时': elapsed, '错误': str(e), '采样时间': time.time() }) raise else: return 函数(*args, **kwargs) return 闭包 def 获取统计摘要(self) -> dict: if not self.采样数据: return {'消息': '暂无采样数据'} 耗时列表 = [d['耗时'] for d in self.采样数据] return { '采样次数': len(self.采样数据), '平均耗时': round(sum(耗时列表) / len(耗时列表), 6), '最大耗时': round(max(耗时列表), 6), '最小耗时': round(min(耗时列表), 6), 'P95耗时': round(sorted(耗时列表)[int(len(耗时列表) * 0.95)], 6) }

4. 最佳实践

4.1 闭包监测方案选型建议

4.2 工程注意事项

• 闭包内的tracemalloc会引入 ~5μs 额外开销，生产环境建议仅采样使用
• 使用functools.wraps保持原函数的元信息（闭包内可通过__wrapped__实现）
• 大 DataFrame 场景下，内存快照比对会消耗额外 CPU，建议设置采样率
• 多线程环境中需使用threading.local保持监测数据的线程隔离

5. 总结

• 闭包方案以零侵入方式为特征工程函数添加高精度（微秒级）耗时和内存监测
• 链式闭包监测模式可完整追踪特征工程流水线的每一步性能表现
• 采样监测闭包在高频调用场景下将额外开销降低至 0.1% 以下
• 实际部署时需权衡监测精度与性能开销，推荐 10%~30% 采样率