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

@@ -115,8 +115,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'])
@@ -260,6 +263,22 @@ 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))
+    elif msg.body['msg'] == 'REPLICATE_ORDERBOOK_SNAPSHOT':
+      timestamp = msg.body['timestamp']
+      symbol    = msg.body['symbol']
+      log_print ("{} received REPLICATE_ORDERBOOK_SNAPSHOT for t= {}".format(self.name, timestamp))
+      if symbol not in self.order_books:
+        log_print ("Orderbook replication request discarded.  Unknown symbol: {}".format(symbol))
+      else:
+        self.order_books[symbol].replicateOrderbookSnapshot()
 
 
   def sendMessage (self, recipientID, msg):

agent/ExperimentalAgent.py

@@ -0,0 +1,35 @@
+import pandas as pd
+
+from agent.TradingAgent import TradingAgent
+
+
+class ExperimentalAgent(TradingAgent):
+
+    def __init__(self, id, name, symbol,
+                 startingCash, execution_timestamp, quantity, is_buy_order, limit_price,
+                 random_state = None):
+        super().__init__(id, name, startingCash, 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.timestamp           = pd.Timestamp("2012-06-21 09:30:02")
+
+    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)
+
+    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,113 @@
+import pandas as pd
+
+from agent.TradingAgent import TradingAgent
+from util.util import log_print
+from util.order.LimitOrder import LimitOrder
+from message.Message import Message
+
+class MarketReplayAgent(TradingAgent):
+
+
+  def __init__(self, id, name, symbol, date, startingCash, log_orders = False, random_state = None):
+    super().__init__(id, name, startingCash, random_state)
+    self.symbol      = symbol
+    self.date        = date
+    self.log_orders = log_orders
+
+
+  def kernelStarting(self, startTime):
+    super().kernelStarting(startTime)
+    self.oracle = self.kernel.oracle
+
+
+  def wakeup (self, currentTime):
+    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
+    elif currentTime == self.oracle.orderbook_df.iloc[0].name:
+      order = self.oracle.trades_df.loc[self.oracle.trades_df.timestamp == currentTime]
+      self.placeMktOpenOrders(order, t=currentTime)
+      self.setWakeup(self.oracle.trades_df.loc[self.oracle.trades_df.timestamp > currentTime].iloc[0].timestamp)
+    elif (currentTime > self.mkt_open) and (currentTime < self.mkt_close):
+      try:
+        order = self.oracle.trades_df.loc[self.oracle.trades_df.timestamp == currentTime]
+        self.placeOrder(currentTime, order)
+        self.setWakeup(self.oracle.trades_df.loc[self.oracle.trades_df.timestamp > currentTime].iloc[0].timestamp)
+      except Exception as e:
+        log_print(e)
+
+
+  def receiveMessage (self, currentTime, msg):
+    super().receiveMessage(currentTime, msg)
+
+
+  def placeMktOpenOrders(self, snapshot_order, t=0):
+      orders_snapshot = self.oracle.orderbook_df.loc[self.oracle.orderbook_df.index == t].T
+      for i in range(0, len(orders_snapshot) - 1, 4):
+          ask_price = orders_snapshot.iloc[i][0]
+          ask_vol = orders_snapshot.iloc[i + 1][0]
+          bid_price = orders_snapshot.iloc[i + 2][0]
+          bid_vol = orders_snapshot.iloc[i + 3][0]
+
+          if snapshot_order.direction.item() == 'BUY' and bid_price == snapshot_order.price.item():
+              bid_vol -= snapshot_order.vol.item()
+          elif snapshot_order.direction.item() == 'SELL' and ask_price == snapshot_order.price.item():
+              ask_vol -= snapshot_order.vol.item()
+
+          self.placeLimitOrder(self.symbol, bid_vol, True, float(bid_price), dollar=False)
+          self.placeLimitOrder(self.symbol, ask_vol, False, float(ask_price), dollar=False)
+      self.placeOrder(snapshot_order.timestamp.item(), snapshot_order)
+
+
+  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()
+        if type == 'NEW':
+            self.placeLimitOrder(self.symbol, vol, direction == 'BUY', float(price), dollar=False, order_id=id)
+        elif type in ['CANCELLATION', 'PARTIAL_CANCELLATION']:
+            existing_order = self.orders.get(id)
+            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)
+            else:
+                self.replicateOrderbookSnapshot(currentTime)
+        elif type in ['EXECUTE_VISIBLE', 'EXECUTE_HIDDEN']:
+            existing_order = self.orders.get(id)
+            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:
+                self.replicateOrderbookSnapshot(currentTime)
+    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 replicateOrderbookSnapshot(self, currentTime):
+       log_print("Received notification of orderbook snapshot replication at: {}".format(currentTime))
+       self.sendMessage(self.exchangeID, Message({"msg": "REPLICATE_ORDERBOOK_SNAPSHOT", "sender": self.id,
+                                                  "symbol": self.symbol, "timestamp": str(currentTime)}))
+       if self.log_orders: self.logEvent('REPLICATE_ORDERBOOK_SNAPSHOT', currentTime)
+
+
+  def getWakeFrequency(self):
+    return self.oracle.trades_df.iloc[0].timestamp - self.mkt_open

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

config/marketreplay.py

