如何在Python/Pandas中将时间序列/向量下采样至固定长度

要实现将任意Pandas Series（普通向量或时间序列）下采样到指定长度，核心是分组聚合——把原序列划分为对应目标长度的若干块，对每块应用聚合逻辑（如均值、求和、极值等）。下面提供一个兼容普通向量和时间索引序列的通用函数，结合Pandas的rolling、分块滑动窗口及resample方法实现需求。

函数实现 #

import pandas as pd
import numpy as np

def downsample_series(series, target_length, agg_func='mean'):
    """
    将Pandas Series下采样到指定长度
    
    参数:
        series: pd.Series - 输入的时间序列/向量
        target_length: int - 目标输出序列长度
        agg_func: str/callable - 聚合函数，如'mean', 'sum', 'max', 'min'或自定义函数
    
    返回:
        pd.Series - 下采样后的序列
    """
    n = len(series)
    if target_length >= n:
        return series.copy()
    
    # 情况1：带时间索引的时间序列，优先用resample
    if isinstance(series.index, pd.DatetimeIndex):
        total_duration = series.index[-1] - series.index[0]
        target_freq = total_duration / target_length
        freq_str = f"{target_freq.total_seconds()}S"
        downsampled = series.resample(freq_str).agg(agg_func)
        # 修正边界误差，确保长度严格匹配目标值
        if len(downsampled) > target_length:
            downsampled = downsampled.iloc[:target_length]
        elif len(downsampled) < target_length:
            last_val = series.iloc[-1] if callable(agg_func) else agg_func([series.iloc[-1]])
            downsampled = pd.concat([downsampled, pd.Series([last_val], index=[series.index[-1]])])
        return downsampled
    
    # 情况2：普通向量（无时间索引），用分块聚合
    window_indices = np.linspace(0, n, target_length + 1, dtype=int)
    downsampled_vals = []
    for i in range(target_length):
        start = window_indices[i]
        end = window_indices[i+1]
        window = series.iloc[start:end]
        downsampled_vals.append(window.agg(agg_func))
    return pd.Series(downsampled_vals, index=series.index[window_indices[:-1]])

测试示例 #

针对题目给出的三个Series，下采样到长度10：

# 定义输入序列
t1 = pd.Series([1,2,3,4,5,6,7,8,6,5,4,3,2,2])
t2 = pd.Series([2,3,4,7,8,9,11,12,13,14,15,16,17,18,19,20])
ts = pd.Series([10,14,18,19,20,22,23,25,27,34,37,41,51,67,69,73,75,82,88,89,90,100,110])

# 下采样到长度10，使用均值聚合
t1_down = downsample_series(t1, target_length=10)
t2_down = downsample_series(t2, target_length=10)
ts_down = downsample_series(ts, target_length=10)

# 输出结果
print("t1下采样结果（长度10）：")
print(t1_down)
print("\nt2下采样结果（长度10）：")
print(t2_down)
print("\nts下采样结果（长度10）：")
print(ts_down)

关键细节说明 #

• 自定义聚合逻辑：可传入自定义函数，比如agg_func=lambda x: x.median()实现中位数聚合
• 时间序列适配：带DatetimeIndex的序列会自动用resample按时间间隔划分窗口，更贴合时间维度的下采样需求
• 边界处理：针对原序列长度与目标长度非整数倍的情况，通过np.linspace划分窗口，确保最后一个窗口包含剩余所有元素
• 滑动窗口替代方案：若需要滑动式下采样而非固定分块，可使用以下逻辑：

  step_size = int(np.floor(len(series)/target_length))
  downsampled = series.rolling(window=step_size).agg(agg_func)[::step_size][:target_length]
  

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