如何用Pandas的rolling/apply实现一阶动态过程计算（替代循环）

核心问题分析 #

你的需求是实现依赖前序计算结果的递推逻辑：b_i = f(c_{i-1})，c_i = a_i + b_i，且要支持分组（groupby）操作。这类递推无法直接用普通rolling窗口实现——因为rolling是固定长度窗口，无法传递跨窗口的状态（比如上一步的c值），因此需要结合累积窗口（expanding）或分组内独立处理的思路。

可行实现方案 #

1. 单数据集（无分组）的实现

可以通过类维护递推状态，配合expanding.apply完成累积计算：

• 用类保存上一步的c值，每次调用计算函数时更新状态；
• expanding.apply会逐步传入累积到当前行的所有数据，我们只需要取当前行的a_i即可。

示例代码：

import pandas as pd

# 示例递推函数f
def f(x):
    return x * 0.5

class RecursiveCalc:
    def __init__(self, initial_b, initial_c):
        self.prev_c = initial_c  # 保存上一步的c值
    
    def compute(self, window):
        # 取当前行的a_i（窗口最后一个元素）
        a_i = window[-1]
        b_i = f(self.prev_c)
        c_i = a_i + b_i
        # 更新状态
        self.prev_c = c_i
        return c_i

# 初始化数据与初始条件
df = pd.DataFrame({'A': [1.0, 2.0, 3.0, 4.0, 5.0]})
initial_b1 = 0.5
initial_c1 = df.loc[0, 'A'] + initial_b1

# 初始化计算器并生成c列
calc = RecursiveCalc(initial_b1, initial_c1)
df['c'] = df['A'].expanding().apply(calc.compute)
# 反向推导b列（除第一行）
df['b'] = df['c'] - df['A']
df.loc[0, 'b'] = initial_b1

2. 支持groupby的实现

分组场景下，每个组的递推状态需要独立维护，最清晰高效的方式是用groupby.apply对每个组单独执行递推逻辑：

• 对每个分组的子DataFrame，初始化组内的b1和c1；
• 在组内执行小范围循环完成递推（组内数据量通常较小，循环开销可忽略）；
• 返回处理后的组，最终合并所有组结果。

示例代码：

# 带分组的示例数据
df_grouped = pd.DataFrame({
    'group': ['X', 'X', 'Y', 'Y', 'Y'],
    'A': [1.0, 2.0, 3.0, 4.0, 5.0]
})

# 每个分组的初始条件（可根据需求自定义）
group_initial = {
    'X': {'b1': 0.5, 'c1': 1.5},
    'Y': {'b1': 1.0, 'c1': 4.0}
}

def process_single_group(group):
    group = group.copy()
    group_name = group['group'].iloc[0]
    init_b = group_initial[group_name]['b1']
    init_c = group_initial[group_name]['c1']
    
    # 初始化第一行
    group.loc[group.index[0], 'b'] = init_b
    group.loc[group.index[0], 'c'] = init_c
    
    # 组内递推
    for idx in range(1, len(group)):
        prev_c = group.iloc[idx-1]['c']
        current_b = f(prev_c)
        current_c = group.iloc[idx]['A'] + current_b
        group.loc[group.index[idx], 'b'] = current_b
        group.loc[group.index[idx], 'c'] = current_c
    return group

# 执行分组处理
result = df_grouped.groupby('group', group_keys=False).apply(process_single_group)

关键注意事项 #

• 避免用全局变量维护递推状态，否则分组时会出现跨组状态污染；
• 若f是向量化函数，可进一步优化组内递推的效率；
• 分组内的小循环并非“低效”——相比全局循环，分组后每个组的数据量更小，且Pandas的groupby底层优化会降低整体开销，同时代码可读性更强。

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