我对问题的理解:给定k 集合Y 和m 集合X 的集合,我们希望找到Y 的子集S,这样对于所有y in S,存在x in X st x 是 y 的子集。
我将假设集合由n-向量表示,零向量和表示包含的向量。这是设置:
import pandas as pd # for drop_duplicates & benchmarks
import numpy as np
np.random.seed(0)
n = 100 # 100 "atomic" elements
m = 1000 # small sets
k = 1000 # large sets
X = pd.DataFrame(np.random.randint(0, 2, size=(m, n))).drop_duplicates().values
Y = pd.DataFrame(np.random.randint(0, 2, size=(k, n))).drop_duplicates().values
# For each row y in Y, we would like to check if there exists a row x in X
# s.t. x represents a subset of Y
def naive(Y, X):
# O(k^2 + m^2)
for i, y in enumerate(Y):
for x in X:
found_subset = False
if (x <= y).all():
yield i
found_subset = True
if found_subset:
break
def naive_as_array(Y, X):
return np.array(list(naive(Y, X)))
naive 函数迭代所有可能满足包含关系的集合对,并在适当的时候进行短路。运行时是O(m + k),其中m = len(combs)。
作为替代方案,我们可以考虑使用以下递归算法一次处理每个元素(从1 到n):
def contains(Y, X):
"""
Y : k x n indicator array specifying sets
X : m x n indicator array specifying sets
output: subset Z of [0..k-1] s.t. i in Z iff there exists x in X s.t.
# x is a subset of Y[i]. Z is represented by a 1D numpy array.
"""
k, n = Y.shape
assert Y.shape[1] == X.shape[1]
detected = np.zeros(k, dtype=np.bool)
inds = np.arange(k)
# utility to account for sets that already have a subset detected
def account_for_detected(Y, inds):
mask = ~detected[inds]
Y = Y[mask]
inds = inds[mask]
return Y, inds
# inductively reduce Y.shape[1] (==X.shape[1])
def f(Y, X, inds):
if Y.shape[0] == 0 or X.shape[0] == 0:
# collection Y is empty; inculsions are impossible
return
# avoid redundant comparisons by dropping sets y in Y
# if it is already known that y contains some element of X
Y, inds = account_for_detected(Y, inds)
if Y.shape[1] == 1:
# Y and X are collections of singletons
Y = np.ravel(Y)
X = np.ravel(X)
X_vals = np.zeros(2, dtype=np.int)
X_vals[X] = 1
if X_vals[0] > 0:
detected[inds] = True
elif X_vals[1] > 0:
detected[inds[Y==1]] = True
return
else:
# make a recursive call
Ymask = Y[:,0] == 0
Xmask = X[:,0] == 0
# if x in X is a subset of y in Y, x[0] <= y[0]
f(Y[Ymask,1:], X[Xmask,1:], inds[Ymask])
# by now, detected is updated in the outer scope
# process the remaining Y's
f(Y[~Ymask,1:], X[:,1:], inds[~Ymask])
# done
# make call at root:
f(Y, X, inds)
# return indices
return np.where(detected)[0]
在1和n之间的步骤d,我们将集合Y拆分为Y0和Y1,其中Y0包含Y中不包含元素d的集合, Y1 包含 Y 中的集合,这些集合确实包含元素 d。同样,我们定义X0 和X1。一个关键的观察是X1 中的集合不能作为Y0 中的集合的子集出现。因此,我们可以减少递归调用中的比较次数。
时间安排:
%timeit contains(Y, X)
%timeit naive_as_array(Y, X)
10 loops, best of 3: 185 ms per loop
1 loop, best of 3: 2.39 s per loop