fix the bug in load chng data, use prefix as issue date instead of dropdate

Author: Jingjing Tang

changehc/delphi_changehc/load_data.py

@@ -4,6 +4,7 @@
 Author: Aaron Rumack
 Created: 2020-10-14
 """
+from datetime import datetime
 # third party
 import pandas as pd
 
@@ -58,7 +59,7 @@ def load_chng_data(filepath, dropdate, base_geo,
     # restrict to start and end date
     data = data[
         (data[Config.DATE_COL] >= Config.FIRST_DATA_DATE) &
-        (data[Config.DATE_COL] < dropdate)
+        (data[Config.DATE_COL] <= dropdate)
         ]
 
     # counts between 1 and 3 are coded as "3 or less", we convert to 1
@@ -76,25 +77,27 @@ def load_chng_data(filepath, dropdate, base_geo,
     return data
 
 
-def load_combined_data(denom_filepath, covid_filepath, dropdate, base_geo,
+def load_combined_data(denom_filepath, covid_filepath, base_geo,
                        backfill_dir, geo, weekday, numtype, backfill_merge_day):
     """Load in denominator and covid data, and combine them.
 
     Args:
         denom_filepath: path to the aggregated denominator data
         covid_filepath: path to the aggregated covid data
-        dropdate: data drop date (datetime object)
         base_geo: base geographic unit before aggregation ('fips')
 
     Returns:
         combined multiindexed dataframe, index 0 is geo_base, index 1 is date
     """
     assert base_geo == "fips", "base unit must be 'fips'"
 
+    # Get issue_date from the filename
+    issue_date = datetime.strptime(covid_filepath.split("/")[-1][:8], "%Y%m%d")
+
     # load each data stream
-    denom_data = load_chng_data(denom_filepath, dropdate, base_geo,
+    denom_data = load_chng_data(denom_filepath, issue_date, base_geo,
                      Config.DENOM_COLS, Config.DENOM_DTYPES, Config.DENOM_COL)
-    covid_data = load_chng_data(covid_filepath, dropdate, base_geo,
+    covid_data = load_chng_data(covid_filepath, issue_date, base_geo,
                      Config.COVID_COLS, Config.COVID_DTYPES, Config.COVID_COL)
 
     # merge data
@@ -109,13 +112,13 @@ def load_combined_data(denom_filepath, covid_filepath, dropdate, base_geo,
 
     # Store for backfill
     merge_backfill_file(backfill_dir, numtype, geo, weekday, backfill_merge_day,
-                        dropdate, test_mode=False, check_nd=25)
-    store_backfill_file(data, dropdate, backfill_dir, numtype, geo, weekday)
+                        issue_date, test_mode=False, check_nd=25)
+    store_backfill_file(data, issue_date, backfill_dir, numtype, geo, weekday)
     return data
 
 
 def load_cli_data(denom_filepath, flu_filepath, mixed_filepath, flu_like_filepath,
-                  covid_like_filepath, dropdate, base_geo,
+                  covid_like_filepath, base_geo,
                   backfill_dir, geo, weekday, numtype, backfill_merge_day):
     """Load in denominator and covid-like data, and combine them.
 
