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 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