exception handling in user code (not ready)

Author: oliver-zehentleitner

README.md

@@ -177,6 +177,16 @@ ubwa.api.cancel_order(orig_client_order_id=orig_client_order_id, symbol="BUSDUSD
 guide on 
 [how to process requests via the Binance WebSocket API](https://medium.lucit.tech/create-and-cancel-orders-via-websocket-on-binance-7f828831404)!
 
+## STREAM SIGNALS - know the state of your streams
+Usually you want to know when a stream is working and when it is not. This can be useful to know that your own system is 
+currently "blind" and you may want to close open positions to be on the safe side, know that indicators will now provide 
+incorrect values or that you have to reload the missing data via REST as an alternative. 
+
+For this purpose, the UNICORN Binance WebSocket API provides so-called 
+[STREAM SIGNALS](https://github.com/LUCIT-Systems-and-Development/unicorn-binance-websocket-api/wiki/%60stream_signals%60)
+, which are used to tell your code in real time when a stream is connected, when it received its first data record, when 
+it was disconnected and stopped, and when the stream cannot be restored.
+
 ## Stop `ubwa` after usage to avoid memory leaks
 
 ```

dev/test_stream_signals.py

@@ -0,0 +1,17 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+from unicorn_binance_websocket_api import BinanceWebSocketApiManager
+import time
+
+
+def receive_stream_signal(self, signal_type=None, stream_id=None, data_record=None, error_msg=None):
+    print(f"Received stream_signal for stream '{ubwa.get_stream_label(stream_id=stream_id)}': "
+          f"{signal_type} - {stream_id} - {data_record} - {error_msg}")
+
+
+with BinanceWebSocketApiManager(process_stream_signals=receive_stream_signal) as ubwa:
+    ubwa.create_stream(channels="trade", markets="btcusdt", stream_label="HEALTHY")
+    print(f"Waiting 5 seconds and then stop the stream ...")
+    time.sleep(5)
+

examples/binance_websocket_stream_signals/README.md

@@ -0,0 +1,40 @@
+# Binance WebSocket Stream Signals
+## Overview
+Usually you want to know when a stream is working and when it is not. This can be useful to know that your own system is 
+currently "blind" and you may want to close open positions to be on the safe side, know that indicators will now provide 
+incorrect values or that you have to reload the missing data via REST as an alternative. 
+
+For this purpose, the UNICORN Binance WebSocket API provides so-called 
+[STREAM SIGNALS](https://github.com/LUCIT-Systems-and-Development/unicorn-binance-websocket-api/wiki/%60stream_signals%60)
+, which are used to tell your code in real time when a stream is connected, when it received its first data record, when 
+it was disconnected and stopped, and when the stream cannot be restored.
+
+In this example, a stream is started and stopped. To keep the example lean and clear, we will not process any data, but 
+only activate and process the stream signals.
+
+## Prerequisites
+Ensure you have Python 3.7+ installed on your system. 
+
+Before running the provided script, install the required Python packages:
+```bash
+pip install -r requirements.txt
+```
+## Get a UNICORN Binance Suite License
+To run modules of the *UNICORN Binance Suite* you need a [valid license](https://shop.lucit.services)!
+
+## Usage
+### Running the Script:
+```bash
+python binance_websocket_stream_signals.py
+```
+
+### Graceful Shutdown:
+The script is designed to handle a graceful shutdown upon receiving a KeyboardInterrupt (e.g., Ctrl+C) or encountering 
+an unexpected exception.
+
+## Logging
+The script employs logging to provide insights into its operation and to assist in troubleshooting. Logs are saved to a 
+file named after the script with a .log extension.
+
+For further assistance or to report issues, please [contact our support team](https://www.lucit.tech/get-support.html) 
+or [visit our GitHub repository](https://github.com/LUCIT-Systems-and-Development/unicorn-binance-websocket-api).

examples/binance_websocket_stream_signals/binance_websocket_stream_signals.py

@@ -0,0 +1,48 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+from unicorn_binance_websocket_api import BinanceWebSocketApiManager
+import asyncio
+import logging
+import os
+
+logging.getLogger("unicorn_binance_websocket_api")
+logging.basicConfig(level=logging.INFO,
+                    filename=os.path.basename(__file__) + '.log',
+                    format="{asctime} [{levelname:8}] {process} {thread} {module}: {message}",
+                    style="{")
+
+
+class BinanceDataProcessor:
+    def __init__(self):
+        self.ubwa = BinanceWebSocketApiManager(process_stream_signals=self.receive_stream_signal)
+
+    async def main(self):
+        print(f"In the Wiki you will find detailed information about the respective stream signal types: "
+              f"https://github.com/LUCIT-Systems-and-Development/unicorn-binance-websocket-api/wiki/%60stream_signals%60")
+        print(f"\r\nExample of invalid API credentials:")
+        self.ubwa.create_stream(channels="arr", markets="!userData", stream_label="INVALID_CREDENTIALS",
+                                api_key="something", api_secret="wrong")
+        await asyncio.sleep(3)
+        print(f"\r\nExample of a healthy stream:")
+        stream_id = self.ubwa.create_stream(channels="trade", markets="btcusdt", stream_label="HEALTHY")
+        await asyncio.sleep(2)
+        print(f"Waiting 5 seconds and then stop the stream ...")
+        await asyncio.sleep(5)
+        self.ubwa.stop_stream(stream_id=stream_id)
+        await asyncio.sleep(5)
+
+    def receive_stream_signal(self, signal_type=None, stream_id=None, data_record=None, error_msg=None):
+        print(f"Received stream_signal for stream '{self.ubwa.get_stream_label(stream_id=stream_id)}': "
+              f"{signal_type} - {stream_id} - {data_record} - {error_msg}")
+
+
+if __name__ == "__main__":
+    bdp = BinanceDataProcessor()
+    try:
+        asyncio.run(bdp.main())
+    except KeyboardInterrupt:
+        print("\r\nGracefully stopping ...")
+    except Exception as e:
+        print(f"\r\nError: {e}\r\nGracefully stopping ...")
+    bdp.ubwa.stop_manager()

examples/binance_websocket_stream_signals/requirements.txt

@@ -0,0 +1 @@
+unicorn-binance-websocket-api

unicorn_binance_websocket_api/manager.py

@@ -855,10 +855,21 @@ def _create_stream_thread(self,
             logger.debug(f"BinanceWebSocketApiManager._create_stream_thread() stream_id={str(stream_id)} "
                          f" - RuntimeError `error: 12` - error_msg: {str(error_msg)}")
         except Exception as error_msg:
-            logger.critical(f"BinanceWebSocketApiManager._create_stream_thread({str(stream_id)} - Unknown Exception - "
-                            f"Please report this issue if your stream does not restart: "
-                            f"https://github.com/LUCIT-Systems-and-Development/unicorn-binance-websocket-api/issues/new/choose"
-                            f" - error_msg: {str(error_msg)}")
+            stream_label = self.get_stream_label(stream_id=stream_id)
+            if stream_label is None:
+                stream_label = ""
+            else:
+                stream_label = f" ({stream_label})"
+            error_msg_wrapper = (f"Exception within a coroutine of stream '{stream_id}'{stream_label}: "
+                                 f"\033[1m\033[31m{type(error_msg).__name__} - {error_msg}\033[0m\r\n"
+                                 f"{traceback.format_exc()}")
+            print(f"\r\n{error_msg_wrapper}")
+            error_msg_wrapper = (f"Exception within to UBWA`s provided `process_asyncio_queue`-coroutine of stream "
+                                 f"'{stream_id}'{stream_label}: "
+                                 f"{type(error_msg).__name__} - {error_msg}\r\n"
+                                 f"{traceback.format_exc()}")
+            logger.critical(error_msg_wrapper)
+            self._crash_stream(stream_id=stream_id, error_msg=error_msg_wrapper)
         finally:
             logger.debug(f"Finally closing the loop stream_id={str(stream_id)}")
             try: