intersticial commit

docs/index.rst

@@ -28,17 +28,17 @@ via zeromq.
 DataSources
 --------------------
 
-A DataSource represents a historical event record, which will be played back during simulation. A simulation may have one or more DataSources, which will be combined in DataFeed. Generally, datasources read records from a persistent store (db, csv file, remote service), format the messages for downstream simulation components, and send them to a PUSH socket. See the base class for all datasources :py:class:`~zipline.sources.DataSource`
+A DataSource represents a historical event record, which will be played back during simulation. A simulation may have one or more DataSources, which will be combined in DataFeed. Generally, datasources read records from a persistent store (db, csv file, remote service), format the messages for downstream simulation components, and send them to a PUSH socket. See the base class for all datasources :py:class:`~zipline.messaging.DataSource` and the module holding all datasources :py:mod:`zipline.sources`
 
 DataFeed
 --------------------
 
-All simulations start with a collection of :py:class:`DataSources <zipline.sources.DataSource>`, which need to be fed to an algorithm. Each :py:class:`~zipline.sources.DataSource`can contain events of differing content (trades, quotes, corporate event) and frequency (quarterly, intraday). To simplify the process of managing the data sources, :py:class:`~zipline.core.DataFeed` can receive events from multiple :py:class:`DataSources <zipline.sources.DataSource>` and combine them into a serial chronological stream. 
+All simulations start with a collection of :py:class:`~zipline.messaging.DataSource`, which need to be fed to an algorithm. Each :py:class:`~zipline.sources.DataSource`can contain events of differing content (trades, quotes, corporate event) and frequency (quarterly, intraday). To simplify the process of managing the data sources, :py:class:`~zipline.core.DataFeed` can receive events from multiple :py:class:`DataSources <zipline.sources.DataSource>` and combine them into a serial chronological stream. 
 
 Transforms
 --------------------
 
-Often, an algorithm will require a running calculation on top of a :py:class:`~zipline.sources.DataSource`, or on the consolidated feed. A simple example is a technical indicator or a moving average. Transforms can be described in :py:class:`~zipline.core.Simulator`'s configuration. Subclass :py:class:`~zipline.transforms.core.Transform` to add your own Transform. Transforms must hold their own state between events, and serialize their current values into messages.
+Often, an algorithm will require a running calculation on top of a :py:class:`~zipline.messaging.DataSource`, or on the consolidated feed. A simple example is a technical indicator or a moving average. Transforms can be described in :py:class:`~zipline.core.Simulator`'s configuration. Subclass :py:class:`~zipline.transforms.core.Transform` to add your own Transform. Transforms must hold their own state between events, and serialize their current values into messages.
 
 
 Data Alignment

docs/zipline.finance.rst

@@ -0,0 +1,27 @@
+finance Package
+===============
+
+:mod:`data` Module
+------------------
+
+.. automodule:: zipline.finance.data
+    :members:
+    :undoc-members:
+    :show-inheritance:
+
+:mod:`risk` Module
+------------------
+
+.. automodule:: zipline.finance.risk
+    :members:
+    :undoc-members:
+    :show-inheritance:
+
+:mod:`trading` Module
+---------------------
+
+.. automodule:: zipline.finance.trading
+    :members:
+    :undoc-members:
+    :show-inheritance:
+

docs/zipline.rst

@@ -9,14 +9,6 @@ zipline Package
     :undoc-members:
     :show-inheritance:
 
-:mod:`cli` Module
------------------
-
-.. automodule:: zipline.cli
-    :members:
-    :undoc-members:
-    :show-inheritance:
-
 :mod:`component` Module
 -----------------------
 
@@ -25,6 +17,30 @@ zipline Package
     :undoc-members:
     :show-inheritance:
 
+:mod:`daemon` Module
+--------------------
+
+.. automodule:: zipline.daemon
+    :members:
+    :undoc-members:
+    :show-inheritance:
+
+:mod:`db` Module
+----------------
+
+.. automodule:: zipline.db
+    :members:
+    :undoc-members:
+    :show-inheritance:
+
+:mod:`host_settings` Module
+---------------------------
+
+.. automodule:: zipline.host_settings
+    :members:
+    :undoc-members:
+    :show-inheritance:
+
 :mod:`messaging` Module
 -----------------------
 
