numpy.amax()将在数组中找到最大值，并且numpy.amin()对最小值执行相同的操作。如果我想同时找到最大值和最小值，我必须调用这两个函数，这需要两次传递（非常大的）数组，这似乎很慢。
numpy API 中是否有一个函数可以仅通过一次数据即可找到最大值和最小值？
问问题
84830 次
12 回答
64
numpy API 中是否有一个函数可以仅通过一次数据即可找到最大值和最小值？
没有。在撰写本文时，还没有这样的功能。（是的，如果有这样的函数，它的性能将明显优于在大数组上调用numpy.amin()和连续。）numpy.amax()
于 2015-04-03T17:15:03.353 回答
35
我不认为两次传递数组是一个问题。 考虑以下伪代码：
minval = array[0]
maxval = array[0]
for i in array:
if i < minval:
minval = i
if i > maxval:
maxval = i
虽然这里只有 1 个循环，但仍有 2 个检查。（而不是有 2 个循环，每个循环 1 个检查）。实际上，您唯一节省的就是 1 个循环的开销。如果数组真的像你说的那样大，那么与实际循环的工作负载相比，这个开销很小。（请注意，这都是在 C 中实现的，因此循环或多或少是免费的）。
编辑对你们中的 4 位投了赞成票并对我有信心的人感到抱歉。你绝对可以优化它。
这是一些 fortran 代码，可以通过以下方式编译成 python 模块f2py（也许Cython大师可以过来并将其与优化的 C 版本进行比较......）：
subroutine minmax1(a,n,amin,amax)
implicit none
!f2py intent(hidden) :: n
!f2py intent(out) :: amin,amax
!f2py intent(in) :: a
integer n
real a(n),amin,amax
integer i
amin = a(1)
amax = a(1)
do i=2, n
if(a(i) > amax)then
amax = a(i)
elseif(a(i) < amin) then
amin = a(i)
endif
enddo
end subroutine minmax1
subroutine minmax2(a,n,amin,amax)
implicit none
!f2py intent(hidden) :: n
!f2py intent(out) :: amin,amax
!f2py intent(in) :: a
integer n
real a(n),amin,amax
amin = minval(a)
amax = maxval(a)
end subroutine minmax2
通过以下方式编译它：
f2py -m untitled -c fortran_code.f90
现在我们可以测试它了：
import timeit
size = 100000
repeat = 10000
print timeit.timeit(
'np.min(a); np.max(a)',
setup='import numpy as np; a = np.arange(%d, dtype=np.float32)' % size,
number=repeat), " # numpy min/max"
print timeit.timeit(
'untitled.minmax1(a)',
setup='import numpy as np; import untitled; a = np.arange(%d, dtype=np.float32)' % size,
number=repeat), '# minmax1'
print timeit.timeit(
'untitled.minmax2(a)',
setup='import numpy as np; import untitled; a = np.arange(%d, dtype=np.float32)' % size,
number=repeat), '# minmax2'
结果对我来说有点惊人：
8.61869883537 # numpy min/max
1.60417699814 # minmax1
2.30169081688 # minmax2
我不得不说，我并不完全理解它。比较只是np.min与minmax1并且minmax2仍然是一场失败的战斗，所以这不仅仅是一个记忆问题......
注释- 将大小增加 1 倍，将10**a重复次数减少 1 倍10**a（保持问题大小不变）确实会改变性能，但并不是以一种看似一致的方式，这表明内存性能和函数调用开销之间存在一些相互作用Python。即使min在 fortran 中比较一个简单的实现，也能比 numpy 高出大约 2 倍……
于 2012-08-30T15:44:12.120 回答
31
您可以使用Numba，这是一个使用 LLVM 的 NumPy 感知动态 Python 编译器。结果实现非常简单明了：
import numpy
import numba
@numba.jit
def minmax(x):
maximum = x[0]
minimum = x[0]
for i in x[1:]:
if i > maximum:
maximum = i
elif i < minimum:
minimum = i
return (minimum, maximum)
numpy.random.seed(1)
x = numpy.random.rand(1000000)
print(minmax(x) == (x.min(), x.max()))
它也应该比 Numpy 的min() & max()实现更快。所有这些都无需编写一行 C/Fortran 代码。
做你自己的性能测试，因为它总是依赖于你的架构、你的数据、你的包版本......
于 2015-11-25T14:30:55.330 回答
28
如果对您有用的话，有一个用于查找 (max-min) 的函数，称为numpy.ptp ：
>>> import numpy
>>> x = numpy.array([1,2,3,4,5,6])
>>> x.ptp()
5
但我认为没有办法通过一次遍历找到最小值和最大值。
于 2012-08-30T15:45:27.897 回答
18
考虑到以下方法，只是为了获得一些关于人们可以预期的数字的想法：
import numpy as np
def extrema_np(arr):
return np.max(arr), np.min(arr)
import numba as nb
@nb.jit(nopython=True)
def extrema_loop_nb(arr):
n = arr.size
max_val = min_val = arr[0]
for i in range(1, n):
item = arr[i]
if item > max_val:
max_val = item
elif item < min_val:
min_val = item
return max_val, min_val
import numba as nb
@nb.jit(nopython=True)
def extrema_while_nb(arr):
n = arr.size
odd = n % 2
if not odd:
n -= 1
max_val = min_val = arr[0]
i = 1
while i < n:
x = arr[i]
y = arr[i + 1]
if x > y:
x, y = y, x
min_val = min(x, min_val)
max_val = max(y, max_val)
i += 2
if not odd:
x = arr[n]
min_val = min(x, min_val)
max_val = max(x, max_val)
return max_val, min_val
%%cython -c-O3 -c-march=native -a
#cython: language_level=3, boundscheck=False, wraparound=False, initializedcheck=False, cdivision=True, infer_types=True
import numpy as np
cdef void _extrema_loop_cy(
long[:] arr,
size_t n,
long[:] result):
cdef size_t i
cdef long item, max_val, min_val
max_val = arr[0]
min_val = arr[0]
for i in range(1, n):
item = arr[i]
if item > max_val:
max_val = item
elif item < min_val:
min_val = item
result[0] = max_val
result[1] = min_val
def extrema_loop_cy(arr):
result = np.zeros(2, dtype=arr.dtype)
_extrema_loop_cy(arr, arr.size, result)
return result[0], result[1]
%%cython -c-O3 -c-march=native -a
#cython: language_level=3, boundscheck=False, wraparound=False, initializedcheck=False, cdivision=True, infer_types=True
import numpy as np
cdef void _extrema_while_cy(
long[:] arr,
size_t n,
long[:] result):
cdef size_t i, odd
cdef long x, y, max_val, min_val
max_val = arr[0]
min_val = arr[0]
odd = n % 2
if not odd:
n -= 1
max_val = min_val = arr[0]
i = 1
while i < n:
x = arr[i]
y = arr[i + 1]
if x > y:
x, y = y, x
min_val = min(x, min_val)
max_val = max(y, max_val)
i += 2
if not odd:
x = arr[n]
min_val = min(x, min_val)
max_val = max(x, max_val)
result[0] = max_val
result[1] = min_val
def extrema_while_cy(arr):
result = np.zeros(2, dtype=arr.dtype)
_extrema_while_cy(arr, arr.size, result)
return result[0], result[1]
（这些方法与此处extrema_loop_*()提出的方法相似，而方法基于此处的代码）extrema_while_*()
以下时间安排：
表示extrema_while_*()最快的，extrema_while_nb()最快的。在任何情况下，extrema_loop_nb()andextrema_loop_cy()解决方案也确实优于 NumPy-only 方法（单独使用np.max()和np.min()）。
最后，请注意，这些都不像np.min()/那样灵活np.max()（在 n-dim 支持、axis参数等方面）。
（此处提供完整代码）
于 2020-01-03T23:28:43.540 回答
13
通常，您可以通过一次处理两个元素并仅将较小的元素与临时最小值进行比较，将较大的元素与临时最大值进行比较，从而减少 minmax 算法的比较量。与幼稚的方法相比，平均而言，只需要 3/4 的比较。
这可以用 c 或 fortran（或任何其他低级语言）实现，并且在性能方面几乎是无与伦比的。我正在使用numba来说明原理并获得一个非常快速、独立于 dtype 的实现：
import numba as nb
import numpy as np
@nb.njit
def minmax(array):
# Ravel the array and return early if it's empty
array = array.ravel()
length = array.size
if not length:
return
# We want to process two elements at once so we need
# an even sized array, but we preprocess the first and
# start with the second element, so we want it "odd"
odd = length % 2
if not odd:
length -= 1
# Initialize min and max with the first item
minimum = maximum = array[0]
i = 1
while i < length:
# Get the next two items and swap them if necessary
x = array[i]
y = array[i+1]
if x > y:
x, y = y, x
# Compare the min with the smaller one and the max
# with the bigger one
minimum = min(x, minimum)
maximum = max(y, maximum)
i += 2
# If we had an even sized array we need to compare the
# one remaining item too.
if not odd:
x = array[length]
minimum = min(x, minimum)
maximum = max(x, maximum)
return minimum, maximum
它肯定比Peque提出的幼稚方法更快：
arr = np.random.random(3000000)
assert minmax(arr) == minmax_peque(arr) # warmup and making sure they are identical
%timeit minmax(arr) # 100 loops, best of 3: 2.1 ms per loop
%timeit minmax_peque(arr) # 100 loops, best of 3: 2.75 ms per loop
正如预期的那样，新的 minmax 实现只需要天真的实现所用时间的大约 3/4 ( 2.1 / 2.75 = 0.7636363636363637)
于 2017-01-19T03:20:54.673 回答
12
没有人提到numpy.percentile，所以我想我会的。如果您要求[0, 100]百分位数，它将为您提供一个包含两个元素的数组，即最小值（第 0 个百分位）和最大值（第 100 个百分位）。
但是，它不能满足 OP 的目的：它并不分别比 min 和 max 快。这可能是由于一些允许非极端百分位数的机器（一个更难的问题，应该需要更长的时间）。
In [1]: import numpy
In [2]: a = numpy.random.normal(0, 1, 1000000)
In [3]: %%timeit
...: lo, hi = numpy.amin(a), numpy.amax(a)
...:
100 loops, best of 3: 4.08 ms per loop
In [4]: %%timeit
...: lo, hi = numpy.percentile(a, [0, 100])
...:
100 loops, best of 3: 17.2 ms per loop
In [5]: numpy.__version__
Out[5]: '1.14.4'
Numpy 的未来版本可能会在特殊情况下跳过正常的百分位数计算（如果仅[0, 100]需要）。在不向接口添加任何内容的情况下，有一种方法可以在一次调用中向 Numpy 询问最小值和最大值（与接受的答案中所说的相反），但是库的标准实现并没有利用这种情况来实现它值得。
于 2018-06-15T13:18:37.837 回答
9
这是一个旧线程，但无论如何，如果有人再次看到这个......
同时查找最小值和最大值时，可以减少比较次数。如果您正在比较的是浮点数（我猜是这样），这可能会为您节省一些时间，尽管不是计算复杂性。
而不是（Python代码）：
_max = ar[0]
_min= ar[0]
for ii in xrange(len(ar)):
if _max > ar[ii]: _max = ar[ii]
if _min < ar[ii]: _min = ar[ii]
您可以先比较数组中的两个相邻值，然后仅将较小的值与当前最小值进行比较，将较大的值与当前最大值进行比较：
## for an even-sized array
_max = ar[0]
_min = ar[0]
for ii in xrange(0, len(ar), 2)): ## iterate over every other value in the array
f1 = ar[ii]
f2 = ar[ii+1]
if (f1 < f2):
if f1 < _min: _min = f1
if f2 > _max: _max = f2
else:
if f2 < _min: _min = f2
if f1 > _max: _max = f1
这里的代码是用 Python 编写的，显然为了提高速度，您可以使用 C 或 Fortran 或 Cython，但这样每次迭代进行 3 次比较，使用 len(ar)/2 次迭代，给出 3/2 * len(ar) 比较。与此相反，以“显而易见的方式”进行比较，每次迭代进行两次比较，导致 2*len(ar) 比较。为您节省 25% 的比较时间。
也许有一天有人会发现这很有用。
于 2013-11-13T13:19:20.077 回答
6
乍一看，似乎可以解决问题：numpy.histogram
count, (amin, amax) = numpy.histogram(a, bins=1)
...但是如果您查看该函数的源代码a.min()，它只是独立调用，a.max()因此无法避免此问题中解决的性能问题。:-(
同样，scipy.ndimage.measurements.extrema看起来是一种可能性，但它也只是简单地调用a.min()和a.max()独立。
于 2014-06-03T17:35:06.480 回答
3
无论如何，这对我来说是值得的，所以我会在这里为可能感兴趣的人提出最困难和最不优雅的解决方案。我的解决方案是在 C++ 中实现一个多线程的 min-max in one pass 算法，并使用它来创建一个 Python 扩展模块。这项工作需要一些开销来学习如何使用 Python 和 NumPy C/C++ API，在这里我将展示代码并为希望走这条路的人提供一些小的解释和参考。
多线程最小值/最大值
这里没有什么太有趣的了。数组被分成大小的块length / workers。为 a 中的每个块计算 min/max future，然后扫描全局 min/max。
// mt_np.cc
//
// multi-threaded min/max algorithm
#include <algorithm>
#include <future>
#include <vector>
namespace mt_np {
/*
* Get {min,max} in interval [begin,end)
*/
template <typename T> std::pair<T, T> min_max(T *begin, T *end) {
T min{*begin};
T max{*begin};
while (++begin < end) {
if (*begin < min) {
min = *begin;
continue;
} else if (*begin > max) {
max = *begin;
}
}
return {min, max};
}
/*
* get {min,max} in interval [begin,end) using #workers for concurrency
*/
template <typename T>
std::pair<T, T> min_max_mt(T *begin, T *end, int workers) {
const long int chunk_size = std::max((end - begin) / workers, 1l);
std::vector<std::future<std::pair<T, T>>> min_maxes;
// fire up the workers
while (begin < end) {
T *next = std::min(end, begin + chunk_size);
min_maxes.push_back(std::async(min_max<T>, begin, next));
begin = next;
}
// retrieve the results
auto min_max_it = min_maxes.begin();
auto v{min_max_it->get()};
T min{v.first};
T max{v.second};
while (++min_max_it != min_maxes.end()) {
v = min_max_it->get();
min = std::min(min, v.first);
max = std::max(max, v.second);
}
return {min, max};
}
}; // namespace mt_np
Python 扩展模块
这就是事情开始变得丑陋的地方......在Python中使用C++代码的一种方法是实现一个扩展模块。该模块可以使用distutils.core标准模块构建和安装。Python 文档中涵盖了这方面的完整描述：https ://docs.python.org/3/extending/extending.html 。 注意：当然还有其他方法可以获得类似的结果，引用https://docs.python.org/3/extending/index.html#extending-index：
本指南仅涵盖作为此版本 CPython 的一部分提供的用于创建扩展的基本工具。Cython、cffi、SWIG 和 Numba 等第三方工具提供了更简单和更复杂的方法来为 Python 创建 C 和 C++ 扩展。
从本质上讲，这条路线可能更具学术性而非实用性。话虽如此，我接下来要做的是，非常接近教程，创建一个模块文件。这本质上是 distutils 知道如何处理您的代码并从中创建 Python 模块的样板。在做任何这些之前，创建一个 Python虚拟环境可能是明智的，这样您就不会污染您的系统包（请参阅https://docs.python.org/3/library/venv.html#module-venv）。
这是模块文件：
// mt_np_forpy.cc
//
// C++ module implementation for multi-threaded min/max for np
#define NPY_NO_DEPRECATED_API NPY_1_7_API_VERSION
#include <python3.6/numpy/arrayobject.h>
#include "mt_np.h"
#include <cstdint>
#include <iostream>
using namespace std;
/*
* check:
* shape
* stride
* data_type
* byteorder
* alignment
*/
static bool check_array(PyArrayObject *arr) {
if (PyArray_NDIM(arr) != 1) {
PyErr_SetString(PyExc_RuntimeError, "Wrong shape, require (1,n)");
return false;
}
if (PyArray_STRIDES(arr)[0] != 8) {
PyErr_SetString(PyExc_RuntimeError, "Expected stride of 8");
return false;
}
PyArray_Descr *descr = PyArray_DESCR(arr);
if (descr->type != NPY_LONGLTR && descr->type != NPY_DOUBLELTR) {
PyErr_SetString(PyExc_RuntimeError, "Wrong type, require l or d");
return false;
}
if (descr->byteorder != '=') {
PyErr_SetString(PyExc_RuntimeError, "Expected native byteorder");
return false;
}
if (descr->alignment != 8) {
cerr << "alignment: " << descr->alignment << endl;
PyErr_SetString(PyExc_RuntimeError, "Require proper alignement");
return false;
}
return true;
}
template <typename T>
static PyObject *mt_np_minmax_dispatch(PyArrayObject *arr) {
npy_intp size = PyArray_SHAPE(arr)[0];
T *begin = (T *)PyArray_DATA(arr);
auto minmax =
mt_np::min_max_mt(begin, begin + size, thread::hardware_concurrency());
return Py_BuildValue("(L,L)", minmax.first, minmax.second);
}
static PyObject *mt_np_minmax(PyObject *self, PyObject *args) {
PyArrayObject *arr;
if (!PyArg_ParseTuple(args, "O", &arr))
return NULL;
if (!check_array(arr))
return NULL;
switch (PyArray_DESCR(arr)->type) {
case NPY_LONGLTR: {
return mt_np_minmax_dispatch<int64_t>(arr);
} break;
case NPY_DOUBLELTR: {
return mt_np_minmax_dispatch<double>(arr);
} break;
default: {
PyErr_SetString(PyExc_RuntimeError, "Unknown error");
return NULL;
}
}
}
static PyObject *get_concurrency(PyObject *self, PyObject *args) {
return Py_BuildValue("I", thread::hardware_concurrency());
}
static PyMethodDef mt_np_Methods[] = {
{"mt_np_minmax", mt_np_minmax, METH_VARARGS, "multi-threaded np min/max"},
{"get_concurrency", get_concurrency, METH_VARARGS,
"retrieve thread::hardware_concurrency()"},
{NULL, NULL, 0, NULL} /* sentinel */
};
static struct PyModuleDef mt_np_module = {PyModuleDef_HEAD_INIT, "mt_np", NULL,
-1, mt_np_Methods};
PyMODINIT_FUNC PyInit_mt_np() { return PyModule_Create(&mt_np_module); }
在此文件中，大量使用了 Python 和 NumPy API，有关更多信息，请参阅：https : //docs.python.org/3/c-api/arg.html#c.PyArg_ParseTuple 和 NumPy ：https ://docs.scipy.org/doc/numpy/reference/c-api.array.html 。
安装模块
接下来要做的是利用 distutils 安装模块。这需要一个安装文件：
# setup.py
from distutils.core import setup,Extension
module = Extension('mt_np', sources = ['mt_np_module.cc'])
setup (name = 'mt_np',
version = '1.0',
description = 'multi-threaded min/max for np arrays',
ext_modules = [module])
要最终安装模块，python3 setup.py install请从您的虚拟环境中执行。
测试模块
最后，我们可以测试一下 C++ 实现是否真的优于 NumPy 的幼稚使用。为此，这是一个简单的测试脚本：
# timing.py
# compare numpy min/max vs multi-threaded min/max
import numpy as np
import mt_np
import timeit
def normal_min_max(X):
return (np.min(X),np.max(X))
print(mt_np.get_concurrency())
for ssize in np.logspace(3,8,6):
size = int(ssize)
print('********************')
print('sample size:', size)
print('********************')
samples = np.random.normal(0,50,(2,size))
for sample in samples:
print('np:', timeit.timeit('normal_min_max(sample)',
globals=globals(),number=10))
print('mt:', timeit.timeit('mt_np.mt_np_minmax(sample)',
globals=globals(),number=10))
以下是我做这一切的结果：
8
********************
sample size: 1000
********************
np: 0.00012079699808964506
mt: 0.002468645994667895
np: 0.00011947099847020581
mt: 0.0020772050047526136
********************
sample size: 10000
********************
np: 0.00024697799381101504
mt: 0.002037393998762127
np: 0.0002713389985729009
mt: 0.0020942929986631498
********************
sample size: 100000
********************
np: 0.0007130410012905486
mt: 0.0019842900001094677
np: 0.0007540129954577424
mt: 0.0029724110063398257
********************
sample size: 1000000
********************
np: 0.0094779249993735
mt: 0.007134920000680722
np: 0.009129883001151029
mt: 0.012836456997320056
********************
sample size: 10000000
********************
np: 0.09471094200125663
mt: 0.0453535050037317
np: 0.09436299200024223
mt: 0.04188535599678289
********************
sample size: 100000000
********************
np: 0.9537652180006262
mt: 0.3957935369980987
np: 0.9624398809974082
mt: 0.4019058070043684
这些远没有线程中的结果显示的那么令人鼓舞，结果表明速度提高了大约 3.5 倍，并且没有包含多线程。我取得的结果在某种程度上是合理的，我预计线程的开销将占主导地位，直到数组变得非常大，此时性能提升将开始接近std::thread::hardware_concurrencyx 增加。
结论
对某些 NumPy 代码的应用程序特定优化肯定有空间，特别是在多线程方面。我不清楚这是否值得努力，但它确实似乎是一个很好的练习（或其他东西）。我认为也许学习一些像 Cython 这样的“第三方工具”可能会更好地利用时间，但谁知道呢。
于 2019-09-21T19:43:36.020 回答
2
受先前答案的启发，我编写了 numba 实现，从二维数组返回轴 = 0 的 minmax。它比调用 numpy min/max 快约 5 倍。也许有人会发现它很有用。
from numba import jit
@jit
def minmax(x):
"""Return minimum and maximum from 2D array for axis=0."""
m, n = len(x), len(x[0])
mi, ma = np.empty(n), np.empty(n)
mi[:] = ma[:] = x[0]
for i in range(1, m):
for j in range(n):
if x[i, j]>ma[j]: ma[j] = x[i, j]
elif x[i, j]<mi[j]: mi[j] = x[i, j]
return mi, ma
x = np.random.normal(size=(256, 11))
mi, ma = minmax(x)
np.all(mi == x.min(axis=0)), np.all(ma == x.max(axis=0))
# (True, True)
%timeit x.min(axis=0), x.max(axis=0)
# 15.9 µs ± 9.4 ns per loop (mean ± std. dev. of 7 runs, 100000 loops each)
%timeit minmax(x)
# 2.62 µs ± 31.3 ns per loop (mean ± std. dev. of 7 runs, 100000 loops each)
于 2020-08-28T12:37:34.667 回答
0
我想出的最短方法是：
mn, mx = np.sort(ar)[[0, -1]]
但是由于它对数组进行排序，所以它不是最有效的。
另一个简短的方法是：
mn, mx = np.percentile(ar, [0, 100])
这应该更有效，但是计算结果，并返回一个浮点数。
于 2019-11-26T14:11:26.893 回答