Server-side compute: add /trendseries endpoint

Author: dshemetov

integrations/server/test_covidcast_endpoints.py

@@ -412,6 +412,7 @@ def match_row(trend, row):
             self.assertEqual(trend["max_date"], first.time_value)
             self.assertEqual(trend["max_value"], first.value)
             self.assertEqual(trend["max_trend"], "steady")
+
         with self.subTest("trend1"):
             trend = trends[1]
             match_row(trend, rows[1])
@@ -440,6 +441,65 @@ def match_row(trend, row):
             self.assertEqual(trend["max_value"], first.value)
             self.assertEqual(trend["max_trend"], "decreasing")
 
+        num_rows = 3
+        time_value_pairs = [(20200331, 0)] + [(20200401 + i, v) for i, v in enumerate(accumulate([num_rows - i for i in range(num_rows)]))]
+        rows = [CovidcastRow(source="jhu-csse", signal="confirmed_cumulative_num", time_value=t, value=v) for t, v in time_value_pairs]
+        self._insert_rows(rows)
+        diffed_rows = self._diff_covidcast_rows(rows)
+        for row in diffed_rows:
+            row.signal = "confirmed_incidence_num"
+        first = diffed_rows[0]
+        last = diffed_rows[-1]
+
+        out = self._fetch("/trendseries", signal="jhu-csse:confirmed_incidence_num", geo=first.geo_pair, date=last.time_value, window="20200401-20200410", basis=1)
+
+        self.assertEqual(out["result"], 1)
+        self.assertEqual(len(out["epidata"]), 3)
+        trends = out["epidata"]
+
+        with self.subTest("trend0, server-side compute"):
+            trend = trends[0]
+            match_row(trend, first)
+            self.assertEqual(trend["basis_date"], None)
+            self.assertEqual(trend["basis_value"], None)
+            self.assertEqual(trend["basis_trend"], "unknown")
+
+            self.assertEqual(trend["min_date"], last.time_value)
+            self.assertEqual(trend["min_value"], last.value)
+            self.assertEqual(trend["min_trend"], "increasing")
+            self.assertEqual(trend["max_date"], first.time_value)
+            self.assertEqual(trend["max_value"], first.value)
+            self.assertEqual(trend["max_trend"], "steady")
+
+        with self.subTest("trend1"):
+            trend = trends[1]
+            match_row(trend, diffed_rows[1])
+            self.assertEqual(trend["basis_date"], first.time_value)
+            self.assertEqual(trend["basis_value"], first.value)
+            self.assertEqual(trend["basis_trend"], "decreasing")
+
+            self.assertEqual(trend["min_date"], last.time_value)
+            self.assertEqual(trend["min_value"], last.value)
+            self.assertEqual(trend["min_trend"], "increasing")
+            self.assertEqual(trend["max_date"], first.time_value)
+            self.assertEqual(trend["max_value"], first.value)
+            self.assertEqual(trend["max_trend"], "decreasing")
+
+        with self.subTest("trend2"):
+            trend = trends[2]
+            match_row(trend, last)
+            self.assertEqual(trend["basis_date"], diffed_rows[1].time_value)
+            self.assertEqual(trend["basis_value"], diffed_rows[1].value)
+            self.assertEqual(trend["basis_trend"], "decreasing")
+
+            self.assertEqual(trend["min_date"], last.time_value)
+            self.assertEqual(trend["min_value"], last.value)
+            self.assertEqual(trend["min_trend"], "steady")
+            self.assertEqual(trend["max_date"], first.time_value)
+            self.assertEqual(trend["max_value"], first.value)
+            self.assertEqual(trend["max_trend"], "decreasing")
+
+
     def test_correlation(self):
         """Request a signal the /correlation endpoint."""
 

src/server/endpoints/covidcast.py

@@ -307,13 +307,21 @@ def handle_trendseries():
     daily_signals, weekly_signals = count_signal_time_types(source_signal_pairs)
     source_signal_pairs, alias_mapper = create_source_signal_alias_mapper(source_signal_pairs)
     geo_pairs = parse_geo_pairs()
+    transform_args = parse_transform_args()
+    jit_bypass = parse_jit_bypass()
 
     time_window, is_day = parse_day_or_week_range_arg("window")
     _verify_argument_time_type_matches(is_day, daily_signals, weekly_signals)
     basis_shift = extract_integer(("basis", "basis_shift"))
     if basis_shift is None:
         basis_shift = 7
 
+    use_server_side_compute = is_day and JIT_COMPUTE and not jit_bypass
+    if use_server_side_compute:
+        pad_length = get_pad_length(source_signal_pairs, transform_args.get("smoother_window_length"))
+        source_signal_pairs, row_transform_generator = get_basename_signals(source_signal_pairs)
+        time_window = pad_time_window(time_window, pad_length)
+
     # build query
     q = QueryBuilder("covidcast", "t")
 
@@ -336,8 +344,20 @@ def handle_trendseries():
     if not is_day:
         shifter = lambda x: shift_week_value(x, -basis_shift)
 
-    def gen(rows):
-        for key, group in groupby((parse_row(row, fields_string, fields_int, fields_float) for row in rows), lambda row: (row["geo_type"], row["geo_value"], row["source"], row["signal"])):
+    if use_server_side_compute:
+        def gen_transform(rows):
+            parsed_rows = (parse_row(row, fields_string, fields_int, fields_float) for row in rows)
+            transformed_rows = row_transform_generator(parsed_rows=parsed_rows, time_pairs=[TimePair("day", [time_window])], transform_args=transform_args)
+            for row in transformed_rows:
+                yield row
+    else:
+        def gen_transform(rows):
+            parsed_rows = (parse_row(row, fields_string, fields_int, fields_float) for row in rows)
+            for row in parsed_rows:
+                yield row
+
+    def gen_trend(rows):
+        for key, group in groupby(rows, lambda row: (row["geo_type"], row["geo_value"], row["source"], row["signal"])):
             geo_type, geo_value, source, signal = key
             if alias_mapper:
                 source = alias_mapper(source, signal)
@@ -352,7 +372,7 @@ def gen(rows):
         raise DatabaseErrorException(str(e))
 
     # now use a generator for sending the rows and execute all the other queries
-    return p(filter_fields(gen(r)))
+    return p(filter_fields(gen_trend(gen_transform(r))))
 
 
 @bp.route("/correlation", methods=("GET", "POST"))

src/server/endpoints/covidcast_utils/trend.py

@@ -75,6 +75,8 @@ def compute_trends(geo_type: str, geo_value: str, signal_source: str, signal_sig
     lookup: Dict[int, float] = OrderedDict()
     # find all needed rows
     for time, value in rows:
+        if value is None:
+            continue
         lookup[time] = value
         if min_value is None or min_value > value:
             min_date = time