move lag and issue_date setting to daily files

Author: nmdefries

claims_hosp/delphi_claims_hosp/backfill.py

@@ -50,6 +50,18 @@ def store_backfill_file(claims_filepath, _end_date, backfill_dir):
     backfilldata = backfilldata.loc[(backfilldata["time_value"] >= _start_date)
                                     & (~backfilldata["fips"].isnull()),
                                     selected_columns]
+
+    backfilldata["lag"] = [(_end_date - x).days for x in backfilldata["time_value"]]
+    backfilldata["time_value"] = backfilldata.time_value.dt.strftime("%Y-%m-%d")
+    backfilldata["issue_date"] = datetime.strftime(_end_date, "%Y-%m-%d")
+
+    backfilldata = backfilldata.astype({
+        "time_value": "string",
+        "issue_date": "string",
+        "fips": "string",
+        "state_id": "string"
+    })
+
     path = backfill_dir + \
         "/claims_hosp_as_of_%s.parquet"%datetime.strftime(_end_date, "%Y%m%d")
     # Store intermediate file into the backfill folder
@@ -101,9 +113,6 @@ def get_date(file_link):
     pdList = []
     for fn in new_files:
         df = pd.read_parquet(fn, engine='pyarrow')
-        issue_date = get_date(fn)
-        df["issue_date"] = issue_date
-        df["lag"] = [(issue_date - x).days for x in df["time_value"]]
         pdList.append(df)
     merged_file = pd.concat(pdList).sort_values(["time_value", "fips"])
     path = backfill_dir + "/claims_hosp_from_%s_to_%s.parquet"%(

claims_hosp/tests/test_backfill.py

@@ -37,7 +37,7 @@ def test_store_backfill_file(self):
         backfill_df = pd.read_parquet(backfill_dir + "/"+ fn, engine='pyarrow')
 
         selected_columns = ['time_value', 'fips', 'state_id',
-                        'num', 'den']
+                        'num', 'den', 'lag', 'issue_date']
         assert set(selected_columns) == set(backfill_df.columns)  
 
         os.remove(backfill_dir + "/" + fn)
@@ -79,9 +79,6 @@ def test_merge_backfill_file(self):
             if "from" in file:
                 continue
             df = pd.read_parquet(file, engine='pyarrow')
-            issue_date = datetime.strptime(file[-16:-8], "%Y%m%d")
-            df["issue_date"] = issue_date
-            df["lag"] = [(issue_date - x).days for x in df["time_value"]]
             pdList.append(df)
             os.remove(file)
         new_files = glob.glob(backfill_dir + "/claims_hosp*.parquet")