@@ -125,24 +128,26 @@ def load_cli_data(denom_filepath, flu_filepath, mixed_filepath, flu_like_filepat
         mixed_filepath: path to the aggregated mixed data
         flu_like_filepath: path to the aggregated flu-like data
         covid_like_filepath: path to the aggregated covid-like data
-        dropdate: data drop date (datetime object)
         base_geo: base geographic unit before aggregation ('fips')
 
     Returns:
         combined multiindexed dataframe, index 0 is geo_base, index 1 is date
     """
     assert base_geo == "fips", "base unit must be 'fips'"
 
+    # Get issue_date from the filename
+    issue_date = datetime.strptime(flu_filepath.split("/")[-1][:8], "%Y%m%d")
+
     # load each data stream
-    denom_data = load_chng_data(denom_filepath, dropdate, base_geo,
+    denom_data = load_chng_data(denom_filepath, issue_date, base_geo,
                      Config.DENOM_COLS, Config.DENOM_DTYPES, Config.DENOM_COL)
-    flu_data = load_chng_data(flu_filepath, dropdate, base_geo,
+    flu_data = load_chng_data(flu_filepath, issue_date, base_geo,
                      Config.FLU_COLS, Config.FLU_DTYPES, Config.FLU_COL)
-    mixed_data = load_chng_data(mixed_filepath, dropdate, base_geo,
+    mixed_data = load_chng_data(mixed_filepath, issue_date, base_geo,
                      Config.MIXED_COLS, Config.MIXED_DTYPES, Config.MIXED_COL)
-    flu_like_data = load_chng_data(flu_like_filepath, dropdate, base_geo,
+    flu_like_data = load_chng_data(flu_like_filepath, issue_date, base_geo,
                      Config.FLU_LIKE_COLS, Config.FLU_LIKE_DTYPES, Config.FLU_LIKE_COL)
-    covid_like_data = load_chng_data(covid_like_filepath, dropdate, base_geo,
+    covid_like_data = load_chng_data(covid_like_filepath, issue_date, base_geo,
                      Config.COVID_LIKE_COLS, Config.COVID_LIKE_DTYPES, Config.COVID_LIKE_COL)
 
     # merge data
@@ -162,30 +167,32 @@ def load_cli_data(denom_filepath, flu_filepath, mixed_filepath, flu_like_filepat
 
     # Store for backfill
     merge_backfill_file(backfill_dir, numtype, geo, weekday, backfill_merge_day,
-                        dropdate, test_mode=False, check_nd=25)
-    store_backfill_file(data, dropdate, backfill_dir, numtype, geo, weekday)
+                        issue_date, test_mode=False, check_nd=25)
+    store_backfill_file(data, issue_date, backfill_dir, numtype, geo, weekday)
     return data
 
 
-def load_flu_data(denom_filepath, flu_filepath, dropdate, base_geo,
+def load_flu_data(denom_filepath, flu_filepath, base_geo,
                   backfill_dir, geo, weekday, numtype, backfill_merge_day):
     """Load in denominator and flu data, and combine them.
 
     Args:
         denom_filepath: path to the aggregated denominator data
         flu_filepath: path to the aggregated flu data
-        dropdate: data drop date (datetime object)
         base_geo: base geographic unit before aggregation ('fips')
 
     Returns:
         combined multiindexed dataframe, index 0 is geo_base, index 1 is date
     """
     assert base_geo == "fips", "base unit must be 'fips'"
 
+    # Get issue_date from the filename
+    issue_date = datetime.strptime(flu_filepath.split("/")[-1][:8], "%Y%m%d")
+
     # load each data stream
-    denom_data = load_chng_data(denom_filepath, dropdate, base_geo,
+    denom_data = load_chng_data(denom_filepath, issue_date, base_geo,
                      Config.DENOM_COLS, Config.DENOM_DTYPES, Config.DENOM_COL)
-    flu_data = load_chng_data(flu_filepath, dropdate, base_geo,
+    flu_data = load_chng_data(flu_filepath, issue_date, base_geo,
                      Config.FLU_COLS, Config.FLU_DTYPES, Config.FLU_COL)
 
     # merge data
@@ -200,6 +207,6 @@ def load_flu_data(denom_filepath, flu_filepath, dropdate, base_geo,
 
     # Store for backfill
     merge_backfill_file(backfill_dir, numtype, geo, weekday, backfill_merge_day,
-                        dropdate, test_mode=False, check_nd=25)
-    store_backfill_file(data, dropdate, backfill_dir, numtype, geo, weekday)
+                        issue_date, test_mode=False, check_nd=25)
+    store_backfill_file(data, issue_date, backfill_dir, numtype, geo, weekday)
     return data

changehc/delphi_changehc/run.py

@@ -183,16 +183,16 @@ def run_module(params: Dict[str, Dict[str, Any]]):
                 )
                 if numtype == "covid":
                     data = load_combined_data(file_dict["denom"],
-                             file_dict["covid"],dropdate_dt,"fips",
+                             file_dict["covid"], "fips",
                              backfill_dir, geo, weekday, numtype,
                              backfill_merge_day)
                 elif numtype == "cli":
                     data = load_cli_data(file_dict["denom"],file_dict["flu"],file_dict["mixed"],
-                             file_dict["flu_like"],file_dict["covid_like"],dropdate_dt,"fips",
+                             file_dict["flu_like"],file_dict["covid_like"], "fips",
                              backfill_dir, geo, weekday, numtype, backfill_merge_day)
                 elif numtype == "flu":
                     data = load_flu_data(file_dict["denom"],file_dict["flu"],
-                             dropdate_dt,"fips",backfill_dir, geo, weekday,
+                             "fips",backfill_dir, geo, weekday,
                              numtype, backfill_merge_day)
                 more_stats = su_inst.update_sensor(
                     data,