tests passing with new lines.py. Still some zmq references left.

pavement.py

@@ -110,7 +110,6 @@ options(
 # Because I'm lazy
 stuff_i_want_in_my_debug_shell = [
     ('qutil', 'zipline.util', []),
-    ('zmq', 'zmq', []),
 ]
 
 @task

tests/test_exception_handling.py

@@ -1,190 +0,0 @@
-import zmq
-
-from unittest2 import TestCase
-from collections import defaultdict
-
-from zipline.test_algorithms import ExceptionAlgorithm, DivByZeroAlgorithm, \
-    InitializeTimeoutAlgorithm, TooMuchProcessingAlgorithm
-from zipline.finance.trading import SIMULATION_STYLE
-from zipline.core.devsimulator import AddressAllocator
-from zipline.lines import SimulatedTrading
-from zipline.gens.transform import StatefulTransform
-from zipline.gens.tradesimulation import MAX_HEARTBEAT_INTERVALS
-
-from zipline.utils.test_utils import (
-    drain_zipline,
-    setup_logger,
-    teardown_logger,
-    ExceptionSource,
-    ExceptionTransform
-)
-
-DEFAULT_TIMEOUT = 15 # seconds
-EXTENDED_TIMEOUT = 90
-
-allocator = AddressAllocator(1000)
-
-class ExceptionTestCase(TestCase):
-
-    leased_sockets = defaultdict(list)
-
-    def setUp(self):
-        self.zipline_test_config = {
-            'sid'                   : 133,
-            'results_socket_uri'    : allocator.lease(1)[0],
-            'simulation_style'      : SIMULATION_STYLE.FIXED_SLIPPAGE
-        }
-        self.ctx = zmq.Context()
-        setup_logger(self)
-
-    def tearDown(self):
-        self.ctx.term()
-        teardown_logger(self)
-
-    def test_datasource_exception(self):
-        self.zipline_test_config['trade_source'] = ExceptionSource()
-        zipline = SimulatedTrading.create_test_zipline(
-            **self.zipline_test_config
-        )
-        output, _ = drain_zipline(self, zipline)
-        assert len(output) == 1
-        assert output[0]['prefix'] == 'EXCEPTION'
-        message = output[0]['payload']
-        for field in ['date', 'message', 'name', 'stack']:
-            assert field in message.keys()
-
-        assert message['message'] == 'integer division or modulo by zero'
-        assert message['name'] == 'ZeroDivisionError'
-
-    def test_tranform_exception(self):
-        exc_tnfm = StatefulTransform(ExceptionTransform)
-        self.zipline_test_config['transforms'] = [exc_tnfm]
-
-        zipline = SimulatedTrading.create_test_zipline(
-            **self.zipline_test_config
-        )
-        output, _ = drain_zipline(self, zipline)
-        assert len(output) == 1
-        assert output[0]['prefix'] == 'EXCEPTION'
-        message = output[0]['payload']
-        for field in ['date', 'message', 'name', 'stack']:
-            assert field in message.keys()
-
-        assert message['message'] == 'An assertion message'
-        assert message['name'] == 'AssertionError'
-
-
-    def test_exception_in_init(self):
-        # Simulation
-        # ----------
-        self.zipline_test_config['algorithm'] = \
-                ExceptionAlgorithm(
-                    'initialize',
-                    self.zipline_test_config['sid']
-                )
-
-        zipline = SimulatedTrading.create_test_zipline(
-            **self.zipline_test_config
-        )
-        output, _ = drain_zipline(self, zipline)
-
-        self.assertEqual(output[-1]['prefix'], 'EXCEPTION')
-        payload = output[-1]['payload']
-        self.assertTrue(payload['date'])
-        self.assertEqual(payload['message'],'Algo exception in initialize')
-        self.assertEqual(payload['name'],'Exception')
-        # make sure our path shortening is working
-        self.assertEqual(payload['stack'][0]['filename'], '/zipline/lines.py')
-        self.assertEqual(payload['stack'][-1]['filename'], '/zipline/test_algorithms.py')
-
-    def test_exception_in_handle_data(self):
-        # Simulation
-        # ----------
-        self.zipline_test_config['algorithm'] = \
-                ExceptionAlgorithm(
-                    'handle_data',
-                    self.zipline_test_config['sid']
-                )
-
-        zipline = SimulatedTrading.create_test_zipline(
-            **self.zipline_test_config
-        )
-
-        output, _ = drain_zipline(self, zipline)
-        self.assertEqual(output[-1]['prefix'], 'EXCEPTION')
-        payload = output[-1]['payload']
-        self.assertTrue(payload['date'])
-        del payload['date']
-        self.assertEqual(payload['message'],'Algo exception in handle_data')
-        self.assertEqual(payload['name'],'Exception')
-        # make sure our path shortening is working
-        self.assertEqual(payload['stack'][0]['filename'], '/zipline/lines.py')
-        self.assertEqual(payload['stack'][-1]['filename'], '/zipline/test_algorithms.py')
-
-    def test_zerodivision_exception_in_handle_data(self):
-
-        # Simulation
-        # ----------
-        self.zipline_test_config['algorithm'] = \
-                DivByZeroAlgorithm(
-                    self.zipline_test_config['sid']
-                )
-
-        zipline = SimulatedTrading.create_test_zipline(
-            **self.zipline_test_config
-        )
-
-        output, _ = drain_zipline(self, zipline)
-
-        self.assertEqual(output[-1]['prefix'], 'EXCEPTION')
-        payload = output[-1]['payload']
-        self.assertTrue(payload['date'])
-        del payload['date']
-        self.assertEqual(payload['message'],'integer division or modulo by zero')
-        self.assertEqual(payload['name'],'ZeroDivisionError')
-        # make sure our path shortening is working
-        self.assertEqual(payload['stack'][0]['filename'], '/zipline/lines.py')
-        self.assertEqual(payload['stack'][-1]['filename'], '/zipline/test_algorithms.py')
-
-    def test_initialize_timeout(self):
-        
-        self.zipline_test_config['algorithm'] = \
-            InitializeTimeoutAlgorithm(
-                self.zipline_test_config['sid']
-            )
-        
-        zipline = SimulatedTrading.create_test_zipline(
-            **self.zipline_test_config
-        )
-        output, _ = drain_zipline(self, zipline)
-        self.assertEqual(output[-1]['prefix'], 'EXCEPTION')
-        payload = output[-1]['payload']
-        self.assertEqual(payload['name'],'TimeoutException')
-        self.assertEqual(payload['message'], 'Call to initialize timed out')
-
-    def test_heartbeat(self):
-        
-        self.zipline_test_config['algorithm'] = \
-            TooMuchProcessingAlgorithm(
-                self.zipline_test_config['sid']
-                )
-        zipline = SimulatedTrading.create_test_zipline(
-            **self.zipline_test_config
-        )
-        output, _ = drain_zipline(self, zipline)
-        
-        # There should be a message for each hearbeat, plus a message
-        # for the final timeout.
-        assert len(output) == MAX_HEARTBEAT_INTERVALS + 1
-
-        # Assert that everything but the last message is a heartbeat log.
-        for message in output[0:-1]:
-            assert message['prefix'] == 'LOG'
-            assert message['payload']['func_name'] == 'log_heartbeats'
-
-        # Assert that the last message is a timeout exception.
-        self.assertEqual(output[-1]['prefix'], 'EXCEPTION')
-        payload = output[-1]['payload']
-        self.assertEqual(payload['name'],'TimeoutException')
-        self.assertEqual(payload['message'], 'Too much time spent in handle_data call')
-        

tests/test_finance.py

@@ -2,7 +2,6 @@
 Tests for the zipline.finance package
 """
 import pytz
-import zmq
 
 from unittest2 import TestCase
 from datetime import datetime, timedelta
@@ -36,9 +35,7 @@ class FinanceTestCase(TestCase):
     def setUp(self):
         self.zipline_test_config = {
             'sid'                   : 133,
-            'results_socket_uri'    : allocator.lease(1)[0]
         }
-        self.ctx = zmq.Context()
 
         setup_logger(self)
 

zipline/lines.py

@@ -59,16 +59,9 @@ before invoking simulate.
                           |      __init__.                  |
                           +---------------------------------+
 """
-import sys
-import zmq
-import os
-from signal import SIGHUP, SIGINT
-import multiprocessing
-from setproctitle import setproctitle
 
 from zipline.test_algorithms import TestAlgorithm
 from zipline.finance.trading import SIMULATION_STYLE
-from zipline.utils.log_utils import ZeroMQLogHandler
 from zipline.utils import factory
 
 from zipline.gens.composites import (
@@ -78,8 +71,6 @@ from zipline.gens.composites import (
 from zipline.gens.tradesimulation import TradeSimulationClient as tsc
 from logbook import Logger
 
-import zipline.protocol as zp
-
 log = Logger('Lines')
 
 
@@ -90,10 +81,27 @@ class SimulatedTrading(object):
             transforms,
             algorithm,
             environment,
-            style,
-            results_socket_uri,
-            context,
-            sim_id):
+            style):
+        """
+        @sources - an iterable of iterables
+        These iterables must yield ndicts that contain:
+        - type :: a ziplines.protocol.DATASOURCE_TYPE
+        - dt :: a milliseconds since epoch timestamp in UTC
+
+        @transforms - An iterable of instances of StatefulTransform.
+
+        @algorithm - An object that implements:
+        `def initialize(self)`
+        `def handle_data(self, data)`
+        `def get_sid_filter(self)`
+        `def set_logger(self, logger)`
+        `def set_order(self, order_callable)`
+
+        @environment - An instance of finance.trading.TradingEnvironment
+
+        @style - protocol.SIMULATION_STYLE
+        """
+
 
         self.date_sorted = date_sorted_sources(*sources)
         self.transforms = transforms
@@ -102,138 +110,12 @@ class SimulatedTrading(object):
                                                 *self.transforms)
         self.trading_client = tsc(algorithm, environment, style)
         self.gen = self.trading_client.simulate(self.with_tnfms)
