处理10**6大小列表的时间复杂性

Question

我最近失败了一个编码挑战，它涉及时间复杂性。在业余时间，我一直在修补它，但仍然不能让它在大清单上迅速发挥作用。我最初考虑过这个问题，重构了它，做了一些渐进式的改进，尝试了使用pandas (结果证明要慢得多)等等。

我有兴趣了解可以使用哪些方法来提高这段代码的执行速度。

Input:在range(1,10**5)中包含未排序整数的最大大小10**6的列表。

任务是从这个任意的构造中计算“总价格”，并返回“总价格”和一个表示未打折的项目的指数的有序列表。

索引i的一个项目的价格按下一个较小/较低的项目贴现。如果items[i+1:]中没有较小的值，则该商品的价格不会打折(或者您可以认为它是由0贴现的)。

示例输入： items = [5, 3, 4, 1, 5]

示例输出： 13, [3, 4]

在这里，items[0]被items[1]贴现，items[1]被items[3]贴现，items[2]被items[3]贴现，items[3]和items[4]不打折。

所以总价格是13，由(5-3) + (3-1) + (4-1) + (1-0) + (5-0)给出。

在大多数情况下，我有一个函数可以非常快地解决这个问题，但是当我们开始接近列表的最大大小时，需要花费更长的时间。例如，长度50000的列表在<1秒内处理。长度为100 K的列表将在<3秒内处理。长度200 K的列表需要<10秒，而400 K大约需要50秒。针对一百万项运行~1000+秒。

为了进行测试，我创建了一个类似于so的大列表，然后将它(或它的切片)传递给函数，如下所示：

data = list(np.array(np.random.randint(1,10**5,(10**6)), dtype='int64'))
total, full_price = get_total(data[:100000])

下面是更快的、非pandas函数：

def get_total(data):
    init_total = sum(data)
    items = data[:] 
    size = len(items)
    discount = [get_discount(items.pop(0),items) for i in range(size)]
    full = [i for (i,v) in enumerate(discount) if v == 0]
    total = init_total - sum(discount)
    return total, full, None

def get_discount(this, _items):
    next_lowest_index, discount = next(((x,val) for x, val in enumerate(_items) if val < this), (np.NaN, 0))
    return discount

我提到我也尝试过pandas，但是即使在小列表(n=1000)上，这段代码也要慢得多。我试着按价值分类：

def frame_total(data):
    if type(data) == list:
        data = pd.DataFrame(data)
    data = data[:].sort_values(0, 'index')
    df = pd.DataFrame({ 'val':data[0],
                        'discount': [0] * data.shape[0]
                        }, dtype='int')
    df.discount = [next(iter(df.loc[(df.index > i) & (df.val < row.val)].sort_index().val),0) 
                   for i,row in df.iterrows()]
    total = data.sum() - df.discount.sum()
    full_indices = list(df[df.discount == 0].sort_index().index)
    return total, full_indices, None

另一种不对输入数据进行排序的是速度不明显的数据：

def frame2(data):
    if type(data) == list:
        data = pd.DataFrame(data)
    data = data[:]
    df = pd.DataFrame({ 'val':data[0],
                        'discount': [0] * data.shape[0]
                        }, dtype='int')
    df.discount = [next(iter(df.val[i+1:].loc[df.val < row.val]),0) for i,row in df.iterrows()]
    total = data.sum() - df.discount.sum()
    full_indices = list(df[df.discount == 0].index)
    return total, full_indices, None

请注意，在列表末尾时，全价商品更有可能存在(随着i的增加，items[i+1:]中存在的任何值< items[i]的可能性都会降低)。我觉得这很重要，但我想不出该如何利用它。

解决了，谢谢@DarrylG和the explanation here

def get_next_smallest(data,default=0):
    """
        returns the discounted value for all items in a list
        discounted value is the next smaller item in the list, e.g.:
        for any n, the next smallest item is the first item in data[n+1:] < data[n]
        provides O(n) complexity solution.
    """
    discounts=[default for i in data] # stores the corresponding next smaller value
    stack = [] # initialize our empty stack
    for i, this in enumerate(data):
        while len(stack) > 0 and this < data[stack[-1]]:
            discounts[stack.pop()] = this
        stack.append(i)
    return discounts

def get_total(data):
    init_total = sum(data)
    default = 0  # should be a value that will NOT be present in the data, like 0 or -1
    discounts = get_next_smallest(data, default)
    full = [i for i,v in enumerate(discounts) if v == default]
    total = init_total - sum(discounts)
    return total, full

Answer 1

按照@Mad物理学家的建议，通过向后迭代您的数据，您可以得到一个需要更少内存的算法，并且速度更快：

def get_total(data):
    tot = sum(data)
    smallest_tail = deque()
    no_discount = []
    i = len(data) - 1 # manually handle the index
    for x in reversed(data):
        while smallest_tail:
            s = smallest_tail[-1]
            if s >= x: # s won't be next smaller for anyone because of x
                smallest_tail.pop()
            else:
                tot -= s
                break
        if not smallest_tail:
            no_discount.append(i)
        smallest_tail.append(x)
        i -= 1
    return tot, list(reversed(no_discount))

与您当前的解决方案(在我的机器上)相比：

:data = list(np.array(np.random.randint(1, 10**5, 10**6, dtype='int64')))
:get_total_dz(data) == get_total(data)
True
:%timeit r = get_total_dz(data) # yours, replacing 'len(stack) > 0' with 'stack'
672 ms ± 6.6 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)
:%timeit r = get_total(data) # mine
435 ms ± 2.29 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)

Answer 2

这里有一个O(N)算法--使用Given an array, find out the next smaller element for each element的算法找到下一个较小的元素

def find_next_smaller_elements(xs):
 " finds next smallest element in O(n) "
    ys=[-1 for x in xs]
    stack=[]
    for i,x in enumerate(xs):
        while len(stack)>0 and x<xs[stack[-1]]:
           ys[stack.pop()]=x
        stack.append(i)
    return ys

def get_total(data):
" Computes desired cost function "
    next_smaller = find_next_smaller_elements(data)

    return sum([ x[0] if x[1] == -1 else x[0]-x[1]  for x in list(zip(data, next_smaller))])

测试(小清单)

data = [5, 3, 4, 1, 5]
print(get_total(data)) # 13

定时测试

for k in [1000, 10000, 100000, 1000000]:
    data = list(np.array(np.random.randint(1,10**5,k, dtype='int64')))
    t0 = time.time()
    ans = get_total(data)
    print(k, time.time()-t0)

结果：

0.0369

• 100000

• No.Items => Time (秒)

• 1000 => 0.0029

• 10000 => 1.96400

因此，一百万个项目在短短2秒内。

Answer 3

这里有一个提示:您可以在一次传递中计算有序的索引。诀窍是沿着列表后退一步：

def find_undiscounted(data):
    skipped = [len(data) - 1]
    current = data[-1]
    for i in range(len(data) - 2, -1, -1):
        if current >= data[i]:
            skipped.append(i)
            current = data[i]
    return skipped[::-1]

一个全面的解决方案将需要一个堆栈，但显然可以在一次通过。如果您决定以这种方式实现collections.deque，请不要忘记使用它。