examples

Author: oliver-zehentleitner

examples/binance_websocket_best_practice/README.md

@@ -0,0 +1,30 @@
+# Binance WebSocket Best Practice
+## Overview
+A best practice example for websockets to Binance in Python.
+
+## 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_best_practice.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_best_practice/binance_websocket_best_practice.py

@@ -0,0 +1,83 @@
+#!/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, print_new_data=False, start_multiplex=True, start_userdata_a=True):
+        self.api_key = "YOUR_BINANCE_API_KEY"
+        self.api_secret = "YOUR_BINANCE_API_SECRET"
+        self.example_database = []
+        self.print_new_data = print_new_data
+        self.start_multiplex = start_multiplex
+        self.start_userdata_a = start_userdata_a
+        self.ubwa = BinanceWebSocketApiManager(exchange='binance.com',
+                                               auto_data_cleanup_stopped_streams=True,
+                                               enable_stream_signal_buffer=True,
+                                               output_default='UnicornFy',
+                                               process_stream_signals=self.receive_stream_signal)
+
+    async def main(self):
+        if self.start_multiplex is True:
+            self.ubwa.create_stream(channels=['trade', 'kline_1m', 'depth5'],
+                                    markets=['btcusdt', 'ethusdt', 'bnbusdt'],
+                                    process_asyncio_queue=self.processing_of_new_data,
+                                    stream_label="multiplex")
+
+        if self.start_userdata_a is True:
+            self.ubwa.create_stream(api_key=self.api_key, api_secret=self.api_secret,
+                                    channels=["arr"], markets=["!userData"],
+                                    process_asyncio_queue=self.processing_of_new_data,
+                                    stream_label="userData_A")
+
+        while self.ubwa.is_manager_stopping() is False:
+            await asyncio.sleep(1)
+            stream_info = \
+                {'multiplex': self.ubwa.get_stream_info(stream_id=self.ubwa.get_stream_id_by_label('multiplex')),
+                 'userData_A': self.ubwa.get_stream_info(stream_id=self.ubwa.get_stream_id_by_label('userData_A'))}
+            status_text = ""
+            if self.start_multiplex is True:
+                status_text += (f"\tStream 'multiplex' is {stream_info['multiplex']['status']} "
+                                f"(last_stream_signal={stream_info['multiplex']['last_stream_signal']})\r\n")
+            if self.start_userdata_a is True:
+                status_text += (f"\tStream 'userData_A' is {stream_info['userData_A']['status']} "
+                                f"(last_stream_signal={stream_info['userData_A']['last_stream_signal']})\r\n")
+            print(f"Status:\r\n\tStored {len(self.example_database)} data records in `self.example_database`\r\n"
+                  f"{status_text}")
+
+    async def processing_of_new_data(self, stream_id=None):
+        print(f"Saving the data from webstream {self.ubwa.get_stream_label(stream_id=stream_id)} to the database ...")
+        while self.ubwa.is_stop_request(stream_id=stream_id) is False:
+            data = await self.ubwa.get_stream_data_from_asyncio_queue(stream_id)
+            self.example_database.append(data)
+            if self.print_new_data is True:
+                print(f"Data record received and added to the database: {data}")
+            self.ubwa.asyncio_queue_task_done(stream_id)
+
+    def receive_stream_signal(self, signal_type=None, stream_id=None, data_record=None, error_msg=None):
+        # More info about `stream_signals`:
+        # https://github.com/LUCIT-Systems-and-Development/unicorn-binance-websocket-api/wiki/%60stream_signals%60
+        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(print_new_data=False, start_multiplex=True, start_userdata_a=True)
+    try:
+        asyncio.run(bdp.main())
+    except KeyboardInterrupt:
+        print("\r\nGracefully stopping ...")
+    except Exception as e:
+        print(f"\r\nERROR: {e}")
+        print("Gracefully stopping ...")
+    bdp.ubwa.stop_manager()

examples/binance_websocket_best_practice/requirements.txt

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