
本文介绍两种不依赖 global 关键字的优雅方案,将跨函数调用需持续维护的状态(如日期标记和已用分钟数)转为局部可控结构:一是使用可变字典传参,二是封装为有状态类,兼顾可读性、可测试性与线程安全性。
本文介绍两种不依赖 global 关键字的优雅方案,将跨函数调用需持续维护的状态(如日期标记和已用分钟数)转为局部可控结构:一是使用可变字典传参,二是封装为有状态类,兼顾可读性、可测试性与线程安全性。
在数据处理中,尤其是使用 pandas.DataFrame.apply() 按行计算时,常需维护上下文状态(例如按天累计免费分钟数)。原代码通过 global 变量 prev_date 和 expen_minutes_used 实现状态延续,但存在明显缺陷:全局污染、难以复用、无法并行、单元测试困难。
✅ 推荐方案一:状态字典传参(轻量、直观)
将状态封装为普通字典,作为显式参数传入函数。因字典是可变对象,函数内部修改会直接反映到外部,无需 global 声明:
from math import ceil
# 初始化状态容器(局部作用域内定义)
state = {'prev_date': None, 'expen_minutes_used': 0}
def calculate_call_cost(row, state):
current_date = row['data of call'].date()
# 日期变更 → 重置当日免费额度
if state['prev_date'] is None or current_date != state['prev_date']:
state['prev_date'] = current_date
state['expen_minutes_used'] = 0
duration_min = ceil(row['Duration in second'] / 60)
expen_minutes = 0
# 分配免费分钟(最多5分钟/天)
if state['expen_minutes_used'] < 5:
expen_minutes = min(5 - state['expen_minutes_used'], duration_min)
state['expen_minutes_used'] += expen_minutes
cheap_minutes = duration_min - expen_minutes
return cheap_minutes * 0.40 + expen_minutes * 3.95
# 使用 lambda 绑定状态字典
calls_data_sorted['Cost of tariff'] = calls_data_sorted.apply(
lambda row: calculate_call_cost(row, state), axis=1
)
total_cost2 = calls_data_sorted['Cost of tariff'].sum()⚠️ 注意:此方式要求 apply 按严格顺序执行(默认满足),且不可用于多线程/并发场景(字典非线程安全)。
✅ 推荐方案二:状态类封装(健壮、可扩展)
面向对象方式更清晰地表达“一个计费上下文”的语义,天然支持实例隔离、方法复用与后续扩展(如添加日志、校验或重置逻辑):
from math import ceil
class TelecomCostCalculator:
def __init__(self):
self.prev_date = None
self.expen_minutes_used = 0
def calculate_call_cost(self, row):
current_date = row['data of call'].date()
if self.prev_date is None or current_date != self.prev_date:
self.prev_date = current_date
self.expen_minutes_used = 0
duration_min = ceil(row['Duration in second'] / 60)
expen_minutes = min(5 - self.expen_minutes_used, duration_min) \
if self.expen_minutes_used < 5 else 0
self.expen_minutes_used += expen_minutes
cheap_minutes = duration_min - expen_minutes
return cheap_minutes * 0.40 + expen_minutes * 3.95
# 实例化并应用
calculator = TelecomCostCalculator()
calls_data_sorted['Cost of tariff'] = calls_data_sorted.apply(
lambda row: calculator.calculate_call_cost(row), axis=1
)
total_cost2 = calls_data_sorted['Cost of tariff'].sum()? 关键优势总结:
- ✅ 彻底消除 global,状态生命周期明确可控;
- ✅ 支持多实例并行计算(如不同用户账单独立计费);
- ✅ 函数/方法职责单一,易于单元测试(可随时新建实例并注入测试数据);
- ✅ 未来扩展成本低(例如增加 reset_daily_counter() 方法或支持配置化资费规则)。
选择字典方案适合简单脚本快速迭代;选择类方案则更适合中大型项目,体现良好的工程实践与可维护性。

















