diff --git a/zipline/messaging.py b/zipline/messaging.py index 671eeea7..23d96b9c 100644 --- a/zipline/messaging.py +++ b/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']) diff --git a/zipline/simulator.py b/zipline/simulator.py deleted file mode 100644 index c64c9a5f..00000000 --- a/zipline/simulator.py +++ /dev/null @@ -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 - diff --git a/zipline/sources.py b/zipline/sources.py index 385ffe0d..4b1a2a18 100644 --- a/zipline/sources.py +++ b/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 - - - - diff --git a/zipline/test/test_messaging.py b/zipline/test/test_messaging.py index 1a42faa1..a8344283 100644 --- a/zipline/test/test_messaging.py +++ b/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) -