def detector_()

in kats/detectors/cusum_detection.py [0:0]

• 45 lines of code
• 10 McCabe index (conditional complexity)

    def detector_(self, **kwargs) -> List[List[CUSUMChangePoint]]:
        """
        Detector method for vectorized version of CUSUM

        Args:
            threshold: Optional; float; significance level, default: 0.01.
            max_iter: Optional; int, maximum iteration in finding the
                changepoint.
            change_directions: Optional; list<str>; a list contain either or
                both 'increase' and 'decrease' to specify what type of change
                want to detect.
            return_all_changepoints: Optional; bool; return all the changepoints
                found, even the insignificant ones.

        Returns:
            A list of tuple of TimeSeriesChangePoint and CUSUMMetadata.
        """
        # Extract all arg values or assign defaults from default vals constant
        threshold = _get_arg("threshold", **kwargs)
        max_iter = _get_arg("max_iter", **kwargs)
        change_directions = _get_arg("change_directions", **kwargs)
        return_all_changepoints = _get_arg("return_all_changepoints", **kwargs)

        # Use array to store the data
        ts_all = self.data.value.to_numpy()
        ts_all = ts_all.astype("float64")
        changes_meta_list = []

        if change_directions is None:
            change_directions = ["increase", "decrease"]

        change_meta_all = {}
        for change_direction in change_directions:
            if change_direction not in {"increase", "decrease"}:
                raise ValueError(
                    "Change direction must be 'increase' or 'decrease.' "
                    f"Got {change_direction}"
                )

            change_meta_all[change_direction] = self._get_change_point_multiple_ts(
                ts_all,
                max_iter=max_iter,
                change_direction=change_direction,
            )

        ret = []
        for col_idx in np.arange(ts_all.shape[1]):
            ts = ts_all[:, col_idx]
            changes_meta = {}
            for change_direction in change_directions:
                change_meta_ = change_meta_all[change_direction]
                change_meta = {
                    k: change_meta_[k][col_idx]
                    if isinstance(change_meta_[k], np.ndarray)
                    or isinstance(change_meta_[k], list)
                    else change_meta_[k]
                    for k in change_meta_
                }
                change_meta["llr"] = self._get_llr(ts, change_meta)
                change_meta["p_value"] = 1 - chi2.cdf(change_meta["llr"], 2)
                change_meta["regression_detected"] = change_meta["p_value"] < threshold
                changes_meta[change_direction] = change_meta
            changes_meta_list.append(changes_meta)
            ret.append(
                self._convert_cusum_changepoints(changes_meta, return_all_changepoints)
            )
        self.changes_meta_list = changes_meta_list
        return ret