基于不同权重随机打乱数组的算法

Question

我有一组想要随机打乱的元素，但每个元素都有不同的优先级或权重。因此，权重较大的元素必须有更多的概率位于结果的顶部。

我有这个数组：

elements = [
  { :id => "ID_1", :weight => 1 },
  { :id => "ID_2", :weight => 2 },
  { :id => "ID_3", :weight => 6 }
]

我想对其进行洗牌，使 id 为 "ID_3" 的元素比元素 "ID_1" 和 ~3 倍 多出 ~6 倍比元素"ID_2" 更多的概率。

 更新

澄清：一旦您选择了第一个位置，其他元素将使用相同的逻辑争夺其余位置。

Answer 1

我可以想到两种方法来解决它，虽然我的直觉告诉我应该修改 Fisher-Yates 以更好地实现它：

O(n*W) 解：（编程简单）

第一种方法，根据权重创建重复项（与您的方法相同），然后填充一个新列表。现在在这个列表上运行一个标准的 shuffle (fisher-yates)。迭代列表并丢弃所有重复项，并仅保留每个元素的第一次出现。这在O(n*W) 中运行，其中n 是列表中元素的数量，W 是平均权重（伪多项式解）。

---

O(nlogn) 解决方案：（显着难以编程）

第二种方法是创建元素权重总和的列表：

sum[i] = weight[0] + ... + weight[i]

现在，在0 到sum[n] 之间画一个数字，并选择sum 大于/等于该随机数的第一个元素。
这将是下一个元素，丢弃该元素，重新创建列表，然后重复。

这在O(n^2*logn) 中运行

可以通过创建二叉树而不是列表来进一步增强，其中每个节点还存储整个子树的权重值。
现在，选择一个元素后，找到匹配的元素（他的总和是第一个比随机选择的数字高的元素），删除节点，并重新计算路径上的权重。
这将需要O(n) 创建树，O(logn) 在每一步找到节点，O(logn) 重新计算总和。重复它直到树用尽，你得到O(nlogn) 解决方案。
这种方法的思想与Order Statistics Trees 非常相似，但使用的是权重之和而不是后代的数量。删除后的查找和平衡与排序统计树类似。

---

二叉树的构造和使用说明。

假设您有elements=[a,b,c,d,e,f,g,h,i,j,k,l,m] 和weights=[1,2,3,1,2,3,1,2,3,1,2,3,1]

首先构造一个几乎完整的二叉树，并填充其中的元素。请注意，树不是二叉搜索树，只是一棵常规树，因此元素的顺序无关紧要 - 我们以后不需要维护它。

你会得到类似下面的树：

图例： w - 该节点的权重，sw - 整个子树的权重之和。

接下来，计算每个子树的权重总和。从叶子开始，计算s.w = w。对于每个其他节点计算s.w = left->s.w + right->s.w，从下往上填充树（post order traversal）。

构建树、填充树并为每个节点计算 s.w. 在 O(n) 中完成。

现在，您需要反复选择一个介于 0 和权重总和之间的随机数（根的 s.w. 值，在我们的例子中为 25）。让这个数字为r，并为每个这样的数字找到匹配的节点。
查找匹配节点是递归完成的

if `r< root.left.sw`:
   go to left son, and repeat. 
else if `r<root.left.sw + root.w`:
   the node you are seeking is the root, choose it. 
else:
   go to `root.right` with `r= r-root.left.sw - root.w`

例如，选择r=10：

Is r<root.left.sw? Yes. Recursively invoke with r=10,root=B (left child)
Is r<root.left.sw No. Is r < root.left.sw + root.w? No. Recursively invoke with r=10-6-2=2, and root=E (right chile)
Is r<root.left.sw? No. Is r < root.left.sw + root.w? Yes. Choose E as next node.

每次迭代都在O(h) = O(logn) 中完成。

现在，您需要删除该节点，并重置树的权重。
一种确保树具有对数权重的删除方法类似于二叉堆：用最右下角的节点替换所选节点，删除新的最右下角节点，然后重新平衡从两个相关节点到的两个分支树。

