Misc fixes, disabled Risk reporting until PR.

zipline/finance/performance.py

@@ -290,17 +290,16 @@ class PerformanceTracker():
         )
 
     def handle_simulation_end(self):
-        assert False
 
-        self.risk_report = risk.RiskReport(
-            self.returns,
-            self.trading_environment
-        )
+        #self.risk_report = risk.RiskReport(
+            #self.returns,
+            #self.trading_environment
+        #)
 
         # Output Results
-        if self.result_stream:
-            # TODO: proper framing
-            self.result_stream.send_pyobj(self.risk_report.to_dict())
+        #if self.result_stream:
+            ## TODO: proper framing
+            #self.result_stream.send_pyobj(self.risk_report.to_dict())
 
         self.result_stream.send_pyobj(None)
 

zipline/protocol_utils.py

@@ -1,6 +1,8 @@
 import copy
 from ctypes import Structure, c_ubyte
 
+from pandas import Series
+
 def Enum(*options):
     """
     Fast enums are very important when we want really tight zmq
@@ -85,3 +87,6 @@ class namedict(object):
 
     def has_attr(self, name):
         return self.__dict__.has_key(name)
+
+    def as_series(self):
+        return Series(self.__dict__, self.__dict__.keys())

zipline/test/client.py

@@ -1,3 +1,5 @@
+from gevent_zeromq import zmq
+
 import zipline.util as qutil
 import zipline.messaging as qmsg
 import zipline.protocol as zp
@@ -14,6 +16,12 @@ class TestClient(qmsg.Component):
         self.received_count     = 0
         self.prev_dt            = None
 
+        self.result_streams     = []
+
+        # Maximum outgoing result streams, really shouldn't ever
+        # need more than 1.
+        self.max_outgoing       = 5
+
     @property
     def get_id(self):
         return "TEST_CLIENT"
@@ -25,6 +33,17 @@ class TestClient(qmsg.Component):
     def open(self):
         self.data_feed = self.connect_result()
 
+    def result_stream(self, zmq_socket, context=None):
+        """
+        Asynchronously grab a socket to stream results out on.
+        """
+        ctx = context or zmq.Context.instance()
+        sock = ctx.socket(zmq.PULL)
+        sock.bind(zmq_socket)
+
+        # Add
+        self.result_streams.append( sock )
+
     def do_work(self):
         socks = dict(self.poll.poll(self.heartbeat_timeout))
 

zipline/test/test_messaging.py

@@ -13,9 +13,7 @@ from zipline.test.transform import DivideByZeroTransform
 
 from nose.tools import timed
 
-# Should not inherit form TestCase since test runners will pick
-# it up as a test. Its a Mixin of sorts at this point.
-class SimulatorTestCase(object):
+class BaseSimulator(object):
 
     # Leased sockets is a defaultdict keyed by the test case.
     # This lets you debug the sockets being allocated in the
@@ -25,7 +23,6 @@ class SimulatorTestCase(object):
     #    'test_orders'      : ['tcp : //127.0.0.1 : 1000', ... ],
     #    'test_performance' : ['tcp : //127.0.0.1 : 1025', ... ],
     # }
-
     leased_sockets = defaultdict(list)
 
     def setUp(self):
@@ -79,6 +76,10 @@ class SimulatorTestCase(object):
     def unallocate_sockets(self):
         self.allocator.reaquire(*self.leased_sockets[self.id()])
 
+# Should not inherit form TestCase since test runners will pick
+# it up as a test. Its a Mixin of sorts at this point.
+class SimulatorTestCase(BaseSimulator):
+
     # -------
     #  Cases
     # -------

zipline/zmq_utils.py

@@ -5,11 +5,11 @@ def ZmqConsole(sock_typ, socket_addr, sock_conn=None, context=None):
 
     context = context or zmq.Context.instance()
     socket = context.socket(zmq.PULL)
-    socket.connect('tcp://127.0.0.1:3141')
+    socket.bind(socket_addr)
 
     def console():
         while True:
-            msg = socket.recv()
+            msg = socket.recv_pyobj()
             print msg
             import pdb; pdb.set_trace()