Migrated the simulator classes to qexec.

zipline/messaging.py

@@ -65,7 +65,7 @@ class Component(object):
                 sock.close()
         except Exception as e:
             qutil.LOGGER.exception("Unexpected error in run for {id}.".format(id=self.get_id()))
-            raise e
+            raise
         finally:
             if(self.context != None):
                 self.context.destroy()
@@ -100,8 +100,7 @@ class Component(object):
             message = self.sync_socket.recv()
         else:
             raise Exception("Sync ack timed out on response for {id}".format(id=self.get_id()))
-            
-            
+
     def bind_data(self):
         return self.bind_pull_socket(self.addresses['data_address'])
         

zipline/simulator.py

@@ -1,49 +0,0 @@
-"""
-Provides simulated data feed services...
-"""
-import multiprocessing
-import json
-import copy
-import threading
-
-import zipline.util as qutil
-import zipline.messaging as qmsg
-
-class SimulatorBase(qmsg.ComponentHost):
-    """
-    Simulator coordinates the launch and communication of source, feed, transform, and merge components.
-    """
-    
-    def __init__(self, addresses, gevent_needed=False):
-        """
-        """    
-        qmsg.ComponentHost.__init__(self, addresses, gevent_needed)
-                                   
-    def simulate(self):
-        self.run()
-        
-    def get_id(self):
-        return "Simulator"
-        
-class ThreadSimulator(SimulatorBase):
-    
-    def __init__(self, addresses):
-        SimulatorBase.__init__(self, addresses)
-        
-    def launch_component(self, component):
-        qutil.LOGGER.info("starting {name}".format(name=component.get_id()))
-        thread = threading.Thread(target=component.run)
-        thread.start()
-        return thread
-    
-class ProcessSimulator(SimulatorBase):        
-    
-    def __init__(self, addresses):
-        SimulatorBase.__init__(self, addresses)
-        
-    def launch_component(self, component):
-        qutil.LOGGER.info("starting {name}".format(name=component.get_id()))
-        proc = multiprocessing.Process(target=component.run)
-        proc.start()
-        return proc
-

zipline/sources.py

@@ -2,15 +2,14 @@
 Provides data handlers that can push messages to a zipline.core.DataFeed
 """
 import datetime
-import json
 import random
 
 import zipline.util as qutil
-import zipline.messaging as qmsg        
-       
+import zipline.messaging as qmsg
+
 class RandomEquityTrades(qmsg.DataSource):
     """Generates a random stream of trades for testing."""
-    
+
     def __init__(self, sid, source_id, count):
         qmsg.DataSource.__init__(self, source_id)
         self.count          = count
@@ -19,22 +18,21 @@ class RandomEquityTrades(qmsg.DataSource):
         self.trade_start    = datetime.datetime.now()
         self.minute         = datetime.timedelta(minutes=1)
         self.price          = random.uniform(5.0, 50.0)
-    
+
     def get_type(self):
-        return 'equity_trade'    
-    
+        return 'equity_trade'
+
     def do_work(self):
         if(self.incr == self.count):
             self.signal_done()
             return
         self.price = self.price + random.uniform(-0.05, 0.05)
-        event = {'sid':self.sid, 
-                 'dt':qutil.format_date(self.trade_start + (self.minute * self.incr)),
-                 'price':self.price, 
-                 'volume':random.randrange(100,10000,100)}
+        event = {
+            'sid'    : self.sid,
+            'dt'     : qutil.format_date(self.trade_start + (self.minute * self.incr)),
+            'price'  : self.price,
+            'volume' : random.randrange(100,10000,100)
+        }
+
         self.send(event)
         self.incr += 1
-          
-       
-        
-        

zipline/test/test_messaging.py

@@ -1,12 +1,17 @@
 """
 Test suite for the messaging infrastructure of QSim.
 """
-#don't worry about excessive public methods pylint: disable=R0904  
+#don't worry about excessive public methods pylint: disable=R0904
+
+# TODO: make sure this can run in parallel... right now this is
+# forbiddeen because we hardcode the ports but it should totally
+# be possible and then we can the suite much faster.
+#
+# nosetests --processes=5
 
 import zipline.util as qutil
 import zipline.messaging as qmsg
 
-from zipline.simulator import ThreadSimulator, ProcessSimulator
 from zipline.transforms.technical import MovingAverage
 from zipline.sources import RandomEquityTrades
 
@@ -14,7 +19,6 @@ from zipline.test.client import TestClient
 
 # Should not inherit form TestCase since test runners will pick
 # it up as a test.
-
 class SimulatorTestCase(object):
 
     def setUp(self):
@@ -24,6 +28,7 @@ class SimulatorTestCase(object):
         Generate some config objects for the datafeed, sources, and transforms.
         """
 
+        # TODO: new logic so we don't have to hardcode these
         self.addresses = {
             'sync_address'   : "tcp://127.0.0.1:10100",
             'data_address'   : "tcp://127.0.0.1:10101",
@@ -32,11 +37,14 @@ class SimulatorTestCase(object):
             'result_address' : "tcp://127.0.0.1:10104"
         }
 
+        # TODO: remove?
         self.addressesblarg = "test"
 
     def get_simulator(self):
+        """
+        Return the simulator instance to be tested.
+        """
         raise NotImplementedError
-        #return ThreadSimulator(self.addresses)
 
     def test_sources_only(self):
 
@@ -48,8 +56,8 @@ class SimulatorTestCase(object):
         sim.simulate()
 
         self.assertEqual(sim.feed.pending_messages(), 0,
-                        "The feed should be drained of all messages, found {n} remaining."
-                            .format(n=sim.feed.pending_messages()))
+            "The feed should be drained of all messages, found {n} remaining."
+                .format(n=sim.feed.pending_messages()))
 
 
     def test_transforms(self):
@@ -62,9 +70,11 @@ class SimulatorTestCase(object):
         sim.register_components([ret1, ret2, mavg1, mavg2, client])
         sim.simulate()
 
-        self.assertEqual(sim.feed.pending_messages(), 0, "The feed should be drained of all messages.")
+        self.assertEqual(sim.feed.pending_messages(), 0, \
+                "The feed should be drained of all messages.")
 
     def dtest_error_in_feed(self):
+
         ret1 = RandomEquityTrades(133, "ret1", 400)
         ret2 = RandomEquityTrades(134, "ret2", 400)
         sources = {"ret1":ret1, "ret2":ret2}
@@ -73,12 +83,9 @@ class SimulatorTestCase(object):
         transforms = {"mavg1":mavg1, "mavg2":mavg2}
         client = TestClient(self, expected_msg_count=0)
         sim = self.get_simulator(sources, transforms, client)
+
+        # TODO: way too long
         sim.feed = DataFeedErr(sources.keys(), sim.data_address, sim.feed_address, sim.performance_address, qmsg.Sync(sim, "DataFeedErrorGenerator"))
         sim.simulate()
 
 
-#class ProcessSimulatorTestCase(SimulatorTestCase):
-
-    #def get_simulator(self):
-        #return ProcessSimulator(self.addresses)
-