@@ -57,14 +73,6 @@ zipline Package
     :undoc-members:
     :show-inheritance:
 
-:mod:`topology` Module
-----------------------
-
-.. automodule:: zipline.topology
-    :members:
-    :undoc-members:
-    :show-inheritance:
-
 :mod:`util` Module
 ------------------
 
@@ -73,19 +81,12 @@ zipline Package
     :undoc-members:
     :show-inheritance:
 
-:mod:`webui` Module
--------------------
-
-.. automodule:: zipline.webui
-    :members:
-    :undoc-members:
-    :show-inheritance:
-
 Subpackages
 -----------
 
 .. toctree::
 
+    zipline.finance
     zipline.test
     zipline.transforms
 

docs/zipline.test.rst

@@ -9,6 +9,38 @@ test Package
     :undoc-members:
     :show-inheritance:
 
+:mod:`dummy` Module
+-------------------
+
+.. automodule:: zipline.test.dummy
+    :members:
+    :undoc-members:
+    :show-inheritance:
+
+:mod:`factory` Module
+---------------------
+
+.. automodule:: zipline.test.factory
+    :members:
+    :undoc-members:
+    :show-inheritance:
+
+:mod:`test_devsimulator` Module
+-------------------------------
+
+.. automodule:: zipline.test.test_devsimulator
+    :members:
+    :undoc-members:
+    :show-inheritance:
+
+:mod:`test_finance` Module
+--------------------------
+
+.. automodule:: zipline.test.test_finance
+    :members:
+    :undoc-members:
+    :show-inheritance:
+
 :mod:`test_messaging` Module
 ----------------------------
 

setup.cfg

@@ -12,6 +12,6 @@ with-xunit=1
 
 
 # Drop into debugger on failure
-#pdb=0
-#pdb-failures=0
+pdb=0
+pdb-failures=0
 

zipline/finance/risk.py

@@ -5,6 +5,7 @@ import numpy as np
 import numpy.linalg as la
 import zipline.util as qutil
 import zipline.db as db
+import zipline.protocol as zp
 from pymongo import ASCENDING, DESCENDING
 
 class daily_return():
@@ -256,12 +257,10 @@ class TradingCalendar(object):
        
 def get_benchmark_data():
     bmQS = db.DbConnection.get()[1].bench_marks.find(
-                                 spec={"symbol" : "GSPC", 
-                                        "date":{"$gte": datetime.datetime.strptime('01/01/1990','%m/%d/%Y').replace(tzinfo = pytz.utc), 
-                                                "$lte": datetime.datetime.strptime('12/31/2010','%m/%d/%Y').replace(tzinfo = pytz.utc)}},
+                                 spec={"symbol" : "GSPC"}, 
                                  sort=[("date",ASCENDING)],
                                  slave_ok=True,
-                                 as_class=qutil.DocWrap)
+                                 as_class=zp.namedict)
     bm_returns = []
     for bm in bmQS:
         bm_r = daily_return(date=bm.date.replace(tzinfo=pytz.utc), returns=bm.returns)

zipline/finance/trading.py

@@ -1,7 +1,11 @@
 import json
 import datetime
+
+from zmq.core.poll import select
+
 import zipline.messaging as qmsg
 import zipline.util as qutil
+import zipline.protocol as zp
 
 class TradeSimulationClient(qmsg.Component):
     
@@ -10,6 +14,7 @@ class TradeSimulationClient(qmsg.Component):
         self.received_count     = 0
         self.prev_dt            = None
     
+    @property
     def get_id(self):
         return "TRADING_CLIENT"
     
@@ -19,7 +24,7 @@ class TradeSimulationClient(qmsg.Component):
     
     def do_work(self):
         #next feed event
