pythonmatplotlibtime-seriessliding-windowmatplotlib-animation

How can I reproduce animation of Backtesting with intermittent refit over time?


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 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))

img

Expected output: gif


Solution

  • You have several issues:

    1. Test patch is sliding bc you pass frames=np.arange into FuncAnimation
    2. In 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())