Sélectionnez k éléments aléatoires dans une liste dont les éléments ont des poids.

Question

La sélection sans pondération (probabilités égales) est magnifiquement décrite. aquí .

Je me demandais s'il existait un moyen de convertir cette approche en une approche pondérée.

Je suis également intéressé par d'autres approches.

Mise à jour : échantillonnage sans remplacement

Answer 1

Si l'échantillonnage est avec remplacement, vous pouvez utiliser cet algorithme (implémenté ici en Python) :

import random

items = [(10, "low"),
         (100, "mid"),
         (890, "large")]

def weighted_sample(items, n):
    total = float(sum(w for w, v in items))
    i = 0
    w, v = items[0]
    while n:
        x = total * (1 - random.random() ** (1.0 / n))
        total -= x
        while x > w:
            x -= w
            i += 1
            w, v = items[i]
        w -= x
        yield v
        n -= 1

C'est O( n + m ) où m est le nombre d'articles.

Pourquoi cela fonctionne-t-il ? Il est basé sur l'algorithme suivant :

def n_random_numbers_decreasing(v, n):
    """Like reversed(sorted(v * random() for i in range(n))),
    but faster because we avoid sorting."""
    while n:
        v *= random.random() ** (1.0 / n)
        yield v
        n -= 1

La fonction weighted_sample est juste cet algorithme fusionné avec une marche de la items pour choisir les éléments sélectionnés par ces nombres aléatoires.

Cela fonctionne à son tour parce que la probabilité que n nombres aléatoires 0.. v seront toutes inférieures à z es P \= ( z/v ) n . Résoudre pour z et vous obtenez z \= vP 1/ n . En substituant un nombre aléatoire à P choisit le plus grand nombre avec la distribution correcte ; et nous pouvons simplement répéter le processus pour sélectionner tous les autres nombres.

Si l'échantillonnage est sans remplacement, vous pouvez placer tous les éléments dans un tas binaire, où chaque nœud met en cache le total des poids de tous les éléments dans ce sous-papier. Construire le tas est O( m ). La sélection d'un élément aléatoire dans le tas, en respectant les poids, est O(log m ). La suppression de cet élément et la mise à jour des totaux mis en cache sont également O(log m ). Vous pouvez donc choisir n en O( m + n journal m ) temps.

(Note : "poids" signifie ici que chaque fois qu'un élément est sélectionné, les autres possibilités sont choisies avec une probabilité proportionnelle à leur poids. Cela ne signifie pas que les éléments apparaissent dans la sortie avec une probabilité proportionnelle à leurs poids).

Voici une implémentation de ce système, abondamment commentée :

import random

class Node:
    # Each node in the heap has a weight, value, and total weight.
    # The total weight, self.tw, is self.w plus the weight of any children.
    __slots__ = ['w', 'v', 'tw']
    def __init__(self, w, v, tw):
        self.w, self.v, self.tw = w, v, tw

def rws_heap(items):
    # h is the heap. It's like a binary tree that lives in an array.
    # It has a Node for each pair in `items`. h[1] is the root. Each
    # other Node h[i] has a parent at h[i>>1]. Each node has up to 2
    # children, h[i<<1] and h[(i<<1)+1].  To get this nice simple
    # arithmetic, we have to leave h[0] vacant.
    h = [None]                          # leave h[0] vacant
    for w, v in items:
        h.append(Node(w, v, w))
    for i in range(len(h) - 1, 1, -1):  # total up the tws
        h[i>>1].tw += h[i].tw           # add h[i]'s total to its parent
    return h

def rws_heap_pop(h):
    gas = h[1].tw * random.random()     # start with a random amount of gas

    i = 1                     # start driving at the root
    while gas >= h[i].w:      # while we have enough gas to get past node i:
        gas -= h[i].w         #   drive past node i
        i <<= 1               #   move to first child
        if gas >= h[i].tw:    #   if we have enough gas:
            gas -= h[i].tw    #     drive past first child and descendants
            i += 1            #     move to second child
    w = h[i].w                # out of gas! h[i] is the selected node.
    v = h[i].v

    h[i].w = 0                # make sure this node isn't chosen again
    while i:                  # fix up total weights
        h[i].tw -= w
        i >>= 1
    return v

def random_weighted_sample_no_replacement(items, n):
    heap = rws_heap(items)              # just make a heap...
    for i in range(n):
        yield rws_heap_pop(heap)        # and pop n items off it.

Answer 2

Si l'échantillonnage est avec remplacement, utilisez le sélection à la roulette (souvent utilisée dans les algorithmes génétiques) :

1. trier les poids
2. calculer les poids cumulés
3. choisir un nombre aléatoire dans [0,1]*totalWeight
4. trouver l'intervalle dans lequel se trouve ce nombre
5. sélectionner les éléments avec l'intervalle correspondant
6. répéter k temps

Si l'échantillonnage est sans remplacement, vous pouvez adapter la technique ci-dessus en retirant l'élément sélectionné de la liste après chaque itération, puis en re-normalisant les poids de façon à ce que leur somme soit égale à 1 (fonction de distribution de probabilité valide).

Answer 3

Je l'ai fait en Ruby

https://github.com/fl00r/pickup

require 'pickup'
pond = {
  "selmon"  => 1,
  "carp" => 4,
  "crucian"  => 3,
  "herring" => 6,
  "sturgeon" => 8,
  "gudgeon" => 10,
  "minnow" => 20
}
pickup = Pickup.new(pond, uniq: true)
pickup.pick(3)
#=> [ "gudgeon", "herring", "minnow" ]
pickup.pick
#=> "herring"
pickup.pick
#=> "gudgeon"
pickup.pick
#=> "sturgeon"

Answer 4

Si vous voulez générer de grands tableaux d'entiers aléatoires avec remplacement vous pouvez utiliser une interpolation linéaire par morceaux. Par exemple, en utilisant NumPy/SciPy :

import numpy
import scipy.interpolate

def weighted_randint(weights, size=None):
    """Given an n-element vector of weights, randomly sample
    integers up to n with probabilities proportional to weights"""
    n = weights.size
    # normalize so that the weights sum to unity
    weights = weights / numpy.linalg.norm(weights, 1)
    # cumulative sum of weights
    cumulative_weights = weights.cumsum()
    # piecewise-linear interpolating function whose domain is
    # the unit interval and whose range is the integers up to n
    f = scipy.interpolate.interp1d(
            numpy.hstack((0.0, weights)),
            numpy.arange(n + 1), kind='linear')
    return f(numpy.random.random(size=size)).astype(int)

Cela n'est pas efficace si vous voulez faire un échantillonnage sans remplacement.

Answer 5

Voici une implémentation de Go de geodns :

package foo

import (
    "log"
    "math/rand"
)

type server struct {
    Weight int
    data   interface{}
}

func foo(servers []server) {
    // servers list is already sorted by the Weight attribute

    // number of items to pick
    max := 4

    result := make([]server, max)

    sum := 0
    for _, r := range servers {
        sum += r.Weight
    }

    for si := 0; si < max; si++ {
        n := rand.Intn(sum + 1)
        s := 0

        for i := range servers {
            s += int(servers[i].Weight)
            if s >= n {
                log.Println("Picked record", i, servers[i])
                sum -= servers[i].Weight
                result[si] = servers[i]

                // remove the server from the list
                servers = append(servers[:i], servers[i+1:]...)
                break
            }
        }
    }

    return result
}