Repository: cmu-delphi/covidcast-indicators
Issue: Combo cases/deaths producing duplicate geos

It seems that combo cases/deaths is pulling some geo ids from both usa-facts _and_ jhu-csse, which causes the archiver to balk. here's a sample file with duplicate geos:

```
$ for f in receiving/*.csv; do echo $f; cut -d, -f1 $f | sort | uniq -d; done 
[...]
receiving/20210227_county_confirmed_7dav_cumulative_num.csv
46000
46075
46107
46129
46135
```

Here's the error we get from archiver:

```
Traceback (most recent call last):
  File "/home/krivard/projects/covid/auto/covidcast-indicators/_delphi_utils_python/delphi_utils/archive.py", line 89, in diff_export_csv
    after_df_cmn = after_df.reindex(common_idx)
  File "/home/krivard/projects/covid/auto/covidcast-indicators/combo_cases_and_deaths/env/lib/python3.8/site-packages/pandas/util/_decorators.py", line 312, in wrapper
    return func(*args, **kwargs)
  File "/home/krivard/projects/covid/auto/covidcast-indicators/combo_cases_and_deaths/env/lib/python3.8/site-packages/pandas/core/frame.py", line 4176, in reindex
    return super().reindex(**kwargs)
  File "/home/krivard/projects/covid/auto/covidcast-indicators/combo_cases_and_deaths/env/lib/python3.8/site-packages/pandas/core/generic.py", line 4811, in reindex
    return self._reindex_axes(
  File "/home/krivard/projects/covid/auto/covidcast-indicators/combo_cases_and_deaths/env/lib/python3.8/site-packages/pandas/core/frame.py", line 4022, in _reindex_axes
    frame = frame._reindex_index(
  File "/home/krivard/projects/covid/auto/covidcast-indicators/combo_cases_and_deaths/env/lib/python3.8/site-packages/pandas/core/frame.py", line 4041, in _reindex_index
    return self._reindex_with_indexers(
  File "/home/krivard/projects/covid/auto/covidcast-indicators/combo_cases_and_deaths/env/lib/python3.8/site-packages/pandas/core/generic.py", line 4877, in _reindex_with_indexers
    new_data = new_data.reindex_indexer(
  File "/home/krivard/projects/covid/auto/covidcast-indicators/combo_cases_and_deaths/env/lib/python3.8/site-packages/pandas/core/internals/managers.py", line 1301, in reindex_indexer
    self.axes[axis]._can_reindex(indexer)
  File "/home/krivard/projects/covid/auto/covidcast-indicators/combo_cases_and_deaths/env/lib/python3.8/site-packages/pandas/core/indexes/base.py", line 3476, in _can_reindex
    raise ValueError("cannot reindex from a duplicate axis")
ValueError: cannot reindex from a duplicate axis
```

Changed files:
combo_cases_and_deaths/delphi_combo_cases_and_deaths/run.py, combo_cases_and_deaths/tests/test_run.py

