Skip to content

Window-based change point detection#

ruptures.detection.window.Window #

Window sliding method.

__init__(self, width=100, model='l2', custom_cost=None, min_size=2, jump=5, params=None) special #

Instanciate with window length.

Parameters:

Name Type Description Default
width int

window length. Defaults to 100 samples.

100
model str

segment model, ["l1", "l2", "rbf"]. Not used if custom_cost is not None.

'l2'
custom_cost BaseCost

custom cost function. Defaults to None.

None
min_size int

minimum segment length.

2
jump int

subsample (one every jump points).

5
params dict

a dictionary of parameters for the cost instance.`

None
Source code in ruptures/detection/window.py
def __init__(
    self, width=100, model="l2", custom_cost=None, min_size=2, jump=5, params=None
):
    """Instanciate with window length.

    Args:
        width (int, optional): window length. Defaults to 100 samples.
        model (str, optional): segment model, ["l1", "l2", "rbf"]. Not used if `custom_cost` is not None.
        custom_cost (BaseCost, optional): custom cost function. Defaults to None.
        min_size (int, optional): minimum segment length.
        jump (int, optional): subsample (one every *jump* points).
        params (dict, optional): a dictionary of parameters for the cost instance.`
    """
    self.min_size = min_size
    self.jump = jump
    self.width = 2 * (width // 2)
    self.n_samples = None
    self.signal = None
    self.inds = None
    if custom_cost is not None and isinstance(custom_cost, BaseCost):
        self.cost = custom_cost
    else:
        if params is None:
            self.cost = cost_factory(model=model)
        else:
            self.cost = cost_factory(model=model, **params)
    self.score = list()

fit(self, signal) #

Compute params to segment signal.

Parameters:

Name Type Description Default
signal array

signal to segment. Shape (n_samples, n_features) or (n_samples,).

required

Returns:

Type Description
Window

self

Source code in ruptures/detection/window.py
def fit(self, signal) -> "Window":
    """Compute params to segment signal.

    Args:
        signal (array): signal to segment. Shape (n_samples, n_features) or (n_samples,).

    Returns:
        self
    """
    # update some params
    if signal.ndim == 1:
        self.signal = signal.reshape(-1, 1)
    else:
        self.signal = signal
    self.n_samples, _ = self.signal.shape
    # indexes
    self.inds = np.arange(self.n_samples, step=self.jump)
    # delete borders
    keep = (self.inds >= self.width // 2) & (
        self.inds < self.n_samples - self.width // 2
    )
    self.inds = self.inds[keep]
    self.cost.fit(signal)
    # compute score
    score = list()
    for k in self.inds:
        start, end = k - self.width // 2, k + self.width // 2
        gain = self.cost.error(start, end)
        if np.isinf(gain) and gain < 0:
            # segment is constant and no improvment possible on start .. end
            score.append(0)
            continue
        gain -= self.cost.error(start, k) + self.cost.error(k, end)
        score.append(gain)
    self.score = np.array(score)
    return self

fit_predict(self, signal, n_bkps=None, pen=None, epsilon=None) #

Helper method to call fit and predict once.

Source code in ruptures/detection/window.py
def fit_predict(self, signal, n_bkps=None, pen=None, epsilon=None):
    """Helper method to call fit and predict once."""
    self.fit(signal)
    return self.predict(n_bkps=n_bkps, pen=pen, epsilon=epsilon)

predict(self, n_bkps=None, pen=None, epsilon=None) #

Return the optimal breakpoints.

Must be called after the fit method. The breakpoints are associated with the signal passed to fit(). The stopping rule depends on the parameter passed to the function.

Parameters:

Name Type Description Default
n_bkps int

number of breakpoints to find before stopping.

None
pen float

penalty value (>0)

None
epsilon float

reconstruction budget (>0)

None

Exceptions:

Type Description
AssertionError

if none of n_bkps, pen, epsilon is set.

BadSegmentationParameters

in case of impossible segmentation configuration

Returns:

Type Description
list

sorted list of breakpoints

Source code in ruptures/detection/window.py
def predict(self, n_bkps=None, pen=None, epsilon=None):
    """Return the optimal breakpoints.

    Must be called after the fit method. The breakpoints are associated with the signal passed
    to [`fit()`][ruptures.detection.window.Window.fit].
    The stopping rule depends on the parameter passed to the function.

    Args:
        n_bkps (int): number of breakpoints to find before stopping.
        pen (float): penalty value (>0)
        epsilon (float): reconstruction budget (>0)

    Raises:
        AssertionError: if none of `n_bkps`, `pen`, `epsilon` is set.
        BadSegmentationParameters: in case of impossible segmentation
            configuration

    Returns:
        list: sorted list of breakpoints
    """
    # raise an exception in case of impossible segmentation configuration
    if not sanity_check(
        n_samples=self.cost.signal.shape[0],
        n_bkps=1 if n_bkps is None else n_bkps,
        jump=self.jump,
        min_size=self.min_size,
    ):
        raise BadSegmentationParameters

    msg = "Give a parameter."
    assert any(param is not None for param in (n_bkps, pen, epsilon)), msg

    bkps = self._seg(n_bkps=n_bkps, pen=pen, epsilon=epsilon)
    return bkps