updates

aat/callback.py

@@ -9,10 +9,6 @@ class Callback(metaclass=ABCMeta):
     def onTrade(self, data: MarketData):
         '''onTrade'''
 
-    @abstractmethod
-    def onReceived(self, data: MarketData):
-        '''onReceived'''
-
     @abstractmethod
     def onOpen(self, data: MarketData):
         '''onOpen'''
@@ -88,7 +84,6 @@ def __init__(self,
                  onCancel=True,
                  onChange=True,
                  onError=True):
-
         if not onTrade:
             setattr(self, 'onTrade', False)
         if not onReceived:

aat/data_source.py

@@ -1,4 +1,5 @@
 from abc import ABCMeta, abstractmethod
+from typing import Set
 from .callback import Callback
 from .structs import TradeRequest, TradeResponse
 from .enums import TickType

aat/exchange.py

@@ -44,6 +44,4 @@ def callback_data(self, data) -> None:
             pass
 
         if res.type != TickType.HEARTBEAT:
-            if self._query_engine:
-                self._query_engine.push(res)
             self.callback(res.type, res)

aat/exchanges/coinbase.py

@@ -1,79 +1,14 @@
 import json
-from datetime import datetime
 from functools import lru_cache
-from ..enums import OrderType, OrderSubType, PairType, TickType
+from .utils.coinbase import CoinbaseMixins
 from ..exchange import Exchange
-from ..structs import MarketData, Instrument
-from ..utils import parse_date, str_to_currency_pair_type, str_to_side, str_to_order_type
 
 
-class CoinbaseExchange(Exchange):
+class CoinbaseExchange(CoinbaseMixins, Exchange):
     @lru_cache(None)
     def subscription(self):
         return [json.dumps({"type": "subscribe", "product_id": self.currencyPairToString(x)}) for x in self.options().currency_pairs]
 
     @lru_cache(None)
     def heartbeat(self):
         return json.dumps({"type": "heartbeat", "on": True})
-
-    def tickToData(self, jsn: dict) -> MarketData:
-        if jsn.get('type') == 'received':
-            return
-        typ = self.strToTradeType(jsn.get('type'), jsn.get('reason', ''))
-        time = parse_date(jsn.get('time')) if jsn.get('time') else datetime.now()
-        price = float(jsn.get('price', 'nan'))
-        volume = float(jsn.get('size', 'nan'))
-        currency_pair = str_to_currency_pair_type(jsn.get('product_id')) if typ != TickType.ERROR else PairType.NONE
-
-        instrument = Instrument(underlying=currency_pair)
-
-        order_type = str_to_order_type(jsn.get('order_type', ''))
-        side = str_to_side(jsn.get('side', ''))
-        remaining_volume = float(jsn.get('remaining_size', 0.0))
-
-        sequence = int(jsn.get('sequence', -1))
-        ret = MarketData(time=time,
-                         volume=volume,
-                         price=price,
-                         type=typ,
-                         instrument=instrument,
-                         remaining=remaining_volume,
-                         side=side,
-                         exchange=self.exchange(),
-                         order_type=order_type,
-                         sequence=sequence)
-        return ret
-
-    def strToTradeType(self, s: str, reason: str = '') -> TickType:
-        if s == 'match':
-            return TickType.TRADE
-        elif s in ('open', 'done', 'change', 'heartbeat'):
-            if reason == 'canceled':
-                return TickType.CANCEL
-            elif reason == 'filled':
-                return TickType.FILL
-            return TickType(s.upper())
-        else:
-            return TickType.ERROR
-
-    def tradeReqToParams(self, req) -> dict:
-        p = {}
-        p['price'] = str(req.price)
-        p['size'] = str(req.volume)
-        p['product_id'] = self.currencyPairToString(req.instrument.currency_pair)
-        p['type'] = self.orderTypeToString(req.order_type)
-
-        if req.order_sub_type == OrderSubType.FILL_OR_KILL:
-            p['time_in_force'] = 'FOK'
-        elif req.order_sub_type == OrderSubType.POST_ONLY:
-            p['post_only'] = '1'
-        return p
-
-    def currencyPairToString(self, cur: PairType) -> str:
-        return cur.value[0].value + '-' + cur.value[1].value
-
-    def orderTypeToString(self, typ: OrderType) -> str:
-        return typ.value.lower()
-
-    def reasonToTradeType(self, s: str) -> TickType:
-        pass

aat/exchanges/gemini.py

@@ -1,17 +1,15 @@
 import aiohttp
 import json
 from aiostream import stream
-from datetime import datetime
 from functools import lru_cache
+from .utils.gemini import GeminiMixins
 from ..define import EXCHANGE_MARKET_DATA_ENDPOINT
