books and parser added

.gitignore

@@ -1,3 +1,6 @@
+.idea/
+dump.txt
+
 # Byte-compiled / optimized / DLL files
 __pycache__/
 *.py[cod]

README.md

@@ -2,12 +2,15 @@
 A Set of Simple Connectors for access To Cryptocurrency Exchanges via websocket based on
  [aiohttp](https://aiohttp.readthedocs.io) .
 
+This is more of a pilot project, if you have any wishes for adding exchanges or expanding functionality, please register issues
+
 ## Installation
 Install ssc2ce with:
 ```bash
 $ pip install ssc2ce
 ```
 
+You can run some examples with  
 ## Bitfinex
 ### Description
 API description look at [Websocket API v2](https://docs.bitfinex.com/v2/docs/ws-general)
@@ -64,7 +67,7 @@ async def subscribe():
     })
 
 
-async def handle_subscription(data):
+def handle_subscription(data):
     method = data.get("method")
     if method and method == "subscription":
         if data["params"]["channel"].startswith("deribit_price_index"):
@@ -88,21 +91,19 @@ except KeyboardInterrupt:
 ## Run examples from a clone
 
 If you clone repository you can run examples from the root directory.
+
 ```bash
-$ PYTHONPATH=.:$PYTHONPATH python examples/basic_example.py
+$ PYTHONPATH=.:$PYTHONPATH python examples/bitfinex_basic_example.py
 ```
 