第一个开关：

然后重新计算：

注意，只需要重新计算两条路径，每个深度最多O(logn)（图中橙色的节点），所以删除和重新计算也是O(logn)。

现在，您为自己获得了一棵新的二叉树，其权重已修改，您可以选择下一个候选者，直到树用尽为止。

Answer 2

我会按如下方式对数组进行洗牌：

代码

def weighted_shuffle(array)
  arr = array.sort_by { |h| -h[:weight] }
  tot_wt = arr.reduce(0) { |t,h| t += h[:weight] }
  ndx_left = arr.each_index.to_a
  arr.size.times.with_object([]) do |_,a|
    cum = 0
    rn = (tot_wt>0) ? rand(tot_wt) : 0
    ndx = ndx_left.find { |i| rn <= (cum += arr[i][:weight]) }
    a << arr[ndx]
    tot_wt -= arr[ndx_left.delete(ndx)][:weight]
  end
end

示例

elements = [
  { :id => "ID_1", :weight => 100 },
  { :id => "ID_2", :weight => 200 },
  { :id => "ID_3", :weight => 600 }
]

def display(arr,n)
  n.times.with_object([]) { |_,a|
    p weighted_shuffle(arr).map { |h| h[:id] } }
end

display(elements,10)
  ["ID_3", "ID_2", "ID_1"]
  ["ID_1", "ID_3", "ID_2"]
  ["ID_1", "ID_3", "ID_2"]
  ["ID_3", "ID_2", "ID_1"]
  ["ID_3", "ID_2", "ID_1"]
  ["ID_2", "ID_3", "ID_1"]
  ["ID_2", "ID_3", "ID_1"]
  ["ID_3", "ID_1", "ID_2"]
  ["ID_3", "ID_1", "ID_2"]
  ["ID_3", "ID_2", "ID_1"]

n = 10_000
pos = elements.each_index.with_object({}) { |i,pos| pos[i] = Hash.new(0) }
n.times { weighted_shuffle(elements).each_with_index { |h,i|
  pos[i][h[:id]] += 1 } }
pos.each { |_,h| h.each_key { |k| h[k] = (h[k]/n.to_f).round(3) } }
  #=> {0=>{"ID_3"=>0.661, "ID_2"=>0.224, "ID_1"=>0.115},
  #    1=>{"ID_2"=>0.472, "ID_3"=>0.278, "ID_1"=>0.251},
  #    2=>{"ID_1"=>0.635, "ID_2"=>0.304, "ID_3"=>0.061}}

这表示，在调用 weighted_shuffle 的 10,000 次中，选择的第一个元素是 66.1% 的时间是“ID_3”，22.4% 的时间是“ID_2”，其余的 11.5% 是“ID_1”时间。 “ID_2”有 47.2% 的次被选中，以此类推。

说明

arr 是要洗牌的哈希数组。随机播放以arr.size 步骤执行。在每个步骤中，我使用提供的权重随机绘制arr 的元素，无需替换。如果h[:weight] 和tot 对于arr 中先前未被选择的所有元素h，则选择这些散列h 中的任何一个的概率为h[:weight]/tot。每一步的选择是通过找到rand(tot) <= p 的第一个累积概率p 来完成的。通过降低权重对element 的元素进行预排序，这最后一步变得更加高效，这是在方法的第一步中完成的：

elements.sort_by { |h| -h[:weight] }
  #=> [{ :id => "ID_3", :weight => 600 },
  #    { :id => "ID_2", :weight => 200 },
  #    { :id => "ID_1", :weight => 100 }]

这是使用arr 的索引数组实现的，称为ndx_left，在该数组上执行迭代。选择索引i 处的哈希h 后，通过减去h[:weight] 来更新tot，并从ndx_left 中删除i。

变体

以下是上述方法的变体：