-from ..enums import Side, OrderType, OrderSubType, PairType, TickType
+from ..enums import TickType
 from ..exchange import Exchange
 from ..logging import LOG as log
-from ..structs import MarketData, Instrument, TradeResponse
-from ..utils import str_to_currency_pair_type, str_to_side
 
 
-class GeminiExchange(Exchange):
+class GeminiExchange(GeminiMixins, Exchange):
     @lru_cache(None)
     def subscription(self):
         return [json.dumps({"type": "subscribe", "product_id": self.currencyPairToString(x)}) for x in self.options().currency_pairs]
@@ -65,105 +63,4 @@ async def get_data_sub_pair(ws, sub=None):
                         pass
 
                     if res.type != TickType.HEARTBEAT:
-                        if self._query_engine:
-                            self._query_engine.push(res)
-
-
-                    if res.type == TickType.TRADE:
-                        self._last = res
-                        self.callback(TickType.TRADE, res)
-                    elif res.type == TickType.RECEIVED:
-                        self.callback(TickType.RECEIVED, res)
-                    elif res.type == TickType.OPEN:
-                        self.callback(TickType.OPEN, res)
-                    elif res.type == TickType.FILL:
-                        self.callback(TickType.FILL, res)
-                    elif res.type == TickType.CANCEL:
-                        self.callback(TickType.CANCEL, res)
-                    elif res.type == TickType.CHANGE:
-                        self.callback(TickType.CHANGE, res)
-                    elif res.type == TickType.HEARTBEAT:
-                        # TODO anything?
-                        pass
-                    else:
-                        self.callback(TickType.ERROR, res)
-
-    def cancel(self, resp: TradeResponse):
-        '''cancel an order'''
-        raise NotImplementedError()
-
-    def cancelAll(self) -> None:
-        '''cancel all orders'''
-        log.critical('Cancelling all active orders')
-        self._client.cancel_all_active_orders()
-
-    def tickToData(self, jsn: dict) -> MarketData:
-        time = datetime.now()
-        price = float(jsn.get('price', 'nan'))
-        volume = float(jsn.get('amount', 0.0))
-        typ = self.strToTradeType(jsn.get('type'))
-        delta = float(jsn.get('delta', 0.0))
-
-        if typ == TickType.CHANGE and not volume:
-            delta = float(jsn.get('delta', 'nan'))
-            volume = delta
-            # typ = self.reasonToTradeType(reason)
-
-        side = str_to_side(jsn.get('side', ''))
-        remaining_volume = float(jsn.get('remaining', 'nan'))
-        sequence = -1
-
-        if 'symbol' not in jsn:
-            return
-
-        currency_pair = str_to_currency_pair_type(jsn.get('symbol'))
-        instrument = Instrument(underlying=currency_pair)
-
-        ret = MarketData(time=time,
-                         volume=volume,
-                         price=price,
-                         type=typ,
-                         instrument=instrument,
-                         remaining=remaining_volume,
-
-                         side=side,
-                         exchange=self.exchange(),
-                         sequence=sequence)
-        return ret
-
-    def strToTradeType(self, s: str) -> TickType:
-        return TickType(s.upper())
-
-    def reasonToTradeType(self, s: str) -> TickType:
-        s = s.upper()
-        if 'CANCEL' in s:
-            return TickType.CANCEL
-        if 'PLACE' in s:
-            return TickType.OPEN
-        if 'INITIAL' in s:
-            return TickType.OPEN
-
-    def tradeReqToParams(self, req) -> dict:
-        p = {}
-        p['price'] = str(req.price)
-        p['size'] = str(req.volume)
-        p['product_id'] = self.currencyPairToString(req.instrument.currency_pair)
-        p['type'] = self.orderTypeToString(req.order_type)
-
-        if p['type'] == OrderType.MARKET:
-            if req.side == Side.BUY:
-                p['price'] = 100000000.0
-            else:
-                p['price'] = .00000001
-
-        if req.order_sub_type == OrderSubType.FILL_OR_KILL:
-            p['time_in_force'] = 'FOK'
-        elif req.order_sub_type == OrderSubType.POST_ONLY:
-            p['post_only'] = '1'
-        return p
-
-    def currencyPairToString(self, cur: PairType) -> str:
-        return cur.value[0].value + cur.value[1].value
-
-    def orderTypeToString(self, typ: OrderType) -> str:
-        return typ.value.lower()
+                        self.callback(res.type, res)

aat/exchanges/kraken.py

@@ -1,12 +1,13 @@
 import json
 from functools import lru_cache
