algorithm - 用于空模型的交换算法的 Scala 版本

Question

我遇到的问题是试图找到一种有效的方法来查找矩阵中的可交换元素，以便为空模型创建实现交换算法。

矩阵由 0 和 1 组成，想法是可以在列之间切换元素，以便矩阵的行和列总数保持不变。

例如，给定以下矩阵:

   c1 c2 c3 c4
r1  0  1  0  0 = 1
r2  1  0  0  1 = 2
r3  0  0  0  0 = 0
r4  1  1  1  1 = 4
   ------------
    2  2  1  2

r1 和 r2 中的 c2 和 c4 列都可以交换，这样总数就不会改变，即:

   c1 c2 c3 c4
r1  0  0  0  1 = 1
r2  1  1  0  0 = 2
r3  0  0  0  0 = 0
r4  1  1  1  1 = 4
   ------------
    2  2  1  2

这一切都需要随机进行，以免引入任何偏差。

我有一个有效的解决方案。我随机选择一行和两列。如果它们产生 10 或 01 模式，那么我随机选择另一行并检查相同的列以查看它们是否产生相反的模式。如果其中任何一个失败，我将重新开始并选择一个新元素。

此方法有效，但我只有大约 10% 的时间“命中”正确的模式。在一个大矩阵或行中很少有 1 的矩阵中，我浪费了很多时间“丢失”。我认为必须有一种更智能的方式来选择矩阵中的元素，但仍然是随机选择。

工作方法的代码是:

def isSwappable(matrix: Matrix): Tuple2[Tuple2[Int, Int], Tuple2[Int, Int]] = {
  val indices = getRowAndColIndices(matrix)

  (matrix(indices._1._1)(indices._2._1), matrix(indices._1._1)(indices._2._2)) match {
    case (1, 0) => {
      if (matrix(indices._1._2)(indices._2._1) == 0 & matrix(indices._1._2)(indices._2._2) == 1) {
        indices
      }
      else {
        isSwappable(matrix)
      }
    }
    case (0, 1) => {
      if (matrix(indices._1._2)(indices._2._1) == 1 & matrix(indices._1._2)(indices._2._2) == 0) {
        indices
      }
      else {
        isSwappable(matrix)
      }
    }
    case _ => {
      isSwappable(matrix)
    }
  }
}

def getRowAndColIndices(matrix: Matrix): Tuple2[Tuple2[Int, Int], Tuple2[Int, Int]] = {
  (getNextIndex(rnd.nextInt(matrix.size), matrix.size), getNextIndex(rnd.nextInt(matrix(0).size), matrix(0).size))
}

def getNextIndex(i: Int, constraint: Int): Tuple2[Int, Int] = {
  val newIndex = rnd.nextInt(constraint)
  newIndex match {
    case `i` => getNextIndex(i, constraint)
    case _ => (i, newIndex)
  }
}

我想出一种更有效的处理方法是删除所有不能使用的行(全 1 或 0)，然后随机选择一个元素。从那里我可以过滤掉行中具有相同值的任何列，然后从剩余的列中进行选择。

一旦选择了第一行和第一列，我就会过滤掉不能提供所需模式的行，然后从剩余的行中进行选择。

这在大多数情况下都有效，但我不知道如何处理的问题是当没有可供选择的列或行时会发生什么？我不想无限循环尝试找到我需要的模式，如果我确实得到一个空的行或列列表可供选择，我需要一种重新开始的方法。

到目前为止，我拥有的那种工作代码(直到我得到一个空列表)是:

def getInformativeRowIndices(matrix: Matrix) = (
  matrix
    .zipWithIndex
    .filter(_._1.distinct.size > 1)
    .map(_._2)
    .toList
  )

def getRowsWithOppositeValueInColumn(col: Int, value: Int, matrix: Matrix) = (
  matrix
    .zipWithIndex
    .filter(_._1(col) != value)
    .map(_._2)
    .toList
  )

def getColsWithOppositeValueInSameRow(row: Int, value: Int, matrix: Matrix) = (
  matrix(row)
    .zipWithIndex
    .filter(_._1 != value)
    .map(_._2)
    .toList
  )

def process(matrix: Matrix): Tuple2[Tuple2[Int, Int], Tuple2[Int, Int]] = {
  val row1Indices = getInformativeRowIndices(matrix)
  if (row1Indices.isEmpty) sys.error("No informative rows")

  val row1 = row1Indices(rnd.nextInt(row1Indices.size))
  val col1 = rnd.nextInt(matrix(0).size)
  val colIndices = getColsWithOppositeValueInSameRow(row1, matrix(row1)(col1), matrix)
  if (colIndices.isEmpty) process(matrix)
  val col2 = colIndices(rnd.nextInt(colIndices.size))
  val row2Indices = getRowsWithOppositeValueInColumn(col1, matrix(row1)(col1), matrix)
    .intersect(getRowsWithOppositeValueInColumn(col2, matrix(row1)(col2), matrix))
  println(row2Indices)
  if (row2Indices.isEmpty) process(matrix)

  val row2 = row2Indices(rnd.nextInt(row2Indices.size))
  ((row1, row2), (col1, col2))
}

我认为递归方法是错误的，在这里不起作用。另外，我真的只是想提高细胞选择的速度，所以任何想法或建议都将不胜感激。

编辑:

我有机会尝试更多，并提出了另一种解决方案，但它似乎并没有比随机选择矩阵中的单元格快多少。另外，我应该补充一点，矩阵需要连续交换大约 30000 次才能被认为是随机的，我需要为每个测试生成 5000 个随机矩阵，我至少还有 5000 个要这样做，所以性能很好的重要。

