Распределение вероятностей в Python - PullRequest
20 голосов
/ 08 февраля 2009

У меня есть набор ключей, каждый из которых имеет переменную неправдоподобия. Я хочу случайным образом выбрать один из этих ключей, но я хочу, чтобы маловероятно, что (ключ, значения) будет более вероятным, чем менее вероятный (более вероятный) объект. Мне интересно, есть ли у вас какие-либо предложения, желательно существующий модуль python, который я мог бы использовать, иначе мне нужно будет сделать это самому.

Я проверил случайный модуль; это, кажется, не обеспечивает это.

Я должен сделать такой выбор много миллионов раз для 1000 различных наборов объектов, каждый из которых содержит 2455 объектов. Каждый набор будет обмениваться объектами между собой, поэтому случайный выбор должен быть динамическим. С 1000 наборов из 2433 объектов, то есть 2,433 миллиона объектов; низкое потребление памяти имеет решающее значение. И поскольку эти варианты не являются основной частью алгоритма, мне нужно, чтобы этот процесс был достаточно быстрым; Время процессора ограничено.

Thx

Обновление:

Хорошо, я пытался продуманно рассмотреть ваши предложения, но время ограничено ...

Я посмотрел на подход бинарного дерева поиска, и он кажется слишком рискованным (сложным и сложным). Все остальные предложения напоминают рецепт ActiveState. Я взял его и немного изменил в надежде сделать его более эффективным:

def windex(dict, sum, max):
    '''an attempt to make a random.choose() function that makes
    weighted choices accepts a dictionary with the item_key and
    certainty_value as a pair like:
    >>> x = [('one', 20), ('two', 2), ('three', 50)], the
    maximum certainty value (max) and the sum of all certainties.'''
    n = random.uniform(0, 1)
    sum = max*len(list)-sum 
    for key, certainty in dict.iteritems():
        weight = float(max-certainty)/sum
        if n < weight:
            break
        n = n - weight
    return key

Я надеюсь получить выигрыш в эффективности от динамического поддержания суммы определений и максимальной уверенности. Любые дальнейшие предложения приветствуются. Вы, ребята, экономите мне столько времени и усилий, но при этом повышаете мою эффективность, это безумие. Спасибо! Спасибо! Thx!

Update2:

Я решил сделать его более эффективным, позволив ему выбирать больше вариантов одновременно. Это приведет к приемлемой потере точности в моем алгоритме, поскольку он носит динамический характер. Во всяком случае, вот что у меня сейчас:

def weightedChoices(dict, sum, max, choices=10):
    '''an attempt to make a random.choose() function that makes
    weighted choices accepts a dictionary with the item_key and
    certainty_value as a pair like:
    >>> x = [('one', 20), ('two', 2), ('three', 50)], the
    maximum certainty value (max) and the sum of all certainties.'''
    list = [random.uniform(0, 1) for i in range(choices)]
    (n, list) = relavate(list.sort())
    keys = []
    sum = max*len(list)-sum 
    for key, certainty in dict.iteritems():
        weight = float(max-certainty)/sum
        if n < weight:
            keys.append(key)
            if list: (n, list) = relavate(list)
            else: break
        n = n - weight
    return keys
def relavate(list):
    min = list[0]
    new = [l - min for l in list[1:]]
    return (min, new)

Я еще не пробовал. Если у вас есть какие-либо комментарии / предложения, пожалуйста, не стесняйтесь. Thx!

Update3:

Я целый день работал над заданной версией ответа Рекса Логана. Вместо 2-х массивов объектов и весов, это фактически специальный класс словаря; что делает вещи довольно сложными, так как код Рекса генерирует случайный индекс ... Я также закодировал тестовый пример, который напоминает то, что произойдет в моем алгоритме (но я не могу знать, пока не попробую!). Основной принцип: чем чаще ключ генерируется случайным образом, тем менее вероятно, что он будет сгенерирован снова:

import random, time
import psyco
psyco.full()