-        (rlist, wlist, xlist) = self.poller.selec([self.result_feed],
+        (rlist, wlist, xlist) = select([self.result_feed],
                                                   [],
                                                   [self.result_feed],
                                                   timeout=self.heartbeat_timeout/1000) #select timeout is in sec
@@ -32,7 +37,7 @@ class TradeSimulationClient(qmsg.Component):
             self.signal_done()
             return #leave open orders hanging? client requests for orders?
             
-        event = json.loads(message)
+        event = zp.MERGE_UNFRAME(message)
         self._handle_event(event)
     
     def connect_order(self):
@@ -51,7 +56,7 @@ class TradeSimulationClient(qmsg.Component):
     
     def order(self, sid, volume):
         order = {'sid':sid, 'volume':volume}
-        self.order_feed.send(json.dumps(order))
+        self.order_feed.send(zp.ORDER_FRAME(order))
     
 
 class TradeSimulator(qmsg.BaseTransform):
@@ -81,7 +86,7 @@ class TradeSimulator(qmsg.BaseTransform):
         """
         
         #next feed event
-        (rlist, wlist, xlist) = self.poller.selec([self.feed_socket],
+        (rlist, wlist, xlist) = select([self.feed_socket],
                                                   [],
                                                   [self.feed_socket],
                                                   timeout=self.heartbeat_timeout/1000) #select timeout is in sec
@@ -94,7 +99,7 @@ class TradeSimulator(qmsg.BaseTransform):
             self.signal_done()
             return #leave open orders hanging? client requests for orders?
             
-        event = json.loads(message)
+        event = qp.FEED_UNFRAME(message)
         
         if self.last_iteration_duration != None:
             self.algo_time = self.last_event_time + self.last_iteration_duration
@@ -113,11 +118,11 @@ class TradeSimulator(qmsg.BaseTransform):
         
         #mark the start time for client's processing of this event.
         self.event_start = datetime.datetime.utcnow()
-        self.result_socket.send(json.dumps(cur_state), self.zmq.NOBLOCK)
+        self.result_socket.send(qf.MERGE_FRAME(cur_state), self.zmq.NOBLOCK)
         
             
         while True: #this loop should also poll for portfolio state req/rep
-            (rlist, wlist, xlist) = self.poller.selec([self.order_socket],
+            (rlist, wlist, xlist) = select([self.order_socket],
                                                       [],
                                                       [self.order_socket],
                                                       timeout=self.heartbeat_timeout/1000) #select timeout is in sec
@@ -131,7 +136,7 @@ class TradeSimulator(qmsg.BaseTransform):
             if order_msg == str(CONTROL_PROTOCOL.DONE):
                 break
 
-            order = json.loads(order_msg)
+            order = qp.ORDER_UNFRAME(order_msg)
             self.add_open_order(order)
             
         #end of order processing loop

zipline/messaging.py

@@ -4,6 +4,7 @@ Commonly used messaging components.
 import json
 import uuid
 import datetime
+import zipline.protocol as zp
 import zipline.util as qutil
 from zipline.component import Component
 
@@ -136,6 +137,23 @@ class ComponentHost(Component):
         raise NotImplementedError
 
 
+class SimulatorBase(ComponentHost):
+    """
+    Simulator coordinates the launch and communication of source, feed, transform, and merge components.
+    """
+
+    def __init__(self, addresses, gevent_needed=False):
+        """
+        """
+        ComponentHost.__init__(self, addresses, gevent_needed)
+
+    def simulate(self):
+        self.run()
+
+    def get_id(self):
+        return "Simulator"
+
+
 class ParallelBuffer(Component):
     """
     Connects to N PULL sockets, publishing all messages received to a PUB
@@ -178,8 +196,8 @@ class ParallelBuffer(Component):
                     self.drain()
                     self.signal_done()
             else:
-                event = json.loads(message)
-                self.append(event[u'id'], event)
+                event = zp.DATASOURCE_UNFRAME(message)
+                self.append(event.source_id, event)
                 self.send_next()
 
     def __len__(self):
@@ -253,7 +271,7 @@ class ParallelBuffer(Component):
 
         event = self.next()
         if(event != None):
-            self.feed_socket.send(json.dumps(event), self.zmq.NOBLOCK)
+            self.feed_socket.send(zp.FEED_FRAME(event), self.zmq.NOBLOCK)
             self.sent_count += 1
 
 
@@ -333,13 +351,13 @@ class BaseTransform(Component):
                 self.signal_done()
                 return
 
-            event = json.loads(message)
+            event = qp.FEED_UNFRAME(message)
             cur_state = self.transform(event)
             #TODO: do we want to relay the datetime again? maybe drop this?
             #cur_state['dt'] = event['dt']
             cur_state['id'] = self.state['name']
 
-            self.result_socket.send(json.dumps(cur_state), self.zmq.NOBLOCK)
+            self.result_socket.send(qp.TRANSFORM_FRAME(cur_state), self.zmq.NOBLOCK)
 
     def transform(self, event):
         """
@@ -372,29 +390,31 @@ class DataSource(Component):
     means looping through all records in a store, converting to a dict, and
     calling send(map).
     """
-    def __init__(self, source_id):
+    def __init__(self, source_id, *args):
         Component.__init__(self)
-        self.id        = source_id
+        self.source_id        = source_id
         self.cur_event = None
+        self.init_ds(args)
+        
+    def init_ds(*args):
+        pass
 
     @property
     def get_id(self):
-        return self.id
+        return self.source_id
 
     def open(self):
         #create the data sink. Based on http://zguide.zeromq.org/py:tasksink2
         self.data_socket = self.connect_data()
 
-    def send(self, event):
-        """
-        event is expected to be a dict
-        sets id and type properties in the dict
-        sends to the data_socket.
-        """
-        event['id'] = self.id
-        event['type'] = self.get_type()
-        self.data_socket.send(json.dumps(event))
-
     def get_type(self):
-        raise NotImplemented
+        raise NotImplementedError
+
+    
+    
+        
+
+
+
+
 

zipline/protocol.py

@@ -42,6 +42,10 @@ you have a strong desire to JSON encode ancient Sanskrit
 """
 
 import msgpack
+import numbers
+import datetime
+import pytz
+import zipline.util as qutil
 #import ujson
 #import ultrajson_numpy
 
@@ -81,8 +85,20 @@ class namedict(object):
     For more complex strcuts use collections.namedtuple:
     """
 
-    def __init__(self, dct):
-        self.__dict__.update(dct)
+    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."""
+        if(key == '_id'):
+            self.__dict__['id'] = value
+        else:
+            self.__dict__[key] = value
+    
+    def merge(self, other_nd):
+        assert isinstance(namedict, other_nd)
+        self.__dict__.update(other_nd.__dict__)
 
 # ================
 # Control Protocol
@@ -145,3 +161,228 @@ COMPONENT_STATE = Enum(
     'DONE'      , # 1
     'EXCEPTION' , # 2
 )
+
+# ==================
+# Datasource Protocol
+# ==================
+
+INVALID_DATASOURCE_FRAME = FrameExceptionFactory('ORDER')
+
+def DATASOURCE_FRAME(ds_id, ds_type, payload):
+    """
+    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(ds_id, basestring)
+    assert isinstance(ds_type, basestring)
+    assert isinstance(payload, basestring) 
+    return msgpack.dumps(tuple([ds_id, ds_type, payload]))
+
+def DATASOURCE_UNFRAME(msg):
+    """
+    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 
+            - dt - a datetime object
+    """
+    try:
+        ds_id, ds_type, payload = msgpack.loads(msg)
+        if(ds_type == "TRADE"):
+            result = {'source_id' : ds_id, 'type' : ds_type}
+            result.update(TRADE_UNFRAME(payload))
+            return namedict(result)
+        else:
+            raise INVALID_DATASOURCE_FRAME(msg)
+            
+    except TypeError:
+        raise INVALID_DATASOURCE_FRAME(msg)
+    except ValueError:
+        raise INVALID_DATASOURCE_FRAME(msg)
+        
+# ==================
+# Feed Protocol
+# ==================
+INVALID_FEED_FRAME = FrameExceptionFactory('FEED')
+
+def FEED_FRAME(event):
+    """
+    :event: a nameddict with at least::
+        - source_id 
+        - type
+    """
+    assert isinstance(event, namedict)
+    source_id = event.source_id
+    ds_type = event.type
+    pack_date(event)
+    del(event.__dict__['dt'])
+    payload = event.__dict__
+    return msgpack.dumps(payload)
+    
+def FEED_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)
+        rval = namedict(payload)
+        unpack_date(rval)
+        return namedict(rval)
+    except TypeError:
+        raise INVALID_TRADE_FRAME(msg)
+    except ValueError:
+        raise INVALID_TRADE_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
+    return msgpack.dumps(tuple([name, value]))
+    
+def TRANSFORM_UNFRAME(msg):
+    try:
+        name, value = msgpack.loads(msg)
+        #TODO: anything we can do to assert more about the content of the dict?
+        assert isinstance(name, basestring)
+        assert payload.has_key('value')
+        return namedict({name : value})
+    except TypeError:
+        raise INVALID_TRANSFORM_FRAME(msg)
+    except ValueError:
+        raise INVALID_TRANSFORM_FRAME(msg)
+        
+
+# ==================
+# Merge Protocol
+# ==================
+INVALID_MERGE_FRAME = FrameExceptionFactory('MERGE')
+
+def MERGE_FRAME(event):
+    """
+    :event: a nameddict with at least::
+        - source_id 
+        - type
+    """
+    assert isinstance(event, namedict)
+    source_id = event.source_id
+    ds_type = event.type
+    payload = event.__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)
+        return namedict(payload)
+    except TypeError:
+        raise INVALID_TRADE_FRAME(msg)
+    except ValueError:
+        raise INVALID_TRADE_FRAME(msg)
+
+    
+# ==================
+# Finance Protocol
+# ==================
+
+INVALID_ORDER_FRAME = FrameExceptionFactory('ORDER')
+INVALID_TRADE_FRAME = FrameExceptionFactory('TRADE')
+
+# ==================
+# Trades
+# ==================
+
+def TRADE_FRAME(event):
+    """:event: should be a namedict with::
+            - ds_id     -- the datasource id sending this trade out
+            - sid       -- the security id
+            - price     -- float of the price printed for the trade
+            - volume    -- int for shares in the trade
+            
+    """
+    assert isinstance(event, namedict)
+    assert isinstance(event.sid, int)
+    assert isinstance(event.price, float)
+    assert isinstance(event.volume, int)
+    pack_date(event)
+    payload = msgpack.dumps(tuple([event.sid, event.price, event.volume, event.epoch, event.micros]))
+    return DATASOURCE_FRAME(ds_id, "TRADE", payload)
+    
+def TRADE_UNFRAME(msg):
+    try:
+        sid, price, volume, epoch, micros = msgpack.loads(msg)
+        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, 'dt' : dt, 'epoch' : epoch, 'micros' : micros})
+        unpack_date(rval)
+        return rval
+    except TypeError:
+        raise INVALID_TRADE_FRAME(msg)
+    except ValueError:
+        raise INVALID_TRADE_FRAME(msg)
+
+# =========
+# Orders
+# =========
+
+def ORDER_FRAME(sid, amount):
+    assert isinstance(sid, int)
+    assert isinstance(amount, int) #no partial shares...   
+    return msgpack.dumps(tuple([sid, amount]))
+    
+
+def ORDER_UNFRAME(msg):
+    try:
+        sid, amount = msgpack.loads(msg)
+        assert isinstance(sid, int)
+        assert isinstance(amount, int)
+
+        return sid, amount
+    except TypeError:
+        raise INVALID_ORDER_FRAME(msg)
+    except ValueError:
+        raise INVALID_ORDER_FRAME(msg)
+        
+# =================
+# Date Helpers
+# =================
+
+def pack_date(event):    
+    assert isinstance(event.dt, datetime.datetime)
+    assert event.dt.tzinfo == pytz.utc #utc only please
+    epoch = long(dt.strftime('%s'))
+    event['epoch'] = epoch
+    event['micros'] = event.dt.microsecond
+    del(event.__dict__['dt'])
+    return event
+
+def unpack_date(payload):
+    assert isinstance(payload['epoch'], numbers.Integral)
+    assert isinstance(payload['micros'], numbers.Integral)
+    dt = datetime.datetime.fromtimestamp(payload['epoch'])
+    dt.replace(microsecond = payload['micros'], tzinfo = pytz.utc)
+    del(payload['epoch'])
+    del(payload['micros'])
+    payload['dt'] = dt
+    return payload

