#8 and #18 in progress

aat/callback.py

@@ -18,8 +18,12 @@ def onOpen(self, data: MarketData):
         '''onOpen'''
 
     @abstractmethod
-    def onDone(self, data: MarketData):
-        '''onDone'''
+    def onFill(self, data: MarketData):
+        '''onFill'''
+
+    @abstractmethod
+    def onCancel(self, data: MarketData):
+        '''onCancel'''
 
     @abstractmethod
     def onChange(self, data: MarketData):
@@ -62,7 +66,10 @@ def onReceived(self, data: MarketData) -> None:
     def onOpen(self, data: MarketData) -> None:
         pass
 
-    def onDone(self, data: MarketData) -> None:
+    def onFill(self, data: MarketData) -> None:
+        pass
+
+    def onCancel(self, data: MarketData) -> None:
         pass
 
     def onChange(self, data: MarketData) -> None:
@@ -77,7 +84,8 @@ def __init__(self,
                  onTrade=True,
                  onReceived=True,
                  onOpen=True,
-                 onDone=True,
+                 onFill=True,
+                 onCancel=True,
                  onChange=True,
                  onError=True):
 
@@ -87,8 +95,10 @@ def __init__(self,
             setattr(self, 'onReceived', False)
         if not onOpen:
             setattr(self, 'onOpen', False)
-        if not onDone:
-            setattr(self, 'onDone', False)
+        if not onFill:
+            setattr(self, 'onFilled', False)
+        if not onCancel:
+            setattr(self, 'onCancelled', False)
         if not onChange:
             setattr(self, 'onChange', False)
         if not onError:
@@ -103,7 +113,10 @@ def onReceived(self, data: MarketData) -> None:
     def onOpen(self, data: MarketData) -> None:
         dlog.info(str(data))
 
-    def onDone(self, data: MarketData) -> None:
+    def onFill(self, data: MarketData) -> None:
+        dlog.info(str(data))
+
+    def onCancel(self, data: MarketData) -> None:
         dlog.info(str(data))
 
     def onChange(self, data: MarketData) -> None:

aat/config.py

@@ -1,12 +1,14 @@
 from .utils import config
-from .enums import TradingType, ExchangeType, PairType
+from .enums import TradingType, ExchangeType, PairType, InstrumentType
+from .structs import Instrument
 
 
 @config
 class ExchangeConfig:
     exchange_types = [ExchangeType], []
     trading_type = TradingType, TradingType.NONE
     currency_pairs = [PairType], [PairType.BTCUSD]
+    instruments = [Instrument], [Instrument(type=InstrumentType.PAIR, underlying=PairType.BTCUSD)]
 
 
 @config

aat/data_source.py

@@ -42,10 +42,10 @@ class StreamingDataSource(DataSource):
     def __init__(self, *args, **kwargs) -> None:
         self._running = False
         self._callbacks = {TickType.TRADE: [],
-                           TickType.RECEIVED: [],
                            TickType.ERROR: [],
                            TickType.OPEN: [],
-                           TickType.DONE: [],
+                           TickType.FILL: [],
+                           TickType.CANCEL: [],
                            TickType.CHANGE: [],
                            TickType.ANALYZE: [],
                            TickType.HALT: [],
@@ -79,14 +79,14 @@ def tickToData(self, jsn):
     def onTrade(self, callback: Callback) -> None:
         self._callbacks[TickType.TRADE].append(callback)
 
-    def onReceived(self, callback: Callback) -> None:
-        self._callbacks[TickType.RECEIVED].append(callback)
-
     def onOpen(self, callback: Callback) -> None:
         self._callbacks[TickType.OPEN].append(callback)
 
-    def onDone(self, callback: Callback) -> None:
-        self._callbacks[TickType.DONE].append(callback)
+    def onFill(self, callback: Callback) -> None:
+        self._callbacks[TickType.FILL].append(callback)
+
+    def onCancel(self, callback: Callback) -> None:
+        self._callbacks[TickType.CANCEL].append(callback)
 
     def onChange(self, callback: Callback) -> None:
         self._callbacks[TickType.CHANGE].append(callback)
@@ -110,22 +110,14 @@ def registerCallback(self, callback: Callback) -> None:
         if not isinstance(callback, Callback):
             raise Exception('%s is not an instance of class '
                             'Callback' % callback)
-
-        if callback.onTrade:
-            self.onTrade(callback.onTrade)
-        if callback.onReceived:
-            self.onReceived(callback.onReceived)
-        if callback.onOpen:
-            self.onOpen(callback.onOpen)
-        if callback.onDone:
-            self.onDone(callback.onDone)
-        if callback.onChange:
-            self.onChange(callback.onChange)
-        if callback.onError:
-            self.onError(callback.onError)
-        if callback.onAnalyze:
-            self.onAnalyze(callback.onAnalyze)
-        if callback.onHalt:
-            self.onHalt(callback.onHalt)
-        if callback.onContinue:
-            self.onContinue(callback.onContinue)
+        for att in ['onTrade',
+                    'onOpen',
+                    'onFill',
+                    'onCancel',
+                    'onChange',
+                    'onError',
+                    'onAnalyze',
+                    'onHalt',
+                    'onContinue']:
+            if hasattr(callback, att):
+                getattr(self, att)(getattr(callback, att))

aat/enums.py

@@ -10,13 +10,14 @@ def members(cls):
 class TickType(BaseEnum):
     # Messages
     TRADE = 'TRADE'  # Match
-    RECEIVED = 'RECEIVED'  # Order received
     OPEN = 'OPEN'  # New Order
-    DONE = 'DONE'  # Order completed, either filled or cancelled
+    FILL = 'FILL'  # Order completed - filled
+    CANCEL = 'CANCEL'  # Order completed - cancelled
     CHANGE = 'CHANGE'  # Order modified
 
     ERROR = 'ERROR'  # Internal error
     ANALYZE = 'ANALYZE'  # Internal
+
     HALT = 'HALT'  # Trading halt
     CONTINUE = 'CONTINUE'  # Trading continue
     EXIT = 'EXIT'  # System exit
@@ -136,13 +137,6 @@ class OrderSubType(BaseEnum):
     # ALL_OR_NOTHING = 3
 
 
-class ChangeReason(BaseEnum):
-    NONE = 'NONE'
-    OPENED = 'OPENED'
-    CANCELLED = 'CANCELLED'
-    FILLED = 'FILLED'
-
-
 class TradeResult(BaseEnum):
     NONE = 'NONE'
     PENDING = 'PENDING'

aat/exchange.py

@@ -8,13 +8,14 @@
 
 
 class Exchange(MarketData, OrderEntry):
-    def __init__(self, exchange_type: ExchangeType, options: ExchangeConfig) -> None:
+    def __init__(self,
+                 exchange_type: ExchangeType,
+                 options: ExchangeConfig,
+                 query_engine=None) -> None:
         super(Exchange, self).__init__()
         self._options = options
         self._exchange = exchange_type
-        self._pending_orders = {}
-        self._messages = {}
-        self._messages_all = []
+        self._query_engine = query_engine
 
     @lru_cache(None)
     def options(self) -> ExchangeConfig:
@@ -33,6 +34,8 @@ async def receive(self) -> None:
 
     def callback_data(self, data) -> None:
         res = self.tickToData(data)
+        if res is None:
+            return
 
         if self._seqnum_enabled and res.type != TickType.HEARTBEAT:
             self.seqnum(res.sequence)
@@ -41,33 +44,6 @@ def callback_data(self, data) -> None:
             pass
 
         if res.type != TickType.HEARTBEAT:
-            if res.type not in self._messages:
-                self._messages[res.type] = [res]
-            else:
-                self._messages[res.type].append(res)
-            self._messages_all.append(res)
-
-        if res.type == TickType.TRADE:
-            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.DONE:
-            self.callback(TickType.DONE, 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 messages(self, by_type=False, instrument=None) -> list:
-        if by_type:
-            if instrument:
-                return {x: [y for y in self._messages[x] if y.instrument == instrument] for x in self._messages}
-            return self._messages
-        if instrument:
-            return [x for x in self._messages_all if x.instrument == instrument]
-        return self._messages_all
+            if self._query_engine:
+                self._query_engine.push(res)
+            self.callback(res.type, res)

aat/exchanges/coinbase.py

@@ -1,7 +1,7 @@
 import json
 from datetime import datetime
 from functools import lru_cache
-from ..enums import OrderType, OrderSubType, PairType, TickType, ChangeReason
+from ..enums import OrderType, OrderSubType, PairType, TickType
 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
@@ -17,8 +17,9 @@ def heartbeat(self):
         return json.dumps({"type": "heartbeat", "on": True})
 
     def tickToData(self, jsn: dict) -> MarketData:
-        typ = self.strToTradeType(jsn.get('type'))
-        reason = jsn.get('reason', '')
+        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'))
@@ -30,35 +31,27 @@ def tickToData(self, jsn: dict) -> MarketData:
         side = str_to_side(jsn.get('side', ''))
         remaining_volume = float(jsn.get('remaining_size', 0.0))
 
-        if reason == 'canceled':
-            reason = ChangeReason.CANCELLED
-        elif reason == '':
-            reason = ChangeReason.NONE
-        elif reason == 'filled':
-            # FIXME
-            reason = ChangeReason.NONE
-            # reason = ChangeReason.FILLED
-        else:
-            reason = ChangeReason.NONE
-
         sequence = int(jsn.get('sequence', -1))
         ret = MarketData(time=time,
                          volume=volume,
                          price=price,
                          type=typ,
                          instrument=instrument,
                          remaining=remaining_volume,
-                         reason=reason,
                          side=side,
                          exchange=self.exchange(),
                          order_type=order_type,
                          sequence=sequence)
         return ret
 
-    def strToTradeType(self, s: str) -> TickType:
+    def strToTradeType(self, s: str, reason: str = '') -> TickType:
         if s == 'match':
             return TickType.TRADE
-        elif s in ('received', 'open', 'done', 'change', 'heartbeat'):
+        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

aat/exchanges/gemini.py

@@ -4,7 +4,7 @@
 from datetime import datetime
 from functools import lru_cache
 from ..define import EXCHANGE_MARKET_DATA_ENDPOINT
-from ..enums import Side, OrderType, OrderSubType, PairType, TickType, ChangeReason
+from ..enums import Side, OrderType, OrderSubType, PairType, TickType
 from ..exchange import Exchange
 from ..logging import LOG as log
 from ..structs import MarketData, Instrument, TradeResponse
@@ -65,11 +65,9 @@ async def get_data_sub_pair(ws, sub=None):
                         pass
 
                     if res.type != TickType.HEARTBEAT:
-                        if res.type not in self._messages:
-                            self._messages[res.type] = [res]
-                        else:
-                            self._messages[res.type].append(res)
-                        self._messages_all.append(res)
+                        if self._query_engine:
+                            self._query_engine.push(res)
+
 
                     if res.type == TickType.TRADE:
                         self._last = res
@@ -78,8 +76,10 @@ async def get_data_sub_pair(ws, sub=None):
                         self.callback(TickType.RECEIVED, res)
                     elif res.type == TickType.OPEN:
                         self.callback(TickType.OPEN, res)
-                    elif res.type == TickType.DONE:
-                        self.callback(TickType.DONE, 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:
@@ -100,7 +100,6 @@ def cancelAll(self) -> None:
     def tickToData(self, jsn: dict) -> MarketData:
         time = datetime.now()
         price = float(jsn.get('price', 'nan'))
-        reason = jsn.get('reason', '')
         volume = float(jsn.get('amount', 0.0))
         typ = self.strToTradeType(jsn.get('type'))
         delta = float(jsn.get('delta', 0.0))
@@ -112,14 +111,6 @@ def tickToData(self, jsn: dict) -> MarketData:
 
         side = str_to_side(jsn.get('side', ''))
         remaining_volume = float(jsn.get('remaining', 'nan'))
-
-        if reason == 'canceled':
-            reason = ChangeReason.CANCELLED
-        elif reason == 'place' or reason == 'initial':
-            reason = ChangeReason.OPENED
-        else:
-            reason = ChangeReason.NONE
-
         sequence = -1
 
         if 'symbol' not in jsn:
@@ -134,7 +125,7 @@ def tickToData(self, jsn: dict) -> MarketData:
                          type=typ,
                          instrument=instrument,
                          remaining=remaining_volume,
-                         reason=reason,
+
                          side=side,
                          exchange=self.exchange(),
                          sequence=sequence)
@@ -146,7 +137,7 @@ def strToTradeType(self, s: str) -> TickType:
     def reasonToTradeType(self, s: str) -> TickType:
         s = s.upper()
         if 'CANCEL' in s:
-            return TickType.DONE
+            return TickType.CANCEL
         if 'PLACE' in s:
             return TickType.OPEN
         if 'INITIAL' in s:

aat/execution.py

@@ -8,7 +8,6 @@
 class Execution(object):
     def __init__(self, options: ExecutionConfig, exchange: Exchange) -> None:
         self.trading_type = options.trading_type
-
         self._ex = exchange
         self._exs = []