Server-side compute: add 7day averaging

dshemetov · dshemetov · commit 5f72ebec464c · 2021-09-28T04:19:03.000-07:00
diff --git a/integrations/server/test_covidcast_endpoints.py b/integrations/server/test_covidcast_endpoints.py
@@ -225,6 +225,33 @@ def test_derived_signals(self):
             expected_values = self._diff_rows(values1) + self._diff_rows(values2)
             self.assertAlmostEqual(out_values, expected_values)
 
+        with self.subTest("smoothed signal"):
+            out = self._fetch("/", signal="jhu-csse:confirmed_7dav_incidence_num", geo=first.geo_pair, time="day:20200401-20200410")
+            self.assertEqual(len(out["epidata"]), len(rows01))
+            out_values = [row["value"] for row in out["epidata"]]
+            values = [None] * 7 + [value for _, value in time_value_pairs]
+            expected_values = self._smooth_rows(self._diff_rows(values))
+            self.assertAlmostEqual(out_values, expected_values)
+
+        with self.subTest("diffed signal and smoothed signal in one request"):
+            out = self._fetch("/", signal="jhu-csse:confirmed_incidence_num;jhu-csse:confirmed_7dav_incidence_num", geo=first.geo_pair, time="day:20200401-20200410")
+            self.assertEqual(len(out["epidata"]), 2*len(rows01) + 6)
+            out_values = [row["value"] for row in out["epidata"]]
+            values = [None] * 7 + [value for _, value in time_value_pairs]
+            expected_diff = self._diff_rows(values)
+            expected_smoothed = self._smooth_rows(expected_diff)
+            expected_values = list(interleave_longest(expected_smoothed, expected_diff))
+            self.assertAlmostEqual(out_values, expected_values)
+
+        with self.subTest("smoothing with time window resizing"):
+            # should fetch 6 extra days
+            out = self._fetch("/", signal="jhu-csse:confirmed_7dav_incidence_num", geo=first.geo_pair, time="day:20200407-20200410")
+            self.assertEqual(len(out["epidata"]), len(rows01) - 6)
+            out_values = [row["value"] for row in out["epidata"]]
+            # an extra None is added because the padding for DIFF_SMOOTH (pad_length = 8) is used
+            values = [None] + [value for _, value in time_value_pairs]
+            expected_values = self._smooth_rows(self._diff_rows(values))
+            self.assertAlmostEqual(out_values, expected_values)
 
         with self.subTest("diffing with time window resizing"):
             # should fetch 1 extra day
@@ -240,6 +267,15 @@ def test_derived_signals(self):
         first = rows[0]
         self._insert_rows(rows)
 
+        with self.subTest("smoothing with a time gap"):
+            # should fetch 1 extra day
+            out = self._fetch("/", signal="jhu-csse:confirmed_7dav_incidence_num", geo=first.geo_pair, time="day:20200401-20200420")
+            self.assertEqual(len(out["epidata"]), len(rows) + 5)
+            out_values = [row["value"] for row in out["epidata"]]
+            values = [None] * 7 + [value for _, value in time_value_pairs][:10] + [None] * 5 + [value for _, value in time_value_pairs][10:]
+            expected_values = self._smooth_rows(self._diff_rows(values))
+            self.assertAlmostEqual(out_values, expected_values)
+
         with self.subTest("diffing with a time gap"):
             # should fetch 1 extra day
             out = self._fetch("/", signal="jhu-csse:confirmed_incidence_num", geo=first.geo_pair, time="day:20200401-20200420")
diff --git a/src/server/endpoints/covidcast_utils/smooth_diff.py b/src/server/endpoints/covidcast_utils/smooth_diff.py
@@ -5,6 +5,7 @@
 from more_itertools import windowed
 from numpy import nan, nan_to_num, array, dot, isnan
 
+from delphi_utils.nancodes import Nans
 
 class PadFillValue(str, Enum):
     first = "first"
@@ -16,6 +17,27 @@ class SmootherKernelValue(str, Enum):
     average = "average"
 
 
