A base de dados original está no arquivo “Cotacoes.csv” e lista as cotações dos preços do ativo WIN (B3) no tempo gráfico de 10 min. Abaixo seguem os 20 primeiros registros, mas a base tem 67624 registros:
DATE;TIME;OPEN;HIGH;LOW;CLOSE;TICKVOL;VOL;SPREAD
2017/09/04;09:00:00;72300;72480;72255;72405;6308;24832;5
2017/09/04;09:10:00;72400;72470;72370;72405;3641;14569;5
2017/09/04;09:20:00;72410;72435;72305;72315;4339;19095;5
2017/09/04;09:30:00;72315;72360;72285;72355;3272;14470;5
2017/09/04;09:40:00;72355;72375;72305;72310;2723;12022;5
2017/09/04;09:50:00;72310;72385;72305;72350;2169;9231;5
2017/09/04;10:00:00;72350;72365;72275;72340;4172;16387;5
2017/09/04;10:10:00;72340;72440;72330;72365;5693;23302;5
2017/09/04;10:20:00;72365;72480;72365;72435;5089;20870;5
2017/09/04;10:30:00;72435;72515;72350;72365;7786;31227;5
2017/09/04;10:40:00;72370;72430;72310;72430;6648;27973;5
2017/09/04;10:50:00;72430;72450;72390;72435;3372;16180;5
2017/09/04;11:00:00;72435;72500;72410;72485;4164;17121;5
2017/09/04;11:10:00;72490;72645;72430;72595;8919;37285;5
2017/09/04;11:20:00;72595;72600;72535;72565;3886;15673;5
2017/09/04;11:30:00;72565;72675;72555;72625;6217;25463;5
2017/09/04;11:40:00;72620;72695;72620;72625;3075;12662;5
2017/09/04;11:50:00;72630;72650;72590;72595;2762;10595;5
2017/09/04;12:00:00;72595;72650;72575;72610;2775;11117;5
Abaixo segue o código:
import os as os
from matplotlib import pyplot as plt
import numpy as np
from datetime import datetime
# Classe Controle rede neural
class cRedeNeural():
def fCriarRede(self, n_inputs = 3, n_hiddena=5, n_hiddenb=4, n_outputs=3, learningrate=0.0015):
# # Pesos
# self.w_ia = 2 * np.random.random((n_hiddena, n_inputs)) – 1
# self.w_ab = 2 * np.random.random((n_hiddenb, n_hiddena)) – 1
# self.w_bo = 2 * np.random.random((n_outputs, n_hiddenb)) – 1
# # Bias
# self.b_ia = 2 * np.random.random((n_hiddena)) – 1
# self.b_ab = 2 * np.random.random((n_hiddenb)) – 1
# self.b_bo = 2 * np.random.random((n_outputs)) – 1
# Pesos
self.w_ia = np.random.default_rng(40).random((n_hiddena, n_inputs))
self.w_ab = np.random.default_rng(41).random((n_hiddenb, n_hiddena))
self.w_bo = np.random.default_rng(42).random((n_outputs, n_hiddenb))
# Bias
self.b_ia = np.random.default_rng(43).random((n_hiddena))
self.b_ab = np.random.default_rng(44).random((n_hiddenb))
self.b_bo = np.random.default_rng(45).random((n_outputs))
# Somas ponderadas das camadas
self.s_ia = np.zeros((n_hiddena))
self.s_ab = np.zeros((n_hiddenb))
self.s_bo = np.zeros((n_outputs))
# Saídas das camadas após a função de ativação
self.z_ia = np.zeros((n_hiddena))
self.z_ab = np.zeros((n_hiddenb))
self.z_bo = np.zeros((n_outputs))
# Deltas das camadas
self.d_ia = np.zeros((n_hiddena))
self.d_ab = np.zeros((n_hiddenb))
self.d_bo = np.zeros((n_outputs))
# Learning rate
self.eta = learningrate
self.erro = list()
def fPropagacao(self, inputs):
self.s_ia = np.dot(self.w_ia, inputs) + self.b_ia
self.z_ia = fRelu(self.s_ia)
self.s_ab = np.dot(self.w_ab, self.s_ia) + self.b_ab
self.z_ab = fRelu(self.s_ab)
self.s_bo = np.dot(self.w_bo, self.s_ab) + self.b_bo
self.z_bo = fSoftmax(self.s_bo)
return self.z_bo
def fretropropagacao(self, x, erro, m):
# Calcula deltas para todas as camadas
self.d_bo = erro * fDerivadaSoftmax(self.z_bo)
self.d_ab = np.dot(self.w_bo.T, self.d_bo) * fDerivadaRelu(self.z_ab)
self.d_ia = np.dot(self.w_ab.T, self.d_ab) * fDerivadaRelu(self.z_ia)
# Atualiza as matrizes de pesos
n_i = x.shape[0]
n_ha = self.d_ia.shape[0]
n_hb = self.d_ab.shape[0]
n_o = self.d_bo.shape[0]
r = np.reshape(self.d_bo, [n_o, 1])
s = np.reshape(self.z_ab, [1, n_hb])
self.w_bo = (self.w_bo * m) – self.eta * np.dot(r, s)
r = np.reshape(self.d_ab,[n_hb,1])
s = np.reshape(self.z_ia,[1,n_ha])
self.w_ab = (self.w_ab * m) – self.eta * np.dot(r, s)
r = np.reshape(self.d_ia,[n_ha,1])
s = np.reshape(x,[1,n_i])
self.w_ia = (self.w_ia * m) – self.eta * np.dot(r, s)
# Atualiza as matrizes de bias
self.b_bo = (self.b_bo * m) – self.eta * self.d_bo
self.b_ab = (self.b_ab * m) – self.eta * self.d_ab
self.b_ia = (self.b_ia * m) – self.eta * self.d_ia
def fTreinarRede(self, x, epocas, lotes, m):
dset = fBaseSplit(x,lotes)
ErroLoteant = 0.0
erromedioant = 0.0
cl = 0
plt.style.use(‘ggplot’)
eixox = list()
eixoy = list()
eixoyy = list()
plt.ion()
for e in range(epocas):
for lote in dset:
for m_lote in lote:
Err = list()
for f in m_lote:
f1 = f[3:]
y_true = f[0:3]
y_pred = self.fPropagacao(f1)
#sErro = fMSE(y_true, y_pred)
sErro = fCross_E(y_true, y_pred)
Err.append(sErro)
ErroLote = np.array(Err).mean()
self.erro.append(ErroLote)
flag = ‘ ■’ # alt 254
if (ErroLote < ErroLoteant ):
flag = ‘ ▼’ # alt 31
elif (ErroLote > ErroLoteant ):
flag = ‘ ▲’ # alt 30
ErroLoteant = ErroLote
erromedio = np.array(self.erro).mean()
flag1 = ‘ ■’ # alt 254
if (erromedio < erromedioant ):
flag1 = ‘ ▼’ # alt 31
elif (erromedio > erromedioant ):
flag1 = ‘ ▲’ # alt 30
erromedioant = erromedio
self.fretropropagacao(f1,ErroLote, m)
eixoy.append(ErroLote)
plt.cla()
plt.clf()
eixox.append(cl)
cl += 1
plt.plot(eixox, eixoy, ‘b’)
plt.pause(0.001)
print(‘Epoca: ‘, ‘{:0>5}’.format(e), ‘/’, ‘{:0>5}’.format(epocas), ‘ – ErroMédio: ‘, f”{erromedio:.5f}” + flag1, ‘ – Erro: ‘, f”{ErroLote:.5f}” + flag)
plt.ioff()
plt.show()
return self.erro
# Função que calcula os valores de entrada da Rede Neural
def fCalcular(s):
s = np.array(s)
s = np.delete(s,(0), axis = 0)
s = np.delete(s,(0), axis = 1)
s = np.delete(s,(0), axis = 1)
# s -> OPEN, HIGH, LOW, CLOSE, TICKVOL, VOL, SPREAD
m = np.zeros((s.shape[0], s.shape[1]+18))
# Ordenacao crescente
m[:,0] = np.arange(m.shape[0])
m[:,1:8] = s[:,:]
a = np.zeros((s.shape[0]))
# Delta do retorno deslocado – Alvo
a = (s[1:,3] / s[:-1,3]) – 1
m[0:-2,8] = a[1:] #m[0:a.shape[0]-1,8] = a[1:]
# 1,0,0 = buy
m[1:,9] = np.where((m[1:,8] > 0) & (m[:-1,8] < 0), 1, 0)
# 0,0,1 = sell
m[1:,11] = np.where((m[1:,8] < 0) & (m[:-1,8] > 0), 1, 0)
# 0,1,0 = hold
m[1:,10] = np.where((m[1:,9] == 0) & (m[1:,11] == 0), 1, 0)
tcandle = np.nan_to_num(s[:,1] – s[:,2])
# Tamanho corpo do candle
m[:,12] = np.nan_to_num((s[:,3] – s[:,0]) / tcandle)
# Tamanho pavio do candle
m[:,13] = np.nan_to_num(1 – m[:,12])
# Distância entre fechamento e a média: Configuração: 9 -> 21 -> 200
m[1:,14] = np.nan_to_num(s[1:,3] – fMediaMovel(s[1:,3],9))
m[1:,15] = np.nan_to_num(s[1:,3] – fMediaMovel(s[1:,3],21))
m[1:,16] = np.nan_to_num(s[1:,3] – fMediaMovel(s[1:,3],200))
# Delta do volume
m[1:,17] = np.nan_to_num((s[1:,5] / s[:-1,5]) – 1)
# Calcula IFR:
calc = np.nan_to_num(fCalculaIFR(s[:,3]), 14)
m[2:,18] = np.where((calc[:-1] == 0) , 0, calc[:-1]) #np.nan_to_num(calc[1:] / calc[:-1] -1))
# Calcula Didi Index: Configuração: 3 -> 8 > 20
mm8 = np.nan_to_num(fMediaMovel(s[1:,3],8))
m[1:,19] = np.nan_to_num(fMediaMovel(s[1:,3],3) – mm8)
m[1:,20] = np.nan_to_num(fMediaMovel(s[1:,3],20) – mm8)
# Calcula ADX
vadx, vdip, vdin = np.nan_to_num(fCalculaADX(s[:,:4],8))
# vadx = adx -> vdip = di+ -> vdin = di-
t1 = np.zeros(vadx.shape[0])
t1[1:] = np.where(((vadx[1:] < 32) & (vadx[1:] < vadx[:-1])) | ((vadx[1:] <= vdip[1:]) & (vadx[1:] <= vdin[1:])), 0, 1)
#t2 = t3 = t4 = np.zeros(vadx.shape[0])
#t2 = np.where((t1 == 1) & (vdip > vdin ), 1, 0)
#t3 = np.where((t1 == 1) & (vdip < vdin ), 1, 0)
#t4 = np.where((vadx[:-3] < vadx[1:-2]) & (vadx[1:-2] > vadx[2:-1]), 1, 0)
# m[1:,21] -> Indica tendência de alta
m[1:,21] = np.where((t1 == 1) & (vdip > vdin ), 1, 0)
# m[1:,22] -> Indica tendência de baixa
m[1:,22] = np.where((t1 == 1) & (vdip < vdin ), 1, 0)
# m[1:,23] -> topo ou fundo
m[4:,23] = np.where((vadx[:-3] < vadx[1:-2]) & (vadx[1:-2] > vadx[2:-1]), 1, 0)
# Calcula distancia entre as Banda Bollinger: Configuracao: 20 -> 2
m[:,24] = np.nan_to_num(fCalculaBBolinger(s[:,3],20,2))
# Normaliza base de dados
ret = m[201:,8:]
minmax = [[min(column), max(column)] for column in zip(*ret)]
for col in (range(ret.shape[1]-3)):
c = col + 3
ret[:,c] = (ret[:,c] – minmax[c][0]) / (minmax[c][1] – minmax[c][0])
# Retorno da base de dados inteira (normalizados)
return ret
# Função que calcula o ADX
def fCalculaADX(x, n=8):
#s -> OPEN, HIGH, LOW, CLOSE, TICKVOL, VOL, SPREAD
vaux = np.zeros((x.shape[0],3))
pDM = np.nan_to_num(fMediaMovel(x[1:,1] – x[:-1,1], n)) # +DM = mma(high[0] – high[-1], periodo)
nDM = np.nan_to_num(fMediaMovel(x[1:,2] – x[:-1,2], n)) # -DM = mma(low[0] – low[-1] , periodo)
vaux[1:,0] = np.nan_to_num(x[1:,1] – x[1:,2]) # high[0] – low[0]
vaux[1:,1] = np.nan_to_num(x[1:,1] – x[:-1,3]) # high[0] – close[-1]
vaux[1:,2] = np.nan_to_num(x[1:,2] – x[:-1,3]) # low[0] – close[-1]
TR = np.amax(vaux,axis=1)
ATR = np.nan_to_num(fMediaMovel(TR, n))
pDI = np.nan_to_num(pDM / ATR[:-1] * 100)
nDI = np.nan_to_num(nDM[:] / ATR[:-1] * 100)
DX = np.nan_to_num(np.abs(pDI – nDI) / np.abs(pDI + nDI) * 100)
ADX = np.nan_to_num(fMediaMovel(DX, n))
return ADX, pDI, nDI
# Função que calcula o IFR
def fCalculaIFR(x, n=14):
ifr = np.eye(x.shape[0], 3)
ifr[0,0] = 0
ifr[1:,0] = x[1:] – x[:-1]
ifr[1:,1] = np.where(ifr[1:,0] > 0, ifr[1:,0], 0)
ifr[1:,2] = np.where(ifr[1:,0] < 0, np.abs(ifr[1:,0]), 0)
try:
ret = 100 – 100 / (1 + (fMediaMovel(ifr[1:,1], n) / fMediaMovel(ifr[1:,2], n)))
except ZeroDivisionError:
ret = 0
return ret
# Função que calcula abertura das Banda de Bollinger -> abrindo = 1 e fechando = -1
def fCalculaBBolinger(x, n=20, k=2):
mma = np.nan_to_num(fMediaMovel(x,n))
s = np.nan_to_num(np.where(mma == 0, 0, (x – mma)**2))
dp = np.zeros(s.shape[0])
dp[n-1:] = (np.convolve(s, np.ones(n), ‘valid’)) ** 0.5
delta = (mma + dp * k) – (mma – dp * k)
ret = np.zeros(s.shape[0])
ret[1:] = np.where(delta[1:] < delta[:-1], -1, 1)
return ret
# Função que calcula a média móvel
def fMediaMovel(x, n):
ret = np.zeros(x.shape[0])
ret[n-1:] = np.convolve(x, np.ones(n), ‘valid’) / n
return ret
# Funcao de ativacao RELU
def fRelu(x):
ret = np.zeros(x.shape[0])
for i in range(len(x)):
ret[i] = max(x[i], 0.0015)
return ret
# Funcao de ativacao SOFTMAX
def fSoftmax(x):
xs = x – x.max()
n = np.exp(xs)
d = np.sum(n)
ret = n/d
return ret
# Calcula derivada – RELU
def fDerivadaRelu(x):
ret = np.zeros(x.shape[0])
for i in range(len(x)):
ret[i] = 1 if x[i] > 0 else 0.0015
return ret
# Calcula derivada – Softmax
def fDerivadaSoftmax(x):
ret = np.zeros(3)
ret = x * (1 – x)
return ret
# Erro pela entropia cruzada
def fCross_E(y_true, y_pred):
ret = (y_true * np.log10(y_pred))
ret = – np.sum(ret)
return ret
def fDerivadaCross_E(y_true, y_pred):
return -y_true/(y_pred + 10**-100)
# Erro médio quadrado
def fMSE(y_true, y_pred):
ret = -np.sum(1/2 * (y_true – y_pred) ** 2)
return ret
# Split a dataset into k folds
def fBaseSplit(dataset, n_folds):
dataset_split = list()
dataset_copy = list(dataset)
fold_size = int(len(dataset) / n_folds)
for i in range(n_folds):
fold = list()
fold.append(dataset_copy[:fold_size])
dataset_split.append(fold)
dataset_copy = dataset_copy[fold_size:]
return dataset_split
def fSalvaLog(a , t, x):
if (type(x) is np.ndarray):
x1 = np.array2string(x, formatter={‘float_kind’:lambda x: “%01.5f” % x})
elif ((type(x) is np.float64) or (type(x) is int)):
x1 = str(x)
else:
x1 = x
a.write(“\n\n” + t)
a.write(“\n” + x1)
#===========================================================================
# Corpo do programa
# Formatacao da saida do print para notacao cientifica
np.set_printoptions(formatter={‘float’:lambda x: ‘%+01.9f ‘ % x})
#Abre arquivoo log.txt
narq = os.getcwd() + ‘\\Log\\Log’ + datetime.now().strftime(‘%Y%m%d%H%M’) + ‘.txt’
arq = open(narq, “w”)
# Carrega base de dados
arquivo = os.getcwd() + ‘\\Cotacoes.csv’
dataset = np.genfromtxt(arquivo, delimiter=’;’)
df = fCalcular(dataset)
#np.savetxt(‘Calculados.csv’, df, fmt=’%01.5f’, header= ‘;Open;High;Low;Close;TickVol;Vol;Spread;Retorno;Buy;Hold;Sell;Candle;Corpo;DeltaMM9;DeltaMM21;DeltaMM200;DeltaVol;DeltaIFR14’, delimiter = ‘;’)
# Filtra base de dados
qTreino = np.int32(df.shape[0] * 0.8)
xTreino = df[:qTreino,1:]
xTeste = df[qTreino:,1:]
cabec = ‘Buy;’
cabec += ‘Hold;’
cabec += ‘Sell;’
cabec += ‘CorpoCandle;’
cabec += ‘PavioCandle;’
cabec += ‘DistanciaMMA9;’
cabec += ‘DistanciaMMA21;’
cabec += ‘DistanciaMM200;’
cabec += ‘DeltaVolume;’
cabec += ‘IFR;’
cabec += ‘DIDIMais;’
cabec += ‘DIDIMenos;’
cabec += ‘ADXAlta;’
cabec += ‘ADXBaixa;’
cabec += ‘ADXTopoFundo;’
cabec += ‘DistanciaBandasBollinger’
np.savetxt(“xTreino.csv”, xTreino, fmt=’%01.5f’, header=cabec, delimiter = “;”)
np.savetxt(“xTeste.csv”, xTeste, fmt=’%01.5f’, header=cabec, delimiter = “;”)
# Executa o Modelo
lote = 10000 # ajuste do erro a cada 100 iteracoes
l_rate = 0.002 # taxa de aprendizado
mo = 0.9 # momentum
nEpocas = 10 # numero de epocas
boolArray = [
True, # Buy
True, # Hold
True, # Sell
True, # Corpo Candle
True, # Pavio Candle
True, # Close – MMA9
True, # Close – MMA21
True, # Close – MM200
True, # Delta do VOLUME
True, # IFR14
True, # DIDI (+)
True, # DIDI(-)
True, # ADX – Tendência de alta
True, # ADX – Tendência de baixa
True, # ADX – topo ou fundo
True # Distancia entre as banda de bollinger
]
ds = xTreino[:,boolArray] # dataset treino
ni = ds.shape[1]-3 # nr de entradas
na = 12 # neurônios na primeira camada oculta
nb = 5 # neurônios na segunda camada oculta
no = 3 # nr de saídas
#Inicializa a rede
p = cRedeNeural()
p.fCriarRede(ni, na, nb, no, l_rate)
fSalvaLog(arq,’**’,’Rede: ‘+ str(ni).replace(‘.’,’,’) + ‘ – ‘ + str(na).replace(‘.’,’,’) + ‘ – ‘ + str(nb).replace(‘.’,’,’)+ ‘ – ‘ + str(no).replace(‘.’,’,’))
fSalvaLog(arq,’**’,’Lote: ‘ + str(lote).replace(‘.’,’,’) + ‘ – Learning rate: ‘ + str(l_rate).replace(‘.’,’,’) + ‘ – Épocas: ‘ + str(nEpocas).replace(‘.’,’,’))
fSalvaLog(arq,’Campos’,np.array(boolArray))
fSalvaLog(arq,’w_ia’,p.w_ia)
fSalvaLog(arq,’b_ia’,p.b_ia)
fSalvaLog(arq,’w_ab’,p.w_ab)
fSalvaLog(arq,’b_ab’,p.b_ab)
fSalvaLog(arq,’w_bo’,p.w_bo)
fSalvaLog(arq,’b_bo’,p.b_bo)
erro = p.fTreinarRede(ds, nEpocas, lote, mo)
erro = np.array(erro)
erromedio = erro.mean()
# Rotina de teste
ds = xTeste[:,boolArray]
eixoy_true = list()
eixoy_pred = list()
eixox = list()
cl = 0
for e in ds:
f1 = e[3:]
y_true = e[0:3]
y_pred = p.fPropagacao(f1)
eixoy_true.append(np.argmax(y_true))
eixoy_pred.append(np.argmax(y_pred))
eixox.append(cl)
cl +=1
print(str(cl) + ” – ” + str(np.argmax(y_true)) + ” – ” + str(np.argmax(y_pred)))
np.savetxt(“ResultadoTeste.csv”, eixox, fmt=’%01.5f’, header=’Contador;True;Pred’, delimiter = “;”)
fSalvaLog(arq,’Erro médio:’,f”{erromedio:.5f}”)
fSalvaLog(arq,’w_ia’,p.w_ia)
fSalvaLog(arq,’b_ia’,p.b_ia)
fSalvaLog(arq,’w_ab’,p.w_ab)
fSalvaLog(arq,’b_ab’,p.b_ab)
fSalvaLog(arq,’w_bo’,p.w_bo)
fSalvaLog(arq,’b_bo’,p.b_bo)
# Fechar arquivo log.txt
arq.close()
plt.plot(erro,’r’)
# plt.scatter(eixox,eixoy_true, color=’b’, s=5, edgecolor=’none’)
# plt.scatter(eixox,eixoy_pred, color=’g’, s=5, edgecolor=’none’)
plt.show()
#===========================================================================