#   ~/git/orch/src/Discover_Weekly/supportFns/make_accel_cols.r

make_accel_cols <- function(col, orderby=dateCol, partitionby=ctp, accel_col.included=TRUE, scale_col.included=accel_col.included, updown_col.included=TRUE, ctp, dateCol="date", nms=colNamesFromVector(col))  {


  if (!missing(dateCol))
    warning("dateCol deprecated -- use 'orderby'")
  if (!missing(ctp))
    warning("ctp deprecated -- use 'partitionby'")

  if (!length(orderby))
    warning("orderby has no length;  accel columns might crash the query")

  partition_clause <- partitionby %>% setdiff(orderby) %>% setdiff(col) %>% 
                      {ifelse(length(.), sprintf("PARTITION BY %s ", commaSep(.)), "")}

  LG <- sprintf("(LAG(%s, 1, NULL) OVER (%sORDER BY %s ASC))", col, partition_clause, orderby)

  diffCol        <- sprintf("(%2$s - %1$s)", LG, col)                                %>% setNames(obj=., nm=paste0(nms, "_increase"))
  percCol        <- sprintf("(%2$s - %1$s)/NULLIF(%1$s, 0)", LG, col)                %>% setNames(obj=., nm=paste0(nms, "_perc_increase"))
  accelCol       <- sprintf("score_accel_before_after(%1$s, %2$s)", LG, col)         %>% setNames(obj=., nm=paste0(nms, "_accel_score"))
  scaledAccelCol <- sprintf("scaled_score_accel_before_after(%1$s, %2$s)", LG, col)  %>% setNames(obj=., nm=paste0(nms, "_accel_score_scaled"))

  updownCol      <- sprintf("sign (%s)", diffCol)                                    %>% setNames(obj=., nm=paste0(nms, "_is_up_or_down"))
  # LG_of_updown   <- sprintf("(LAG(%s, 1, NULL) OVER (%sORDER BY %s ASC))", updownCol, partition_clause, orderby)
  # local_extrema  <- sprintf("sign(%s - %s)", updownCol, LG_of_updown)                %>% setNames(obj=., nm=paste0(nms, "_is_local_extrema"))

  ret <- c(diffCol, percCol, if (accel_col.included) accelCol, if (scale_col.included) scaledAccelCol, if (updown_col.included) updownCol)

  ## regroup the output according to the input
  if (length(col) > 1) {
    col_order <- lapply(nms, function(x) paste0(x, c("_increase", "_perc_increase", "_accel_score", "_accel_score_scaled", "_is_up_or_down", "_is_local_extrema"))) %>% unlist(use.names=FALSE)
    col_order %<>% intersect(names(ret)) %>% c(., setdiff(names(ret), .))
    if (all(col_order %in% names(ret)) && all(names(ret) %in% col_order))
      ret <- ret[col_order]
  }
  return(ret)
}

make_local_extrema_col <- function(col, orderby, partitionby, nms=colNamesFromVector(col))  {

  if (!length(orderby))
    warning("orderby has no length;  accel columns might crash the query")

  partition_clause <- partitionby %>% setdiff(orderby) %>% setdiff(col) %>% 
                      {ifelse(length(.), sprintf("PARTITION BY %s ", commaSep(.)), "")}

  # LG <- sprintf("(LAG(%s, 1, NULL) OVER (%sORDER BY %s ASC))", col, partition_clause, orderby)

  # diffCol        <- sprintf("(%2$s - %1$s)", LG, col)                                %>% setNames(obj=., nm=paste0(nms, "_increase"))
  # percCol        <- sprintf("(%2$s - %1$s)/%1$s", LG, col)                           %>% setNames(obj=., nm=paste0(nms, "_perc_increase"))
  # accelCol       <- sprintf("score_accel_before_after(%1$s, %2$s)", LG, col)         %>% setNames(obj=., nm=paste0(nms, "_accel_score"))
  # scaledAccelCol <- sprintf("scaled_score_accel_before_after(%1$s, %2$s)", LG, col)  %>% setNames(obj=., nm=paste0(nms, "_accel_score_scaled"))

  LG_of_updown   <- sprintf("(LAG(%s, 1, NULL) OVER (%sORDER BY %s ASC))", col, partition_clause, orderby)
  local_extrema  <- sprintf("sign(%s - %s)", col, LG_of_updown)                %>% setNames(obj=., nm=paste0(nms, "_is_local_extrema"))

  ret <- c(local_extrema)

  ## regroup the output according to the input
  if (length(col) > 1) {
    col_order <- lapply(nms, function(x) paste0(x, c("_increase", "_perc_increase", "_accel_score", "_accel_score_scaled", "_is_up_or_down", "_is_local_extrema"))) %>% unlist(use.names=FALSE)
    col_order %<>% intersect(names(ret)) %>% c(., setdiff(names(ret), .))
    if (all(col_order %in% names(ret)) && all(names(ret) %in% col_order))
      ret <- ret[col_order]
  }
  return(ret)
}

