USER
can you use a sigmoid function here to get the output in a signal 0,1
pip install keras_tuner
import pandas as pd #We use Pandas to preprocess the data
import numpy as np
import matplotlib.pyplot as plt #We used Matplotlib library to visualize the data. It helped us know the data and find important trends.
import keras #We use Keras to implement the machine learning models.
import math
import tensorflow as tf
from numpy import array
from tensorflow.keras.models import Sequential,Model,save_model
from tensorflow.keras.layers import LSTM, Dense, Dropout, Flatten, Input
from sklearn.preprocessing import StandardScaler
from sklearn.metrics import mean_squared_error, mean_absolute_error, r2_score
import seaborn as sns # Visualization
sns.set_style('white', { 'axes.spines.right': False, 'axes.spines.top': False})
import io
from google.colab import files
from keras_tuner import RandomSearch
from keras_tuner import Objective
#uploaded = files.upload()
#df = pd.read_csv(io.BytesIO(uploaded['VSIN1-a.csv']))
df = pd.read_excel("/content/BSAP1-a.xls").dropna()#.reset_index().drop(columns=['index'])
#df['Date'] = pd.to_datetime(df['Date'], format='%Y%m%d')
df['<DTYYYYMMDD>'] = pd.to_datetime(df['<DTYYYYMMDD>'], format='%Y%m%d')
#df=df[::-1].reset_index(drop=True)
df
#df = df[~(df == 0).any(axis=1)]
#df=df.reset_index(drop=True)
dataset = df[['<CLOSE>']].dropna().values
dataset = dataset.astype('float64')
'''
df['dif'] = (df['<HIGH>']-df['<LOW>']).dropna()
df['dif1'] = (df['<CLOSE>'].diff()).dropna()
dataset = df[['<CLOSE>','dif', "dif1"]].dropna().values
dataset = dataset.astype('float64')
'''
'''
df.drop(df.tail(1).index,inplace=True)
dataset = df[['<CLOSE>']].dropna().values
dataset = dataset.astype('float64')
'''
dataset
plt.plot(df['<DTYYYYMMDD>'],df['<CLOSE>'])
plt.grid()
plt.show()
# lookback -> timestep
def create_dataset(dataset,look_back):
data_x, data_y = [],[] #data_x is data and data_y is label
for i in range(len(dataset)-look_back): #we want if data be beyond len(sequendatasetce), the command will not continue
# اگر داده بیشتر از
#len(sequendatasetce)
#باشد، فرمان ادامه نمییابد.
data_x.append(dataset[i:(i+look_back),:])
data_y.append(dataset[i+look_back,:])
return np.array(data_x) , np.array(data_y)
train_size = int(len(dataset) * 0.80)
train , test = dataset[:train_size,:] , dataset[train_size:,:]
scaler = StandardScaler()
train = scaler.fit_transform(train)
test = scaler.fit_transform(test)
n_steps = 5 #timestep or look_up
train_x , train_y = create_dataset(train, n_steps)
test_x , test_y = create_dataset(test, n_steps)
print(train_x.shape , train_y.shape)
print(test_x.shape , test_y.shape)
trainxr = np.reshape(train_x,(train_x.shape[0],train_x.shape[1],1))
testxr = np.reshape(test_x,(test_x.shape[0],test_x.shape[1],1))
train_x = trainxr
test_x = testxr
print(trainxr.shape)
print(testxr.shape)
def R2_Score(y_true, y_pred):
SS_res = tf.reduce_sum(tf.square(y_true - y_pred))
SS_tot = tf.reduce_sum(tf.square(y_true - tf.reduce_mean(y_true)))
return 1 - SS_res/(SS_tot + tf.keras.backend.epsilon())
def rmse(y_true, y_pred):
return tf.sqrt(tf.reduce_mean(tf.square(y_pred - y_true)))
def build_model(hp):
model = Sequential()
model.add(LSTM(units=hp.Int('units1', min_value = 64, max_value = 512, step = 32), input_shape=(train_x.shape[1], train_x.shape[2]), return_sequences=True,
kernel_regularizer=keras.regularizers.l1_l2(l1=0.01, l2=0.01)))
model.add(Dropout(hp.Choice('Dropout', values = [0.2, 0.3, 0.4])))
model.add(LSTM(units=hp.Int('units2', min_value = 64, max_value = 256, step = 32)))
model.add(Dense(units=1, activation='linear'))
model.compile(loss = 'mse', optimizer = keras.optimizers.Adam(hp.Choice('learning_rate', values = [1e-2, 1e-3, 1e-4])), metrics=[rmse])
#early_stopping = tf.keras.callbacks.EarlyStopping(monitor='val_loss', patience=5)
#history = model.fit(train_x, train_y, validation_split=0.2, callbacks=[early_stopping], epochs=100)
return model
tuner = RandomSearch(
build_model,
objective = Objective('val_rmse', direction='min'),
max_trials=3,
executions_per_trial=2,
directory='project_dir',
project_name='test5')
tuner.search(train_x, train_y,
epochs=10,
validation_data=(test_x, test_y))
best_model = tuner.get_best_models(num_models=1)[0]
best_model.save('/content/savemodel')
# loading model
#new_model = tf.keras.saving.load_model('/content/savedata')
predict_train = best_model.predict(train_x)
predict_test = best_model.predict(test_x)
print('predicted y(train):', np.reshape(predict_train[:5],-1))
print('real y(train):', train_y[:5])
predict_train = scaler.inverse_transform(predict_train)
trainy = scaler.inverse_transform(train_y)
predict_test = scaler.inverse_transform(predict_test)
testy = scaler.inverse_transform(test_y)
Answer1 = pd.DataFrame({
"Predicted": predict_train.ravel(),
"real": trainy.ravel()
})
Answer1.head()
#train
Answer1.plot(title="outcome of training data", figsize=(20,10));
plt.grid()
<div dir=rtl>
<font face="XB Zar" size=5>
<hr />
نمایش ارور
</font>
</div>
error=Answer1['real']-Answer1['Predicted']
error.plot(title="error of training data")
plt.grid()
errorr=(Answer1['real']-Answer1['Predicted'])/Answer1['real']
errorr.plot(title="error of training data (%)")
plt.grid()
<div dir=rtl>
<font face="XB Zar" size=5>
<hr />
ساخت دیتافریم و رسم مقادیر آزمایشی واقعی و پیشبینی شده
</font>
</div>
Answer2 = pd.DataFrame({
"Predicted": predict_test.ravel(),
"real": testy.ravel()
})
Answer2.tail()
#test
Answer2.plot(title="outcome of test data", figsize=(20,10));
plt.grid()
error1=Answer2['real']-Answer2['Predicted']
error1.plot(title="error of test data")
plt.grid()
<div dir=rtl>
<font face="XB Zar" size=5>
<hr />
نمایش درصد خطای مدل
</font>
</div>
error2=(Answer2['real']-Answer2['Predicted'])/Answer2['real']
error2.plot(title="error of test data")
plt.grid()
score = best_model.evaluate(test_x, test_y, verbose = 0)
print('Test loss:', score[0])
predict_train
trainy
train_score = math.sqrt(mean_squared_error(trainy[:,0],predict_train[:,0]))#.reshape(-1)
print('RMSE of trian', train_score)
test_score = math.sqrt(mean_squared_error(test_y[:,0].reshape(-1),predict_test[:,0]))
print('RMSE of test', test_score)
RMSE of trian 943.3192753323997
RMSE of test 26628.738837928475
$(∑ y-x)/n$
train_score1 = math.sqrt(mean_absolute_error(trainy[:,0].reshape(-1),predict_train[:,0]))
print('RMAE of train', train_score1)
test_score1 = math.sqrt(mean_absolute_error(test_y[:,0].reshape(-1),predict_test[:,0]))
print('RMAE of test', test_score1)
RMAE of train 12.706159751886348
RMAE of test 6.442473942269808
print("Train data R2 score:", r2_score(trainy[:,0].reshape(-1), predict_train[:,0]))
print("Test data R2 score:", r2_score(test_y[:,0].reshape(-1), predict_test[:,0]))
price_today = testy[0][-1]
predicted_price = np.round(predict_test[-1][0], 2)
change_percent = np.round(100 - (price_today * 100)/predicted_price, 2)
end_date=df.iloc[-1]['<DTYYYYMMDD>']
print(f'The close price for Iran Khodro at {end_date} was {price_today}')
print(f'The predicted close price is {predicted_price} ({change_percent}%)')
<a name="lstmpred10"></a>
#### Predicting next 10 days
x_input=test[len(test)-n_steps:].reshape(1,-1)
temp_input=list(x_input)
temp_input=temp_input[0].tolist()
lst_output=[]
i=0
print('How many days ahead do you want to predict?')
pred_days = int(input())
while(i<pred_days):
if(len(temp_input)>n_steps):
x_input=np.array(temp_input[1:])
#print("{} day input {}".format(i,x_input))
x_input = x_input.reshape(1,-1)
x_input = x_input.reshape((1, n_steps, 1))
yhat = best_model.predict(x_input, verbose=0)
#print("{} day output {}".format(i,yhat))
temp_input.extend(yhat[0].tolist())
temp_input=temp_input[1:]
#print(temp_input)
lst_output.extend(yhat.tolist())
i=i+1
else:
x_input = x_input.reshape((1, n_steps,1))
yhat = best_model.predict(x_input, verbose=0)
temp_input.extend(yhat[0].tolist())
lst_output.extend(yhat.tolist())
i=i+1
print("Output of predicted next days: ", scaler.inverse_transform(np.array(lst_output[-1]).reshape(-1,1)).reshape(1,-1))
print("Output of predicted next days: ", scaler.inverse_transform(np.array(lst_output).reshape(-1,1)).reshape(1,-1))