marketTrade/notebooks/autogen/decisionManager.py

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

# In[4]:


import pandas as pd
import datetime
import numpy as np

import pickle
from signals import *
from dealManager import *
from trandeVoter import *
from riskManager import riskManager


# In[5]:


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


# In[6]:


class decsionManager():

    
    def __init__(self,name):
        self.name = name
        self.RM = riskManager()
        self.DM = DealManager()
        self.TV = trandeVoter(name)
        self.SA = signalAgrigator()
        pass
    
    #вытащенный из signalAgrigator метод теста для сигналов
    def getSignalTest(self,data: pd.DataFrame(),reqSig: dict, batchSize=30, dataType='candel') -> dict:
        
        self.SA.mode = 'retroFast'
        t.SA.createSingnalInstances(
            data = data,
            dictAgrigSignal = reqSig,
            dataType='candel',
            batchSize=30
        )
        ans = t.SA.getAns(data)
        return ans
    
    #метод для генерации матрицы вероятностей.
    def generateMatrixProbability(self, 
                                  data: pd.DataFrame(),
                                  reqSig: dict, 
                                  target: str, 
                                  batchSize=30, 
                                  #dataType='candel'
                                 ):
        data=data.reset_index(drop=True)
        
        t.SA.createSingnalInstances(
            data = data,
            dictAgrigSignal = reqSig,
            dataType='candel',
            batchSize=batchSize
        )
        self.TV.createMatrixAmounts(reqSig.keys())
        for i in range(data.shape[0]-batchSize-1):
            sigAns=self.SA.getAns(data[i:i+batchSize])
            rightAns=self.getRetroStepAns(data[target][i],data[target][i+1])
            self.TV.setDecisionBySignals(self.KostilEbaniy(sigAns),rightAns)
        self.TV.generateMatrixProbability()
    
    #без коментариев блять
    def KostilEbaniy(self,d):
        ans={}
        for i in d.keys():
            if d[i] == 0:
                ans[i] = 'none'
            elif d[i] == 1:
                ans[i] = 'up'
            elif d[i] == -1:
                ans[i] = 'down'
        return ans
    
    #тож понятная хуита
    def getRetroStepAns(self, value1,value2):
        
        if value1 == value2:
            ans = 'none'
        elif value1 < value2:
            ans = 'up'
        else:
            ans = 'down'
        return ans
        
    #метод для онлай получения решения по сигналу
    def getSignal(self,data: pd.DataFrame(),reqSig: dict, dataType='candel') -> dict:
        data=data.reset_index(drop=True)
        self.SA.mode = 'online'
        t.SA.createSingnalInstances(
            data = data,
            dictAgrigSignal = reqSig,
            dataType='candel',
            batchSize=30
        )
        ans = t.SA.getAns(data)
        return ans        
    
    
    #Создание сигналов. Вызывать перед getOnlineAns
    def crateSignals(self,data: pd.DataFrame(),reqSig: dict, dataType='candel'):
        data=data.reset_index(drop=True)
        self.SA.mode = 'online'
        t.SA.createSingnalInstances(
            data = data,
            dictAgrigSignal = reqSig,
            dataType='candel',
            batchSize=30
        )
        
    
    def getOnlineAns(self,data: pd.DataFrame(),price):
        sigAns = self.SA.getAns(data)
        prob = self.TV.getDecisionBySignals(sigAns)
        ans = self.RM.getDecision(sigAns,prob,price)
        return ans
        
        
        
        


# In[ ]:





# In[7]:


t= decsionManager('TEST')


# In[8]:


t.__dict__


# In[9]:


reqSig={
    'BB1':{
        'className':signal_BB,
        'indParams':{'MeanType':'SMA','window':15,'valueType':'close','kDev':2.5},
        'signalParams':{'source':'close','target':'close'},
        'batchSize':15
    },
    'BB2':{
        'className':signal_BB,
        'indParams':{'MeanType':'SMA','window':15,'valueType':'close','kDev':2.5},
        'signalParams':{'source':'close','target':'close'},
        'batchSize':20
    }
}


# In[10]:


reqSig.keys()


# In[11]:


t.SA.__dict__


# In[12]:


t.generateMatrixProbability(df_candle[:10000],reqSig,'close',40)


# In[ ]:





# In[ ]:





# In[ ]:





# In[ ]:





# In[ ]:





# In[ ]:





# In[13]:


mop = t.TV.matrixProbability
mop


# In[ ]:





# In[ ]:





# In[14]:


t.getSignal(df_candle[:10000],reqSig)


# In[15]:


t.getSignalTest(df_candle[:10000],reqSig,40)


# In[ ]:





# In[16]:


t.SA.createSingnalInstances(
    data = df_candle[:10000],
    dictAgrigSignal = reqSig,
    dataType='candel',
    batchSize=30
)


# In[ ]:





# In[ ]:





# In[ ]:





# In[17]:


reqSig={
    'BB1':{
        'className':signal_BB,
        'indParams':{'MeanType':'SMA','window':15,'valueType':'close','kDev':2.5},
        'signalParams':{'source':'close','target':'close'},
        'batchSize':15
    },
    'BB2':{
        'className':signal_BB,
        'indParams':{'MeanType':'SMA','window':15,'valueType':'close','kDev':2.5},
        'signalParams':{'source':'close','target':'close'},
        'batchSize':20
    }
}


# In[18]:


t=decsionManager(reqSig)


# In[ ]:





# In[ ]:


import pickle


# In[ ]:


dictionary_data = {"a": 1, "b": 2}


a_file = open("data.pkl", "wb")

pickle.dump(reqSig, a_file)

a_file.close()


a_file = open("data.pkl", "rb")

output = pickle.load(a_file)

print(output)



a_file.close()


# In[ ]: