问题 计算Vandermonde矩阵的有效方法


我正在计算 Vandermonde matrix 对于一个相当大的1D阵列。这样做的自然而干净的方法就是使用 np.vander()。但是,我发现这是约。 慢2.5倍 而不是基于列表理解的方法。

In [43]: x = np.arange(5000)
In [44]: N = 4

In [45]: %timeit np.vander(x, N, increasing=True)
155 µs ± 205 ns per loop (mean ± std. dev. of 7 runs, 10000 loops each)

# one of the listed approaches from the documentation
In [46]: %timeit np.flip(np.column_stack([x**(N-1-i) for i in range(N)]), axis=1)
65.3 µs ± 235 ns per loop (mean ± std. dev. of 7 runs, 10000 loops each)

In [47]: np.all(np.vander(x, N, increasing=True) == np.flip(np.column_stack([x**(N-1-i) for i in range(N)]), axis=1))
Out[47]: True

我试图了解瓶颈的位置以及原生的实现原因 np.vander() 是〜 2.5倍 比较慢。

效率对我的实施至关重要。所以,也欢迎更快的替代品!


9353
2018-01-14 01:45


起源

“......原生np.vander()的C实现......” 供参考: numpy.vander 是用Python编写的,使用对其他numpy函数的调用: github.com/numpy/numpy/blob/... - Warren Weckesser
以你为榜样 x, 你给 x = np.arange(5000)。你的实际数据是整数吗? - Warren Weckesser
@WarrenWeckesser谢谢,更新了这个问题。是的,我的 x 将会 int32 数据类型! - kmario23
在此处添加摘要。对于整数,我的第一种方法目前是您和我的帖子中所有选项中最快的。对于花车,我的第二种方法是最快的。 - coldspeed
如果您的问题得到了解答,请考虑 投票,接受最有帮助的答案。单击灰色勾选将其切换为绿色。它表明您的问题已经解决,并有助于社区。谢谢! - coldspeed


答案:


这里有一些方法,其中一些方法(在我的计算机上)比目前发布的方法快得多。

我认为最重要的观察是,它在很大程度上取决于你想要多少度数。指数(我认为是小整数指数的特殊情况)只对小指数范围有意义。指数越多,基于乘法的方法就越好。

我想强调一下 multiply.accumulate 基于方法(ma)这类似于numpy的内置方法,但更快(而不是因为我在检查上吝啬 - nnc,numpy-no-checks表明了这一点)。除了最小的指数范围之外,它对我来说实际上是最快的。

由于我不理解的原因,numpy实现做了三件我认识最慢且不必要的事情:(1)它产生了很多基本向量的副本。 (2)它使它们不连续。 (3)我相信强制缓冲就地积累。

