Numpy 中的快速线性插值 / Scipy "along a path"
Fast linear interpolation in Numpy / Scipy "along a path"
假设我有来自山上 3 个（已知）海拔高度的气象站的数据。具体来说，每个站点每分钟记录一次其所在位置的温度测量值。我有两种我想执行的插值。我希望能够快速执行每一项。
那么让我们设置一些数据：
import numpy as np
from scipy.interpolate import interp1d
import pandas as pd
import seaborn as sns
np.random.seed(0)
N, sigma = 1000., 5
basetemps = 70 + (np.random.randn(N) * sigma)
midtemps = 50 + (np.random.randn(N) * sigma)
toptemps = 40 + (np.random.randn(N) * sigma)
alltemps = np.array([basetemps, midtemps, toptemps]).T # note transpose!
trend = np.sin(4 / N * np.arange(N)) * 30
trend = trend[:, np.newaxis]
altitudes = np.array([500, 1500, 4000]).astype(float)
finaltemps = pd.DataFrame(alltemps + trend, columns=altitudes)
finaltemps.index.names, finaltemps.columns.names = ['Time'], ['Altitude']
finaltemps.plot()
太好了，所以我们的温度看起来是这样的：
将所有时间插值到同一高度：
我认为这个非常简单。假设我想每次获取 1,000 海拔高度的温度。我可以只使用内置的 scipy 插值方法：
interping_function = interp1d(altitudes, finaltemps.values)
interped_to_1000 = interping_function(1000)
fig, ax = plt.subplots(1, 1, figsize=(8, 5))
finaltemps.plot(ax=ax, alpha=0.15)
ax.plot(interped_to_1000, label='Interped')
ax.legend(loc='best', title=finaltemps.columns.name)
这很好用。让我们看看速度：
%%timeit
res = interp1d(altitudes, finaltemps.values)(1000)
#-> 1000 loops, best of 3: 207 µs per loop
插值"along a path":
所以现在我有第二个相关问题。假设我知道徒步旅行队的海拔高度随时间的变化，我想通过对时间数据进行线性插值来计算他们（移动）位置的温度。 特别是，我知道远足派对地点的时间与我知道气象站温度的时间相同。我可以毫不费力地做到这一点：
location = np.linspace(altitudes[0], altitudes[-1], N)
interped_along_path = np.array([interp1d(altitudes, finaltemps.values[i, :])(loc)
for i, loc in enumerate(location)])
fig, ax = plt.subplots(1, 1, figsize=(8, 5))
finaltemps.plot(ax=ax, alpha=0.15)
ax.plot(interped_along_path, label='Interped')
ax.legend(loc='best', title=finaltemps.columns.name)
所以这确实很好用，但重要的是要注意上面的关键行是使用列表理解来隐藏大量工作。在前面的例子中，scipy 正在为我们创建一个单一的插值函数，并在大量数据上对其进行一次评估。在这种情况下，scipy 实际上是在构造 N 个单独的插值函数，并在少量数据上对每个插值函数进行一次评估。这感觉本质上是低效的。这里（在列表理解中）潜伏着一个 for 循环，而且，这感觉很松弛。
不足为奇，这比之前的情况慢得多：
%%timeit
res = np.array([interp1d(altitudes, finaltemps.values[i, :])(loc)
for i, loc in enumerate(location)])
#-> 10 loops, best of 3: 145 ms per loop
因此第二个示例比第一个慢 1,000。 IE。与繁重的工作是 "make a linear interpolation function" 步骤的想法一致......在第二个示例中发生了 1,000 次，但在第一个示例中只发生了一次。
所以，问题是：是否有更好的方法来解决第二个问题？例如，是否有一个很好的方法来设置二维插值（也许可以处理这样的情况，即已知徒步旅行地点的时间 而不是 对温度进行采样的时间）？还是有一种特别巧妙的方式来处理这里时间排队的事情？还是其他？
我会提供一点进步。在第二种情况下（插值 "along a path"），我们正在制作许多不同的插值函数。我们可以尝试的一件事是只制作一个插值函数（一个在所有时间都在高度维度上进行插值的函数，如上面的第一种情况）并一遍又一遍地评估该函数（以矢量化方式）。这将为我们提供比我们想要的更多的数据（它将为我们提供 1,000 x 1,000 矩阵而不是 1,000 元素向量）。但是这样我们的目标结果就会沿着对角线。所以问题是，在更复杂的参数上调用单个函数 运行 是否比创建许多函数并使用简单参数调用它们更快？
答案是肯定的！
关键是 scipy.interpolate.interp1d 返回的插值函数能够接受 numpy.ndarray 作为其输入。因此，您可以通过输入矢量输入以 C 速度有效地多次调用插值函数。 IE。这比编写一个 for 循环要快得多，后者在标量输入上一遍又一遍地调用插值函数。因此，虽然我们计算了很多我们最终丢弃的数据点，但我们通过不构建我们几乎不使用的许多不同的插值函数来节省更多时间。
old_way = interped_along_path = np.array([interp1d(altitudes, finaltemps.values[i, :])(loc)
for i, loc in enumerate(location)])
# look ma, no for loops!
new_way = np.diagonal(interp1d(altitudes, finaltemps.values)(location))
# note, `location` is a vector!
abs(old_way - new_way).max()
#-> 0.0
还有：
%%timeit
res = np.diagonal(interp1d(altitudes, finaltemps.values)(location))
#-> 100 loops, best of 3: 16.7 ms per loop
所以这种方法让我们的效率提高了 10 倍！谁能做得更好？或者建议一种完全不同的方法？
两个值 y1、y2 在位置 x1 和 x2 之间关于点 xi 的线性插值很简单：
yi = y1 + (y2-y1) * (xi-x1) / (x2-x1)
通过一些向量化的 Numpy 表达式，我们可以 select 数据集中的相关点并应用上述函数：
I = np.searchsorted(altitudes, location)
x1 = altitudes[I-1]
x2 = altitudes[I]
time = np.arange(len(alltemps))
y1 = alltemps[time,I-1]
y2 = alltemps[time,I]
xI = location
yI = y1 + (y2-y1) * (xI-x1) / (x2-x1)
麻烦的是有些点位于已知范围的边界（甚至之外），应该考虑到这一点：
I = np.searchsorted(altitudes, location)
same = (location == altitudes.take(I, mode='clip'))
out_of_range = ~same & ((I == 0) | (I == altitudes.size))
I[out_of_range] = 1 # Prevent index-errors
x1 = altitudes[I-1]
x2 = altitudes[I]
time = np.arange(len(alltemps))
y1 = alltemps[time,I-1]
y2 = alltemps[time,I]
xI = location
yI = y1 + (y2-y1) * (xI-x1) / (x2-x1)
yI[out_of_range] = np.nan
幸运的是，Scipy已经提供了ND插值，这也很容易处理不匹配的时间，例如：
from scipy.interpolate import interpn
time = np.arange(len(alltemps))
M = 150
hiketime = np.linspace(time[0], time[-1], M)
location = np.linspace(altitudes[0], altitudes[-1], M)
xI = np.column_stack((hiketime, location))
yI = interpn((time, altitudes), alltemps, xI)
这是一个基准代码（实际上没有任何 pandas，我确实包含了其他答案中的解决方案）：
import numpy as np
from scipy.interpolate import interp1d, interpn
def original():
return np.array([interp1d(altitudes, alltemps[i, :])(loc)
for i, loc in enumerate(location)])
def OP_self_answer():
return np.diagonal(interp1d(altitudes, alltemps)(location))
def interp_checked():
I = np.searchsorted(altitudes, location)
same = (location == altitudes.take(I, mode='clip'))
out_of_range = ~same & ((I == 0) | (I == altitudes.size))
I[out_of_range] = 1 # Prevent index-errors
x1 = altitudes[I-1]
x2 = altitudes[I]
time = np.arange(len(alltemps))
y1 = alltemps[time,I-1]
y2 = alltemps[time,I]
xI = location
yI = y1 + (y2-y1) * (xI-x1) / (x2-x1)
yI[out_of_range] = np.nan
return yI
def scipy_interpn():
time = np.arange(len(alltemps))
xI = np.column_stack((time, location))
yI = interpn((time, altitudes), alltemps, xI)
return yI
N, sigma = 1000., 5
basetemps = 70 + (np.random.randn(N) * sigma)
midtemps = 50 + (np.random.randn(N) * sigma)
toptemps = 40 + (np.random.randn(N) * sigma)
trend = np.sin(4 / N * np.arange(N)) * 30
trend = trend[:, np.newaxis]
alltemps = np.array([basetemps, midtemps, toptemps]).T + trend
altitudes = np.array([500, 1500, 4000], dtype=float)
location = np.linspace(altitudes[0], altitudes[-1], N)
funcs = [original, interp_checked, scipy_interpn]
for func in funcs:
print(func.func_name)
%timeit func()
from itertools import combinations
outs = [func() for func in funcs]
print('Output allclose:')
print([np.allclose(out1, out2) for out1, out2 in combinations(outs, 2)])
我的系统结果如下：
original
10 loops, best of 3: 184 ms per loop
OP_self_answer
10 loops, best of 3: 89.3 ms per loop
interp_checked
1000 loops, best of 3: 224 µs per loop
scipy_interpn
1000 loops, best of 3: 1.36 ms per loop
Output allclose:
[True, True, True, True, True, True]
与最快的方法相比，Scipy 的 interpn 在速度方面有所下降，但就其通用性和易用性而言，它绝对是最佳选择。
对于一个固定的时间点，可以利用下面的插值函数：
g(a) = cc[0]*abs(a-aa[0]) + cc[1]*abs(a-aa[1]) + cc[2]*abs(a-aa[2])
其中 a 是徒步旅行者的高度，aa 具有 3 测量值的向量 altitudes 和 cc 是具有系数的向量。需要注意三点：
- 对于对应于
aa 的给定温度 (alltemps)，可以通过使用 np.linalg.solve(). 求解线性矩阵方程来确定 cc
g(a) 很容易向量化为 (N,) 维 a 和 (N, 3) 维 cc（分别包括 np.linalg.solve()）。g(a)称为一阶单变量样条核（三点）。使用 abs(a-aa[i])**(2*d-1) 会将样条顺序更改为 d。这种方法可以解释为 Gaussian Process in Machine Learning.
所以代码是：
import matplotlib.pyplot as plt
import numpy as np
import seaborn as sns
# generate temperatures
np.random.seed(0)
N, sigma = 1000, 5
trend = np.sin(4 / N * np.arange(N)) * 30
alltemps = np.array([tmp0 + trend + sigma*np.random.randn(N)
for tmp0 in [70, 50, 40]])
# generate attitudes:
altitudes = np.array([500, 1500, 4000]).astype(float)
location = np.linspace(altitudes[0], altitudes[-1], N)
def doit():
""" do the interpolation, improved version for speed """
AA = np.vstack([np.abs(altitudes-a_i) for a_i in altitudes])
# This is slighty faster than np.linalg.solve(), because AA is small:
cc = np.dot(np.linalg.inv(AA), alltemps)
return (cc[0]*np.abs(location-altitudes[0]) +
cc[1]*np.abs(location-altitudes[1]) +
cc[2]*np.abs(location-altitudes[2]))
t_loc = doit() # call interpolator
# do the plotting:
fg, ax = plt.subplots(num=1)
for alt, t in zip(altitudes, alltemps):
ax.plot(t, label="%d feet" % alt, alpha=.5)
ax.plot(t_loc, label="Interpolation")
ax.legend(loc="best", title="Altitude:")
ax.set_xlabel("Time")
ax.set_ylabel("Temperature")
fg.canvas.draw()
测量时间得出：
In [2]: %timeit doit()
10000 loops, best of 3: 107 µs per loop
更新：我在doit()中替换了原来的列表理解
导入速度提高 30%（对于 N=1000）。
此外，根据要求进行比较，@moarningsun 在我的机器上的基准代码块：
10 loops, best of 3: 110 ms per loop
interp_checked
10000 loops, best of 3: 83.9 µs per loop
scipy_interpn
1000 loops, best of 3: 678 µs per loop
Output allclose:
[True, True, True]
请注意 N=1000 是一个相对较小的数字。使用 N=100000 产生结果：
interp_checked
100 loops, best of 3: 8.37 ms per loop
%timeit doit()
100 loops, best of 3: 5.31 ms per loop
这表明此方法比 interp_checked 方法更适合大型 N。
假设我有来自山上 3 个（已知）海拔高度的气象站的数据。具体来说，每个站点每分钟记录一次其所在位置的温度测量值。我有两种我想执行的插值。我希望能够快速执行每一项。
那么让我们设置一些数据：
import numpy as np
from scipy.interpolate import interp1d
import pandas as pd
import seaborn as sns
np.random.seed(0)
N, sigma = 1000., 5
basetemps = 70 + (np.random.randn(N) * sigma)
midtemps = 50 + (np.random.randn(N) * sigma)
toptemps = 40 + (np.random.randn(N) * sigma)
alltemps = np.array([basetemps, midtemps, toptemps]).T # note transpose!
trend = np.sin(4 / N * np.arange(N)) * 30
trend = trend[:, np.newaxis]
altitudes = np.array([500, 1500, 4000]).astype(float)
finaltemps = pd.DataFrame(alltemps + trend, columns=altitudes)
finaltemps.index.names, finaltemps.columns.names = ['Time'], ['Altitude']
finaltemps.plot()
太好了，所以我们的温度看起来是这样的：
将所有时间插值到同一高度：
我认为这个非常简单。假设我想每次获取 1,000 海拔高度的温度。我可以只使用内置的 scipy 插值方法：
interping_function = interp1d(altitudes, finaltemps.values)
interped_to_1000 = interping_function(1000)
fig, ax = plt.subplots(1, 1, figsize=(8, 5))
finaltemps.plot(ax=ax, alpha=0.15)
ax.plot(interped_to_1000, label='Interped')
ax.legend(loc='best', title=finaltemps.columns.name)
这很好用。让我们看看速度：
%%timeit
res = interp1d(altitudes, finaltemps.values)(1000)
#-> 1000 loops, best of 3: 207 µs per loop
插值"along a path":
所以现在我有第二个相关问题。假设我知道徒步旅行队的海拔高度随时间的变化，我想通过对时间数据进行线性插值来计算他们（移动）位置的温度。 特别是，我知道远足派对地点的时间与我知道气象站温度的时间相同。我可以毫不费力地做到这一点：
location = np.linspace(altitudes[0], altitudes[-1], N)
interped_along_path = np.array([interp1d(altitudes, finaltemps.values[i, :])(loc)
for i, loc in enumerate(location)])
fig, ax = plt.subplots(1, 1, figsize=(8, 5))
finaltemps.plot(ax=ax, alpha=0.15)
ax.plot(interped_along_path, label='Interped')
ax.legend(loc='best', title=finaltemps.columns.name)
所以这确实很好用，但重要的是要注意上面的关键行是使用列表理解来隐藏大量工作。在前面的例子中，scipy 正在为我们创建一个单一的插值函数，并在大量数据上对其进行一次评估。在这种情况下，scipy 实际上是在构造 N 个单独的插值函数，并在少量数据上对每个插值函数进行一次评估。这感觉本质上是低效的。这里（在列表理解中）潜伏着一个 for 循环，而且，这感觉很松弛。
不足为奇，这比之前的情况慢得多：
%%timeit
res = np.array([interp1d(altitudes, finaltemps.values[i, :])(loc)
for i, loc in enumerate(location)])
#-> 10 loops, best of 3: 145 ms per loop
因此第二个示例比第一个慢 1,000。 IE。与繁重的工作是 "make a linear interpolation function" 步骤的想法一致......在第二个示例中发生了 1,000 次，但在第一个示例中只发生了一次。
所以，问题是：是否有更好的方法来解决第二个问题？例如，是否有一个很好的方法来设置二维插值（也许可以处理这样的情况，即已知徒步旅行地点的时间 而不是 对温度进行采样的时间）？还是有一种特别巧妙的方式来处理这里时间排队的事情？还是其他？
我会提供一点进步。在第二种情况下（插值 "along a path"），我们正在制作许多不同的插值函数。我们可以尝试的一件事是只制作一个插值函数（一个在所有时间都在高度维度上进行插值的函数，如上面的第一种情况）并一遍又一遍地评估该函数（以矢量化方式）。这将为我们提供比我们想要的更多的数据（它将为我们提供 1,000 x 1,000 矩阵而不是 1,000 元素向量）。但是这样我们的目标结果就会沿着对角线。所以问题是，在更复杂的参数上调用单个函数 运行 是否比创建许多函数并使用简单参数调用它们更快？
答案是肯定的！
关键是 scipy.interpolate.interp1d 返回的插值函数能够接受 numpy.ndarray 作为其输入。因此，您可以通过输入矢量输入以 C 速度有效地多次调用插值函数。 IE。这比编写一个 for 循环要快得多，后者在标量输入上一遍又一遍地调用插值函数。因此，虽然我们计算了很多我们最终丢弃的数据点，但我们通过不构建我们几乎不使用的许多不同的插值函数来节省更多时间。
old_way = interped_along_path = np.array([interp1d(altitudes, finaltemps.values[i, :])(loc)
for i, loc in enumerate(location)])
# look ma, no for loops!
new_way = np.diagonal(interp1d(altitudes, finaltemps.values)(location))
# note, `location` is a vector!
abs(old_way - new_way).max()
#-> 0.0
还有：
%%timeit
res = np.diagonal(interp1d(altitudes, finaltemps.values)(location))
#-> 100 loops, best of 3: 16.7 ms per loop
所以这种方法让我们的效率提高了 10 倍！谁能做得更好？或者建议一种完全不同的方法？
两个值 y1、y2 在位置 x1 和 x2 之间关于点 xi 的线性插值很简单：
yi = y1 + (y2-y1) * (xi-x1) / (x2-x1)
通过一些向量化的 Numpy 表达式，我们可以 select 数据集中的相关点并应用上述函数：
I = np.searchsorted(altitudes, location)
x1 = altitudes[I-1]
x2 = altitudes[I]
time = np.arange(len(alltemps))
y1 = alltemps[time,I-1]
y2 = alltemps[time,I]
xI = location
yI = y1 + (y2-y1) * (xI-x1) / (x2-x1)
麻烦的是有些点位于已知范围的边界（甚至之外），应该考虑到这一点：
I = np.searchsorted(altitudes, location)
same = (location == altitudes.take(I, mode='clip'))
out_of_range = ~same & ((I == 0) | (I == altitudes.size))
I[out_of_range] = 1 # Prevent index-errors
x1 = altitudes[I-1]
x2 = altitudes[I]
time = np.arange(len(alltemps))
y1 = alltemps[time,I-1]
y2 = alltemps[time,I]
xI = location
yI = y1 + (y2-y1) * (xI-x1) / (x2-x1)
yI[out_of_range] = np.nan
幸运的是，Scipy已经提供了ND插值，这也很容易处理不匹配的时间，例如：
from scipy.interpolate import interpn
time = np.arange(len(alltemps))
M = 150
hiketime = np.linspace(time[0], time[-1], M)
location = np.linspace(altitudes[0], altitudes[-1], M)
xI = np.column_stack((hiketime, location))
yI = interpn((time, altitudes), alltemps, xI)
这是一个基准代码（实际上没有任何 pandas，我确实包含了其他答案中的解决方案）：
import numpy as np
from scipy.interpolate import interp1d, interpn
def original():
return np.array([interp1d(altitudes, alltemps[i, :])(loc)
for i, loc in enumerate(location)])
def OP_self_answer():
return np.diagonal(interp1d(altitudes, alltemps)(location))
def interp_checked():
I = np.searchsorted(altitudes, location)
same = (location == altitudes.take(I, mode='clip'))
out_of_range = ~same & ((I == 0) | (I == altitudes.size))
I[out_of_range] = 1 # Prevent index-errors
x1 = altitudes[I-1]
x2 = altitudes[I]
time = np.arange(len(alltemps))
y1 = alltemps[time,I-1]
y2 = alltemps[time,I]
xI = location
yI = y1 + (y2-y1) * (xI-x1) / (x2-x1)
yI[out_of_range] = np.nan
return yI
def scipy_interpn():
time = np.arange(len(alltemps))
xI = np.column_stack((time, location))
yI = interpn((time, altitudes), alltemps, xI)
return yI
N, sigma = 1000., 5
basetemps = 70 + (np.random.randn(N) * sigma)
midtemps = 50 + (np.random.randn(N) * sigma)
toptemps = 40 + (np.random.randn(N) * sigma)
trend = np.sin(4 / N * np.arange(N)) * 30
trend = trend[:, np.newaxis]
alltemps = np.array([basetemps, midtemps, toptemps]).T + trend
altitudes = np.array([500, 1500, 4000], dtype=float)
location = np.linspace(altitudes[0], altitudes[-1], N)
funcs = [original, interp_checked, scipy_interpn]
for func in funcs:
print(func.func_name)
%timeit func()
from itertools import combinations
outs = [func() for func in funcs]
print('Output allclose:')
print([np.allclose(out1, out2) for out1, out2 in combinations(outs, 2)])
我的系统结果如下：
original
10 loops, best of 3: 184 ms per loop
OP_self_answer
10 loops, best of 3: 89.3 ms per loop
interp_checked
1000 loops, best of 3: 224 µs per loop
scipy_interpn
1000 loops, best of 3: 1.36 ms per loop
Output allclose:
[True, True, True, True, True, True]
Scipy 的 interpn 在速度方面有所下降，但就其通用性和易用性而言，它绝对是最佳选择。
对于一个固定的时间点，可以利用下面的插值函数：
g(a) = cc[0]*abs(a-aa[0]) + cc[1]*abs(a-aa[1]) + cc[2]*abs(a-aa[2])
其中 a 是徒步旅行者的高度，aa 具有 3 测量值的向量 altitudes 和 cc 是具有系数的向量。需要注意三点：
- 对于对应于
aa的给定温度 (alltemps)，可以通过使用np.linalg.solve(). 求解线性矩阵方程来确定 g(a)很容易向量化为 (N,) 维a和 (N, 3) 维cc（分别包括np.linalg.solve()）。g(a)称为一阶单变量样条核（三点）。使用abs(a-aa[i])**(2*d-1)会将样条顺序更改为d。这种方法可以解释为 Gaussian Process in Machine Learning. 的简化版本
cc
所以代码是：
import matplotlib.pyplot as plt
import numpy as np
import seaborn as sns
# generate temperatures
np.random.seed(0)
N, sigma = 1000, 5
trend = np.sin(4 / N * np.arange(N)) * 30
alltemps = np.array([tmp0 + trend + sigma*np.random.randn(N)
for tmp0 in [70, 50, 40]])
# generate attitudes:
altitudes = np.array([500, 1500, 4000]).astype(float)
location = np.linspace(altitudes[0], altitudes[-1], N)
def doit():
""" do the interpolation, improved version for speed """
AA = np.vstack([np.abs(altitudes-a_i) for a_i in altitudes])
# This is slighty faster than np.linalg.solve(), because AA is small:
cc = np.dot(np.linalg.inv(AA), alltemps)
return (cc[0]*np.abs(location-altitudes[0]) +
cc[1]*np.abs(location-altitudes[1]) +
cc[2]*np.abs(location-altitudes[2]))
t_loc = doit() # call interpolator
# do the plotting:
fg, ax = plt.subplots(num=1)
for alt, t in zip(altitudes, alltemps):
ax.plot(t, label="%d feet" % alt, alpha=.5)
ax.plot(t_loc, label="Interpolation")
ax.legend(loc="best", title="Altitude:")
ax.set_xlabel("Time")
ax.set_ylabel("Temperature")
fg.canvas.draw()
测量时间得出：
In [2]: %timeit doit()
10000 loops, best of 3: 107 µs per loop
更新：我在doit()中替换了原来的列表理解
导入速度提高 30%（对于 N=1000）。
此外，根据要求进行比较，@moarningsun 在我的机器上的基准代码块：
10 loops, best of 3: 110 ms per loop
interp_checked
10000 loops, best of 3: 83.9 µs per loop
scipy_interpn
1000 loops, best of 3: 678 µs per loop
Output allclose:
[True, True, True]
请注意 N=1000 是一个相对较小的数字。使用 N=100000 产生结果：
interp_checked
100 loops, best of 3: 8.37 ms per loop
%timeit doit()
100 loops, best of 3: 5.31 ms per loop
这表明此方法比 interp_checked 方法更适合大型 N。