+def _smoother(values: List[Number], kernel: Optional[Union[List[Number], SmootherKernelValue]] = None) -> Number:
+    """Basic smoother.
+
+    If kernel passed, uses the kernel as summation weights. If something is wrong,
+    defaults to the mean.
+    """
+    try:
+        if kernel and isinstance(kernel, list):
+            kernel = array(kernel, copy=False)
+            values = array(values, copy=False)
+            smoothed_value = dot(values, kernel)
+            return smoothed_value
+        else:
+            raise ValueError
+    except (ValueError, TypeError):
+        try:
+            smoothed_value = array(values).mean()
+        except (ValueError, TypeError):
+            smoothed_value = None
+
+    return smoothed_value
 
 
 def generate_smooth_rows(
@@ -52,8 +74,25 @@ def generate_smooth_rows(
     **kwargs:
         Container for non-shared parameters with other computation functions.
     """
-    for window in windowed(rows, smoother_window_length):
-        yield smoothed_row(window)
+    # Validate params.
+    if not isinstance(smoother_window_length, int) or (isinstance(smoother_window_length, int) and smoother_window_length < 1):
+        smoother_window_length = 7
+    if isinstance(smoother_kernel, list):
+        smoother_window_length = len(smoother_kernel)
+    if not isinstance(nan_fill_value, Number):
+        nan_fill_value = nan
+    if not isinstance(smoother_kernel, (list, SmootherKernelValue)):
+        smoother_kernel = SmootherKernelValue.average
+
+    for window in windowed(rows, smoother_window_length):  # Iterable[List[Dict]]
+        if all(e is None for e in window):
+            continue
+        value = _smoother(nan_to_num([e.get("value") if e else nan_fill_value for e in window], nan=nan_fill_value), smoother_kernel)
+        value = float(round(value, 8)) if value and not isnan(value) else None
+        last_item = list(filter(lambda e: e is not None, window))[-1]
+        item = last_item.copy() # inherit values of last time stamp
+        item.update({"value": value, "stderr": None, "sample_size": None, "missing_value": Nans.NOT_MISSING if value is not None else Nans.NOT_APPLICABLE, "missing_stderr": Nans.NOT_APPLICABLE, "missing_sample_size": Nans.NOT_APPLICABLE})
+        yield item
 
 
 def generate_row_diffs(rows: Iterable[Dict], nan_fill_value: Number = nan, **kwargs) -> Iterable[Dict]:
diff --git a/tests/server/endpoints/covidcast_utils/test_smooth_diff.py b/tests/server/endpoints/covidcast_utils/test_smooth_diff.py
@@ -5,10 +5,90 @@
 from more_itertools import windowed
 
 from delphi.epidata.server.utils import date_to_time_value, CovidcastRecords
-from delphi.epidata.server.endpoints.covidcast_utils.smooth_diff import generate_row_diffs, generate_smooth_rows
+from delphi.epidata.server.endpoints.covidcast_utils.smooth_diff import generate_row_diffs, generate_smooth_rows, _smoother
 
 
 class TestStreaming:
+    def test__smoother(self):
+        assert _smoother(list(range(7)), [1] * 7) == sum(range(7))
+        assert _smoother([1] * 6, list(range(7))) == sum([1] * 6) / 6
+
+
+    def test_generate_smooth_rows(self):
+        # an empty dataframe should return an empty dataframe
+        data = DataFrame({})
+        smoothed_df = DataFrame.from_records(generate_smooth_rows(data.to_dict(orient='records')))
+        expected_df = DataFrame({})
+        assert_frame_equal(smoothed_df, expected_df)
+
+        # a dataframe with a single entry should return a single nan value
+        data = CovidcastRecords(
+            time_values=[20210501],
+            values=[1.0]
+        ).as_dataframe()
+        smoothed_df = DataFrame.from_records(generate_smooth_rows(data.to_dict(orient='records')))
+        expected_df = CovidcastRecords(
+            time_values=[20210501],
+            values=[None],
+            stderrs=[None],
+            sample_sizes=[None]
+        ).as_dataframe()
+        assert_frame_equal(smoothed_df, expected_df)
+
+        data = CovidcastRecords(
+            time_values=date_range("2021-05-01", "2021-05-10"),
+            values=chain(range(7), [None, 2., 1.])
+        ).as_dataframe()
+
+        # regular window, nan fill
+        smoothed_df = DataFrame.from_records(generate_smooth_rows(data.to_dict(orient='records')))
+        expected_df = CovidcastRecords(
+            time_values=date_range("2021-05-07", "2021-05-10"),
+            values=(sum(x)/len(x) if None not in x else None for x in windowed(chain(range(7), [None, 2., 1.]), 7)),
+            stderrs=[None]*4,
+            sample_sizes=[None]*4,
+        ).as_dataframe()
+        assert_frame_equal(smoothed_df, expected_df)
+
+        # regular window, 0 fill
+        smoothed_df = DataFrame.from_records(generate_smooth_rows(data.to_dict(orient='records'), nan_fill_value=0.))
+        expected_df = CovidcastRecords(
+            time_values=date_range("2021-05-07", "2021-05-10"),
+            values=(sum(x)/len(x) if None not in x else None for x in windowed(chain(range(7), [0., 2., 1.]), 7)),
+            stderrs=[None]*4,
+            sample_sizes=[None]*4,
+        ).as_dataframe()
+        assert_frame_equal(smoothed_df, expected_df)
+
+        # regular window, different window length
+        smoothed_df = DataFrame.from_records(generate_smooth_rows(data.to_dict(orient='records'), smoother_window_length = 8))
+        expected_df = CovidcastRecords(
+            time_values=date_range("2021-05-08", "2021-05-10"),
+            values=(sum(x)/len(x) if None not in x else None for x in windowed(chain(range(7), [None, 2., 1.]), 8)),
+            stderrs=[None]*3,
+            sample_sizes=[None]*3,
+        ).as_dataframe()
+        assert_frame_equal(smoothed_df, expected_df)
+
+        # regular window, different kernel
+        smoothed_df = DataFrame.from_records(generate_smooth_rows(data.to_dict(orient='records'), smoother_kernel = list(range(8))))
+        expected_df = CovidcastRecords(
+            time_values=date_range("2021-05-08", "2021-05-10"),
+            values=(sum([i * j for i, j in zip(x, range(8))])/len(x) if None not in x else None for x in windowed(chain(range(7), [None, 2., 1.]), 8)),
+            stderrs=[None]*3,
+            sample_sizes=[None]*3,
+        ).as_dataframe()
+        assert_frame_equal(smoothed_df, expected_df)
+
+        # conflicting smoother args validation
+        smoothed_df = DataFrame.from_records(generate_smooth_rows(data.to_dict(orient='records'), smoother_kernel=[1/7.]*7, smoother_window_length=10))
+        expected_df = CovidcastRecords(
+            time_values=date_range("2021-05-07", "2021-05-10"),
+            values=(sum([i * j for i, j in zip(x, [1/7.]*7)]) if None not in x else None for x in windowed(chain(range(7), [None, 2., 1.]), 7)),
+            stderrs=[None]*4,
+            sample_sizes=[None]*4,
+        ).as_dataframe()
+        assert_frame_equal(smoothed_df, expected_df)
 
 
     def test_generate_row_diffs(self):