我想提到的另一件事是小范围的最快(out_e_1 基本上是手动版本 ma通过简单的预防措施促进更大的dtype,减慢了两倍以上(safe_e_1 可以说有点用词不当)。

调用基于广播的方法 bc_* 哪里 * 表示广播轴(b表示基数,e表示exp)'作弊'表示结果是不连续的。

计时(最好的3):

rep=100 n_b=5000 n_e=4 b_tp=<class 'numpy.int32'> e_tp=<class 'numpy.int32'>
vander                0.16699657 ms
bc_b                  0.09595204 ms
bc_e                  0.07959786 ms
ma                    0.10755240 ms
nnc                   0.16459018 ms
out_e_1               0.02037535 ms
out_e_2               0.02656622 ms
safe_e_1              0.04652272 ms
safe_e_2              0.04081079 ms
cheat bc_e_cheat            0.04668466 ms
rep=100 n_b=5000 n_e=8 b_tp=<class 'numpy.int32'> e_tp=<class 'numpy.int32'>
vander                0.25086462 ms
bc_b             apparently failed
bc_e             apparently failed
ma                    0.15843041 ms
nnc                   0.24713077 ms
out_e_1          apparently failed
out_e_2          apparently failed
safe_e_1              0.15970622 ms
safe_e_2              0.19672418 ms
bc_e_cheat       apparently failed
rep=100 n_b=5000 n_e=4 b_tp=<class 'float'> e_tp=<class 'numpy.int32'>
vander                0.16225773 ms
bc_b                  0.53315020 ms
bc_e                  0.56200830 ms
ma                    0.07626799 ms
nnc                   0.16059748 ms
out_e_1               0.03653416 ms
out_e_2               0.04043702 ms
safe_e_1              0.04060494 ms
safe_e_2              0.04104209 ms
cheat bc_e_cheat            0.52966076 ms
rep=100 n_b=5000 n_e=8 b_tp=<class 'float'> e_tp=<class 'numpy.int32'>
vander                0.24542852 ms
bc_b                  2.03353578 ms
bc_e                  2.04281270 ms
ma                    0.11075758 ms
nnc                   0.24212880 ms
out_e_1               0.14809043 ms
out_e_2               0.19261359 ms
safe_e_1              0.15206112 ms
safe_e_2              0.19308420 ms
cheat bc_e_cheat            1.99176601 ms

码:

import numpy as np
import types
from timeit import repeat

prom={np.dtype(np.int32): np.dtype(np.int64), np.dtype(float): np.dtype(float)}

def RI(k, N, dt, top=100):
    return np.random.randint(0, top if top else N, (k, N)).astype(dt)

def RA(k, N, dt, top=None):
    return np.add.outer(np.zeros((k,), int), np.arange(N)%(top if top else N)).astype(dt)

def RU(k, N, dt, top=100):
    return (np.random.random((k, N))*(top if top else N)).astype(dt)

def data(k, N_b, N_e, dt_b, dt_e, b_fun=RI, e_fun=RA):
    b = list(b_fun(k, N_b, dt_b))
    e = list(e_fun(k, N_e, dt_e))
    return b, e

def f_vander(b, e):
    return np.vander(b, len(e), increasing=True)

def f_bc_b(b, e):
    return b[:, None]**e

def f_bc_e(b, e):
    return np.ascontiguousarray((b**e[:, None]).T)

def f_ma(b, e):
    out = np.empty((len(b), len(e)), prom[b.dtype])
    out[:, 0] = 1
    np.multiply.accumulate(np.broadcast_to(b, (len(e)-1, len(b))), axis=0, out=out[:, 1:].T)
    return out

def f_nnc(b, e):
    out = np.empty((len(b), len(e)), prom[b.dtype])
    out[:, 0] = 1
    out[:, 1:] = b[:, None]
    np.multiply.accumulate(out[:, 1:], out=out[:, 1:], axis=1)
    return out

def f_out_e_1(b, e):
    out = np.empty((len(b), len(e)), b.dtype)
    out[:, 0] = 1
    out[:, 1] = b
    out[:, 2] = c = b*b
    for i in range(3, len(e)):
        c*=b
        out[:, i] = c
    return out

def f_out_e_2(b, e):
    out = np.empty((len(b), len(e)), b.dtype)
    out[:, 0] = 1
    out[:, 1] = b
    out[:, 2] = b*b
    for i in range(3, len(e)):
        out[:, i] = out[:, i-1] * b
    return out

def f_safe_e_1(b, e):
    out = np.empty((len(b), len(e)), prom[b.dtype])
    out[:, 0] = 1
    out[:, 1] = b
    out[:, 2] = c = (b*b).astype(prom[b.dtype])
    for i in range(3, len(e)):
        c*=b
        out[:, i] = c
    return out

def f_safe_e_2(b, e):
    out = np.empty((len(b), len(e)), prom[b.dtype])
    out[:, 0] = 1
    out[:, 1] = b
    out[:, 2] = b*b
    for i in range(3, len(e)):
        out[:, i] = out[:, i-1] * b
    return out

def f_bc_e_cheat(b, e):
    return (b**e[:, None]).T

for params in [(100, 5000, 4, np.int32, np.int32),
               (100, 5000, 8, np.int32, np.int32),
               (100, 5000, 4, float, np.int32),
               (100, 5000, 8, float, np.int32)]:
    k = params[0]
    dat = data(*params)
    ref = f_vander(dat[0][0], dat[1][0])
    print('rep={} n_b={} n_e={} b_tp={} e_tp={}'.format(*params))
    for name, func in list(globals().items()):
        if not name.startswith('f_') or not isinstance(func, types.FunctionType):
            continue
        try:
            assert np.allclose(ref, func(dat[0][0], dat[1][0]))
            if not func(dat[0][0], dat[1][0]).flags.c_contiguous:
                print('cheat', end=' ')
            print("{:16s}{:16.8f} ms".format(name[2:], np.min(repeat(
                'f(b.pop(), e.pop())', setup='b, e = data(*p)', globals={'f':func, 'data':data, 'p':params}, number=k)) * 1000 / k))
        except:
            print("{:16s} apparently failed".format(name[2:]))

4
2018-01-16 05:55



如果您有最新消息,请随时更新答案 发现 ;) - kmario23


广播取幂怎么样?

%timeit (x ** np.arange(N)[:, None]).T
43 µs ± 348 ns per loop (mean ± std. dev. of 7 runs, 10000 loops each)

完整性检查 -

np.all((x ** np.arange(N)[:, None]).T == np.vander(x, N, increasing=True))
True

需要注意的是,只有输入数组才能实现此加速 x 有一个 dtype 的 int。正如@Warren Weckesser在一篇评论中所指出的那样,对于浮点阵列,广播的取幂速度变慢了。


至于为什么 np.vander 很慢,看看吧 源代码  -

x = asarray(x)
if x.ndim != 1:
    raise ValueError("x must be a one-dimensional array or sequence.")
if N is None:
    N = len(x)

v = empty((len(x), N), dtype=promote_types(x.dtype, int))
tmp = v[:, ::-1] if not increasing else v

if N > 0:
    tmp[:, 0] = 1
if N > 1:
    tmp[:, 1:] = x[:, None]
    multiply.accumulate(tmp[:, 1:], out=tmp[:, 1:], axis=1)

return v

除了你的功能之外,这个函数必须满足更多的用例,因此它使用了更通用的计算方法,这种方法可靠,但速度较慢(我特别指出 multiply.accumulate)。


令人感兴趣的是,我找到了计算Vandermonde矩阵的另一种方法,结果如下:

%timeit x[:, None] ** np.arange(N)
150 µs ± 230 ns per loop (mean ± std. dev. of 7 runs, 10000 loops each)

它做同样的事情,但速度慢得多。答案在于操作是广播的,但是 低效

另一方面,为 float 数组,这实际上最终表现最好。


9
2018-01-14 01:51



我只是注意到输入数组时的power方法要慢得多 x 是浮点值的数组而不是整数。再试一次你的时间比较,但是 x = np.arange(5000.0)。 (这个例子 x 在这个问题中是一个整数数组,所以这对@ kmario23来说可能无关紧要。) - Warren Weckesser
@WarrenWeckesser似乎是这样 x[:, None] * np.arange(N) 成为花车最快的。 - coldspeed