I'm experimenting with 1D time-series data and trying to reproduce the following approach via animation over my own data in GoogleColab notebook.
It's about reproducing the animation of this approach: Backtesting with intermittent refit that has been introduced in skforecast package.
Backtesting with intermittent refit
The model is retrained every n iterations of prediction. This method is often used when the frequency of retraining and prediction is different. It can be implemented using either a fixed or rolling origin, providing flexibility in adapting the model to new data.
Following is my code:
import numpy as np
import matplotlib.pyplot as plt
from matplotlib.animation import FuncAnimation
from matplotlib.patches import Rectangle
import pandas as pd
from IPython.display import HTML
# create data
df = pd.DataFrame({
"TS_24hrs": np.arange(0, 274),
"count" : np.abs(np.sin(2 * np.pi * np.arange(0, 274) / 7) + np.random.normal(0, 100.1, size=274)) # generate sesonality
})
# Define the initial width for training and test data
TRAIN_WIDTH = 100
TEST_WIDTH = 1
# Define the delay for refitting the model
REFIT_DELAY = 10
# Define the delay for adding test data to train data
ADD_DELAY = 10
# create plot
plt.style.use("ggplot") # <-- set overall look
fig, ax = plt.subplots( figsize=(10,4))
# plot data
plt.plot(df['TS_24hrs'], df['count'], 'r-', linewidth=0.5, label='data or y')
# make graph beautiful
plt.plot([], [], 'g-', label="Train", linewidth=8, alpha=0.3) # <-- dummy legend entry
plt.plot([], [], 'b-', label="Test", linewidth=8, alpha=0.3) # <-- dummy legend entry
plt.xticks([0, 50, 100, 150, 200, 250, df['TS_24hrs'].iloc[-1]], visible=True, rotation="horizontal")
plt.title('Time-series backtesting with intermittent refit')
plt.ylabel('count', fontsize=15)
plt.xlabel('Timestamp [24hrs]', fontsize=15)
plt.grid(True)
plt.legend(loc="upper left")
fig.tight_layout(pad=1.2)
TRAIN_WIDTH = 25
TEST_WIDTH = 10
Y_LIM = 300 #ax.get_ylim()
def init():
rects = [Rectangle((0, 0), TRAIN_WIDTH, Y_LIM, alpha=0.3, facecolor='green'),
Rectangle((0 + TRAIN_WIDTH, 0), TEST_WIDTH, Y_LIM, alpha=0.3, facecolor='blue')]
patches = []
for rect in rects:
patches.append(ax.add_patch(rect))
return patches
# Initialize the start points for training and test data
train_data_start = 0
test_data_start = TRAIN_WIDTH
# Initialize the counter for refitting the model
refit_counter = REFIT_DELAY
# Initialize the counter for adding test data to train data
add_counter = ADD_DELAY
def update(x_start):
global train_data_start, test_data_start, refit_counter, add_counter, TRAIN_WIDTH
# Check if the model needs to be refitted
if refit_counter == REFIT_DELAY:
# Update the positions of train and test data with refit
patches[0].xy = (x_start + test_data_start - TRAIN_WIDTH , 0)
patches[1].xy = (x_start + test_data_start, 0)
# Reset the counter for refitting the model
refit_counter = 0
else:
# Update the positions of train and test data without refit
TRAIN_WIDTH += TEST_WIDTH # Increase the most data width
patches[0].set_width(TRAIN_WIDTH)
patches[0].xy = (x_start + test_data_start - TRAIN_WIDTH - 10 , 0)
patches[1].xy = (x_start + test_data_start, 0)
# Increase the counter for refitting the model
refit_counter += 1
# Check if the test data needs to be added to train data
if add_counter == ADD_DELAY:
# Move the training and test data one step forward
train_data_start += TEST_WIDTH # Add the width of the test to the widest
test_data_start += 1
# Reset the counter for adding test data to train data
add_counter = 0
else:
# Increase the counter for adding test data to train data
add_counter += 1
return patches
# Create "Train" and "Test" areas
patches = init()
ani = FuncAnimation(
fig,
update,
frames=np.arange(0, df.shape[0] - TRAIN_WIDTH - TEST_WIDTH), # All starting points
interval=70,
blit=True
)
HTML(ani.to_html5_video())
My current output is:
from matplotlib.animation import FuncAnimation, PillowWriter
ani.save("TLI.gif", dpi=100, writer=PillowWriter(fps=50))
Expected output:
You have several issues:
frames=np.arange
into FuncAnimation
update()
the logic is flaky.PLease inspect the changes I've introduced, it will help you to understand better further development direction.
# Define the delay for refitting the model
REFIT_DELAY = 5 # steps
TRAIN_WIDTH = 25
TEST_WIDTH = 10
Y_LIM = 300
def init():
rects = [Rectangle((0, 0), TRAIN_WIDTH, Y_LIM, alpha=0.3, facecolor='green'),
Rectangle((0 + TRAIN_WIDTH, 0), TEST_WIDTH, Y_LIM, alpha=0.3, facecolor='blue')]
patches = []
for rect in rects:
patches.append(ax.add_patch(rect))
return patches
# Initialize the start points for training and test data
train_data_start = 0
test_data_start = TRAIN_WIDTH
# Initialize the counter for refitting the model
refit_counter = 0
def update(x_start):
global test_data_start, refit_counter
patches[1].xy = (x_start + test_data_start, 0)
# Check if the model needs to be refitted
if refit_counter == REFIT_DELAY:
patches[0].set_width(x_start + test_data_start)
refit_counter = 0
# Increase the counter for refitting the model
refit_counter += 1
return patches
# Create "Train" and "Test" areas
patches = init()
ani = FuncAnimation(
fig,
update,
# frames=np.arange(0, df.shape[0] - TRAIN_WIDTH - TEST_WIDTH), # All starting points
frames=np.linspace(0, 250 - TEST_WIDTH, 15),
interval=300,
blit=True
)
HTML(ani.to_html5_video())