-        self.results_uri = results_socket_uri
-        self.results_socket = None
-        self.context = context
-        self.sim_id = sim_id
 
-        # optional process if we fork simulate into an
-        # independent process.
-        self.proc = None
-        self.send_sighup = False
-        self.logger = Logger(sim_id)
-        self.print_logger = Logger('Print')
+    def __iter__(self):
+        return self.gen
 
-        # exit status flag
-        self.success = False
-
-    def simulate(self, blocking=True, send_sighup=False):
-
-        # for non-blocking,
-        if blocking:
-            self.run_gen()
-        else:
-            self.send_sighup = send_sighup
-            return self.fork_and_sim()
-
-    def fork_and_sim(self):
-        self.proc = multiprocessing.Process(target=self.run_gen)
-        self.proc.start()
-        return self.proc
-
-    def run_gen(self):
-        setproctitle(self.sim_id)
-        self.open()
-        if self.zmq_out:
-            with self.zmq_out.threadbound():
-                self.stream_results()
-            # if no log socket, just run the algo normally
-        else:
-            self.stream_results()
-
-    def stream_results(self):
-        assert self.results_socket, \
-            "Results socket must exist to stream results"
-        try:
-            for event in self.gen:
-                if 'daily_perf' in event:
-                    msg = zp.PERF_FRAME(event)
-                else:
-                    msg = zp.RISK_FRAME(event)
-                self.results_socket.send(msg)
-
-            self.signal_done()
-            self.success = True
-        except Exception as exc:
-            self.handle_exception(exc)
-        finally:
-            # not much to do besides log our exit.
-            self.close()
-
-    def signal_done(self):
-        # notify monitor we're done
-        done_frame = zp.DONE_FRAME('success')
-        self.results_socket.send(done_frame)
-
-    def close(self):
-        log.info("Closing Simulation: {id}".format(id=self.sim_id))
-        if self.results_socket:
-            self.results_socket.close()
-        if self.proc and self.send_sighup:
-            ppid = os.getppid()
-            if self.success:
-                log.warning("Sending SIGHUP")
-                os.kill(ppid, SIGHUP)
-            else:
-                log.warning("Sending SIGINT")
-                os.kill(ppid, SIGINT)
-
-    def handle_exception(self, exc):
-        self.signal_exception(exc)
-
-    def signal_exception(self, exc=None):
-        """
-        All exceptions inside any component should boil back to
-        this handler.
-
-        Will inform the system that the component has failed and how it
-        has failed.
-        """
-        exc_type, exc_value, exc_traceback = sys.exc_info()
-
-        try:
-            log.exception('{id} sending exception to result stream.'\
-                .format(id=self.sim_id))
-            msg = zp.EXCEPTION_FRAME(
-                    exc_traceback,
-                    exc_type.__name__,
-                    exc_value.message
-                )
-
-            self.results_socket.send(msg)
-        except:
-            log.exception("Exception while reporting simulation exception.")
-
-    def open(self):
-        if not self.context:
-            self.context = zmq.Context()
-        if self.results_uri:
-            sock = self.context.socket(zmq.PUSH)
-            sock.connect(self.results_uri)
-            self.results_socket = sock
-            self.setup_logging()
-
-    def setup_logging(self):
-        assert self.results_socket
-        # The filter behavior is: matches are logged, mismatches
-        # are bubbled. If bubble is True, matches are also
-        # bubbled. Since we do not want user logs in our system
-        # logs, we set bubble to False.
-        self.zmq_out = ZeroMQLogHandler(
-            socket=self.results_socket,
-            filter=lambda r, h: r.channel in ['Print', 'AlgoLog'],
-            bubble=False
-        )
-
-    def join(self):
-        if self.proc:
-            self.proc.join()
-
-    def get_pids(self):
-        if self.proc:
-            return [self.proc.pid]
-        else:
-            return []
+    def __next__(self):
+        return self.gen.next()
 
     @staticmethod
     def create_test_zipline(**config):
