Merge pull request abides-sim#1 from mamahfouz/master

market replay changes

README.md

@@ -1,8 +1,21 @@
+# ABIDES: Agent-Based Interactive Discrete Event Simulation environment
+
+> ABIDES is an Agent-Based Interactive Discrete Event Simulation environment. ABIDES is designed from the ground up to support AI agent research in market applications. While simulations are certainly available within trading firms for their own internal use, there are no broadly available high-fidelity market simulation environments. We hope that the availability of such a platform will facilitate AI research in this important area. ABIDES currently enables the simulation of tens of thousands of trading agents interacting with an exchange agent to facilitate transactions. It supports configurable pairwise network latencies between each individual agent as well as the exchange. Our simulator's message-based design is modeled after NASDAQ's published equity trading protocols ITCH and OUCH. 
+
 Please see our arXiv paper for preliminary documentation:
 
 https://arxiv.org/abs/1904.12066
 
-
 Please see the wiki for tutorials and example configurations:
 
 https://github.com/abides-sim/abides/wiki
+
+## Quickstart
+```
+mkdir project
+cd project
+
+git clone https://github.com/abides-sim/abides.git
+cd abides
+pip install -r requirements.txt
+```

agent/ExchangeAgent.py

@@ -9,8 +9,6 @@
 from util.OrderBook import OrderBook
 from util.util import log_print
 
-import sys
-
 import jsons as js
 import numpy as np
 import pandas as pd
@@ -68,17 +66,25 @@ def kernelInitializing (self, kernel):
 
     # Obtain opening prices (in integer cents).  These are not noisy right now.
     for symbol in self.order_books:
-      self.order_books[symbol].last_trade = self.oracle.getDailyOpenPrice(symbol, self.mkt_open)
-      log_print ("Opening price for {} is {}", symbol, self.order_books[symbol].last_trade)
+      try:
+        self.order_books[symbol].last_trade = self.oracle.getDailyOpenPrice(symbol, self.mkt_open)
+        log_print ("Opening price for {} is {}", symbol, self.order_books[symbol].last_trade)
+      except AttributeError as e:
+        log_print(str(e))
 
 
   # The exchange agent overrides this to additionally log the full depth of its
   # order books for the entire day.
   def kernelTerminating (self):
     super().kernelTerminating()
 
-    # Skip order book dump if requested.
-    if self.book_freq is None: return
+    if self.book_freq is None:
+      for symbol in self.order_books:
+        book = self.order_books[symbol]
+        dfLog = pd.DataFrame([book.mid_dict, book.bid_levels_price_dict, book.bid_levels_size_dict,
+                              book.ask_levels_price_dict, book.ask_levels_size_dict]).T
+        dfLog.columns = ['mid_price', 'bid_level_prices', 'bid_level_sizes', 'ask_level_prices', 'ask_level_sizes']
+        self.writeLog(dfLog, filename='orderbook_{}'.format(symbol))
 
     # Iterate over the order books controlled by this exchange.
     for symbol in self.order_books:
@@ -115,8 +121,11 @@ def kernelTerminating (self):
         quotes = sorted(dfLog.index.get_level_values(1).unique())
         min_quote = quotes[0]
         max_quote = quotes[-1]
-        quotes = range(min_quote, max_quote+1)
-
+        try:
+          quotes = range(min_quote, max_quote+1)
+        except Exception as e:
+          quotes = np.arange(min_quote, max_quote + 0.01, step=0.01)
+
         # Restructure the log to have multi-level rows of all possible pairs of time and quote
         # with volume as the only column.
         filledIndex = pd.MultiIndex.from_product([time_idx, quotes], names=['time','quote'])
@@ -134,7 +143,7 @@ def kernelTerminating (self):
         # to the exchange agent log.
         self.writeLog(df, filename='orderbook_{}'.format(symbol))
 
-        print ("Order book archival complete.")
+    print ("Order book archival complete.")
 
 
   def receiveMessage (self, currentTime, msg):
@@ -260,7 +269,15 @@ def receiveMessage (self, currentTime, msg):
       else:
         # Hand the order to the order book for processing.
         self.order_books[order.symbol].cancelOrder(deepcopy(order))
-
+    elif msg.body['msg'] == 'MODIFY_ORDER':
+      order = msg.body['order']
+      new_order = msg.body['new_order']
+      log_print ("{} received MODIFY_ORDER: {}, new order: {}".format(self.name, order, new_order))
+      if order.symbol not in self.order_books:
+        log_print ("Modification request discarded.  Unknown symbol: {}".format(order.symbol))
+      else:
+        self.order_books[order.symbol].modifyOrder(deepcopy(order), deepcopy(new_order))
+
 
   def sendMessage (self, recipientID, msg):
     # The ExchangeAgent automatically applies appropriate parallel processing pipeline delay

