Merge pull request #7 from quantopian/master_finance_merge

Master finance merge

setup.cfg

@@ -2,16 +2,16 @@
 verbosity=2
 detailed-errors=1
 
-with-xcoverage=1
-cover-package=zipline
-cover-erase=1
-cover-html=1
-cover-html-dir=docs/_build/html/cover
-with-xunit=1 
+#with-xcoverage=1
+#cover-package=zipline
+#cover-erase=1
+#cover-html=1
+#cover-html-dir=docs/_build/html/cover
+#with-xunit=1 
 #processes=2
 
 
 # Drop into debugger on failure
-pdb=0
-pdb-failures=0
+#pdb=0
+#pdb-failures=0
 

zipline/finance/trading.py

@@ -1,5 +1,6 @@
-import json
 import datetime
+import pytz
+import math
 
 from zmq.core.poll import select
 
@@ -17,7 +18,7 @@ class TradeSimulationClient(qmsg.Component):
     
     @property
     def get_id(self):
-        return "TRADING_CLIENT"
+        return str(zp.FINANCE_COMPONENT.TRADING_CLIENT)
     
     def open(self):
         self.result_feed = self.connect_result()
@@ -25,107 +26,80 @@ class TradeSimulationClient(qmsg.Component):
     
     def do_work(self):
         #next feed event
-        (rlist, wlist, xlist) = select([self.result_feed],
-                                                  [],
-                                                  [self.result_feed],
-                                                  timeout=self.heartbeat_timeout/1000) #select timeout is in sec
-        #
-        #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()
-            return #leave open orders hanging? client requests for orders?
+        socks = dict(self.poll.poll(self.heartbeat_timeout))
+
+        if self.result_feed in socks and socks[self.result_feed] == self.zmq.POLLIN:   
+            msg = self.result_feed.recv()
+
+            if msg == str(zp.CONTROL_PROTOCOL.DONE):
+                qutil.LOGGER.info("Client is DONE!")
+                self.signal_done()
+                return
             
-        event = zp.MERGE_UNFRAME(message)
-        self._handle_event(event)
+            event = zp.MERGE_UNFRAME(msg)
+            self._handle_event(event)
     
     def connect_order(self):
         return self.connect_push_socket(self.addresses['order_address'])
     
     def _handle_event(self, event):
-        self.event_queue.append(event)
-        if event.ALGO_TIME <= event.dt:
-            #event occurred in the present, send the queue to be processed
-            self.handle_events(self.event_queue)
-        self.order_socket.send(str(zp.CONTROL_PROTOCOL.DONE))
+        self.handle_event(event)
+        #signal done to order source.
+        self.order_socket.send(str(zp.ORDER_PROTOCOL.BREAK))
     
-    def handle_events(self, event_queue):
+    def handle_event(self, event):
         raise NotImplementedError    
     
     def order(self, sid, amount):
         self.order_socket.send(zp.ORDER_FRAME(sid, amount))
         
-    
-
-class TradeSimulator(qmsg.BaseTransform):
-    
-    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
+    def signal_order_done(self):
+        self.order_socket.send(str(zp.ORDER_PROTOCOL.DONE))
         
+class OrderDataSource(qmsg.DataSource):
+    """DataSource that relays orders from the client"""
+
+    def __init__(self, simulation_dt):
+        """
+        :param simulation_time: datetime in UTC timezone, sets the start time of simulation. orders
+            will be timestamped relative to this datetime.
+                event = {
+                    'sid'    : an integer for security id,
+                    'dt'     : datetime object,
+                    'price'  : float for price,
+                    'volume' : integer for volume
+                }
+        """
+        qmsg.DataSource.__init__(self, zp.FINANCE_COMPONENT.ORDER_SOURCE)
+        self.simulation_dt = simulation_dt
+        self.last_iteration_duration = datetime.timedelta(seconds=0)
+        self.sent_count         = 0
+
     @property
-    def get_id(self):
-        return "ALGO_TIME"    
-    
+    def get_type(self):
+        return zp.DATASOURCE_TYPE.ORDER
+        
     def open(self):
-        qmsg.BaseTransform.open(self)
+        qmsg.DataSource.open(self)
         self.order_socket = self.bind_order()
         
     def bind_order(self):
         return self.bind_pull_socket(self.addresses['order_address'])
 
