Issue #67: Black linter refactoring

.github/workflows/linter.yml

@@ -21,4 +21,6 @@ jobs:
           GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
           DEFAULT_BRANCH: develop
           # DISABLE_ERRORS: true    # avoid crashing PR
-          # VALIDATE_ALL_CODEBASE: false    # check only new changes
+          # VALIDATE_ALL_CODEBASE: false    # check only new changes
+        with:
+          BLACK_LINE_LENGTH: 120

analyst/__init__.py

@@ -3,4 +3,4 @@
 Analyse all predictions and make decision whether to make purchase or not.
 Get information from Prediction Module and give issues commands to Trading Module.
 
-"""
+"""

analyst/analyst_service.py

@@ -18,13 +18,11 @@ def analyse_threshold(self, vector):
         return vector >= self.threshold
 
     def make_trading_decision(self, vector):
-        """ Decide if purchasing cryptocurrency is profitable based on the probability of price growth.
+        """Decide if purchasing cryptocurrency is profitable based on the probability of price growth.
 
         :arg vector: probability of cryptocurrency price growth.
         :return: True if it is profitable to buy/keep crypto, False otherwise (to sell).
         """
         decision = self.analyse_threshold(vector)
         self.logger.info(f"Vector: {vector}, Decision: {decision}")
         return decision
-
-

logs/clear_logs.py

@@ -3,11 +3,11 @@
 import os
 
 
-def clear_all_logs(log_dir='./'):
+def clear_all_logs(log_dir="./"):
     files = os.listdir(log_dir)
 
     for file in files:
-        if file.endswith('.log'):
+        if file.endswith(".log"):
             os.remove(os.path.join(log_dir, file))
 
 

predictor/model_testing/testing.py

@@ -7,35 +7,36 @@
 class ModelTesting:
     def __init__(self, data_set, train_coefficient, output):
         self.data_set = data_set
-        self.testing_data = data_set[int(len(data_set) * train_coefficient):]
+        self.testing_data = data_set[int(len(data_set) * train_coefficient) :]
 
         self.train_coefficient = train_coefficient
         self.output_column = output
         self.lines = []
 
-        self._add_line(self.data_set[self.output_column], 'real_data')
+        self._add_line(self.data_set[self.output_column], "real_data")
 
     def _add_line(self, data_set, name: str):
-        line1, = plt.plot(data_set, label=name)
+        (line1,) = plt.plot(data_set, label=name)
         self.lines.append(line1)
 
     def add_model(self, model, input_layers: list):
-        test_x, test_labels = reconstruct_data(np.array(self.testing_data[input_layers]), 
-                                               np.array(self.testing_data[self.output_column]), 
-                                               model.NUM_OF_PREV_ITEMS)
-
+        test_x, test_labels = reconstruct_data(
+            np.array(self.testing_data[input_layers]),
+            np.array(self.testing_data[self.output_column]),
+            model.NUM_OF_PREV_ITEMS,
+        )
+
         test_predict = model.make_prediction(test_x)
 
         test_score = mean_squared_error(test_labels, test_predict)
-        print(f'[{model.__class__.__name__}|{model.model_name}] Score on test set: {test_score} MSE')
+        print(f"[{model.__class__.__name__}|{model.model_name}] Score on test set: {test_score} MSE")
 
         number_of_unlabeled_data = int(len(self.data_set) * self.train_coefficient) + 1 + model.NUM_OF_PREV_ITEMS
-        predict_plot = np.array([[np.nan]] * number_of_unlabeled_data).astype('float32')
-        predict_plot = np.concatenate((predict_plot.astype('float32'), 
-                                                test_predict.astype('float32')))
-
+        predict_plot = np.array([[np.nan]] * number_of_unlabeled_data).astype("float32")
+        predict_plot = np.concatenate((predict_plot.astype("float32"), test_predict.astype("float32")))
+
         self._add_line(predict_plot, model.model_name)
 
     def show_graph(self):
         plt.legend(handles=self.lines)
-        plt.show() # TODO show date on axis
+        plt.show()  # TODO show date on axis

predictor/models/data_processing.py

@@ -1,9 +1,9 @@
 import numpy as np
 
-def reconstruct_data(input_data, output_data, n: int) -> (np.array, np.array):
 
+def reconstruct_data(input_data, output_data, n: int) -> (np.array, np.array):
     """
