From 28c380c245b5aa6048b7eb11b47b562885e8d7df Mon Sep 17 00:00:00 2001 From: fawce Date: Fri, 20 Apr 2012 00:58:24 -0400 Subject: [PATCH] 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 | 12 ++++--- zipline/finance/trading.py | 64 ++++++++++++++++++++------------------ zipline/messaging.py | 21 +++++-------- 3 files changed, 47 insertions(+), 50 deletions(-) diff --git a/zipline/component.py b/zipline/component.py index e4af443c..4e1d6b7f 100644 --- a/zipline/component.py +++ b/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 diff --git a/zipline/finance/trading.py b/zipline/finance/trading.py index 9a950e61..228572ec 100644 --- a/zipline/finance/trading.py +++ b/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} diff --git a/zipline/messaging.py b/zipline/messaging.py index 03764807..6cd4154d 100644 --- a/zipline/messaging.py +++ b/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()