向量化还是一种计算元组最长增长子序列的有效方法

Question

使用pandas/python，我想计算每个DTE组的元组的最长递增子序列，但效率很高，有1300万行。现在，使用apply/iteration大约需要10个小时

我的问题大致如下：

^{tb1}$

import pandas as pd
pd.DataFrame({
    'DTE': [1,1,1,1,1,1,2,2,2],
    'Strike': [100,200,300,400,500,600,100,200,300],
    'Bid': [10,16,17,11,12,13,10,15,16],
    'Ask': [11,17,18,12,13,14,30,20,21],
})

我想：

• 按DTE对这些进行分组。这里我们有两个组（DTE1和DTE2）。然后在每组中
• 查找最长的成对递增子序列。排序顺序由Strike确定，这对于每个DTE组都是唯一的。因此，100次罢工之后是200次罢工。
  • 因此，200次罢工的出价和要求必须大于或等于（不严格）100次罢工的出价和要求
  • 任何介于两者之间的罢工都将被删除，因为这两个罢工都没有出价，也没有要求增加价值

在这种情况下，答案是：

^{tb2}$

每个组只保留最长的递增子序列，而不只是任何递增子序列。所有其他行都将被删除

注意O(nlogn)的标准最长递增子序列算法不起作用。请参见https://www.quora.com/How-can-the-SPOJ-problem-LIS2-be-solved了解原因。对于标准O（nlogn）LIS解决方案，示例组DTE 2值将失败。我目前正在使用O（n^2）的标准LIS解决方案。有一个更复杂的O（nlog^2n），但我不认为这是我的瓶颈

由于每一行必须引用前面的行，才能计算出该点上最长的递增子序列，因此您似乎无法并行执行此操作？这意味着你不能矢量化？这是否意味着加速的唯一方法就是使用cython？还是有其他并行解决方案

我当前的解决方案如下所示：

def modify_lsc_row(row, df, longest_lsc):
    lsc_predecessor_count = 0
    lsc_predecessor_index = -1
    df_predecessors = df[(df['Bid'] <= row.Bid) &
                         (df['Ask'] <= row.Ask) &
                         (df['lsc_count'] != -1)]
    if len(df_predecessors) > 0:
        df_predecessors = df_predecessors[(df_predecessors['lsc_count'] == df_predecessors['lsc_count'].max())]
        lsc_predecessor_index = df_predecessors.index.max()
        lsc_predecessor_count = df_predecessors.at[lsc_predecessor_index, 'lsc_count']

    new_predecessor_count = lsc_predecessor_count + 1
    df.at[row.name, 'lsc_count'] = new_predecessor_count
    df.at[row.name, 'prev_index'] = lsc_predecessor_index

    if new_predecessor_count >= longest_lsc.lsc_count:
        longest_lsc.lsc_count = new_predecessor_count
        longest_lsc.lsc_index = row.name

def longest_increasing_bid_ask_subsequence(df):
    original_columns = df.columns
    df.sort_values(['Strike'], ascending=True, inplace=True)

    df.set_index(['Strike'], inplace=True)
    assert df.index.is_unique

    longest_lsc = LongestLsc()
    longest_lsc.lsc_index = df.index.max()
    longest_lsc.lsc_count = 1

    df['lsc_count'] = -1

    df.apply(lambda row: modify_lsc_row(row, df, longest_lsc),
                              axis=1)

    while longest_lsc.lsc_index != -1:
        df.at[longest_lsc.lsc_index, 'keep'] = True
        longest_lsc.lsc_index = df.at[longest_lsc.lsc_index, 'prev_index']

    df.dropna(inplace=True)


    return df.reset_index()[original_columns]


df_groups = df.groupby(['DTE'], group_keys=False, as_index=False)
df_groups.apply(longest_increasing_bid_ask_subsequence)

更新：https://stackoverflow.com/users/15862569/alexander-volkovsky已经提到我可以使用pandarallel来并行每个DTE，因为它们都是独立的。这确实使速度提高了5倍左右。然而，我想把它加快很多（特别是最长的递增子序列的实际优化）。另外，pandarallel似乎无法使用pycharm（这似乎是一个已知的问题https://github.com/nalepae/pandarallel/issues/76）

更新：使用https://stackoverflow.com/users/15862569/alexander-volkovsky建议：即numba，numpy。Pandarallel实际上减慢了速度，因为我的东西越来越快（可能是由于开销）。所以我把它去掉了。10小时->；2.8分钟。这是成功的关键。一些最大的减速是将n^2改为使用numba。也不要使用pandas groupby apply，即使只是用于numba函数。我发现groupby+apply==groupby+pd.concat的时间。您可以使用Alexander所说的删除pd.concat，只需选择最后要保留的行（而不是将所有不同的df组合并在一起）。大量其他小型优化大多是通过使用line profiler发现的

更新代码如下：

@njit
def set_list_indices(bids, asks, indices, indices_to_keep):
    entries = len(indices)

    lis_count = np.full(entries, 0)
    prev_index = np.full(entries, -1)

    longest_lis_count = -1
    longest_lis_index = -1

    for i in range(entries):
        predecessor_counts = np.where((bids <= bids[i]) & (asks <= asks[i]), lis_count, 0)
        best_predecessor_index = len(predecessor_counts) - np.argmax(predecessor_counts[::-1]) - 1

        if best_predecessor_index < i:
            prev_index[i] = best_predecessor_index

        new_count = predecessor_counts[best_predecessor_index] + 1
        lis_count[i] = new_count

        if new_count >= longest_lis_count:
            longest_lis_count = new_count
            longest_lis_index = i

    while longest_lis_index != -1:
        indices_to_keep[indices[longest_lis_index]] = True
        longest_lis_index = prev_index[longest_lis_index]


# necessary for lis algo, and groupby will preserve the order
df = df.sort_values(['Strike'], ascending=True)

# necessary for rows that were dropped. need reindexing for lis algo
df = df.reset_index(drop=True)

df_groups = df.groupby(['DTE'])

row_indices_to_keep = np.full(len(df.index), False, dtype=bool)
for name, group in df_groups:
    bids = group['Bid'].to_numpy()
    asks = group['Ask'].to_numpy()
    indices = group.index.to_numpy()
    set_list_indices(bids, asks, indices, row_indices_to_keep)

df = df.iloc[row_indices_to_keep]

Answer 1

查找最长递增子序列的算法复杂度是多少

This article提供了一个复杂度为O（n logn）的算法。 Upd：不工作。 您甚至不需要修改代码，因为在python中，比较适用于元组：assert (1, 2) < (3, 4)

>>> seq=[(10, 11), (16, 17), (17, 18), (11, 12), (12, 13), (13, 14)]
>>> subsequence(seq)
[(10, 11), (11, 12), (12, 13), (13, 14)]

Since each row must refer to the previous rows to have already computed the longest increasing subsequence at that point, it seems you cannot do this in parallel?

是的，但是您可以为每个DTE并行计算序列。您可以尝试类似于pandarallel的方法在.groupby()之后进行并行聚合

from pandarallel import pandarallel
pandarallel.initialize()

# just an example of usage:
df.groupby("DTE").parallel_apply(subsequence)

还可以尝试摆脱pandas（速度非常慢），使用原始numpy数组和python结构。您可以使用O（n^2）算法计算LIS索引，然后使用df.iloc选择所需的行