diff --git a/etc/jenkins.sh b/etc/jenkins.sh index 8a649664..b05bb26d 100755 --- a/etc/jenkins.sh +++ b/etc/jenkins.sh @@ -1,5 +1,5 @@ #!/bin/bash - +set -e #setup virtualenvironment export VIRTUALENVWRAPPER_PYTHON=/usr/bin/python2.7 export WORKON_HOME=/mnt/jenkins_backups/virtual_envs diff --git a/zipline/finance/trading.py b/zipline/finance/trading.py index cc5ca523..2402e553 100644 --- a/zipline/finance/trading.py +++ b/zipline/finance/trading.py @@ -61,14 +61,17 @@ class TradeSimulationClient(qmsg.Component): class TradeSimulator(qmsg.BaseTransform): - def __init__(self): + def __init__(self, expected_orders): qmsg.BaseTransform.__init__(self, "") self.open_orders = {} self.algo_time = None self.event_start = None self.last_event_time = None self.last_iteration_duration = None - + self.expected_orders = expected_orders + self.order_count = 0 + self.trade_count = 0 + @property def get_id(self): return "ALGO_TIME" @@ -91,13 +94,15 @@ class TradeSimulator(qmsg.BaseTransform): [], [self.feed_socket], timeout=self.heartbeat_timeout/1000) #select timeout is in sec - # + self.trade_count += 1 #no more orders, should be an error condition if len(rlist) == 0 or len(xlist) > 0: raise Exception("unexpected end of feed stream") message = rlist[0].recv() if message == str(zp.CONTROL_PROTOCOL.DONE): self.signal_done() + if(self.expected_orders > 0): + assert self.expected_orders == self.order_count return #leave open orders hanging? client requests for orders? event = zp.FEED_UNFRAME(message) @@ -143,7 +148,7 @@ class TradeSimulator(qmsg.BaseTransform): def add_open_order(self, sid, amount): - pass + self.order_count = self.order_count + 1 def process_orders(self, event): #TODO put real fill logic here, return a list of fills diff --git a/zipline/protocol.py b/zipline/protocol.py index 864f4274..9367b703 100644 --- a/zipline/protocol.py +++ b/zipline/protocol.py @@ -210,9 +210,7 @@ def DATASOURCE_UNFRAME(msg): - dt - a datetime object """ try: - qutil.LOGGER.info("unpacking {msg}".format(msg = msg)) ds_type, payload = msgpack.loads(msg) - qutil.LOGGER.info("unpacked a datasource frame! {ds_type} - {payload}".format(ds_type=ds_type, payload=payload) ) assert isinstance(ds_type, basestring) if(ds_type == "TRADE"): return TRADE_UNFRAME(payload) @@ -366,12 +364,10 @@ def TRADE_FRAME(event): assert isinstance(event.price, float) assert isinstance(event.volume, int) PACK_DATE(event) - qutil.LOGGER.info("event is: {event}".format(event=event.__dict__)) return msgpack.dumps(tuple([event.sid, event.price, event.volume, event.epoch, event.micros, event.type, event.source_id])) def TRADE_UNFRAME(msg): try: - qutil.LOGGER.info("about to unpack TRADE: {trade}".format(trade=msg)) sid, price, volume, epoch, micros, source_type, source_id = msgpack.loads(msg) assert isinstance(sid, int) @@ -381,7 +377,6 @@ def TRADE_UNFRAME(msg): assert isinstance(micros, numbers.Integral) rval = namedict({'sid' : sid, 'price' : price, 'volume' : volume, 'epoch' : epoch, 'micros' : micros, 'type' : source_type, 'source_id' : source_id}) UNPACK_DATE(rval) - qutil.LOGGER.info("unpacked Trade: {trade}".format(trade=rval.__dict__)) return rval except TypeError: raise INVALID_TRADE_FRAME(msg) diff --git a/zipline/test/client.py b/zipline/test/client.py index c8b7674a..33b9bf29 100644 --- a/zipline/test/client.py +++ b/zipline/test/client.py @@ -56,7 +56,7 @@ class TestTradingClient(TradeSimulationClient): def handle_events(self, event_queue): #place an order for 100 shares of sid:133 - if(self.incr >= self.count): + if(self.incr < self.count): self.order(133, 100) self.incr += 1 diff --git a/zipline/test/test_finance.py b/zipline/test/test_finance.py index c255e0a3..06beafb1 100644 --- a/zipline/test/test_finance.py +++ b/zipline/test/test_finance.py @@ -91,12 +91,12 @@ class FinanceTestCase(ThreadPoolExecutorMixin, TestCase): # --------------------- set1 = SpecificEquityTrades("flat-133",factory.create_trade_history(133, - [10.0,10.0,10.0,10.0], - [100,100,100,100], - datetime.datetime.strptime("02/15/2012","%m/%d/%Y"), + [10.0,10.0,10.0,10.0,10.0,10.0,10.0,10.0,10.0,10.0,10.0,10.0,10.0,10.0,10.0,10.0], + [100,100,100,100,100,100,100,100,100,100,100,100,100,100,100,100], + datetime.datetime.strptime("02/1/2012","%m/%d/%Y"), datetime.timedelta(days=1))) - client = TestTradingClient(1) - order_sim = TradeSimulator() + client = TestTradingClient(10) + order_sim = TradeSimulator(expected_orders=10) sim.register_components([client, order_sim, set1]) sim.register_controller( con ) diff --git a/zipline/test/test_monitor.py b/zipline/test/test_monitor.py index 5835b779..0fea158d 100644 --- a/zipline/test/test_monitor.py +++ b/zipline/test/test_monitor.py @@ -13,6 +13,7 @@ from gevent_zeromq import zmq ctx = zmq.Context() +#TODO: disabled by prefixing the test methods with a d class TestControlProtocol(TestCase): def setUpController(self): @@ -40,7 +41,7 @@ class TestControlProtocol(TestCase): msg.join() self.assertEqual(msg.value, message) - def test_control_message(self): + def dtest_control_message(self): sub = self.controller.message_listener(context=ctx) message = gevent.spawn(self.asyncMessage, sub) @@ -55,7 +56,7 @@ class TestControlProtocol(TestCase): sub.close() push.close() - def test_control_delivery(self): + def dtest_control_delivery(self): # Assert that the number of messages sent on the wire is # the number of messages received, ie we don't drop any. # This is of course depenendent on the topology of the