use single reqrep socket remove unused fuctions add symbol list

Author: traderpedroso

ejtraderMT/api/mql.py

@@ -35,16 +35,9 @@ class Functions:
     def __init__(self, host=None, debug=None):
         self.HOST = host or "localhost"
         self.SYS_PORT = 15555  # REP/REQ port
-        self.DATA_PORT = 15556  # PUSH/PULL port
-        self.LIVE_PORT = 15557  # PUSH/PULL port
-        self.EVENTS_PORT = 15558  # PUSH/PULL port
-        self.INDICATOR_DATA_PORT = 15559  # REP/REQ port
-        self.CHART_DATA_PORT = 15560  # PUSH
-        self.debug = debug or True
 
         # ZeroMQ timeout in seconds
-        sys_timeout = 1000 * 1000
-        data_timeout = 1000 * 1000
+        sys_timeout = 1000
 
         # initialise ZMQ context
         context = zmq.Context()
@@ -56,44 +49,17 @@ def __init__(self, host=None, debug=None):
             self.sys_socket.RCVTIMEO = sys_timeout
             self.sys_socket.connect("tcp://{}:{}".format(self.HOST, self.SYS_PORT))
 
-            self.data_socket = context.socket(zmq.PULL)
-            # set port timeout
-            self.data_socket.RCVTIMEO = data_timeout
-            self.data_socket.connect("tcp://{}:{}".format(self.HOST, self.DATA_PORT))
-
-            self.indicator_data_socket = context.socket(zmq.PULL)
-            # set port timeout
-            self.indicator_data_socket.RCVTIMEO = data_timeout
-            self.indicator_data_socket.connect(
-                "tcp://{}:{}".format(self.HOST, self.INDICATOR_DATA_PORT)
-            )
-            self.chart_data_socket = context.socket(zmq.PUSH)
-            # set port timeout
-            # TODO check if port is listening and error handling
-            self.chart_data_socket.connect(
-                "tcp://{}:{}".format(self.HOST, self.CHART_DATA_PORT)
-            )
-
         except zmq.ZMQError:
             raise zmq.ZMQBindError("Binding ports ERROR")
         except KeyboardInterrupt:
             self.sys_socket.close()
             self.sys_socket.term()
-            self.data_socket.close()
-            self.data_socket.term()
-            self.indicator_data_socket.close()
-            self.indicator_data_socket.term()
-            self.chart_data_socket.close()
-            self.chart_data_socket.term()
             pass
 
     def _send_request(self, data: dict) -> None:
         """Send request to server via ZeroMQ System socket"""
         try:
             self.sys_socket.send_json(data)
-            msg = self.sys_socket.recv_string()
-            # terminal received the request
-            assert msg == "OK", "Something wrong on server side"
         except AssertionError as err:
             raise zmq.NotDone(err)
         except zmq.ZMQError:
@@ -102,50 +68,11 @@ def _send_request(self, data: dict) -> None:
     def _pull_reply(self):
         """Get reply from server via Data socket with timeout"""
         try:
-            msg = self.data_socket.recv_json()
+            msg = self.sys_socket.recv_json()
         except zmq.ZMQError:
             raise zmq.NotDone("Data socket timeout ERROR")
         return msg
 
-    def _indicator_pull_reply(self):
-        """Get reply from server via Data socket with timeout"""
-        try:
-            msg = self.indicator_data_socket.recv_json()
-        except zmq.ZMQError:
-            raise zmq.NotDone("Indicator Data socket timeout ERROR")
-        if self.debug:
-            logging.info("ZMQ INDICATOR DATA REPLY: ", msg)
-        return msg
-
-    def live_socket(self, context=None):
-        """Connect to socket in a ZMQ context"""
-        try:
-            context = context or zmq.Context.instance()
-            socket = context.socket(zmq.PULL)
-            socket.connect("tcp://{}:{}".format(self.HOST, self.LIVE_PORT))
-        except zmq.ZMQError:
-            raise zmq.ZMQBindError("Live port connection ERROR")
-        return socket
-
-    def streaming_socket(self, context=None):
-        """Connect to socket in a ZMQ context"""
-        try:
-            context = context or zmq.Context.instance()
-            socket = context.socket(zmq.PULL)
-            socket.connect("tcp://{}:{}".format(self.HOST, self.EVENTS_PORT))
-        except zmq.ZMQError:
-            raise zmq.ZMQBindError("Data port connection ERROR")
-        return socket
-
-    def _push_chart_data(self, data: dict) -> None:
-        """Send message for chart control to server via ZeroMQ chart data socket"""
-        try:
-            if self.debug:
-                logging.info("ZMQ PUSH CHART DATA: ", data, " -> ", data)
-            self.chart_data_socket.send_json(data)
-        except zmq.ZMQError:
-            raise zmq.NotDone("Sending request ERROR")
-
     def Command(self, **kwargs) -> dict:
         """Construct a request dictionary from default and send it to server"""
 
@@ -185,58 +112,6 @@ def Command(self, **kwargs) -> dict:
         # return server reply
         return self._pull_reply()
 
