PELT Changepoints Detection

changepoints_detection.py

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from ruptures import Pelt
from scipy.stats import ttest_ind

# Generate synthetic data with clear changepoints
np.random.seed(42)
timestamps = pd.date_range('2023-10-26', periods=60, freq='D')
latency = np.concatenate([
    np.random.normal(100, 5, 20),
    np.random.normal(150, 5, 20),
    np.random.normal(200, 5, 20)
])

data = {
    'timestamp': timestamps,
    'latency': latency
}
df = pd.DataFrame(data)

# Smooth latency using a 3-day moving average
df['smoothed_latency'] = df['latency'].rolling(window=3).mean()
df['smoothed_latency'].fillna(method='bfill', inplace=True)

# Changepoint detection using Pelt algorithm
model = Pelt(model="rbf").fit(np.array(df['smoothed_latency']).reshape(-1, 1))
changepoints = model.predict(pen=5)

# Ensure changepoints are within the range of indices
changepoints = [cp for cp in changepoints if cp < len(df)]

# Plot original and smoothed latency with changepoints
plt.figure(figsize=(12, 6))
plt.plot(df['timestamp'], df['latency'], label='Original Latency', marker='o')
plt.plot(df['timestamp'], df['smoothed_latency'], label='Smoothed Latency', marker='o')
for cp in changepoints:
    plt.axvline(x=df['timestamp'].iloc[cp], color='r', linestyle='--', label='Changepoint' if cp == changepoints[0] else "")
plt.xlabel('Timestamp')
plt.ylabel('Latency (ms)')
plt.title('API Latency with Changepoints')
plt.legend()
plt.grid(True)
plt.show()

print("Changepoints:", changepoints)