@@ -297,10 +179,6 @@ class SimulatedTrading(object):
         if not simulation_style:
             simulation_style = SIMULATION_STYLE.FIXED_SLIPPAGE
 
-        zmq_context = config.get('zmq_context', None)
-        simulation_id = config.get('simulation_id', 'test_simulation')
-        results_socket_uri = config.get('results_socket_uri', None)
-
         #-------------------
         # Trade Source
         #-------------------
@@ -341,52 +219,7 @@ class SimulatedTrading(object):
                 test_algo,
                 trading_environment,
                 simulation_style,
-                results_socket_uri,
-                zmq_context,
-                simulation_id)
+                )
         #-------------------
 
         return sim
-
-
-class SimulatedTradingLite(object):
-    """
-    SimulatedTrading without multiprocess and without zmq.
-    Useful for profiling the core logic and for rapid testing
-    of new features.
-    """
-    def __init__(self,
-            sources,
-            transforms,
-            algorithm,
-            environment,
-            style):
-        """
-        @sources - an iterable of iterables
-        These iterables must yield ndicts that contain:
-        - type :: a ziplines.protocol.DATASOURCE_TYPE
-        - dt :: a milliseconds since epoch timestamp in UTC
-
-        @transforms - An iterable of instances of StatefulTransform.
-
-        @algorithm - An object that implements:
-        `def initialize(self)`
-        `def handle_data(self, data)`
-        `def get_sid_filter(self)`
-        `def set_logger(self, logger)`
-        `def set_order(self, order_callable)`
-
-        @environment - An instance of finance.trading.TradingEnvironment
-
-        @style - protocol.SIMULATION_STYLE
-        """
-        self.date_sorted = date_sorted_sources(*sources)
-        self.transforms = transforms
-        # Formerly merged_transforms.
-        self.with_tnfms = sequential_transforms(self.date_sorted,
-                                                *self.transforms)
-        self.trading_client = tsc(algorithm, environment, style)
-        self.gen = self.trading_client.simulate(self.with_tnfms)
-
-    def get_results(self):
-        return self.gen

