さくら VPS CNN 豪ドル/円予測
さくら VPS CNN 豪ドル/円４時間足予測ファイル
さくら VPS CNN 豪ドル/円４時間予測ファイル keras-go-cnn-4h.py
の内容を表示します。
colab の中で動作する
keras-go-cnn-4h.ipynb
との差分は赤字にしています。
# -*- coding: utf-8 -*- #!/usr/bin/python3 #import sys #sys.path.append( # '/content/drive/MyDrive/Colab Notebooks/my-modules' #) import os os.environ['TF_CPP_MIN_LOG_LEVEL']='1' import datetime import numpy as np import pandas as pd from matplotlib import pyplot as plt from tensorflow import keras from mplfinance.original_flavor import candlestick_ohlc def addBusinessDays_4h(current_date, business_days_to_add_h): # current_date : 処理する初日の日時 例) 2022-02-03 00:00:00 # business_days_to_add_h : 4時間ずつ延長する回数 cur_date8 = [] # 21/03/30 12～ cur_date4 = [] # 3/30 man_f = 1 #月曜フラグ 0 or 1 doyo_f = 0 #土曜フラグ 0 or 1 current_date0 = current_date weekday0 = current_date0.weekday() if(weekday0 == 0): #月 man_f = 0 while business_days_to_add_h > 0: current_date += datetime.timedelta(hours=4) weekday = current_date.weekday() hour_doyo = current_date.hour if weekday == 6: # sunday = 6 continue if((weekday == 5) and (hour_doyo == 8)): #土 doyo_f = 1 if((weekday == 5) and (doyo_f == 1)): #土 continue if((weekday == 0) and (man_f == 1)): #月 man_f = 0 continue # 年/月/日 時刻として文字列にする処理 # 例 22/03/30 12～ e11 = current_date.strftime("%y/%m/%d %H") e22 = e11.replace("/0", "/") e22 = e22 + "～" e33 = e22[3:] # 22/ 削除 cur_date8.append(e22) cur_date4.append(e33) business_days_to_add_h -= 1 return cur_date8, cur_date4 xl_df = pd.read_csv( # "drive/MyDrive/Colab Notebooks/my_data/colab_mane_chart_go_4h.csv", "/home/yamada/public_html/manep_4h/mane_chart_go_4h.csv", encoding="cp932") raw_data0 = xl_df["終値(売り)"].values # print("raw_data0.shape:", raw_data0.shape) # print("raw_data0") # print(raw_data0) # plt.plot(range(len(raw_data0)), raw_data0) # plt.show() raw_data = raw_data0.copy() # 行列の平均、標準偏差を求めます。 mean = np.mean(raw_data) # print("Mean", mean) raw_data -= mean std = np.std(raw_data) # print("Std ",std) # 標準偏差値に変換 raw_data /= std # print("各標準偏差値") # print(raw_data) # 連続デ－タとする。一つおきは、2 sampling_rate = 1 # 過去20間隔デ－タをひとまとまりとして時系列予測する sequence_length = 20 delay = sampling_rate * sequence_length # print("delay:", delay) batch_size = 32 # 適当 # 検証デ－タのスタ－ト値 num_half_samples = int(0.5 * len(raw_data)) train_dataset = keras.utils.timeseries_dataset_from_array( raw_data, targets=raw_data[delay:], sampling_rate=sampling_rate, sequence_length=sequence_length, batch_size=batch_size, ) val_dataset = keras.utils.timeseries_dataset_from_array( raw_data[:-1], targets=raw_data[delay:], sampling_rate=sampling_rate, sequence_length=sequence_length, batch_size=batch_size, start_index=num_half_samples, ) test_dataset = keras.utils.timeseries_dataset_from_array( raw_data, targets=None, sampling_rate=sampling_rate, sequence_length=sequence_length, batch_size=batch_size, ) # numpy ndarray 配列に変換して表示 # 訓練デ－タセット表示 """ itr = 0 for samples, targets in train_dataset: samples_n = samples.numpy() targets_n = targets.numpy() if itr == 0: print("Start in-train:", samples_n[0]) print("Start tar-train:", targets_n[0]) itr = itr + 1 print("End in-train:", samples_n[-1]) print("End tar-train:", targets_n[-1]) # 検証デ－タセット表示 itv = 0 for samples_v, targets_v in val_dataset: samples_vn = samples_v.numpy() targets_vn = targets_v.numpy() if itv == 0: print("Start in-val:", samples_vn[0]) print("Start tar-val:", targets_vn[0]) itv = itv + 1 print("End in-val:", samples_vn[-1]) print("End tar-val:", targets_vn[-1]) """ # テストデ－タセット表示 i = 0 for inputs_t in test_dataset: inputs_n = inputs_t.numpy() if i == 0: # print("Start test:", inputs_n[0]) i = i + 1 # print("End test:", inputs_n[-1]) from keras import layers # from keras import initializers inputs = keras.Input(shape=(sequence_length, 1)) # x = layers.LSTM(20, kernel_initializer='zeros')(inputs) # 3行追加 x = layers.Conv1D(7, 15, activation="tanh")(inputs) x = layers.MaxPooling1D(2)(x) x = layers.GlobalAveragePooling1D()(x) # カ－ネルサイズは 15 に微調整 outputs = layers.Dense(1)(x) model = keras.Model(inputs, outputs) # print("モデルア－キテクチャ") # print(model.summary()) #---ここまで keras-test-52.ipynb callbacks_list = [ keras.callbacks.EarlyStopping( monitor="val_loss", patience=6, ), keras.callbacks.ModelCheckpoint( # "drive/MyDrive/Colab Notebooks/my_data/jena_dense_cnn_4h.keras", "/home/yamada/public_html/colab/jena_dense_go_cnn_4h.keras", save_best_only=True, ) ] model.compile(optimizer="rmsprop", loss="mse", metrics=["mae"]) history = model.fit( train_dataset, epochs=40, # verbose=0, verbose=0, validation_data=val_dataset, callbacks=callbacks_list) loss = history.history["mae"] # 平均絶対誤差(MAE) val_loss = history.history["val_mae"] epochs = range(1, len(loss) + 1) """ plt.plot(epochs, loss, "bo", label="Training MAE") plt.plot(epochs, val_loss, "b", label="Validation MAE") plt.title("Training and Validation MAE") plt.legend() plt.show() """ #---ここまで keras-test-53.ipynb model = keras.models.load_model( # "drive/MyDrive/Colab Notebooks/my_data/jena_dense_cnn_4h.keras") "/home/yamada/public_html/colab/jena_dense_go_cnn_4h.keras") # pre = model.predict(test_dataset) pre = model.predict(test_dataset, verbose=0) pre1 = np.reshape(pre, (-1)) # print(f"Test 予測値:") # print(pre1) future_test = inputs_t[-1:] # print("future_test 最初の配列値") # print(future_test) future_result = [] for i in range(6): test_data_f = np.reshape(future_test, (1, 20, 1)) # batch_predict = model.predict(test_data_f) batch_predict = model.predict(test_data_f, verbose=0) future_test = np.delete(future_test, 0) future_test = np.append(future_test, batch_predict) future_result = np.append(future_result, batch_predict) # print("future_result :") # print(future_result) # ここまで colab-21 まとめ len_raw_data = len(raw_data) xx1 = np.arange(sequence_length, len_raw_data + 1) xx3 = np.arange(len_raw_data, len_raw_data + 6) # plt.plot(xx1, pre) # plt.plot(xx3, future_result) # plt.show() #ここまで colab-22①まとめ pre_chg = pre.copy() pre_chg *= std pre_chg += mean pre_chg1 = np.reshape(pre_chg, (-1)) # print("pre_chg1:") # print(pre_chg1) f_result = future_result.copy() f_result *= std f_result += mean """ print("f_result:") print(f_result) plt.plot(range(len(raw_data0)), raw_data0) plt.plot(xx1, pre_chg) plt.plot(xx3, f_result) plt.show() """ #ここまで colab-22②まとめ # ここから４時間ロ－ソク足表示 Open = xl_df["始値(売り)"].values High = xl_df["高値(売り)"].values Low = xl_df["安値(売り)"].values Close = xl_df["終値(売り)"].values Date = xl_df["日付"].values Idx = xl_df.index # X 軸日付処理 xDate = [] xD = [] dayf = 0 for i, key in enumerate(Date): day00 = str(key)[11:19] if((dayf == 1) and (day00 == '00:00:00')): dayf = 0 else: if((dayf == 0) and (day00 == '00:00:00')): e4 = str(key)[4:10] e6 = e4.replace("/0", "/") e8 = e6.lstrip("/") xDate.append(e8) xD.append(i) dayf = 1 tstr = Date[-1] if(dayf == 0): e4l = tstr[7:14] e6l = e4l.replace("/0", "/") e8l = e6l.lstrip("/") xDate.append(e8l) xD.append(i) # ロ－ソク４時間足表示 ohlc = zip( Idx, Open, High, Low, Close) fig = plt.figure( figsize=(8.34, 5.56)) ax = fig.add_subplot(1,1,1) #④ ax.grid() plt.plot(xx1, pre_chg) plt.plot(xx3, f_result, 'bo') candlestick_ohlc( ax, ohlc, width=0.5, alpha = 1, colorup='r', colordown='g') plt.xticks(xD, xDate) plt.title('AUS$ / JPY chart') plt.xlabel('Date') plt.ylabel('Yen') plt.show plt.savefig( # 'colab_mane_chart_go_cnn_4h.png') '/home/yamada/public_html/manep-img/colab_mane_chart_go_cnn_4h.png') # ここから４時間足予測最終日時取得 tstr = Date[-1] tdatetime = datetime.datetime.strptime(tstr, '%Y/%m/%d %H:%M:%S') lastday = addBusinessDays_4h(tdatetime, 6) lastday_pd = pd.DataFrame(lastday[0]) f_result_pd = pd.DataFrame(np.round(f_result, 3)) # 有効数字3桁に変更 df_concat = pd.concat([lastday_pd, f_result_pd], axis = 1) df_concat.to_csv( # 'colab_mane_chart_go_cnn_4h.csv', '/home/yamada/public_html/manep-img/colab_mane_chart_go_cnn_4h.csv', header=False, index=False)
ここまでさくら VPS CNN 豪ドル/円４時間足予測ファイル keras-go-cnn-4h.py がまとまりました。
引き続き本ファイルの動作確認をしていきます。