如何通过一段时间的间歇性改装来重现回测动画？

Question

我正在试验一维时间序列数据，并尝试通过动画在 GoogleColab 笔记本中我自己的数据上重现以下方法。

这是关于重现此方法的动画：间歇性改装的回测已在skforecast包中引入。

间歇性改装回测

每 n 次预测迭代都会重新训练模型。当再训练和预测的频率不同时，经常使用这种方法。它可以使用固定或滚动原点来实现，从而提供使模型适应新数据的灵活性。

以下是我的代码：

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

我当前的输出是：

from matplotlib.animation import FuncAnimation, PillowWriter
ani.save("TLI.gif", dpi=100, writer=PillowWriter(fps=50))

预期输出：

Answer 1

您有几个问题：

1. 测试补丁正在滑动，因为您将

   frames=np.arange

   传递到

   FuncAnimation

2. 在

   update()

   中逻辑很不稳定。

请检查我介绍的更改，它将帮助您更好地了解进一步的发展方向。

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