Update statistics.py (#32)

* Update statistics.py

* add percentile(autocorrelation) statistic
* add power statistic

* fix flake8

* Update requirements.txt

* add tests

requirements/requirements.txt

@@ -3,6 +3,7 @@ tensorflow==2.9.1
 tensorflow_probability==0.17.0
 scipy>=1.7.3
 numpy>=1.21.6
+statsmodels
 dtaidistance==2.3.10
 networkx
 optuna

tests/test_metrics.py

@@ -19,7 +19,7 @@ def test_statistics():
     assert (tsgm.metrics.statistics.axis_max_s(ts, axis=1) == [11, 21]).all()
     assert (tsgm.metrics.statistics.axis_min_s(ts, axis=1) == [0, -11]).all()
 
-    assert (tsgm.metrics.statistics.axis_max_s(ts, axis=2) == [21, 11,  8]).all()
+    assert (tsgm.metrics.statistics.axis_max_s(ts, axis=2) == [21, 11, 8]).all()
     assert (tsgm.metrics.statistics.axis_min_s(ts, axis=2) == [0, -11, 1]).all()
 
     assert (tsgm.metrics.statistics.axis_mode_s(ts, axis=None) == [1]).all()
@@ -30,6 +30,13 @@ def test_statistics():
 
     assert (tsgm.metrics.statistics.axis_percentile_s(ts, axis=None, percentile=50) - np.asarray([2]) < eps).all()
 
+    assert (tsgm.metrics.statistics.axis_percautocorr_s(ts, axis=None) - np.asarray([-0.245016]) < eps).all()
+    assert (tsgm.metrics.statistics.axis_percautocorr_s(ts, axis=1) - np.asarray([-0.48875, -0.48875]) < eps).all()
+
+    assert (tsgm.metrics.statistics.axis_power_s(ts, axis=None) - np.asarray([74.5]) < eps).all()
+    assert (tsgm.metrics.statistics.axis_power_s(ts, axis=1) - np.asarray([1869.61111111, 15148.72222222]) < eps).all()
+    assert (tsgm.metrics.statistics.axis_power_s(ts, axis=2) - np.asarray([36587.13, 7321., 1253.13]) < eps).all()
+
     # Now, checking with tf.Tensor
     ts_tf = tf.convert_to_tensor(ts)
 
@@ -39,7 +46,7 @@ def test_statistics():
     assert (tsgm.metrics.statistics.axis_max_s(ts_tf, axis=1) == [11, 21]).all()
     assert (tsgm.metrics.statistics.axis_min_s(ts_tf, axis=1) == [0, -11]).all()
 
-    assert (tsgm.metrics.statistics.axis_max_s(ts_tf, axis=2) == [21, 11,  8]).all()
+    assert (tsgm.metrics.statistics.axis_max_s(ts_tf, axis=2) == [21, 11, 8]).all()
     assert (tsgm.metrics.statistics.axis_min_s(ts_tf, axis=2) == [0, -11, 1]).all()
 
     assert (tsgm.metrics.statistics.axis_mode_s(ts_tf, axis=None) == [1]).all()
@@ -49,6 +56,13 @@ def test_statistics():
 
     assert (tsgm.metrics.statistics.axis_percentile_s(ts_tf, axis=None, percentile=50) - np.asarray([2]) < eps).all()
 
+    assert (tsgm.metrics.statistics.axis_percautocorr_s(ts_tf, axis=None) - np.asarray([-0.245016]) < eps).all()
+    assert (tsgm.metrics.statistics.axis_percautocorr_s(ts_tf, axis=1) - np.asarray([-0.48875, -0.48875]) < eps).all()
+
+    assert (tsgm.metrics.statistics.axis_power_s(ts_tf, axis=None) - np.asarray([74.5]) < eps).all()
+    assert (tsgm.metrics.statistics.axis_power_s(ts_tf, axis=1) - np.asarray([1869.61111111, 15148.72222222]) < eps).all()
+    assert (tsgm.metrics.statistics.axis_power_s(ts_tf, axis=2) - np.asarray([36587.13, 7321., 1253.13]) < eps).all()
+
 
 def test_distance_metric():
     ts = np.array([[[0, 2], [11, -11], [1, 2]], [[10, 21], [1, -1], [6, 8]]])
@@ -66,7 +80,7 @@ def test_distance_metric():
     )
     assert dist_metric(ts, diff_ts) > dist_metric(ts, sim_ts)
     stat_results = dist_metric.stats(ts)
-    
+
     assert len(stat_results) == 6
     assert dist_metric._discrepancy(dist_metric.stats(ts), dist_metric.stats(sim_ts)) == dist_metric(ts, sim_ts)
     assert dist_metric(ts, sim_ts) != dist_metric(ts, diff_ts)
@@ -83,7 +97,6 @@ def test_distance_metric():
     assert dist_metric(ds, ds_diff) == dist_metric(ds_diff, ds)
 
 
-
 class MockEvaluator:
     def evaluate(self, D: tsgm.dataset.Dataset, Dtest: tsgm.dataset.Dataset) -> float:
         return 0.42
@@ -127,7 +140,7 @@ def test_downstream_performance_metric():
     assert downstream_perf_metric(D1, D2, D_test) == downstream_perf_metric(ts, diff_ts, test_ts)
     assert downstream_perf_metric(D1, D2, D_test) == downstream_perf_metric(D1, diff_ts, D_test)
     assert downstream_perf_metric(D1, D2, D_test) == downstream_perf_metric(ts, D2, D_test)
-    mean, std =  downstream_perf_metric(D1, D2, D_test, return_std=True)
+    mean, std = downstream_perf_metric(D1, D2, D_test, return_std=True)
     assert mean == 0 and std == 0
 
 

tsgm/metrics/statistics.py

@@ -2,6 +2,8 @@
 import numpy as np
 import scipy
 import functools
+import tensorflow as tf
+from statsmodels.tsa.stattools import acf
 
 import tsgm
 
@@ -15,6 +17,14 @@ def _validate_axis(axis: typing.Optional[int]):
     assert axis == 1 or axis == 2 or axis is None
 
 
+def _apply_percacf(x):
+    return np.percentile(acf(x), .75)
+
+
+def _apply_power(x):
+    return np.power(x, 2).sum() / len(x)
+
+
 def axis_max_s(ts: tsgm.types.Tensor, axis: typing.Optional[int]) -> tsgm.types.Tensor:
     _validate_axis(axis)
 
@@ -49,3 +59,17 @@ def axis_percentile_s(ts: tsgm.types.Tensor, axis: typing.Optional[int], percent
     _validate_axis(axis)
 
     return np.percentile(ts, percentile, axis=axis).flatten()
+
+
+def axis_percautocorr_s(ts: tsgm.types.Tensor, axis: typing.Optional[int]) -> tsgm.types.Tensor:
+    _validate_axis(axis)
+
+    return np.array([_apply_percacf(tf.reshape(ts, [-1]))]) if axis is None else \
+        np.apply_along_axis(_apply_percacf, 0, np.apply_along_axis(_apply_percacf, axis, ts))
+
+
+def axis_power_s(ts: tsgm.types.Tensor, axis: typing.Optional[int]) -> tsgm.types.Tensor:
+    _validate_axis(axis)
+
+    return np.array([_apply_power(tf.reshape(ts, [-1]))]) if axis is None else \
+        np.apply_along_axis(_apply_power, 0, np.apply_along_axis(_apply_power, axis, ts))