Your question is Real-Time Anomaly Detection. Start with the requirements on the right.
Run and submit as often as you like. When you're ready, talk me through your approach or go straight to the code.
Mi Home telemetry arrives as a timestamp-ordered stream of sensor events. For each device, detect an event as anomalous when its value differs from the rolling median of the device's previous accepted values by more than a configured threshold.
An anomalous event must not be added to the device's baseline, because a spike should not influence detection of later events. Each device maintains at most the most recent window_size accepted values. Events from different devices have independent baselines.
Implement detect_anomalies(events, window_size, threshold, min_history). events is a list of dictionaries with integer timestamp, string device_id, and numeric value fields, ordered by nondecreasing timestamp. Return a list of integer indices for anomalous events, in stream order.
An event is anomalous only when the device has at least min_history accepted values and:
abs(value - rolling_median) > threshold
For an even-sized window, the rolling median is the average of the two middle values. Events with insufficient history are accepted without detection and become part of the baseline.
def detect_anomalies(events, window_size, threshold, min_history):