zipline/sources.py

@@ -5,13 +5,23 @@ import datetime
 import random
 
 import zipline.util as qutil
-import zipline.messaging as qmsg
+import zipline.messaging as zm
+import zipline.protocol as zp
 
-class RandomEquityTrades(qmsg.DataSource):
+class TradeDataSource(zm.DataSource):
+    
+    def send(event):
+        """ :param dict event: is a trade event with data as per :py:func: `zipline.protocol.TRADE_FRAME`
+            :rtype: None
+        """
+        message = zp.TRADE_FRAME(self.get_id, sid, price, volume, dt)
+        self.data_socket.send(message)
+
+class RandomEquityTrades(TradeDataSource):
     """Generates a random stream of trades for testing."""
     
     def __init__(self, sid, source_id, count):
-        qmsg.DataSource.__init__(self, source_id)
+        zm.DataSource.__init__(self, source_id)
         self.count          = count
         self.incr           = 0
         self.sid            = sid
@@ -29,33 +39,29 @@ class RandomEquityTrades(qmsg.DataSource):
             self.signal_done()
             return
         
-        self.price = self.price + random.uniform(-0.05, 0.05)
-        event = {
-            'sid'    : self.sid,
-            'dt'     : qutil.format_date(self.trade_start + (self.minute * self.incr)),
-            'price'  : self.price,
-            'volume' : random.randrange(100,10000,100)
-        }
-        
-        self.send(event)
-        self.incr += 1
-    
+        self.price = self.price + random.uniform(-0.05, 0.05)        
+        self.send(self.sid, self.price, random.randrange(100,10000,100), qutil.format_date(self.trade_start + (self.minute * self.incr)))
+        self.incr += 1        
+
+    def send(self, sid, price, volume, dt):
+        message = zp.TRADE_FRAME(self.get_id(), sid, price, volume, dt)
+        self.data_socket.send(message)
 
 
-class SpecificEquityTrades(qmsg.DataSource):
+class SpecificEquityTrades(TradeDataSource):
     """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 = {
-                    'sid'    : self.sid,
-                    'dt'     : qutil.format_date(self.trade_start + (self.minute * self.incr)),
-                    'price'  : self.price,
-                    'volume' : random.randrange(100,10000,100)
+                    'sid'    : an integer for security id,
+                    'dt'     : datetime object,
+                    'price'  : float for price,
+                    'volume' : integer for volume
                 }
         """
