优化“差分函数”的计算

Question

我的代码调用了许多“差分函数”来计算“Yin algorithm”（基频提取器）。

差分函数（论文中的eq. 6）定义为：

这是我对差异函数的实现：

def differenceFunction(x, W, tau_max):
   df = [0] * tau_max
   for tau in range(1, tau_max):
      for j in range(0, W - tau):
          tmp = long(x[j] - x[j + tau])
          df[tau] += tmp * tmp
   return df

例如：

x = np.random.randint(0, high=32000, size=2048, dtype='int16')
W = 2048
tau_max = 106
differenceFunction(x, W, tau_max)

有没有办法优化这种双循环计算（仅使用 python，最好不使用除 numpy 之外的其他库）？

编辑：更改代码以避免索引错误（j 循环，@Elliot 答案）

EDIT2：将代码更改为使用 x[0]（j 循环，@hynekcer 注释）

Answer 1

编辑：将速度提高到 220 µs - 请参阅最后的编辑 - 直接版本

可以通过Autocorrelation function 或类似地通过卷积轻松评估所需的计算。 Wiener-Khinchin 定理允许使用两个快速傅里叶变换 (FFT) 计算自相关，时间复杂度 O(n log n)。 我使用来自Scipy 包的加速卷积函数fftconvolve。一个优点是在这里很容易解释它为什么起作用。一切都是矢量化的，在 Python 解释器级别没有循环。

from scipy.signal import fftconvolve

def difference_by_convol(x, W, tau_max):
    x = np.array(x, np.float64)
    w = x.size
    x_cumsum = np.concatenate((np.array([0.]), (x * x).cumsum()))
    conv = fftconvolve(x, x[::-1])
    df = x_cumsum[w:0:-1] + x_cumsum[w] - x_cumsum[:w] - 2 * conv[w - 1:]
    return df[:tau_max + 1]

• 与Elliot's answer 中的differenceFunction_1loop 函数相比：FFT 更快：430 µs 与原来的 1170 µs 相比。大约tau_max >= 40 开始更快。数值精度很高。与精确整数结果相比，最高相对误差小于 1E-14。 （因此它可以很容易地四舍五入到精确的长整数解。）
• 参数tau_max 对算法并不重要。它最终只会限制输出。索引 0 处的零元素被添加到输出中，因为在 Python 中索引应该从 0 开始。
• 参数W在Python中并不重要。大小更适合自省。
• 最初将数据转换为 np.float64 以防止重复转换。它快了半个百分点。由于溢出，任何小于 np.int64 的类型都是不可接受的。
• 所需的差分函数是双能量减去自相关函数。这可以通过卷积来评估：correlate(x, x) = convolve(x, reversed(x)。
• “从 Scipy v0.19 开始，正常 convolve 会自动选择此方法或基于估计哪个更快的直接方法。”这种启发式方法不适合这种情况，因为卷积计算的 tau 比 tau_max 多得多，而且它必须被比直接方法更快的 FFT 压倒。
• 它也可以通过没有 Scipy 的 Numpy ftp 模块通过将答案 Calculate autocorrelation using FFT in matlab 重写为 Python 来计算（在最后）。我认为上面的解决方案可以更容易理解。

---

证明：（对于 Pythonistas :-)

最初的朴素实现可以写成：

df = [sum((x[j] - x[j + t]) ** 2   for j in range(w - t))  for t in range(tau_max + 1)]

tau_max < w.

按规则派生(a - b)**2 == a**2 + b**2 - 2 * a * b

df = [  sum(x[j] ** 2 for j in range(w - t))
      + sum(x[j] ** 2 for j in range(t, w))
      - 2 * sum(x[j] * x[j + t] for j in range(w - t))
      for t in range(tau_max + 1)]

在x_cumsum = [sum(x[j] ** 2 for j in range(i)) for i in range(w + 1)] 的帮助下替换前两个元素，这可以在线性时间内轻松计算。用输出大小为len(x) + len(x) - 1 的卷积conv = convolvefft(x, reversed(x), mode='full') 替换sum(x[j] * x[j + t] for j in range(w - t))。

