Merge branch 'byebye_threadsim' of github.com:quantopian/zipline into byebye_threadsim

tests/test_exception_handling.py

@@ -9,7 +9,7 @@ from zipline.finance.trading import SIMULATION_STYLE
 from zipline.core.devsimulator import AddressAllocator
 from zipline.lines import SimulatedTrading
 
-from zipline.utils.test_utils import drain_zipline
+from zipline.utils.test_utils import drain_zipline, check
 
 DEFAULT_TIMEOUT = 15 # seconds
 EXTENDED_TIMEOUT = 90
@@ -50,7 +50,12 @@ class ExceptionTestCase(TestCase):
         )
 
         output, _ = drain_zipline(self, zipline)
-        self.assertEqual(output, ['EXCEPTION'])
+        self.assertEqual(len(output), 1)
+        self.assertEqual(output[-1]['prefix'], 'EXCEPTION')
+        payload = output[-1]['payload']
+        check(self, payload, INITIALIZE_STACK_TB)
+
+        import nose.tools; nose.tools.set_trace()
         self.assertTrue(zipline.sim.ready())
         self.assertFalse(zipline.sim.exception)
 
@@ -61,3 +66,31 @@ class ExceptionTestCase(TestCase):
         #   - define more zipline failure modes: exception in other
         #   components, exception in Monitor, etc. write tests
         #   for those scenarios.
+
+
+
+INITIALIZE_STACK_TB =\
+[{'file': '/Users/fawce/projects/qexec/zipline_repo/zipline/core/component.py',
+  'line': 'self._run()',
+  'lineno': 229,
+  'method': 'run'},
+ {'file': '/Users/fawce/projects/qexec/zipline_repo/zipline/core/component.py',
+  'line': 'self.open()',
+  'lineno': 208,
+  'method': '_run'},
+ {'file': '/Users/fawce/projects/qexec/zipline_repo/zipline/components/tradesimulation.py',
+  'line': 'self.initialize_algo()',
+  'lineno': 73,
+  'method': 'open'},
+ {'file': '/Users/fawce/projects/qexec/zipline_repo/zipline/components/tradesimulation.py',
+  'line': 'self.do_op(self.algorithm.initialize)',
+  'lineno': 83,
+  'method': 'initialize_algo'},
+ {'file': '/Users/fawce/projects/qexec/zipline_repo/zipline/components/tradesimulation.py',
+  'line': 'callable_op(*args, **kwargs)',
+  'lineno': 205,
+  'method': 'do_op'},
+ {'file': '/Users/fawce/projects/qexec/zipline_repo/zipline/test_algorithms.py',
+  'line': 'raise Exception("Algo exception in initialize")',
+  'lineno': 161,
+  'method': 'initialize'}]

zipline/components/tradesimulation.py

@@ -187,7 +187,8 @@ class TradeSimulationClient(Component):
 
     def exception_callback(self, exc_type, exc_value, exc_traceback):
         if self.results_socket:
-            self.out_socket.send("EXCEPTION")
+            msg = zp.EXCEPTION_FRAME(exc_traceback)
+            self.out_socket.send(msg)
 
     def do_op(self, callable_op, *args, **kwargs):
         """ Wrap a callable operation with the zmq logbook
@@ -225,8 +226,8 @@ class TradeSimulationClient(Component):
         with log_pipeline.threadbound(), self.stdout_capture(self.logger, ''):
             self.algorithm.handle_data('data')
 
-    def connect_order(self):
-        return self.connect_push_socket(self.addresses['order_address'])
+    #def connect_order(self):
+    #    return self.connect_push_socket(self.addresses['order_address'])
 
     def order(self, sid, amount):
         order = zp.ndict({

zipline/protocol.py

@@ -118,6 +118,7 @@ import msgpack
 import numbers
 import datetime
 import pytz
+import traceback
 
 from collections import namedtuple
 
@@ -503,6 +504,20 @@ def convert_transactions(transactions):
 def RISK_FRAME(risk):
     return BT_UPDATE_FRAME('RISK', risk)
 
+def EXCEPTION_FRAME(exception_tb):
+    stack_list = traceback.extract_tb(exception_tb)
+    rlist = []
+    for stack in stack_list:
+        rstack = {
+            'file'      : stack[0],
+            'lineno'    : stack[1],
+            'method'    : stack[2],
+            'line'      : stack[3]
+        }
+        rlist.append(rstack)
+
+    return BT_UPDATE_FRAME('EXCEPTION', rlist)
+
 def BT_UPDATE_FRAME(prefix, payload):
     """
     Frames prepared by RISK_FRAME and PERF_FRAME methods are sent via the same

zipline/utils/test_utils.py

@@ -1,5 +1,55 @@
 import zmq
 import zipline.protocol as zp
+from datetime import datetime
+import blist
+from zipline.utils.date_utils import EPOCH
+from itertools import izip
+
+
+def check_list(test, a, b, label):
+    test.assertTrue(isinstance(a, (list, blist.blist)))
+    test.assertTrue(isinstance(b, (list, blist.blist)))
+    i = 0
+    for a_val, b_val in izip(a, b):
+        check(test, a_val, b_val, label + "[" + str(i) + "]")
+
+
+def check_dict(test, a, b, label):
+    test.assertTrue(isinstance(a, dict))
+    test.assertTrue(isinstance(b, dict))
+    for key in a.keys():
+        # ignore the extra fields used by dictshield
+        if key in ['progress']:
+            continue
+        test.assertTrue(a.has_key(key), "missing key at: " + label + "." + key)
+        test.assertTrue(b.has_key(key), "missing key at: " + label + "." + key)
+        a_val = a[key]
+        b_val = b[key]
+        check(test, a_val, b_val, label + "." + key)
+
+
+def check_datetime(test, a, b, label):
+    test.assertTrue(isinstance(a, datetime))
+    test.assertTrue(isinstance(b, datetime))
+    test.assertEqual(EPOCH(a), EPOCH(b), "mismatched dates " + label)
+
+
+def check(test, a, b, label=None):
+    """
+    Check equality for arbitrarily nested dicts and lists that terminate
+    in types that allow direct comparisons (string, ints, floats, datetimes)
+    """
+    if not label:
+        label = '<root>'
+    if isinstance(a, dict):
+        check_dict(test, a, b, label)
+    elif isinstance(a, (list, blist.blist)):
+        check_list(test, a, b, label)
+    elif isinstance(a, datetime):
+        check_datetime(test, a, b, label)
+    else:
+        test.assertEqual(a, b, "mismatch on path: " + label)
+
 
 def drain_zipline(test, zipline):
     assert test.ctx, "method expects a valid zmq context"
@@ -19,15 +69,14 @@ def drain_zipline(test, zipline):
         msg = test.receiver.recv()
         if msg == str(zp.CONTROL_PROTOCOL.DONE):
             break
-        elif msg == "EXCEPTION":
-            output.append(msg)
-            break
         else:
             update = zp.BT_UPDATE_UNFRAME(msg)
             output.append(update)
             if update['prefix'] == 'PERF':
                 transaction_count += \
                     len(update['payload']['daily_perf']['transactions'])
+            elif update['prefix'] == 'EXCEPTION':
+                break
 
     del test.receiver