-        qmsg.DataSource.__init__(self, source_id)
+        zm.DataSource.__init__(self, source_id)
         self.event_list = event_list
 
     def get_type(self):
@@ -67,6 +73,6 @@ class SpecificEquityTrades(qmsg.DataSource):
             return
         
         event = self.event_list.pop(0)
-        self.send(event)
+        self.send(zp.namedict(event))
         
 

zipline/test/client.py

@@ -38,7 +38,7 @@ class TestClient(qmsg.Component):
                 return
 
             self.received_count += 1
-            event = json.loads(msg)
+            event = zp.MERGE_UNFRAME(msg)
             if(self.prev_dt != None):
                 if(not event['dt'] >= self.prev_dt):
                     raise Exception("Message out of order: {date} after {prev}".format(date=event['dt'], prev=prev_dt))
@@ -49,6 +49,8 @@ class TestClient(qmsg.Component):
 
 class TestTradingClient(TradeSimulationClient):
     
+    def __init__(self):
+        TradeSimulationClient.__init__(self)
     
     def handle_events(self, event_queue):
         #place an order for 100 shares of sid:133

zipline/test/dummy.py

@@ -6,7 +6,7 @@ import threading
 import mock
 from collections import defaultdict
 from zipline.monitor import Controller
-from zipline.messaging import ComponentHost
+from zipline.messaging import SimulatorBase
 import zipline.util as qutil
 
 class DummyAllocator(object):
