在不知道元素总数的情况下从数据流中随机拆分元素

Question

给定一个“分割比率”，我试图将一个数据集随机分成两组。问题是，我事先不知道数据集包含多少项目。我的库从输入流中一一接收数据，并期望将数据返回到两个输出流。理想情况下，生成的两个数据集应完全按照给定的拆分比例拆分。

插图：

                            ┌─► stream A
 input stream ──► LIBRARY ──┤
                            └─► stream B

例如，给定30/70 的拆分比率，流 A 应接收来自输入流的 30% 的元素，流 B 接收剩余的 70%。订单必须保留。

---

到目前为止我的想法：

想法 1：为每个元素“掷骰子”

显而易见的方法：对于每个元素，算法随机决定该元素应该进入流 A 还是 B。问题是，生成的数据集可能与预期的拆分比率相差甚远。给定50/50 的拆分比率，生成的数据拆分可能有些遥远（对于非常小的数据集，甚至可能是100/0）。目标是使得到的分流比尽可能接近所需的分流比。

思路2：使用缓存，随机化缓存数据

另一个想法是在传递它们之前缓存固定数量的元素。这将导致缓存 1000 个元素并打乱数据（或它们相应的索引以保持顺序稳定），将它们拆分并传递结果数据集。这应该工作得很好，但我不确定随机化对于大型数据集是否真的是随机的（我想在查看分布时会有模式）。

这两种算法都不是最优的，所以希望你能帮助我。

---

背景

这是关于基于层的数据科学工具，其中每一层通过流从前一层接收数据。该层预计在传递数据（向量）之前将其拆分为训练和测试集。输入数据的范围可以从几个元素到永无止境的数据流（因此，流）。代码是用 JavaScript 开发的，但是这个问题更多的是关于算法而不是实际的实现。

Answer 1

您可以调整概率，因为它偏离所需的比率。

这是一个示例，以及对调整概率的各种级别的测试。随着我们增加调整，我们看到分流器与理想比率的偏差较小，但这也意味着它的随机性较小（知道之前的值，您可以预测下一个值）。

// rateStrictness = 0 will lead to "rolling the dice" for each invocations
// higher values of rateStrictness will lead to strong "correcting" forces
function* splitter(desiredARate, rateStrictness = .5) {
	let aCount = 0, bCount = 0;

	while (true) {

		let actualARate = aCount / (aCount + bCount);
		let aRate = desiredARate + (desiredARate - actualARate) * rateStrictness;
		if (Math.random() < aRate) {
			aCount++;
			yield 'a';
		} else {
			bCount++;
			yield 'b';
		}
	}
}

let test = (desiredARate, rateStrictness) => {
	let s = splitter(desiredARate, rateStrictness);
	let values = [...Array(1000)].map(() => s.next().value);
	let aCount = values.map((_, i) => values.reduce((count, v, j) => count + (v === 'a' && j <= i), 0));
	let aRate = aCount.map((c, i) => c / (i + 1));
	let deviation = aRate.map(a => a - desiredARate);
	let avgDeviation = deviation.reduce((sum, dev) => sum + dev, 0) / deviation.length;
	console.log(`inputs: desiredARate = ${desiredARate}; rateStrictness = ${rateStrictness}; average deviation = ${avgDeviation}`);
};

test(.5, 0);
test(.5, .25);
test(.5, .5);
test(.5, .75);
test(.5, 1);
test(.5, 10);
test(.5, 100);

Answer 2

掷骰子两次怎么样：首先决定是应该随机选择流还是应该考虑比率。然后对于第一种情况，掷骰子，对于第二种情况，取比率。一些伪代码：

  const toA =
    Math.random() > 0.5 // 1 -> totally random, 0 -> totally equally distributed
      ? Math.random() > 0.7
      :  (numberA / (numberA + numberB) > 0.7);

这只是我想出的一个想法，我还没有尝试过......

Answer 3

