The scikit-multiflow package can detect data drift using an algorithm known as adaptive windowing (ADWIN) that detects data drift over a stream of data.

import numpy as np from skmultiflow.drift_detection.adwin import ADWIN adwin = ADWIN() # Simulating a data stream as a normal distribution of 1's and 0's data_stream = np.random.randint(2, size=2000) # Artificially shift the data from index 999 to 2000 # by replacing the i value with a greater one for i in range(999, 2000): data_stream[i] = np.random.randint(5, high=10) previous_variance = 0 # Add the stream elements to ADWIN and check if drift has been detected for i in range(2000): adwin.add_element(data_stream[i]) if adwin.detected_change(): print("Change detected in value {}, at index {}".format(data_stream[i], i)) print("Current variance: {}. Previous variance {}".format(adwin.variance, previous_variance)) previous_variance = adwin.variance