Support for authenticated data channel on Gemini

Author: bmoscon

cryptofeed/auth/__init__.py

@@ -0,0 +1,30 @@
+'''
+Copyright (C) 2017-2021  Bryant Moscon - [email protected]
+
+Please see the LICENSE file for the terms and conditions
+associated with this software.
+'''
+import base64
+import hashlib
+import hmac
+import json
+import time
+
+
+def generate_token(key_id: str, key_secret: str, request: str, account_name=None, payload=None) -> dict:
+    if not payload:
+        payload = {}
+    payload['request'] = request
+    payload['nonce'] = int(time.time() * 1000)
+
+    if account_name:
+        payload['account'] = account_name
+
+    b64_payload = base64.b64encode(json.dumps(payload).encode('utf-8'))
+    signature = hmac.new(key_secret.encode('utf-8'), b64_payload, hashlib.sha384).hexdigest()
+
+    return {
+        'X-GEMINI-PAYLOAD': b64_payload.decode(),
+        'X-GEMINI-APIKEY': key_id,
+        'X-GEMINI-SIGNATURE': signature
+    }

cryptofeed/auth/gemini.py

@@ -100,3 +100,7 @@ class MarketInfoCallback(Callback):
 
 class TransactionsCallback(Callback):
     pass
+
+
+class OrderInfoCallback(Callback):
+    pass

cryptofeed/callback.py

@@ -45,6 +45,7 @@
 WHALE_ALERT = 'WHALE_ALERT'
 
 
+# Market Data
 L2_BOOK = 'l2_book'
 L3_BOOK = 'l3_book'
 BOOK_DELTA = 'book_delta'
@@ -59,6 +60,10 @@
 MARKET_INFO = 'market_info'
 TRANSACTIONS = 'transactions'
 
+# Account Data / Authenticated Channels
+ORDER_INFO = 'order_info'
+
+
 BUY = 'buy'
 SELL = 'sell'
 BID = 'bid'

cryptofeed/defines.py

