(已编辑以包含不同且更准确的方法)
应该注意,并不是所有的混洗分数p(混洗元素的数量除以元素的总数)的值都是可访问的。
p 的可能值取决于输入的大小和重复元素的数量。
也就是说,我可以提出两种可能的方法:
- 将您的输入分成正确大小的 pinned 和 unpinned 索引,然后打乱 unpinned 索引。
import numpy as np
def partial_shuffle(arr, p=1.0):
n = arr.size
k = round(n * p)
shuffling = np.arange(n)
shuffled = np.random.choice(n, k, replace=False)
shuffling[shuffled] = np.sort(shuffled)
return arr[shuffling]
方法 (1) 的主要优点是可以使用np.random.choice() 和高级索引以矢量化形式轻松实现。
另一方面,只要您愿意接受某些洗牌可能会因为重复值或仅仅因为洗牌索引与未洗牌的索引意外重合而返回一些未洗牌的元素,则此方法效果很好。
这会导致p 的请求 值通常大于观察到的实际 值。
如果需要一个相对更准确的 p 值,可以尝试对 p 参数执行搜索,在输出中给出所需的值,或者通过反复试验。
- 实现Fisher-Yates shuffle 的变体,您可以:(a) 拒绝交换值相同的头寸,(b) 只选择尚未访问过的随机头寸进行交换。
def partial_shuffle_eff(arr, p=1.0, inplace=False, tries=2.0):
if not inplace:
arr = arr.copy()
n = arr.size
k = round(n * p)
tries = round(n * tries)
seen = set()
i = l = t = 0
while i < n and l < k:
seen.add(i)
j = np.random.randint(i, n)
while j in seen and t < tries:
j = np.random.randint(i, n)
t += 1
if arr[i] != arr[j]:
arr[i], arr[j] = arr[j], arr[i]
l += 2
seen.add(j)
while i in seen:
i += 1
return arr
虽然这种方法可以得到更准确的值p,但它仍然受到目标交换次数必须是偶数的限制。
此外,对于具有大量唯一性的输入,第二个 while (while j in seen ...) 可能是一个无限循环,因此应设置尝试次数上限。
最后,您需要使用显式循环,这会导致执行速度慢得多,除非您可以使用 Numba 的 JIT 编译,这会显着加快您的执行速度。
import numba as nb
partial_shuffle_eff_nb = nb.njit(partial_shuffle_eff)
partial_shuffle_eff_nb.__name__ = 'partial_shuffle_eff_nb'
为了测试部分改组的准确性,我们可以使用(百分比)Hamming distance:
def hamming_distance(a, b):
assert(a.shape == b.shape)
return np.count_nonzero(a == b)
def percent_hamming_distance(a, b):
return hamming_distance(a, b) / len(a)
def shuffling_fraction(a, b):
return 1 - percent_hamming_distance(a, b)
我们可能会观察到类似这样的行为:
funcs = (
partial_shuffle,
partial_shuffle_eff,
partial_shuffle_eff_nb
)
n = 12
m = 3
arrs = (
np.zeros(n, dtype=int),
np.arange(n),
np.repeat(np.arange(m), n // m),
np.repeat(np.arange(3), 2),
np.repeat(np.arange(3), 3),
)
np.random.seed(0)
for arr in arrs:
print(" " * 24, arr)
for func in funcs:
shuffled = func(arr, 0.5)
print(f"{func.__name__:>24s}", shuffled, shuffling_fraction(arr, shuffled))
# [0 0 0 0 0 0 0 0 0 0 0 0]
# partial_shuffle [0 0 0 0 0 0 0 0 0 0 0 0] 0.0
# partial_shuffle_eff [0 0 0 0 0 0 0 0 0 0 0 0] 0.0
# partial_shuffle_eff_nb [0 0 0 0 0 0 0 0 0 0 0 0] 0.0
# [ 0 1 2 3 4 5 6 7 8 9 10 11]
# partial_shuffle [ 0 8 2 3 6 5 7 4 9 1 10 11] 0.5
# partial_shuffle_eff [ 3 8 11 0 4 5 6 7 1 9 10 2] 0.5
# partial_shuffle_eff_nb [ 9 10 11 3 4 5 6 7 8 0 1 2] 0.5
# [0 0 0 0 1 1 1 1 2 2 2 2]
# partial_shuffle [0 0 2 0 1 2 1 1 2 2 1 0] 0.33333333333333337
# partial_shuffle_eff [1 1 1 0 0 1 0 0 2 2 2 2] 0.5
# partial_shuffle_eff_nb [1 2 1 0 1 0 0 1 0 2 2 2] 0.5
# [0 0 1 1 2 2]
# partial_shuffle [0 0 1 1 2 2] 0.0
# partial_shuffle_eff [1 1 0 0 2 2] 0.6666666666666667
# partial_shuffle_eff_nb [1 2 0 1 0 2] 0.6666666666666667
# [0 0 0 1 1 1 2 2 2]
# partial_shuffle [0 0 1 1 0 1 2 2 2] 0.2222222222222222
# partial_shuffle_eff [0 1 2 1 0 1 2 2 0] 0.4444444444444444
# partial_shuffle_eff_nb [0 0 1 0 2 1 2 1 2] 0.4444444444444444
或者,对于更接近您的用例的输入:
n = 4500
m = 3
arr = np.repeat(np.arange(m), n // m)
np.random.seed(0)
for func in funcs:
shuffled = func(arr, 0.5)
print(f"{func.__name__:>24s}", shuffling_fraction(arr, shuffled))
# partial_shuffle 0.33777777777777773
# partial_shuffle_eff 0.5
# partial_shuffle_eff_nb 0.5
最后是一些小的基准测试:
n = 4500
m = 3
arr = np.repeat(np.arange(m), n // m)
np.random.seed(0)
for func in funcs:
print(f"{func.__name__:>24s}", end=" ")
%timeit func(arr, 0.5)
# partial_shuffle 213 µs ± 6.36 µs per loop (mean ± std. dev. of 7 runs, 1000 loops each)
# partial_shuffle_eff 10.9 ms ± 194 µs per loop (mean ± std. dev. of 7 runs, 100 loops each)
# partial_shuffle_eff_nb 172 µs ± 1.79 µs per loop (mean ± std. dev. of 7 runs, 10000 loops each)