agent/ExperimentalAgent.py

@@ -0,0 +1,37 @@
+from agent.TradingAgent import TradingAgent
+
+
+class ExperimentalAgent(TradingAgent):
+
+    def __init__(self, id, name, symbol,
+                 starting_cash, execution_timestamp, quantity, is_buy_order, limit_price,
+                 log_orders = False, random_state = None):
+        super().__init__(id, name, starting_cash, random_state)
+        self.symbol              = symbol
+        self.execution_timestamp = execution_timestamp
+        self.quantity            = quantity
+        self.is_buy_order        = is_buy_order
+        self.limit_price         = limit_price
+        self.log_orders          = log_orders
+
+    def kernelStarting(self, startTime):
+        super().kernelStarting(startTime)
+
+    def wakeup(self, currentTime):
+        super().wakeup(currentTime)
+        self.last_trade[self.symbol] = 0
+        if not self.mkt_open or not self.mkt_close:
+            return
+        elif (currentTime > self.mkt_open) and (currentTime < self.mkt_close):
+            if currentTime == self.execution_timestamp:
+                self.placeLimitOrder(self.symbol, self.quantity, self.is_buy_order, self.limit_price, dollar=False)
+                if self.log_orders: self.logEvent('LIMIT_ORDER', {'agent_id': self.id, 'dollar': False, 'fill_price': None,
+                                                                  'is_buy_order': self.is_buy_order, 'limit_price': self.limit_price,
+                                                                  'order_id': 1, 'quantity': self.quantity, 'symbol': self.symbol,
+                                                                  'time_placed': str(currentTime)})
+
+    def receiveMessage(self, currentTime, msg):
+        super().receiveMessage(currentTime, msg)
+
+    def getWakeFrequency(self):
+        return self.execution_timestamp - self.mkt_open

agent/MarketReplayAgent.py

@@ -0,0 +1,90 @@
+import pandas as pd
+
+from agent.TradingAgent import TradingAgent
+from util.order.LimitOrder import LimitOrder
+from util.util import log_print
+
+
+class MarketReplayAgent(TradingAgent):
+
+
+  def __init__(self, id, name, type, symbol, date, starting_cash, log_orders = False, random_state = None):
+    super().__init__(id, name, type, starting_cash=starting_cash, log_orders=log_orders, random_state = random_state)
+    self.symbol      = symbol
+    self.date        = date
+    self.log_orders  = log_orders
+    self.state       = 'AWAITING_WAKEUP'
+
+
+  def kernelStarting(self, startTime):
+    super().kernelStarting(startTime)
+    self.oracle = self.kernel.oracle
+
+  def kernelStopping (self):
+    super().kernelStopping()
+
+  def wakeup (self, currentTime):
+    self.state = 'INACTIVE'
+    try:
+        super().wakeup(currentTime)
+        self.last_trade[self.symbol] = self.oracle.getDailyOpenPrice(self.symbol, self.mkt_open)
+        if not self.mkt_open or not self.mkt_close:
+          return
+        order = self.oracle.trades_df.loc[self.oracle.trades_df.timestamp == currentTime]
+        wake_up_time = self.oracle.trades_df.loc[self.oracle.trades_df.timestamp > currentTime].iloc[0].timestamp
+        if (currentTime > self.mkt_open) and (currentTime < self.mkt_close):
+          self.state = 'ACTIVE'
+          try:
+            self.placeOrder(currentTime, order)
+          except Exception as e:
+            log_print(e)
+        self.setWakeup(wake_up_time)
+    except Exception as e:
+        log_print(str(e))
+
+
+  def receiveMessage (self, currentTime, msg):
+    super().receiveMessage(currentTime, msg)
+
+
+  def placeOrder(self, currentTime, order):
+    if len(order) == 1:
+        type        = order.type.item()
+        id          = order.order_id.item()
+        direction   = order.direction.item()
+        price       = order.price.item()
+        vol         = order.vol.item()
+
+        existing_order = self.orders.get(id)
+
+        if type == 'NEW':
+            self.placeLimitOrder(self.symbol, vol, direction == 'BUY', float(price), dollar=False, order_id=id)
+        elif type in ['CANCELLATION', 'PARTIAL_CANCELLATION']:
+            if existing_order:
+                if type == 'CANCELLATION':
+                    self.cancelOrder(existing_order)
+                elif type == 'PARTIAL_CANCELLATION':
+                    new_order = LimitOrder(self.id, currentTime, self.symbol, vol, direction == 'BUY', float(price),
+                                           dollar=False, order_id=id)
+                    self.modifyOrder(existing_order, new_order)
+        elif type in ['EXECUTE_VISIBLE', 'EXECUTE_HIDDEN']:
+            if existing_order:
+                if existing_order.quantity == vol:
+                    self.cancelOrder(existing_order)
+                else:
+                    new_vol = existing_order.quantity - vol
+                    if new_vol == 0:
+                        self.cancelOrder(existing_order)
+                    else:
+                        executed_order = LimitOrder(self.id, currentTime, self.symbol, new_vol, direction == 'BUY', float(price),
+                                                    dollar=False, order_id=id)
+                        self.modifyOrder(existing_order, executed_order)
+                        self.orders.get(id).quantity = new_vol
+    else:
+        orders = self.oracle.trades_df.loc[self.oracle.trades_df.timestamp == currentTime]
+        for index, order in orders.iterrows():
+            self.placeOrder(currentTime, order = pd.DataFrame(order).T)
+
+
+  def getWakeFrequency(self):
+    return self.oracle.trades_df.iloc[0].timestamp - self.mkt_open

