anomaly_classification

anomaly_classification ¤

AnomalyClassificationEvents ¤

AnomalyClassificationEvents(
    dataframe: DataFrame,
    signal_uuid: str,
    *,
    event_uuid: str = "quality:anomaly_type",
    value_column: str = "value_double",
    time_column: str = "systime"
)

Bases: Base

Quality: Anomaly Classification

Classify anomalies in a numeric signal by type: spike, drift, oscillation, flatline, or level shift.

Methods: - classify_anomalies: Detect and classify anomalous windows. - detect_flatline: Signal stuck at constant value. - detect_oscillation: Periodic instability detection. - detect_drift: Short-term slope-based drift events.

detect_flatline ¤

detect_flatline(
    min_duration: str = "1min",
    tolerance: float = 1e-06,
    *,
    interval_end: str = "next_change",
    max_gap: str | Timedelta | None = None
) -> pd.DataFrame

Detect intervals where the signal is stuck at a constant value.

A run starts at its first sample (the value was already stuck there) and continues while each sample is within tolerance of the previous one. Samples at the same timestamp (duplicates) add no duration: a run whose samples all share one timestamp counts as a single sample.

Parameters:

Name	Type	Description	Default
`min_duration`	`str`	Minimum duration to qualify as a flatline.	`'1min'`
`tolerance`	`float`	Maximum step between consecutive samples to count as flat.	`1e-06`
`interval_end`	`str`	`"next_change"` (default) ends a run at the next differing sample, so a historian with a deadband -- one row, then nothing until the value moves -- still reports a stuck sensor. A single-sample run counts only when the gap after it is longer than the signal's median sampling interval. `"last_sample"` ends a run at its own last sample and needs at least two samples.	`'next_change'`
`max_gap`	`str \| Timedelta \| None`	Longest time between two samples a run may span (e.g. `"15min"`). A longer gap is "no data": the run before it ends at its last sample and a new run starts after it, even if the value is unchanged. `None` (default): no limit.	`None`

Returns:

Type	Description
`DataFrame`	DataFrame with columns: start, end, duration_seconds, uuid, source_uuid, duration, stuck_value. `duration` is a Timedelta; `duration_seconds` is the same span in float seconds.

detect_oscillation ¤

detect_oscillation(
    window: str = "1min", min_crossings: int = 6
) -> pd.DataFrame

Detect windows with excessive zero-crossings of the detrended signal.

Parameters:

Name	Type	Description	Default
`window`	`str`	Window size for analysis.	`'1min'`
`min_crossings`	`int`	Minimum zero-crossings to flag oscillation.	`6`

Returns:

Type	Description
`DataFrame`	DataFrame with columns: start, end, duration_seconds, uuid, source_uuid, crossing_count, amplitude.

detect_drift ¤

detect_drift(
    window: str = "1h",
    min_slope: float = 0.01,
    slope_per: str = "second",
) -> pd.DataFrame

Detect windows where the rolling slope exceeds a threshold.

A least-squares line is fitted to each non-overlapping window. With the default slope_per="second" the x-axis is elapsed time, so slope and min_slope are in signal units per second regardless of the sampling rate (and irregular sampling is handled correctly). With slope_per="sample" the x-axis is the sample index, so slope is in signal units per sample and the meaning of min_slope depends on the sampling rate.

NaN values are left out of the fit. A window whose fit fails (e.g. all of its samples share one timestamp) is skipped with a :class:~ts_shape.errors.DataQualityWarning.

Parameters:

Name	Type	Description	Default
`window`	`str`	Window size for slope computation.	`'1h'`
`min_slope`	`float`	Minimum absolute slope to flag, in signal units per sample or per second (see `slope_per`).	`0.01`
`slope_per`	`str`	`"second"` (default) or `"sample"`.	`'second'`

Returns:

Type	Description
`DataFrame`	DataFrame with columns: start, end, duration_seconds, uuid, source_uuid, slope, direction.

classify_anomalies ¤

classify_anomalies(
    window: str = "10min",
    n_sigma: float = 3.0,
    slope_per: str = "second",
) -> pd.DataFrame

Detect anomalous windows and classify by type.

Types: spike, drift, oscillation, flatline, level_shift.

A window is a drift when its least-squares slope exceeds 0.1 × the signal's overall std per sample. With the default slope_per="second" the slope is fitted against elapsed time and compared with 0.1 × std divided by the signal's median sampling interval (seconds): irregular sampling is handled correctly and the slope reported in details is in signal units per second. With slope_per="sample" the slope is fitted against the sample index, so the test depends on the sampling rate; on regularly sampled data both settings flag the same windows.

Parameters:

Name	Type	Description	Default
`window`	`str`	Window size for analysis.	`'10min'`
`n_sigma`	`float`	Z-score threshold for spike detection.	`3.0`
`slope_per`	`str`	`"second"` (default) or `"sample"`.	`'second'`

Returns:

Type	Description
`DataFrame`	DataFrame with columns: start, end, duration_seconds, uuid, source_uuid, anomaly_type, severity, details.