Feature/data versioning and optimization

docs/api/covidcast.md

@@ -94,6 +94,38 @@ and lists.
 The current set of signals available for each data source is returned by the
 [`covidcast_meta`](covidcast_meta.md) endpoint.
 
+#### Optional
+
+The default API behavior is to return the most recently issued value for each `time_value` selected.
+
+We also provide access to previous versions of data using the optional parameters below.
+
+| Parameter | Description | Type |
+| --- | --- | --- |
+| `as_of` | maximum time unit (e.g., date) when the signal data were published (return most recent for each `time_value`) | time value (e.g., 20200401) |
+| `issues` | time unit (e.g., date) when the signal data were published (return all matching records for each `time_value`) | `list` of time values (e.g., 20200401) |
+| `lag` | time delta (e.g. days) between when the underlying events happened and when the data were published | integer |
+
+Use cases:
+
+* To pretend like you queried the API on June 1, such that the returned results
+  do not include any updates which became available after June 1, use
+  `as_of=20200601`.
+* To retrieve only data that was published or updated on June 1, and exclude
+  records whose most recent update occured earlier than June 1, use
+  `issues=20200601`.
+* To retrieve all data that was published between May 1 and June 1, and exclude
+  records whose most recent update occured earlier than May 1, use
+  `issues=20200501-20200601`. The results will include all matching issues for
+  each `time_value`, not just the most recent.
+* To retrieve only data that was published or updated exactly 3 days after the
+  underlying events occurred, use `lag=3`.
+
+NB: Each issue in the versioning system contains only the records that were
+added or updated during that time unit; we exclude records whose values remain
+the same as a previous issue. If you have a research problem that would require
+knowing when an unchanged value was last confirmed, please get in touch.
+
 ### Response
 
 | Field | Description | Type |
@@ -106,6 +138,8 @@ The current set of signals available for each data source is returned by the
 | `epidata[].value` | value (statistic) derived from the underlying data source | float |
 | `epidata[].stderr` | approximate standard error of the statistic with respect to its sampling distribution, `null` when not applicable | float |
 | `epidata[].sample_size` | number of "data points" used in computing the statistic, `null` when not applicable | float |
+| `epidata[].issue` | time unit (e.g. date) when this statistic was published | integer |
+| `epidata[].lag` | time delta (e.g. days) between when the underlying events happened and when this statistic was published | integer |
 | `message` | `success` or error message | string |
 
 **Note:** `result` code 2, "too many results", means that the number of results

integrations/acquisition/covidcast/test_covidcast_meta_caching.py

@@ -66,12 +66,14 @@ def test_caching(self):
     self.cur.execute('''
       insert into covidcast values
         (0, 'src', 'sig', 'day', 'state', 20200422, 'pa',
-          123, 1, 2, 3, 456, 1)
+          123, 1, 2, 3, 456, 1, 20200422, 0),
+        (0, 'src', 'sig', 'day', 'state', 20200422, 'wa',
+          789, 1, 2, 3, 456, 1, 20200423, 1)
     ''')
     self.cur.execute('''
       insert into covidcast values
         (100, 'src', 'wip_sig', 'day', 'state', 20200422, 'pa',
-          456, 4, 5, 6, 789, -1)
+          456, 4, 5, 6, 789, -1, 20200422, 0)
     ''')
 
     self.cnx.commit()
@@ -90,12 +92,15 @@ def test_caching(self):
         'geo_type': 'state',
         'min_time': 20200422,
         'max_time': 20200422,
-        'num_locations': 1,
-        'last_update': 123,
-        'min_value': 1,
-        'max_value': 1,
-        'mean_value': 1,
-        'stdev_value': 0,
+        'num_locations': 2,
+        'last_update': 789,
+        'min_value': 1.0,
+        'max_value': 1.0,
+        'mean_value': '1.0000000',
+        'stdev_value': '0.0000000',
+        'max_issue': 20200423,
+        'min_lag': 0,
+        'max_lag': 1
       }
     ])
     epidata1={'result':1,'message':'success','epidata':epidata1}

integrations/acquisition/covidcast/test_csv_uploading.py

@@ -1,6 +1,7 @@
 """Integration tests for covidcast's CSV-to-database uploading."""
 
 # standard library
+from datetime import date
 import os
 import unittest
 from unittest.mock import MagicMock
@@ -95,11 +96,24 @@ def test_uploading(self):
     response = Epidata.covidcast(
         'src-name', 'test', 'day', 'state', 20200419, '*')
 
+
+    expected_issue_day=date.today()
+    expected_issue=expected_issue_day.strftime("%Y%m%d")
+    def apply_lag(expected_epidata):
+      for dct in expected_epidata:
+        dct['issue'] = int(expected_issue)
+        time_value_day = date(year=dct['time_value'] // 10000,
+                              month=dct['time_value'] % 10000 // 100,
+                              day= dct['time_value'] % 100)
+        expected_lag = (expected_issue_day - time_value_day).days
+        dct['lag'] = expected_lag
+      return expected_epidata
+
     # verify data matches the CSV
     # NB these are ordered by geo_value
     self.assertEqual(response, {
       'result': 1,
-      'epidata': [
+      'epidata': apply_lag([
         {
           'time_value': 20200419,
           'geo_value': 'ca',
@@ -124,19 +138,20 @@ def test_uploading(self):
           'sample_size': 20,
           'direction': None,
         },
-       ],
+    ]),
       'message': 'success',
     })
 
     # request CSV data from the API on WIP signal
     response = Epidata.covidcast(
       'src-name', 'wip_prototype', 'day', 'state', 20200419, '*')
 
