#================ Store_Avgs.r .us() .us() lib(forecast, quietly=TRUE) # lib(mi) setScience(proj="chartio", create=TRUE, subl=FALSE, load=FALSE) verbose.dbcon.off() ## CLUSTERS cluster.in <- getOption("db.defaultcluster.in") cluster.out <- getOption("db.defaultcluster.out") ## If FALSE, this script will skip over the whole section where plots of missing data are created createPlots <- FALSE refresh_DTs <- TRUE ## For DT.stores schema.out <- "bi" tbl.out <- "gpu" verbose <- assignIfNotExist("verbose", TRUE) ## How far back to look at data, Number of months. monthsBack <- 24 ## Set the DB to where we will query from message(sprintf("Will run the GPU queries against cluster #%02i", cluster.in)) setDBall(cluster=cluster.in) ## Load files from other projects DT.iTunes544 <- itunes_544(out="DT") DT.stores <- get_dim_store (refresh=refresh_DTs) DT.transacs <- get_dim_transacs (refresh=refresh_DTs) kCols <- c("store", "transac_typeid") kColsDate <- c("date", kCols) dimCols <- c(store="storeid", periodid="accountingperiodid", transac_typeid="transactiontypeid") ## Aggregate ACCOUNTING by dimCols { BackUpOrRestore("DT.storeavg", clear=TRUE, verbose=FALSE) Q.storeavg <- makeQry(tbl="fact_sales", schema="production" , colsToAgg=c(paidunits="sales", "gross") , colsToPull=dimCols , dateCol = "accountingperiodid" , minDate = sprintf("SELECT max(periodid) FROM bi.period_view WHERE perioddate <= sysdate - INTERVAL '%i MONTH'", monthsBack) ) DT.storeavg <- runQry(Q.storeavg, cluster=cluster.in) addDateCols.periodid_(DT.storeavg, drop=TRUE) # ## no need to set key, we will change it immediately # setkeyIfNot(DT.storeavg, kColsDate, organize=TRUE, verbose=FALSE) BackUpOrRestore("DT.storeavg", verbose=FALSE, verboseRestore=TRUE) } ## --------------------------------------------------------------- ## ## Calculate GPU ## That is, avg sale per unit, by store, period, transactype ## --------------------------------------------------------------- ## DT.storeavg[, GPU := gross / paidunits] ## Any non-finite GPU should be set to NA (division by 0) DT.storeavg[!is.finite(GPU), GPU := NA] ## --------------------------------------------------------------- ## #================ Store_Avgs- GPU Calc.r ================ # ## Set key with date column LAST, for forecasting purposes setkeyIfNot(DT.storeavg, c(kCols, "date"), organize=TRUE) ## dont use kColsDate ## FIND ANY $0 gross stores or NA gross [there should not be any NAs at all] tmp.zerogross <- DT.storeavg[gross==0 | is.na(gross)] if (nrow(tmp.zerogross)) { msg <- sprintf("There %s %i groups with $0 gross [These rows will be removed]. Shown below :\n %s\n", plrl("are", nrow(tmp.zerogross)), nrow(tmp.zerogross), pasteC(capture.output(tmp.zerogross), C="\n ")) ## If there are very few gross==0 (generally, just one store), drop those. Otherwise, fail if (nrow(tmp.zerogross) / nrow(DT.storeavg) < .001) { DT.storeavg <- assignWithInfo("DT.storeavg", DT.storeavg[gross!=0], info=sprintf("Modified self; Removed %i rows, keeping only rows where !(gross == 0 | is.na(gross))", nrow(DT.storeavg) - nrow(DT.storeavg[gross!=0]))) # this will NOT preserve NAs in gross } else stop (msg) } ## CREATE TWO FORECASTS COLUMNS -- by kCols (ie, currently store + transac) ## -------------------------------------------- ## Get all unique dates, then we will iterate over them, in reverse order ## GPU_Est_Fore :: Forecasting over the lastN values (using forecast() function) ## GPU_Est_MA :: 3 Month moving average lastN <- 13 AllDates <- DT.storeavg[, seq(min(date), max(date), by="month")] for (D in rev(AllDates) ) { verboseMsg(verbose, "Calculating GPU_Est_Fore & GPU_Est_MA for date == '", as.character(as.Date(D)), "'\n", simple=TRUE) DT.storeavg[date <= D & !is.na(GPU) ## GPU must not be NA since this errors forecast() ## &&& 20141019 ## TODO: something like {ff <- forecast(tail(GPU, lastN), 1); list(list(ff$model), ff$mean) , `:=`( GPU_Est_Fore = if (length(GPU) < 3) NA_real_ else forecast(tail(GPU, lastN), 1)[["mean"]], # , `:=`( GPU_Est_Fore = if (length(GPU) < 3) NA_real_ else {browser(); forecast(tail(GPU, lastN), 1)[["mean"]]}, # GPU_Est_MA = {z <- mean(tail(GPU, 3), na.rm=TRUE); if(is.na(z)) NA_real_ else z} GPU_Est_MA = mean(tail(GPU, 3), na.rm=TRUE) ) , by=kCols ] } ## SEE DT.storeavg[store == 1 & transac_typeid %in% 6:12][order(transac_typeid)] DT.storeavg[store == 286 & transac_typeid %in% 1:12][order(transac_typeid)] DT.storeavg[store == 2 ][order(transac_typeid)] ## PLOTTING, optional. (Shows what data is available) if (createPlots && .Pfm == "Darwin") source("~rsaporta/git/orch/src/chartio/Plot GPU Table.r") browser(expr=inDebugMode(c("GPU", "Automation")), text="Not in a function, but in the file '~/git/orch/src/chartio/Create GPU Table from analytics.r' being sourced") ## Add a month to the tail of DT.storeavg, for each group in kcols. ## ----------------------------------------------------------------- ## do this by taking a two-month seq and banking just the last value ## then rbind'ing this to the original DT.storeavg ## We save it to a new table, to preserve the original ## TODO 20141014: Consider calculating DT.nextmonth JUST BY store. ## Pros: Uniformity across store. ## Cons: Smaller stores (eg storeid==2) will have sevearl transac types that will be blank for sevearl months ## TODO 20140920: (We do not want to add dates for groups that have not occurred in over two months) ## However, this does NOT work, because it introduces bugs downstream DT.nextmonth <- setkey(DT.storeavg[date >= max(date) - 63, list(date=seq(max(date), length.out=2, by="month")[[-1L]]), keyby=kCols]) ## Consider adding a filter column -- but how to get it to DT.storeavg2 ? ## Also, what is the implication of leaving it as is? DT.nextmonth <- setkey(DT.storeavg[, list(date=seq(max(date), length.out=2, by="month")[[-1L]]), keyby=kCols]) DT.storeavg2 <- setkeyv(rbind(DT.storeavg, DT.nextmonth, fill=TRUE), key(DT.storeavg)) setInfo(DT.storeavg2, appendInfo("rbind of DT.storeavg with rows added for 'next month'", DT=DT.storeavg)) ## Confirm that the last row in each store-trans is NA stopifnot(DT.storeavg2[, is.na(tail(GPU, 1)), by=kCols][, V1]) ## Shift the ESTs down one row (ie, they are for the future, not for the present) DT.storeavg2[, GPU_Est_MA := shiftDown(GPU_Est_MA), by=kCols] DT.storeavg2[, GPU_Est_Fore := shiftDown(GPU_Est_Fore), by=kCols] ## 2014-01-27 ## Something changed in either the forecast package or the data.table package ## where GPU_Est_Fore is throwing a warning. The warning comes just when displaying the vector, ## but then when calling a simple operation on it (such as addition or division) ## an error gets thrown: ## EXAMPLE: ## R > DT.storeavg2[, GPU_Est_Fore ] ## # Warning: series is corrupt: length 3133 with 'tsp' implying 1 ## R > DT.storeavg2[, GPU - GPU_Est_Fore] ## # Error in NextMethod(.Generic) : invalid time series parameters specified ## SIMPLE SOLUTION ## Converting to numeric. ## This looses the time.series attributes, but then again ## (1) not sure if they are needed downstream ## (2) the attribs still exist in DT.storeavg ## Convert the timeseries to numeric (Added 2015-01-27) DT.storeavg2[, GPU_Est_Fore := as.numeric(GPU_Est_Fore)] ## Calculate the diff between estimates and stat DT.storeavg2[, `:=`( err.fore = (GPU - GPU_Est_Fore) / GPU , err.ma = (GPU - GPU_Est_MA) / GPU )] ## Determine which one is more accurate ## If both are equal or both NA, then neither is "better" ## Otherwise whichever is SMALLER is better. (ie, MA is better when err.fore > err.ma) DT.storeavg2[!equals(abs(err.ma), abs(err.fore), tol=1e-4, na.check=TRUE) , MA_better_than_Fore := removeNA(abs(err.fore), Inf) > removeNA(abs(err.ma), Inf)] ## < SEE > DT.storeavg2[, table(MA_better_than_Fore, useNA="always")] ## By definition, anytime GPU is NA, then MA_better_than_Fore will be NA as well ## If this CONFIRM fails, then we need to modify the next line (where 'use_MA' is defined) to account for NAs in GPU stopifnot(DT.storeavg2[, !(is.na(GPU) & !is.na(MA_better_than_Fore) )]) ## Which inds to use for error checking DT.storeavg2[, indsForErrChecking := date > Sys.Date() - 150] ## See DT.storeavg2[(indsForErrChecking)] DT.storeavg2[ ## Looking back over the lst four months, is the ratio of MA_better_than_Fore greater than .5 DT.storeavg2[(indsForErrChecking) , list(use_MA = sumn(MA_better_than_Fore) / max(1, sum(!is.na(MA_better_than_Fore))) >= .5) , keyby=kCols ] , use_MA := i.use_MA , allow.cartesian=TRUE ] ## CLEAN UP ANY NA's from paidunits DT.storeavg2[!is.finite(GPU_Est_MA), GPU_Est_MA := NA] DT.storeavg2[!is.finite(GPU_Est_Fore), GPU_Est_Fore := NA] ## Weighted betweeen the two ## -------------------------------------------------------------- ## ## We need to account for NAs in the two Estimates. However, note that Est_MA is NEVER NA (except for the first month, by kcols) ## Thus, we only need to account for those rows where Est_Fore is NA ## CONFIRM: No NAs in Est_MA for date > minDate_by_kCol stopifnot(DT.storeavg2[DT.storeavg2[, list(minDate_by_kCol = min(date)), keyby=kCols]][date > minDate_by_kCol][, !is.na(GPU_Est_MA)]) ## CONFIRM: If GPU_Est_MA is NA, then so is GPU_Est_Fore stopifnot(DT.storeavg2[is.na(GPU_Est_MA), is.na(GPU_Est_Fore)]) ## ## When GPU_Est_Fore is NA, then use 100% Est_MA DT.storeavg2[is.na(GPU_Est_Fore), GPU_estimate2 := GPU_Est_MA] ## When GPU_Est_Fore is not NA, then use 80x20% based on which is more accurae DT.storeavg2[!is.na(GPU_Est_Fore) & use_MA, GPU_estimate2 := GPU_Est_MA * 0.80 + GPU_Est_Fore * 0.20] # <~~ Note the reverse parameters bsed on (use_MA) DT.storeavg2[!is.na(GPU_Est_Fore) & !use_MA, GPU_estimate2 := GPU_Est_MA * 0.20 + GPU_Est_Fore * 0.80] # <~~ Note the reverse parameters bsed on (use_MA) ## Determine which GPU estimate to use DT.storeavg2[, GPU_estimate := ifelse(use_MA, GPU_Est_MA, GPU_Est_Fore)] DT.storeavg2[is.na(GPU_estimate), GPU_estimate := ifelse(!is.na(GPU_Est_MA), GPU_Est_MA, GPU_Est_Fore)] ## Use a blend of Forecast and MA DT.storeavg2[!is.na(GPU_Est_MA) & !is.na(GPU_Est_MA), GPU_blended := (2*GPU_Est_MA + 3*GPU_Est_Fore)/5 ] # DT.storeavg2[!is.na(GPU_blended), GPU_estimate := GPU_blended] ## TESTING DIFFERENT METHODS estimates <- c("GPU_estimate", "GPU_estimate2", "GPU_blended") ### DIFFERENT WAYS OF COUNTING #### ------------------------------------------- { SimpleErrors <- DT.storeavg2[(indsForErrChecking), c(setNames(nm=estimates, lapply(estimates, function(x) abs(get(x)-GPU))), GPU=GPU), keyby=kColsDate] SimpleSumOfErrors <- SimpleErrors [, lapply(.SD, sumn), keyby=kCols, .SDcols=estimates] ## No Date SimpleSumOfErrors[, estimates[apply(.SD, 1, which.min)], .SDcols=estimates, keyby=store][, {counts <- table(V1); nwhich(counts==max(counts))}] SimpleSumOfErrors[, estimates[apply(.SD, 1, which.min)], .SDcols=estimates, keyby=transac_typeid][, {counts <- table(V1); nwhich(counts==max(counts))}] SimpleSumOfErrors <- SimpleErrors [, lapply(.SD, sumn), keyby=transac_typeid, .SDcols=estimates] ## No Date SimpleSumOfErrors[, estimates[apply(.SD, 1, which.min)], .SDcols=estimates, keyby=transac_typeid][, {counts <- table(V1); nwhich(counts==max(counts))}] SimpleSumOfErrors <- SimpleErrors [, lapply(.SD, sumn), keyby=store, .SDcols=estimates] ## No Date SimpleSumOfErrors[, estimates[apply(.SD, 1, which.min)], .SDcols=estimates, keyby=store][, {counts <- table(V1); nwhich(counts==max(counts))}] } # -------------------------------------------------------------------- # GPU ESTIMATE -- # GPU ESTIMATE -- ## CREATE the GPU.EST data.table. This is what will ultimately make it into the DB # GPU ESTIMATE -- setkeyIfNot(DT.storeavg2, key(DT.nextmonth)) # GPU ESTIMATE -- DT.GPU.Est <- DT.storeavg2[DT.nextmonth][, list(date, store, transac_typeid, label_sc_group, GPU_estimate)] # GPU ESTIMATE -- # GPU ESTIMATE -- ## Some values are still NA. Take the last known value # GPU ESTIMATE -- ## ... find the NAs, order by date, take the last value. # GPU ESTIMATE -- DT.GPU.Fillers <- DT.storeavg2[ unique(DT.GPU.Est[is.na(GPU_estimate), list(store, transac_typeid)], by=NULL) # GPU ESTIMATE -- ][!is.na(GPU) # GPU ESTIMATE -- ][order(date) # GPU ESTIMATE -- , list(GPU_estimate_filler=tail(GPU, 1)) # GPU ESTIMATE -- , keyby=kCols] # GPU ESTIMATE -- # GPU ESTIMATE -- ## Match keys for filling # GPU ESTIMATE -- matchKey(DT.GPU.Est, DT.GPU.Fillers, key(DT.GPU.Fillers)) # GPU ESTIMATE -- # GPU ESTIMATE -- ## FILL by adding a column, then any NAs get filled, confirm all is good, then drop the column # GPU ESTIMATE -- DT.GPU.Est[DT.GPU.Fillers, GPU_estimate_filler := GPU_estimate_filler] # GPU ESTIMATE -- # GPU ESTIMATE -- DT.GPU.Est[is.na(GPU_estimate), GPU_estimate := GPU_estimate_filler] # GPU ESTIMATE -- ## Confirm there are no NAs for the estimates # GPU ESTIMATE -- stopifnot(DT.GPU.Est[, !is.na(GPU_estimate)]) # GPU ESTIMATE -- ## Drop the column added # GPU ESTIMATE -- DT.GPU.Est[, GPU_estimate_filler := NULL] # -------------------------------------------------------------------- #================ Store_Avgs - 2 Analytics.r lastDateInMonth <- function(x) { lubridate::ceiling_date(x+1, unit="month") - 1 } firstDateInMonth <- function(x) { lubridate::floor_date(x, unit="month") } iTunesDaysInMonth <- function(x) { ## Every 3rd month is 35 days, the rest are 28 days ifelse(data.table::month(x) %% 3, 35, 28) } ## FORM QUERIES Qry.MonthlyAnalytics <- sprintf(" SELECT storeid AS store, transactiontypeid AS transac_typeid, %s, max(download_activity_date) AS max_date, sum(paidunits) AS paidunits FROM fact_analytics WHERE (download_activity_date BETWEEN sysdate - INTERVAL '%i MONTH' AND sysdate) %s -- ## I was previously removing NULL labelids, due to an issue matching to SC. But there should not be any NULLs in labelid -- AND NOT labelid is NULL GROUP BY 1, 2, 3 ORDER BY store" , c(iTunes = itunes_544(out="SQL", assign.to.envir=FALSE, dateCol="download_activity_date", dateColAS="date"), OtherStores = "to_char(download_activity_date, 'YYYY-MM-01')") , monthsBack , c(iTunes = "AND storeid = 1", OtherStores = "AND NOT (storeid = 1)") ) ## Add names to the QRY for lapply message("Pulling the raw data - Two queries. One for iTunes 544 and one for all the other stores") setattr(Qry.MonthlyAnalytics, "names", c("iTunes", "OtherStores")) setQry(Qry.MonthlyAnalytics) ## EXECUTE QUERIES # setDBall(cluster=cluster.in) ## until I fix runQry, call this again, in case connection was closed off listOfDTs <- lapply(Qry.MonthlyAnalytics, runQry, cluster=cluster.in) ## Collapse into a single DT DT.MonthlyAnalytics <- rbindlist(listOfDTs) setInfo(DT.MonthlyAnalytics, paste0("rbindlist of RAW PULL using Qry.MonthlyAnalytics.\nPulls sum(paidunits) from fact_analytics with \ndownload_activity_date adjusted to monthFloor or 544 for iTunes\nCluster pulled from = ", cluster.in, "")) ## ----- CLEAN THE DATA ----- ## BackUpOrRestore("DT.MonthlyAnalytics", clear=TRUE, verbose=FALSE) ## CLEAN 'Downloaded Ringtone' and 'Pre Cut Ringtone' ## ## According to James K, "Downloaded Ringtone" & "Pre Cut Ringtone" are treated ## the same by Accounting reports. They are only distinguished in analytics. ## ## transac_typeid transac_type_abbr transac_type ## 4 DR Downloaded Ringtone ## 26 PR Pre Cut Ringtone ## DT.MonthlyAnalytics[transac_typeid == 26, transac_typeid := 4] ## Storeid ==355, Myxer, has messed up transactype. Change 12 to 4 DT.MonthlyAnalytics[store == 355 & transac_typeid == 12, transac_typeid := 4] DT.MonthlyAnalytics <- changeAndAggregate(DT.MonthlyAnalytics, changeFunc=identity, colsToAgg="paidunits") ## Drop any missing label_sc_group if ("label_sc_group" %in% names(DT.MonthlyAnalytics)) DT.MonthlyAnalytics <- assignWithInfo("DT.MonthlyAnalytics", DT.MonthlyAnalytics[!is.na(label_sc_group)], info=sprintf("Modified self; Removed %i rows, keeping only rows where !(is.na(label_sc_group))", nrow(DT.MonthlyAnalytics) - nrow(DT.MonthlyAnalytics[!is.na(label_sc_group)]))) # this will NOT preserve NAs in gross # DT.MonthlyAnalytics <- DT.MonthlyAnalytics[!is.na(label_sc_group)] ## Convert Dates from Strings dateCols <- c("date", "max_date") DT.MonthlyAnalytics[, (dateCols) := lapply(.SD, as.Date), .SDcols=dateCols] ## Apply the same max date across the whole store, regardless of group. Bank the old max_date, for inspection at some other time setnames(DT.MonthlyAnalytics, "max_date", "max_date_bygrp") DT.MonthlyAnalytics[, max_date := max(max_date_bygrp, na.rm=TRUE), by="store,date"] ## Set keys, prepare for merge setkeyIfNot(DT.MonthlyAnalytics, kCols, verbose=FALSE) setcolorderpt(DT.MonthlyAnalytics, c(kCols, "date", "max_date")) ## ----- CLEAN THE DATA ----- ## ## -------------------------------------------- ## ## We will first add in the itunes start/end months ## then we will put in the regular start/end months setkeyIfNot(DT.iTunes544, "month", warnForColNameInEnv=FALSE, verbose=FALSE) setkeyIfNot(DT.MonthlyAnalytics, kColsDate, verbose=FALSE) # DT.MonthlyAnalytics[store==1 & date == '2013-10-01'] DT.MonthlyAnalytics[DT.iTunes544, `:=`(month_end=i.end, month_start=i.start), allow.cartesian=TRUE] DT.MonthlyAnalytics[(store != 'iTunes' & store != 1), `:=`(month_end=lastDateInMonth(date), month_start=firstDateInMonth(date) )] ## Check if max_date > month_end if (nrow(DT.MonthlyAnalytics[max_date>month_end])) { warning ("Some rows have a max_date beyond the month_end") print(DT.MonthlyAnalytics[max_date>month_end]) } DT.MonthlyAnalytics[, days_in_month := as.numeric(month_end - month_start + 1)] DT.MonthlyAnalytics[, days_present := as.numeric(max_date - (month_start-1) )] DT.MonthlyAnalytics[, days_missing := as.numeric(month_end - max_date)] stopifnot(DT.MonthlyAnalytics[, days_in_month == days_present + days_missing]) ## When month is full, we will have max_date == month_end. No (-1) needed. ## Set keys, prepare for merge setkeyIfNot(DT.MonthlyAnalytics, kColsDate, verbose=FALSE) setcolorderpt(DT.MonthlyAnalytics, kColsDate) ## Calculate expected paidunits, simple extrapolation by total number of days ## paidunits_expected_ratio is (total_days_in_month / days_present) == ((days_present + days_missing) / days_present) DT.MonthlyAnalytics[, paidunits_expected_ratio := days_in_month / days_present ] # dont use as integer, in case of overflow DT.MonthlyAnalytics[, paidunits_expected := round(paidunits / days_present * days_in_month)] # dont use as integer, in case of overflow ## FOR Dev'ing - Take backup BackUpOrRestore("DT.MonthlyAnalytics") #================ [NOT IN ANY OTHER FILE] ### DETERMINE What transaction types are missing from each DB, by store missing.minDate <- "2013-11-01" DT.missing <- merge(DT.MonthlyAnalytics[date >= missing.minDate & !is.na(paidunits), list(in.analytics=TRUE), keyby=list(store, transac_typeid)], DT.storeavg[date>'2013-11-01' & !is.na(paidunits), list(in.accounting=TRUE), keyby=list(store, transac_typeid)], all=TRUE) setInfo(DT.missing, "merge of DT.MonthlyAnalytics & DT.storeavg\nIdentifies store-transactiontype combinations that exist in one table but not another.\nie identifies a group that is analyitcs but not in accounting or vice versa") ## Clean up 'in.*' fields DT.missing[is.na(in.analytics), in.analytics := FALSE] DT.missing[is.na(in.accounting), in.accounting := FALSE] DT.missing[, store_in_analytics := any(in.analytics), by=store] ## Identify where missing_from DT.missing[, missing_from := ifelse(store_in_analytics & !in.analytics, "analytics", ifelse(!in.accounting, "accounting", NA))] key.bak.DT.missing <- key(DT.missing) setkey(DT.missing, transac_typeid) [DT.transacs, `:=`(transac_type=i.transac_type, transac_type_abbr=i.transac_type_abbr), allow=TRUE] setkey(DT.missing, store) [DT.stores, `:=`(store_name=i.store_name_short), allow=TRUE] setkeyIfNot(DT.missing, key.bak.DT.missing, verbose=FALSE) rm(key.bak.DT.missing) ## OUTPUT { cat("\t\t**** The following store+transaction_type are missing (looking back to ",missing.minDate,") ****\n\t\t", pasteR(93), "\n", sep="") print(DT.missing[!is.na(missing_from)] [order(missing_from, decreasing=TRUE) , list(storeid = store, store_name, transac_typeid, transac_type_abbr, transac_type, missing_from)]) } ## ADD IN THE GPU. Use Estimates when we dont have the full month. ## Keys for merging setkeyIfNot(DT.MonthlyAnalytics , kColsDate, verbose=FALSE) setkeyIfNot(DT.storeavg2 , kColsDate, verbose=FALSE) ## MERGE IN THE GPU INFO ## DT.MonthlyAnalytics[DT.storeavg2, `:=`(GPU_using = i.GPU, GPU_is_estimate = FALSE)] DT.MonthlyAnalytics[DT.storeavg2, `:=`(GPU_estimate = i.GPU_estimate, GPU_is_estimate = FALSE)] ## If we have three or less days of info for the group AND the estimated GPU for that group is not NA, ## Then replace the using with the estimate, and set the "GPU_is_estimate" flag to true DT.MonthlyAnalytics[(max_date_bygrp - month_start < 4) & !is.na(GPU_estimate) , `:=`(GPU_using = GPU_estimate, GPU_is_estimate = TRUE) ] ## If GPU_using is NA and the Estimate is not NA, use the estimate DT.MonthlyAnalytics[(is.na(GPU_using)) & !is.na(GPU_estimate) , `:=`(GPU_using = GPU_estimate, GPU_is_estimate = TRUE) ] ## Lastly, if there are days_missing, we consider the GPU_using to be an estimate (because it will change) DT.MonthlyAnalytics[days_missing != 0, GPU_is_estimate := TRUE ] ## TAKE A BACKUP jesusForData(DT.MonthlyAnalytics, info="No SC Group - Prior to inserting imputed values") ### ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ ### ### For now, the last month will take the previous month ### ### ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ ### ## Determine if each NA in GPU_using is in the middle of other values or not ## Values: before, middle, after DT.MonthlyAnalytics[, NA_location_gpu_using := factor(NA, levels=c("before", "middle", "after"))] DT.MonthlyAnalytics[is.na(GPU_using), NA_location_gpu_using := "middle"] invisible(suppressWarnings( DT.MonthlyAnalytics[, NA_location_gpu_using := { NA_location_gpu_using[date > max( date[!is.na(GPU_using)], na.rm=TRUE)] <- "after" NA_location_gpu_using[date < min( date[!is.na(GPU_using)], na.rm=TRUE)] <- "before" # NA_location_gpu_using[sum(is.na(GPU_using)) = .N] NA_location_gpu_using } , by=kCols] )) ## All the "after" rows are the ones that need filling DT.last <- DT.MonthlyAnalytics[NA_location_gpu_using == "after", kColsDate, with=FALSE] ## We will need to go back N many months, where N are the number of gaps currently present DT.last[, monthsback := .N, by=kCols] ## datefuture is where the old values are being assigned into DT.last[, datefuture := date] ## subtract back some months= DT.last[, date := lubridate:::.quick_month_add(date, -monthsback), by=monthsback] setkeyIfNot(DT.last, kColsDate, verbose=FALSE) DT.last[DT.MonthlyAnalytics, `:=`(GPU_using=i.GPU_using, NA_location_gpu_using=as.character(i.NA_location_gpu_using)), allow=TRUE] ## Check values. if (DT.last[NA_location_gpu_using != "middle", any(is.na(GPU_using)) ]) { warning ("Some NA's in DT.last[, GPU_using]") } else if (all(is.na(DT.last$NA_location_gpu_using))) { invisible(DT.last[, NA_location_gpu_using := NULL]) } ## Adjust dates back DT.last[, datefrom := date] DT.last[, date := datefuture] ## setkeys for merging matchKey(DT.MonthlyAnalytics, DT.last, kColsDate) ## Merge in the GPUs back to DT.MonthlyAnalytics DT.MonthlyAnalytics[DT.last, GPU_last := i.GPU_using, allow=TRUE] ## Confirm, there are no values that would be overwritten if (nrow(DT.MonthlyAnalytics[!is.na(GPU_last) & !is.na(GPU_using)])) warning("Some GPU_using & GPU_last are both non-NA") ## Assign over from GPU_last to GPU_using DT.MonthlyAnalytics[!is.na(GPU_last) & is.na(GPU_using), `:=` (GPU_using = GPU_last, GPU_is_estimate = TRUE) ] DT.MonthlyAnalytics[, GPU_last := NULL] ### ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ ### if (nrow(DT.MonthlyAnalytics[is.nan(GPU_using)])) { warning ("There are NaN's in GPU_using:") print(DT.MonthlyAnalytics[is.nan(GPU_using)]) } ## Take a manually moving average, removing NAs counter <- 4 while (counter > 0 && any(is.na(DT.MonthlyAnalytics$GPU_using))) { ## Expand greater at each iteration width <- {(-(2+(4-counter))):(3+(4-counter))} DT.MonthlyAnalytics[, c("GPU_using", "GPU_is_estimate") := { G <- GPU_using G_nas <- is.na(G) inds <- which(G_nas) # browser(expr=all(G_nas)) if (length(inds) && !all(G_nas)) { G_notnas <- setdiff(seq(G), inds) G[inds] <- sapply(inds, function(i) ## Index from 3back to 4forward, intersecting with current seq. ## Note that mean() will return NaN when mean(G[ intersect(i + width, G_notnas) ]) ) } list(G, (GPU_is_estimate | G_nas)) } , by=kCols] counter <- counter - 1 } ### ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ ### # skip for now # ## ---- IMPUTE ---- ### # skip for now # results_from_imputing <- mi(DT.MonthlyAnalytics[, list(date, store, transac_typeid, label_sc_group, GPU_using)], n.iter=5000, n.imp=6, max.minutes=15) # skip for now # imputedValues <- tail(results_from_imputing@imp, 1)[[1]] # skip for now # ## ---- IMPUTE ---- ### # skip for now # # skip for now # # skip for now # ## Check that it is the correct length # skip for now # if (length(imputedValues$GPU_using@random) == DT.MonthlyAnalytics[, sum(is.na(GPU_using))]) { # skip for now # ## Info for user # skip for now # message("Attempting to insert imputed values") # skip for now # nacount.prev <- DT.MonthlyAnalytics[, sum(is.na(GPU_using))] # skip for now # # skip for now # ## Adding an extra step of first creating it's own column so that I can double check that Ive lined them up right # skip for now # DT.MonthlyAnalytics[is.na(GPU_using), GPU_imputed := imputedValues$GPU_using@random] # skip for now # # skip for now # ## ERROR CHECK -- Exactly one column should be NA # skip for now # stopifnot(DT.MonthlyAnalytics[, all(xor(is.na(GPU_using), is.na(GPU_imputed)))]) # skip for now # # skip for now # ## Assign the value and note that it is an estimate and that it was imputed # skip for now # DT.MonthlyAnalytics[, GPU_was_imputed := FALSE] # skip for now # DT.MonthlyAnalytics[is.na(GPU_using) # skip for now # , `:=`(GPU_using = GPU_imputed, # skip for now # GPU_is_estimate = TRUE , # skip for now # GPU_was_imputed = TRUE # skip for now # )] # skip for now # # skip for now # ## Drop the column # skip for now # DT.MonthlyAnalytics[, GPU_imputed := NULL] # skip for now # # skip for now # nacount.post <- DT.MonthlyAnalytics[, sum(is.na(GPU_using))] # skip for now # message("Cleared ", nacount.prev - nacount.post, " NA values.\n\t ", nacount.post, " remain.") # skip for now # } ## Error check to see if there are any NA's remaining if (nrow(DT.missing2 <- DT.MonthlyAnalytics[is.na(GPU_using)])) { hr <- pasteR(65) message("\n", hr, "\n\t\tSome values of 'GPU_using' are still NA\n", hr) print(DT.missing2[order(store, transac_typeid, date)] ) cat("\n\n ", pasteR(12), " The corresponding Transsaction Types are ", pasteR(12),"\n\n") print(DT.transacs[.(sort(unique(DT.missing2$transac_typeid))), list(transac_typeid, transac_type_abbr, transac_type)]) message("\n", hr, "\n") } else { DT.MonthlyAnalytics[, GPU_estimate := NULL] rm(DT.missing2) } setcolorderpt(DT.MonthlyAnalytics, c("date", "store", "transac_typeid", "max_date_bygrp", "max_date", "month_start", "month_end", "days_in_month", "days_present", "days_missing", "paidunits", "paidunits_expected", "GPU_is_estimate", "GPU_using", "GPU_estimate")) jesusForData(DT.MonthlyAnalytics, info="After_filling_GPU_using") ## NOTE: Currently, the data is needed on both clusters. Thus iterating BackUpOrRestore("cluster.out", verboseRestore=TRUE, verbose=FALSE) # <~~ This line can be removed once cluster issue resolved for (cluster.out in unique(c(cluster.in, cluster.out))) { # <~~ This line can be removed once cluster issue resolved ## Change the DB to where we will export to dbDisconnectAll() message(sprintf("Uploading the DT.MonthlyAnalytics data to cluster #%02i", cluster.out)) setDBall(cluster=cluster.out) ## CREATE bi.gpu ( fromerlly DS_scratch.GPU ) ## Insert the table into SQL warning("2015-01-20 Reminder to RICK -- it is not clear if bi.gpu is using accounting_month or activity_month544") ingestIntoSQL(DT.MonthlyAnalytics , nmsToChange = c(activity_month544="date", storeid="store") , tbl = tbl.out , schema = schema.out , drop = TRUE , numericDecimals=c(22, 9) ) ## TODO: This can be settings in params for ingestIntoSQL qUpdatePerms(tbl=tbl.out, schema=schema.out, user="orcdpipeline") } # <~~ This line can be removed once cluster issue resolved BackUpOrRestore("cluster.out") # <~~ This line can be removed once cluster issue resolved x <- saveImageTo() cat("GPU Update image saved to\n '", x, "'\n", sep="") ## 'RETURN' x