Win11 WSL2 Ubuntu Python FFT & IFFT
豪ドル/円４時間足大容量デ－タ FFT 解析予測ファイルまとめ
ここまで、豪ドル/円４時間足大容量デ－タ FFT 解析予測について小出しにその詳細を説明してきました。
引き続き、これを一つのファイル fft-go-long-4h.py にまとめます。
# -*- coding: utf-8 -*-
#!/usr/bin/python3
import os
os.environ['TF_CPP_MIN_LOG_LEVEL']='1'
import datetime
from datetime import datetime as dt
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(
"/home/yamada/public_html/manep-w/mane_chart_go_long_4h.csv",
encoding="cp932")
Open = xl_df["始値(売り)"].values
High = xl_df["高値(売り)"].values
Low = xl_df["安値(売り)"].values
Close = xl_df["終値(売り)"].values
Date = xl_df["日付"].values
Idx = xl_df.index
tstr = Date[-1]
raw_data0 = Close.copy()
raw_data = Close.copy()
Close1 = Close.copy()
tdatetime = dt.strptime(tstr, '%Y/%m/%d %H:%M:%S')
# 文字列をdatetimeに変換するのがstrptime()関数
# datetime.datetime.strptime(文字列, 書式指定文字列)
lastday = addBusinessDays_4h(tdatetime, 6)
Date_100 = Date[-100:]
xDate = []
xD = []
dayf = 0
for i, key in enumerate(Date_100):
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)
# 行列の平均、標準偏差を求めます。
mean = np.mean(raw_data)
print("Mean", mean)
raw_data -= mean
std = np.std(raw_data)
print("Std ",std)
# 標準偏差値に変換
# 約 5000個あり
raw_data /= std
print("各標準偏差値")
print(raw_data)
# for d_a_s in raw_data:
# print(d_a_s)
print(len(raw_data))
raw_data00 = raw_data.copy()
raw_data100l = raw_data00[-100:]
print("raw_data100l")
print(raw_data100l)
Idx0 = Idx.copy()
Idx_100 = Idx0[-100:] - Idx0[-100]
Idx_100 = Idx_100
print("Idx_100")
print(Idx_100)
# 仮の予測値を求める
z = np.polyfit(Idx_100, raw_data100l, 4) #4
p = np.poly1d(z)
Idx12 = Idx_100[-12:] + 12
print("仮の予測値12個 Idx12")
print(Idx12)
print("仮の予測値10個 p(Idx12)")
print(p(Idx12))
# 仮の予測値と大容量データ合体
raw_data_12 = np.hstack((raw_data, p(Idx12)))
print("仮の予測値と大容量データ合体")
print(raw_data_12)
fs = len(raw_data_12)
print("fs")
print(fs)
t = np.linspace(0, fs-1, fs)
print("t")
print(t)
plt.plot(t, raw_data_12)
plt.show()
# フーリエ変換(FFT)実行
f = np.fft.fft(raw_data_12)
print("f[5000]")
print(f[5000])
f_abs = np.abs(f)
plt.plot(f_abs[:int(fs/2)+1])
plt.show()
# ロ－パスフィルタ処理を行うため、配列 f[800]～f[fs-1] の値を０でクリア。
fc = 800 # cut off 800
f[fc:] = 0 # low pass filter
# ロ－パスフィルタ処理後のフーリエ変換(FFT)の結果も見ておきましょう。
f_abs = np.abs(f)
plt.plot(f_abs[:int(fs/2)+1])
plt.show()
# 逆フーリエ変換(IFFT)の実行
f_ifft = np.fft.ifft(f*2)
# 逆フーリエ変換(IFFT)
f_real = f_ifft.real
# 実数部のみにする
print("f_real")
print(f_real)
plt.plot(t, f_real)
plt.show()
f_real_112 = f_real[-112:]
t_112 = t[-112:] - t[-112]
print("f_real_112")
print(f_real_112)
print("t_112")
print(t_112)
t_106 = t_112[:106]
print("t_106")
print(t_106)
plt.plot(t_112, f_real_112)
plt.show()
# 標準偏差値を実値に変換
result_112 = f_real_112.copy()
result_112 *= std
result_112 += mean
print("result_112")
print(result_112)
result_106 = result_112[:106]
print("result_106")
print(result_106)
Open_100 = Open[-100:]
High_100 = High[-100:]
Low_100 = Low[-100:]
Close_100 = Close1[-100:]
Idx_100 = Idx0[-100:] - Idx0[-100]
plt.plot(Idx_100, Close_100)
plt.plot(t_106, result_106)
plt.show()
ohlc = zip(
Idx_100, Open_100, High_100, Low_100, Close_100)
fig = plt.figure(
figsize=(8.34, 5.56))
# python スクリプトと Jupyter とでは、
# matplotlib の図のサイズが違うので注意
ax = fig.add_subplot(1,1,1)
ax.grid()
# 解析結果表示はここに挿入
plt.plot(t_106, result_106)
# 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.savefig(
# '/home/yamada/public_html/manep-img/fft_go_long_4h.png')
plt.show()
print("予測値 = 最終日時 +1～+6本")
valhe = np.round(result_106[100:], 3) # 3桁まで表示
valhe_pd = pd.DataFrame(valhe)
lastday_pd = pd.DataFrame(lastday[0])
# この２つを pandas concat 関数で横(列)方向へ連結します。
# 横(列)方向指定 axis = 1 を忘れないでください。
df_concat = pd.concat([lastday_pd, valhe_pd], axis = 1)
print(df_concat)
df_concat.to_csv(
'/home/yamada/public_html/manep-img/fft_go_long_4h.csv',
header=False, index=False)
ここまでで、豪ドル/円４時間足大容量デ－タ FFT 解析予測ファイル fft-go-long-4h.py がまとまりました。
引き続き、このファイルの動作確認をします。