python - Pythonのnumpy/scipyでベクトル化された相関関数を高速化しますか？

Question

行列内の列のペアワイズ相関をとる関数（組み込みのようpdistにscipy.stats）を作成しましたが、引数で指定された欠落値を処理できますna_values。すなわち：

def my_pdist(X, dist_func, na_values=["NA"]):
    X = array(X, dtype=object)
    num_rows, num_cols = X.shape
    dist_matrix = []
    for col1 in range(num_cols):
        pdist_row = []
        for col2 in range(num_cols):
            pairs = array([[x, y] for x, y in zip(X[:, col1], X[:, col2]) \
                           if (x not in na_values) and (y not in na_values)])
            if len(pairs) == 0:
                continue
            dist = dist_func(pairs[:, 0],
                             pairs[:, 1])
            pdist_row.append(dist)
        dist_matrix.append(pdist_row)
    dist_matrix = array(dist_matrix)
    return dist_matrix

ここdist_funcで、は距離メトリックを指定する関数です。この機能を高速化する方法はありますか？使用例は次のとおりです。

def spearman_dist(u, v, na_vals=["NA"]):
    matrix = [[x, y] for x, y in zip(u, v) \
              if (u not in na_vals) and (v not in na_vals)]
    matrix = array(matrix)
    spearman = scipy.stats.spearmanr(matrix[:, 0], matrix[:, 1])[0]
    return 1 - spearman

my_pdist(X, spearman_dist, na_values=["NA"])

どうすればこれをベクトル化/高速化できますか？

Answer 1

いくつかの提案があります：

1. 'object' 型の配列は使用しないでください。これにより、生の値ではなく python オブジェクトを操作する必要があるため、numpy は組み込みの最適化を使用できなくなります。float 配列を使用する場合は、「NA」の代わりに np.nan を使用できます。整数配列の場合、良い/悪い値のマスクを別の配列に格納するのが最善かもしれません (これにはマスクされた配列を使用することもできますが、少し扱いに​​くいと思います)。

2. この行がほとんどの時間を占めているに違いありません。

   pairs = array([[x, y] for x, y in zip(X[:, col1], X[:, col2]) \
                      if (x not in na_values) and (y not in na_values)])
   

   したがって、次のように内側のループを高速化できます。

   x1 = X[:, col1]
   x2 = X[:, col2]
   mask = ~np.isnan(x1) * ~np.isnan(x2)
   if mask.sum() == 0:
       continue
   dist = dist_func(x1[mask], x2[mask])
   

3. list.append を使用して dist_matrix を構築するのではなく、空の配列から始めて、要素を埋めていきます。

   dist_matrix = np.empty((num_cols, num_cols))
   for col1 in range(num_cols):
       for col2 in range(num_cols):
           ...
           dist_matrix[col1, col2] = dist
   

4. range(num_cols) を 2 回反復しているため、実際にはほとんどの距離値を 2 回計算しています。これは最適化できます：

   dist_matrix = np.empty((num_cols, num_cols))
   for col1 in range(num_cols):
       for col2 in range(col1, num_cols):
           ...
           dist_matrix[col1, col2] = dist
           dist_matrix[col2, col1] = dist
   

5. for ループをまったく使用せずに計算全体を実行できる場合もありますが、これは dist_func の詳細に依存します。

Answer 2

na_vals を numpy のMasked Arraysに置き換えてみてください。