快速计算条件函数的方法

5
什么是计算类似函数的最快方法?
# here x is just a number
def f(x):
    if x >= 0:
        return np.log(x+1)
    else:
        return -np.log(-x+1)

一种可能的方法是:

# here x is an array
def loga(x)
    cond = [x >= 0, x < 0]
    choice = [np.log(x+1), -np.log(-x+1)
    return np.select(cond, choice)

但似乎numpy是逐个遍历数组元素的。 有没有类似于np.exp(x)的概念性方法来实现更好的性能?


你尝试过这些解决方案中的任何一个吗? - Divakar
@Divakar 对不起,我还没有检查的可能性。我会在最近的时间尝试并标记一个答案。 - ichernob
4个回答

6
    def f(x):
        return (x/abs(x)) * np.log(1+abs(x))

不错!这可以工作。数学做得好。与 numexpr 和/或重复使用 abs(x) 应该是赢家。 - Divakar
3
建议使用 numpy.sign(x) 而不是 x/abs(x),以避免可能出现的除零错误。另一种方法是使用 np.copysign(np.log(1 + abs(x)), x) - Mark Dickinson
这一切都很不错。但是像abs(),np.sign()这样的函数可能会返回0。Mark的解决方案真的很好,但是我想找到比np.tanh(x)更快的解决方案。这可能吗? - ichernob

3
在这种情况下,掩码技术可以帮助解决问题 -
def mask_vectorized_app(x):
    out = np.empty_like(x)
    mask = x>=0
    mask_rev = ~mask
    out[mask] = np.log(x[mask]+1)
    out[mask_rev] = -np.log(-x[mask_rev]+1)
    return out

介绍一下numexpr模块,它对于我们的技术有很大帮助。
import numexpr as ne

def mask_vectorized_numexpr_app(x):
    out = np.empty_like(x)
    mask = x>=0
    mask_rev = ~mask

    x_masked = x[mask]
    x_rev_masked = x[mask_rev]
    out[mask] = ne.evaluate('log(x_masked+1)')
    out[mask_rev] = ne.evaluate('-log(-x_rev_masked+1)')
    return out

@user2685079的帖子的启发,然后使用对数性质:log(A**B) = B*log(A),我们可以将符号推入对数计算中,这使我们能够更多地使用numexpr的评估表达式,像这样 -

s = (-2*(x<0))+1 # np.sign(x)
out = ne.evaluate('log( (abs(x)+1)**s)')

使用比较计算 sign 可以用另一种方式得到 s -

s = (-2*(x<0))+1

最后,我们可以将这个内容推入到numexpr评估表达式中 -
def mask_vectorized_numexpr_app2(x):
    return ne.evaluate('log( (abs(x)+1)**((-2*(x<0))+1))')

运行时测试

循环比较的方法 -

def loopy_app(x):
    out = np.empty_like(x)
    for i in range(len(out)):
        out[i] = f(x[i])
    return out

定时和验证 -

In [141]: x = np.random.randn(100000)
     ...: print np.allclose(loopy_app(x), mask_vectorized_app(x))
     ...: print np.allclose(loopy_app(x), mask_vectorized_numexpr_app(x))
     ...: print np.allclose(loopy_app(x), mask_vectorized_numexpr_app2(x))
     ...: 
True
True
True

In [142]: %timeit loopy_app(x)
     ...: %timeit mask_vectorized_numexpr_app(x)
     ...: %timeit mask_vectorized_numexpr_app2(x)
     ...: 
10 loops, best of 3: 108 ms per loop
100 loops, best of 3: 3.6 ms per loop
1000 loops, best of 3: 942 µs per loop

使用@user2685079的解决方案,使用np.sign替换第一个部分,然后进行有和无numexpr评估 -

In [143]: %timeit np.sign(x) * np.log(1+abs(x))
100 loops, best of 3: 3.26 ms per loop

In [144]: %timeit np.sign(x) * ne.evaluate('log(1+abs(x))')
1000 loops, best of 3: 1.66 ms per loop

请问你可以把我的时间加进去吗? - piRSquared
1
@piRSquared,我没有numba。你介意把我的加入你的吗? - Divakar
添加了最新的 app2 - piRSquared
@piRSquared 确认一下 - 你使用的是最新的一行代码版本,对吧? - Divakar
是的,我在我的答案中包含了函数定义以消除歧义。 - piRSquared

2

使用 numba

Numba 可以让你通过高性能函数在 Python 中加速应用程序。只需添加几个注释,面向数组和数学运算的 Python 代码就可以即时编译为本机机器指令,与 C、C++ 和 Fortran 的性能相当,无需切换语言或 Python 解释器。

Numba 在导入时、运行时或静态编译(使用包含的 pycc 工具)时使用 LLVM 编译器基础架构生成优化后的机器代码。Numba 支持将 Python 编译成可在 CPU 或 GPU 硬件上运行,并设计与 Python 科学软件堆栈集成。

Numba 项目由 Continuum Analytics 和 The Gordon and Betty Moore Foundation(Grant GBMF5423)支持。

from numba import njit
import numpy as np

@njit
def pir(x):
    a = np.empty_like(x)
    for i in range(a.size):
        x_ = x[i]
        _x = abs(x_)
        a[i] = np.sign(x_) * np.log(1 + _x)
    return a

Accuracy

np.isclose(pir(x), f(x)).all()

True

时间

x = np.random.randn(100000)

# My proposal
%timeit pir(x)
1000 loops, best of 3: 881 µs per loop

# OP test
%timeit f(x)
1000 loops, best of 3: 1.26 ms per loop

# Divakar-1
%timeit mask_vectorized_numexpr_app(x)
100 loops, best of 3: 2.97 ms per loop

# Divakar-2
%timeit mask_vectorized_numexpr_app2(x)
1000 loops, best of 3: 621 µs per loop

函数定义

from numba import njit
import numpy as np

@njit
def pir(x):
    a = np.empty_like(x)
    for i in range(a.size):
        x_ = x[i]
        _x = abs(x_)
        a[i] = np.sign(x_) * np.log(1 + _x)
    return a

import numexpr as ne

def mask_vectorized_numexpr_app(x):
    out = np.empty_like(x)
    mask = x>=0
    mask_rev = ~mask

    x_masked = x[mask]
    x_rev_masked = x[mask_rev]
    out[mask] = ne.evaluate('log(x_masked+1)')
    out[mask_rev] = ne.evaluate('-log(-x_rev_masked+1)')
    return out

def mask_vectorized_numexpr_app2(x):
    return ne.evaluate('log( (abs(x)+1)**((-2*(x<0))+1))')


def f(x):
    return (x/abs(x)) * np.log(1+abs(x))

我的最新应用程序2有点不同 :) 抱歉给您带来麻烦。您能否更新一下?不认为您需要再次列出那些内容。 - Divakar
1
@Divakar 没有问题 :-). 最新的那个真的很快。 - piRSquared

1
你可以使用np.where替代np.select,稍微提高第二个解决方案的速度:
def loga(x):
    cond = [x >= 0, x < 0]
    choice = [np.log(x+1), -np.log(-x+1)]
    return np.select(cond, choice)

def logb(x):
    return np.where(x>=0, np.log(x+1), -np.log(-x+1))

In [16]: %timeit loga(arange(-1000,1000))
10000 loops, best of 3: 169 µs per loop

In [17]: %timeit logb(arange(-1000,1000))
10000 loops, best of 3: 98.3 µs per loop

In [18]: np.all(loga(arange(-1000,1000)) == logb(arange(-1000,1000)))
Out[18]: True

网页内容由stack overflow 提供, 点击上面的
可以查看英文原文,
原文链接