marketTrade/notebooks/autogen/Indicators.py

#!/usr/bin/env python
# coding: utf-8

# In[2]:


import pandas as pd
import datetime
import numpy as np
#import plotly as pl

#import plotly.graph_objs as go
#from plotly.offline import init_notebook_mode, iplot
#from plotly.subplots import make_subplots
#init_notebook_mode()

import CoreTraidMath 
import CoreDraw


# In[3]:


class coreIndicator():
    def __init__(self,
                 data=pd.DataFrame(),
                 options={},
                 showMode='None',
                ):
        '''
        showMode = None/Ind/PartOf        
        '''
        self.data=data
        self.showMode=showMode
        self.options=options
        self.overlayInd=None #True/False
        self.ans=None
        self.figDict=None
        
    def getAns(self,data=None):
        if type(data)!=type(None):
            self.data=data
        self.ans=self.getCalculate()
        if self.showMode=='Ind' or self.showMode=='PartOf':
            self.figDict=self.getFigDict()
            if self.showMode=='Ind':
                self.getFig()
        return self.ans
    def getFig(self,row=1):
        CoreDraw.coreDraw(self.figDict,True)
    def getCalculate(self):
        return "Error"
    def getFigDict(self):
        return "Error"

class indicatorAgrigator():
    '''
    Тема чисто для отладки
    jj=indicatorAgrigator().runAll([o1,o2],df_candle[:30])
    #jj.createIndFromList([o1,o2])
    #jj.calculateInd(df_candle[:30])
    
    '''
    def __init__(self):
        self.indList=None
        self.data=None
    def createInd(self,classDict):
        return classDict['name'](
                options=classDict['params'],
                showMode=classDict['showMode']
            )
        
        
    def createIndFromList(self,indList):
        self.indList=indList
        ans=[]
        for i in self.indList:
            ans.append(self.createInd(i))
        self.indList=ans
        return ans
    
    def calculateInd(self,data):
        self.data=data
        for i in self.indList:
            #i.getAns(data)
            i.data=self.data
            i.ans=i.getCalculate()
            i.figDict=i.getFigDict()
            #i.getFig()
    def agrigateFig(self):
        req=[[]]
        
        for i in self.indList:
            if i.overlayInd==True:
                req[0].append(i)
            else:
                req.append([i])
        CoreDraw.agrigateFig(req,True)
    def runAll(self,indList,df,needDraw=False):
        self.createIndFromList(indList)
        self.calculateInd(df)
        if needDraw:
            self.agrigateFig()

        
        


# In[4]:


class ind_BB(coreIndicator):
    
    def getCalculate(self):
        self.overlayInd=True
        ans={}
        opMA={'dataType':'ohcl',
            'action':'findMean',
            'actionOptions':{
                'MeanType':self.options['MeanType'],
                'valueType':self.options['valueType'],
                'window':self.options['window']
            }
        }
        ans['BB']=CoreTraidMath.CoreMath(self.data,opMA).ans
        opSTD={'dataType':'ohcl',
            'action':'findSTD',
            'actionOptions':{'valueType':self.options['valueType'],'window':self.options['window']}
        }
        ans['STD']=CoreTraidMath.CoreMath(self.data,opSTD).ans
        ans['pSTD']=ans['BB']+ans['STD']*self.options['kDev']
        ans['mSTD']=ans['BB']-ans['STD']*self.options['kDev']
        ans['x']=np.array(self.data['date'][self.options['window']-1:].to_list())
        return ans
    def getFigDict(self,row=1):
        req=[]
        
        req.append({
            'vtype':'Scatter',
            'df':pd.DataFrame(
                {'value':self.ans['BB'],'date':self.ans['x']}) ,
            'row':row,
            'col':1,
            'name':'BB'
            
        })
        req.append({
            'vtype':'Scatter',
            'df':pd.DataFrame(
                {'value':self.ans['pSTD'],'date':self.ans['x']}) ,
            'row':row,
            'col':1,
            'name':'pSTD'
            
        })
        req.append({
            'vtype':'Scatter',
            'df':pd.DataFrame(
                {'value':self.ans['mSTD'],'date':self.ans['x']}) ,
            'row':row,
            'col':1,
            'name':'mSTD'
            
        })
        
        return req


# In[5]:


class ind_OCHL(coreIndicator):
    def getCalculate(self):
        self.overlayInd=True
    def getFigDict(self,row=1):
        req=[]
        
        req.append({
                'vtype':'OCHL',
                'df':self.data,
                'row':1,
                'col':1,
                'name':'OHCL'

        })
        return req


# In[7]:


df_candle = pd.read_csv("../data/EURUSD_price_candlestick.csv")
df_candle.rename(columns={'timestamp': 'date'}, inplace=True)
df_candle


# In[8]:


o1={
    'name':ind_OCHL,
    'params':{},
    'showMode':'PartOf',
}
o2={
    'name':ind_BB,
    'params':{'MeanType':'SMA','window':25,'valueType':'low','kDev':2},
    'showMode':'PartOf',
}
jj=indicatorAgrigator().runAll([o1,o2],df_candle[:300],True)
#jj.createIndFromList([o1,o2])
#jj.calculateInd(df_candle[:30])


# In[9]:


op={'MeanType':'SMA','window':5,'valueType':'low','kDev':2}
a=ind_BB(df_candle[:100],op,'PartOf')


# In[10]:


a.getAns()


# In[11]:


b=ind_OCHL(df_candle[:30],{},'Ind')
b.getAns(df_candle[:100])


# In[12]:


opc={'MeanType':'SMA','window':20,'valueType':'low','kDev':2}
c=ind_BB(df_candle[:100],opc,'PartOf')
c.getAns()


# In[13]:


hhh = CoreDraw.agrigateFig([[b,a,c]],True)


# In[14]:


import indicators


# In[15]:


op_1={'MeanType':'SMA','window':5,'valueType':'low','kDev':2}
test_1=indicators.ind_BB(df_candle[:100],op)
test_1.getAns()


# In[ ]: