Merge pull request #1711 from cmu-delphi/ndefries/training-date-from-model

Fetch modeling date from model filenames

Author: krivard

backfill_corrections/delphiBackfillCorrection/R/beta_prior_estimation.R

@@ -48,7 +48,8 @@ objective <- function(theta, x, prob, ...) {
 #' @param start the initialization of the the points in nlm
 #' @param base_pseudo_denom the pseudo counts added to denominator if little data for training
 #' @param base_pseudo_num the pseudo counts added to numerator if little data for training
-#' @param training_end_date the most recent training date
+#' @template training_end_date-template
+#' @template training_start_date-template
 #' @param model_save_dir directory containing trained models
 #' 
 #' @importFrom stats nlm predict
@@ -58,7 +59,8 @@ objective <- function(theta, x, prob, ...) {
 est_priors <- function(train_data, prior_test_data, geo, value_type, dw, taus, 
                        covariates, response, lp_solver, lambda, 
                        indicator, signal, geo_level, signal_suffix, 
-                       training_end_date, model_save_dir, start=c(0, log(10)),
+                       training_end_date, training_start_date,
+                       model_save_dir, start=c(0, log(10)),
                        base_pseudo_denom=1000, base_pseudo_num=10,
                        train_models = TRUE, make_predictions = TRUE) {
   sub_train_data <- train_data %>% filter(train_data[[dw]] == 1)
@@ -76,6 +78,7 @@ est_priors <- function(train_data, prior_test_data, geo, value_type, dw, taus,
                                            geo=geo, dw=dw, tau=tau,
                                            value_type=value_type,
                                            training_end_date=training_end_date,
+                                           training_start_date=training_start_date,
                                            beta_prior_mode=TRUE)
       model_path <- file.path(model_save_dir, model_file_name)
 
@@ -123,7 +126,8 @@ frac_adj_with_pseudo <- function(data, dw, pseudo_num, pseudo_denom, num_col, de
 #' @template train_data-template
 #' @param test_data testing data
 #' @param prior_test_data testing data for the lag -1 model
-#' @param training_end_date the most recent training date
+#' @template training_end_date-template
+#' @template training_start_date-template
 #' @param model_save_dir directory containing trained models
 #' @template indicator-template
 #' @template signal-template
@@ -141,7 +145,8 @@ frac_adj_with_pseudo <- function(data, dw, pseudo_num, pseudo_denom, num_col, de
 frac_adj <- function(train_data, test_data, prior_test_data, 
                      indicator, signal, geo_level, signal_suffix,
                      lambda, value_type, geo, 
-                     training_end_date, model_save_dir, 
+                     training_end_date, training_start_date,
+                     model_save_dir,
                      taus, lp_solver,
                      train_models = TRUE,
                      make_predictions = TRUE) {
@@ -177,7 +182,7 @@ frac_adj <- function(train_data, test_data, prior_test_data,
     pseudo_counts <- est_priors(train_data, prior_test_data, geo, value_type, cov, taus, 
                                 pre_covariates, "log_value_target", lp_solver, lambda, 
                                 indicator, signal, geo_level, signal_suffix, 
-                                training_end_date, model_save_dir,
+                                training_end_date, training_start_date, model_save_dir,
                                 train_models = train_models,
                                 make_predictions = make_predictions)
     pseudo_denum = pseudo_counts[1]

backfill_corrections/delphiBackfillCorrection/R/io.R

@@ -22,34 +22,46 @@ read_data <- function(input_dir) {
 #' @template lambda-template
 #' @template value_type-template
 #' @template export_dir-template
-#' @param training_end_date the most recent training date
+#' @template training_end_date-template
+#' @template training_start_date-template
 #'
 #' @importFrom readr write_csv
 #' @importFrom stringr str_interp str_split
 export_test_result <- function(test_data, coef_data, indicator, signal, 
-                               geo_level, geo, signal_suffix, lambda,
-                               training_end_date,
+                               geo_level, signal_suffix, lambda,
+                               training_end_date, training_start_date,
                                value_type, export_dir) {
   base_name <- generate_filename(indicator=indicator, signal=signal,
                                  geo_level=geo_level, signal_suffix=signal_suffix,
                                  lambda=lambda, training_end_date=training_end_date,
-                                 geo=geo, value_type=value_type, model_mode=FALSE)
+                                 training_start_date=training_start_date,
+                                 value_type=value_type, model_mode=FALSE)
 
-  signal_info <- str_interp("indicator ${indicator} signal ${signal} geo ${geo} value_type ${value_type}")
+  signal_info <- str_interp("indicator ${indicator} signal ${signal} geo_level ${geo_level} value_type ${value_type}")
+  
+  components <- c(indicator, signal, signal_suffix)
+  signal_dir <- paste(components[components != ""], collapse="_")
+  
+  dir.create(file.path(export_dir, signal_dir), showWarnings = FALSE)
+  
   if (nrow(test_data) == 0) {
     warning(str_interp("No test data available for ${signal_info}"))
   } else {
     msg_ts(str_interp("Saving predictions to disk for ${signal_info} "))
     pred_output_file <- str_interp("prediction_${base_name}")
-    write_csv(test_data, file.path(export_dir, pred_output_file))
+    
+    prediction_col <- colnames(test_data)[grepl("^predicted", colnames(test_data))]
+    expected_col <- c("time_value", "issue_date", "lag", "geo_value", 
+                      "target_date", "wis", prediction_col)
+    write_csv(test_data[expected_col], file.path(export_dir, signal_dir, pred_output_file))
   }
 
   if (nrow(coef_data) == 0) {
     warning(str_interp("No coef data available for ${signal_info}"))
   } else {
     msg_ts(str_interp("Saving coefficients to disk for ${signal_info}"))
-  coef_output_file <- str_interp("coefs_${base_name}")
-  write_csv(coef_data, file.path(export_dir, coef_output_file))
+    coef_output_file <- str_interp("coefs_${base_name}")
+    write_csv(coef_data, file.path(export_dir, signal_dir, coef_output_file))
   }
 }
 
@@ -99,13 +111,13 @@ subset_valid_files <- function(files_list, file_type = c("daily", "rollup"), par
   switch(file_type,
          daily = {
            start_dates <- as.Date(
-             sub("^.*/.*_as_of_([0-9]{8}).parquet$", "\\1", files_list),
+             sub("^.*/.*_as_of_([0-9]{8})[.]parquet$", "\\1", files_list),
              format = date_format
            )
            end_dates <- start_dates
          },
          rollup = {
-           rollup_pattern <- "^.*/.*_from_([0-9]{8})_to_([0-9]{8}).parquet$"
+           rollup_pattern <- "^.*/.*_from_([0-9]{8})_to_([0-9]{8})[.]parquet$"
            start_dates <- as.Date(
              sub(rollup_pattern, "\\1", files_list),
              format = date_format
@@ -117,12 +129,9 @@ subset_valid_files <- function(files_list, file_type = c("daily", "rollup"), par
          }
   )
 
-  # Start_date depends on if we're doing model training or just corrections.
-  n_addl_days <- params$ref_lag
-  if (params$train_models) {
-    n_addl_days <- n_addl_days + params$training_days
-  }
-
+  ## TODO: right now, this gets both training and testing data regardless of
+  #  which mode is selected
+  n_addl_days <- params$ref_lag + params$training_days
   start_date <- TODAY - n_addl_days
   end_date <- TODAY - 1
 
@@ -146,7 +155,7 @@ create_name_pattern <- function(indicator, signal,
                                 file_type = c("daily", "rollup")) {
   file_type <- match.arg(file_type)
   switch(file_type,
-         daily = str_interp("${indicator}_${signal}_as_of_[0-9]{8}.parquet$"),
-         rollup = str_interp("${indicator}_${signal}_from_[0-9]{8}_to_[0-9]{8}.parquet$")
+         daily = str_interp("${indicator}_${signal}_as_of_[0-9]{8}[.]parquet$"),
+         rollup = str_interp("${indicator}_${signal}_from_[0-9]{8}_to_[0-9]{8}[.]parquet$")
   )
 }

backfill_corrections/delphiBackfillCorrection/R/main.R

@@ -8,14 +8,13 @@
 #' @template signal_suffixes-template
 #' @template indicator-template
 #' @template signal-template
-#' @param training_end_date the most recent training date
 #' 
 #' @importFrom dplyr %>% filter select group_by summarize across everything group_split ungroup
 #' @importFrom tidyr drop_na
 #' @importFrom rlang .data .env
 #' 
 #' @export
-run_backfill <- function(df, params, training_end_date,
+run_backfill <- function(df, params,
                          refd_col = "time_value", lag_col = "lag", issued_col = "issue_date",
                          signal_suffixes = c(""), indicator = "", signal = "") {
   df <- filter(df, .data$lag < params$ref_lag + 30) # a rough filtration to save memory
@@ -57,13 +56,14 @@ run_backfill <- function(df, params, training_end_date,
         coef_list[[key]] <- list()
       }
     }
-
+    
     msg_ts("Splitting data into geo groups")
     group_dfs <- group_split(df, .data$geo_value)
 
     # Build model for each location
     for (subdf in group_dfs) {
       geo <- subdf$geo_value[1]
+      
       msg_ts(str_interp("Processing ${geo} geo group"))
 
       min_refd <- min(subdf[[refd_col]])
@@ -115,9 +115,9 @@ run_backfill <- function(df, params, training_end_date,
           combined_df <- combined_df %>% filter(.data$lag < params$ref_lag)
 
           geo_train_data <- combined_df %>%
-            filter(.data$issue_date < training_end_date) %>%
-            filter(.data$target_date <= training_end_date) %>%
-            filter(.data$target_date > training_end_date - params$training_days) %>%
+            filter(.data$issue_date < params$training_end_date) %>%
+            filter(.data$target_date <= params$training_end_date) %>%
+            filter(.data$target_date > params$training_start_date) %>%
             drop_na()
           geo_test_data <- combined_df %>%
             filter(.data$issue_date %in% params$test_dates) %>%
@@ -135,7 +135,8 @@ run_backfill <- function(df, params, training_end_date,
                                      indicator = indicator, signal = signal,
                                      geo_level = geo_level, signal_suffix = signal_suffix,
                                      lambda = params$lambda, value_type = value_type, geo = geo,
-                                     training_end_date = training_end_date,
+                                     training_end_date = params$training_end_date,
+                                     training_start_date = params$training_start_date,
                                      model_save_dir = params$cache_dir,
                                      taus = params$taus,
                                      lp_solver = params$lp_solver,
@@ -178,7 +179,9 @@ run_backfill <- function(df, params, training_end_date,
               lambda = params$lambda, test_lag = test_lag, geo = geo,
               value_type = value_type, model_save_dir = params$cache_dir,
               indicator = indicator, signal = signal, geo_level = geo_level,
-              signal_suffix =signal_suffix, training_end_date = training_end_date,
+              signal_suffix =signal_suffix,
+              training_end_date = params$training_end_date,
+              training_start_date = params$training_start_date,
               train_models = params$train_models,
               make_predictions = params$make_predictions
             )
@@ -199,23 +202,24 @@ run_backfill <- function(df, params, training_end_date,
           }# End for test lags
         }# End for value types
       }# End for signal suffixes
-      
-      if (params$make_predictions) {
-        for (value_type in params$value_types) {
-          for (signal_suffix in signal_suffixes) {
-            key <- make_key(value_type, signal_suffix)
-            test_combined <- bind_rows(test_data_list[[key]]) 
-            coef_combined <- bind_rows(coef_list[[key]]) 
-            export_test_result(test_combined, coef_combined, 
-                               indicator, signal, 
-                               geo_level, geo, signal_suffix, params$lambda,
-                               training_end_date,
-                               value_type, export_dir=params$export_dir)
-          }
+    }# End for geo list
+    
+    if (params$make_predictions) {
+      for (value_type in params$value_types) {
+        for (signal_suffix in signal_suffixes) {
+          key <- make_key(value_type, signal_suffix)
+          test_combined <- bind_rows(test_data_list[[key]]) 
+          coef_combined <- bind_rows(coef_list[[key]]) 
+          export_test_result(test_combined, coef_combined, 
+                             indicator=indicator, signal=signal,
+                             signal_suffix=signal_suffix,
+                             geo_level=geo_level, lambda=params$lambda,
+                             training_end_date=params$training_end_date,
+                             training_start_date=params$training_start_date,
+                             value_type=value_type, export_dir=params$export_dir)
         }
       }
-      
-    }# End for geo list
+    }
   }# End for geo type
 }
 
@@ -236,14 +240,10 @@ main <- function(params) {
 
   if (params$train_models) {
     msg_ts("Removing stored models")
-    files_list <- list.files(params$cache_dir, pattern="*.model", full.names = TRUE)
+    files_list <- list.files(params$cache_dir, pattern="[.]model$", full.names = TRUE)
     file.remove(files_list)
   }
 
-  training_end_date <- as.Date(readLines(
-    file.path(params$cache_dir, "training_end_date.txt")))
-  msg_ts(str_interp("training_end_date is ${training_end_date}"))
-
   ## Set default number of cores for mclapply to half of those available.
   if (params$parallel) {
     cores <- detectCores()
@@ -255,7 +255,18 @@ main <- function(params) {
       options(mc.cores = min(params$parallel_max_cores, max(floor(cores / 2), 1L)))
     }
   }
-  
+
+  # Training start and end dates are the same for all indicators, so we can fetch
+  # at the beginning.
+  result <- get_training_date_range(params)
+  params$training_start_date <- result$training_start_date
+  params$training_end_date <- result$training_end_date
+
+  msg_ts(paste0(
+    str_interp("training_start_date is ${params$training_start_date}, "),
+    str_interp("training_end_date is ${params$training_end_date}")
+  ))
+
   # Loop over every indicator + signal combination.
   for (group_i in seq_len(nrow(INDICATORS_AND_SIGNALS))) {
     input_group <- INDICATORS_AND_SIGNALS[group_i,]
@@ -302,14 +313,8 @@ main <- function(params) {
     training_days_check(input_data$issue_date, params$training_days)
 
     # Perform backfill corrections and save result
-    run_backfill(input_data, params, training_end_date,
+    run_backfill(input_data, params,
       indicator = input_group$indicator, signal = input_group$signal,
       signal_suffixes = input_group$name_suffix)
-
-    if (params$train_models) {
-      # Save the training end date to a text file.
-      writeLines(as.character(TODAY),
-                 file.path(params$cache_dir, "training_end_date.txt"))
-    }
   }
 }