removed heartbeat on every iteration, resulted in a 50% reduction in processing time per day.

However, the big lesson from this experiment is that the major bottleneck is in the feedback loop. By eliminating the feedback loop and instead putting the transaction simulation into the client end of the zipline, I think we could accelerate by 10x.

zipline/component.py

@@ -86,6 +86,7 @@ class Component(object):
         self.start_tic         = None
         self.stop_tic          = None
         self.note              = None
+        self.confirmed         = False
 
         # Humanhashes make this way easier to debug because they
         # stick in your mind unlike a 32 byte string of random hex.
@@ -235,12 +236,13 @@ class Component(object):
         """
         Send a synchronization request to the host.
         """
+        if not self.confirmed:
+            # TODO: proper framing
+            self.sync_socket.send(self.get_id + ":RUN")
 
-        # TODO: proper framing
-        self.sync_socket.send(self.get_id + ":RUN")
-
-        self.receive_sync_ack() # blocking
-
+            self.receive_sync_ack() # blocking
+            self.confirmed = True
+            
     def runtime(self):
         if self.ready() and self.start_tic and self.stop_tic:
             return self.stop_tic - self.start_tic

zipline/finance/trading.py

@@ -38,7 +38,7 @@ class TradeSimulationClient(qmsg.Component):
         self.current_dt             = trading_environment.period_start
         self.last_iteration_dur     = datetime.timedelta(seconds=0)
         self.algorithm              = None
-        self.max_wait               = datetime.timedelta(seconds=7)
+        self.max_wait               = datetime.timedelta(seconds=60)
         self.last_msg_dt            = datetime.datetime.utcnow()
         
         assert self.trading_environment.frame_index != None
@@ -207,7 +207,6 @@ class OrderDataSource(qmsg.DataSource):
         qmsg.DataSource.__init__(self, zp.FINANCE_COMPONENT.ORDER_SOURCE)
         self.sent_count         = 0
         self.recv_count         = Counter()
-        self.works              = 0
 
     @property
     def get_type(self):
@@ -222,14 +221,14 @@ class OrderDataSource(qmsg.DataSource):
 
     def do_work(self):    
         
-        self.works += 1
+        self.recv_count['work_loops'] += 1
 
         #pull all orders from client.
         count = 0
         
         # one iteration of the client could include several orders
         # so iterate until the client signals a break or a close.
-        while True:
+        # while True:
             # poll all the sockets
             # we reduce the timeout here by a factor of 2, because we need
             # to potentially receive the client's done message before the 
@@ -237,34 +236,37 @@ class OrderDataSource(qmsg.DataSource):
             
             # this will block for timeout/2, and return an empty dict if there
             # are no messages.
-            socks = dict(self.poll.poll(self.heartbeat_timeout/2))
-
-            # see if the poller has results for the result_feed
-            if self.order_socket in socks and \
-                socks[self.order_socket] == self.zmq.POLLIN:
-                
-                order_msg = self.order_socket.recv()
             
-                if order_msg == str(zp.ORDER_PROTOCOL.DONE):
-                    qutil.LOGGER.info("order source is done")
-                    self.signal_done()
-                    self.recv_count['done'] += 1
-                    return
-                
-                if order_msg == str(zp.ORDER_PROTOCOL.BREAK):
-                    # send a blank message to avoid an empty buffer
-                    # in the feed
-                    self.recv_count['break'] += 1
-                    if count == 0:
-                        self.send(namedict({}))
-                    break
+        socks = dict(self.poll.poll())
 
-                order = zp.ORDER_UNFRAME(order_msg)
-                self.recv_count['order'] += 1
-                #send the order along
-                self.send(order)
-                count += 1
-                self.sent_count += 1
+        # see if the poller has results for the result_feed
+        if self.order_socket in socks and \
+            socks[self.order_socket] == self.zmq.POLLIN:
+            
+            order_msg = self.order_socket.recv()
+        
+            if order_msg == str(zp.ORDER_PROTOCOL.DONE):
+                qutil.LOGGER.info("order source is done")
+                self.signal_done()
+                self.recv_count['done'] += 1
+                return
+            
+            if order_msg == str(zp.ORDER_PROTOCOL.BREAK):
+                # send a blank message to avoid an empty buffer
+                # in the feed
+                self.recv_count['break'] += 1
+                if self.sent_count == 0:
+                    self.send(namedict({}))
+                self.sent_count = 0
+                return
+
+            order = zp.ORDER_UNFRAME(order_msg)
+            self.recv_count['order'] += 1
+            #send the order along
+            self.send(order)
+            count += 1
+            self.sent_count += 1
+            
                 
 
 class TransactionSimulator(qmsg.BaseTransform):
@@ -435,7 +437,7 @@ class TransactionSimulator(qmsg.BaseTransform):
                 dt.replace(tzinfo = pytz.utc), 
                 direction
             )
-        else:
+        elif len(orders) > 0:
             warning = """
 Calculated a zero volume transaction on trade: 
 {event} 

zipline/messaging.py

@@ -110,7 +110,7 @@ class ComponentHost(Component):
             self.launch_component(component)
         self.launch_controller()
 
-    def is_timed_out(self):
+    def is_running(self):
         """
         DEPRECATED, left in for compatability for now.
         """
@@ -119,23 +119,16 @@ class ComponentHost(Component):
 
         if len(self.components) == 0:
             qutil.LOGGER.info("Component register is empty.")
-            return True
+            return False
 
-        for source, last_dt in self.sync_register.iteritems():
-            if (cur_time - last_dt) > self.timeout:
-                qutil.LOGGER.info(
-                    "Time out for {source}. Current component registery: {reg}".
-                    format(source=source, reg=self.components)
-                )
-                return True
-
-        return False
+        return True
 
     def loop(self, lockstep=True):
 
-        while not self.is_timed_out():
-            # wait for synchronization request
-            socks = dict(self.sync_poller.poll(self.heartbeat_timeout)) #timeout after 2 seconds.
+        while self.is_running():
+            # wait for synchronization request at start, and DONE at end.
+            # don't timeout.
+            socks = dict(self.sync_poller.poll()) 
 
             if self.sync_socket in socks and socks[self.sync_socket] == self.zmq.POLLIN:
                 msg = self.sync_socket.recv()