add pipeop IPCW draft

DESCRIPTION

@@ -140,6 +140,7 @@ Collate:
     'PipeOpCrankCompositor.R'
     'PipeOpDistrCompositor.R'
     'PipeOpPredClassifSurvDiscTime.R'
+    'PipeOpPredClassifSurvIPCW.R'
     'PipeOpTransformer.R'
     'PipeOpPredTransformer.R'
     'PipeOpPredRegrSurv.R'
@@ -148,6 +149,7 @@ Collate:
     'PipeOpSurvAvg.R'
     'PipeOpTaskRegrSurv.R'
     'PipeOpTaskSurvClassifDiscTime.R'
+    'PipeOpTaskSurvClassifIPCW.R'
     'PipeOpTaskSurvRegr.R'
     'PipeOpTaskTransformer.R'
     'PredictionDataDens.R'

NAMESPACE

@@ -73,13 +73,15 @@ export(PipeOpBreslow)
 export(PipeOpCrankCompositor)
 export(PipeOpDistrCompositor)
 export(PipeOpPredClassifSurvDiscTime)
+export(PipeOpPredClassifSurvIPCW)
 export(PipeOpPredRegrSurv)
 export(PipeOpPredSurvRegr)
 export(PipeOpPredTransformer)
 export(PipeOpProbregr)
 export(PipeOpSurvAvg)
 export(PipeOpTaskRegrSurv)
 export(PipeOpTaskSurvClassifDiscTime)
+export(PipeOpTaskSurvClassifIPCW)
 export(PipeOpTaskSurvRegr)
 export(PipeOpTaskTransformer)
 export(PipeOpTransformer)
@@ -97,6 +99,7 @@ export(as_task_surv)
 export(assert_surv)
 export(breslow)
 export(pecs)
+export(pipeline_survtoclassif_IPCW)
 export(pipeline_survtoclassif_disctime)
 export(pipeline_survtoregr)
 export(plot_probregr)

R/PipeOpPredClassifSurvIPCW.R

@@ -0,0 +1,71 @@
+#' @title PipeOpPredClassifSurvIPCW
+#' @name mlr_pipeops_trafopred_classifsurv_IPCW
+#'
+#' @description
+#' Transform [PredictionClassif] to [PredictionSurv].
+#'
+#' @section Dictionary:
+#' This [PipeOp][mlr3pipelines::PipeOp] can be instantiated via the
+#' [dictionary][mlr3misc::Dictionary] [mlr3pipelines::mlr_pipeops]
+#' or with the associated sugar function [mlr3pipelines::po()]:
+#' ```
+#' PipeOpPredClassifSurvIPCW$new()
+#' mlr_pipeops$get("trafopred_classifsurv_IPCW")
+#' po("trafopred_classifsurv_IPCW")
+#' ```
+#'
+#' @section Input and Output Channels:
+#' The input is a [PredictionClassif] and a [data.table] containing observed times
+#' and row ids both generated by [PipeOpTaskSurvClassifIPCW].
+#' The output is the input [PredictionClassif] transformed to a [PredictionSurv].
+#' Only works during prediction phase.
+#'
+#' @family PipeOps
+#' @family Transformation PipeOps
+#' @export
+PipeOpPredClassifSurvIPCW = R6Class(
+  "PipeOpPredClassifSurvIPCW",
+  inherit = mlr3pipelines::PipeOp,
+
+  public = list(
+    #' @description
+    #' Creates a new instance of this [R6][R6::R6Class] class.
+    #' @param id (character(1))\cr
+    #' Identifier of the resulting object.
+    initialize = function(id = "trafopred_classifsurv_IPCW") {
+      super$initialize(
+        id = id,
+        input = data.table(
+          name = c("input", "data"),
+          train = c("NULL", "NULL"),
+          predict = c("PredictionClassif", "data.table")
+        ),
+        output = data.table(
+          name = "output",
+          train = "NULL",
+          predict = "PredictionSurv"
+        )
+      )
+    }
+  ),
+
+  private = list(
+    .predict = function(input) {
+      pred = input[[1]]
+      data = input[[2]]
+
+      p = PredictionSurv$new(row_ids = data$ids,
+                             truth = Surv(time = data$times,
+                                          event = as.integer(pred$truth)),
+                             crank = pred$prob[, 2])
+      list(p)
+    },
+
+    .train = function(input) {
+      self$state = list()
+      list(input)
+    }
+  )
+)
+
+register_pipeop("trafopred_classifsurv_IPCW", PipeOpPredClassifSurvIPCW)

R/PipeOpTaskSurvClassifIPCW.R

@@ -0,0 +1,133 @@
+#' @title PipeOpTaskSurvClassifIPCW
+#' @name mlr_pipeops_trafotask_survclassif_IPCW
+#' @template param_pipelines
+#'
+#' @description
+#' Transform [TaskSurv] to [TaskClassif][mlr3::TaskClassif] using IPCW (Vock et al., 2016).
+#'
+#' @section Dictionary:
+#' This [PipeOp][mlr3pipelines::PipeOp] can be instantiated via the
+#' [dictionary][mlr3misc::Dictionary] [mlr3pipelines::mlr_pipeops]
+#' or with the associated sugar function [mlr3pipelines::po()]:
+#' ```
+#' PipeOpTaskSurvClassifIPCW$new()
+#' mlr_pipeops$get("trafotask_survclassif_IPCW")
+#' po("trafotask_survclassif_IPCW")
+#' ```
+#'
+#' @section Input and Output Channels:
+#' [PipeOpTaskSurvClassifIPCW] has one input channel named "input", and two
+#' output channels, one named "output" and the other "data".
+#'
+#' During training, the "output" is the "input" [TaskSurv] transformed to a
+#' [TaskClassif][mlr3::TaskClassif].
+#' The target column is named `"status"` and indicates whether an event occurred
+#' in each time interval.
+#' The transformed task now has the property "weights".
+#' The "data" is NULL.
+#'
+#' During prediction, the "input" [TaskSurv] is transformed to the "output"
+#' [TaskClassif][mlr3::TaskClassif] with `"status"` as target.
+#' The "data" is a [data.table] containing the "time" of each subject as well
+#' as corresponding "row_ids".
+#' This "data" is only meant to be used with the [PipeOpPredClassifSurvIPCW].
+#'
+#' @section Parameters:
+#' The parameters are
+#'
+#' * `cutoff_time :: numeric()`\cr
+#' Cutoff time for IPCW. Observations with time larger than `cutoff_time` are censored.
+#' * `eps :: numeric()`\cr
+#' Small value to replace `0` survival probabilities with to prevent infinite weights.
+#'
+#' @references
+#' `r format_bib("vock_2016")`
+#'
+#' @family PipeOps
+#' @family Transformation PipeOps
+#' @export
+PipeOpTaskSurvClassifIPCW = R6Class(
+  "PipeOpTaskSurvClassifIPCW",
+  inherit = mlr3pipelines::PipeOp,
+
+  public = list(
+    #' @description
+    #' Creates a new instance of this [R6][R6::R6Class] class.
+    initialize = function(id = "trafotask_survclassif_IPCW") {
+      param_set = ps(
+        cutoff_time = p_dbl(lower = 0, special_vals = list()),
+        eps = p_dbl(lower = 0, default = 1e-6)
+      )
+      super$initialize(
+        id = id,
+        param_set = param_set,
+        input = data.table(
+          name    = "input",
+          train   = "TaskSurv",
+          predict = "TaskSurv"
+        ),
+        output = data.table(
+          name    = c("output", "data"),
+          train   = c("TaskClassif", "NULL"),
+          predict = c("TaskClassif", "data.table")
+        )
+      )
+    }
+  ),
+
+  private = list(
+    .predict = function(input) {
+      data = input[[1]]$data()
+      data$status = factor(data$status, levels = c("0", "1"))
+      task = TaskClassif$new(id = input[[1]]$id, backend = data,
+                             target = "status", positive = "1")
+
+      time = data[[input[[1]]$target_names[1]]]
+      data = data.table(ids = input[[1]]$row_ids, times = time)
+      list(task, data)
+    },
+
+    .train = function(input) {
+      data = input[[1]]$data()
+      time_var = input[[1]]$target_names[1]
+      status_var = input[[1]]$target_names[2]
+
+      cutoff_time = self$param_set$values$cutoff_time
+      eps = self$param_set$values$eps
+
+      if (cutoff_time >= max(data[[time_var]])) {
+        stop("Cutoff time must be smaller than the maximum event time.")
+      }
+
+      # transform data and calculate weights
+      times = data[[time_var]]
+      times[times > cutoff_time] = cutoff_time
+
+      status = data[[status_var]]
+      status[times == cutoff_time] = 0
+
+      cens = survival::survfit(Surv(times, 1 - status) ~ 1)
+      cens$surv[length(cens$surv)] = cens$surv[length(cens$surv)-1]
+      cens$surv[cens$surv == 0] = eps
+
+      weights = rep(1/cens$surv, table(times))
+
+      # add weights to original data
+      data[["ipc_weights"]] = weights
+      data[status_var == 0 & time_var < cutoff_time, "ipc_weights" := 0]
+      data[[status_var]] = factor(data[[status_var]], levels = c("0", "1"))
+      data[[time_var]] = NULL
+
+      # create new task
+      task = TaskClassif$new(id = paste0(input[[1]]$id, "_IPCW"), backend = data,
+                             target = status_var, positive = "1")
+
+      task$set_col_roles("ipc_weights", roles = "weight")
+
+      self$state = list()
+      list(task, NULL)
+    }
+  )
+)
+
+register_pipeop("trafotask_survclassif_IPCW", PipeOpTaskSurvClassifIPCW)

R/PipeOpTaskSurvRegr.R

@@ -21,7 +21,7 @@
 #'
 #' * `method::character(1))`\cr
 #' Method to use for dealing with censoring. Options are `"ipcw"` (Vock et al., 2016): censoring
-#' is column is removed and a `weights` column is added, weights are inverse estimated survival
+#' column is removed and a `weights` column is added, weights are inverse estimated survival
 #'  probability of the censoring distribution evaluated at survival time;
 #'  `"mrl"` (Klein and Moeschberger, 2003): survival time of censored
 #' observations is transformed to the observed time plus the mean residual life-time at the moment

R/pipelines.R

@@ -610,10 +610,66 @@ pipeline_survtoclassif_disctime = function(learner, cut = NULL, max_time = NULL,
   gr
 }
 
+#' @name mlr_graphs_survtoclassif_IPCW
+#' @title Survival to Classification Reduction Pipeline using IPCW
+#' @description Wrapper around multiple [PipeOp][mlr3pipelines::PipeOp]s to help in creation
+#' of complex survival reduction methods.
+#'
+#' @param learner [LearnerClassif][mlr3::LearnerClassif]\cr
+#' Classification learner to fit the transformed [TaskClassif][mlr3::TaskClassif].
+#' @param cutoff_time `numeric()`\cr
+#' Cutoff time for IPCW. Observations with time larger than `cutoff_time` are censored.
+#' @param eps `numeric()`\cr
+#' Small value to replace `0` survival probabilities with to prevent infinite weights.
+#' @param output `numeric()`\cr
+#' If not set to "classif" (default) then the prediction is transformed to a crank.
+#' @param graph_learner `logical(1)`\cr
+#' If `TRUE` returns wraps the [Graph][mlr3pipelines::Graph] as a
+#' [GraphLearner][mlr3pipelines::GraphLearner] otherwise (default) returns as a `Graph`.
+#'
+#' @details
+#' The pipeline consists of the following steps:
+#' \enumerate{
+#' \item [PipeOpTaskSurvClassifIPCW] Converts [TaskSurv] to a [TaskClassif][mlr3::TaskClassif].
+#' \item A [LearnerClassif] is fit and predicted on the new `TaskClassif`.
+#' \item Optionally: [PipeOpPredClassifSurvIPCW] transforms the resulting [PredictionClassif][mlr3::PredictionClassif]
+#' to [PredictionSurv].
+#' }
+#'
+#' @return [mlr3pipelines::Graph] or [mlr3pipelines::GraphLearner]
+#' @family pipelines
+#'
+#' @export
+pipeline_survtoclassif_IPCW = function(learner, cutoff_time = NULL, eps = 1e-6, output = "classif", graph_learner = FALSE) {
+  assert_true("prob" %in% learner$predict_types)
+
+  gr = mlr3pipelines::Graph$new()
+  gr$add_pipeop(mlr3pipelines::po("trafotask_survclassif_IPCW", cutoff_time = cutoff_time))
+  gr$add_pipeop(mlr3pipelines::po("learner", learner, predict_type = "prob"))
+
+  gr$add_edge(src_id = "trafotask_survclassif_IPCW", dst_id = learner$id, src_channel = "output", dst_channel = "input")
+
+  if (output != "classif") {
+    gr$add_pipeop(mlr3pipelines::po("trafopred_classifsurv_IPCW"))
+    gr$add_pipeop(mlr3pipelines::po("nop"))
+
+    gr$add_edge(src_id = learner$id, dst_id = "trafopred_classifsurv_IPCW", src_channel = "output", dst_channel = "input")
+    gr$add_edge(src_id = "trafotask_survclassif_IPCW", dst_id = "nop", src_channel = "data", dst_channel = "input")
+    gr$add_edge(src_id = "nop", dst_id = "trafopred_classifsurv_IPCW", src_channel = "output", dst_channel = "data")
+  }
+
+  if (graph_learner) {
+    gr = mlr3pipelines::GraphLearner$new(gr)
+  }
+
+  gr
+}
+
 register_graph("survaverager", pipeline_survaverager)
 register_graph("survbagging", pipeline_survbagging)
 register_graph("crankcompositor", pipeline_crankcompositor)
 register_graph("distrcompositor", pipeline_distrcompositor)
 register_graph("probregr", pipeline_probregr)
 register_graph("survtoregr", pipeline_survtoregr)
 register_graph("survtoclassif_disctime", pipeline_survtoclassif_disctime)
+register_graph("survtoclassif_IPCW", pipeline_survtoclassif_IPCW)