+
     # verify data matches the CSV
     # NB these are ordered by geo_value
     self.assertEqual(response, {
       'result': 1,
-      'epidata': [
+      'epidata': apply_lag([
         {
           'time_value': 20200419,
           'geo_value': 'me',
@@ -161,7 +176,7 @@ def test_uploading(self):
           'sample_size': 300,
           'direction': None,
         },
-       ],
+       ]),
       'message': 'success',
     })
 

integrations/acquisition/covidcast/test_direction_updating.py

@@ -61,23 +61,23 @@ def test_uploading(self):
     self.cur.execute('''
       insert into covidcast values
         (0, 'src', 'sig', 'day', 'state', 20200228, 'ca',
-          123, 2, 0, 0, 0, NULL),
+          123, 2, 0, 0, 0, NULL, 20200228, 0),
         (0, 'src', 'sig', 'day', 'state', 20200229, 'ca',
-          123, 6, 0, 0, 0, NULL),
+          123, 6, 0, 0, 0, NULL, 20200229, 0),
         (0, 'src', 'sig', 'day', 'state', 20200301, 'ca',
-          123, 5, 0, 0, 0, NULL),
+          123, 5, 0, 0, 0, NULL, 20200301, 0),
         (0, 'src', 'sig', 'day', 'state', 20200511, 'fl',
-          123, 1, 0, 0, 0, NULL),
+          123, 1, 0, 0, 0, NULL, 20200511, 0),
         (0, 'src', 'sig', 'day', 'state', 20200512, 'fl',
-          123, 2, 0, 0, 0, NULL),
+          123, 2, 0, 0, 0, NULL, 20200512, 0),
         (0, 'src', 'sig', 'day', 'state', 20200517, 'fl',
-          123, 2, 0, 0, 0, NULL),
+          123, 2, 0, 0, 0, NULL, 20200517, 0),
         (0, 'src', 'sig', 'day', 'state', 20200615, 'tx',
-          123, 9, 0, 0, 456, NULL),
+          123, 9, 0, 0, 456, NULL, 20200615, 0),
         (0, 'src', 'sig', 'day', 'state', 20200616, 'tx',
-          123, 5, 0, 0, 456, NULL),
+          123, 5, 0, 0, 456, NULL, 20200616, 0),
         (0, 'src', 'sig', 'day', 'state', 20200617, 'tx',
-          123, 1, 0, 0, 456, 1)
+          123, 1, 0, 0, 456, 1, 20200617, 0)
     ''')
     self.cnx.commit()
 
@@ -100,6 +100,8 @@ def test_uploading(self):
           'stderr': 0,
           'sample_size': 0,
           'direction': None,
+          'issue': 20200228,
+          'lag': 0
         },
         {
           'time_value': 20200229,
@@ -108,6 +110,8 @@ def test_uploading(self):
           'stderr': 0,
           'sample_size': 0,
           'direction': None,
+          'issue': 20200229,
+          'lag': 0
         },
         {
           'time_value': 20200301,
@@ -116,6 +120,8 @@ def test_uploading(self):
           'stderr': 0,
           'sample_size': 0,
           'direction': 1,
+          'issue': 20200301,
+          'lag': 0
         },
         {
           'time_value': 20200511,
@@ -124,6 +130,8 @@ def test_uploading(self):
           'stderr': 0,
           'sample_size': 0,
           'direction': None,
+          'issue': 20200511,
+          'lag': 0
         },
         {
           'time_value': 20200512,
@@ -132,6 +140,8 @@ def test_uploading(self):
           'stderr': 0,
           'sample_size': 0,
           'direction': None,
+          'issue': 20200512,
+          'lag': 0
         },
         {
           'time_value': 20200517,
@@ -140,6 +150,8 @@ def test_uploading(self):
           'stderr': 0,
           'sample_size': 0,
           'direction': 0,
+          'issue': 20200517,
+          'lag': 0
         },
         {
           'time_value': 20200615,
@@ -148,6 +160,8 @@ def test_uploading(self):
           'stderr': 0,
           'sample_size': 0,
           'direction': None,
+          'issue': 20200615,
+          'lag': 0
         },
         {
           'time_value': 20200616,
@@ -156,6 +170,8 @@ def test_uploading(self):
           'stderr': 0,
           'sample_size': 0,
           'direction': None,
+          'issue': 20200616,
+          'lag': 0
         },
         {
           'time_value': 20200617,
@@ -164,6 +180,8 @@ def test_uploading(self):
           'stderr': 0,
           'sample_size': 0,
           'direction': 1,
+          'issue': 20200617,
+          'lag': 0
         },
        ],
       'message': 'success',