@@ -26,21 +26,6 @@ class DummyAllocator(object):
     def reaquire(self, *conn):
         pass
 
-class SimulatorBase(ComponentHost):
-    """
-    Simulator coordinates the launch and communication of source, feed, transform, and merge components.
-    """
-
-    def __init__(self, addresses, gevent_needed=False):
-        """
-        """
-        ComponentHost.__init__(self, addresses, gevent_needed)
-
-    def simulate(self):
-        self.run()
-
-    def get_id(self):
-        return "Simulator"
 
 class ThreadSimulator(SimulatorBase):
 

zipline/test/factory.py

@@ -1,5 +1,6 @@
 import datetime
 import pytz
+import zipline.util as qutil
 import zipline.finance.risk as risk
 
 def createReturns(daycount, start):
@@ -64,20 +65,15 @@ def create_trade(sid, price, amount, datetime):
     row['dt'] = datetime
     row['price'] = price
     row['volume'] = amount
-    row['exchange_code'] = "fake exchange"
-    db = getTickDB()
-    db.equity.trades.minute.insert(row,safe=True)
-    dw = DocWrap()
-    dw.store = row
-    return dw
+    return row
  
 def create_trade_history(sid, prices, amounts, start_time, interval):
     i = 0
     trades = []
-    current = start_time
+    current = start_time.replace(tzinfo = pytz.utc)
     while i < len(prices):
         if(risk.trading_calendar.is_trading_day(current)):  