这是一种结合了您的两种想法的方法：它使用缓存。只要缓存中的元素数量可以处理，如果流结束，我们仍然可以接近目标分布，我们只是掷骰子。如果没有，我们将其添加到缓存中。当输入流结束时，我们将缓存中的元素打乱并发送它们以尝试接近分布。如果分布在随机性方面偏离太多，我不确定这是否比仅仅强制元素转到 x 有任何好处。

请注意，这种方法不会保留原始输入流的顺序。可以添加一些其他内容，例如缓存限制和放宽分布错误（此处使用 0）。如果您需要保持顺序，可以通过发送缓存值并推送到缓存当前的值来完成，而不是在缓存中仍有元素时仅发送当前的值。

let shuffle = (array) => array.sort(() => Math.random() - 0.5);

function* generator(numElements) {
  for (let i = 0; i < numElements;i++) yield i; 
}

function* splitter(aGroupRate, generator) {
  let cache = [];
  let sentToA = 0;
  let sentToB = 0;
  let bGroupRate = 1 - aGroupRate;
  let maxCacheSize = 0;
  
  let sendValue = (value, group) => {
      sentToA += group == 0;
      sentToB += group == 1;
      return {value: value, group: group};
  }
  
  function* retRandomGroup(value, expected) {
    while(Math.random() > aGroupRate != expected) {
      if (cache.length) {
        yield sendValue(cache.pop(), !expected);
      } else {
        yield sendValue(value, !expected);
        return;
      } 
    }
    yield sendValue(value, expected);
  }
  
  for (let value of generator) {
    if (sentToA + sentToB == 0) {
      yield sendValue(value, Math.random() > aGroupRate);
      continue;
    }
    
    let currentRateA = sentToA / (sentToA + sentToB);
        
    if (currentRateA <= aGroupRate) {
      // can we handle current value going to b group?
      if ((sentToA + cache.length) / (sentToB + sentToA + 1 + cache.length) >= aGroupRate) {
        for (val of retRandomGroup(value, 1)) yield val;
        continue;
      }
    }
    
    if (currentRateA > aGroupRate) {
      // can we handle current value going to a group?
      if (sentToA / (sentToB + sentToA + 1 + cache.length) <= aGroupRate) {
        for (val of retRandomGroup(value, 0)) yield val;
        continue;
      }
    }  
    
    cache.push(value);
    maxCacheSize = Math.max(maxCacheSize, cache.length)
  }
  
  shuffle(cache);
  
  let totalElements = sentToA + sentToB + cache.length;
  
  while (sentToA < totalElements * aGroupRate) {
    yield {value: cache.pop(), group: 0}
    sentToA += 1;
  }
  
  while (cache.length) {
    yield {value: cache.pop(), group: 1}
  }  
  
  yield {cache: maxCacheSize}
}

function test(numElements, aGroupRate) {
  let gen = generator(numElements);
  let sentToA = 0;
  let total = 0;
  let cacheSize = null;
  let split = splitter(aGroupRate, gen);
  for (let val of split) {
    if (val.cache != null) cacheSize = val.cache;
    else {
      sentToA += val.group == 0;
      total += 1
    }
  }
  console.log("required rate for A group", aGroupRate, "actual rate", sentToA / total, "cache size used", cacheSize);
}

test(3000, 0.3)
test(5000, 0.5)
test(7000, 0.7)

Answer 4

假设您必须为流向 A 的数据项保持给定的比率 R，例如根据您的示例，R = 0.3。然后在收到每个数据项计数 项目总数和传递到流 A 的项目，并根据哪种选择使您更接近目标比率 R，决定每个项目是否进入 A。

对于任何大小的数据集，这应该是您所能做的最好的事情。至于随机性，生成的流 A 和 B 应该与您的输入流一样随机。

让我们看看前几次迭代的结果：

示例：R = 0.3

N : 到目前为止处理的项目总数（最初为 0）