-    Match each element in output_data (except the first n element) 
+    Match each element in output_data (except the first n element)
     to list of previous n elements of input_data.
 
     The data will be reshaped so network could see what output should be to the given input data.
@@ -12,12 +12,12 @@ def reconstruct_data(input_data, output_data, n: int) -> (np.array, np.array):
     the length of input_data should be = length of output_data.
     Input and output data could be the same or different.
     Input data could contain multiple or single features.
-    
+
 
     For example - input_data = output_data = [[1], [2], [3], [4], [5], [6], [7], [8]], n=3
     The result - [1, 2, 3] -> 4; [2, 3, 4] -> [5] ...
     The result data will be returned in form:
-    (array([[1, 2, 3], [2, 3, 4], [3, 4, 5], [4, 5, 6], [5, 6, 7]]), 
+    (array([[1, 2, 3], [2, 3, 4], [3, 4, 5], [4, 5, 6], [5, 6, 7]]),
     array([4, 5, 6, 7, 8]))
 
     :param input_data: array
@@ -29,14 +29,14 @@ def reconstruct_data(input_data, output_data, n: int) -> (np.array, np.array):
     x, y = [], []
 
     for i in range(len(input_data) - n):
-        a = input_data[i:(i + n)]
+        a = input_data[i : (i + n)]
         x.append(a)
         y.append(output_data[i + n])
 
     return np.array(x), np.array(y)
 
 
 def train_test_split(data, train_cof):
-    train, test = data[0:int(len(data) * train_cof), :], data[int(len(data) * train_cof):len(data), :]
+    train, test = data[0 : int(len(data) * train_cof), :], data[int(len(data) * train_cof) : len(data), :]
 
     return train, test

predictor/models/model_handler.py

@@ -4,7 +4,7 @@
 
 
 class ModelHandler(ABC):
-    PATH_TO_CONF = os.path.join(os.path.dirname(__file__), 'fortune-nn-configs')
+    PATH_TO_CONF = os.path.join(os.path.dirname(__file__), "fortune-nn-configs")
 
     def __init__(self):
         super().__init__()
@@ -19,7 +19,7 @@ def make_prediction(self, data_set: list) -> list:
 
 
 class ModelTrainer(ABC):
-    PATH_TO_CONF = os.path.join(os.path.dirname(__file__), 'fortune-nn-configs')
+    PATH_TO_CONF = os.path.join(os.path.dirname(__file__), "fortune-nn-configs")
 
     def __init__(self, input_shape: tuple):
         super().__init__()

predictor/models/sf1/snowfall_model.py

@@ -7,25 +7,24 @@
 
 
 class Snowfall(ModelHandler):
-    def __init__(self, model_name='model_15m_50:1_c-c'):
+    def __init__(self, model_name="model_15m_50:1_c-c"):
         super().__init__()
         self.model_name = model_name
 
         path_to_model = os.path.join(self.PATH_TO_CONF, model_name)
         self.model = keras.models.load_model(path_to_model)
 
-        with open(os.path.join(ModelHandler.PATH_TO_CONF, model_name, 'input_scaler'),'rb') as f:
+        with open(os.path.join(ModelHandler.PATH_TO_CONF, model_name, "input_scaler"), "rb") as f:
             self.input_scaler = pickle.load(f)
 
-        with open(os.path.join(ModelHandler.PATH_TO_CONF, model_name, 'output_scaler'),'rb') as f:
+        with open(os.path.join(ModelHandler.PATH_TO_CONF, model_name, "output_scaler"), "rb") as f:
             self.output_scaler = pickle.load(f)
 
         self.input_shape = self.model.layers[0].input_shape[1:]
 
         self.NUM_OF_PREV_ITEMS = self.model.layers[0].input_shape[1]
 
     def make_prediction(self, data_set):
-
         n_data_set = np.reshape(data_set, (-1, self.input_shape[1]))
 
         # min-max normalization (inverse to (0, 1) range)

predictor/models/sf1/snowfall_training.py

@@ -12,6 +12,7 @@
 from predictor.models.data_processing import reconstruct_data
 from predictor.models.model_handler import ModelTrainer
 
+
 class SnowfallTestTrain(ModelTrainer):
     def __init__(self, shape):
         super().__init__(shape)
@@ -43,16 +44,16 @@ def train(self, train_data, label_data):
 
         train_x, train_y = reconstruct_data(data_transformed, label_transformed, self.NUM_OF_PREV_ITEMS)
 
-        self.model.compile(loss='mean_squared_error', optimizer='adam')
-        self.model.fit(train_x, train_y, epochs=self.epochs, batch_size=self.batch_size , verbose=self.verbose)
+        self.model.compile(loss="mean_squared_error", optimizer="adam")
+        self.model.fit(train_x, train_y, epochs=self.epochs, batch_size=self.batch_size, verbose=self.verbose)
 
     def save_model(self, path):
         self.model.save(path)
 
-        with open(os.path.join(path, 'input_scaler'), 'wb') as save_file:
-            pickle.dump(self.input_scaler, save_file)      
+        with open(os.path.join(path, "input_scaler"), "wb") as save_file:
+            pickle.dump(self.input_scaler, save_file)
+
+        with open(os.path.join(path, "output_scaler"), "wb") as save_file:
+            pickle.dump(self.output_scaler, save_file)
 
-        with open(os.path.join(path, 'output_scaler'), 'wb') as save_file:
-            pickle.dump(self.output_scaler, save_file)                    
-
         print(f"Model 'sf' saved to - {path}")

predictor/predictor_service.py

@@ -14,7 +14,7 @@ def predict(self, data: list) -> int:
         """
         Make predictions using a list of data.
 
-        This method uses the `model_handler` to predict the next number based on the input `data`. 
+        This method uses the `model_handler` to predict the next number based on the input `data`.
         The input `data` should be of the correct size, which can be accessed using `self.model_handler.NUM_OF_PREV_ITEMS`.
 
         :param data: list of numbers of correct size
@@ -23,6 +23,6 @@ def predict(self, data: list) -> int:
         if np.shape(data)[0] == self.model_handler.NUM_OF_PREV_ITEMS:
             return self.model_handler.predict_next(data)
         else:
-            massage = f"The number of previous items should be {self.model_handler.NUM_OF_PREV_ITEMS}, but not - {np.shape(data)[0]}"         
+            massage = f"The number of previous items should be {self.model_handler.NUM_OF_PREV_ITEMS}, but not - {np.shape(data)[0]}"
             self.logger.error(massage)
             return False

project.py

@@ -17,13 +17,13 @@
 
 
 class Fortune:
-    CONFIG_PATH = './src/config/'
+    CONFIG_PATH = "./src/config/"
 
     def __init__(self, **kwargs):
-        log_setup.configurate_logs(self.CONFIG_PATH + 'log_config.yml')
+        log_setup.configurate_logs(self.CONFIG_PATH + "log_config.yml")
         self.logger = logging.getLogger(__class__.__name__)
         self.exit_flag = asyncio.Event()
-        self.run_pigamma = kwargs['pigamma']
+        self.run_pigamma = kwargs["pigamma"]
 
         self.client = api_binance.configure_binance_api(self.CONFIG_PATH + api_binance.API.CLIENT_CONFIG_FILE)
         self.trader = trader_service.Trader()
@@ -35,9 +35,7 @@ def __init__(self, **kwargs):
 
     def configure_pigamma_wrapper(self):
         discord_bot.configure_pigamma(
-            self.CONFIG_PATH + discord_bot.PiGamma.CONFIG_FILE,
-            self.stats_queue,
-            self.exit_flag
+            self.CONFIG_PATH + discord_bot.PiGamma.CONFIG_FILE, self.stats_queue, self.exit_flag
         )
 
     def process_iteration(self):

src/__init__.py

@@ -10,5 +10,4 @@
 import src.generate_config
 import src.api_binance
 
-print("*** ARA Development | Fortune ***\n"
-      "All rights reserved!")
+print("*** ARA Development | Fortune ***\n" "All rights reserved!")

src/api_binance.py

@@ -18,14 +18,14 @@
 
 def convert_timestamp_to_str(timestamp):
     dt_object = datetime.date.fromtimestamp(timestamp / 1000)
-    return dt_object.strftime('%-d %b %Y')
+    return dt_object.strftime("%-d %b %Y")
 
 
 class API(Client):
-    """ Binance API """
+    """Binance API"""
 
-    CLIENT_CONFIG_FILE = 'api_config.json'
-    BTCUSDT = 'BTCUSDT'
+    CLIENT_CONFIG_FILE = "api_config.json"
+    BTCUSDT = "BTCUSDT"
 
     def __init__(self, api_key, api_secret):
         """
@@ -47,10 +47,10 @@ def _get_new_price(self, symbol):
             self.logger.error(f"Unable to retrieve the latest price!\n{exception}")
             return None
         else:
-            return response_['price']
+            return response_["price"]
 
     def _update_price(self, symbol, interval):
-        """ Retrieves the latest price for a given trading symbol and puts it into the price_queue.
+        """Retrieves the latest price for a given trading symbol and puts it into the price_queue.
 
         :param symbol: The trading pair symbol (e.g., 'BTCUSDT') for which to retrieve the price.
         :param interval: The interval length in minutes for fetching the klines data.
@@ -67,7 +67,7 @@ def _update_price(self, symbol, interval):
             time.sleep(interval_in_seconds)
 
     def launch_price_update_subprocess(self, symbol, interval):
-        """ Launches a subprocess to update the price for the given symbol at the specified interval.
+        """Launches a subprocess to update the price for the given symbol at the specified interval.
 
         This function creates a new subprocess to continuously update the price for the specified trading symbol
         at the given interval. The subprocess runs the `_update_price` method internally.
@@ -86,7 +86,7 @@ def launch_price_update_subprocess(self, symbol, interval):
         self.logger.info("Price update subprocess was launched")
 
     def terminate_price_update_subprocess(self):
-        """ Terminates the subprocess responsible for updating the price.
+        """Terminates the subprocess responsible for updating the price.
 
         This function terminates the subprocess that is responsible for updating the price of a trading symbol.
         If no subprocess is currently running, the function returns without taking any action.
@@ -107,7 +107,7 @@ def terminate_price_update_subprocess(self):
         self.logger.info("Price update subprocess was terminated.")
 
     def await_price_update(self):
-        """ Waits for a price update by retrieving the latest price from the price queue.
+        """Waits for a price update by retrieving the latest price from the price queue.
 
         If the price queue is not empty, this function retrieves and discards all existing prices
         until the queue becomes empty. If the price queue is empty, the function waits and blocks until
@@ -127,7 +127,7 @@ def await_price_update(self):
         return float(price)
 
     def load_price_history(self, symbol, interval):
-        """ Loads historical klines data for a specified symbol and time interval.
+        """Loads historical klines data for a specified symbol and time interval.
 
         This function retrieves historical klines data for the specified trading symbol with the provided time interval.
         The data is fetched starting from '1 Jan 2000' up to the current date and time.
@@ -141,7 +141,7 @@ def load_price_history(self, symbol, interval):
         """
 
         all_data = []
-        end_date = datetime.datetime.now().strftime('%-d %b %Y')
+        end_date = datetime.datetime.now().strftime("%-d %b %Y")
         limit = 1000
 
         while True:
@@ -173,7 +173,7 @@ def _check_column_for_duplicates(self, data, column_index):
         self.logger.info("No data duplication was detected")
 
     def save_price_history_csv(self, symbol, interval, file_path):
-        """ Save price history data to CSV
+        """Save price history data to CSV
 
         :param symbol: The trading pair symbol (e.g., 'BTCUSDT') for which to load historical data.
         :param interval: The time interval for the klines data (e.g., Client.KLINE_INTERVAL_15MINUTE).
@@ -184,7 +184,7 @@ def save_price_history_csv(self, symbol, interval, file_path):
         data = self.load_price_history(symbol, interval)
         self._check_column_for_duplicates(data, 0)
 
-        with open(file_path, 'w', newline='') as csv_file:
+        with open(file_path, "w", newline="") as csv_file:
             csv_writer = csv.writer(csv_file)
             csv_writer.writerow(kline_metric)
             csv_writer.writerows(data)
@@ -201,7 +201,7 @@ def load_last_prices(self, symbol, interval, limit=5):
 
 
 def configure_binance_api(config_file):
-    """ Configures the Binance API client.
+    """Configures the Binance API client.
 
     This function reads the API configuration data from the specified file, creates a Binance API client instance,
     and returns it.
@@ -217,8 +217,8 @@ def configure_binance_api(config_file):
     return client
 
 
-if __name__ == '__main__':
-    api = configure_binance_api('./config/api_config.json')
+if __name__ == "__main__":
+    api = configure_binance_api("./config/api_config.json")
     prices = api.load_price_history(api.BTCUSDT, Client.KLINE_INTERVAL_15MINUTE)
     api.save_price_history_csv(api.BTCUSDT, api.KLINE_INTERVAL_15MINUTE, "./tmp.csv")
     print(len(prices))