energykit.anomaly

Smart meter anomaly detection with financial impact quantification.

energykit.anomaly — smart meter anomaly detection with financial impact.

class energykit.anomaly.AnomalySummary(n_anomalies: int, anomaly_rate_pct: float, total_excess_kwh: float, total_estimated_cost_usd: float, worst_event: Series | None, top_anomalies_df: DataFrame, anomalies_df: DataFrame)[source]

Bases: object

High-level summary of a MeterAnomalyDetector.detect() run.

n_anomalies

Type:: int

anomaly_rate_pct

Type:: float

total_excess_kwh

Sum of excess energy across all anomalies (above baseline).

Type:: float

total_estimated_cost_usd

Estimated dollar impact of all anomalies.

Type:: float

worst_event

Row from the anomaly DataFrame with the highest estimated_cost_usd.

Type:: pd.Series

top_anomalies_df

Top 10 most expensive anomaly events.

Type:: pd.DataFrame

anomalies_df

Full detection result.

Type:: pd.DataFrame

anomalies_df: DataFrame

anomaly_rate_pct: float

n_anomalies: int

top_anomalies_df: DataFrame

total_estimated_cost_usd: float

total_excess_kwh: float

worst_event: Series | None

class energykit.anomaly.MeterAnomalyDetector(z_threshold: float = 2.5, overnight_hours: Tuple[int, int] = (0, 5), sustained_window: int = 3, use_isolation_forest: bool = False)[source]

Bases: object

Detect energy meter anomalies with financial impact estimation.

Parameters:

z_threshold (float, default 2.5) – Number of standard deviations above/below the seasonal baseline that classify a reading as an anomaly. Lower = more sensitive.
overnight_hours (tuple of int, default (0, 5)) – Half-open interval [start, stop) of hours considered “off-peak overnight”. Anomalies in this window get labelled overnight.
sustained_window (int, default 3) – Minimum number of consecutive anomalous readings to be labelled sustained_elevation (instead of individual spike).
use_isolation_forest (bool, default False) – If True and scikit-learn is available, adds a second layer of detection using IsolationForest on the residuals. Useful for multivariate or pattern-based anomalies.

Examples

>>> detector = MeterAnomalyDetector(z_threshold=2.5)
>>> detector.fit(training_series)      # learn seasonal baseline
>>> result = detector.detect(new_series, energy_price=0.15)
>>> print(result)
AnomalySummary(n=23, rate=0.26%, waste=312.4 kWh, cost=$46.86)

detect(series: Series, energy_price: float = 0.15) → AnomalySummary[source]

Detect anomalies in meter readings and estimate financial impact.

Parameters:

series (pd.Series) – Meter readings with a pd.DatetimeIndex.
energy_price (float, default 0.15) – Energy price in $/kWh. Used to estimate the cost of excess energy.

Return type:

AnomalySummary

fit(series: Series) → MeterAnomalyDetector[source]

Learn baseline patterns from historical meter data.

Parameters:: series (pd.Series) – Meter readings (kW or kWh) with a pd.DatetimeIndex. More historical data (≥60 days) produces more reliable baselines.
Return type:: self

Meter Anomaly Detector

energykit.anomaly.detector

Smart meter anomaly detection with financial impact quantification.

The key insight missing from every other anomaly detection library in the energy space: anomaly detection is only useful if it tells you what it’s costing you. This module detects anomalies and estimates the dollar value of each one.

Detection strategy

Uses a seasonal baseline approach: for each (hour-of-day, day-of-week) slot, compute the median historical reading. An anomaly is a reading that deviates beyond a z-score threshold from that baseline. This approach:

Requires no heavy dependencies (pure numpy/pandas/sklearn)

Handles daily and weekly seasonality

Is robust to missing data (median is robust to outliers)

Works equally well for 15-minute, hourly, and daily data

Anomaly types

