add real time candle processing from Binance (#23)

* add real time candle processing from binance

* lint

Author: idanya

README.md

@@ -155,6 +155,14 @@ Running this example pipeline will load historical data from Binance to MongoDB:
    ```shell
    python main.py pipeline run examples/pipeline-templates/build_daily_binance_loader.json
    ```
+
+* Realtime Crypto pipeline from Binance can be found in [examples/pipeline-templates/build_realtime_binance.json](src/examples/pipeline-templates/build_realtime_binance.json). 
+Running this example pipeline will process realtime, second candles from Binance:
+
+   ```shell
+   python main.py pipeline run examples/pipeline-templates/build_realtime_binance.json
+   ```
+
 
 
 ## Virtual environment

src/examples/pipeline-templates/build_realtime_binance.json

@@ -0,0 +1,33 @@
+{
+  "__class__": "pipeline.pipeline:Pipeline",
+  "source": {
+    "__class__": "pipeline.sources.binance_realtime:BinanceRealtimeSource",
+    "binanceProvider": {
+      "apiKey": "",
+      "apiSecret": "",
+      "enableWebsocket": true
+    },
+    "symbols": [
+      "BTCUSDT"
+    ],
+    "timeSpan": 1
+  },
+  "processor": {
+    "__class__": "pipeline.processors.technicals:TechnicalsProcessor",
+    "next_processor": {
+      "__class__": "pipeline.processors.candle_cache:CandleCache",
+      "next_processor": {
+        "__class__": "pipeline.processors.storage_provider_sink:StorageSinkProcessor",
+        "storage_provider": {
+          "__class__": "storage.mongodb_storage:MongoDBStorage",
+          "host": "localhost",
+          "port": 27017,
+          "database": "algo-trader",
+          "username": "root",
+          "password": "root"
+        }
+      }
+    }
+  },
+  "terminator": null
+}

src/logger/__init__.py

@@ -1,10 +1,13 @@
 import logging
+import os
 import pathlib
 import time
+from rich.logging import RichHandler
 
 
 def setup_logger():
     log_dir = pathlib.Path(__file__).parent.parent.joinpath('logs').resolve()
+    level = logging.DEBUG if os.environ.get('DEBUG') else logging.INFO
 
     if not pathlib.Path.exists(log_dir):
         pathlib.Path.mkdir(log_dir)
@@ -16,11 +19,11 @@ def setup_logger():
     logging.basicConfig(
         filename=f'{log_dir}/{time.strftime("algo-trader.%y%m%d_%H%M%S.log")}',
         filemode="w",
-        level=logging.INFO,
+        level=level,
         format=recfmt, datefmt=timefmt)
     logger = logging.getLogger()
-    console = logging.StreamHandler()
-    console.setLevel(logging.INFO)
+    console = RichHandler()
+    console.setLevel(level)
     logger.addHandler(console)
 
     logging.getLogger('ibapi.wrapper').setLevel(logging.WARNING)

src/main.py

@@ -25,6 +25,7 @@ def save_pipeline_spec(filename: str, pipeline: Pipeline):
 
 
 def generate_example_templates():
+    save_pipeline_spec('build_realtime_binance.json', LoadersPipelines.build_realtime_binance())
     save_pipeline_spec('build_daily_binance_loader.json', LoadersPipelines.build_daily_binance_loader())
     save_pipeline_spec('build_daily_yahoo_loader.json', LoadersPipelines.build_daily_yahoo_loader())
     save_pipeline_spec('backtest_mongo_source_rsi_strategy.json', BacktestPipelines.build_mongodb_backtester())

src/pipeline/builders/loaders.py

@@ -16,6 +16,7 @@
 from pipeline.reverse_source import ReverseSource
 from pipeline.source import Source
 from pipeline.sources.binance_history import BinanceHistorySource
+from pipeline.sources.binance_realtime import BinanceRealtimeSource
 from pipeline.sources.ib_history import IBHistorySource
 from pipeline.sources.mongodb_source import MongoDBSource
 from pipeline.sources.yahoo_finance_history import YahooFinanceHistorySource
@@ -75,6 +76,21 @@ def build_daily_binance_loader(days_back: int = DEFAULT_DAYS_BACK) -> Pipeline:
 
         return Pipeline(source, processor)
 
+    @staticmethod
+    def build_realtime_binance() -> Pipeline:
+        mongodb_storage = MongoDBStorage()
+
+        symbols = AssetsProvider.get_crypto_symbols()
+
+        provider = BinanceProvider(enable_websocket=True)
+        source = BinanceRealtimeSource(provider, symbols, TimeSpan.Second)
+
+        sink = StorageSinkProcessor(mongodb_storage)
+        cache_processor = CandleCache(sink)
+        processor = TechnicalsProcessor(cache_processor)
+
+        return Pipeline(source, processor)
+
     @staticmethod
     def build_returns_calculator(days_back: int = DEFAULT_DAYS_BACK) -> Pipeline:
         mongodb_storage = MongoDBStorage()

src/pipeline/pipeline.py

@@ -1,6 +1,8 @@
 import logging
 from typing import Optional, Dict
 
+from rich.progress import Progress, TextColumn, BarColumn
+
 from entities.serializable import Serializable, Deserializable
 from pipeline.processor import Processor
 from pipeline.shared_context import SharedContext
@@ -34,8 +36,14 @@ def deserialize(cls, data: Dict):
 
     def run(self, context: SharedContext) -> None:
         self.logger.info('Starting pipeline...')
-        for candle in self.source.read():
-            self.processor.process(context, candle)
+
+        with Progress(TextColumn('{task.completed} Candle(s) processed'), BarColumn()) as progress:
+            processing_task = progress.add_task("Processing", total=None)
+
+            for candle in self.source.read():
+                self.logger.debug('Processing candle: %s\r', candle.serialize())
+                self.processor.process(context, candle)
+                progress.update(processing_task, advance=1)
 
         if self.terminator:
             self.terminator.terminate(context)

src/pipeline/sources/binance_realtime.py

@@ -0,0 +1,40 @@
+from queue import Queue
+from typing import List, Dict, Iterator
+
+from entities.candle import Candle
+from entities.timespan import TimeSpan
+from pipeline.source import Source
+from providers.binance import BinanceProvider
+
+
+class BinanceRealtimeSource(Source):
+
+    def __init__(self, binance_provider: BinanceProvider, symbols: List[str], time_span: TimeSpan):
+        self.binance_provider = binance_provider
+        self.symbols = symbols
+        self.time_span = time_span
+        self.queue = Queue()
+
+    def read(self) -> Iterator[Candle]:
+        for symbol in self.symbols:
+            self.binance_provider.start_kline_socket(symbol, self.time_span, self._on_candle)
+
+        while self.binance_provider.is_socket_alive():
+            yield self.queue.get()
+
+    def _on_candle(self, candle: Candle):
+        self.queue.put(candle)
+
+    def serialize(self) -> Dict:
+        obj = super().serialize()
+        obj.update({
+            'binanceProvider': self.binance_provider.serialize(),
+            'symbols': self.symbols,
+            'timeSpan': self.time_span.value,
+        })
+        return obj
+
+    @classmethod
+    def deserialize(cls, data: Dict):
+        provider = BinanceProvider.deserialize(data.get('binanceProvider'))
+        return cls(provider, data.get('symbols'), TimeSpan(data.get('timeSpan')))

src/providers/binance.py

@@ -27,10 +27,13 @@ def __init__(self, api_key: Optional[str] = '', api_secret: Optional[str] = '',
             self.wsManager.start()
 
     def stop(self):
-        if self.wsManager.isAlive():
+        if self.is_socket_alive():
             self.logger.info('Stopping websocket manager...')
             self.wsManager.stop()
 
+    def is_socket_alive(self) -> bool:
+        return self.wsManager.is_alive()
+
     def start_kline_socket(self, symbol: str, interval: TimeSpan, handler: StreamedCandleCallback):
         self.logger.info(f'Starting kline socket for {symbol}...')
         self.wsManager.kline(symbol=symbol, id=1,