-            trades.append(create_trade(sid, priceList[i], amtList[i], current))
+            trades.append(create_trade(sid, prices[i], amounts[i], current))
             current = current + interval
             i += 1
         else:

zipline/test/test_finance.py

@@ -1,46 +1,32 @@
 """Tests for the zipline.finance package"""
+import datetime
 import mock
+import pytz
 import zipline.host_settings
 from unittest2 import TestCase
-from zipline.test.test_devsimulator import ThreadSimulator, DummyAllocator
-from zipline.test.test_messaging import SimulatorTestCase
 import zipline.test.factory as factory
-from zipline.monitor import Controller
-from zipline.messaging import DataSource
 import zipline.util as qutil
 import zipline.db as db
-import zipline.host_settings
+import zipline.finance.risk as risk
 
-class FinanceTestCase(SimulatorTestCase, TestCase):
+from zipline.test.client import TestTradingClient
+from zipline.test.dummy import ThreadPoolExecutorMixin
+from zipline.sources import SpecificEquityTrades
 
-    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,
-        )
-
-    #
+class FinanceTestCase(ThreadPoolExecutorMixin, TestCase):
+    
+    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")
+        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):
 
-        # Base Simuation
+        # Just verify sending and receiving orders.
         # --------------
 
         # Allocate sockets for the simulator components
@@ -64,9 +50,9 @@ class FinanceTestCase(SimulatorTestCase, TestCase):
         set1 = SpecificEquityTrades("flat-133",factory.create_trade_history(133,    
                                                                             [10.0,10.0,10.0,10.0], 
                                                                             [100,100,100,100], 
-                                                                            datetime.datetime.utcnow(), 
+                                                                            datetime.datetime.strptime("02/15/2012","%m/%d/%Y"),
                                                                             datetime.timedelta(days=1)))
-        client = TestTradingClient(self, expected_msg_count=4)
+        client = TestTradingClient()
 
         sim.register_components([set1, client])
         sim.register_controller( con )

zipline/util.py

@@ -49,30 +49,3 @@ def format_date(dt):
         return None
     dt_str = dt.strftime('%Y/%m/%d-%H:%M:%S') + "." + str(dt.microsecond / 1000)
     return dt_str
-
-class DocWrap():
-    def __init__(self, store=None):
-        if(store == None):
-            self.store = {}
-        else:
-            self.store = store.copy()
-        if(self.store.has_key('_id')):
-            self.store['id'] = self.store['_id']
-            del(self.store['_id'])
-        
-    def __setitem__(self,key,value):
-        if(key == '_id'):
-            self.store['id'] = value
-        else:
-            self.store[key] = value
-        
-    def __getitem__(self, key):
-        if self.store.has_key(key):
-            return self.store[key]
-            
-    def __getattr__(self,attrname):
-        if self.store.has_key(attrname):
-            return self.store[attrname]
-        else:
-            raise AttributeError("No attribute named {name}".format(name=attrname))
-