Python 随机样本生成器（适应庞大的人口规模）

Question

您可能知道random.sample(population,sample_size) 会快速返回一个随机样本，但是如果您事先不知道样本的大小怎么办？您最终会对整个人口进行抽样，或对其进行洗牌，这是相同的。但这可能是浪费的（如果大多数样本量与总体规模相比都很小），甚至是不可行的（如果总体规模很大，内存不足）。另外，如果您的代码需要在选择示例的下一个元素之前从这里跳转到那里怎么办？

附：在为TSP 处理simulated annealing 时，我遇到了优化随机样本的需求。在我的代码中，采样重新启动了数十万次，每次我都不知道我是否需要选择 1 个元素或 100% 的总体元素。

Answer 1

起初，我会将人口分成几块。进行块采样的功能可以很容易地成为一个生成器，能够处理任意大小的样本。这也允许您将函数设为生成器。

想象无限人口，人口块为 512，样本大小为 8。这意味着您可以根据需要收集尽可能多的样本，并且为了将来减少再次对已经采样的空间进行采样（对于 1024 个块，这意味着 8196 个样本来自其中您可以再次采样）。

同时，这允许并行处理，这在样本非常大的情况下可能是可行的。

考虑内存填充的示例

import random

population = [random.randint(0, 1000) for i in range(0, 150000)]

def sample_block(population, block_size, sample_size):
    block_number = 0
    while 1:
        try:
            yield random.sample(population[block_number * block_size:(block_number + 1) * block_size], sample_size)
            block_number += 1
        except ValueError:
            break

sampler = sample_block(population, 512, 8)
samples = []

try:
    while 1:
        samples.extend(sampler.next())
except StopIteration:
    pass

print random.sample(samples, 200)

如果填充在脚本外部（文件、块），唯一的修改是您必须将适当的块加载到内存中。无限人口抽样的概念证明：

import random
import time

def population():
    while 1:
        yield random.randint(0, 10000)

def reduced_population(samples):
    for sample in samples:
        yield sample

def sample_block(generator, block_size, sample_size):

    block_number = 0
    block = []
    while 1:
        block.append(generator.next())
        if len(block) == block_size:
            s = random.sample(block, sample_size)
            block_number += 1
            block = []
            print 'Sampled block {} with result {}.'.format(block_number, s)
            yield s

samples = []
result = []
reducer = sample_block(population(), 512, 12)

try:
    while 1:
        samples.append(reducer.next())
        if len(samples) == 1000:
            sampler = sample_block(reduced_population(samples), 1000, 15)
            result.append(list(sampler))
            time.sleep(5)
except StopIteration:
    pass

理想情况下，您还可以收集样本并再次对其进行采样。

Answer 2

我（在 Python 2.7.9 中）编写了一个随机采样器生成器（索引），其速度仅取决于样本大小（它应该是 O(ns log(ns))，其中 ns 是样本大小）。因此，当样本量与总体规模相比较小时，它快速，因为它完全不依赖于总体规模。它不建立任何人口集合，它只是选择随机索引并在采样索引上使用一种二分法以避免重复并保持然后排序。给定一个可迭代的population，下面是如何使用itersample 生成器：

import random
sampler=itersample(len(population))
next_pick=sampler.next() # pick the next random (index of) element

或

import random
sampler=itersample(len(population))
sample=[]
for index in sampler:
    # do something with (index of) picked element
    sample.append(index) # build a sample
    if some_condition: # stop sampling when needed
        break

如果您需要实际元素而不仅仅是索引，只需在需要时将population iterable 应用于索引（第一个和第二个示例分别为population[sampler.next()] 和population[index]）。

一些测试的结果表明速度并不取决于人口规模，所以如果你需要从 1000 亿人口中随机挑选 10 个元素，你只需支付 10 个（记住，我们事先不知道我们会选择多少个元素，否则你最好使用random.sample)。

Sampling 1000 from 1000000
Using itersample 0.0324 s

Sampling 1000 from 10000000
Using itersample 0.0304 s

Sampling 1000 from 100000000
Using itersample 0.0311 s

Sampling 1000 from 1000000000
Using itersample 0.0329 s

其他测试证实，运行时间与样本量的线性关系略大于线性：

Sampling 100 from 1000000000
Using itersample 0.0018 s

Sampling 1000 from 1000000000
Using itersample 0.0294 s

Sampling 10000 from 1000000000
Using itersample 0.4438 s

Sampling 100000 from 1000000000
Using itersample 8.8739 s

最后，这里是生成器函数itersample：

import random
def itersample(c): # c: population size
    sampled=[]
    def fsb(a,b): # free spaces before middle of interval a,b
        fsb.idx=a+(b+1-a)/2
        fsb.last=sampled[fsb.idx]-fsb.idx if len(sampled)>0 else 0
        return fsb.last
    while len(sampled)<c:
        sample_index=random.randrange(c-len(sampled))
        a,b=0,len(sampled)-1
        if fsb(a,a)>sample_index:
            yielding=sample_index
            sampled.insert(0,yielding)
            yield yielding
        elif fsb(b,b)<sample_index+1:
            yielding=len(sampled)+sample_index
            sampled.insert(len(sampled),yielding)
            yield yielding
        else: # sample_index falls inside sampled list
            while a+1<b:
                if fsb(a,b)<sample_index+1:
                    a=fsb.idx
                else:
                    b=fsb.idx
            yielding=a+1+sample_index
            sampled.insert(a+1,yielding)
            yield yielding

Answer 3

这就是生成器的用途，我相信。这是通过生成器/产量进行 Fisher-Yates-Knuth 采样的示例，您可以一一获取事件并在需要时停止。

代码更新

import random
import numpy
import array

class populationFYK(object):
    """
    Implementation of the Fisher-Yates-Knuth shuffle
    """
    def __init__(self, population):
        self._population = population      # reference to the population
        self._length     = len(population) # lengths of the sequence
        self._index      = len(population)-1 # last unsampled index
        self._popidx     = array.array('i', range(0,self._length))

        # array module vs numpy
        #self._popidx     = numpy.empty(self._length, dtype=numpy.int32)
        #for k in range(0,self._length):
        #    self._popidx[k] = k


    def swap(self, idx_a, idx_b):
        """
        Swap two elements in population
        """
        temp = self._popidx[idx_a]
        self._popidx[idx_a] = self._popidx[idx_b]
        self._popidx[idx_b] = temp

    def sample(self):
        """
        Yield one sampled case from population
        """
        while self._index >= 0:
            idx = random.randint(0, self._index) # index of the sampled event

            if idx != self._index:
                self.swap(idx, self._index)

            sampled = self._population[self._popidx[self._index]] # yielding it

            self._index -= 1 # one less to be sampled

            yield sampled

    def index(self):
        return self._index

    def restart(self):
        self._index = self._length - 1
        for k in range(0,self._length):
            self._popidx[k] = k

if __name__=="__main__":
    population = [1,3,6,8,9,3,2]

    gen = populationFYK(population)

    for k in gen.sample():
        print(k)

Answer 4

您可以通过在 [0...N[ 范围内选择 K 个非重复随机数并将它们视为索引来从大小为 N 的总体中获取大小为 K 的样本。

选项 a)

您可以使用众所周知的样本方法生成这样的索引样本。

random.sample(xrange(N), K)

来自Python docs about random.sample：

要从整数范围中选择样本，请使用 xrange() 对象作为参数。这对于从大量人口中进行采样特别快速且节省空间

选项 b)

如果您不喜欢 random.sample 已经返回列表而不是非重复随机数的惰性生成器这一事实，您可以使用Format-Preserving Encryption 来加密计数器。

通过这种方式，您可以获得真正的随机索引生成器，您可以选择任意数量的索引并随时停止，而不会出现任何重复，从而为您提供动态大小的样本集。

这个想法是构建一个加密方案来加密从 0 到 N 的数字。现在，对于每次您想从总体中获取样本时，您选择一个随机密钥进行加密并开始加密从0, 1, 2, ... 向前（这是计数器）。由于每个好的加密都会创建一个看起来随机的 1:1 映射，因此您最终会得到可以用作索引的非重复随机整数。 这个惰性生成期间的存储需求只是初始键加上计数器的当前值。

Generating non-repeating random numbers in Python 已经讨论了这个想法。甚至还有一个 python sn-p 链接：formatpreservingencryption.py

使用这个 sn-p 的示例代码可以这样实现：

def itersample(population):
    # Get the size of the population
    N = len(population)
    # Get the number of bits needed to represent this number
    bits = (N-1).bit_length()
    # Generate some random key
    key = ''.join(random.choice(string.ascii_letters + string.digits) for _ in range(32))
    # Create a new crypto instance that encrypts binary blocks of width <bits>
    # Thus, being able to encrypt all numbers up to the nearest power of two
    crypter = FPEInteger(key=key, radix=2, width=bits)

    # Count up 
    for i in xrange(1<<bits):
        # Encrypt the current counter value
        x = crypter.encrypt(i)
        # If it is bigger than our population size, just skip it
        # Since we generate numbers up to the nearest power of 2, 
        # we have to skip up to half of them, and on average up to one at a time
        if x < N:
            # Return the randomly chosen element
            yield population[x]

Answer 5

这是另一个想法。因此，对于庞大的人口，我们希望保留一些有关选定记录的信息。在您的情况下，您为每个选定的记录保留一个整数索引 - 32 位或 64 位整数，加上一些代码来进行合理的搜索 wrt 选择/未选择。在大量选定记录的情况下，这种记录保存可能会令人望而却步。我建议对选定的索引集使用布隆过滤器。误报匹配是可能的，但不会出现误报，因此没有获得重复记录的风险。它确实引入了轻微的偏差——误报记录将被排除在抽样之外。但是内存效率很好，对于 1% 的误报概率，每个元素需要少于 10 位。因此，如果您选择 5% 的人口并有 1% 的误报，那么您错过了 0.0005 个人口，这取决于要求可能没问题。如果您想要更低的误报，请使用更多位。但是内存效率会好很多，尽管我希望每个记录样本可以执行更多代码。

抱歉，没有代码