@@ -6,14 +6,16 @@
 '''
 import logging
 from decimal import Decimal
+from typing import List, Tuple, Callable
 
 from sortedcontainers import SortedDict as sd
 from yapic import json
 
+from cryptofeed.auth.gemini import generate_token
 from cryptofeed.connection import AsyncConnection
-from cryptofeed.defines import BID, ASK, BUY, GEMINI, L2_BOOK, SELL, TRADES
+from cryptofeed.defines import BID, ASK, BUY, GEMINI, L2_BOOK, SELL, TRADES, ORDER_INFO
 from cryptofeed.feed import Feed
-from cryptofeed.standards import symbol_exchange_to_std, timestamp_normalize
+from cryptofeed.standards import symbol_exchange_to_std, timestamp_normalize, is_authenticated_channel
 
 
 LOG = logging.getLogger('feedhandler')
@@ -22,8 +24,9 @@
 class Gemini(Feed):
     id = GEMINI
 
-    def __init__(self, **kwargs):
-        super().__init__('wss://api.gemini.com/v2/marketdata/', **kwargs)
+    def __init__(self, sandbox=False, **kwargs):
+        auth_api = 'wss://api.gemini.com' if not sandbox else 'wss://api.sandbox.gemini.com'
+        super().__init__({'public': 'wss://api.gemini.com/v2/marketdata/', 'auth': f'{auth_api}/v1/order/events'}, **kwargs)
 
     def __reset(self, pairs):
         for pair in pairs:
@@ -69,7 +72,43 @@ async def _trade(self, msg: dict, timestamp: float):
                             timestamp=timestamp_normalize(self.id, msg['timestamp']),
                             receipt_timestamp=timestamp)
 
-    async def message_handler(self, msg: str, conn, timestamp: float):
+    async def _order(self, msg: dict, timestamp: float):
+        if msg['type'] == "initial" or msg['type'] == "booked":
+            status = "active"
+        elif msg['type'] == "fill":
+            status = 'filled'
+        else:
+            status = msg['type']
+
+        keys = ('executed_amount', 'remaining_amount', 'original_amount', 'price', 'avg_execution_price', 'total_spend')
+        data = {k: Decimal(msg[k]) for k in keys if k in msg}
+
+        await self.callback(ORDER_INFO, feed=self.id,
+                            symbol=symbol_exchange_to_std(msg['symbol'].upper()),  # This uses the REST endpoint format (lower case)
+                            status=status,
+                            order_id=msg['order_id'],
+                            side=BUY if msg['side'].lower() == 'buy' else SELL,
+                            order_type=msg['order_type'],
+                            timestamp=msg['timestampms'] / 1000.0,
+                            receipt_timestamp=timestamp,
+                            **data
+                            )
+
+    async def message_handler_orders(self, msg: str, conn: AsyncConnection, timestamp: float):
+        msg = json.loads(msg, parse_float=Decimal)
+
+        if isinstance(msg, list):
+            for entry in msg:
+                await self._order(entry, timestamp)
+        elif isinstance(msg, dict):
+            if msg['type'] == 'subscription_ack':
+                LOG.info('%s: Authenticated successfully', self.id)
+            elif msg['type'] == 'heartbeat':
+                return
+            else:
+                await self._order(msg, timestamp)
+
+    async def message_handler(self, msg: str, conn: AsyncConnection, timestamp: float):
         msg = json.loads(msg, parse_float=Decimal)
 
         if msg['type'] == 'l2_updates':
@@ -83,9 +122,27 @@ async def message_handler(self, msg: str, conn, timestamp: float):
         else:
             LOG.warning('%s: Invalid message type %s', self.id, msg)
 
-    async def subscribe(self, conn: AsyncConnection):
-        pairs = self.symbols if not self.subscription else list(set.union(*list(self.subscription.values())))
-        self.__reset(pairs)
+    def connect(self) -> List[Tuple[AsyncConnection, Callable[[None], None], Callable[[str, float], None]]]:
+        authenticated = []
+        public = []
+        ret = []
+
+        for channel in self.subscription or self.channels:
+            if is_authenticated_channel(channel):
+                authenticated.extend(self.subscription.get(channel) or self.symbols)
+            else:
+                public.extend(self.subscription.get(channel) or self.symbols)
+
+        if authenticated:
+            header = generate_token(self.key_id, self.key_secret, "/v1/order/events", self.config.gemini.account_name)
+            symbols = '&'.join([f"symbolFilter={s.lower()}" for s in authenticated])  # needs to match REST format (lower case)
+
+            ret.append(self._connect_builder(f"{self.address['auth']}?{symbols}", None, header=header, sub=self._empty_subscribe, handler=self.message_handler_orders))
+        if public:
+            ret.append(self._connect_builder(self.address['public'], list(set(public))))
+
+        return ret
 
-        await conn.send(json.dumps({"type": "subscribe",
-                                    "subscriptions": [{"name": "l2", "symbols": pairs}]}))
+    async def subscribe(self, conn: AsyncConnection, options=None):
+        self.__reset(options)
+        await conn.send(json.dumps({"type": "subscribe", "subscriptions": [{"name": "l2", "symbols": options}]}))

cryptofeed/exchange/gemini.py

@@ -5,6 +5,7 @@
 associated with this software.
 '''
 from collections import defaultdict
+from functools import partial
 import logging
 import os
 from typing import Tuple, Callable, Union, List
@@ -15,7 +16,7 @@
 from cryptofeed.defines import (ASK, BID, BOOK_DELTA, FUNDING, FUTURES_INDEX, L2_BOOK, L3_BOOK, LIQUIDATIONS,
                                 OPEN_INTEREST, MARKET_INFO, TICKER, TRADES, TRANSACTIONS, VOLUME)
 from cryptofeed.exceptions import BidAskOverlapping, UnsupportedDataFeed
-from cryptofeed.standards import feed_to_exchange, get_exchange_info, load_exchange_symbol_mapping, symbol_std_to_exchange
+from cryptofeed.standards import feed_to_exchange, get_exchange_info, load_exchange_symbol_mapping, symbol_std_to_exchange, is_authenticated_channel
 from cryptofeed.util.book import book_delta, depth
 
 
@@ -25,12 +26,14 @@
 class Feed:
     id = 'NotImplemented'
 