A : 到目前为止传递给流 A 的数字（最初为 0）

第一次迭代

N = 0 ; A = 0 ; R = 0.3
if next item goes to stream A then 
    n = N + 1
    a = A + 1
    r = a / n = 1
else if next item goes to stream B
    n = N + 1
    a  = A
    r = a / n = 0

So first item goes to stream B since 0 is closer to 0.3

第二次迭代

N = 1 ; A = 0 ; R = 0.3
if next item goes to stream A then 
    n = N + 1
    a = A + 1
    r = a / n = 0.5
else if next item goes to stream B
    n = N + 1
    a = A
    r = a / n = 0

So second item goes to stream A since 0.5 is closer to 0.3

第三次迭代

N = 2 ; A = 1 ; R = 0.3
if next item goes to stream A then 
    n = N + 1
    a = A + 1
    r = a / n = 0.66
else if next item goes to stream B
    n = N + 1
    a = A
    r = a / n = 0.5

So third item goes to stream B since 0.5 is closer to 0.3

第四次迭代

N = 3 ; A = 1 ; R = 0.3
if next item goes to stream A then 
    n = N + 1
    a = A + 1
    r = a / n = 0.5
else if next item goes to stream B
    n = N + 1
    a = A
    r = a / n = 0.25

So third item goes to stream B since 0.25 is closer to 0.3

所以这里是决定每个数据项的伪代码：

if (((A + 1) / (N + 1)) - R) < ((A / (N + 1)) - R ) then
    put the next data item on stream A
    A = A + 1
    N = N + 1
 else
    put the next data item on B
    N = N + 1

正如下面的 cmets 中所讨论的，这在 OP 所期望的意义上不是随机的。因此，一旦我们知道下一个项目的正确目标流，我们就会掷硬币来决定我们是真的把它放在那里，还是引入错误。

if (((A + 1) / (N + 1)) - R) < ((A / (N + 1)) - R ) then
    target_stream = A
else 
    target_stream = B

if random() < 0.5 then
    if target_stream == A then
        target_stream = B
    else
        target_stream = A

if target_stream == A then   
    put the next data item on stream A
    A = A + 1
    N = N + 1
 else
    put the next data item on B
    N = N + 1

现在这可能会导致总体上任意大的错误。所以我们必须设置一个误差限制 L 并检查当即将引入误差时得到的比率与目标 R 相差多远：

if (((A + 1) / (N + 1)) - R) < ((A / (N + 1)) - R ) then
    target_stream = A
else 
    target_stream = B

if random() < 0.5 then
    if target_stream == A then
        if abs((A / (N + 1)) - R) < L then
            target_stream = B
    else
        if abs(((A + 1) / (N + 1)) - R) < L then
            target_stream = A

if target_stream == A then   
    put the next data item on stream A
    A = A + 1
    N = N + 1
 else
    put the next data item on B
    N = N + 1

所以我们得到了它：一个接一个地处理数据项，我们知道放置下一个项目的正确流，然后我们引入随机局部错误，我们能够用 L 限制整体错误。

Answer 5

查看您编写的两个数字（块大小为 1000，概率拆分为 0.7），您可能对只为每个元素掷骰子的简单方法没有任何问题。 谈到概率和高数，你有law of large numbers。

这意味着，您确实存在将流非常不均匀地分成 0 和 1000 个元素的风险，但实际上这不太可能发生。当您谈论测试和训练集时，我也不希望您的概率分裂远离 0.7。如果您被允许缓存，您仍然可以将其用于前 100 个元素，这样您就可以确保有足够的数据来执行大数定律。

这是binomial distribution，n=1000，p=.7

如果你想用其他参数重现图像

import pandas as pd
import matplotlib.pyplot as plt
from scipy.stats import binom
index = np.arange(binom.ppf(0.01, n, p), binom.ppf(0.99, n, p))
pd.Series(index=index, data=binom.pmf(x, n, p)).plot()
plt.show()