具有最小周期和最大值的 Numpy 滚动窗口（类似于 Pandas 滚动但 Numpy）

Question

我希望（无法在任何地方看到此内容）使用 numpy 而不是 pandas.rolling（主要是为了速度）创建一个滑动窗口。 但是，滑动窗口也必须是窗口中实例的最小和最大数量的函数，当窗口无法构造时返回NaN。这类似于 pandas.rolling，其参数设置为窗口大小（最大）和 min_periods。例如：

设置 Min_periods = 3 和 Max_periods = 7，请参阅下面的预期窗口示例：

index  values  intended_window
0      10      np.nan
1      11      np.nan
2      12      [10,11,12]
3      13      [10,11,12,13]
4      14      [10,11,12,13,14]
5      15      [10,11,12,13,14,15]
6      16      [10,11,12,13,14,15,16]
7      17      [11,12,13,14,15,16,17]
8      18      [12,12,14,15,16,17,18]
9      19      [13,14,15,16,17,18,19]

我看到了当滑动窗口不需要最大值或最小值时如何构造此滑动窗口的示例，例如

def rolling_window(a, window):
  shp = a.shape[:-1] + (a.shape[-1] - window + 1, window)
  strides = a.strides + (a.strides[-1],)
  return np.lib.stride_tricks.as_strided(a, shape=shp, strides=strides)

有谁知道我可以如何扩展它以返回上例中的窗口？

Answer 1

请尝试以下方法。

def dataframe_striding(dataframe, window):
    '''
    Parameters
    ----------
    dataframe : Input Dataframe, in this case df with columns ['index', 'values'] present.
    window : Tuple denoting the window size.

    Returns
    -------
    dataframe : Pandas Dataframe

    '''
    lower_end, upper_end = window 
    if lower_end > upper_end:
        raise ValueError('Check window size!')

    results = []
    for i in range(dataframe.shape[0]):
        l = [k for k in dataframe['values'][:i+1]]        
        if len(l) < lower_end:                 # checks for minimum window length
            l = np.nan
            results.append(l)
        elif lower_end <= len(l) <= upper_end: # checks for required window length
            results.append(l)
        else:                                  # checks for maximum window length
            l = l[-upper_end:]
            results.append(l)
        
     dataframe['rolling_output'] = results     # appends output to input dataframe
     return dataframe 

# run above function #
final_df = dataframe_striding(df, window = (4,6))

Answer 2

values = np.linspace(1, 10, num=10)
window_column = []
for i in range(len(values)):
    if i - 7 < 0:
        t = 0
    else:
        t = i - 7
    window = values[t:i]
    if len(window) < 3:
        window_column.append(np.nan)
    else:
        window_column.append(window)