diff --git a/combo_cases_and_deaths/delphi_combo_cases_and_deaths/run.py b/combo_cases_and_deaths/delphi_combo_cases_and_deaths/run.py
index 558cd694..a78b4a35 100755
--- a/combo_cases_and_deaths/delphi_combo_cases_and_deaths/run.py
+++ b/combo_cases_and_deaths/delphi_combo_cases_and_deaths/run.py
@@ -134,6 +134,14 @@ def combine_usafacts_and_jhu(signal, geo, date_range, issue_range=None, fetcher=
             # se and sample size are required later so we add them back.
             combined_df["se"] = combined_df["sample_size"] = None
         combined_df.rename({geo: "geo_id"}, axis=1, inplace=True)
+
+    # default sort from API is ORDER BY signal, time_value, geo_value, issue
+    # we want to drop all but the most recent (last) issue
+    combined_df.drop_duplicates(
+        subset=["geo_id", "timestamp"],
+        keep="last",
+        inplace=True
+    )
     return combined_df
 
 def extend_raw_date_range(params, sensor_name):
diff --git a/combo_cases_and_deaths/tests/test_run.py b/combo_cases_and_deaths/tests/test_run.py
index b87018bf..6596f4f4 100644
--- a/combo_cases_and_deaths/tests/test_run.py
+++ b/combo_cases_and_deaths/tests/test_run.py
@@ -39,36 +39,56 @@ All variants: {variants}
 Date-extended variants: {variants_changed}
 """
 
-
-def test_unstable_sources():
+@patch("covidcast.covidcast.signal")
+def test_unstable_sources(mock_covidcast_signal):
     """Verify that combine_usafacts_and_jhu assembles the combined data
     frame correctly for all cases where 0, 1, or both signals are
     available.
     """
-    placeholder = lambda geo: pd.DataFrame(
-        [(date.today(),"pr" if geo == "state" else "72000",1,1,1)],
-        columns="time_value geo_value value stderr sample_size".split())
-    fetcher10 = lambda *x, **k: placeholder(x[-1]) if x[0] == "usa-facts" else None
-    fetcher01 = lambda *x, **k: placeholder(x[-1]) if x[0] == "jhu-csse" else None
-    fetcher11 = lambda *x, **k: placeholder(x[-1])
-    fetcher00 = lambda *x, **k: None
-
+    def jhu(geo, c=[0]):
+        c[0] += 1
+        return pd.DataFrame(
+            [(date.fromordinal(c[0]),"pr" if geo == "state" else "72000",1,1,1)],
+            columns="time_value geo_value value stderr sample_size".split())
+    def uf(geo, c=[0]):
+        c[0] += 1
+        return pd.DataFrame(
+            [(date.fromordinal(c[0]),"ny" if geo == "state" else "36000",1,1,1)],
+            columns="time_value geo_value value stderr sample_size".split())
+    def make_mock(geo):
+        return [
+            # 1 0
+            uf(geo), None,
+            # 0 1
+            None, jhu(geo),
+            # 1 1
+            uf(geo), jhu(geo),
+            # 0 0
+            None, None
+        ]
+
+    geos = ["state", "county", "msa", "nation", "hhs"]
+    outputs = [df for g in geos for df in make_mock(g)]
+    mock_covidcast_signal.side_effect = outputs[:]
+    
     date_range = [date.today(), date.today()]
 
-    for geo in ["state", "county", "msa", "nation", "hhs"]:
-        for (fetcher, expected_size) in [
-                (fetcher00, 0),
-                (fetcher01, 0 if geo == "msa" else 1),
-                (fetcher10, 1),
-                (fetcher11, 1 if geo in ["msa", "nation", "hhs"] else 2)
+    calls = -1
+    for geo in geos:
+        for config, expected_size in [
+                ("1 0", 1),
+                ("0 1", 0 if geo == "msa" else 1),
+                ("1 1", 1 if geo in ["msa", "nation", "hhs"] else 2),
+                ("0 0", 0)
         ]:
-            df = combine_usafacts_and_jhu("", geo, date_range, fetcher=fetcher)
+            calls += 1
+            df = combine_usafacts_and_jhu("", geo, date_range, fetcher=mock_covidcast_signal)
             assert df.size == expected_size * len(COLUMN_MAPPING), f"""
 Wrong number of rows in combined data frame for the number of available signals.
 
-input for {geo}:
-{fetcher('usa-facts',geo)}
-{fetcher('jhu-csse',geo)}
+input for {geo} {config}:
+{outputs[2*calls]}
+{outputs[2*calls + 1]}
 
 output:
 {df}
@@ -77,6 +97,32 @@ expected rows: {expected_size}
 """
 
 
+@patch("covidcast.covidcast.signal")
+def test_multiple_issues(mock_covidcast_signal):
+    """Verify that only the most recent issue is retained."""
+    mock_covidcast_signal.side_effect = [
+        pd.DataFrame({
+            "geo_value": ["01000", "01000"],
+            "value": [1, 10],
+            "timestamp": [20200101, 20200101],
+            "issue": [20200102, 20200104]
+        }),
+        None
+    ]
+    result = combine_usafacts_and_jhu("confirmed_incidence_num", "county", date_range=(0, 1), fetcher=mock_covidcast_signal)
+    pd.testing.assert_frame_equal(
+        result,
+        pd.DataFrame(
+            {
+                "geo_id": ["01000"],
+                "val": [10],
+                "timestamp": [20200101],
+                "issue": [20200104]
+            },
+            index=[1]
+        )
+    )
+
 def test_compute_special_geo_dfs():
     test_df = pd.DataFrame({"geo_id": ["01000", "01001"],
                             "val": [50, 100],