Ticker support on cryptostore + fix redis host issue

config.yaml

@@ -44,13 +44,16 @@ exchanges:
             max_depth: 10
             book_delta: true
         trades: [XBTUSD]
+        ticker: [XBTUSD]
     COINBASE:
         retries: -1
         l3_book:
             symbols: [BTC-USD, ETH-USD]
             book_delta: true
             book_interval: 10000
         trades: [BTC-USD, ETH-USD, ETH-BTC]
+        ticker: [BTC-USD]
+
 
 # Where to store the data. Currently arctic, influx, elastic, and parquet are supported. More than one can be enabled
 storage: [arctic, influx]

cryptostore/aggregator/kafka.py

@@ -7,7 +7,7 @@
 import json
 import logging
 
-from cryptofeed.defines import L2_BOOK, L3_BOOK, TRADES
+from cryptofeed.defines import L2_BOOK, L3_BOOK, TRADES, TICKER
 
 from cryptostore.engines import StorageEngines
 from cryptostore.aggregator.cache import Cache
@@ -59,7 +59,7 @@ def read(self, exchange, dtype, pair):
             if dtype in {L2_BOOK, L3_BOOK}:
                 update = book_flatten(update, update['timestamp'], update['delta'])
                 ret.extend(update)
-            if dtype == TRADES:
+            if dtype in {TRADES, TICKER}:
                 ret.append(update)
         return ret
 

cryptostore/aggregator/redis.py

@@ -9,7 +9,7 @@
 import json
 import time
 
-from cryptofeed.defines import TRADES, L2_BOOK, L3_BOOK
+from cryptofeed.defines import TRADES, L2_BOOK, L3_BOOK, TICKER
 
 from cryptostore.aggregator.util import book_flatten
 from cryptostore.aggregator.cache import Cache
@@ -51,8 +51,8 @@ def read(self, exchange, dtype, pair):
                         if k in u:
                             u[k] = float(u[k])
                 ret.extend(update)
-            if dtype == TRADES:
-                for k in ('size', 'amount', 'price', 'timestamp'):
+            if dtype in {TRADES, TICKER}:
+                for k in ('size', 'amount', 'price', 'timestamp', 'bid', 'ask'):
                     if k in update:
                         update[k] = float(update[k])
                 ret.append(update)

cryptostore/collector.py

@@ -9,7 +9,7 @@
 import logging
 
 from cryptofeed import FeedHandler
-from cryptofeed.defines import TRADES, L2_BOOK, L3_BOOK, BOOK_DELTA
+from cryptofeed.defines import TRADES, L2_BOOK, L3_BOOK, BOOK_DELTA, TICKER
 
 
 LOG = logging.getLogger('cryptostore')
@@ -48,23 +48,27 @@ def run(self):
                 self.exchange_config[callback_type] = self.exchange_config[callback_type]['symbols']
 
             if cache == 'redis':
-                if self.config['redis']['ip']:
-                    kwargs = {'ip': self.config['redis']['ip'], 'port': self.config['redis']['port'], 'numeric_type': float}
+                if 'ip' in self.config['redis'] and self.config['redis']['ip']:
+                    kwargs = {'host': self.config['redis']['ip'], 'port': self.config['redis']['port'], 'numeric_type': float}
                 else:
                     kwargs = {'socket': self.config['redis']['socket'], 'numeric_type': float}
-                from cryptofeed.backends.redis import TradeStream, BookStream, BookDeltaStream
+                from cryptofeed.backends.redis import TradeStream, BookStream, BookDeltaStream, TickerStream
                 trade_cb = TradeStream
                 book_cb = BookStream
                 book_up = BookDeltaStream if delta else None
+                ticker_cb = TickerStream
             elif cache == 'kafka':
-                from cryptofeed.backends.kafka import TradeKafka, BookKafka, BookDeltaKafka
+                from cryptofeed.backends.kafka import TradeKafka, BookKafka, BookDeltaKafka, TickerKafka
                 trade_cb = TradeKafka
                 book_cb = BookKafka
                 book_up = BookDeltaKafka if delta else None
+                ticker_cb = TickerKafka
                 kwargs = {'host': self.config['kafka']['ip'], 'port': self.config['kafka']['port']}
 
             if callback_type == TRADES:
                 cb[TRADES] = [trade_cb(**kwargs)]
+            elif callback_type == TICKER:
+                cb[TICKER] = [ticker_cb(**kwargs)]
             elif callback_type == L2_BOOK:
                 cb[L2_BOOK] = [book_cb(key=L2_BOOK, **kwargs)]
                 if book_up:

cryptostore/data/arctic.py

@@ -5,7 +5,7 @@
 associated with this software.
 '''
 import pandas as pd
-from cryptofeed.defines import TRADES, L2_BOOK, L3_BOOK
+from cryptofeed.defines import TRADES, L2_BOOK, L3_BOOK, TICKER
 
 from cryptostore.data.store import Store
 from cryptostore.engines import StorageEngines
@@ -32,6 +32,10 @@ def write(self, exchange, data_type, pair, timestamp):
             df['date'] = pd.to_datetime(df['timestamp'], unit='s')
             df = df.drop(['pair', 'feed', 'amount'], axis=1)
             chunk_size = 'H'
+        elif data_type == TICKER:
+            df['date'] = pd.to_datetime(df['timestamp'], unit='s')
+            df = df.drop(['pair', 'feed'], axis=1)
+            chunk_size = 'D'
         elif data_type in { L2_BOOK, L3_BOOK }:
             df['date'] = pd.to_datetime(df['timestamp'], unit='s')
             chunk_size = 'T'

cryptostore/data/influx.py

@@ -7,7 +7,7 @@
 from decimal import Decimal
 from collections import defaultdict
 
-from cryptofeed.defines import TRADES, L2_BOOK, L3_BOOK
+from cryptofeed.defines import TRADES, L2_BOOK, L3_BOOK, TICKER
 import requests
 
 from cryptostore.data.store import Store
@@ -44,6 +44,11 @@ def write(self, exchange, data_type, pair, timestamp):
                     ts += 1
                 used_ts[pair].add(ts)
                 agg.append(f'{data_type}-{exchange},pair={pair} side="{entry["side"]}",id="{entry["id"]}",amount={entry["amount"]},price={entry["price"]},timestamp={entry["timestamp"]} {ts}')
+        elif data_type == TICKER:
+            for entry in self.data:
+                ts = int(Decimal(entry["timestamp"]) * 1000000000)
+                agg.append(f'{data_type}-{exchange},pair={pair} bid={entry["bid"]},ask={entry["ask"]},timestamp={entry["timestamp"]} {ts}')
+
         elif data_type == L2_BOOK:
             if len(self.data):
                 ts = int(Decimal(self.data[0]["timestamp"]) * 1000000000)