目前的解决方案(除了随机单元格选择是:

1. 从矩阵中随机选择2行
2. 从另一行中减去一行并将其放入数组
3. 如果新数组同时包含 1 和 -1，那么我们可以交换

减法的逻辑是这样的:

  0  1  0  0
- 1  0  0  1
---------------
 -1  1  0 -1

执行此操作的方法如下所示:

 def findSwaps(matrix: Matrix, iterations: Int): Boolean = {
   var result = false

   val mtxLength = matrix.length

   val row1 = rnd.nextInt(mtxLength)
   val row2 = getNextIndex(row1, mtxLength)

   val difference = subRows(matrix(row1), matrix(row2))

   if (difference.min == -1 & difference.max == 1) {
     val zeroOne = difference.zipWithIndex.filter(_._1 == -1).map(_._2)
     val oneZero = difference.zipWithIndex.filter(_._1 == 1).map(_._2)

     val col1 = zeroOne(rnd.nextInt(zeroOne.length))
     val col2 = oneZero(rnd.nextInt(oneZero.length))

     swap(matrix, row1, row2, col1, col2)
     result = true
   }
   result
 }

矩阵行减法看起来像这样:

 def subRows(a: Array[Int], b: Array[Int]): Array[Int] = (a, b).zipped.map(_ - _)

实际的交换看起来像这样:

 def swap(matrix: Matrix, row1: Int, row2: Int, col1: Int, col2: Int) = {

   val temp = (matrix(row1)(col1), matrix(row1)(col2))
   matrix(row1)(col1) = matrix(row2)(col1)
   matrix(row1)(col2) = matrix(row2)(col2)

   matrix(row2)(col1) = temp._1
   matrix(row2)(col2) = temp._2
   matrix
 }

这比以前好得多，因为我尝试交换的成功率在 80% 到 90% 之间(随机选择单元格时只有大约 10%)但是......它仍然需要大约 2.5 分钟才能完成生成 1000 个随机矩阵。

关于如何提高速度有什么想法吗？

Answer 1

我将假设矩阵很大，因此(矩阵大小的平方)数量级的存储不可行(出于速度或内存的原因)。

如果你有一个稀疏矩阵，你可以在一个集合的每一列中输入每个 1 的索引(这里我展示了做事的紧凑方式，但你可能希望用 while 循环迭代以提高速度):

val mtx = Array(Array(0,1,0,0),Array(1,0,0,1),Array(0,0,0,0),Array(1,1,1,1))
val cols = mtx.transpose.map(x => x.zipWithIndex.filter(_._1==1).map(_._2).toSet)

现在对于每一列，当且仅当以下两个集合非空时，后面的列包含兼容对(至少一个):

def xorish(a: Set[Int], b: Set[Int]) = (a--b, b--a)

所以答案将涉及计算这些集合并测试它们是否都是非空的。

现在的问题是“随机抽样”是什么意思。随机抽样单个 1,0 对与随机抽样可能的交换不同。要了解这一点，请考虑以下事项:

1 0       1 0
1 0       1 0
1 0       1 0
0 1       1 0
0 1       1 0
0 1       0 1

左边的两列有九种可能的交换。右边的两个只有五种可能的交换。但是，如果您正在寻找 (1,0) 模式，您将只在左侧采样 3 次，而在右侧采样 5 次；如果您正在寻找 (1,0) 或 (0,1)，您将采样 6 和 6，这再次扭曲了概率。解决这个问题的唯一方法是要么不聪明，然后随机抽取第二次(在第一种情况下，3/5 的时间可以使用交换，而只有 1/5 个)，或者基本上计算每个可能的交换对(或至少有多少对)并从该预定义集合中选择。

如果我们想做后者，我们注意到对于每对不同的列，我们可以计算要交换的两个集合，并且我们知道大小和乘积是可能性的总数。为了避免实例化所有的可能性，我们可以创建

val poss = {
  for (i<-cols.indices; j <- (i+1) until cols.length) yield 
    (i, j, (cols(i)--cols(j)).toArray, (cols(j)--cols(i)).toArray)
}.filter{ case (_,_,a,b) => a.length>0 && b.length>0 }

然后数一数有多少:

val cuml = poss.map{ case (_,_,a,b) => a.size*b.size }.scanLeft(0)(_ + _).toArray

现在随机选择一个数字，我们选择一个介于 0 和 cuml.last 之间的数字，然后选择这是哪个桶以及桶中的哪个项目:

def pickItem(cuml: Array[Int], poss: Seq[(Int,Int,Array[Int],Array[Int])]) = {
  val n = util.Random.nextInt(cuml.last)
  val k = {
    val i = java.util.Arrays.binarySearch(cuml,n)
    if (i<0) -i-2 else i
  }
  val j = n - cuml(k)
  val bucket = poss(k)
  (
    bucket._1, bucket._2, 
    bucket._3(j % bucket._3.size), bucket._4(j / bucket._3.size)
  )
}

这最终会返回随机选择的 (c1,c2,r1,r2)。

现在您有了坐标，您可以根据需要创建新矩阵。 (最有效的方法可能是就地交换条目，然后在您想重试时交换回来。)

请注意，这仅适用于来自同一起始矩阵的大量独立交换。如果您想迭代地执行此操作并保持独立性，那么您最好还是随机执行此操作，除非矩阵非常稀疏，此时值得将矩阵简单地存储在一些标准稀疏矩阵中格式(即通过非零条目的索引)并对它们进行操作(可能使用可变集和更新策略，因为单个交换的结果仅限于 n 中的条目 n*n矩阵)。