-The deribit_private.py example uses [python-dotenv](https://github.com/theskumar/python-dotenv), you must either install it if you want the example to work right out of the box,
+To run some examples, you may need additional modules, you can install them from the `requirements.txt` file.
+
 ```bash
-$ pip install python-dotenv
-```
-or make the corresponding changes, removed followed code.
-```python
-from dotenv import load_dotenv
-dotenv_path = os.path.join(os.path.dirname(__file__), '.env')
-load_dotenv(dotenv_path)
+$ pip install -r requirements.txt
 ```
+
 To run the private.py example, you must either fill in the .env file or set the environment variables DERIBIT_CLIENT_ID and DERIBIT_CLIENT_SECRET. Look at .env_default. 
+
 ```bash
-$ PYTHONPATH=.:$PYTHONPATH DERIBIT_CLIENT_ID=YOU_ACCESS_KEY DERIBIT_CLIENT_SECRET=YOU_ACCESS_SECRET python examples/private.py
+$ PYTHONPATH=.:$PYTHONPATH DERIBIT_CLIENT_ID=YOU_ACCESS_KEY DERIBIT_CLIENT_SECRET=YOU_ACCESS_SECRET python examples/deribit_private.py
 ```

examples/bitfinex_book.py

@@ -2,21 +2,23 @@
 import asyncio
 import logging
 
-from ssc2ce.bitfinex import Bitfinex, L2Book
+from ssc2ce.bitfinex import Bitfinex, BitfinexL2Book
 
-logging.basicConfig(format='%(asctime)s %(name)s %(funcName)s %(levelname)s %(message)s', level=logging.INFO)
+logging.basicConfig(
+    format='%(asctime)s %(name)s %(funcName)s %(levelname)s %(message)s',
+    level=logging.INFO)
 logger = logging.getLogger("bitfinex-basic-example")
 
 conn = Bitfinex()
 
 
 class BookMaintainer:
     def __init__(self, instrument):
-        self.book = L2Book(instrument)
+        self.book = BitfinexL2Book(instrument)
         self.check_sum = None
         self.active = False
-        self.top_bid = [0, 0]
-        self.top_ask = [0, 0]
+        self.top_bid = 0.
+        self.top_ask = 0.
 
     async def handle_book(self, message, connector):
         if type(message[1]) is str:
@@ -29,10 +31,11 @@ async def handle_book(self, message, connector):
             self.book.handle_snapshot(message)
             self.book.time = message[-1]
 
-        if self.top_ask[0] != self.book.asks[0][0] or self.top_ask[0] != self.book.asks[0][0]:
-            self.top_ask = self.book.asks[0].copy()
-            self.top_bid = self.book.bids[0].copy()
-            print(f"{self.book.instrument} bid:{self.top_bid[0]} ask:{self.top_ask[0]}")
+        if self.top_ask != self.book.top_ask_price() or self.top_bid != self.book.top_bid_price():
+            self.top_ask = self.book.top_ask_price()
+            self.top_bid = self.book.top_bid_price()
+            logger.info(
+                f"{self.book.instrument()} bid:{self.top_bid} ask:{self.top_ask}")
 
 
 btc = BookMaintainer("BTC-USD")

examples/book_watcher.py

@@ -0,0 +1,23 @@
+from ssc2ce.deribit.l2_book import L2Book
+
+
+class BookWatcher:
+    def __init__(self, parser, print_it: bool = True):
+        self.betters = {}
+        self.print_it = print_it
+        parser.set_on_book_setup(self.handle_book_setup)
+        parser.set_on_book_update(self.handle_book_update)
+
+    def handle_book_setup(self, book: L2Book):
+        top = [book.top_bid_price(), book.top_ask_price()]
+        self.betters[book.instrument()] = top
+        if self.print_it:
+            print(f"{book.instrument()} bid:{top[0]} ask:{top[1]}")
+
+    def handle_book_update(self, book: L2Book):
+        top = self.betters[book.instrument()]
+        if top[1] != book.top_ask_price() or top[0] != book.top_bid_price():
+            top[1] = book.top_ask_price()
+            top[0] = book.top_bid_price()
+            if self.print_it:
+                print(f"{book.instrument()} bid:{top[0]} ask:{top[1]}")

examples/deribit_basic_example.py

@@ -8,18 +8,21 @@
 pending = {}
 
 
-async def handle_instruments(data: dict):
+def handle_instruments(data: dict):
     del pending[data["id"]]
-    print(json.dumps(data))
+    print("handle_instruments", json.dumps(data))
     if not pending:
-        await subscribe()
+        asyncio.ensure_future(subscribe())
 
 
-async def handle_currencies(data: dict):
+async def get_instruments(symbol):
+    request_id = await conn.get_instruments(symbol, callback=handle_instruments)
+    pending[request_id] = symbol
+
+
+def handle_currencies(data: dict):
     for currency in data["result"]:
-        symbol = currency["currency"]
-        id = await conn.get_instruments(symbol, callback=handle_instruments)
-        pending[id] = symbol
+        asyncio.ensure_future(get_instruments(currency["currency"]))
 
 
 async def get_currencies():
@@ -30,12 +33,13 @@ async def subscribe():
     await conn.send_public(request={
         "method": "public/subscribe",
         "params": {
-            "channels": ["deribit_price_index.btc_usd"]
+            "channels": ["deribit_price_index.btc_usd",
+                         "deribit_price_index.eth_usd"]
         }
     })
 
 
-async def handle_subscription(data):
+def handle_subscription(data):
     method = data.get("method")
     if method and method == "subscription":
         if data["params"]["channel"].startswith("deribit_price_index"):

examples/deribit_book.py

@@ -2,15 +2,34 @@
 import asyncio
 import json
 import logging
+import sys
+
+from examples.book_watcher import BookWatcher
 from ssc2ce import Deribit
-from ssc2ce.deribit.l2_book import L2Book
+
+if len(sys.argv) > 1 and "cpp" in sys.argv:
+    from importlib import util
+
+    ssc2ce_cpp_spec = util.find_spec("ssc2ce_cpp")
+    if ssc2ce_cpp_spec:
+        from ssc2ce_cpp import DeribitParser
+    else:
+        print("You must install the ssc2ce_cpp module to use its features.\n pip install ssc2ce_cpp")
+        exit(1)
+else:
+    from ssc2ce.deribit.parser import DeribitParser
 
 conn = Deribit()
+parser = DeribitParser()
+watcher = BookWatcher(parser)
+
+# parser.set_on_unprocessed_message(conn.handle_message)
+conn.on_message = parser.parse
 
 pending = {}
-books = {}
+instruments = []
 
-logging.basicConfig(format='%(asctime)s %(name)s %(funcName)s %(levelname)s %(message)s', level=logging.INFO)
+logging.basicConfig(format='%(asctime)s %(name)s %(funcName)s %(levelname)s %(message)s', level=logging.DEBUG)
 logger = logging.getLogger("deribit-book")
 
 
@@ -19,17 +38,18 @@ async def handle_instruments(data: dict):
     del pending[request_id]
     print(json.dumps(data))
     if not pending:
-        await subscribe_books(list(books.keys()))
+        await conn.send_public(request={
+            "method": "public/subscribe",
+            "params": {
+                "channels": [f"book.{i}.raw" for i in instruments]
+            }
+        })
 
 
 async def handle_currencies(data: dict):
     for currency in data["result"]:
         symbol = currency["currency"]
-        instrument = symbol+"-PERPETUAL"
-        book = L2Book(instrument)
-        book.top_bid = [0., 0.]
-        book.top_ask = [0., 0.]
-        books[instrument] = book
+        instruments.append(symbol + "-PERPETUAL")
         request_id = await conn.get_instruments(symbol, kind="future", callback=handle_instruments)
         pending[request_id] = symbol
 
@@ -38,39 +58,7 @@ async def get_currencies():
     await conn.get_currencies(handle_currencies)
 
 
-async def subscribe_books(instruments: list):
-    await conn.send_public(request={
-        "method": "public/subscribe",
-        "params": {
-            "channels": [f"book.{i}.raw" for i in instruments]
-        }
-    })
-
-
-async def handle_subscription(data):
-    method = data.get("method")
-    if method and method == "subscription":
-        params = data["params"]
-        channel = params["channel"]
-        if channel.startswith("book."):
-            params_data = params["data"]
-            instrument = params_data["instrument_name"]
-            book: L2Book = books[instrument]
-            if "prev_change_id" in params_data:
-                book.handle_update(params_data)
-            else:
-                book.handle_snapshot(params_data)
-
-            if book.top_ask[0] != book.asks[0][0] or book.top_bid[0] != book.bids[0][0]:
-                book.top_ask = book.asks[0].copy()
-                book.top_bid = book.bids[0].copy()
-                print(f"{instrument} bid:{book.top_bid[0]} ask:{book.top_ask[0]}")
-        else:
-            print("Unknown channel", json.dumps(data))
-
-
 conn.on_connect_ws = get_currencies
-conn.method_routes += [("subscription", handle_subscription)]
 
 loop = asyncio.get_event_loop()
 

examples/deribit_parser.py

@@ -0,0 +1,47 @@
+import asyncio
+import logging
+from time import time
+
+from examples.book_watcher import BookWatcher
+
+import sys
+
+if len(sys.argv) > 1 and "cpp" in sys.argv:
+    from importlib import util
+
+    ssc2ce_cpp_spec = util.find_spec("ssc2ce_cpp")
+    if ssc2ce_cpp_spec:
+        from ssc2ce_cpp import DeribitParser
+    else:
+        print("You must install the ssc2ce_cpp module to use its features.\n pip install ssc2ce_cpp")
+        exit(1)
+else:
+    from ssc2ce.deribit.parser import DeribitParser
+
+logging.basicConfig(format='%(asctime)s %(name)s %(funcName)s %(levelname)s %(message)s', level=logging.INFO)
+logger = logging.getLogger("deribit-parser")
+
+
+class FileController:
+    def __init__(self):
+        self.parser = DeribitParser()
+        self.counter = 0
+        self.watcher = BookWatcher(self.parser, False)
+
+    def run(self, filename: str):
+        start = time()
+        i = 0
+        with open(filename) as f:
+            for line in f:
+                i += 1
+                if not self.parser.parse(line):
+                    self.handle_message(line)
+
+        logger.info(f"{i} in {time() - start} sec. {self.counter}")
+
+    def handle_message(self, message: str) -> None:
+        self.counter += 1
+        # logger.info(message[:-1])
+
+
+FileController().run("dump.txt")