def __init__()

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

• 40 lines of code
• 16 McCabe index (conditional complexity)

    def __init__(self, data: TimeSeriesData, parameters: NormalKnownParameters):

        # \mu \sim N(\mu0, \frac{1}{\lambda0})
        # x \sim N(\mu,\frac{1}{\lambda})

        empirical = parameters.empirical
        mean_prior = parameters.mean_prior
        mean_prec_prior = parameters.mean_prec_prior
        known_prec = parameters.known_prec
        self.parameters = parameters
        self._maxT = len(data)

        # hyper parameters for mean and precision
        self.mu_0 = mean_prior
        self.lambda_0 = mean_prec_prior
        self.lambda_val = known_prec
        if data.is_univariate():
            self._data_shape = self._maxT
        else:
            # Multivariate
            self.P = data.value.values.shape[1]
            # If the user didn't specify the priors as multivariate
            # then we assume the same prior(s) over all time series.
            if self.mu_0 is not None and isinstance(self.mu_0, float):
                self.mu_0 = np.repeat(self.mu_0, self.P)
            if self.mu_0 is not None and isinstance(self.lambda_0, float):
                self.lambda_0 = np.repeat(self.lambda_0, self.P)
            if self.mu_0 is not None and isinstance(self.lambda_val, float):
                self.lambda_val = np.repeat(self.lambda_val, self.P)
            self._data_shape = (self._maxT, self.P)

        # For efficiency, we simulate a dynamically growing list with
        # insertions at the start, by a fixed size array with a pointer
        # where we grow the array from the end of the array. This
        # makes insertions constant time and means we can use
        # vectorized computation throughout.
        self._mean_arr_num = np.zeros(self._data_shape)
        self._std_arr = np.zeros(self._data_shape)
        self._ptr = 0

        # if priors are going to be decided empirically,
        # we ignore these settings above
        # Also, we need to pass on the data in this case
        if empirical:
            check_data(data)
            self._find_empirical_prior(data)

        if (
            self.lambda_0 is not None
            and self.lambda_val is not None
            and self.mu_0 is not None
        ):
            # We set these here to avoid recomputing the linear expression
            # throughout + avoid unnecessarily zeroing the memory etc.
            self._mean_arr = np.repeat(
                np.expand_dims(self.mu_0 * self.lambda_0, axis=0), self._maxT, axis=0
            )
            self._prec_arr = np.repeat(
                np.expand_dims(self.lambda_0, axis=0), self._maxT, axis=0
            )
        else:
            raise ValueError("Priors for NormalKnownPrec should not be None.")