-    def __init__(self, address: Union[dict, str], symbols=None, channels=None, subscription=None, config: Union[Config, dict, str] = None, callbacks=None, max_depth=None, book_interval=1000, snapshot_interval=False, checksum_validation=False, cross_check=False, origin=None):
+    def __init__(self, address: Union[dict, str], sandbox=False, symbols=None, channels=None, subscription=None, config: Union[Config, dict, str] = None, callbacks=None, max_depth=None, book_interval=1000, snapshot_interval=False, checksum_validation=False, cross_check=False, origin=None):
         """
         address: str, or dict
             address to be used to create the connection.
             The address protocol (wss or https) will be used to determine the connection type.
             Use a "str" to pass one single address, or a dict of option/address
+        sandbox: bool
+            For authenticated channels, run against the sandbox websocket (when True)
         max_depth: int
             Maximum number of levels per side to return in book updates
         book_interval: int
@@ -69,15 +72,20 @@ def __init__(self, address: Union[dict, str], symbols=None, channels=None, subsc
         self.origin = origin
         self.checksum_validation = checksum_validation
         self.ws_defaults = {'ping_interval': 10, 'ping_timeout': None, 'max_size': 2**23, 'max_queue': None, 'origin': self.origin}
-        key_id = os.environ.get(f'CF_{self.id}_KEY_ID') or self.config[self.id.lower()].key_id
-        load_exchange_symbol_mapping(self.id, key_id=key_id)
+        self.key_id = os.environ.get(f'CF_{self.id}_KEY_ID') or self.config[self.id.lower()].key_id
+        self.key_secret = os.environ.get(f'CF_{self.id}_KEY_SECRET') or self.config[self.id.lower()].key_secret
+
+        load_exchange_symbol_mapping(self.id, key_id=self.key_id)
 
         if subscription is not None and (symbols is not None or channels is not None):
             raise ValueError("Use subscription, or channels and symbols, not both")
 
         if subscription is not None:
             for channel in subscription:
                 chan = feed_to_exchange(self.id, channel)
+                if is_authenticated_channel(channel):
+                    if not self.key_id or not self.key_secret:
+                        raise ValueError("Authenticated channel subscribed to, but no auth keys provided")
                 self.subscription[chan].update([symbol_std_to_exchange(symbol, self.id) for symbol in subscription[channel]])
                 self.normalized_symbols.extend(self.subscription[chan])
 
@@ -86,6 +94,9 @@ def __init__(self, address: Union[dict, str], symbols=None, channels=None, subsc
             self.symbols = [symbol_std_to_exchange(symbol, self.id) for symbol in symbols]
         if channels:
             self.channels = list(set([feed_to_exchange(self.id, chan) for chan in channels]))
+            if any(is_authenticated_channel(chan) for chan in channels):
+                if not self.key_id or not self.key_secret:
+                    raise ValueError("Authenticated channel subscribed to, but no auth keys provided")
 
         self.l3_book = {}
         self.l2_book = {}
@@ -112,6 +123,17 @@ def __init__(self, address: Union[dict, str], symbols=None, channels=None, subsc
             if not isinstance(callback, list):
                 self.callbacks[key] = [callback]
 
+    def _connect_builder(self, address: str, options: list, header=None, sub=None, handler=None):
+        """
+        Helper method for building a custom connect tuple
+        """
+        subscribe = partial(self.subscribe if not sub else sub, options=options)
+        conn = AsyncConnection(address, self.id, extra_headers=header, **self.ws_defaults)
+        return conn, subscribe, handler if handler else self.message_handler
+
+    async def _empty_subscribe(self, conn: AsyncConnection, **kwargs):
+        return
+
     def connect(self) -> List[Tuple[AsyncConnection, Callable[[None], None], Callable[[str, float], None]]]:
         """
         Generic connection method for exchanges. Exchanges that require/support

cryptofeed/feed.py

@@ -1,15 +1,12 @@
-import base64
-import hashlib
-import hmac
 import logging
 from decimal import Decimal
-from time import sleep, time
+from time import sleep
 
 import pandas as pd
 import requests
 from sortedcontainers.sorteddict import SortedDict as sd
-from yapic import json
 
+from cryptofeed.auth.gemini import generate_token
 from cryptofeed.defines import BID, ASK, BUY, CANCELLED, FILLED, GEMINI, LIMIT, OPEN, PARTIAL, SELL
 from cryptofeed.rest.api import API, request_retry
 from cryptofeed.standards import normalize_trading_options, symbol_exchange_to_std, symbol_std_to_exchange