def weighted_shuffle_variant(array)
   arr = array.sort_by { |h| -h[:weight] }
   tot_wt = arr.reduce(0) { |t,h| t += h[:weight] }
   n = arr.size
   n.times.with_object([]) do |_,a|
     cum = 0
     rn = (tot_wt>0) ? rand(tot_wt) : 0
     h, ndx = arr.each_with_index.find { |h,_| rn <= (cum += h[:weight]) }
     a << h
     tot_wt -= h[:weight]
     arr[ndx] = arr.pop
   end
 end

而不是维护arr 中尚未被选择的元素的索引数组，arr 被原地修改并在选择每个元素时将大小减小一。如果选择了元素arr[i]，则将最后一个元素复制到偏移i，并删除arr的最后一个元素：

arr[i] = arr.pop 

基准测试

复制elementsh[:weight] 次中的每个元素h，然后改组然后uniqifying 结果的方法效率极低。如果这不明显，这里有一个基准。我将我的weighted_shuffle 与@Mori 的解决方案进行了比较，该解决方案代表了“复制、随机播放、删除”方法：

def mori_shuffle(array)
  array.flat_map { |h| [h[:id]] * h[:weight] }.shuffle.uniq
end

require 'benchmark'

def test_em(nelements, ndigits)
  puts "\nelements.size=>#{nelements}, weights have #{ndigits} digits\n\n"
  mx = 10**ndigits
  elements = nelements.times.map { |i| { id: i, weight: rand(mx) } }
  Benchmark.bm(15 "mori_shuffle", "weighted_shuffle") do |x|
    x.report { mori_shuffle(elements) }
    x.report { weighted_shuffle(elements) }
  end
end

elements.size=>3, weights have 1 digits

                      user     system      total        real
mori_shuffle      0.000000   0.000000   0.000000 (  0.000068)
weighted_shuffle  0.000000   0.000000   0.000000 (  0.000051)

elements.size=>3, weights have 2 digits

                      user     system      total        real
mori_shuffle      0.000000   0.000000   0.000000 (  0.000035)
weighted_shuffle  0.010000   0.000000   0.010000 (  0.000026)

elements.size=>3, weights have 3 digits

                      user     system      total        real
mori_shuffle      0.000000   0.000000   0.000000 (  0.000161)
weighted_shuffle  0.000000   0.000000   0.000000 (  0.000027)

elements.size=>3, weights have 4 digits

                      user     system      total        real
mori_shuffle      0.000000   0.000000   0.000000 (  0.000854)
weighted_shuffle  0.000000   0.000000   0.000000 (  0.000026)

elements.size=>20, weights have 2 digits

                      user     system      total        real
mori_shuffle      0.000000   0.000000   0.000000 (  0.000089)
weighted_shuffle  0.000000   0.000000   0.000000 (  0.000090)

elements.size=>20, weights have 3 digits

                      user     system      total        real
mori_shuffle      0.000000   0.000000   0.000000 (  0.000771)
weighted_shuffle  0.000000   0.000000   0.000000 (  0.000071)

elements.size=>20, weights have 4 digits

                      user     system      total        real
mori_shuffle      0.000000   0.000000   0.000000 (  0.005895)
weighted_shuffle  0.000000   0.000000   0.000000 (  0.000073)

elements.size=>100, weights have 2 digits

                      user     system      total        real
mori_shuffle      0.000000   0.000000   0.000000 (  0.000446)
weighted_shuffle  0.000000   0.000000   0.000000 (  0.000683)

elements.size=>100, weights have 3 digits

                      user     system      total        real
mori_shuffle      0.010000   0.000000   0.010000 (  0.003765)
weighted_shuffle  0.000000   0.000000   0.000000 (  0.000659)

elements.size=>100, weights have 4 digits

                      user     system      total        real
mori_shuffle      0.030000   0.010000   0.040000 (  0.034982)
weighted_shuffle  0.000000   0.000000   0.000000 (  0.000638)

elements.size=>100, weights have 5 digits

                      user     system      total        real
mori_shuffle      0.550000   0.040000   0.590000 (  0.593190)
weighted_shuffle  0.000000   0.000000   0.000000 (  0.000623)

elements.size=>100, weights have 6 digits

                      user     system      total        real