df = [x_cumsum[w - t] + x_cumsum[w] - x_cumsum[t]
      - 2 * convolve(x, x[::-1])[w - 1 + t]
      for t in range(tau_max + 1)]

通过向量表达式进行优化：

df = x_cumsum[w:0:-1] + x_cumsum[w] - x_cumsum[:w] - 2 * conv[w - 1:]

每一步也可以通过测试数据进行数值测试和比较

---

编辑：直接由 Numpy FFT 实施的解决方案。

def difference_fft(x, W, tau_max):
    x = np.array(x, np.float64)
    w = x.size
    tau_max = min(tau_max, w)
    x_cumsum = np.concatenate((np.array([0.]), (x * x).cumsum()))
    size = w + tau_max 
    p2 = (size // 32).bit_length()
    nice_numbers = (16, 18, 20, 24, 25, 27, 30, 32)
    size_pad = min(x * 2 ** p2 for x in nice_numbers if x * 2 ** p2 >= size)
    fc = np.fft.rfft(x, size_pad)
    conv = np.fft.irfft(fc * fc.conjugate())[:tau_max]
    return x_cumsum[w:w - tau_max:-1] + x_cumsum[w] - x_cumsum[:tau_max] - 2 * conv

它比我之前的解决方案快两倍多，因为卷积的长度被限制为在W + tau_max 之后具有小素因数的最接近的“好”数，而不是评估完整的2 * W。也不需要像使用 `fftconvolve(x, reversed(x)) 那样将相同的数据转换两次。

In [211]: %timeit differenceFunction_1loop(x, W, tau_max)
1.1 ms ± 4.51 µs per loop (mean ± std. dev. of 7 runs, 1000 loops each)

In [212]: %timeit difference_by_convol(x, W, tau_max)
431 µs ± 5.69 µs per loop (mean ± std. dev. of 7 runs, 1000 loops each)

In [213]: %timeit difference_fft(x, W, tau_max)
218 µs ± 685 ns per loop (mean ± std. dev. of 7 runs, 1000 loops each)

对于 tau_max >= 20，最新的解决方案比 Eliot 的 difference_by_convol 更快。该比率不太依赖于数据大小，因为开销成本的比率相似。

Answer 2

首先，您应该考虑数组的边界。您最初编写的代码将获得IndexError。 您可以通过矢量化内循环来获得显着的加速

import numpy as np

# original version
def differenceFunction_2loop(x, W, tau_max):
   df = np.zeros(tau_max, np.long)
   for tau in range(1, tau_max):
      for j in range(0, W - tau): # -tau eliminates the IndexError
          tmp = np.long(x[j] -x[j + tau])
          df[tau] += np.square(tmp)
   return df

# vectorized inner loop
def differenceFunction_1loop(x, W, tau_max):
    df = np.zeros(tau_max, np.long)
    for tau in range(1, tau_max):
        tmp = (x[:-tau]) - (x[tau:]).astype(np.long)
        df[tau] = np.dot(tmp, tmp)
    return df

x = np.random.randint(0, high=32000, size=2048, dtype='int16')
W = 2048
tau_max = 106
twoloop = differenceFunction_2loop(x, W, tau_max)
oneloop = differenceFunction_1loop(x, W, tau_max)

# confirm that the result comes out the same. 
print(np.all(twoloop == oneloop))
# True

现在进行一些基准测试。在ipython 我得到以下内容

In [103]: %timeit twoloop = differenceFunction_2loop(x, W, tau_max)
1 loop, best of 3: 2.35 s per loop

In [104]: %timeit oneloop = differenceFunction_1loop(x, W, tau_max)
100 loops, best of 3: 8.23 ms per loop

所以，大约提高了 300 倍。

Answer 3

与优化算法相反，您可以使用 numba.jit 优化解释器：

import timeit

import numpy as np
from numba import jit


def differenceFunction(x, W, tau_max):
    df = [0] * tau_max
    for tau in range(1, tau_max):
        for j in range(0, W - tau):
            tmp = int(x[j] - x[j + tau])
            df[tau] += tmp * tmp
    return df


@jit
def differenceFunction2(x, W, tau_max):
    df = np.ndarray(shape=(tau_max,))
    for tau in range(1, tau_max):
        for j in range(0, W - tau):
            tmp = int(x[j] - x[j + tau])
            df[tau] += tmp * tmp
    return df


x = np.random.randint(0, high=32000, size=2048, dtype='int16')
W = 2048
tau_max = 106
differenceFunction(x, W, tau_max)


print('old',
      timeit.timeit('differenceFunction(x, W, tau_max)', 'from __main__ import differenceFunction, x, W, tau_max',
                    number=20) / 20)
print('new',
      timeit.timeit('differenceFunction2(x, W, tau_max)', 'from __main__ import differenceFunction2, x, W, tau_max',
                    number=20) / 20)

结果：

old 0.18265145074453273
new 0.016223197058214667

您可以结合优化算法和numba.jit 以获得更好的结果。

Answer 4

这是使用列表理解的另一种方法。它花费的时间大约不到原始函数的十分之一，但不超过Elliot's answer。还是把它放在那里吧。

import numpy as np
import time

# original version
def differenceFunction_2loop(x, W, tau_max):
   df = np.zeros(tau_max, np.long)
   for tau in range(1, tau_max):
      for j in range(0, W - tau): # -tau eliminates the IndexError
          tmp = np.long(x[j] -x[j + tau])
          df[tau] += np.square(tmp)
   return df

# vectorized inner loop
def differenceFunction_1loop(x, W, tau_max):
    df = np.zeros(tau_max, np.long)
    for tau in range(1, tau_max):
        tmp = (x[:-tau]) - (x[tau:]).astype(np.long)
        df[tau] = np.dot(tmp, tmp)
    return df

# with list comprehension
def differenceFunction_1loop_listcomp(x, W, tau_max):
    df = [sum(((x[:-tau]) - (x[tau:]).astype(np.long))**2) for tau in range(1, tau_max)]
    return [0] + df[:]

x = np.random.randint(0, high=32000, size=2048, dtype='int16')
W = 2048
tau_max = 106

s = time.clock()
twoloop = differenceFunction_2loop(x, W, tau_max)
print(time.clock() - s)

s = time.clock()
oneloop = differenceFunction_1loop(x, W, tau_max)
print(time.clock() - s)

s = time.clock()
listcomprehension = differenceFunction_1loop_listcomp(x, W, tau_max)
print(time.clock() - s)

# confirm that the result comes out the same. 
print(np.all(twoloop == listcomprehension))
# True

性能结果（大约）：

differenceFunction_2loop() = 0.47s
differenceFunction_1loop() = 0.003s
differenceFunction_1loop_listcomp() = 0.033s

Answer 5

我不知道如何找到嵌套循环问题的替代方案，但对于算术函数，您可以使用 numpy 库。比手动操作要快。

import numpy as np
tmp = np.subtract(long(x[j] ,x[j + tau])

Answer 6

我会这样做：

>>> x = np.random.randint(0, high=32000, size=2048, dtype='int16')
>>> tau_max = 106
>>> res = np.square((x[tau_max:] - x[:-tau_max]))

但是我相信这不是最快的方法。

Answer 7

我试图弄清楚最快的答案，但我想出了一个更快、更简单的解决方案。

def autocorrelation(x):
    result = np.correlate(x, x, mode='full')
    return result[result.size // 2:]

def difference(x):
    return np.dot(x, x) + (x * x)[::-1].cumsum()[::-1] - 2 * autocorrelation(x)

解决方案基于the YIN paper 中定义的difference 函数。

%%timeit
difference(frame)

140 µs ± 438 ns per loop (mean ± std. dev. of 7 runs, 10000 loops each)