@@ -0,0 +1,146 @@
+from Kernel import Kernel
+
+from agent.MarketReplayAgent import MarketReplayAgent
+from agent.ExchangeAgent import ExchangeAgent
+from agent.ExperimentalAgent import ExperimentalAgent
+from util.oracle.OrderBookOracle import OrderBookOracle
+
+from util import util
+from util.order import LimitOrder
+
+import datetime as dt
+import numpy as np
+import pandas as pd
+import sys
+import argparse
+
+parser = argparse.ArgumentParser(description='Options for Market Replay Agent Config.')
+
+# General Config for all agents
+parser.add_argument('-c', '--config', required=True,
+                    help='Name of config file to execute')
+parser.add_argument('-s', '--seed', type=int, default=None,
+                    help='numpy.random.seed() for simulation')
+parser.add_argument('-l', '--log_dir', default=None,
+                    help='Log directory name (default: unix timestamp at program start)')
+parser.add_argument('-v', '--verbose', action='store_true',
+                    help='Maximum verbosity!')
+parser.add_argument('-o', '--log_orders', action='store_true',
+                    help='Log every order-related action by every agent.')
+parser.add_argument('--config_help', action='store_true',
+                    help='Print argument options for this config file')
+
+args, remaining_args = parser.parse_known_args()
+
+log_orders = args.log_orders
+
+if args.config_help:
+  parser.print_help()
+  sys.exit()
+
+# Simulation Start Time
+simulation_start_time = dt.datetime.now()
+print ("Simulation Start Time: {}".format(simulation_start_time))
+
+# Random Seed Config
+seed = args.seed
+if not seed: seed = int(pd.Timestamp.now().timestamp() * 1000000) % (2**32 - 1)
+np.random.seed(seed)
+print ("Configuration seed: {}".format(seed))
+
+random_state = np.random.RandomState(seed=np.random.randint(low=1))
+
+util.silent_mode = not args.verbose
+LimitOrder.silent_mode = not args.verbose
+print ("Silent mode: {}".format(util.silent_mode))
+
+######################## Agents Config #########################################################################
+
+# 1) Symbols
+symbols = ['AAPL']
+print("Symbols traded: {}".format(symbols))
+
+# 2) Historical Date to simulate
+date = '2012-06-21'
+date_pd = pd.to_datetime(date)
+print("Historical Simulation Date: {}".format(date))
+
+agents = []
+
+# 3) ExchangeAgent Config
+num_exchanges = 1
+mkt_open  = date_pd + pd.to_timedelta('09:30:00')
+mkt_close = date_pd + pd.to_timedelta('09:30:05')
+print("ExchangeAgent num_exchanges: {}".format(num_exchanges))
+print("ExchangeAgent mkt_open: {}".format(mkt_open))
+print("ExchangeAgent mkt_close: {}".format(mkt_close))
+
+ea = ExchangeAgent(id = 0,
+                   name = 'Exchange_Agent',
+                   type = 'ExchangeAgent',
+                   mkt_open = mkt_open,
+                   mkt_close = mkt_close,
+                   symbols = symbols,
+                   log_orders=log_orders,
+                   book_freq = '1s',
+                   pipeline_delay = 0,
+                   computation_delay = 0,
+                   stream_history = 10,
+                   random_state = random_state)
+
+agents.extend([ea])
+
+# 4) MarketReplayAgent Config
+num_mr_agents  = 1
+cash_mr_agents = 10000000
+
+mr_agents = [MarketReplayAgent(id      = 1,
+                              name    = "Market_Replay_Agent",
+                              symbol  = symbols[0],
+                              date    = date,
+                              startingCash = cash_mr_agents,
+                              random_state = random_state)]
+agents.extend(mr_agents)
+
+# 5) ExperimentalAgent Config
+num_exp_agents  = 1
+cash_exp_agents = 10000000
+
+exp_agents = [ExperimentalAgent(id      = 2,
+                               name    = "Experimental_Agent",
+                               symbol  = symbols[0],
+                               startingCash = cash_exp_agents,
+                               execution_timestamp = pd.Timestamp("2012-06-21 09:30:02"),
+                               quantity = 1000,
+                               is_buy_order = True,
+                               limit_price = 500,
+                               random_state = random_state)]
+agents.extend(exp_agents)
+#######################################################################################################################
+
+# 6) Kernel Parameters
+kernel = Kernel("Market Replay Kernel", random_state = random_state)
+
+kernelStartTime = date_pd + pd.to_timedelta('09:30:00')
+kernelStopTime = date_pd + pd.to_timedelta('09:30:05')
+defaultComputationDelay = 0
+latency = np.zeros((3, 3))
+noise = [ 0.0 ]
+
+# 7) Data Oracle
+oracle = OrderBookOracle(symbol='AAPL',
+                         date='2012-06-21',
+                         orderbook_file_path='C:/_code/py/air/abides_open_source/abides/data/LOBSTER/AAPL_2012-06-21_34200000_57600000_orderbook_10.csv',
+                         message_file_path='C:/_code/py/air/abides_open_source/abides/data/LOBSTER/AAPL_2012-06-21_34200000_57600000_message_10.csv',
+                         num_price_levels=10)
+
+kernel.runner(agents = agents, startTime = kernelStartTime,
+              stopTime = kernelStopTime, agentLatency = latency,
+              latencyNoise = noise,
+              defaultComputationDelay = defaultComputationDelay,
+              defaultLatency=0,
+              oracle = oracle, log_dir = args.log_dir)
+
+simulation_end_time = dt.datetime.now()
+print ("Simulation End Time: {}".format(simulation_end_time))
+print ("Time taken to run simulation: {}".format(simulation_end_time - simulation_start_time))