survivalist.base

  1# This program is free software: you can redistribute it and/or modify
  2# it under the terms of the GNU General Public License as published by
  3# the Free Software Foundation, either version 3 of the License, or
  4# (at your option) any later version.
  5#
  6# This program is distributed in the hope that it will be useful,
  7# but WITHOUT ANY WARRANTY; without even the implied warranty of
  8# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  9# GNU General Public License for more details.
 10#
 11# You should have received a copy of the GNU General Public License
 12# along with this program.  If not, see <http://www.gnu.org/licenses/>.
 13import numpy as np
 14
 15
 16class SurvivalAnalysisMixin:
 17    def _predict_function(self, func_name, baseline_model, prediction, return_array, **kwargs):
 18        fns = getattr(baseline_model, func_name)(prediction)
 19
 20        if not return_array:
 21            return fns
 22
 23        times = baseline_model.unique_times_
 24        arr = np.empty((prediction.shape[0], times.shape[0]), dtype=float)
 25        for i, fn in enumerate(fns):
 26            arr[i, :] = fn(times)
 27        return arr
 28
 29    def _predict_survival_function(self, baseline_model, prediction, return_array, **kwargs):
 30        """Return survival functions.
 31
 32        Parameters
 33        ----------
 34        baseline_model : survivalist.linear_model.coxph.BreslowEstimator
 35            Estimator of baseline survival function.
 36
 37        prediction : array-like, shape=(n_samples,)
 38            Predicted risk scores.
 39
 40        return_array : bool
 41            If True, return a float array of the survival function
 42            evaluated at the unique event times, otherwise return
 43            an array of :class:`survivalist.functions.StepFunction` instances.
 44
 45        Returns
 46        -------
 47        survival : ndarray
 48        """
 49        return self._predict_function(
 50            "get_survival_function",
 51            baseline_model,
 52            prediction,
 53            return_array,
 54            **kwargs,
 55        )
 56
 57    def _predict_cumulative_hazard_function(self, baseline_model, prediction, return_array, **kwargs):
 58        """Return cumulative hazard functions.
 59
 60        Parameters
 61        ----------
 62        baseline_model : survivalist.linear_model.coxph.BreslowEstimator
 63            Estimator of baseline cumulative hazard function.
 64
 65        prediction : array-like, shape=(n_samples,)
 66            Predicted risk scores.
 67
 68        return_array : bool
 69            If True, return a float array of the cumulative hazard function
 70            evaluated at the unique event times, otherwise return
 71            an array of :class:`survivalist.functions.StepFunction` instances.
 72
 73        Returns
 74        -------
 75        cum_hazard : ndarray
 76        """
 77        return self._predict_function(
 78            "get_cumulative_hazard_function",
 79            baseline_model,
 80            prediction,
 81            return_array,
 82            **kwargs,
 83        )
 84
 85    def score(self, X, y):
 86        """Returns the concordance index of the prediction.
 87
 88        Parameters
 89        ----------
 90        X : array-like, shape = (n_samples, n_features)
 91            Test samples.
 92
 93        y : structured array, shape = (n_samples,)
 94            A structured array containing the binary event indicator
 95            as first field, and time of event or time of censoring as
 96            second field.
 97
 98        Returns
 99        -------
100        cindex : float
101            Estimated concordance index.
102        """
103        from .metrics import concordance_index_censored
104
105        name_event, name_time = y.dtype.names
106
107        try:
108            risk_score = self.predict(X)
109        except Exception as e:
110            print("Exception", e)
111            preds = self.predict(X)
112            # print("preds", preds)
113            risk_score = preds.mean
114            print("risk_score", risk_score)
115        if not getattr(self, "_predict_risk_score", True):
116            risk_score *= -1  # convert prediction on time scale to risk scale
117        print("y[name_event]", y[name_event])
118        print("y[name_time]", y[name_time])
119        print("y[name_event].shape", y[name_event].shape)
120        print("y[name_time].shape", y[name_time].shape)
121        print("risk_score", risk_score)
122        result = concordance_index_censored(
123            y[name_event], y[name_time], risk_score)
124        return result[0]
125
126    def _more_tags(self):
127        return {"requires_y": True}