-    def do_work(self):
-        """
-        Pulls one message from the event feed, then
-        loops on orders until client sends DONE message.
-        """
-        
-        #next feed event
-        (rlist, wlist, xlist) = select([self.feed_socket],
-                                                  [],
-                                                  [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)
-        
-        if self.last_iteration_duration != None:
-            self.algo_time = self.last_event_time + self.last_iteration_duration
-        else:
-            self.algo_time = event.dt #base case, first event we're transporting.
-        
-        self.last_event_time = event.dt
-        
-        if self.algo_time < self.last_event_time:
-            #compress time, move algo's clock to the time of this event
-            self.algo_time = self.last_event_time
-            
-        #self.process_orders(event)
-
+    def do_work(self):    
         #mark the start time for client's processing of this event.
         self.event_start = datetime.datetime.utcnow()
-        self.result_socket.send(zp.TRANSFORM_FRAME('ALGO_TIME', self.algo_time), self.zmq.NOBLOCK)
+        self.simulation_dt = self.simulation_dt + self.last_iteration_duration
         
-            
-        while True: #this loop should also poll for portfolio state req/rep
+        #TODO: if this is the first iteration, break deadlock by sending a dummy order
+        if(self.sent_count == 0):
+            self.send_dummy()
+        
+        #pull all orders from client.
+        orders = []
+        order_dt = None
+        count = 0
+        while True:
             (rlist, wlist, xlist) = select([self.order_socket],
                                            [],
                                            [self.order_socket],
@@ -136,20 +110,133 @@ class TradeSimulator(qmsg.BaseTransform):
                 continue
                 
             order_msg = rlist[0].recv()
-            if order_msg == str(zp.CONTROL_PROTOCOL.DONE):
+            if order_msg == str(zp.ORDER_PROTOCOL.DONE):
+                self.signal_done()
+                return
+                
+            if order_msg == str(zp.ORDER_PROTOCOL.BREAK):
                 qutil.LOGGER.info("order loop finished")
                 break
 
             sid, amount = zp.ORDER_UNFRAME(order_msg)
-            self.add_open_order(sid, amount)
+            #send the order along
             
-        #end of order processing loop
-        self.last_iteration_duration = datetime.datetime.utcnow() - self.event_start
-        
+            self.last_iteration_duration = datetime.datetime.utcnow() - self.event_start
+            dt = self.simulation_dt + self.last_iteration_duration
+            order_event = zp.namedict({"sid":sid, "amount":amount, "dt":dt})
             
-    def add_open_order(self, sid, amount):
-        self.order_count = self.order_count + 1
+            self.send(order_event)
+            count += 1
+            self.sent_count += 1
+    
+        #TODO: we have to send at least one dummy order per do_work iteration or the feed will block waiting for our messages.
+        if(count == 0):
+            self.send_dummy()
+            self.sent_count += 1
         
-    def process_orders(self, event):
-        #TODO put real fill logic here, return a list of fills
-        return [{'sid':133, 'amount':-100}]
+    def send_dummy(self):
+        dt = self.simulation_dt + self.last_iteration_duration
+        dummy_order = zp.namedict({"sid":0, "amount":0, "dt":dt})
+        self.send(dummy_order)
+    
+    
+
+class TransactionSimulator(qmsg.BaseTransform):
+    
+    def __init__(self): 
+        qmsg.BaseTransform.__init__(self, zp.TRANSFORM_TYPE.TRANSACTION)
+        self.open_orders                = {}
+        self.order_count                = 0
+        self.trade_windwo                = datetime.timedelta(seconds=30)
+        self.orderTTL                   = datetime.timedelta(days=1)
+        self.volume_share               = 0.05
+        self.commission                 = 0.03
+            
+    def transform(self, event):
+        """
+        Pulls one message from the event feed, then
+        loops on orders until client sends DONE message.
+        """
+        #TODO: need a way to send a placeholder txn, to avoid blocking merge... maybe customize merge to not block on txn?
+        if(event.type == zp.DATASOURCE_TYPE.ORDER):
+            self.add_open_order(event)
+            self.state['value'] = None
+        elif(event.type == zp.DATASOURCE_TYPE.TRADE):
+            txn = self.apply_trade_to_open_orders(event)
+            self.state['value'] = txn
+        else:
+            self.state['value'] = None
+            qutil.LOGGER.info("unexpected event type in transform: {etype}".format(etype=event.type))
+        #TODO: what to do if we get another kind of datasource event.type?
+        
+        return self.state
+            
+    def add_open_order(self, event):
+        """Orders are captured in a buffer by sid. No calculations are done here.
+            Amount is explicitly converted to an int.
+            Orders of amount zero are ignored.
+        """
+        event.amount = int(event.amount)
+        if event.amount == 0:
+            qutil.LOGGER.debug("requested to trade zero shares of {sid}".format(sid=event.sid))
+            return
+            
+        self.order_count += 1
+        
+        if(not self.open_orders.has_key(event.sid)):
+            self.open_orders[event.sid] = []
+        self.open_orders[event.sid].append(event)
+     
+    def apply_trade_to_open_orders(self, event):
+        
+        if(event.volume == 0):
+            #there are zero volume events bc some stocks trade less frequently than once per minute.
+            return self.create_dummy_txn(event.dt)
+            
+        if self.open_orders.has_key(event.sid):
+            orders = self.open_orders[event.sid] 
+        else:
+            return None
+            
+        remaining_orders = []
+        total_order = 0        
+        dt = event.dt
+    
+        for order in orders:
+            #we're using minute bars, so allow orders within 30 seconds of the trade
+            if((order.dt - event.dt) < self.trade_windwo):
+                total_order += order.amount
+                if(order.dt > dt):
+                    dt = order.dt
+            #if the order still has time to live (TTL) keep track
+            elif((self.algo_time - order.dt) < self.orderTTL):
+                remaining_orders.append(order)
+    
+        self.open_orders[event.sid] = remaining_orders
+    
+        if(total_order != 0):
+            direction = total_order / math.fabs(total_order)
+        else:
+            direction = 1
+            
+        volume_share = (direction * total_order) / event.volume
+        if volume_share > .25:
+            volume_share = .25
+        amount = volume_share * event.volume * direction
+        impact = (volume_share)**2 * .1 * direction * event.price
+        return self.create_transaction(event.sid, amount, event.price + impact, dt.replace(tzinfo = pytz.utc), direction)
+    
+        
+    def create_transaction(self, sid, amount, price, dt, direction):                
+        txn = {'sid'                : sid, 
+                'amount'             : int(amount), 
+                'dt'                 : dt, 
+                'price'              : price, 
+                'commission'          : self.commission * amount * direction,
+                'source_id'          : zp.FINANCE_COMPONENT.TRANSACTION_SIM
+                }
+        return zp.namedict(txn) 
+                
+                
+                
+

zipline/messaging.py

@@ -3,10 +3,10 @@ Commonly used messaging components.
 """
 
 import datetime
-import ujson as json
 
 import zipline.util as qutil
 from zipline.component import Component
+import zipline.protocol as zp
 from zipline.protocol import CONTROL_PROTOCOL, COMPONENT_TYPE, \
     COMPONENT_STATE
 
@@ -220,20 +220,27 @@ class ParallelBuffer(Component):
                     self.signal_done()
             else:
                 try:
-                    event = json.loads(message)
-
-                # JSON deserialization error
-                except ValueError as exc:
+                    event = self.unframe(message)
+                # deserialization error
+                except zp.INVALID_DATASOURE_FRAME as exc:
                     return self.signal_exception(exc)
 
                 try:
-                    self.append(event[u'id'], event)
+                    self.append(event)
                     self.send_next()
 
                 # Invalid message
-                except KeyError as exc:
+                except zp.INVALID_DATASOURCE_FRAME as exc:
                     return self.signal_exception(exc)
 
+    #
+    def unframe(self, msg):
+        return zp.DATASOURCE_UNFRAME(msg)
+
+    def frame(self, event):
+        return zp.FEED_FRAME(event)
+    
+
     # -------------
     # Flow Control
     # -------------
@@ -254,17 +261,16 @@ class ParallelBuffer(Component):
             return
 
         event = self.next()
-
-        if event != None:
-            self.feed_socket.send(json.dumps(event), self.zmq.NOBLOCK)
+        if(event != None):
+            self.feed_socket.send(self.frame(event), self.zmq.NOBLOCK)
             self.sent_count += 1
 
-    def append(self, source_id, value):
+    def append(self, event):
         """
         Add an event to the buffer for the source specified by
         source_id.
         """
-        self.data_buffer[source_id].append(value)
+        self.data_buffer[event.source_id].append(event)
         self.received_count += 1
 
     def next(self):
@@ -274,17 +280,22 @@ class ParallelBuffer(Component):
         if not(self.is_full() or self.draining):
             return
 
-        cur = None
-        earliest = None
+        cur_source = None
+        earliest_source = None
+        earliest_event = None
+        #iterate over the queues of events from all sources (1 queue per datasource)
         for events in self.data_buffer.values():
             if len(events) == 0:
                 continue
-            cur = events
-            if (earliest == None) or (cur[0]['dt'] <= earliest[0]['dt']):
-                earliest = cur
+            cur_source = events
+            first_in_list = events[0]
+            
+            if (earliest_event == None) or (first_in_list.dt <= earliest_event.dt):
+                earliest_event = first_in_list
+                earliest_source = cur_source
 
-        if earliest != None:
-            return earliest.pop(0)
+        if earliest_event != None:
+            return earliest_source.pop(0)
 
     def is_full(self):
         """
@@ -350,15 +361,32 @@ class MergedParallelBuffer(ParallelBuffer):
         if(not(self.is_full() or self.draining)):
             return
 
+        #
         #get the raw event from the passthrough transform.
-        result = self.data_buffer["PASSTHROUGH"].pop(0)['value']
+        result = self.data_buffer[zp.TRANSFORM_TYPE.PASSTHROUGH].pop(0).PASSTHROUGH
         for source, events in self.data_buffer.iteritems():
-            if source == "PASSTHROUGH":
+            if source == zp.TRANSFORM_TYPE.PASSTHROUGH:
                 continue
             if len(events) > 0:
                 cur = events.pop(0)
-                result[source] = cur['value']
+                result.merge(cur)
         return result
+        
+    def unframe(self, msg):
+        return zp.TRANSFORM_UNFRAME(msg)   
+    
+    def frame(self, event):
+        return zp.MERGE_FRAME(event)
+
+    def append(self, event):
+        """
+        :param event: a namedict with one entry. key is the name of the transform, value is the transformed value.
+        Add an event to the buffer for the source specified by
+        source_id.
+        """
+
+        self.data_buffer[event.keys()[0]].append(event)
+        self.received_count += 1
 
 
 class BaseTransform(Component):
@@ -425,14 +453,12 @@ class BaseTransform(Component):
                 return
 
             try:
-                event = json.loads(message)
-            except ValueError as exc:
+                event = self.unframe(message)
+            except zp.INVALID_FEED_FRAME as exc:
                 return self.signal_exception(exc)
 
             try:
                 cur_state = self.transform(event)
-                cur_state['dt'] = event['dt']
-                cur_state['id'] = self.state['name']
 
             # This is overloaded, so it can fail in all sorts of
             # unknown ways. Its best to catch it in the
@@ -441,12 +467,18 @@ class BaseTransform(Component):
                 return self.signal_exception(exc)
 
             try:
-                json_frame = json.dumps(cur_state)
-            except ValueError as exc:
+                transform_frame = self.frame(cur_state)
+            except zp.INVALID_TRANSFORM_FRAME as exc:
                 return self.signal_exception(exc)
 
-            self.result_socket.send(json_frame, self.zmq.NOBLOCK)
-
+            self.result_socket.send(transform_frame, self.zmq.NOBLOCK)
+            
+    def frame(self, cur_state):
+        return zp.TRANSFORM_FRAME(cur_state['name'], cur_state['value'])
+        
+    def unframe(self, msg):
+        return zp.FEED_UNFRAME(msg)
+        
     def transform(self, event):
         """
         Must return the transformed value as a map with::
@@ -476,7 +508,6 @@ class PassthroughTransform(BaseTransform):
 
     def __init__(self):
         BaseTransform.__init__(self, "PASSTHROUGH")
-
         self.init()
 
     def init(self):
@@ -486,8 +517,9 @@ class PassthroughTransform(BaseTransform):
     def get_type(self):
         return COMPONENT_TYPE.CONDUIT
 
+    #TODO, could save some cycles by skipping the _UNFRAME call and just setting value to original msg string.
     def transform(self, event):
-        return { 'value': event }
+        return {'name':zp.TRANSFORM_TYPE.PASSTHROUGH, 'value': zp.FEED_FRAME(event) }
 
 
 class DataSource(Component):
@@ -520,14 +552,17 @@ class DataSource(Component):
         """
         Emit data.
         """
-        assert isinstance(event, dict)
+        assert isinstance(event, zp.namedict)
 
-        event['id'] = self.id
-        event['type'] = self.get_type()
+        event['source_id'] = self.get_id
+        event['type'] = self.get_type
 
         try:
-            json_frame = json.dumps(event)
-        except ValueError as exc:
+            ds_frame = self.frame(event)
+        except zp.INVALID_DATASOURCE_FRAME as exc:
             return self.signal_exception(exc)
 
-        self.data_socket.send(json_frame)
+        self.data_socket.send(ds_frame)
+
+    def frame(self, event):
+        return zp.DATASOURCE_FRAME(event)

zipline/protocol.py

@@ -10,7 +10,7 @@ Notes
 
 Msgpack
 -------
-Msgpack is the fastest seriaization protocol in Python at the
+Msgpack is the fastest serialization protocol in Python at the
 moment. Its 100% C is typically orders of magnitude faster than
 json and pickle making it awesome for ZeroMQ.
 
@@ -39,12 +39,88 @@ that anything outside of UTF8 can cause serious problems, so if
 you have a strong desire to JSON encode ancient Sanskrit
 ( admit it, we all do ), just say no.
 
+Data Structures
+===============
+
+Enum
+----
+
+Classic C style enumeration::
+
+    opts = Enum('FOO', 'BAR')
+
+    opts.FOO # 0
+    opts.BAR # 1
+    opts.FOO = opts.BAR # False
+
+Oh, and if you do this::
+
+    protocol.Enum([1,2,3])
+
+Your interpreter will segfault, think of this like an extreme assert.
+
+Namedict
+--------
+
+Namedicts are dict like objects that have fields accessible by attribute lookup
+as well as being indexable and iterable::
+
+    HEARTBEAT_PROTOCOL = namedict({
+        'REQ' : b'\x01',
+        'REP' : b'\x02',
+    })
+
+    HEARTBEAT_PROTOCOL.REQ # syntactic sugar
+    HEARTBEAT_PROTOCOL.REP # oh suga suga
+
+    HEARTBEAT_PROTOCOL['REQ'] # classic dictionary index
+
+Namedtuple
+----------
+
+From the standard library, namedtuples are great for specifying
+containers for spec'ing data container objects::
+
+    from collections import namedtuple
+
+    Person = namedtuple('Person', 'name age gender')
+    bob = Person(name='Bob', age=30, gender='male')
+
+    bob.name   # 'Bob'
+    bob.age    # 30
+    bob.gender # male
+
+    # The slots on the tuple are also finite and read-only. This
+    # is a good thing, keeps us honest!
+
+    bob.hobby = 'underwater archery'
+    # Will raise:
+    # AttributeError: 'Person' object has no attribute 'hobby'
+
+    bob.name = 'joe'
+    # Will raise:
+    # AttributeError: can't set attribute
+
+    # Namedtuples are normally read-only, but you can change the
+    # internals using a private operation.
+    bob._replace(gender='female')
+
+    # You can also dump out to dictionary form:
+    OrderedDict([('name', 'Bob'), ('age', 30), ('gender', 'male')])
+
+    # Or JSON.
+    json.dumps(bob._asdict())
+    '{"gender":"male","age":30,"name":"Bob"}'
+
 """
 
 import msgpack
 import numbers
 import datetime
 import pytz
+import copy
+from collections import namedtuple
+
 import zipline.util as qutil
 #import ujson
 #import ultrajson_numpy
@@ -68,12 +144,13 @@ def FrameExceptionFactory(name):
     class InvalidFrame(Exception):
         def __init__(self, got):
             self.got = got
+
         def __str__(self):
             return "Invalid {framecls} Frame: {got}".format(
                 framecls = name,
                 got = self.got,
             )
-            
+
     return InvalidFrame
 
 class namedict(object):
@@ -90,24 +167,42 @@ class namedict(object):
     def __init__(self, dct=None):
         if(dct):
             self.__dict__.update(dct)
-    
+
     def __setitem__(self, key, value):
-        """Required for use by pymongo as_class parameter to find."""
+        """
+        Required for use by pymongo as_class parameter to find.
+        """
         if(key == '_id'):
             self.__dict__['id'] = value
         else:
             self.__dict__[key] = value
-    
+
+    def __getitem__(self, key):
+        return self.__dict__[key]
+
+    def keys(self):
+        return self.__dict__.keys()
+
+    def as_dict(self):
+        # shallow copy is O(n)
+        return copy.copy(self.__dict__)
+
+    def delete(self, key):
+        del(self.__dict__[key])
+
     def merge(self, other_nd):
         assert isinstance(other_nd, namedict)
         self.__dict__.update(other_nd.__dict__)
-        
+
     def __repr__(self):
         return "namedict: " + str(self.__dict__)
-    
+
     def __eq__(self, other):
-        return self.__dict__ == other.__dict__
-    
+        # !!!!!!!!!!!!!!!!!!!!
+        # !!!! DANGEROUS !!!!!
+        # !!!!!!!!!!!!!!!!!!!!
+        return other != None and self.__dict__ == other.__dict__
+
     def has_attr(self, name):
         return self.__dict__.has_key(name)
 
@@ -187,47 +282,59 @@ INVALID_DATASOURCE_FRAME = FrameExceptionFactory('DATASOURCE')
 
 def DATASOURCE_FRAME(event):
     """
-    wraps any datasource payload with id and type, so that unpacking may choose the write
-    UNFRAME for the payload.
-    ::ds_id:: an identifier that is unique to the datasource in the context of a component host (e.g. Simulator
+    Wraps any datasource payload with id and type, so that unpacking may choose
+    the write UNFRAME for the payload.
+
+    ::ds_id:: an identifier that is unique to the datasource in the context of
+    a component host (e.g. Simulator
     ::ds_type:: a string denoting the datasource type. Must be on of::
         TRADE
         (others to follow soon)
     ::payload:: a msgpack string carrying the payload for the frame
     """
+
     assert isinstance(event.source_id, basestring)
-    assert isinstance(event.type, basestring)
-    if(event.type == "TRADE"):
+    assert isinstance(event.type, int), 'Unexpected type %s' % (event.type)
+
+    if(event.type == DATASOURCE_TYPE.TRADE):
         return msgpack.dumps(tuple([event.type, TRADE_FRAME(event)]))
+    elif(event.type == DATASOURCE_TYPE.ORDER):
+        return msgpack.dumps(tuple([event.type, ORDER_SOURCE_FRAME(event)]))
     else:
         raise INVALID_DATASOURCE_FRAME(str(event))
-    
+
 def DATASOURCE_UNFRAME(msg):
     """
-    extracts payload, and calls correct UNFRAME method based on the datasource type passed along
+    Extracts payload, and calls correct UNFRAME method based on the datasource
+    type passed along.
+
     returns a dict containing at least::
         - source_id
         - type
+
     other properties are added based on the datasource type::
         - TRADE::
             - sid - int security identifier
             - price - float
-            - volume - int 
+            - volume - int
             - dt - a datetime object
     """
+
     try:
         ds_type, payload = msgpack.loads(msg)
-        assert isinstance(ds_type, basestring)
-        if(ds_type == "TRADE"):
+        assert isinstance(ds_type, int)
+        if(ds_type == DATASOURCE_TYPE.TRADE):
             return TRADE_UNFRAME(payload)
+        elif(ds_type == DATASOURCE_TYPE.ORDER):
+            return ORDER_SOURCE_UNFRAME(payload)
         else:
             raise INVALID_DATASOURCE_FRAME(msg)
-            
+
     except TypeError:
         raise INVALID_DATASOURCE_FRAME(msg)
     except ValueError:
         raise INVALID_DATASOURCE_FRAME(msg)
-        
+
 # ==================
 # Feed Protocol
 # ==================
@@ -236,16 +343,16 @@ INVALID_FEED_FRAME = FrameExceptionFactory('FEED')
 def FEED_FRAME(event):
     """
     :event: a nameddict with at least::
-        - source_id 
+        - source_id
         - type
     """
     assert isinstance(event, namedict)
     source_id = event.source_id
     ds_type = event.type
     PACK_DATE(event)
-    payload = event.__dict__
+    payload = event.as_dict()
     return msgpack.dumps(payload)
-    
+
 def FEED_UNFRAME(msg):
     try:
         payload = msgpack.loads(msg)
@@ -258,54 +365,42 @@ def FEED_UNFRAME(msg):
         raise INVALID_FEED_FRAME(msg)
     except ValueError:
         raise INVALID_FEED_FRAME(msg)
-        
+
 # ==================
 # Transform Protocol
 # ==================
 INVALID_TRANSFORM_FRAME = FrameExceptionFactory('TRANSFORM')
 
 def TRANSFORM_FRAME(name, value):
-    """
-    :event: a nameddict with at least::
-        - source_id 
-        - type
-    """
     assert isinstance(name, basestring)
-    assert value != None
-    
-    if(name == 'ALGO_TIME'):
-        value = PACK_ALGO_DT(value)
-        
+    if value == None:
+        return msgpack.dumps(tuple([name, TRANSFORM_TYPE.EMPTY]))
+    if(name == TRANSFORM_TYPE.TRANSACTION):
+        value = TRANSACTION_FRAME(value)
     return msgpack.dumps(tuple([name, value]))
-    
+
 def TRANSFORM_UNFRAME(msg):
     """
     :rtype: namedict with <transform_name>:<transform_value>
     """
     try:
+
         name, value = msgpack.loads(msg)
+        if(value == TRANSFORM_TYPE.EMPTY):
+            return namedict({name : None})
         #TODO: anything we can do to assert more about the content of the dict?
         assert isinstance(name, basestring)
-        if(name == "PASSTHROUGH"):
+        if(name == TRANSFORM_TYPE.PASSTHROUGH):
             value = FEED_UNFRAME(value)
-        elif(name == "ALGO_TIME"):
-            value = UNPACK_ALGO_DT(value)
+        elif(name == TRANSFORM_TYPE.TRANSACTION):
+            value = TRANSACTION_UNFRAME(value)
+
         return namedict({name : value})
     except TypeError:
         raise INVALID_TRANSFORM_FRAME(msg)
     except ValueError:
         raise INVALID_TRANSFORM_FRAME(msg)
 
-def PACK_ALGO_DT(value):
-    value = namedict({'dt' : value})
-    PACK_DATE(value)
-    return value.__dict__
-
-def UNPACK_ALGO_DT(value):    
-    value = namedict(value)
-    UNPACK_DATE(value)
-    return value.dt
-
 # ==================
 # Merge Protocol
 # ==================
@@ -314,27 +409,30 @@ INVALID_MERGE_FRAME = FrameExceptionFactory('MERGE')
 def MERGE_FRAME(event):
     """
     :event: a nameddict with at least::
-        - source_id 
+        - source_id
         - type
     """
     assert isinstance(event, namedict)
-    assert isinstance(event.dt, datetime.datetime)
     PACK_DATE(event)
-    if(event.has_attr('ALGO_TIME')):
-        event.ALGO_TIME = PACK_ALGO_DT(event.ALGO_TIME)
-    payload = event.__dict__
+    if(event.has_attr(TRANSFORM_TYPE.TRANSACTION)):
+        if(event.TRANSACTION == None):
+            event.TRANSACTION = TRANSFORM_TYPE.EMPTY
+        else:
+            event.TRANSACTION = TRANSACTION_FRAME(event.TRANSACTION)
+    payload = event.as_dict()
     return msgpack.dumps(payload)
-    
+
 def MERGE_UNFRAME(msg):
     try:
         payload = msgpack.loads(msg)
         #TODO: anything we can do to assert more about the content of the dict?
         assert isinstance(payload, dict)
         payload = namedict(payload)
-        if(payload.has_attr('ALGO_TIME')):
-            payload.ALGO_TIME = UNPACK_ALGO_DT(payload.ALGO_TIME)
-        assert isinstance(payload.epoch, numbers.Integral)
-        assert isinstance(payload.micros, numbers.Integral)
+        if(payload.has_attr(TRANSFORM_TYPE.TRANSACTION)):
+            if(payload.TRANSACTION == TRANSFORM_TYPE.EMPTY):
+                payload.TRANSACTION = None
+            else:
+                payload.TRANSACTION = TRANSACTION_UNFRAME(payload.TRANSACTION)
         UNPACK_DATE(payload)
         return payload
     except TypeError:
@@ -342,16 +440,15 @@ def MERGE_UNFRAME(msg):
     except ValueError:
         raise INVALID_MERGE_FRAME(msg)
 
-    
+
 # ==================
 # Finance Protocol
 # ==================
-
 INVALID_ORDER_FRAME = FrameExceptionFactory('ORDER')
 INVALID_TRADE_FRAME = FrameExceptionFactory('TRADE')
 
 # ==================
-# Trades
+# Trades - Should only be called from inside DATASOURCE_ (UN)FRAME.
 # ==================
 
 def TRADE_FRAME(event):
@@ -361,27 +458,42 @@ def TRADE_FRAME(event):
             - price     -- float of the price printed for the trade
             - volume    -- int for shares in the trade
             - dt        -- datetime for the trade
-            
+
     """
     assert isinstance(event, namedict)
     assert isinstance(event.source_id, basestring)
-    assert event.type == "TRADE"
+    assert event.type == DATASOURCE_TYPE.TRADE
     assert isinstance(event.sid, int)
     assert isinstance(event.price, float)
     assert isinstance(event.volume, int)
     PACK_DATE(event)
-    return msgpack.dumps(tuple([event.sid, event.price, event.volume, event.epoch, event.micros, event.type, event.source_id]))
-    
+    return msgpack.dumps(tuple([
+        event.sid,
+        event.price,
+        event.volume,
+        event.epoch,
+        event.micros,
+        event.type,
+        event.source_id
+    ]))
+
 def TRADE_UNFRAME(msg):
     try:
-        sid, price, volume, epoch, micros, source_type, source_id = msgpack.loads(msg)
-        
+        packed = msgpack.loads(msg)
+        sid, price, volume, epoch, micros, source_type, source_id = packed
+
         assert isinstance(sid, int)
         assert isinstance(price, float)
         assert isinstance(volume, int)
-        assert isinstance(epoch, numbers.Integral)
-        assert isinstance(micros, numbers.Integral)
-        rval = namedict({'sid' : sid, 'price' : price, 'volume' : volume, 'epoch' : epoch, 'micros' : micros, 'type' : source_type, 'source_id' : source_id})
+        rval = namedict({
+            'sid'       : sid,
+            'price'     : price,
+            'volume'    : volume,
+            'epoch'     : epoch,
+            'micros'    : micros,
+            'type'      : source_type,
+            'source_id' : source_id
+        })
         UNPACK_DATE(rval)
         return rval
     except TypeError:
@@ -390,14 +502,14 @@ def TRADE_UNFRAME(msg):
         raise INVALID_TRADE_FRAME(msg)
 
 # =========
-# Orders
+# Orders - from client to order source
 # =========
 
 def ORDER_FRAME(sid, amount):
     assert isinstance(sid, int)
-    assert isinstance(amount, int) #no partial shares...   
+    assert isinstance(amount, int) #no partial shares...
     return msgpack.dumps(tuple([sid, amount]))
-    
+
 
 def ORDER_UNFRAME(msg):
     try:
@@ -410,26 +522,166 @@ def ORDER_UNFRAME(msg):
         raise INVALID_ORDER_FRAME(msg)
     except ValueError:
         raise INVALID_ORDER_FRAME(msg)
-        
+
+#
+# ==================
+# TRANSACTIONS - Should only be called from inside TRANSFORM_(UN)FRAME.
+# ==================
+
+def TRANSACTION_FRAME(event):
+    assert isinstance(event, namedict)
+    assert isinstance(event.sid, int)
+    assert isinstance(event.price, float)
+    assert isinstance(event.commission, float)
+    assert isinstance(event.amount, int)
+    PACK_DATE(event)
+    return msgpack.dumps(tuple([
+        event.sid,
+        event.price,
+        event.amount,
+        event.commission,
+        event.epoch,
+        event.micros
+    ]))
+
+def TRANSACTION_UNFRAME(msg):
+    try:
+        sid, price, amount, commission, epoch, micros = msgpack.loads(msg)
+
+        assert isinstance(sid, int)
+        assert isinstance(price, float)
+        assert isinstance(commission, float)
+        assert isinstance(amount, int)
+        rval = namedict({
+            'sid'        : sid,
+            'price'      : price,
+            'amount'     : amount,
+            'commission' : commission,
+            'epoch'      : epoch,
+            'micros'     : micros
+        })
+
+        UNPACK_DATE(rval)
+        return rval
+    except TypeError:
+        raise INVALID_TRADE_FRAME(msg)
+    except ValueError:
+        raise INVALID_TRADE_FRAME(msg)
+
+
+# =========
+# Orders - from order source to feed
+#        - should only be called from inside DATASOURCE_(UN)FRAME
+# =========
+
+def ORDER_SOURCE_FRAME(event):
+    assert isinstance(event.sid, int)
+    assert isinstance(event.amount, int) #no partial shares...
+    assert isinstance(event.source_id, basestring)
+    assert event.type == DATASOURCE_TYPE.ORDER
+    PACK_DATE(event)
+    return msgpack.dumps(tuple([
+        event.sid,
+        event.amount,
+        event.epoch,
+        event.micros,
+        event.source_id,
+        event.type
+    ]))
+
+
+def ORDER_SOURCE_UNFRAME(msg):
+    try:
+        sid, amount, epoch, micros, source_id, source_type = msgpack.loads(msg)
+        event = namedict({
+            "sid"       : sid,
+            "amount"    : amount,
+            "epoch"     : epoch,
+            "micros"    : micros,
+            "source_id" : source_id,
+            "type"      : source_type
+        })
+        assert isinstance(sid, int)
+        assert isinstance(amount, int)
+        assert isinstance(source_id, basestring)
+        assert isinstance(source_type, int)
+        UNPACK_DATE(event)
+        return event
+    except TypeError:
+        raise INVALID_ORDER_FRAME(msg)
+    except ValueError:
+        raise INVALID_ORDER_FRAME(msg)
+
 # =================
 # Date Helpers
 # =================
 
-def PACK_DATE(event):    
+def PACK_DATE(event):
+    """
+    Packs the datetime property of event into msgpack'able longs.
+    This function should be called purely for its side effects. 
+    The event's 'dt' property is replaced by two longs: epoch and micros. 
+    Epoch is the unix epoch time in UTC, and micros is the microsecond 
+    property of the original event.dt datetime object.
+    
+    PACK_DATE and UNPACK_DATE are inverse operations. 
+    
+    :param event: event must a namedict with a property named 'dt' that is a datetime.
+    :rtype: None
+    """
     assert isinstance(event.dt, datetime.datetime)
     assert event.dt.tzinfo == pytz.utc #utc only please
     epoch = long(event.dt.strftime('%s'))
     event['epoch'] = epoch
     event['micros'] = event.dt.microsecond
-    del(event.__dict__['dt'])
-    return event
+    event.delete('dt')
 
-def UNPACK_DATE(payload):
-    assert isinstance(payload.epoch, numbers.Integral)
-    assert isinstance(payload.micros, numbers.Integral)
-    dt = datetime.datetime.fromtimestamp(payload.epoch)
-    dt = dt.replace(microsecond = payload.micros, tzinfo = pytz.utc)
-    del(payload.__dict__['epoch'])
-    del(payload.__dict__['micros'])
-    payload['dt'] = dt
-    return payload
+def UNPACK_DATE(event):
+    """
+    Unpacks the datetime property of event from msgpack'able longs.
+    This function should be called purely for its side effects. 
+    The event's 'dt' property is created by reading two longs: epoch and micros. 
+    
+    UNPACK_DATE and PACK_DATE are inverse operations. 
+    
+    :param event: event must a namedict with::
+            - a property named 'epoch' that is an integral representing the unix \
+             epoch time in UTC
+            - a property named 'micros' that is an integral the microsecond \
+            property of the original event.dt datetime object
+    :rtype: None
+    """
+    assert isinstance(event.epoch, numbers.Integral)
+    assert isinstance(event.micros, numbers.Integral)
+    dt = datetime.datetime.fromtimestamp(event.epoch)
+    dt = dt.replace(microsecond = event.micros, tzinfo = pytz.utc)
+    event.delete('epoch')
+    event.delete('micros')
+    event.dt = dt
+
+
+DATASOURCE_TYPE = Enum(
+    'ORDER',
+    'TRADE'
+)
+
+ORDER_PROTOCOL = Enum(
+    'DONE',
+    'BREAK'
+)
+
+
+#Transform type needs to be a namedict to facilitate merging.
+TRANSFORM_TYPE = namedict({
+    'TRANSACTION' : 'TRANSACTION', #needed?
+    'PASSTHROUGH' : 'PASSTHROUGH',
+    'EMPTY'       : ''
+})
+
+
+FINANCE_COMPONENT = namedict({
+    'TRADING_CLIENT'   : 'TRADING_CLIENT',
+    'PORTFOLIO_CLIENT' : 'PORTFOLIO_CLIENT',
+    'ORDER_SOURCE'     : 'ORDER_SOURCE',
+    'TRANSACTION_SIM'  : 'TRANSACTION_SIM'
+})

zipline/simulator.py

@@ -0,0 +1,85 @@
+"""
+Simulator hosts all the components necessary to execute a simluation. See :py:method""
+"""
+
+import threading
+import mock
+from collections import defaultdict
+from zipline.monitor import Controller
+from zipline.messaging import ComponentHost
+import zipline.util as qutil
+
+class AddressAllocator(object):
+
+    def __init__(self, ns):
+        self.idx = 0
+        self.sockets = [
+            'tcp://127.0.0.1:%s' % (10000 + n)
+            for n in xrange(ns)
+        ]
+
+    def lease(self, n):
+        sockets = self.sockets[self.idx:self.idx+n]
+        self.idx += n
+        return sockets
+
+    def reaquire(self, *conn):
+        pass
+
+#
+class Simulator(ComponentHost):
+
+    def __init__(self, addresses):
+        ComponentHost.__init__(self, addresses)
+        self.subthreads = []
+        self.running = False
+
+    def launch_controller(self):
+        thread = threading.Thread(target=self.controller.run)
+        thread.start()
+
+        self.subthreads.append(thread)
+        return thread
+
+    def simulate(self):
+        thread = threading.Thread(target=self.run)
+        thread.start()
+
+        self.subthreads.append(thread)
+        self.running = True
+
+        return thread
+
+    def did_clean_shutdown(self):
+        return not any([t.isAlive() for t in self.subthreads])
+
+    def shutdown(self):
+        """
+        Destroy all tracked components.
+        """
+
+        if not self.running:
+            return
+
+        try:
+            self.controller.shutdown(context=self.context)
+        except:
+            import pdb; pdb.set_trace()
+
+        for component in self.components.itervalues():
+            component.shutdown()
+
+        for thread in self.subthreads:
+            if thread.is_alive():
+                thread._Thread__stop()
+
+        self.running = False
+
+        assert self.did_clean_shutdown()
+
+    def launch_component(self, component):
+        thread = threading.Thread(target=component.run)
+        thread.start()
+
+        self.subthreads.append(thread)
+        return thread

zipline/sources.py

@@ -5,23 +5,28 @@ import datetime
 import random
 import pytz
 
-import zipline.util as qutil
 import zipline.messaging as zm
 import zipline.protocol as zp
 
+
 class TradeDataSource(zm.DataSource):
-    
+
     def send(self, event):
-        """ :param dict event: is a trade event with data as per :py:func: `zipline.protocol.TRADE_FRAME`
-            :rtype: None
+        """
+        :param dict event: is a trade event with data as per
+                           :py:func: `zipline.protocol.TRADE_FRAME`
+        :rtype: None
         """
         event.source_id = self.get_id
         message = zp.DATASOURCE_FRAME(event)
         self.data_socket.send(message)
 
+
 class RandomEquityTrades(TradeDataSource):
-    """Generates a random stream of trades for testing."""
-    
+    """
+    Generates a random stream of trades for testing.
+    """
+
     def __init__(self, sid, source_id, count):
         zm.DataSource.__init__(self, source_id)
         self.count          = count
@@ -30,32 +35,42 @@ class RandomEquityTrades(TradeDataSource):
         self.trade_start    = datetime.datetime.now().replace(tzinfo=pytz.utc)
         self.minute         = datetime.timedelta(minutes=1)
         self.price          = random.uniform(5.0, 50.0)
-    
-    
+
+
     def get_type(self):
-        return 'equity_trade'
-    
-    
+        zp.COMPONENT_TYPE.SOURCE
+
     def do_work(self):
         if(self.incr == self.count):
             self.signal_done()
             return
-        
-        self.price = self.price + random.uniform(-0.05, 0.05)        
-        self._send(self.sid, self.price, random.randrange(100,10000,100), self.trade_start + (self.minute * self.incr))
-        self.incr += 1        
 
-    def _send(self, sid, price, volume, dt):
-        event = zp.namedict({'source_id': self.get_id, "type" : "TRADE", "sid":sid, "price":price, "volume":volume, "dt":dt})
+        self.price = self.price + random.uniform(-0.05, 0.05)
+        volume = random.randrange(100,10000,100)
+
+        event = zp.namedict({
+            "type"      : zp.DATASOURCE_TYPE.TRADE,
+            "sid"       : self.sid,
+            "price"     : self.price,
+            "volume"    : volume,
+            "dt"        : self.trade_start + (self.minute * self.incr),
+        })
+
         self.send(event)
 
+        self.incr += 1
+
 
 class SpecificEquityTrades(TradeDataSource):
-    """Generates a random stream of trades for testing."""
+    """
+    Generates a random stream of trades for testing.
+    """
 
     def __init__(self, source_id, event_list):
         """
-        :event_list: should be a chronologically ordered list of dictionaries in the following form:
+        :event_list: should be a chronologically ordered list of dictionaries
+                     in the following form:
+
                 event = {
                     'sid'    : an integer for security id,
                     'dt'     : datetime object,
@@ -67,14 +82,14 @@ class SpecificEquityTrades(TradeDataSource):
         self.event_list = event_list
 
     def get_type(self):
-        return 'equity_trade'
+        zp.COMPONENT_TYPE.SOURCE
 
     def do_work(self):
         if(len(self.event_list) == 0):
             self.signal_done()
             return
-        
+
         event = self.event_list.pop(0)
         self.send(zp.namedict(event))
-        
+
 

zipline/test/client.py

@@ -1,26 +1,18 @@
-#import logging
-import ujson as json
-
 import zipline.util as qutil
 import zipline.messaging as qmsg
+import zipline.protocol as zp
 from zipline.protocol import CONTROL_PROTOCOL, COMPONENT_TYPE
-
-#qlogger = logging.getLogger('qexec')
+from zipline.finance.trading import TradeSimulationClient
 
 class TestClient(qmsg.Component):
 
-    def __init__(self, utest, expected_msg_count=0):
+    def __init__(self):
         qmsg.Component.__init__(self)
-
-        # Generally shouldn't reference unit tests here.
-        self.utest              = utest
-        self.expected_msg_count = expected_msg_count
-
         self.init()
 
     def init(self):
-        self.received_count = 0
-        self.prev_dt        = None
+        self.received_count     = 0
+        self.prev_dt            = None
 
     @property
     def get_id(self):
@@ -46,20 +38,15 @@ class TestClient(qmsg.Component):
             if msg == str(CONTROL_PROTOCOL.DONE):
                 qutil.LOGGER.info("Client is DONE!")
                 self.signal_done()
-                #self.utest.assertEqual(
-                    #self.expected_msg_count, self.received_count,
-                    #"The client should have received ({n}) the same number of \
-                    #messages as the feed sent ({m})."
-                    #.format(n=self.received_count, m=self.expected_msg_count))
                 return
 
             self.received_count += 1
 
             try:
-                event = json.loads(msg)
+                event = self.unframe(msg)
 
-            # JSON deserialization error
-            except ValueError as exc:
+            # deserialization error
+            except zp.INVALID_MERGE_FRAME as exc:
                 return self.signal_exception(exc)
 
             if self.prev_dt != None:
@@ -70,7 +57,29 @@ class TestClient(qmsg.Component):
                         )
                     )
             else:
-                self.prev_dt = event['dt']
+                self.prev_dt = event.dt
 
             if self.received_count % 100 == 0:
                 qutil.LOGGER.info("received {n} messages".format(n=self.received_count))
+            
+    def unframe(self, msg):
+        return zp.MERGE_UNFRAME(msg)
+
+
+class TestTradingClient(TradeSimulationClient):
+    
+    def __init__(self, sid, amount, order_count):
+        TradeSimulationClient.__init__(self)
+        self.count = order_count
+        self.sid = sid
+        self.amount = amount
+        self.incr = 0
+    
+    def handle_event(self, event):
+        #place an order for 100 shares of sid:133
+        if(self.incr < self.count):
+            self.order(self.sid, self.amount)
+            self.incr += 1
+        else:
+            self.signal_order_done()
+            self.signal_done()

zipline/test/dummy.py

@@ -1,128 +0,0 @@
-"""
-Dummy simulator for test/development on Zipline.
-"""
-
-import threading
-import mock
-from collections import defaultdict
-from zipline.monitor import Controller
-from zipline.messaging import SimulatorBase
-import zipline.util as qutil
-
-class DummyAllocator(object):
-
-    def __init__(self, ns):
-        self.idx = 0
-        self.sockets = [
-            'tcp://127.0.0.1:%s' % (10000 + n)
-            for n in xrange(ns)
-        ]
-
-    def lease(self, n):
-        sockets = self.sockets[self.idx:self.idx+n]
-        self.idx += n
-        return sockets
-
-    def reaquire(self, *conn):
-        pass
-
-
-class ThreadSimulator(SimulatorBase):
-
-    def __init__(self, addresses):
-        SimulatorBase.__init__(self, addresses)
-
-    def launch_controller(self):
-        thread = threading.Thread(target=self.controller.run)
-        thread.start()
-        self.cuc = thread
-        return thread
-
-    def launch_component(self, component):
-        thread = threading.Thread(target=component.run)
-        thread.start()
-        return thread
-        
-class ExecutorMixinBase(object):
-    """Abstract base to allow mixin for tests that need a dummy simulator."""
-    leased_sockets = defaultdict(list)
-
-    def setUp(self):
-        self.setup_logging()
-
-        # TODO: how to make Nose use this cross-process????
-        self.setup_allocator()
-
-    def tearDown(self):
-        pass
-        #self.unallocate_sockets()
-
-        # Assert the sockets were properly cleaned up
-        #self.assertEmpty(self.leased_sockets[self.id()].values())
-
-        # Assert they were returned to the heap
-        #self.allocator.socketheap.assert
-
-    def get_simulator(self):
-        """
-        Return a new simulator instance to be tested.
-        """
-        raise NotImplementedError
-
-    def get_controller(self):
-        """
-        Return a new controler for simulator instance to be tested.
-        """
-        raise NotImplementedError
-
-    def setup_allocator(self):
-        """
-        Setup the socket allocator for this test case.
-        """
-        raise NotImplementedError
-
-    def allocate_sockets(self, n):
-        """
-        Allocate sockets local to this test case, track them so
-        we can gc after test run.
-        """
-
-        assert isinstance(n, int)
-        assert n > 0
-
-        leased = self.allocator.lease(n)
-
-        self.leased_sockets[self.id()].extend(leased)
-        return leased
-
-    def unallocate_sockets(self):
-        self.allocator.reaquire(*self.leased_sockets[self.id()])
-
-class ThreadPoolExecutorMixin(ExecutorMixinBase):
-    """Dummy server using threads."""
-    allocator = DummyAllocator(100)
-
-    def setup_logging(self):
-        qutil.configure_logging()
-
-        # lazy import by design
-        self.logger = mock.Mock()
-
-    def setup_allocator(self):
-        pass
-
-    def get_simulator(self, addresses):
-        return ThreadSimulator(addresses)
-
-    def get_controller(self):
-        # Allocate two more sockets
-        controller_sockets = self.allocate_sockets(2)
-
-        return Controller(
-            controller_sockets[0],
-            controller_sockets[1],
-            logging = self.logger,
-        )
-
-
-

zipline/test/factory.py

@@ -2,101 +2,50 @@ import datetime
 import pytz
 import zipline.util as qutil
 import zipline.finance.risk as risk
+import zipline.protocol as zp
 
-def createReturns(daycount, start):
-    i = 0
-    test_range = []
-    current = start.replace(tzinfo=pytz.utc)
-    one_day = datetime.timedelta(days = 1)
-    while i < daycount: 
-        i += 1
-        r = daily_return(current, random.random())
-        test_range.append(r)
-        current = current + one_day
-    return [ x for x in test_range if(risk.trading_calendar.is_trading_day(x.date)) ]
 
-def createReturnsFromRange(start, end):
-    current = start.replace(tzinfo=pytz.utc)
-    end = end.replace(tzinfo=pytz.utc)
-    one_day = datetime.timedelta(days = 1)
-    test_range = []
-    i = 0
-    while current <= end: 
-        current = current + one_day
-        if(not risk.trading_calendar.is_trading_day(current)):
-            continue
-        r = daily_return(current, random.random())
-        i += 1
-        test_range.append(r)
-        
-    return test_range
-def createReturnsFromList(returns, start):
-    current = start.replace(tzinfo=pytz.utc)
-    one_day = datetime.timedelta(days = 1)
-    test_range = []
-    i = 0
-    while len(test_range) < len(returns): 
-        if(risk.trading_calendar.is_trading_day(current)):
-            r = daily_return(current, returns[i])
-            i += 1
-            test_range.append(r)
-        current = current + one_day
-    return test_range
-    
-
-def createAlgo(filename):
-    algo = Algorithm()
-    algo.code = getCodeFromFile(filename)
-    algo.title = filename
-    algo._id = pymongo.objectid.ObjectId()
-    hostedAlgo = HostedAlgorithm(algo)
-    return hostedAlgo       
-    
-def getCodeFromFile(filename):
-    rVal = None
-    with open('./test/algo_samples/' + filename, 'r') as f:
-        rVal = f.read()
-    return rVal
-    
-                          
 def create_trade(sid, price, amount, datetime):
-    row = {}
-    row['source_id'] = "test_factory"
-    row['type'] = "TRADE"
-    row['sid'] = sid
-    row['dt'] = datetime
-    row['price'] = price
-    row['volume'] = amount
+    row = {
+        'source_id' : "test_factory",
+        'type'      : zp.DATASOURCE_TYPE.TRADE,
+        'sid'       : sid,
+        'dt'        : datetime,
+        'price'     : price,
+        'volume'    : amount
+    }
     return row
- 
+
 def create_trade_history(sid, prices, amounts, start_time, interval):
     i = 0
     trades = []
     current = start_time.replace(tzinfo = pytz.utc)
     while i < len(prices):
-        if(risk.trading_calendar.is_trading_day(current)):  
+        if(risk.trading_calendar.is_trading_day(current)):
             trades.append(create_trade(sid, prices[i], amounts[i], current))
             current = current + interval
             i += 1
         else:
             current = current + datetime.timedelta(days=1)
-        
-    return trades 
-    
+
+    return trades
+
 def createTxn(sid, price, amount, datetime, btrid=None):
-    txn = Transaction(sid=sid, amount=amount, dt = datetime, 
+    txn = Transaction(sid=sid, amount=amount, dt = datetime,
                       price=price, transaction_cost=-1*price*amount)
     return txn
-    
+
 def createTxnHistory(sid, priceList, amtList, startTime, interval):
-    i = 0
     txns = []
     current = startTime
-    while i < len(priceList):
-        if(risk.trading_calendar.is_trading_day(current)): 
-            txns.append(createTxn(sid,priceList[i],amtList[i], current))
+
+    for price, amount in zip(priceList, amtList):
+
+        if risk.trading_calendar.is_trading_day(current):
+            txns.append(createTxn(sid, price, amount, current))
             current = current + interval
-            i += 1
+
         else:
-            current = current + datetime.timedelta(days=1) 
-    return txns
+            current = current + datetime.timedelta(days=1)
+
+    return txns

zipline/test/test_finance.py

@@ -1,9 +1,10 @@
 """Tests for the zipline.finance package"""
-import datetime
 import mock
 import pytz
 import host_settings
 from unittest2 import TestCase
+from datetime import datetime, timedelta
+
 import zipline.test.factory as factory
 import zipline.util as qutil
 import zipline.db as db
@@ -11,19 +12,29 @@ import zipline.finance.risk as risk
 import zipline.protocol as zp
 
 from zipline.test.client import TestTradingClient
-from zipline.test.dummy import ThreadPoolExecutorMixin
 from zipline.sources import SpecificEquityTrades
-from zipline.finance.trading import TradeSimulator
+from zipline.finance.trading import TransactionSimulator, OrderDataSource
+from zipline.simulator import AddressAllocator, Simulator
+from zipline.monitor import Controller
 
 
-class FinanceTestCase(ThreadPoolExecutorMixin, TestCase):
-            
+class FinanceTestCase(TestCase):
+
+    def setUp(self):
+        qutil.configure_logging()
+
     def test_trade_feed_protocol(self):
-        trades = 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"),
-                                            datetime.timedelta(days=1))
+
+        # TODO: Perhaps something more self-documenting for variables names?
+        a = 133
+        b = [10] * 4
+        c = [100] * 4
+
+        ts = datetime.strptime("02/15/2012","%m/%d/%Y")
+        dt = timedelta(days=1)
+
+        trades = factory.create_trade_history( a, b, c, ts, dt )
+
         for trade in trades:
             #simulate data source sending frame
             msg = zp.DATASOURCE_FRAME(zp.namedict(trade))
@@ -35,8 +46,12 @@ class FinanceTestCase(ThreadPoolExecutorMixin, TestCase):
             recovered_feed = zp.FEED_UNFRAME(feed_msg)
             #do a transform
             trans_msg = zp.TRANSFORM_FRAME('helloworld', 2345.6)
-            #simulate passthrough transform -- passthrough shouldn't even unpack the msg, just resend.
-            passthrough_msg = zp.TRANSFORM_FRAME('PASSTHROUGH', feed_msg)
+            #simulate passthrough transform -- passthrough shouldn't even
+            # unpack the msg, just resend.
+
+            passthrough_msg = zp.TRANSFORM_FRAME(zp.TRANSFORM_TYPE.PASSTHROUGH,\
+                    feed_msg)
+
             #merge unframes transform and passthrough
             trans_recovered = zp.TRANSFORM_UNFRAME(trans_msg)
             pt_recovered = zp.TRANSFORM_UNFRAME(passthrough_msg)
@@ -46,34 +61,73 @@ class FinanceTestCase(ThreadPoolExecutorMixin, TestCase):
             merged_msg = zp.MERGE_FRAME(pt_recovered.PASSTHROUGH)
             #unframe the merge and validate values
             event = zp.MERGE_UNFRAME(merged_msg)
-            
+
             #check the transformed value, should only be in event, not trade.
             self.assertTrue(event.helloworld == 2345.6)
-            del(event.__dict__['helloworld'])
-            
+            event.delete('helloworld')
+
             self.assertEqual(zp.namedict(trade), event)
-            
+
     def test_order_protocol(self):
+        #client places an order
         order_msg = zp.ORDER_FRAME(133, 100)
+
+        #order datasource receives
         sid, amount = zp.ORDER_UNFRAME(order_msg)
         self.assertEqual(sid, 133)
         self.assertEqual(amount, 100)
-    
+
+        #order datasource datasource frames the order
+        order_dt = datetime.utcnow().replace(tzinfo=pytz.utc)
+        order_event = zp.namedict({
+            "sid"        : sid,
+            "amount"     : amount,
+            "dt"         : order_dt,
+            "source_id"  : zp.FINANCE_COMPONENT.ORDER_SOURCE,
+            "type"       : zp.DATASOURCE_TYPE.ORDER
+        })
+
+
+        order_ds_msg = zp.DATASOURCE_FRAME(order_event)
+
+        #transaction transform unframes
+        recovered_order = zp.DATASOURCE_UNFRAME(order_ds_msg)
+
+        self.assertEqual(order_dt, recovered_order.dt)
+
+        #create a transaction from the order
+        txn = zp.namedict({
+            'sid'        : recovered_order.sid,
+            'amount'     : recovered_order.amount,
+            'dt'         : recovered_order.dt,
+            'price'      : 10.0,
+            'commission' : 0.50
+        })
+
+        #frame that transaction
+        txn_msg = zp.TRANSFORM_FRAME(zp.TRANSFORM_TYPE.TRANSACTION, txn)
+
+        #unframe
+        recovered_tx = zp.TRANSFORM_UNFRAME(txn_msg).TRANSACTION
+        self.assertEqual(recovered_tx.sid, 133)
+        self.assertEqual(recovered_tx.amount, 100)
+
     def test_trading_calendar(self):
-        known_trading_day = datetime.datetime.strptime("02/24/2012","%m/%d/%Y")
-        known_holiday     = datetime.datetime.strptime("02/20/2012", "%m/%d/%Y") #president's day
-        saturday          = datetime.datetime.strptime("02/25/2012", "%m/%d/%Y")
+        known_trading_day = datetime.strptime("02/24/2012","%m/%d/%Y")
+        known_holiday     = datetime.strptime("02/20/2012", "%m/%d/%Y") #president's day
+        saturday          = datetime.strptime("02/25/2012", "%m/%d/%Y")
         self.assertTrue(risk.trading_calendar.is_trading_day(known_trading_day))
         self.assertFalse(risk.trading_calendar.is_trading_day(known_holiday))
         self.assertFalse(risk.trading_calendar.is_trading_day(saturday))
-    
+
     def test_orders(self):
 
         # Just verify sending and receiving orders.
         # --------------
 
         # Allocate sockets for the simulator components
-        sockets = self.allocate_sockets(6)
+        allocator = AddressAllocator(8)
+        sockets = allocator.lease(8)
 
         addresses = {
             'sync_address'   : sockets[0],
@@ -84,36 +138,45 @@ class FinanceTestCase(ThreadPoolExecutorMixin, TestCase):
             'order_address'  : sockets[5]
         }
 
-        sim = self.get_simulator(addresses)
-        con = self.get_controller()
+        con = Controller(
+            sockets[6],
+            sockets[7],
+            logging = qutil.LOGGER
+        )
+
+        sim = Simulator(addresses)
 
         # Simulation Components
         # ---------------------
 
-        set1 = SpecificEquityTrades("flat-133",factory.create_trade_history(133,    
-                                                                            [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(10)
-        order_sim = TradeSimulator(expected_orders=10)
+        # TODO: Perhaps something more self-documenting for variables names?
+        a = 133
+        b = [10.0] * 16
+        c = [100] * 16
+        ts = datetime.strptime("02/1/2012","%m/%d/%Y")
+        dt = timedelta(days=1)
 
-        sim.register_components([client, order_sim, set1])
+        trade_history = factory.create_trade_history( a, b, c, ts, dt )
+
+        set1 = SpecificEquityTrades("flat-133",)
+
+        #client sill send 10 orders for 100 shares of 133
+        client = TestTradingClient(133, 100, 10)
+        ts = datetime.strptime("02/1/2012","%m/%d/%Y").replace(tzinfo=pytz.utc)
+
+        order_source = OrderDataSource(ts)
+        transaction_sim = TransactionSimulator()
+
+        sim.register_components([client, order_source, transaction_sim, set1])
         sim.register_controller( con )
 
         # Simulation
         # ----------
-        sim.simulate()
+        sim_context = sim.simulate()
+        sim_context.join()
 
-        # Stop Running
-        # ------------
 
-        # TODO: less abrupt later, just shove a StopIteration
-        # down the pipe to make it stop spinning
-        sim.cuc._Thread__stop()
-
-        self.assertEqual(sim.feed.pending_messages(), 0,
-            "The feed should be drained of all messages, found {n} remaining."
-            .format(n=sim.feed.pending_messages())
-        )
-        
+        # TODO: Make more assertions about the final state of the components.
+        self.assertEqual(sim.feed.pending_messages(), 0, \
+            "The feed should be drained of all messages, found {n} remaining." \
+            .format(n=sim.feed.pending_messages()))

zipline/test/test_messaging.py

@@ -11,6 +11,7 @@ from zipline.sources import RandomEquityTrades
 from zipline.test.client import TestClient
 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.
@@ -73,6 +74,7 @@ class SimulatorTestCase(object):
     #  Cases
     # -------
 
+    @timed(2)
     def test_simple(self):
 
         # Simple test just to make sure that the archiecture is
@@ -100,7 +102,7 @@ class SimulatorTestCase(object):
 
         ret1 = RandomEquityTrades(133, "ret1", 1)
         ret2 = RandomEquityTrades(134, "ret2", 1)
-        client = TestClient(self, expected_msg_count=ret1.count + ret2.count)
+        client = TestClient()
 
         sim.register_controller( con )
         sim.register_components([ret1, ret2, client])
@@ -149,7 +151,7 @@ class SimulatorTestCase(object):
         ret1 = RandomEquityTrades(133, "ret1", 1)
         ret2 = RandomEquityTrades(134, "ret2", 1)
         fail_transform = DivideByZeroTransform("fail")
-        client = TestClient(self, expected_msg_count=ret1.count + ret2.count)
+        client = TestClient()
 
         sim.register_controller( con )
         sim.register_components([ret1, ret2, fail_transform, client])
@@ -194,7 +196,7 @@ class SimulatorTestCase(object):
 
         ret1 = RandomEquityTrades(133, "ret1", 400)
         ret2 = RandomEquityTrades(134, "ret2", 400)
-        client = TestClient(self, expected_msg_count=ret1.count + ret2.count)
+        client = TestClient()
 
         sim.register_controller( con )
         sim.register_components([ret1, ret2, client])
@@ -240,7 +242,7 @@ class SimulatorTestCase(object):
         ret2 = RandomEquityTrades(134, "ret2", 5000)
         mavg1 = MovingAverage("mavg1", 30)
         mavg2 = MovingAverage("mavg2", 60)
-        client = TestClient(self, expected_msg_count=10000)
+        client = TestClient()
 
         sim.register_components([ret1, ret2, mavg1, mavg2, client])
         sim.register_controller( con )