added Pearson correlation

Author: Tomasz Latkowski

methods/selection.py

@@ -1,61 +1,72 @@
-import tensorflow as tf
 import pandas as pd
+import tensorflow as tf
 
 data_file = '../data/autism.tsv'
 df = pd.read_csv(data_file, sep='\t', header=None, index_col=0).T
 
 
-def fisher(data, num_instances: list, top_k=10):
+def fisher(data, num_instances: list, top_k_features=10):
     """
     Performs Fisher feature selection method according to the following formula:
     D(f) = (m1(f) - m2(f) / (std1(f) - std2(f))
 
     :param data:
     :param num_instances:
-    :param top_k:
+    :param top_k_features:
     :return: the list of most significant features.
     """
     assert len(num_instances) == 2, "Fisher selection method can be performed for two-class problems."
+    data = tf.convert_to_tensor(data)
+    _, num_features = data.get_shape().as_list()
+    if top_k_features < num_features:
+        top_k_features = num_features
     class1, class2 = tf.split(data, num_instances)
     mean1, std1 = tf.nn.moments(class1, axes=0)
     mean2, std2 = tf.nn.moments(class2, axes=0)
-    fisher_coeffs = tf.abs((mean1 - mean2)) / (std1 + std2)
-    return tf.nn.top_k(fisher_coeffs, k=top_k)
+    fisher_coeffs = tf.abs(mean1 - mean2) / (std1 + std2)
+    return tf.nn.top_k(fisher_coeffs, k=top_k_features)
 
 
-def feature_correlation_with_class(data, num_instances: list, top_k=10):
+def feature_correlation_with_class(data, num_instances: list, top_k_features=10):
     """
     Makes feature correlation with class selection according to the following formula:
     D(f) = [(m1(f) - m(f))^2 + (m2(f) - m(f))^2] / 2*sigma(f)^2
     :return: the list of most significant features.
     """
     data = tf.convert_to_tensor(data)
+    _, num_features = data.get_shape().as_list()
+    if top_k_features < num_features:
+        top_k_features = num_features
     class1, class2 = tf.split(data, num_instances)
     mean1, std1 = tf.nn.moments(class1, axes=0)
     mean2, std2 = tf.nn.moments(class2, axes=0)
     mean, std = tf.nn.moments(data, axes=0)
-    corr_coeffs = (tf.square(mean1 - mean) + tf.square(mean2 - mean)) / 2*tf.square(std) # FIXME sth is wrong
-    return tf.nn.top_k(corr_coeffs, k=top_k)
+    corr_coeffs = (tf.square(mean1 - mean) + tf.square(mean2 - mean)) / 2*tf.square(std)
+    return tf.nn.top_k(corr_coeffs, k=top_k_features)
 
 
-def t_test(data, num_instances: list, top_k=10):
+def t_test(data, num_instances: list, top_k_features=10):
     """
     Makes feature correlation with class selection according to the following formula:
     D(f) = [(m1(f) - m(f))^2 + (m2(f) - m(f))^2] / 2*sigma(f)^2
     :return: the list of most significant features.
     """
+    data = tf.convert_to_tensor(data)
+    _, num_features = data.get_shape().as_list()
+    if top_k_features < num_features:
+        top_k_features = num_features
     class1, class2 = tf.split(data, num_instances)
     mean1, std1 = tf.nn.moments(class1, axes=0)
     mean2, std2 = tf.nn.moments(class2, axes=0)
-    t_test_coeffs = (mean1 - mean2) / tf.sqrt(tf.square(std1)/num_instances[0] + tf.square(std2) / num_instances[1])
-    return tf.nn.top_k(t_test_coeffs, k=top_k)
+    t_test_coeffs = tf.abs(mean1 - mean2) / tf.sqrt(tf.square(std1)/num_instances[0] + tf.square(std2) / num_instances[1])
+    return tf.nn.top_k(t_test_coeffs, k=top_k_features)
 
 with tf.Session() as session:
     input_data = df.as_matrix()
     instances_per_class = [82, 64]
-    fisher_coeffs = session.run(fisher(data=input_data, num_instances=instances_per_class))
-    corr_coeffs = session.run(feature_correlation_with_class(data=input_data, num_instances=instances_per_class))
-    t_test_coeff = session.run(t_test(data=input_data, num_instances=instances_per_class))
+    fisher_coeffs = session.run(fisher(data=input_data, num_instances=instances_per_class, top_k_features=5))
+    corr_coeffs = session.run(feature_correlation_with_class(data=input_data, num_instances=instances_per_class, top_k_features=5))
+    t_test_coeff = session.run(t_test(data=input_data, num_instances=instances_per_class, top_k_features=5))
     print(fisher_coeffs)
     print(corr_coeffs)
     print(t_test_coeff)

tests/pearson.py

@@ -0,0 +1,33 @@
+import tensorflow as tf
+from utils.statistics import pearson_correlation
+import numpy as np
+
+
+class PearsonTest(tf.test.TestCase):
+
+    def testPearsonCoefficientValueForTwoVectors(self):
+        with self.test_session() as test_session:
+            x1 = np.array([2., 3., 4.])
+            x2 = np.array([3., 1., 5.])
+            actual_pearson_coefficient = test_session.run(pearson_correlation(x1, x2))
+            correct_pearson_coefficient = tf.constant([.5])
+            self.assertEqual(actual_pearson_coefficient, correct_pearson_coefficient.eval())
+
+    def testNegativePearsonCoefficientValueForTwoVectors(self):
+        with self.test_session() as test_session:
+            x1 = np.array([1., 2., 3.])
+            x2 = np.array([-1., -2., -3.])
+            actual_pearson_coefficient = test_session.run(pearson_correlation(x1, x2))
+            correct_pearson_coefficient = tf.constant([-1.])
+            self.assertEqual(actual_pearson_coefficient, correct_pearson_coefficient.eval())
+
+    def testPositivePearsonCoefficientValueForTwoVectors(self):
+        with self.test_session() as test_session:
+            x1 = np.array([1., 2., 3.])
+            x2 = np.array([1., 2., 3.])
+            actual_pearson_coefficient = test_session.run(pearson_correlation(x1, x2))
+            correct_pearson_coefficient = tf.constant([1.])
+            self.assertEqual(actual_pearson_coefficient, correct_pearson_coefficient.eval())
+
+if __name__ == '__main__':
+    tf.test.main()

utils/statistics.py

@@ -1,7 +1,15 @@
+import tensorflow as tf
 
 
-def pearson_correlation():
-    pass
+def pearson_correlation(x1, x2):
+    x1 = tf.convert_to_tensor(x1)
+    x2 = tf.convert_to_tensor(x2)
+    m1, std1 = tf.nn.moments(x1, axes=0)
+    m2, std2 = tf.nn.moments(x2, axes=0)
+    l = tf.reduce_sum((x1 - m1) * (x2 - m2))
+    i = tf.reduce_sum((x1 - m1) ** 2) * tf.reduce_sum((x2 - m2) ** 2)
+    p = tf.sqrt(i)
+    return l/p
 
 
 def f_test():