2D滚动窗口分位数的最快方法是什么？

Question

我希望能够计算一个大型的二维矩阵的滚动分位数，其维度为(1e6, 1e5)，按列进行计算。由于需要执行此操作数千次，并且计算成本非常高，因此我正在寻找最快的方式。在实验中使用了窗口大小为1000和q值为0.1。

import numpy as np
import pandas as pd
import multiprocessing as mp
from functools import partial
import numba as nb
X = np.random.random((10000,1000)) # Original array has dimensions of about (1e6, 1e5)

我的当前方法：

Pandas：%timeit：每次循环5.8秒±15.5毫秒

def pd_rolling_quantile(X, window, q):
    return pd.DataFrame(X).rolling(window).quantile(quantile=q)

Numpy Strided: %timeit: 2min 42s ± 3.29 s per loop 的翻译结果为：Numpy跨步：%timeit：每次循环2分42秒±3.29秒。

def strided_app(a, L, S):
    nrows = ((a.size-L)//S)+1
    n = a.strides[0]
    return np.lib.stride_tricks.as_strided(a, shape=(nrows,L), strides=(S*n,n))
def np_1d(x, window, q):
    return np.pad(np.percentile(strided_app(x, window, 1), q*100, axis=-1), (window-1, 0) , mode='constant')
def np_rolling_quantile(X, window, q):
    results = []
    for i in np.arange(X.shape[1]):
        results.append(np_1d(X[:,i], window, q))
    return np.column_stack(results)

多进程：%timeit:每次循环1.13秒±27.6毫秒

def mp_rolling_quantile(X, window, q):
    pool = mp.Pool(processes=12)
    results = pool.map(partial(pd_rolling_quantile, window=window, q=q), [X[:,i] for i in np.arange(X.shape[1])])
    pool.close()
    pool.join()
    return np.column_stack(results)

Numba: %timeit: 每次循环2分28秒±182毫秒

@nb.njit
def nb_1d(x, window, q):
    out = np.zeros(x.shape[0])
    for i in np.arange(x.shape[0]-window+1)+window:
        out[i-1] = np.quantile(x[i-window:i], q=q)
    return out
def nb_rolling_quantile(X, window, q):
    results = []
    for i in np.arange(X.shape[1]):
        results.append(nb_1d(X[:,i], window, q))
    return np.column_stack(results)

时间不太理想，理想情况下我希望能将速度提高10-50倍。如果您有任何建议如何加速它，我将不胜感激。也许有人有使用更低级别语言（Cython）或其他基于Numpy/Numba/Tensorflow的方法来加速的想法。谢谢！

- Franc Weser

Answer 1

我建议使用新的rolling-quantilespackage。为了说明，即使是构建每个列的单独过滤器的相对简单的方法也优于上述单线程的pandas实验。

pipes = [rq.Pipeline(rq.LowPass(window=1000, quantile=0.1)) for i in range(1000)]
%timeit [pipe.feed(X[:, i]) for i, pipe in enumerate(pipes)]
1.34 s ± 7.76 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)

对抗

df = pd.DataFrame(X)
%timeit df.rolling(1000).quantile(0.1)
5.63 s ± 27 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)

正如您所展示的那样，这两种方式都可以通过multiprocessing轻松地并行化。

- Myrl Marmarelis