zipline/utils/test_utils.py

@@ -1,7 +1,4 @@
 import multiprocessing
-import zmq
-import time
-import zipline.protocol as zp
 from datetime import datetime
 import blist
 from zipline.utils.date_utils import EPOCH
@@ -65,69 +62,18 @@ def check(test, a, b, label=None):
         test.assertEqual(a, b, "mismatch on path: " + label)
 
 
-def drain_zipline(test, zipline, p_blocking=False):
-    assert test.ctx, "method expects a valid zmq context"
-    assert test.zipline_test_config, "method expects a valid test config"
-    assert isinstance(test.zipline_test_config, dict)
-    assert test.zipline_test_config['results_socket_uri'], \
-            "need to specify a socket address for logs/perf/risk"
-    test.receiver = create_receiver(
-        test.zipline_test_config['results_socket_uri'],
-        test.ctx
-    )
-    # Bind and connect are asynch, so allow time for bind before
-    # starting the zipline (TSC connects internally).
-    time.sleep(1)
-
-    # start the simulation
-    zipline.simulate(blocking=p_blocking)
-    output, transaction_count = drain_receiver(test.receiver)
-    # some processes will exit after the message stream is
-    # finished. We block here to avoid collisions with subsequent
-    # ziplines.
-    zipline.join()
-
-    return output, transaction_count
-
-
-def create_receiver(socket_addr, ctx):
-    receiver = ctx.socket(zmq.PULL)
-    receiver.bind(socket_addr)
-
-    return receiver
-
-
-def drain_receiver(receiver, count=None):
+def assert_single_position(test, zipline):
     output = []
     transaction_count = 0
     msg_counter = 0
