深層学習用のインジケータ作成器を作ってみました。

先日作成したプログラムに食わせるインジケータを作成したので。

python 3.5.6
Platform: Windows-10-10.0.17134-SP0
Chainer: 5.0.0
NumPy: 1.15.2
CuPy: Not Available
iDeep: Not Available

ta-lib

使用に際して食わせるデータは以下。

ndarray,float32

close_float->終値
high_float->高値
low_float->安値

length->ローソク足の使用する本数

例)200期間のsmaを使う

sma200 = util.SMA(200,close_float)

とやって終値を渡してやると、以下の処理を済ませた値が返ってきます。

200期間の移動平均を算出
↓
冒頭部分に作られる[nan]が返された部分を削除
↓
残った部分の数値を使って正規化
↓
削除されて寸足らずになった部分をゼロ埋めしてclose_floatと同じ長さにする
↓
正規化されてスキマをゼロ埋めして200期間になったsmaを返す

しかしこれ…全部正規化する必要あるのかな…？

util.py

import numpy as np
import talib
from sklearn import preprocessing

#-------------各種インジケーター-------------

def ADX(length,high_float, low_float, close_float):
    #ADX - Average Directional Movement Index
    h_f = high_float.copy()
    l_f = low_float.copy()
    c_f = close_float.copy()
    cache = np.c_[c_f, talib.ADX(h_f, l_f, c_f,timeperiod = length)]

    adx = []

    for i in cache:
        if np.isnan(i[1]) != True:
            adx.append(i[1])

    adx = np.array(adx,dtype='float32')
    adx = preprocessing.minmax_scale(adx)

    while len(adx) < len(close_float):
        adx = np.insert(adx,0,0)
    return adx

def PDI(length,high_float, low_float, close_float):
    #PLUS_DI - Plus Directional Indicator
    h_f = high_float.copy()
    l_f = low_float.copy()
    c_f = close_float.copy()
    cache = np.c_[c_f, talib.PLUS_DI(h_f, l_f, c_f,timeperiod = length)]

    pdi = []

    for i in cache:
        if np.isnan(i[1]) != True:
            pdi.append(i[1])

    pdi = np.array(pdi,dtype='float32')
    pdi = preprocessing.minmax_scale(pdi)

    while len(pdi) < len(close_float):
        pdi = np.insert(pdi,0,0)
    return pdi

def MDI(length,high_float, low_float, close_float):
    #MINUS_DI - Minus Directional Indicator
    h_f = high_float.copy()
    l_f = low_float.copy()
    c_f = close_float.copy()
    cache = np.c_[c_f, talib.MINUS_DI(h_f, l_f, c_f,timeperiod = length)]

    mdi = []

    for i in cache:
        if np.isnan(i[1]) != True:
            mdi.append(i[1])

    mdi = np.array(mdi,dtype='float32')
    mdi = preprocessing.minmax_scale(mdi)

    while len(mdi) < len(close_float):
        mdi = np.insert(mdi,0,0)
    return mdi

def SMA(length,close_float):
    # 単純移動平均(SMA -> Simple Moving Average)
    c_f = close_float.copy()
    cache = np.c_[c_f, talib.SMA(c_f,timeperiod = length)]

    sma = []

    for i in cache:
        if np.isnan(i[1]) != True:
            sma.append(i[1])

    sma = np.array(sma,dtype='float32')
    sma = preprocessing.minmax_scale(sma)

    while len(sma) < len(close_float):
        sma = np.insert(sma,0,0)
    return sma

def RSI(length, close_float):
    # RSI: Relative Strength Index
    c_f = close_float.copy()
    cache = np.c_[c_f, talib.RSI(c_f,timeperiod = length)]

    rsi = []

    for i in cache:
        if np.isnan(i[1]) != True:
            rsi.append(i[1])

    rsi = np.array(rsi,dtype='float32')
    rsi = preprocessing.minmax_scale(rsi)

    while len(rsi) < len(close_float):
        rsi = np.insert(rsi,0,0)
    return rsi

def MACD(len_short, len_long, len_signal, close_float,):
    # MACD: Moving Average Convergence/Divergence
    #戻り値
    #return_1 -> 短期平均(len_short) - 長期平均(len_long)
    #return_2 -> シグナル(len_signal)
    #return_3 -> return_1 - return_2
    output = c_f = close_float.copy()
    cache = talib.MACD(c_f,len_short,len_long,len_signal)

    return_1 = []
    return_2 = []
    return_3 = []

    for i in range(len(cache[0])):
        if np.isnan(cache[0][i]) != True:
            return_1.append(cache[0][i])
            return_2.append(cache[1][i])
            return_3.append(cache[2][i])

    return_1 = np.array(return_1,dtype='float32')
    return_1 = preprocessing.minmax_scale(return_1)
    return_2 = np.array(return_2,dtype='float32')
    return_2 = preprocessing.minmax_scale(return_2)
    return_3 = np.array(return_3,dtype='float32')
    return_3 = preprocessing.minmax_scale(return_3)

    while len(return_1) < len(close_float):
        return_1 = np.insert(return_1,0,0)
        return_2 = np.insert(return_2,0,0)
        return_3 = np.insert(return_3,0,0)

    return return_1,return_2,return_3