Блог

У меня есть временной ряд со столбцами date и price. Я использую LSTM, и когда я тестирую данные, они хорошо соответствуют графику.

Теперь я хочу предсказать следующие переходящие X дней с сегодняшнего дня, когда данные не существуют. Я не могу найти способ сделать это. Каждый пример, на который я смотрю, примеры показывают только обратное тестирование на исторических данных.

Нужно ли мне все еще разделять данные на тест и обучение? Я не уверен, что я вкладываю в model.predict(?) это, чтобы дать мне прогноз на будущее.

 import pandas as pd
from datetime import date
import matplotlib.pyplot as plt
from sklearn.preprocessing import MinMaxScaler
import numpy as np
from keras.models import Sequential
from keras.layers import LSTM,Dropout,Dense
import datetime
import math
from datetime import datetime, timedelta
from sklearn import model_selection
from sklearn.metrics import mean_squared_error
plt.style.use('fivethirtyeight')

def create_dataset(dataset, look_back=1):
    dataX, dataY = [], []
    for i in range(len(dataset)-look_back-1):
        a = dataset[i:(i look_back), 0]
        dataX.append(a)
        dataY.append(dataset[i   look_back, 0])
    return np.array(dataX), np.array(dataY)

def createForecastDF(num_days,COLUMN_NAME):
    datelist = pd.date_range(datetime.today(), periods=num_days).date.tolist()
    datelist = [date_obj.strftime('%d/%m/%Y') for date_obj in datelist]
    for i in range (0,len(datelist)):
        datelist[i] = [datelist[i],1]
    forecast_df = pd.DataFrame(datelist, columns = ['Date',COLUMN_NAME])
    forecast_df = forecast_df.set_index('Date')
    return forecast_df

def back_test(data,look_back):

    # Apply scaler
    scaler = MinMaxScaler(feature_range = (0,1))
    dataset = scaler.fit_transform(data)

    # Split Data into train and test
    training_set, testing_set = model_selection.train_test_split(dataset, test_size=0.33, shuffle=False)

    # reshape into X=t and Y=t 1
    trainX, trainY = create_dataset(training_set, look_back)
    testX, testY = create_dataset(testing_set, look_back)
    
    # reshape input to be [samples, time steps, features]
    trainX = np.reshape(trainX, (trainX.shape[0], 1, trainX.shape[1]))
    testX = np.reshape(testX, (testX.shape[0], 1, testX.shape[1]))

    model = Sequential() h
    model.add(LSTM(4, input_shape=(1, look_back)))
    model.add(Dense(1))
    model.compile(loss='mean_squared_error', optimizer='adam')
    model.fit(trainX, trainY, epochs=100, batch_size=1, verbose=2)
    
    trainPredict = model.predict(trainX)
    testPredict = model.predict(testX)
    
    # invert predictions
    trainPredict = scaler.inverse_transform(trainPredict)
    trainY = scaler.inverse_transform([trainY])
    testPredict = scaler.inverse_transform(testPredict)
    testY = scaler.inverse_transform([testY])
    
    # calculate root mean squared error
    trainScore = math.sqrt(mean_squared_error(trainY[0], trainPredict[:,0]))
    print('Train Score: %.2f RMSE' % (trainScore))
    testScore = math.sqrt(mean_squared_error(testY[0], testPredict[:,0]))
    print('Test Score: %.2f RMSE' % (testScore))
    
    trainPredictPlot = np.empty_like(dataset)
    trainPredictPlot[:, :] = np.nan
    trainPredictPlot[look_back:len(trainPredict) look_back, :] = trainPredict

    # shift test predictions for plotting
    testPredictPlot = np.empty_like(dataset)
    testPredictPlot[:, :] = np.nan
    testPredictPlot[len(trainPredict) (look_back*2) 1:len(dataset)-1, :] = testPredict

    # plot baseline and predictions
    plt.plot(scaler.inverse_transform(dataset))
    plt.plot(trainPredictPlot)
    plt.plot(testPredictPlot)
    plt.title("Training Dataset")
    plt.show()


def forecast(data,look_back,num_historial_data,df_future):
    
    final_data = data[:num_historial_data]
    
    # Apply scaler
    scaler = MinMaxScaler(feature_range = (0,1))
    dataset = scaler.fit_transform(final_data)

    # reshape into X=t and Y=t 1
    trainX, trainY = create_dataset(dataset, look_back)

    # reshape input to be [samples, time steps, features]
    trainX = np.reshape(trainX, (trainX.shape[0], 1, trainX.shape[1]))

    model = Sequential()
    model.add(LSTM(4, input_shape=(1, look_back)))
    model.add(Dense(1))
    model.compile(loss='mean_squared_error', optimizer='adam')
    model.fit(trainX, trainY, epochs=100, batch_size=1, verbose=2)
    
    trainPredict = model.predict(trainX)
    
    # invert predictions
    trainPredict = scaler.inverse_transform(trainPredict)
    trainY = scaler.inverse_transform([trainY])




COLUMN_NAME = "Close"

# Make sure your data frame has only one column beside index [Date, col]
historical_df = pd.read_csv('dataset.csv', index_col='Date')
historical_df = historical_df[[COLUMN_NAME]]

# Variables
num_historial_data = len(historical_df)
forecast_length = 26
today = date.today()

# 1. Creating future dates df
df_future = createForecastDF(100,COLUMN_NAME)

# 2. Merge dataframes together
dataframe = historical_df.append(df_future)

# 3.Back Test Data
back_test(dataframe[:num_historial_data], forecast_length)

# 4. Forecast Data
#forecast(dataframe, forecast_length,num_historial_data,df_future)

 

1. У вас есть данные: серия [дни, цены]. Разделите его для обучения, тестирования, проверки. (современное состояние). Вы должны обучить его так, использовать:

 history = model.fit(X_train, y_train,
                    validation_data=(X_valid, y_valid),
                    epochs=100, 
                    callbacks=tf.keras.callbacks.EarlyStopping(patience=5))
model.evaluate(X_valid, y_valid) 

2. Идея состоит в том, чтобы использовать последние значения вашего ряда для составления прогнозов:

 #n_steps length of your time series/day

X_new, Y_new = series[:, :n_steps], series[:, n_steps:]
X = X_new

for step_ahead in range(10):
    y_pred_one = model.predict(X[:, step_ahead:])[:, np.newaxis,:]
    X = np.concatenate([X,y_pred_one], axis=1)
    
Y_pred = X[:, n_steps:] 
# the next 10 values predicted, using the last val. predicted as input