matplotlib 我怎样才能重现随着时间的推移间歇改装回测的动画?

hpxqektj  于 2023-10-24  发布在  其他
关注(0)|答案(1)|浏览(116)

我正在用1D时间序列数据进行实验,并试图通过GoogleColab notebook中我自己的数据的动画来重现以下方法。
这是关于重现这种方法的动画:* 间歇性改装的回测 * 已经在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))

预期输出:

zf2sa74q

zf2sa74q1#

您有几个问题:
1.测试补丁正在滑动,因为您将frames=np.arange传递到FuncAnimation
1.在update()中,逻辑是x1。
请检查我所介绍的变化,它将帮助您更好地了解进一步的发展方向。

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

相关问题