滑动窗口处理DataFrame二进制值转换的代码问题求助

需求描述 #

给定一个仅包含0和1的DataFrame（数千行和列），需以非重叠固定窗口（窗口大小为4）遍历数据列，每个窗口内遵循以下规则处理：

• 将窗口内所有1转换为0
• 将1的前后相邻位置（单个或连续多个1的前后）转换为1
• 若1位于窗口首尾，仅将自身转为0（因首尾无外侧位置，无需处理外侧）

处理示例（id=1的行） #

window 1 : 0 ,1 ,1 ,1 → 1, 0,0,0
window 2 : 0 ,0 ,1 ,1 → 0, 1,0,0
window 3 : 1 ,0 ,0 ,0 → 0, 1,0,0

测试DataFrame #

import pandas as pd

data = {
    'Ids': {0: 4, 1: 5, 2: 1},
    'val1': {0: 0, 1: 0, 2: 0},
    'val2': {0: 1, 1: 0, 2: 1},
    'val3': {0: 1, 1: 1, 2: 1},
    'val4': {0: 1, 1: 1, 2: 1},
    'val5': {0: 0, 1: 1, 2: 0},
    'val6': {0: 0, 1: 0, 2: 1},
    'val7': {0: 1, 1: 0, 2: 1},
    'val8': {0: 1, 1: 1, 2: 1},
    'val9': {0: 1, 1: 1, 2: 0},
    'val10': {0: 1, 1: 0, 2: 0},
    'val11': {0: 0, 1: 1, 2: 1},
    'val12': {0: 0, 1: 0, 2: 1},
}
df = pd.DataFrame(data)

期望输出 #

val1    val2    val3    val4    val5    val6 val7 val8  val9    val10   val11   val12
1        0      0       0        0      1     0     0   0   0   1   0
0        1      0       0        0      1     1     0   0   1   0   1
1        0      0       0        1      0     0     0   0   1   0   0

现有代码 #

def pre_snpback(df, window_size):
    output_df = pd.DataFrame()

    for i in range(0, df.shape[1]-window_size, window_size ):
        window = df.iloc[:, i + 1:i + 1 + window_size]
        
        print(window)
        
        x = window.shift(1, axis=1).fillna(0).astype(int)
        
        y = window.shift(-1, axis=1).fillna(0).astype(int)

        df_w = (x | y) * ~window.astype(bool)
        
        df_w.reset_index(drop=True, inplace=True)

        output_df = pd.concat([output_df,df_w], axis=1)
        
    return output_df

new_df = pre_snpback(df, 4)

问题及测试用例 #

上述代码应用于大型数据集时结果不符合预期，提供测试用例（窗口大小32）：

data = {f'{i}': 1 if 406 <= i <= 413 else 0 for i in range(384, 417)}
df = pd.DataFrame([data])
pre_snpback(df, 32)

代码问题分析 #

1. 列范围遗漏：原循环终止条件df.shape[1]-window_size会丢弃最后一组不足窗口大小的列。例如原测试DataFrame有12个数据列，窗口大小为4时，仅处理前8列，剩余4列完全被忽略，导致输出列数不足。
2. 效率低下：多次调用pd.concat拼接结果，对于数千列的数据集会产生严重的性能损耗。
3. 场景兼容性差：未考虑无Ids列的测试场景，且列名可能在拼接过程中丢失。

修正后的代码 #

import pandas as pd

def pre_snpback(df, window_size, has_ids=True):
    # 提取数据列，根据是否包含Ids列调整
    if has_ids:
        data_cols = df.columns[1:]
        ids_col = df[['Ids']]
    else:
        data_cols = df.columns
        ids_col = pd.DataFrame()
    
    output_dfs = []
    
    # 遍历所有非重叠窗口，包含最后一组不足窗口大小的列
    for start in range(0, len(data_cols), window_size):
        end = start + window_size
        window_cols = data_cols[start:end]
        window = df[window_cols]
        
        # 计算当前列的前后相邻列的1位置
        x = window.shift(1, axis=1).fillna(0).astype(int)
        y = window.shift(-1, axis=1).fillna(0).astype(int)
        
        # 应用规则：1的前后位置设为1，原1的位置设为0
        df_w = (x | y) * (~window.astype(bool)).astype(int)
        output_dfs.append(df_w)
    
    # 一次性拼接所有窗口结果
    output_df = pd.concat(output_dfs, axis=1)
    # 合并Ids列（如果存在）
    if not ids_col.empty:
        output_df = pd.concat([ids_col, output_df], axis=1)
    
    return output_df

# 测试原DataFrame
data = {
    'Ids': {0: 4, 1: 5, 2: 1},
    'val1': {0: 0, 1: 0, 2: 0},
    'val2': {0: 1, 1: 0, 2: 1},
    'val3': {0: 1, 1: 1, 2: 1},
    'val4': {0: 1, 1: 1, 2: 1},
    'val5': {0: 0, 1: 1, 2: 0},
    'val6': {0: 0, 1: 0, 2: 1},
    'val7': {0: 1, 1: 0, 2: 1},
    'val8': {0: 1, 1: 1, 2: 1},
    'val9': {0: 1, 1: 1, 2: 0},
    'val10': {0: 1, 1: 0, 2: 0},
    'val11': {0: 0, 1: 1, 2: 1},
    'val12': {0: 0, 1: 0, 2: 1},
}
df = pd.DataFrame(data)
new_df = pre_snpback(df, 4)
print(new_df.drop('Ids', axis=1))

# 测试窗口大小32的用例
data_test = {f'{i}': 1 if 406 <= i <= 413 else 0 for i in range(384, 417)}
df_test = pd.DataFrame([data_test])
result_test = pre_snpback(df_test, 32, has_ids=False)
print(result_test)

修正说明 #

1. 修复列范围问题：循环改为range(0, len(data_cols), window_size)，确保所有数据列（包括最后一组不足窗口大小的列）都被处理。
2. 提升性能：用列表收集每个窗口的处理结果，最后一次性拼接，避免多次pd.concat的性能损耗。
3. 增强兼容性：新增has_ids参数，兼容有无Ids列的场景，同时保留原列名。
4. 逻辑一致性：保留原窗口内的处理逻辑，确保与示例规则匹配。

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