250 lines
3.8 KiB
Python
250 lines
3.8 KiB
Python
#!/usr/bin/env python
|
|
# coding: utf-8
|
|
|
|
# In[2]:
|
|
|
|
|
|
import pandas as pd
|
|
import datetime
|
|
import numpy as np
|
|
|
|
import CoreTraidMath
|
|
import CoreDraw
|
|
from tqdm import tqdm
|
|
|
|
from indicators_v2 import *
|
|
|
|
|
|
# In[3]:
|
|
|
|
|
|
df_candle = pd.read_csv(r"../data/EURUSD_price_candlestick.csv")
|
|
df_candle.rename(columns={'timestamp': 'date'}, inplace=True)
|
|
df_candle
|
|
|
|
|
|
# In[4]:
|
|
|
|
|
|
class coreSignalTrande:
|
|
|
|
def __init__(self, name: str, req: dict, dataType: str):
|
|
self.name = name
|
|
self.agrigateInds = self.createIndicatorsInstance(req)
|
|
self.params = req['params']
|
|
self.dataType = dataType
|
|
|
|
|
|
def createIndicatorsInstance(self,req: dict) -> dict:
|
|
return indicatorsAgrigator(req['indicators'])
|
|
|
|
def getIndAns(self, dataDict: dict) -> dict:
|
|
return self.agrigateInds.getAns(dataDict)
|
|
|
|
def getAns(self, data: pd.DataFrame(), indDataDict: dict) -> dict:
|
|
return self.getSigAns(data, self.getIndAns(indDataDict))
|
|
|
|
|
|
|
|
class sig_BB(coreSignalTrande):
|
|
"""
|
|
ind keys:
|
|
ind_BB
|
|
"""
|
|
|
|
def __init__(self, name: str, req:dict):
|
|
super().__init__(name, req, 'ochl')
|
|
|
|
def getSigAns(self, data: pd.DataFrame(), indAnsDict: dict) -> dict:
|
|
|
|
lastValue = data[self.params['source']].to_list()[-1]
|
|
if lastValue>indAnsDict['ind_BB']['pSTD'][-1]:
|
|
ans='down'
|
|
elif lastValue<indAnsDict['ind_BB']['mSTD'][-1]:
|
|
ans='up'
|
|
else:
|
|
ans='none'
|
|
|
|
return ans
|
|
|
|
|
|
|
|
# In[5]:
|
|
|
|
|
|
class signalsAgrigator:
|
|
|
|
def __init__ (self,req:dict):
|
|
self.signals = self.createSignalsInstance(req)
|
|
|
|
def createSignalsInstance(self, siganlsDict: dict) -> dict:
|
|
ans = {}
|
|
for i in siganlsDict.keys():
|
|
ans[i]=siganlsDict[i]['className'](name = i, req = siganlsDict[i])
|
|
return ans
|
|
|
|
def getAns(self, dataDict: dict) -> dict:
|
|
ans = {}
|
|
for i in dataDict.keys():
|
|
ans[i] = self.signals[i].getAns(data = dataDict[i]['signalData'],
|
|
indDataDict = dataDict[i]['indicatorData'])
|
|
return ans
|
|
|
|
|
|
# In[ ]:
|
|
|
|
|
|
|
|
|
|
|
|
# In[6]:
|
|
|
|
|
|
sigreq= {
|
|
'params':{'source':'close','target':'close'},
|
|
'indicators':{
|
|
'ind_BB':{
|
|
'className':ind_BB,
|
|
'params':{'MeanType':'SMA','window':15,'valueType':'close','kDev':2.5}
|
|
}
|
|
}
|
|
}
|
|
|
|
indReqDict ={'ind_BB':df_candle[:1000]}
|
|
|
|
|
|
# In[7]:
|
|
|
|
|
|
sigAgrReq = {
|
|
'sig_BB':{
|
|
'className':sig_BB,
|
|
'params':{'source':'close','target':'close'},
|
|
'indicators':{
|
|
'ind_BB':{
|
|
'className':ind_BB,
|
|
'params':{'MeanType':'SMA','window':15,'valueType':'close','kDev':2.5}
|
|
}
|
|
}
|
|
},
|
|
'sig_BB_2':{
|
|
'className':sig_BB,
|
|
'params':{'source':'close','target':'close'},
|
|
'indicators':{
|
|
'ind_BB':{
|
|
'className':ind_BB,
|
|
'params':{'MeanType':'SMA','window':30,'valueType':'close','kDev':2}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
sigAgrData = {
|
|
'sig_BB':{
|
|
'signalData': df_candle[990:1000],
|
|
'indicatorData' :{'ind_BB': df_candle[:1000]}
|
|
},
|
|
'sig_BB_2':{
|
|
'signalData': df_candle[990:1000],
|
|
'indicatorData' :{'ind_BB': df_candle[:1000]}
|
|
}
|
|
}
|
|
|
|
|
|
# In[ ]:
|
|
|
|
|
|
|
|
|
|
|
|
# In[8]:
|
|
|
|
|
|
ttt=signalsAgrigator(sigAgrReq)
|
|
|
|
|
|
# In[9]:
|
|
|
|
|
|
ttt.__dict__
|
|
|
|
|
|
# In[10]:
|
|
|
|
|
|
ttt.signals['sig_BB'].__dict__
|
|
|
|
|
|
# In[11]:
|
|
|
|
|
|
ttt.getAns(sigAgrData)
|
|
|
|
|
|
# In[ ]:
|
|
|
|
|
|
|
|
|
|
|
|
# In[ ]:
|
|
|
|
|
|
|
|
|
|
|
|
# In[12]:
|
|
|
|
|
|
list({'ttt':2}.keys())[0]
|
|
|
|
|
|
# In[13]:
|
|
|
|
|
|
test = sig_BB('sig_BB', sigreq)
|
|
|
|
|
|
# In[14]:
|
|
|
|
|
|
test.__dict__
|
|
|
|
|
|
# In[ ]:
|
|
|
|
|
|
|
|
|
|
|
|
# In[15]:
|
|
|
|
|
|
test.agrigateInds.__dict__
|
|
|
|
|
|
# In[16]:
|
|
|
|
|
|
ians = test.getIndAns(indReqDict)
|
|
ians
|
|
|
|
|
|
# In[17]:
|
|
|
|
|
|
test.getAns(df_candle[:100],indReqDict)
|
|
|
|
|
|
# In[ ]:
|
|
|
|
|
|
|
|
|
|
|
|
# In[ ]:
|
|
|
|
|
|
|
|
|