market_trade/core/decisionManager_v2.py

@@ -1,4 +1,5 @@
 import os
+import pickle
 
 import pandas as pd
 import datetime
@@ -6,156 +7,181 @@ import numpy as np
 
 from tqdm import tqdm
 
-from market_trade.core.indicators_v2 import *
-from market_trade.core.signals_v2 import *
-from market_trade.core.dealManager import *
-from market_trade.core.trandeVoter import *
-from market_trade.core.riskManager import *
-import pickle
+from market_trade.core.indicators_v2 import ind_BB
+from market_trade.core.signals_v2 import sig_BB, SignalsAggregator
+from market_trade.core.dealManager import DealManager
+from market_trade.core.trandeVoter import TradeVoter
+from market_trade.core.riskManager import RiskManager
 
 
-class decsionManager:
-    '''
-sigAgrReq = {
-    'sig_BB':{
-        'className':sig_BB,
-        'params':{'source':'close','target':'close'},
-        'indicators':{
-            'ind_BB':{
-                'className':ind_BB,
-                'params':{'MeanType':'SMA','window':30,'valueType':'close','kDev':2.5}
-            }            
+class DecisionManager:
+    """Manages trading decisions based on signals, probability voting, and risk management.
+
+    Coordinates the entire decision-making pipeline:
+    1. Signals from indicators
+    2. Probability-based voting (TradeVoter)
+    3. Risk assessment (RiskManager)
+    4. Deal tracking (DealManager)
+
+    Example configuration:
+        sig_config = {
+            'sig_BB': {
+                'className': sig_BB,
+                'params': {'source': 'close', 'target': 'close'},
+                'indicators': {
+                    'ind_BB': {
+                        'className': ind_BB,
+                        'params': {'MeanType': 'SMA', 'window': 30, '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]}
-    }    
-}
+    def __init__(self, name: str, sig_dict: dict):
+        """Initialize DecisionManager with configuration.
 
-
-sigAgrRetroTemplate = {
-    'sig_BB':{
-        'signalData': None,
-        'indicatorData' :{'ind_BB': None}
-    },
-    'sig_BB_2':{
-        'signalData': None,
-        'indicatorData' :{'ind_BB': None}
-    }     
-}    
-    
-    
-    
-    
-    
-    
-    '''
-    
-    
-    
-    
-    def __init__(self,name, sigDict: dict):
-        self.RM = riskManager()
+        Args:
+            name: Identifier for this decision manager instance.
+            sig_dict: Dictionary of signal configurations.
+        """
+        self.RM = RiskManager()
         self.DM = DealManager()
-        self.TV = trandeVoter(name)
-        self.SA = signalsAgrigator(sigDict)
-        self.sigDict = sigDict
-        
-        
-    def getOnlineAns(self, signalsAns: dict, price: float) -> dict:
-        probabilityDecsion = self.TV.getDecisionBySignals(self.getSignalsAns(signalsAns))
-        RMD = self.RM.getDecision(probabilityDecision=probabilityDecsion, price=price, deals = self.DM.deals)
-        return RMD
-        
-    def getSignalsAns(self, signalsDataDict: dict) -> dict:
-        return self.SA.getAns(signalsDataDict)
-    
-    def getRightAns(self,value_1, value_2):
-        
-        ans=''
-        
+        self.TV = TradeVoter(name)
+        self.SA = SignalsAggregator(sig_dict)
+        self.sig_dict = sig_dict
+
+    def get_online_answer(self, signals_ans: dict, price: float) -> dict:
+        """Get trading decision for current market conditions.
+
+        Args:
+            signals_ans: Dictionary of signal data.
+            price: Current market price.
+
+        Returns:
+            Risk-adjusted decision dictionary.
+        """
+        probability_decision = self.TV.get_decision_by_signals(self.get_signals_answer(signals_ans))
+        rmd = self.RM.get_decision(
+            probability_decision=probability_decision,
+            price=price,
+            deals=self.DM.deals
+        )
+        return rmd
+
+    def get_signals_answer(self, signals_data_dict: dict) -> dict:
+        """Get answers from all configured signals.
+
+        Args:
+            signals_data_dict: Dictionary of signal data inputs.
+
+        Returns:
+            Dictionary of signal results.
+        """
+        return self.SA.get_answer(signals_data_dict)
+
+    def get_right_answer(self, value_1: float, value_2: float) -> str:
+        """Determine correct direction based on value comparison.
+
+        Args:
+            value_1: First value (current).
+            value_2: Second value (next).
+
+        Returns:
+            Direction: 'down' if value decreases, 'up' if increases, 'none' if same.
+        """
         if value_1 > value_2:
             ans = 'down'
         elif value_1 < value_2:
             ans = 'up'
         else:
             ans = 'none'
-        
+
         return ans
-        
-    def getRetroTrendAns(self, retroTemplateDict: dict, data: pd.DataFrame(), window: int) -> list:
-        
-        reqSig={} 
+
+    def get_retro_trend_answer(self, retro_template_dict: dict, data: pd.DataFrame, window: int) -> dict:
+        """Run retrospective analysis on historical data.
+
+        Slides a window through historical data to generate training data
+        for probability matrix generation.
+
+        Args:
+            retro_template_dict: Template defining signal structure.
+            data: Historical market data.
+            window: Size of sliding window.
+
+        Returns:
+            Dictionary with 'signalsAns' and 'rightAns' lists.
+        """
+        req_sig = {}
         ans = {
-            'signalsAns':[],
-            'rightAns':[]
-            
+            'signalsAns': [],
+            'rightAns': []
         }
         target = ''
-        
-        
-        for k in tqdm(range(data.shape[0]-window-1)):
-            for i in retroTemplateDict.keys():
-                reqSig[i] = {'signalData': data[k:k+window], 'indicatorData':{}}
-                target = self.SA.signals[i].params['target']
-                for j in retroTemplateDict[i]['indicatorData'].keys():
-                    reqSig[i]['indicatorData'][j] = data[k:k+window]
-                    
-            sigAns = self.getSignalsAns(reqSig)
-            rightAns = self.getRightAns(data[target][k], data[target][k+1])
 
-            ans['signalsAns'].append(sigAns)
-            ans['rightAns'].append(rightAns)
+        for k in tqdm(range(data.shape[0] - window - 1)):
+            for i in retro_template_dict.keys():
+                req_sig[i] = {'signalData': data[k:k+window], 'indicatorData': {}}
+                target = self.SA.signals[i].params['target']
+                for j in retro_template_dict[i]['indicatorData'].keys():
+                    req_sig[i]['indicatorData'][j] = data[k:k+window]
+
+            sig_ans = self.get_signals_answer(req_sig)
+            right_ans = self.get_right_answer(data[target][k], data[target][k+1])
+
+            ans['signalsAns'].append(sig_ans)
+            ans['rightAns'].append(right_ans)
 
         return ans
- 
 
-    def generateMatrixProbabilityFromDict(self, dictSignals: dict) -> dict:
-        self.TV.createMatrixAmounts(dictSignals['signalsAns'][0].keys())
-        for i in range(len(dictSignals['signalsAns'])):
-            self.TV.setDecisionBySignals(signalDecisions = dictSignals['signalsAns'][i],
-                                        trande = dictSignals['rightAns'][i])
-        self.TV.generateMatrixProbability()
-        
-    def createDump(self,postfix='') -> str:
-        dataDict = {
-            'RM':self.RM,
-            'DM':self.DM,
-            'TV':self.TV,
-            'SA':self.SA,
-            'sigDict':self.sigDict
+    def generate_matrix_probability_from_dict(self, dict_signals: dict) -> None:
+        """Generate probability matrices from retrospective signal data.
+
+        Args:
+            dict_signals: Dictionary containing 'signalsAns' and 'rightAns' from retro analysis.
+        """
+        self.TV.create_matrix_amounts(dict_signals['signalsAns'][0].keys())
+        for i in range(len(dict_signals['signalsAns'])):
+            self.TV.set_decision_by_signals(
+                signal_decisions=dict_signals['signalsAns'][i],
+                trande=dict_signals['rightAns'][i]
+            )
+        self.TV.generate_matrix_probability()
+
+    def create_dump(self, postfix: str = '') -> str:
+        """Save decision manager state to pickle file.
+
+        Args:
+            postfix: Optional postfix for filename.
+
+        Returns:
+            Absolute path to saved file.
+        """
+        data_dict = {
+            'RM': self.RM,
+            'DM': self.DM,
+            'TV': self.TV,
+            'SA': self.SA,
+            'sigDict': self.sig_dict
         }
-        fileName='data_'+postfix+'.pickle'
-        with open(fileName, 'wb') as f:
-            pickle.dump(dataDict, f)
-            
-        return os.path.abspath(fileName)
-    
-    def loadDump(self,path: str) -> None:
-        
+        file_name = 'data_' + postfix + '.pickle'
+        with open(file_name, 'wb') as f:
+            pickle.dump(data_dict, f)
+
+        return os.path.abspath(file_name)
+
+    def load_dump(self, path: str) -> None:
+        """Load decision manager state from pickle file.
+
+        Args:
+            path: Path to pickle file.
+        """
         with open(path, 'rb') as f:
-            dataDict = pickle.load(f)
-        
-        self.RM = dataDict['RM']
-        self.DM = dataDict['DM']
-        self.TV = dataDict['TV']
-        self.SA = dataDict['SA']
-        self.sigDict = dataDict['sigDict']
+            data_dict = pickle.load(f)
+
+        self.RM = data_dict['RM']
+        self.DM = data_dict['DM']
+        self.TV = data_dict['TV']
+        self.SA = data_dict['SA']
+        self.sig_dict = data_dict['sigDict']

market_trade/core/riskManager.py

@@ -3,27 +3,54 @@ import datetime
 import numpy as np
 import random
 
-class riskManager:
-    
-    def __init__(self,commision=0.04):
-        self.commision = commision
-        pass
-    def getDecision(self,probabilityDecision, price, deals=None) -> dict:
+
+class RiskManager:
+    """Manages risk assessment and position sizing for trading decisions.
+
+    Evaluates trading decisions from probability-based signals and applies
+    risk management rules including commission calculations and profit targets.
+    """
+
+    def __init__(self, commission: float = 0.04):
+        """Initialize RiskManager with commission rate.
+
+        Args:
+            commission: Commission rate as decimal (default 0.04 = 4%).
+        """
+        self.commission = commission
+
+    def get_decision(self, probability_decision: dict, price: float, deals: pd.DataFrame = None) -> dict:
+        """Evaluate trading decision with risk management rules.
+
+        Args:
+            probability_decision: Dictionary containing 'trande' direction from TradeVoter.
+            price: Current market price.
+            deals: DataFrame of active positions (optional).
+
+        Returns:
+            Dictionary with 'decision' ('buy', 'sell', 'none') and additional fields:
+                - For 'buy': includes 'amount' field
+                - For 'sell': includes 'deals' list of position UUIDs to close
+        """
         ans = {}
         ans['decision'] = 'none'
-        if probabilityDecision['trande'] == 'up':
+
+        if probability_decision['trande'] == 'up':
             ans['decision'] = 'buy'
             ans['amount'] = 1
-        elif probabilityDecision['trande'] == 'none':
+
+        elif probability_decision['trande'] == 'none':
             ans['decision'] = 'none'
-        elif probabilityDecision['trande'] == 'down': 
-            for i in range(deals.shape[0]):
-                ans['decision'] = 'None'
-                ans['deals'] = []
-                row = deals.iloc[i]
-                if row.startPrice < price*pow(1+self.commission,2):
-                    ans['decision'] = 'sell'
-                    ans['deals'].append(row.name)
+
+        elif probability_decision['trande'] == 'down':
+            if deals is not None:
+                for i in range(deals.shape[0]):
+                    ans['decision'] = 'none'
+                    ans['deals'] = []
+                    row = deals.iloc[i]
+                    # Check if position is profitable after commission
+                    if row.startPrice < price * pow(1 + self.commission, 2):
+                        ans['decision'] = 'sell'
+                        ans['deals'].append(row.name)
+
         return ans
-        
-    

market_trade/core/trandeVoter.py

@@ -3,82 +3,72 @@ import datetime
 import numpy as np
 #import random
 
-class trandeVoter():
-    
-    def __init__(self,name):
-       
-        self.name = name # просто имя
-        self.trandeValuesList = ['up','none','down'] #словарь трегдов
-        self.matrixAmounts = None # матрица сумм
-        self.keysMatrixAmounts = None #ключи матрицы сумм, техническое поле
-        self.matrixProbability = None # матрица вероятностей
-        
-    
-    #функция которая создает df с заданным набором колонок и индексов. индексы - уникальные соотношения 
-    def createDFbyNames(self, namesIndex, namesColoms,defaultValue=0.0):
-        df = pd.DataFrame(dict.fromkeys(namesColoms, [defaultValue]*pow(3,len(namesIndex))),
-                          index=pd.MultiIndex.from_product([self.trandeValuesList]*len(namesIndex), names=namesIndex)
-                            #,columns=namesColoms
+class TradeVoter():
+
+    def __init__(self, name):
+
+        self.name = name  # Instance identifier
+        self.trade_values_list = ['up', 'none', 'down']  # Valid trade directions
+        self.matrix_amounts = None  # Sum matrix for signal combinations
+        self.keys_matrix_amounts = None  # Matrix keys, technical field
+        self.matrix_probability = None  # Probability matrix for decision making
+
+
+    # Function to create DataFrame with specified columns and indices. Indices are unique combinations.
+    def create_df_by_names(self, names_index, column_names, default_value=0.0):
+        df = pd.DataFrame(dict.fromkeys(column_names, [default_value]*pow(3, len(names_index))),
+                          index=pd.MultiIndex.from_product([self.trade_values_list]*len(names_index), names=names_index)
                         )
-        return(df)
-    
-    #создание матрицы сумм с дефолтным значением
-    def createMatrixAmounts(self,namesIndex: list) -> pd.DataFrame():
-        self.matrixAmounts = self.createDFbyNames(namesIndex,self.trandeValuesList,0)
-        self.keysMatrixAmounts = self.matrixAmounts.to_dict('tight')['index_names']
-        self.createMatrixProbability(namesIndex)
-        return(self.matrixAmounts)
-    
-    #создание матрицы вероятностей с дефолтным значением
-    def createMatrixProbability(self,namesIndex: list) -> pd.DataFrame():
-        self.matrixProbability = self.createDFbyNames(namesIndex,self.trandeValuesList)
-        return(self.matrixProbability)
-    
-    #установка значений в матрицы сумм. signalDecisions - значения индикаторов key:value; trande - реальное значение
-    def setDecisionBySignals(self,signalDecisions: dict,trande: str) -> None:
-        buff=[]
-        for i in self.keysMatrixAmounts:
-            buff.append(signalDecisions[i])
-        self.matrixAmounts.loc[tuple(buff),trande] += 1
-        
-    #заполнение матрицы вероятностей вычисляемыми значениями из матрицы сумм
-    def generateMatrixProbability(self) -> None:
-        for i in range(self.matrixAmounts.shape[0]):
-            print(self.matrixAmounts)
-            rowSum=sum(self.matrixAmounts.iloc[i]) + 1
-            self.matrixProbability.iloc[i]['up'] = self.matrixAmounts.iloc[i]['up'] / rowSum
-            self.matrixProbability.iloc[i]['none'] = self.matrixAmounts.iloc[i]['none'] / rowSum
-            self.matrixProbability.iloc[i]['down'] = self.matrixAmounts.iloc[i]['down'] / rowSum
+        return df
 
-    #получение рещения из матрицы вероятностей по заданным значениям сигналов
-    def getDecisionBySignals(self,signalDecisions: dict) -> dict:
+    # Create sum matrix with default value
+    def create_matrix_amounts(self, names_index: list) -> pd.DataFrame:
+        self.matrix_amounts = self.create_df_by_names(names_index, self.trade_values_list, 0)
+        self.keys_matrix_amounts = self.matrix_amounts.to_dict('tight')['index_names']
+        self.create_matrix_probability(names_index)
+        return self.matrix_amounts
+
+    # Create probability matrix with default value
+    def create_matrix_probability(self, names_index: list) -> pd.DataFrame:
+        self.matrix_probability = self.create_df_by_names(names_index, self.trade_values_list)
+        return self.matrix_probability
+
+    # Set values in sum matrix. signalDecisions - indicator values key:value; trande - actual value
+    def set_decision_by_signals(self, signal_decisions: dict, trande: str) -> None:
+        buff = []
+        for i in self.keys_matrix_amounts:
+            buff.append(signal_decisions[i])
+        self.matrix_amounts.loc[tuple(buff), trande] += 1
+
+    # Fill probability matrix with calculated values from sum matrix
+    def generate_matrix_probability(self) -> None:
+        for i in range(self.matrix_amounts.shape[0]):
+            print(self.matrix_amounts)
+            row_sum = sum(self.matrix_amounts.iloc[i]) + 1
+            self.matrix_probability.iloc[i]['up'] = self.matrix_amounts.iloc[i]['up'] / row_sum
+            self.matrix_probability.iloc[i]['none'] = self.matrix_amounts.iloc[i]['none'] / row_sum
+            self.matrix_probability.iloc[i]['down'] = self.matrix_amounts.iloc[i]['down'] / row_sum
+
+    # Get decision from probability matrix based on signal values
+    def get_decision_by_signals(self, signal_decisions: dict) -> dict:
         ans = {}
-        spliceSearch =self.matrixProbability.xs(tuple(signalDecisions.values()),
-                                                       level=list(signalDecisions.keys())
-                                                      ) 
-        ans['probability'] = spliceSearch.to_dict('records')[0]
-        ans['trande'] = spliceSearch.iloc[0].idxmax()
+        splice_search = self.matrix_probability.xs(tuple(signal_decisions.values()),
+                                                       level=list(signal_decisions.keys())
+                                                      )
+        ans['probability'] = splice_search.to_dict('records')[0]
+        ans['trande'] = splice_search.iloc[0].idxmax()
         return ans
-    
-    #получение матриц вероятностей и суммы в видей словарей
-    def getMatrixDict(self) -> dict:
-        ans={}
-        ans['amounts'] = self.matrixAmounts.to_dict('tight')
-        ans['probability'] = self.matrixProbability.to_dict('tight')
+
+    # Get probability and sum matrices as dictionaries
+    def get_matrix_dict(self) -> dict:
+        ans = {}
+        ans['amounts'] = self.matrix_amounts.to_dict('tight')
+        ans['probability'] = self.matrix_probability.to_dict('tight')
         return ans
-    
-    #установка матриц вероятностей и суммы в видей словарей
-    def setMatrixDict(self,matrixDict: dict) -> dict:
-        if matrixDict['amounts'] != None:
-            self.matrixAmounts = pd.DataFrame.from_dict(y['amounts'], orient='tight')
-        if matrixDict['probability'] != None:
-            self.matrixProbability = pd.DataFrame.from_dict(y['probability'], orient='tight') 
-        
-        
 
-    
-    
-    
-    
-
-        
+    # Set probability and sum matrices from dictionaries
+    def set_matrix_dict(self, matrix_dict: dict) -> dict:
+        if matrix_dict['amounts'] != None:
+            self.matrix_amounts = pd.DataFrame.from_dict(y['amounts'], orient='tight')
+        if matrix_dict['probability'] != None:
+            self.matrix_probability = pd.DataFrame.from_dict(y['probability'], orient='tight')