class ProbDict():
    """
    Modified version of Rex Logans RandomObject class. The more a key is randomly
    chosen, the more unlikely it will further be randomly chosen. 
    """
    def __init__(self,keys_weights_values={}):
        self._kw=keys_weights_values
        self._keys=self._kw.keys()
        self._len=len(self._keys)
        self._findSeniors()
        self._effort = 0.15
        self._fails = 0
    def __iter__(self):
        return self.next()
    def __getitem__(self, key):
        return self._kw[key]
    def __setitem__(self, key, value):
        self.append(key, value)
    def __len__(self):
        return self._len
    def next(self):
        key=self._key()
        while key:
            yield key
            key = self._key()
    def __contains__(self, key):
        return key in self._kw
    def items(self):
        return self._kw.items()
    def pop(self, key):  
        try:
            (w, value) = self._kw.pop(key)
            self._len -=1
            if w == self._seniorW:
                self._seniors -= 1
                if not self._seniors:
                    #costly but unlikely:
                    self._findSeniors()
            return [w, value]
        except KeyError:
            return None
    def popitem(self):
        return self.pop(self._key())
    def values(self):
        values = []
        for key in self._keys:
            try:
                values.append(self._kw[key][1])
            except KeyError:
                pass
        return values
    def weights(self):
        weights = []
        for key in self._keys:
            try:
                weights.append(self._kw[key][0])
            except KeyError:
                pass
        return weights
    def keys(self, imperfect=False):
        if imperfect: return self._keys
        return self._kw.keys()
    def append(self, key, value=None):
        if key not in self._kw:
            self._len +=1
            self._kw[key] = [0, value]
            self._keys.append(key)
        else:
            self._kw[key][1]=value
    def _key(self):
        for i in range(int(self._effort*self._len)):
            ri=random.randint(0,self._len-1) #choose a random object
            rx=random.uniform(0,self._seniorW)
            rkey = self._keys[ri]
            try:
                w = self._kw[rkey][0]
                if rx >= w: # test to see if that is the value we want
                    w += 1
                    self._warnSeniors(w)
                    self._kw[rkey][0] = w
                    return rkey
            except KeyError:
                self._keys.pop(ri)
        # if you do not find one after 100 tries then just get a random one
        self._fails += 1 #for confirming effectiveness only
        for key in self._keys:
            if key in self._kw:
                w = self._kw[key][0] + 1
                self._warnSeniors(w)
                self._kw[key][0] = w
                return key
        return None
    def _findSeniors(self):
        '''this function finds the seniors, counts them and assess their age. It
        is costly but unlikely.'''
        seniorW = 0
        seniors = 0
        for w in self._kw.itervalues():
            if w >= seniorW:
                if w == seniorW:
                    seniors += 1
                else:
                    seniorsW = w
                    seniors = 1
        self._seniors = seniors
        self._seniorW = seniorW
    def _warnSeniors(self, w):
        #a weight can only be incremented...good
        if w >= self._seniorW:
            if w == self._seniorW:
                self._seniors+=1
            else:
                self._seniors = 1
                self._seniorW = w
def test():
    #test code
    iterations = 200000
    size = 2500
    nextkey = size 


    pd = ProbDict(dict([(i,[0,i]) for i in xrange(size)]))
    start = time.clock()
    for i in xrange(iterations):
        key=pd._key()
        w=pd[key][0]
        if random.randint(0,1+pd._seniorW-w):
            #the heavier the object, the more unlikely it will be removed
            pd.pop(key)
        probAppend = float(500+(size-len(pd)))/1000
        if random.uniform(0,1) < probAppend:
            nextkey+=1
            pd.append(nextkey)
    print (time.clock()-start)*1000/iterations, "msecs / iteration with", pd._fails, "failures /", iterations, "iterations"
    weights = pd.weights()
    weights.sort()
    print "avg weight:", float(sum(weights))/pd._len, max(weights), pd._seniorW, pd._seniors, len(pd), len(weights)
    print weights
test()

Любые комментарии по-прежнему приветствуются. @Darius: ваши двоичные деревья слишком сложны для меня; и я не думаю, что его листья могут быть удалены эффективно ... Thx все

Ответы [ 12 ]

26 голосов
/ 08 февраля 2009

Этот рецепт activestate дает простой для понимания подход, особенно версию в комментариях, которая не требует предварительной нормализации ваших весов:

import random

def weighted_choice(items):
    """items is a list of tuples in the form (item, weight)"""
    weight_total = sum((item[1] for item in items))
    n = random.uniform(0, weight_total)
    for item, weight in items:
        if n < weight:
            return item
        n = n - weight
    return item

Это будет медленно, если у вас большой список предметов. Бинарный поиск, вероятно, был бы лучше в этом случае ... но также был бы более сложным для записи, для небольшого выигрыша, если у вас небольшой размер выборки. Вот пример подхода двоичного поиска в python , если вы хотите следовать по этому маршруту.

(Я бы порекомендовал провести быстрое тестирование производительности обоих методов в вашем наборе данных. Производительность различных подходов к этому алгоритму часто немного неинтуитивна.)


Редактировать: Я воспользовался собственным советом, так как мне было любопытно, и сделал несколько тестов.

Я сравнил четыре подхода:

Функция weighted_choice выше.

Функция выбора бинарного поиска, например, так:

def weighted_choice_bisect(items):
    added_weights = []
    last_sum = 0

    for item, weight in items:
        last_sum += weight
        added_weights.append(last_sum)

    return items[bisect.bisect(added_weights, random.random() * last_sum)][0]

Версия компиляции 1:

def weighted_choice_compile(items):
    """returns a function that fetches a random item from items

    items is a list of tuples in the form (item, weight)"""
    weight_total = sum((item[1] for item in items))
    def choice(uniform = random.uniform):
        n = uniform(0, weight_total)
        for item, weight in items:
            if n < weight:
                return item
            n = n - weight
        return item
    return choice

Версия компиляции 2:

def weighted_choice_bisect_compile(items):
    """Returns a function that makes a weighted random choice from items."""
    added_weights = []
    last_sum = 0

    for item, weight in items:
        last_sum += weight
        added_weights.append(last_sum)

    def choice(rnd=random.random, bis=bisect.bisect):
        return items[bis(added_weights, rnd() * last_sum)][0]
    return choice

Затем я создал большой список вариантов:

choices = [(random.choice("abcdefg"), random.uniform(0,50)) for i in xrange(2500)]

И слишком простая функция профилирования:

def profiler(f, n, *args, **kwargs):
    start = time.time()
    for i in xrange(n):
        f(*args, **kwargs)
    return time.time() - start

Результаты:

(Секунды, потраченные на 1000 обращений к функции.)

  • Простой нескомпилированный: 0,918624162674
  • Двоичный файл без компиляции: 1.01497793198
  • Простой скомпилирован: 0.287325024605
  • Двоичный код: 0,00327413797379

«Скомпилированные» результаты включают среднее время, затрачиваемое на компиляцию функции выбора один раз. (Я рассчитал 1000 компиляций, затем разделил это время на 1000 и добавил результат ко времени функции выбора.)

Итак: если у вас есть список элементов + весов, которые меняются очень редко, двоичный скомпилированный метод будет на намного самым быстрым.

6 голосов
/ 09 февраля 2009

В комментариях к оригинальному сообщению Николас Леонард предполагает, что обмен и выборка должны быть быстрыми. Вот идея для этого случая; Я не пробовал это.

Если бы только выборка была быстрой, мы могли бы использовать массив значений вместе с текущей суммой их вероятностей и выполнить бинарный поиск по текущей сумме (с ключом, являющимся однородным случайным числом) - O (log (n)) операция. Но обмен потребует обновления всех значений промежуточной суммы, появляющихся после обмена записями - операция O (n). (Не могли бы вы обменять только предметы в конце их списков? Я предполагаю, что нет.)

Итак, давайте стремиться к O (log (n)) в обеих операциях. Вместо массива сохраняйте двоичное дерево для каждого набора, из которого производится выборка. Лист содержит значение выборки и ее (ненормализованную) вероятность. Узел ветвления содержит общую вероятность своих дочерних элементов.

Для выборки сгенерируйте равномерное случайное число x между 0 и общей вероятностью корня и спустите дерево. В каждой ветви выберите левого ребенка, если у левого ребенка есть полная вероятность <= x. В противном случае вычтите вероятность левого ребенка из x и идите направо. Верните значение листа, которого вы достигли.

Чтобы обменять, удалите лист из его дерева и отрегулируйте ветви, которые ведут к нему (уменьшая их общую вероятность и вырезая все узлы с одним дочерним ветвями). Вставьте лист в дерево назначения: у вас есть выбор, куда его поместить, так что держите его сбалансированным. Выбор случайного ребенка на каждом уровне, вероятно, достаточно хорош - вот с чего я начну. Увеличьте вероятность каждого родительского узла, вернитесь к корню.

Теперь выборка и обмен в среднем составляют O (log (n)). (Если вам нужен гарантированный баланс, простой способ - добавить еще одно поле к узлам ветви, содержащее количество листьев во всем поддереве. При добавлении листа на каждом уровне выбирайте дочерний элемент с меньшим количеством листьев. Это оставляет возможность дерево становится неуравновешенным исключительно за счет удалений, это не может быть проблемой, если между наборами имеется достаточно равномерный трафик, но если это так, тогда выбирайте ротации во время удаления, используя информацию о количестве листьев на каждом узле в вашем обходе.

Обновление: По запросу, вот базовая реализация. Не настроил это вообще. Использование:

>>> t1 = build_tree([('one', 20), ('two', 2), ('three', 50)])
>>> t1
Branch(Leaf(20, 'one'), Branch(Leaf(2, 'two'), Leaf(50, 'three')))
>>> t1.sample()
Leaf(50, 'three')
>>> t1.sample()
Leaf(20, 'one')
>>> t2 = build_tree([('four', 10), ('five', 30)])
>>> t1a, t2a = transfer(t1, t2)
>>> t1a
Branch(Leaf(20, 'one'), Leaf(2, 'two'))
>>> t2a
Branch(Leaf(10, 'four'), Branch(Leaf(30, 'five'), Leaf(50, 'three')))

Код:

import random

def build_tree(pairs):
    tree = Empty()
    for value, weight in pairs:
        tree = tree.add(Leaf(weight, value))
    return tree

def transfer(from_tree, to_tree):
    """Given a nonempty tree and a target, move a leaf from the former to
    the latter. Return the two updated trees."""
    leaf, from_tree1 = from_tree.extract()
    return from_tree1, to_tree.add(leaf)

class Tree:
    def add(self, leaf):
        "Return a new tree holding my leaves plus the given leaf."
        abstract
    def sample(self):
        "Pick one of my leaves at random in proportion to its weight."
        return self.sampling(random.uniform(0, self.weight))
    def extract(self):
        """Pick one of my leaves and return it along with a new tree
        holding my leaves minus that one leaf."""
        return self.extracting(random.uniform(0, self.weight))        

class Empty(Tree):
    weight = 0
    def __repr__(self):
        return 'Empty()'
    def add(self, leaf):
        return leaf
    def sampling(self, weight):
        raise Exception("You can't sample an empty tree")
    def extracting(self, weight):
        raise Exception("You can't extract from an empty tree")

class Leaf(Tree):
    def __init__(self, weight, value):
        self.weight = weight
        self.value = value
    def __repr__(self):
        return 'Leaf(%r, %r)' % (self.weight, self.value)
    def add(self, leaf):
        return Branch(self, leaf)
    def sampling(self, weight):
        return self
    def extracting(self, weight):
        return self, Empty()

def combine(left, right):
    if isinstance(left, Empty): return right
    if isinstance(right, Empty): return left
    return Branch(left, right)

class Branch(Tree):
    def __init__(self, left, right):
        self.weight = left.weight + right.weight
        self.left = left
        self.right = right
    def __repr__(self):
        return 'Branch(%r, %r)' % (self.left, self.right)
    def add(self, leaf):
        # Adding to a random branch as a clumsy way to keep an
        # approximately balanced tree.
        if random.random() < 0.5:
            return combine(self.left.add(leaf), self.right)
        return combine(self.left, self.right.add(leaf))
    def sampling(self, weight):
        if weight < self.left.weight:
            return self.left.sampling(weight)
        return self.right.sampling(weight - self.left.weight)
    def extracting(self, weight):
        if weight < self.left.weight:
            leaf, left1 = self.left.extracting(weight)
            return leaf, combine(left1, self.right)
        leaf, right1 = self.right.extracting(weight - self.left.weight)
        return leaf, combine(self.left, right1)

Обновление 2: В , отвечая на другую проблему , Джейсон Орендорфф отмечает, что двоичные деревья можно поддерживать идеально сбалансированными, представляя их в массиве, подобно классической структуре кучи. (Это также экономит место, затрачиваемое на указатели.) См. Мои комментарии к этому ответу о том, как адаптировать его код к этой проблеме.

2 голосов
/ 19 января 2014

Примерно через 3 года ...

Если вы используете numpy, возможно, самый простой вариант - использовать np.random.choice, который принимает список возможных значений и необязательную последовательность вероятностей, связанную с каждым значением:

import numpy as np

values = ('A', 'B', 'C', 'D')
weights = (0.5, 0.1, 0.2, 0.2)

print ''.join(np.random.choice(values, size=60, replace=True, p=weights))
# ACCADAACCDACDBACCADCAAAAAAADACCDCAADDDADAAACCAAACBAAADCADABA
2 голосов
/ 08 февраля 2009

Вот классический способ сделать это в псевдокоде, где random.random () дает вам случайное значение с плавающей точкой от 0 до 1.

let z = sum of all the convictions
let choice = random.random() * z 
iterate through your objects:
    choice = choice - the current object's conviction
    if choice <= 0, return this object
return the last object

Для примера: представьте, что у вас есть два объекта, один с весом 2, другой с весом 4. Вы генерируете число от 0 до 6. Если choice находится между 0 и 2, что произойдет с 2/6 = 1/3 вероятности, тогда он будет вычтен на 2 и выбран первый объект. Если выбор между 2 и 6, что произойдет с вероятностью 4/6 = 2/3, то первое вычитание будет по-прежнему иметь выбор> 0, а второе вычитание сделает выбор второго объекта.

2 голосов
/ 08 февраля 2009

Я предлагаю вам перенести эту PHP-реализацию взвешенного случайного на Python. В частности, второй алгоритм на основе бинарного поиска помогает решить ваши проблемы со скоростью.

2 голосов
/ 08 февраля 2009

Я бы использовал этот рецепт . Вам нужно будет добавить вес к вашим объектам, но это простое соотношение и поместить их в список кортежей (объект, осуждение / (сумма убеждений)). Это должно быть легко сделать с использованием понимания списка.

2 голосов
/ 08 февраля 2009

Вы хотите дать каждому объекту вес. Чем больше вес, тем больше вероятность, что это произойдет. Точнее, probx = weight / sum_all_weights.

Затем сгенерируйте случайное число в диапазоне от 0 до sum_all_weights и сопоставьте его с каждым объектом.

Этот код позволяет генерировать случайный индекс, и он отображается при создании объекта для скорости. Если все ваши наборы объектов имеют одинаковое распределение, вы можете обойтись только одним объектом RandomIndex.

import random

class RandomIndex:
    def __init__(self, wlist):
        self._wi=[]
        self._rsize=sum(wlist)-1
        self._m={}
        i=0
        s=wlist[i]
        for n in range(self._rsize+1):
            if n == s:
                i+=1
                s+=wlist[i]
            self._m[n]=i    

    def i(self):
        rn=random.randint(0,self._rsize)
        return self._m[rn]


sx=[1,2,3,4]


wx=[1,10,100,1000] #weight list
ri=RandomIndex(wx)

cnt=[0,0,0,0]

for i in range(1000):
    cnt[ri.i()] +=1  #keep track of number of times each index was generated

print(cnt)  
1 голос
/ 12 января 2010
1 голос
/ 09 февраля 2009

Вот лучший ответ для специального распределения вероятностей, тот, который Ответ Рекса Логана , кажется, ориентирован на. Распределение выглядит следующим образом: каждый объект имеет целочисленный вес от 0 до 100, а его вероятность пропорциональна его весу. Поскольку это принятый в настоящее время ответ, я думаю, об этом стоит подумать.

Так что держите массив из 101 бункера. Каждая корзина содержит список всех объектов с определенным весом. Каждая корзина также знает общий вес всех своих объектов.

Для выборки: выбрать корзину наугад пропорционально ее общему весу. (Используйте один из стандартных рецептов для этого - линейный или бинарный поиск.) Затем случайным образом выберите объект из корзины случайно.

Для переноса объекта: удалите его из корзины, поместите его в корзину в цели и обновите вес обеих корзин. (Если вы используете бинарный поиск для выборки, вы также должны обновить текущие суммы, которые используют. Это все еще достаточно быстро, так как не так много бинов.)

1 голос
/ 08 февраля 2009

Очень простой и легкий способ сделать это - установить веса для каждого из значений, и это не потребует большого количества памяти.

Возможно, вы могли бы использовать хеш / словарь для этого.

То, что вы хотите сделать, это иметь случайное число x , умноженное и суммированное на весь набор вещей, которые вы хотите выбрать, и разделить полученный результат на количество объектов в вашем наборе .

Псевдо-код:

objectSet = [(object1, weight1), ..., (objectN, weightN)]
sum = 0
rand = random()
for obj, weight in objectSet
    sum = sum+weight*rand
choice = objectSet[floor(sum/objectSet.size())]

РЕДАКТИРОВАТЬ : Я просто подумал о том, насколько медленным будет мой код с очень большими наборами (это O (n)). Следующий псевдокод O (log (n)), и в основном использует двоичный поиск.

objectSet = [(object1, weight1), ..., (objectN, weightN)]
sort objectSet from less to greater according to weights
choice = random() * N # where N is the number of objects in objectSet
do a binary search until you have just one answer

Существуют реализации бинарного поиска в Python по всей сети, поэтому нет необходимости повторять здесь.

Добро пожаловать на сайт PullRequest, где вы можете задавать вопросы и получать ответы от других членов сообщества.
...