agent/RandomAgent.py

@@ -0,0 +1,43 @@
+from agent.TradingAgent import TradingAgent
+import numpy as np
+import pandas as pd
+
+
+class RandomAgent(TradingAgent):
+
+
+    def __init__(self, id, name, symbol, startingCash,
+                 buy_price_range = [90, 105], sell_price_range = [95, 110], quantity_range = [50, 500],
+                 random_state = None):
+        super().__init__(id, name, startingCash, random_state)
+        self.symbol           = symbol
+        self.buy_price_range  = buy_price_range
+        self.sell_price_range = sell_price_range
+        self.quantity_range   = quantity_range
+
+
+    def kernelStarting(self, startTime):
+        super().kernelStarting(startTime)
+
+
+    def wakeup(self, currentTime):
+        super().wakeup(currentTime)
+        self.last_trade[self.symbol] = 0
+        if not self.mkt_open or not self.mkt_close:
+            return
+        elif (currentTime > self.mkt_open) and (currentTime < self.mkt_close):
+            direction = np.random.randint(0, 2)
+            price = np.random.randint(self.buy_price_range[0], self.buy_price_range[1]) \
+                if direction == 1 else np.random.randint(self.sell_price_range[0], self.sell_price_range[1])
+            quantity = np.random.randint(self.quantity_range[0], self.quantity_range[1])
+            self.placeLimitOrder(self.symbol, quantity, direction, price, dollar=False)
+        delta_time = self.random_state.exponential(scale=1.0 / 0.005)
+        self.setWakeup(currentTime + pd.Timedelta('{}ms'.format(int(round(delta_time)))))
+
+
+    def receiveMessage(self, currentTime, msg):
+        super().receiveMessage(currentTime, msg)
+
+
+    def getWakeFrequency(self):
+        return pd.Timedelta('1ms')

agent/TradingAgent.py

@@ -253,8 +253,8 @@ def getOrderStream (self, symbol, length=1):
 
   # Used by any Trading Agent subclass to place a limit order.  Parameters expect:
   # string (valid symbol), int (positive share quantity), bool (True == BUY), int (price in cents).
-  def placeLimitOrder (self, symbol, quantity, is_buy_order, limit_price, ignore_risk = False):
-    order = LimitOrder(self.id, self.currentTime, symbol, quantity, is_buy_order, limit_price)
+  def placeLimitOrder (self, symbol, quantity, is_buy_order, limit_price, dollar=True, order_id=None, ignore_risk = False):
+    order = LimitOrder(self.id, self.currentTime, symbol, quantity, is_buy_order, limit_price, dollar, order_id)
 
     if quantity > 0:
       # Test if this order can be permitted given our at-risk limits.
@@ -299,6 +299,15 @@ def cancelOrder (self, order):
     # Log this activity.
     if self.log_orders: self.logEvent('CANCEL_SUBMITTED', js.dump(order))
 
+  # Used by any Trading Agent subclass to modify any existing limitorder.  The order must currently
+  # appear in the agent's open orders list.
+  def modifyOrder (self, order, newOrder):
+    self.sendMessage(self.exchangeID, Message({ "msg" : "MODIFY_ORDER", "sender": self.id,
+                                                "order" : order, "new_order" : newOrder}))
+
+    # Log this activity.
+    if self.log_orders: self.logEvent('MODIFY_ORDER', js.dump(order))
+
 
   # Handles ORDER_EXECUTED messages from an exchange agent.  Subclasses may wish to extend,
   # but should still call parent method for basic portfolio/returns tracking.

config/impact.py

@@ -7,7 +7,6 @@
 from util.oracle.MeanRevertingOracle import MeanRevertingOracle
 from util import util
 
-import datetime as dt
 import numpy as np
 import pandas as pd
 import sys