-    while True:
-        msg = receiver.recv()
+    # start the simulation
+    for update in zipline:
         msg_counter += 1
-        update = zp.BT_UPDATE_UNFRAME(msg)
         output.append(update)
-        if update['prefix'] == 'PERF':
+        if update.has_key('daily_perf'):
             transaction_count += \
-                len(update['payload']['daily_perf']['transactions'])
-        elif update['prefix'] == 'EXCEPTION':
-            break
-        elif update['prefix'] == 'DONE':
-            break
+                len(update['daily_perf']['transactions'])
 
-        if count and msg_counter >= count:
-            break
-
-    receiver.close()
-    del receiver
-
-    return output, transaction_count
-
-
-def assert_single_position(test, zipline, blocking=False):
-    output, transaction_count = drain_zipline(test,
-                                              zipline,
-                                              p_blocking=blocking)
-    test.assertEqual(output[-1]['prefix'], 'DONE')
 
     test.assertEqual(
         test.zipline_test_config['order_count'],
@@ -137,8 +83,7 @@ def assert_single_position(test, zipline, blocking=False):
     # the final message is the risk report, the second to
     # last is the final day's results. Positions is a list of
     # dicts.
-    perfs = [x for x in output if x['prefix'] == 'PERF']
-    closing_positions = perfs[-2]['payload']['daily_perf']['positions']
+    closing_positions = output[-2]['daily_perf']['positions']
 
     test.assertEqual(
         len(closing_positions),