-    def indicator_construct_and_send(self, **kwargs) -> dict:
-        """Construct a request dictionary from default and send it to server"""
-
-        # default dictionary
-        request = {
-            "action": None,
-            "actionType": None,
-            "id": None,
-            "symbol": None,
-            "chartTF": None,
-            "fromDate": None,
-            "toDate": None,
-            "name": None,
-            "params": None,
-            "linecount": None,
-        }
-
-        # update dict values if exist
-        for key, value in kwargs.items():
-            if key in request:
-                request[key] = value
-            else:
-                raise KeyError("Unknown key in **kwargs ERROR")
-
-        # send dict to server
-        self._send_request(request)
-
-        # return server reply
-        return self._indicator_pull_reply()
-
-    def chart_data_construct_and_send(self, **kwargs) -> dict:
-        """Construct a request dictionary from default and send it to server"""
-
-        # default dictionary
-        message = {
-            "action": None,
-            "actionType": None,
-            "chartId": None,
-            "indicatorChartId": None,
-            "data": None,
-        }
-
-        # update dict values if exist
-        for key, value in kwargs.items():
-            if key in message:
-                message[key] = value
-            else:
-                raise KeyError("Unknown key in **kwargs ERROR")
-
-        # send dict to server
-        self._push_chart_data(message)
-
 
 class Metatrader:
     def __init__(
@@ -290,7 +165,6 @@ def symbols(self):
         except Exception as e:
             logging.info(f"Error while processing. Error message: {str(e)}")
 
-        self.__api.Command(action="RESET")
         if df is not None and isinstance(df, dict):
             try:
                 if df["data"]:
@@ -349,63 +223,48 @@ def _calendar(self, data):
             toDate = start_date
             toDate += delta2
             toDate = toDate.strftime("%d/%m/%Y")
-            attempts = 0
-            success = False
 
-            while not success and attempts < 5:
-                try:
-                    df = self.__api.Command(
-                        action="CALENDAR",
-                        actionType="DATA",
-                        symbol=symbol,
-                        fromDate=self.__date_to_timestamp(fromDate),
-                        toDate=self.__date_to_timestamp(toDate),
-                    )
-                    success = True
+            try:
+                df = self.__api.Command(
+                    action="CALENDAR",
+                    actionType="DATA",
+                    symbol=symbol,
+                    fromDate=self.__date_to_timestamp(fromDate),
+                    toDate=self.__date_to_timestamp(toDate),
+                )
+            except Exception as e:
+                logging.info(
+                    f"Error while processing {symbol}. Error message: {str(e)}"
+                )
+                pass
 
-                except Exception as e:
-                    logging.info(
-                        f"Error while processing {symbol}. Error message: {str(e)}"
-                    )
-                    attempts += 1
-            if attempts == 5 and not success:
-                # print(f"Check if {symbol} is avalible from {fromDate}")
+            try:
+                df = pd.DataFrame(df["data"])
+                df.columns = [
+                    "date",
+                    "currency",
+                    "impact",
+                    "event",
+                    "country",
+                    "actual",
+                    "forecast",
+                    "previous",
+                ]
+                df["date"] = pd.to_datetime(df["date"], errors="coerce")
+                df = df.dropna(subset=["date"])
+                df = df.set_index("date")
+                df.index = pd.to_datetime(df.index)
+            except Exception as e:
+                logging.info(
+                    f"Error while processing {symbol} Dataframe. Error message: {str(e)}"
+                )
                 pass
-            self.__api.Command(action="RESET")
-            if df is not None and isinstance(df, dict):
-                try:
-                    if df["data"]:
-                        df = pd.DataFrame(df["data"])
-                        df.columns = [
-                            "date",
-                            "currency",
-                            "impact",
-                            "event",
-                            "country",
-                            "actual",
-                            "forecast",
-                            "previous",
-                        ]
-                        df["date"] = pd.to_datetime(df["date"], errors="coerce")
-                        df = df.dropna(subset=["date"])
-                        df = df.set_index("date")
-                        df.index = pd.to_datetime(df.index)
-                except Exception as e:
-                    logging.info(
-                        f"Error while processing {symbol} Dataframe. Error message: {str(e)}"
-                    )
-                    pass
 
             appended_data.append(df)
             start_date += delta
         pbar.close()
-        try:
-            df = pd.concat(appended_data)
-        except Exception as e:
-            logging.info(
-                f"Error while processing {symbol} Dataframe. Error message: {str(e)}"
-            )
-            pass
+
+        df = pd.concat(appended_data)
 
         if self.__database:
             start(self.__save_to_db, data=[df], repeat=1, max_threads=20)
@@ -608,7 +467,6 @@ def _event(self):
                 pass
 
     def price(self, symbol, chartTF):
-        self.__api.Command(action="RESET")
         self._allsymbol_ = symbol
         self._allchartTF = chartTF
         for active in symbol:
@@ -618,7 +476,6 @@ def price(self, symbol, chartTF):
         return self.__priceQ.get()
 
     def event(self, symbol, chartTF):
-        self.__api.Command(action="RESET")
         self._allsymbol_ = symbol
         self._allchartTF = chartTF
         for active in symbol:
@@ -816,14 +673,6 @@ def __historyThread_save(self, data):
                         logging.info(f"Check if {active} is avalible from {fromDate}")
                         pass
 
-                    try:
-                        self.__api.Command(action="RESET")
-                    except Exception as e:
-                        logging.info(
-                            f"Error while processing RESET. Error message: {str(e)}"
-                        )
-                        pass
-
                     if data is not None and isinstance(data, dict):
                         try:
                             if data["data"]:
@@ -875,13 +724,6 @@ def __historyThread_save(self, data):
                         logging.info(f"Check if {active} is avalible from {fromDate}")
                         pass
 
-                    try:
-                        self.__api.Command(action="RESET")
-                    except Exception as e:
-                        logging.info(
-                            f"Error while processing commad Reset {active}. Error message: {str(e)}"
-                        )
-                        pass
                     if data is not None and isinstance(data, dict):
                         try:
                             if data["data"]: