产生大量随机变量

Question

我正在尝试找出在Python中生成许多随机数字的最佳方法。困难的部分是在运行时之前我不知道需要多少个数字。 我有一个程序，它每次使用一个随机数，但需要多次这样做。到目前为止我尝试了以下几种方法： - 使用random.random()逐个生成随机数 - 使用np.random.rand()逐个生成随机数 - 使用np.random.rand(N)批量生成N个随机数 - 使用np.random.rand(N)批量生成N个随机数，并在使用完第一个N个后再生成新的一批（我尝试了两种不同的实现，但都比逐个生成随机数慢） 在下面的脚本中，我比较了这三种方法的结果（对于均匀分布和正态分布的随机数）。 我不知道函数p是否真的必要，但我想在每种情况下使用相同的随机数来进行等效的操作，这似乎是最简单的方法。

#!/bin/python3

import time
import random
import numpy as np

def p(x):
    pass

def gRand(n):
    for i in range(n):
        p(random.gauss(0,1))

def gRandnp1(n):
    for i in range(n):
        p(np.random.randn())

def gRandnpN(n):
    rr=np.random.randn(n)
    for i in rr:
        p(i)

def uRand(n):
    for i in range(n):
        p(random.random())

def uRandnp1(n):
    for i in range(n):
        p(np.random.rand())

def uRandnpN(n):
    rr=np.random.rand(n)
    for i in rr:
        p(i)

tStart=[]
tEnd=[]
N=1000000
for f in [uRand, uRandnp1, uRandnpN]:
    tStart.append(time.time())
    f(N)
    tEnd.append(time.time())

for f in [gRand, gRandnp1, gRandnpN]:
    tStart.append(time.time())
    f(N)
    tEnd.append(time.time())

print(np.array(tEnd)-np.array(tStart))

这个计时脚本的输出示例如下：
[ 0.26499939 0.45400381 0.19900227 1.57501364 0.49000382 0.23000193]
前三个数字是在[0,1)上均匀分布的随机数，后三个数字是正态分布的随机数(mu=0, sigma=1)。

对于任何类型的随机变量，这三种方法中最快的方法是一次性生成所有随机数，将它们存储在数组中，并迭代该数组。问题是，在运行程序之后，我不知道需要多少这些数字。

我想做的是批量生成随机数。然后当我使用一个批次中的所有数字时，我只需重新填充存储它们的对象。问题是，我不知道有没有一个干净的方法来实现这一点。我想到的一个解决方案如下：

N=1000000
numRepop=4
N1=N//numRepop
__rands__=[]
irand=-1

def repop():
    global __rands__
    __rands__=np.random.rand(N1)

repop()

def myRand():
    global irand
    try:
        irand += 1
        return __rands__[irand]
    except:
        irand=1
        repop()
        return __rands__[0]

但实际上这比其他选项都要慢。

如果我将numpy数组转换为列表，然后弹出元素，那么我的性能就与仅使用numpy一次生成随机变量类似：

__r2__=[]

def repop2():
    global __r2__
    rr=np.random.rand(N1)
    __r2__=rr.tolist()

repop2()

def myRandb():
    try:
        return __r2__.pop()
    except:
        repop2()
        return __r2__.pop()

有更好的方式来处理这个问题吗？

编辑：我的意思是更快的方式。我也希望使用确定性（伪）随机数。

- kevin

---

(1) 对于这种基准测试要小心：返回自纪元以来的秒数作为浮点数。请注意，尽管时间始终返回为浮点数，但并非所有系统都提供比1秒更高的精度。虽然此函数通常返回非递减值，但如果在两次调用之间系统时钟被设置回，则它可以返回低于先前调用的值。 (2) 如果你只需要移动索引以选择位置，为什么要弹出呢？不需要删除对象。 - sascha

“更好的方式”是什么意思？除了性能之外，您还有其他要求吗？对于某些目的，您需要加密安全的随机数，或者您可能需要超过默认的32位随机性。 - Håken Lid

@Håken Lid，“更好”我只是指更快。 - kevin

@sascha (1) 那是我不知道的一个好观点，但我做的时间测试结果比较一致，所以暂时我不会过于担心。(2) 起初我尝试使用索引（使用 repop 和 myRand），但这比使用 pop()（使用 repop2 和 myRandb）更慢。 - kevin

@dawg 这并没有为这个问题增添任何内容。正如OP所明确表示的那样，我们只对非加密目的感兴趣。你提出的两种方法都非常缓慢、不确定（这是一个有用的特性），而且通常也可能提供质量更差的随机数（针对此任务）。是的，最后一句话有些极端，如果你愿意可以忽略它（我相信PRNG比熵估计更可靠）。 - sascha

Answer 1

如果一次生成许多数字更快，您可以创建一个会缓存批次的生成器。这适用于Python 3.5。

def randoms(batchsize=10000):
    while True:
        yield from numpy.random.rand(batchsize)

不知道它是否比你的其他实现更快，但它是一个永无止境的生成器。

您可以像使用任何迭代器一样使用它：

prng = randoms()
for _ in range(1000000):
    foo(next(prng))

或者像这样（但循环永远不会退出）：

for x in randoms():
    foo(x)

编辑:

我尝试自己进行基准测试，我认为差异主要是由于Python中函数调用的额外成本。我已经尝试通过在所有情况下循环一个range来使基准测试更具可比性，并且使用预生成的数组的优势较小。

通过使用微小的优化技巧，将numpy.random.rand分配给本地变量，可以大大加快函数调用的速度，因此我得到了几乎同样好的速度。

为了比较，我还包括生成器方法。

def randoms(batchsize):
    rand = numpy.random.rand
    while True:
        yield from rand(batchsize)
​
def test_generator(times):
    rand = randoms(1000).__next__
    for n in range(times):
        rand()

def test_rand(times):
    for n in range(times):
        numpy.random.rand() 

def test_rand_micro_opt(times):
    rand = numpy.random.rand
    for n in range(times):
        rand()

def test_array(times):
    array = numpy.random.rand(times)
    for n in range(times):
        array[n]
​
# ipython / jupyter magic %timeit command        
%timeit -n 1000 test_generator(10000)
%timeit -n 1000 test_rand(10000)
%timeit -n 1000 test_rand_micro_opt(10000)
%timeit -n 1000 test_array(10000)
​
1000 loops, best of 3: 2.09 ms per loop
1000 loops, best of 3: 2.93 ms per loop
1000 loops, best of 3: 1.74 ms per loop
1000 loops, best of 3: 1.57 ms per loop

- Håken Lid

---

我现在进行了一些测试，这比简单地使用“random”模块的“random”函数要慢两倍以上。测试是timeit(lambda: [next(prng) for _ in range(1000000)], number=10)与timeit(lambda: [random() for _ in range(1000000)], number=10)。我还尝试将prng.__next__存储在变量中并使用它，但效果不大。 - Stefan Pochmann

是的。它基于这样一个假设，即制作数字批次将节省大量时间，因为OP的基准测试似乎表明如此。但我认为速度差异主要是由于昂贵的函数调用造成的，当您直接在数组上循环时不会出现这种情况。使用生成器并不会减少函数调用。 - Håken Lid

我也测试了生成器版本，但它的速度仅比我编辑后答案中最慢的测试略快。 - Håken Lid

最好还是在答案中包含那个测试，最好使用 rand = prng.__next__。 - Stefan Pochmann

你说得对，那确实有所不同。然而，简单的微观优化仍然更快。 - Håken Lid

Answer 2

您可以通过不一直查找模块及其函数来显著加快速度。

def uRand_2(n):
    r = random.random
    for i in range(n):
        p(r())

def uRandnp1_2(n):
    r = np.random.rand
    for i in range(n):
        p(r())

你的版本在我的电脑上计时：

[ 0.14439154  0.24865651  0.13786387  0.85637093  0.28924942  0.13338685]

我上面的两个版本（对应你的前两个）：

[ 0.10629296  0.15638423]

噢，我不明白为什么要调用p。我认为这只会增加噪音并模糊实际随机数生成的速度。以下是我在不调用p的情况下使用r()的时间:

[ 0.04560113  0.1083169]

- Stefan Pochmann

---

我想知道为什么numpy函数这么慢。还有numpy.random.random()似乎比numpy.random.rand()生成单个值时快大约两倍。不过标准库版本仍然更快。 - Håken Lid

@HåkenLid 可能是参数的问题。我刚刚尝试了 timeit('f()', 'def f(size=None): pass')，速度大约是 timeit('f()', 'def f(**args): pass') 的两倍。 - Stefan Pochmann

Answer 3

不是非常漂亮，但应该可以工作：

import numpy as np

class BatchedPRNG(object):
    def __init__(self, seed=0, batch_size=10000, dist='uniform'):
        self.prng = np.random.RandomState(seed)         # own random-stream !
        self.batch_size = batch_size
        self.dist = dist
        self.index = 0
        if self.dist == 'uniform':
            self.pool = self.prng.random_sample(size=self.batch_size)
        else:
            self.pool = self.prng.normal(size=self.batch_size)

    def sample_one(self):
        if self.index < self.batch_size:
            self.index += 1
            return self.pool[self.index-1]
        else:
            self.index = 1
            if self.dist == 'uniform':
                self.pool = self.prng.random_sample(size=self.batch_size)
            else:
                self.pool = self.prng.normal(size=self.batch_size)
            return self.pool[self.index-1]

dist = BatchedPRNG()
for i in range(11):
    print(dist.sample_one())

这遵循了封装/面向对象的思想，每次需要新样本时都要进行函数调用。它还使用自己的PRNG流，因此代码其他部分对np.random.X的全局调用不会改变该对象的内部状态。

显然，如果您想使用其他分布或需要其他功能，则需要修改此内容。

不幸的是，您还需要注意基准测试。

编辑：出乎意料的慢

- sascha