Your question is Real-Time Anomaly Detection Function. 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.
Cambridge Mobile Telematics DriveWell receives time-ordered raw sensor readings from many vehicles. Implement a streaming anomaly detector that identifies readings which differ significantly from the recent behavior of the same sensor while avoiding a full data copy or repeated sorting.
Each reading is represented as [timestamp, sensor_id, value]. For every sensor, compare the current value with the preceding window_size accepted readings for that sensor. Use the rolling median as the center and the population standard deviation as the scale. A reading is anomalous when abs(value - median) > threshold * max(stddev, 1e-12). Readings with fewer than two prior values for their sensor are never anomalous. The current reading must be added to its sensor window whether or not it is anomalous.
Return the zero-based indices of anomalous readings in their original input order. Timestamps are already ordered globally and do not affect the calculation.
def detect_anomalies(readings, window_size, threshold):