このプログラムの目的
株価のテクニカル指標を算出し，データセットとして保存することです．
work_share
├04_get_stock_price_ver2
├Dockerfile
├docker-compose.yml
└src
├dataset
├original_data_2010-01-01_2023-03-01_1d
├time_cluster_result
├get_stock_price.py
├make_original_data.py
├make_time_cluster_dataset.py
├calculate_nearest_codes.py
├make_technical_data.py (これを作成)
└stocks_code.xls使用ライブラリ
import pandas as pd
import os
import talibデータセットのロード
データセットを読み込みます．
def df_split(df):
train_df = df.loc[df['eval_type']=='train', :].reset_index(drop=True)
test_df = df.loc[df['eval_type']=='test', :].reset_index(drop=True)
return train_df, test_df
def load_dataset(args):
data = {
'close':pd.read_pickle(args['close_dataset_path']),
'high':pd.read_pickle(args['open_dataset_path']),
'low':pd.read_pickle(args['low_dataset_path']),
'high':pd.read_pickle(args['high_dataset_path'])
}
for key in data.keys():
data[key] = data[key].fillna(method='ffill')
return dataテクニカル指標の保存
numpyを使ってテクニカル指標を計算しても良いのですが，今回はtalibというライブラリを利用してテクニカル指標を計算します．本章で利用した指標以外にも多くのテクニカル指標が計算可能ですので，必要だと思う指標を利用してください．
各テクニカル指標にはパラメータがありますが，本章では適当な値を指定しました．好みがあれば，自分でカスタマイズしてください．
移動平均
def make_moving_average_data(data, args):
def make_ma(df, timeperiod):
return pd.DataFrame({
f'{col}':talib.MA(df[col], timeperiod=timeperiod)
for col in df.columns if col != 'eval_type'
})
make_ma(data['close'], 5).to_pickle(f'{args["save_dir"]}/ma_5.dfpkl')
make_ma(data['close'], 25).to_pickle(f'{args["save_dir"]}/ma_25.dfpkl')
make_ma(data['close'], 75).to_pickle(f'{args["save_dir"]}/ma_75.dfpkl')指数平滑移動平均
def make_ema_data(data, args):
def make_ema(df, timeperiod):
return pd.DataFrame({
f'{col}':talib.EMA(df[col], timeperiod=timeperiod)
for col in df.columns if col != 'eval_type'
})
make_ema(data['close'], 5).to_pickle(f'{args["save_dir"]}/ema_5.dfpkl')
make_ema(data['close'], 25).to_pickle(f'{args["save_dir"]}/ema_25.dfpkl')
make_ema(data['close'], 75).to_pickle(f'{args["save_dir"]}/ema_75.dfpkl')ボリンジャーバンド
def make_bbands_data(data, args):
uppers = pd.DataFrame()
middles = pd.DataFrame()
lowers = pd.DataFrame()
for col in data['close'].columns:
if col != 'eval_type':
upper, middle, lower = talib.BBANDS(data['close'][col], timeperiod=25)
uppers[col] = upper
middles[col] = middle
lowers[col] = lower
uppers.to_pickle(f'{args["save_dir"]}/bbands_upper.dfpkl')
middles.to_pickle(f'{args["save_dir"]}/bbands_middle.dfpkl')
lowers.to_pickle(f'{args["save_dir"]}/bbands_lower.dfpkl')RSI
def make_rsi_data(data, args):
def make_rsi(df, timeperiod):
return pd.DataFrame({
f'{col}':talib.RSI(df[col], timeperiod=timeperiod)
for col in df.columns if col != 'eval_type'
})
make_rsi(data['close'], 14).to_pickle(f'{args["save_dir"]}/rsi_14.dfpkl')ストキャスティクス
def make_stoch_data(data, args):
slowk_df = pd.DataFrame()
slowd_df = pd.DataFrame()
for col in data['close'].columns:
if col != 'eval_type':
slowk_df[col], slowd_df[col] = talib.STOCH(data['high'][col], data['low'][col], data['close'][col], fastk_period=5, slowk_period=3, slowd_period=3)
slowk_df.to_pickle(f'{args["save_dir"]}/stoch_slowk.dfpkl')
slowd_df.to_pickle(f'{args["save_dir"]}/stoch_slowd.dfpkl')DMI
def make_dmi_data(data, args):
plus_df = pd.DataFrame()
minus_df = pd.DataFrame()
adx_df = pd.DataFrame()
for col in data['close'].columns:
if col != 'eval_type':
plus_df[col] = talib.PLUS_DI(data['high'][col], data['low'][col], data['close'][col], timeperiod=14)
minus_df[col] = talib.MINUS_DI(data['high'][col], data['low'][col], data['close'][col], timeperiod=14)
adx_df[col] = talib.ADX(data['high'][col], data['low'][col], data['close'][col], timeperiod=14)
plus_df.to_pickle(f'{args["save_dir"]}/dim_plus.dfpkl')
minus_df.to_pickle(f'{args["save_dir"]}/dim_minus.dfpkl')
adx_df.to_pickle(f'{args["save_dir"]}/dim_adx.dfpkl')MACD
def make_macd_data(data, args):
macd_df = pd.DataFrame()
signal_df = pd.DataFrame()
hist_df = pd.DataFrame()
for col in data['close'].columns:
if col != 'eval_type':
macd_df[col], signal_df[col], hist_df[col] = talib.MACD(data['close'][col])
macd_df.to_pickle(f'{args["save_dir"]}/macd_main.dfpkl')
signal_df.to_pickle(f'{args["save_dir"]}/macd_signal.dfpkl')
hist_df.to_pickle(f'{args["save_dir"]}/macd_hist.dfpkl')実行
def run(args):
data = load_dataset(args)
make_moving_average_data(data, args)
make_ema_data(data, args)
make_bbands_data(data, args)
make_rsi_data(data, args)
make_stoch_data(data, args)
make_dmi_data(data, args)
make_macd_data(data, args)
def experiment():
dataset_dir = 'dataset'
technical_data_save_dir = f'./{dataset_dir}/technical_data'
os.makedirs(technical_data_save_dir, exist_ok=True)
args = {
'close_dataset_path':f'./{dataset_dir}/original_dataset_Close.dfpkl',
'open_dataset_path':f'./{dataset_dir}/original_dataset_Open.dfpkl',
'low_dataset_path':f'./{dataset_dir}/original_dataset_Low.dfpkl',
'high_dataset_path':f'./{dataset_dir}/original_dataset_High.dfpkl',
'save_ret_path':f'./{dataset_dir}/nearest_calculate_result.json',
'save_dir':technical_data_save_dir
}
run(args)
if __name__ == '__main__':
experiment()実行エラー
筆者の環境ではTalibを実行しようとするとエラーが発生しました．
ImportError: numpy.core.multiarray failed to importこれはnumpyのアップグレードで回避できます．
pip install numpy --upgradeしかし，numpyをアップグレードすると，この節のプログラムは動きますが，No.03-03 時系列クラスタリングのtslearnが動かなくなります．
なので，numpyをアップグレードした後にNo.04-03を実行したい場合はDockerコンテナを終了し，再度buildして環境をリセットする必要があります．
