基于DENA公式的Pandas 10分钟转15分钟数据重采样实现方案

首先，你的核心需求是基于DENA-Netzstudie II（2010）第109页的加权公式，将10分钟间隔的时间序列数据转换为15分钟平均值。现有实现已经能完成基础功能，但在性能、代码简洁性和健壮性上还有不少优化空间，下面是具体的优化建议和重构后的代码：

现有实现的可优化点 #

• 手动遍历索引寻找起始时间的方式冗余，可直接利用Pandas的时间序列工具简化
• cal_ten_to_fifteen_min采用逐元素循环，大数据量下性能较差，可改用向量化计算提升效率
• 递归处理多列的方式效率低，且frame.drop(column, axis=1)未重新赋值，实际未生效
• 裸except会捕获所有异常，不利于调试，应针对性处理索引越界问题
• 数据截断逻辑不够严谨，容易出现索引长度不匹配的情况

重构后的实现方案 #

1. 向量化实现DENA加权计算 #

用numpy向量化操作替代Python循环，大幅提升性能：

import pandas as pd
import numpy as np

def dena_10to15_weighted(arr: np.ndarray) -> np.ndarray:
    """
    基于DENA Netzstudie II (2010) S.109的公式，将10分钟值数组转换为15分钟值数组
    :param arr: 10分钟平均值的numpy数组
    :return: 15分钟平均值的numpy数组
    """
    # 计算15分钟数据的总长度：每6个10分钟对应4个15分钟
    full_cycles = len(arr) // 6
    if full_cycles == 0:
        raise ValueError("输入数据至少需要包含6个10分钟周期（1小时）")
    
    # 截断到完整周期，避免边界计算出错
    arr = arr[:full_cycles * 6]
    n_15min = full_cycles * 4
    
    # 生成15分钟数据的索引n（从1开始）
    n = np.arange(1, n_15min + 1)
    # 计算对应的10分钟数据索引m
    m = ((3 * n + (n % 2)) // 2) - 1
    # 计算权重g
    g_m = np.abs((2 * m % 3) - 1) + 1
    g_m1 = np.abs((2 * (m + 1) % 3) - 1) + 1
    
    # 计算加权平均值，适配0-based数组索引
    try:
        fifteen_min_vals = (arr[m - 1] * g_m + arr[m] * g_m1) / 3
    except IndexError as e:
        raise ValueError("输入数据长度不符合DENA公式的周期要求") from e
    
    return fifteen_min_vals

2. 优化DataFrame转换函数 #

用更简洁的方式处理索引和多列转换，避免递归和不必要的拷贝：

def resample_10to15_dena(df: pd.DataFrame, columns) -> pd.DataFrame:
    """
    将10分钟间隔的DataFrame转换为15分钟间隔，应用DENA加权公式
    :param df: 输入的10分钟平均值DataFrame（索引为DatetimeIndex）
    :param columns: 需要转换的列名（字符串或列表）
    :return: 转换后的15分钟平均值DataFrame
    """
    # 校验输入索引类型
    if not isinstance(df.index, pd.DatetimeIndex):
        raise TypeError("输入DataFrame的索引必须是DatetimeIndex")
    
    # 找到第一个完整小时的起始点（分钟为0）
    start_idx = df.index[df.index.minute == 0][0]
    filtered_df = df.loc[start_idx:]
    
    # 生成目标15分钟索引，对齐到完整周期
    end_idx = filtered_df.index[-1].floor('15T')
    target_index = pd.date_range(start_idx, end_idx, freq='15T')
    
    # 统一列参数为列表格式
    if isinstance(columns, str):
        columns = [columns]
    
    # 批量处理需要转换的列
    result_dict = {}
    for col in columns:
        arr = filtered_df[col].values
        result_dict[col] = dena_10to15_weighted(arr)
    
    # 构建结果DataFrame
    result_df = pd.DataFrame(result_dict, index=target_index[:len(next(iter(result_dict.values())))])
    
    # 保留并对齐不需要转换的列（用前向填充适配时间序列）
    non_transform_cols = [col for col in df.columns if col not in columns]
    if non_transform_cols:
        non_transform_df = df[non_transform_cols].reindex(target_index, method='ffill')
        result_df = pd.concat([result_df, non_transform_df], axis=1)
    
    return result_df

3. 测试用例验证 #

用你提供的测试数据验证转换效果：

# 生成测试数据
index = pd.date_range('2020-01-01 00:00', '2020-01-01 02:00', freq='10T')
values = list(range(1, 14))  # 12个10分钟数据（2小时），对应8个15分钟数据
column_name = '10_min_avg'
df = pd.DataFrame(values, index, [column_name])

# 执行转换
avg_15_min = resample_10to15_dena(df, column_name)
print(avg_15_min)

关键优化说明 #

1. 向量化计算：用numpy数组替代Python列表循环，大数据量下性能提升数倍
2. 严谨的边界处理：主动截断非完整周期数据，明确抛出异常便于调试
3. 简洁的索引生成：利用Pandas时间戳方法直接生成目标索引，无需手动遍历
4. 批量列处理：避免递归调用，直接循环处理多列，代码更易读且效率更高
5. 非转换列对齐：对不需要转换的列采用reindex+ffill的方式对齐到15分钟索引，保证数据完整性

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