stop("TODO:  \n\n You want to spot when the accel score is growing or shrinking.   Anytime Accel score is growing, that's what we want!")


# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

## TESTS: 

if (FALSE)
{
  # make_accel_cols(c("streams_accel_score", "streams_accel_score_scaled"), orderby="weekof", partitionby="isrc", accel_col.included=TRUE) %>% makeQry(tbl="TTTTT")
  make_accel_cols(c(score="streams_accel_score_scaled"), orderby="weekof", partitionby="isrc", accel_col.included=TRUE, nms="metascore")
}

if (FALSE)
{
  dateCol <- "weekof"
  ctp_group <- c("isrc")
  accel_col <- "streams_accel_score_scaled"
  where <- list(isrc = 'USMEI1300005')
  where <- NULL
  colsToPull_accel <- c(ctp_group, dateCol, "streams", "streams_perc_increase", streams_accel=accel_col, 
                        make_accel_cols(c(metascore=accel_col), orderby=dateCol, partitionby=ctp_group, accel_col.included=TRUE)
                      )
  qry.metascore_inner <- makeQry(colsToPull=colsToPull_accel, schema="spot_stream_counts", tbl="by_isrc", whereIn=where, orderby="weekof", limit=NULL, colsToAgg=NULL)

  
  colsToPull_accel_outer=c("*", make_local_extrema_col(c(metascore="metascore_is_up_or_down"), orderby=dateCol, partitionby=ctp_group))
  qry.metascore_outer <- makeQry(colsToPull=colsToPull_accel_outer, colsToAgg=NULL, schema=NULL, tbl=qry.metascore_inner, orderby="weekof", limit=NULL)

  sfPopulateTable(qry=qry.metascore_outer, tbl="accel_metascore_by_isrc", schema="spot_stream_counts", over=TRUE)
}


if (FALSE)
{
 DT.accel_accel <- sfQry("
SELECT
isrc,
weekof,
streams, 
STREAMS_PERC_INCREASE

----------------------------------- 
, ' | ' as \"_\"
, STREAMS_ACCEL_SCORE
----------------------------------- 
, (streams_accel_score - (LAG(streams_accel_score, 1, NULL) OVER (PARTITION BY isrc ORDER BY   weekof ASC))) AS score_increase
, (streams_accel_score - (LAG(streams_accel_score, 1 , NULL) OVER (PARTITION BY isrc ORDER BY   weekof ASC)))/(LAG(streams_accel_score, 1, NULL) OVER (PARTITION BY isrc ORDER BY   weekof ASC)) AS score_perc_increase
, score_accel_before_after((LAG(streams_accel_score, 1, NULL) OVER (PARTITION BY isrc ORDER BY   weekof ASC)), streams_accel_score) AS score_accel_score
, scaled_score_accel_before_after((LAG(streams_accel_score, 1 , NULL) OVER (PARTITION BY isrc ORDER BY   weekof ASC)), streams_accel_score) AS score_accel_score_scaled
, sign ((streams_accel_score - (LAG(streams_accel_score, 1, NULL) OVER (PARTITION BY isrc ORDER BY   weekof ASC)))) AS score_is_up_or_down



--  , STREAMS_ACCEL_SCORE
--  ----------------------------
--  , (streams_accel_score_scaled - (LAG(streams_accel_score_scaled, 1 , NULL) OVER (PARTITION BY isrc ORDER BY   weekof ASC))) AS streams_accel_score_scaled_increase
--  , (streams_accel_score_scaled - (LAG(streams_accel_score_scaled, 1 , NULL) OVER (PARTITION BY isrc ORDER BY   weekof ASC)))/(LAG(streams_accel_score_scaled, 1, NULL) OVER (PARTITION BY isrc ORDER BY   weekof ASC)) AS streams_accel_score_scaled_perc_increase
--  , score_accel_before_after((LAG(streams_accel_score_scaled, 1, NULL) OVER (PARTITION BY isrc ORDER BY   weekof ASC)), streams_accel_score_scaled) AS streams_accel_score_scaled_accel_score
--  , scaled_score_accel_before_after((LAG(streams_accel_score_scaled, 1, NULL) OVER (PARTITION BY isrc ORDER BY   weekof ASC)), streams_accel_score_scaled) AS streams_accel_score_scaled_accel_score_scaled
--  , sign ((streams_accel_score_scaled - (LAG(streams_accel_score_scaled, 1 , NULL) OVER (PARTITION BY isrc ORDER BY   weekof ASC)))) AS streams_accel_score_scaled_is_up_or_down

FROM spot_stream_counts.by_isrc_TEST
where isrc = 'USMEI1300005'
order by weekof;
")
print(formnumb(DT.accel_accel, round=FALSE)[])


# DT.explore <- DT.accel_accel[, list(isrc, weekof, streams, updown=streams_accel_score_is_up_or_down)]
# DT.explore[, d := sign(diffNA(updown))]
# ## CASE 
#   WHEN d = 0
#    then continue direction
#   WHEN d < 0
#    then local_max
#   WHEN d > 0
#    then local_min

# DT.explore

}