mori_shuffle      5.560000   0.380000   5.940000 (  5.944749)
weighted_shuffle  0.010000   0.000000   0.010000 (  0.000636)

weighted_shuffle和weighted_shuffle_variant的比较

考虑到基准引擎已经全部预热，我不妨比较一下我建议的两种方法。结果相似，weighted_shuffle 具有一致的优势。以下是一些典型的结果：

elements.size=>20, weights have 3 digits

                               user     system      total        real
weighted_shuffle           0.000000   0.000000   0.000000 (  0.000062)
weighted_shuffle_variant   0.000000   0.000000   0.000000 (  0.000108)

elements.size=>20, weights have 4 digits

                               user     system      total        real
weighted_shuffle           0.000000   0.000000   0.000000 (  0.000060)
weighted_shuffle_variant   0.000000   0.000000   0.000000 (  0.000089)

elements.size=>100, weights have 2 digits

                               user     system      total        real
weighted_shuffle           0.000000   0.000000   0.000000 (  0.000666)
weighted_shuffle_variant   0.000000   0.000000   0.000000 (  0.000871)

elements.size=>100, weights have 4 digits

                               user     system      total        real
weighted_shuffle           0.000000   0.000000   0.000000 (  0.000625)
weighted_shuffle_variant   0.000000   0.000000   0.000000 (  0.000803)

elements.size=>100, weights have 6 digits

                               user     system      total        real
weighted_shuffle           0.000000   0.000000   0.000000 (  0.000664)
weighted_shuffle_variant   0.000000   0.000000   0.000000 (  0.000773)

与weighted_shuffle 相比，weighted_shuffle_variant 不维护尚未选择的elements（的副本）元素的索引数组（节省时间）。相反，它将数组中的选定元素替换为数组的最后一个元素，然后pops 最后一个元素，导致数组的大小在每一步都减小一。不幸的是，这会通过减少权重来破坏元素的顺序。相比之下，weighted_shuffle 通过降低权重的顺序来保持考虑元素的优化。总的来说，后者的权衡似乎比前者更重要。

Answer 3

基于@amit suggestion：

def self.random_suffle_with_weight(elements, &proc)
  consecutive_chain = []
  elements.each do |element|
    proc.call(element).times { consecutive_chain << element }
  end

  consecutive_chain.shuffle.uniq
end

Answer 4

Weighted Random Sampling (2005; Efraimidis, Spirakis) 为此提供了一个非常优雅的算法。实现超级简单，运行在O(n log(n)):

def weigthed_shuffle(items, weights):
    order = sorted(range(len(items)), key=lambda i: -random.random() ** (1.0 / weights[i]))
    return [items[i] for i in order]

Answer 5

我有我的解决方案，但我认为可以改进：

module Utils
  def self.random_suffle_with_weight(elements, &proc)
    # Create a consecutive chain of element
    # on which every element is represented
    # as many times as its weight.
    consecutive_chain = []
    elements.each do |element|
      proc.call(element).times { consecutive_chain << element }
    end

    # Choosine one element randomly from
    # the consecutive_chain and remove it for the next round
    # until all elements has been chosen.
    shorted_elements = []
    while(shorted_elements.length < elements.length)
      random_index = Kernel.rand(consecutive_chain.length)
      selected_element = consecutive_chain[random_index]
      shorted_elements << selected_element
      consecutive_chain.delete(selected_element)
    end

    shorted_elements
  end
end

测试：

def test_random_suffle_with_weight
  element_1 = { :id => "ID_1", :weight => 10 }
  element_2 = { :id => "ID_2", :weight => 20 }
  element_3 = { :id => "ID_3", :weight => 60 }
  elements = [element_1, element_2, element_3]

  Kernel.expects(:rand).with(90).returns(11)
  Kernel.expects(:rand).with(70).returns(1)
  Kernel.expects(:rand).with(60).returns(50)

  assert_equal([element_2, element_1, element_3], Utils.random_suffle_with_weight(elements) { |e| e[:weight] })
end

Answer 6

elements.flat_map { |h| [h[:id]] * h[:weight] }.shuffle.uniq