@@ -63,26 +60,14 @@ def helper():
         return helper()
 
     def _post(self, command: str, payload=None):
-        if not payload:
-            payload = {}
-        payload['request'] = command
-        payload['nonce'] = int(time() * 1000)
-        if self.config.account_name:
-            payload['account'] = self.config.account_name
+        headers = generate_token(self.config.key_id, self.config.key_secret, command, account_name=self.config.account_name, payload=payload)
+
+        headers['Content-Type'] = "text/plain"
+        headers['Content-Length'] = "0"
+        headers['Cache-Control'] = "no-cache"
 
         api = self.api if not self.sandbox else self.sandbox_api
         api = f"{api}{command}"
-        b64_payload = base64.b64encode(json.dumps(payload).encode('utf-8'))
-        signature = hmac.new(self.config.key_secret.encode('utf-8'), b64_payload, hashlib.sha384).hexdigest()
-
-        headers = {
-            'Content-Type': "text/plain",
-            'Content-Length': "0",
-            'X-GEMINI-APIKEY': self.config.key_id,
-            'X-GEMINI-PAYLOAD': b64_payload,
-            'X-GEMINI-SIGNATURE': signature,
-            'Cache-Control': "no-cache"
-        }
 
         resp = requests.post(api, headers=headers)
         self._handle_error(resp, LOG)

cryptofeed/rest/gemini.py

@@ -20,7 +20,7 @@
                                 KRAKEN, KRAKEN_FUTURES, OKCOIN, OKEX, POLONIEX, PROBIT, UPBIT, WHALE_ALERT)
 from cryptofeed.defines import (FILL_OR_KILL, IMMEDIATE_OR_CANCEL, LIMIT, MAKER_OR_CANCEL, MARKET, UNSUPPORTED)
 from cryptofeed.defines import (FUNDING, FUTURES_INDEX, L2_BOOK, L3_BOOK, LIQUIDATIONS, OPEN_INTEREST, MARKET_INFO,
-                                TICKER, TRADES, TRANSACTIONS, VOLUME)
+                                TICKER, TRADES, TRANSACTIONS, VOLUME, ORDER_INFO)
 from cryptofeed.exceptions import UnsupportedDataFeed, UnsupportedTradingOption, UnsupportedSymbol
 from cryptofeed.symbols import gen_symbols, _exchange_info
 
@@ -250,6 +250,9 @@ def timestamp_normalize(exchange, ts):
     },
     FUTURES_INDEX: {
         BYBIT: 'instrument_info.100ms'
+    },
+    ORDER_INFO: {
+        GEMINI: ORDER_INFO
     }
 }
 
@@ -321,3 +324,7 @@ def raise_error():
     if ret == UNSUPPORTED:
         raise_error()
     return ret
+
+
+def is_authenticated_channel(channel: str) -> bool:
+    return channel in (ORDER_INFO)

cryptofeed/standards.py

@@ -194,7 +194,7 @@ def coinbase_symbols() -> Dict[str, str]:
         raise_failure_explanation('COINBASE', why, {"": r})
 
 
-def gemini_symbols() -> Dict[str, str]:
+def gemini_symbols(*args) -> Dict[str, str]:
     r = None
     try:
         r = requests.get('https://api.gemini.com/v1/symbols')

cryptofeed/symbols.py

@@ -0,0 +1,23 @@
+'''
+Copyright (C) 2017-2021  Bryant Moscon - [email protected]
+
+Please see the LICENSE file for the terms and conditions
+associated with this software.
+'''
+from cryptofeed import FeedHandler
+from cryptofeed.defines import GEMINI, ORDER_INFO
+
+
+async def order(**kwargs):
+    print(f"Order Update: {kwargs}")
+
+
+def main():
+    f = FeedHandler(config="config.yaml")
+    f.add_feed(GEMINI, sandbox=True, subscription={ORDER_INFO: ['BTC-USD', 'ETH-USD']}, callbacks={ORDER_INFO: order})
+
+    f.run()
+
+
+if __name__ == '__main__':
+    main()