spike: Single-timestep outlier: instantaneous spike (equipment fault, data error).
sustained_elevation: ≥3 consecutive readings above threshold: equipment left running, HVAC fault.
overnight: Anomaly during hours typically considered off (0:00–5:00). High risk for energy theft or after-hours equipment left on.
sudden_drop: Reading far below baseline: meter malfunction, curtailment event, or grid interruption.

Usage

>>> from energykit.anomaly import MeterAnomalyDetector
>>>
>>> detector = MeterAnomalyDetector(z_threshold=2.5)
>>> detector.fit(historical_series)          # learns seasonal baseline
>>> anomalies = detector.detect(series, energy_price=0.15)
>>>
>>> # DataFrame: is_anomaly, score, anomaly_type, excess_kwh, estimated_cost_usd
>>> wasteful = anomalies[anomalies.estimated_cost_usd > 5].sort_values(
...     "estimated_cost_usd", ascending=False
... )
>>> print(wasteful.head(10))

class energykit.anomaly.detector.AnomalySummary(n_anomalies: int, anomaly_rate_pct: float, total_excess_kwh: float, total_estimated_cost_usd: float, worst_event: Series | None, top_anomalies_df: DataFrame, anomalies_df: DataFrame)[source]

Bases: object

High-level summary of a MeterAnomalyDetector.detect() run.

n_anomalies

Type:: int

anomaly_rate_pct

Type:: float

total_excess_kwh

Sum of excess energy across all anomalies (above baseline).

Type:: float

total_estimated_cost_usd

Estimated dollar impact of all anomalies.

Type:: float

worst_event

Row from the anomaly DataFrame with the highest estimated_cost_usd.

Type:: pd.Series

top_anomalies_df

Top 10 most expensive anomaly events.

Type:: pd.DataFrame

anomalies_df

Full detection result.

Type:: pd.DataFrame

anomalies_df: DataFrame

anomaly_rate_pct: float

n_anomalies: int

top_anomalies_df: DataFrame

total_estimated_cost_usd: float

total_excess_kwh: float

worst_event: Series | None

class energykit.anomaly.detector.MeterAnomalyDetector(z_threshold: float = 2.5, overnight_hours: Tuple[int, int] = (0, 5), sustained_window: int = 3, use_isolation_forest: bool = False)[source]

Bases: object

Detect energy meter anomalies with financial impact estimation.

Parameters:

z_threshold (float, default 2.5) – Number of standard deviations above/below the seasonal baseline that classify a reading as an anomaly. Lower = more sensitive.
overnight_hours (tuple of int, default (0, 5)) – Half-open interval [start, stop) of hours considered “off-peak overnight”. Anomalies in this window get labelled overnight.
sustained_window (int, default 3) – Minimum number of consecutive anomalous readings to be labelled sustained_elevation (instead of individual spike).
use_isolation_forest (bool, default False) – If True and scikit-learn is available, adds a second layer of detection using IsolationForest on the residuals. Useful for multivariate or pattern-based anomalies.

Examples

>>> detector = MeterAnomalyDetector(z_threshold=2.5)
>>> detector.fit(training_series)      # learn seasonal baseline
>>> result = detector.detect(new_series, energy_price=0.15)
>>> print(result)
AnomalySummary(n=23, rate=0.26%, waste=312.4 kWh, cost=$46.86)

detect(series: Series, energy_price: float = 0.15) → AnomalySummary[source]

Detect anomalies in meter readings and estimate financial impact.

Parameters:

series (pd.Series) – Meter readings with a pd.DatetimeIndex.
energy_price (float, default 0.15) – Energy price in $/kWh. Used to estimate the cost of excess energy.

Return type:

AnomalySummary

fit(series: Series) → MeterAnomalyDetector[source]

Learn baseline patterns from historical meter data.

Parameters:: series (pd.Series) – Meter readings (kW or kWh) with a pd.DatetimeIndex. More historical data (≥60 days) produces more reliable baselines.
Return type:: self