+from .utils.kraken import KrakenMixins
 from ..config import ExchangeConfig
 from ..enums import ExchangeType, OrderType, PairType, TickType
 from ..exchange import Exchange
 from ..structs import MarketData
 
 
-class KrakenExchange(Exchange):
+class KrakenExchange(KrakenMixins, Exchange):
     def __init__(self, exchange_type: ExchangeType, options: ExchangeConfig) -> None:
         super(KrakenExchange, self).__init__(exchange_type, options)
         self._last = None

aat/exchanges/poloniex.py

@@ -1,13 +1,12 @@
 import json
 from functools import lru_cache
 from ..config import ExchangeConfig
-from ..enums import ExchangeType, OrderType, PairType, TickType
+from ..enums import ExchangeType
 from ..exchange import Exchange
-from ..structs import MarketData
-from .utils.poloniex import POLONIEX_PAIR_ID
+from .utils.poloniex import POLONIEX_PAIR_ID, PoloniexMixins
 
 
-class PoloniexExchange(Exchange):
+class PoloniexExchange(PoloniexMixins, Exchange):
     def __init__(self, exchange_type: ExchangeType, options: ExchangeConfig) -> None:
         super(PoloniexExchange, self).__init__(exchange_type, options)
         self._last = None
@@ -21,18 +20,3 @@ def subscription(self):
     @lru_cache(None)
     def heartbeat(self):
         return json.dumps({"command": "subscribe", "channel": "1010"})
-
-    def tickToData(self, jsn: dict) -> MarketData:
-        raise NotImplementedError()
-
-    def strToTradeType(self, s: str) -> TickType:
-        raise NotImplementedError()
-
-    def tradeReqToParams(self, req) -> dict:
-        raise NotImplementedError()
-
-    def currencyPairToString(self, cur: PairType) -> str:
-        return cur.value[0].value + '_' + cur.value[1].value
-
-    def orderTypeToString(self, typ: OrderType) -> str:
-        raise NotImplementedError()

aat/exchanges/utils/coinbase.py

@@ -0,0 +1,68 @@
+from datetime import datetime
+from ...enums import OrderType, OrderSubType, PairType, TickType
+from ...structs import MarketData, Instrument
+from ...utils import parse_date, str_to_currency_pair_type, str_to_side, str_to_order_type
+
+
+class CoinbaseMixins(object):
+    def tickToData(self, jsn: dict) -> MarketData:
+        if jsn.get('type') == 'received':
+            return
+        typ = self.strToTradeType(jsn.get('type'), jsn.get('reason', ''))
+        time = parse_date(jsn.get('time')) if jsn.get('time') else datetime.now()
+        price = float(jsn.get('price', 'nan'))
+        volume = float(jsn.get('size', 'nan'))
+        currency_pair = str_to_currency_pair_type(jsn.get('product_id')) if typ != TickType.ERROR else PairType.NONE
+
+        instrument = Instrument(underlying=currency_pair)
+
+        order_type = str_to_order_type(jsn.get('order_type', ''))
+        side = str_to_side(jsn.get('side', ''))
+        remaining_volume = float(jsn.get('remaining_size', 0.0))
+
+        sequence = int(jsn.get('sequence', -1))
+        ret = MarketData(time=time,
+                         volume=volume,
+                         price=price,
+                         type=typ,
+                         instrument=instrument,
+                         remaining=remaining_volume,
+                         side=side,
+                         exchange=self.exchange(),
+                         order_type=order_type,
+                         sequence=sequence)
+        return ret
+
+    def strToTradeType(self, s: str, reason: str = '') -> TickType:
+        if s == 'match':
+            return TickType.TRADE
+        elif s in ('open', 'done', 'change', 'heartbeat'):
+            if reason == 'canceled':
+                return TickType.CANCEL
+            elif reason == 'filled':
+                return TickType.FILL
+            return TickType(s.upper())
+        else:
+            return TickType.ERROR
+
+    def tradeReqToParams(self, req) -> dict:
+        p = {}
+        p['price'] = str(req.price)
+        p['size'] = str(req.volume)
+        p['product_id'] = self.currencyPairToString(req.instrument.currency_pair)
+        p['type'] = self.orderTypeToString(req.order_type)
+
+        if req.order_sub_type == OrderSubType.FILL_OR_KILL:
+            p['time_in_force'] = 'FOK'
+        elif req.order_sub_type == OrderSubType.POST_ONLY:
+            p['post_only'] = '1'
+        return p
+
+    def currencyPairToString(self, cur: PairType) -> str:
+        return cur.value[0].value + '-' + cur.value[1].value
+
+    def orderTypeToString(self, typ: OrderType) -> str:
+        return typ.value.lower()
+
+    def reasonToTradeType(self, s: str) -> TickType:
+        pass