this is a hotfix to the accidental commit on master, but I lost my bearings again and added pycco, so this is a bit more than a hotfix now.

etc/jenkins.sh

@@ -27,13 +27,17 @@ pip freeze
 
 #documentation output
 paver apidocs html
+pycco ./zipline/*.py -d ./docs/_build/html/pycco/
+pycco ./zipline/finance/*.py -d ./docs/_build/html/pycco/finance
+pycco ./zipline/test/*.py -d ./docs/_build/html/pycco/test
+pycco ./zipline/transforms/*.py -d ./docs/_build/html/pycco/transforms
 
 #run all the tests in test. see setup.cfg for flags.
 nosetests --config=jenkins_setup.cfg 
 
 #run pylint checks
 cp ./pylint.rcfile /mnt/jenkins/.pylintrc #default location for config file...
-pylint -f parseable zipline | tee pylint.out
+pylint -f parseable zipline > pylint.out
 
 #run sloccount analysis
 sloccount --wide --details ./  > sloccount.sc

etc/requirements_dev.txt

@@ -3,7 +3,7 @@ ipython==0.12
 # For debugger
 fancycompleter==0.2
 pyrepl==0.8.2
-Pygments==1.4
+Pygments==1.5
 pdbpp==0.7.2
 
 
@@ -28,4 +28,11 @@ nose==1.1.2
 coverage==3.5.1
 mock==0.7.2
 nosexcover==1.0.7
-pylint==0.25.1
+pylint==0.25.1
+
+# 
+Markdown==2.1.1
+Pycco==0.3.0
+pystache==0.4.0
+smartypants==1.6.0.3
+wsgiref==0.1.2

pylint.rcfile

@@ -115,7 +115,7 @@ no-docstring-rgx=__.*__
 [FORMAT]
 
 # Maximum number of characters on a single line.
-max-line-length=130
+max-line-length=85
 
 # Maximum number of lines in a module
 max-module-lines=1000

zipline/finance/risk.py

@@ -80,33 +80,32 @@ class DailyReturn():
 class RiskMetrics():
     def __init__(self, start_date, end_date, returns, trading_environment):
 
-        self.treasury_curves = trading_environment.treasury_curves
+        self.treasury_curves = trading_environment.treasury_curves        
         self.start_date = start_date
         self.end_date = end_date
         self.trading_environment = trading_environment
         self.algorithm_period_returns, self.algorithm_returns = \
             self.calculate_period_returns(returns)
-
+            
         benchmark_returns = [
-            x for x in self.trading_environment.benchmark_returns
-            if x.date >= returns[0].date and x.date <= returns[-1].date
+                    x for x in self.trading_environment.benchmark_returns
+                    if x.date >= returns[0].date and x.date <= returns[-1].date
         ]
-
+        
         self.benchmark_period_returns, self.benchmark_returns = \
             self.calculate_period_returns(benchmark_returns)
 
         if(len(self.benchmark_returns) != len(self.algorithm_returns)):
             message = "Mismatch between benchmark_returns ({bm_count}) and \
             algorithm_returns ({algo_count}) in range {start} : {end}"
-            message.format(
+            message = message.format(
                 bm_count=len(self.benchmark_returns),
                 algo_count=len(self.algorithm_returns),
                 start=start_date,
                 end=end_date
             )
+            raise Exception(message)
 
-            # TODO: vestigal?
-            #raise Exception(messge)
 
         self.trading_days = len(self.benchmark_returns)
         self.benchmark_volatility = self.calculate_volatility(self.benchmark_returns)

zipline/finance/trading.py

@@ -29,14 +29,6 @@ class TradeSimulationClient(qmsg.Component):
         )
         
         self.perf = perf.PerformanceTracker(self.trading_environment)
-        ##################################################################
-        # TODO: the next line of code need refactoring from RealDiehl
-        # The below sets up the performance object to trigger a full risk
-        # report with rolling periods over the entire test duration. We
-        # would prefer something more explicit than a callback.
-        ##################################################################
-        self.on_done = self.perf.handle_simulation_end
-        
     
     @property
     def get_id(self):
@@ -45,7 +37,7 @@ class TradeSimulationClient(qmsg.Component):
     def set_algorithm(self, algorithm):
         """
         :param algorithm: must implement the algorithm protocol. See 
-        algorithm_protocol.rst.
+        :py:mod:`zipline.test.algorithm`
         """
         self.algorithm = algorithm 
         #register the trading_client's order method with the algorithm
@@ -56,42 +48,71 @@ class TradeSimulationClient(qmsg.Component):
         self.order_socket = self.connect_order()
     
     def do_work(self):
-        #next feed event
+        # poll all the sockets
         socks = dict(self.poll.poll(self.heartbeat_timeout))
 
+        # see if the poller has results for the result_feed
         if self.result_feed in socks and \
             socks[self.result_feed] == self.zmq.POLLIN:   
             
+            # get the next message from the result feed
             msg = self.result_feed.recv()
-
+            
+            # if the feed is done, shut 'er down
             if msg == str(zp.CONTROL_PROTOCOL.DONE):
                 qutil.LOGGER.info("Client is DONE!")
-                self.run_algorithm()
+                # signal the performance tracker that the simulation has
+                # ended. Perf will internally calculate the full risk report.
+                self.perf.handle_simulation_end()
+                # shutdown the feedback loop to the OrderDataSource
                 self.signal_order_done()
+                # signal Simulator, our ComponentHost, that this component is
+                # done and Simulator needn't block exit on this component.
                 self.signal_done()
                 return
             
+            # result_feed is a merge component, so unframe accordingly
             event = zp.MERGE_UNFRAME(msg)
             
-            if(event.TRANSACTION != None):
-                self.txn_count += 1
+            # update performance and relay the event to the algorithm
+            self.process_event(event)
             
-            #filter order flow out of the events sent to callbacks
-            if event.source_id != zp.FINANCE_COMPONENT.ORDER_SOURCE:
-                #mark the start time for client's processing of this event.
-                event_start = datetime.datetime.utcnow()
-                self.queue_event(event)
-                
-                if event.dt >= self.current_dt:
-                    self.run_algorithm()
-                
-                #update time based on receipt of the order
-                self.last_iteration_dur = datetime.datetime.utcnow() - event_start
-                    
-                self.current_dt = self.current_dt + self.last_iteration_dur
-            
-            #signal done to order source.
+             # signal done to order source.
             self.order_socket.send(str(zp.ORDER_PROTOCOL.BREAK))
+
+    def process_event(self, event):
+        # track the number of transactions, for testing purposes.
+        if(event.TRANSACTION != None):
+            self.txn_count += 1
+        
+        #filter order flow out of the events sent to callbacks
+        if event.source_id != zp.FINANCE_COMPONENT.ORDER_SOURCE:
+            
+            # the performance class needs to process each event, without 
+            # skipping. Algorithm should wait until the performance has been 
+            # updated, so that down stream components can safely assume that
+            # performance is up to date. Note that this is done before we
+            # mark the time for the algorithm's processing, thereby not
+            # running the algo's clock for performance book keeping.
+            self.perf.process_event(event)
+            
+            # mark the start time for client's processing of this event.
+            event_start = datetime.datetime.utcnow()
+            
+            # queue the event.
+            self.queue_event(event)
+            
+            # if the event is later than our current time, run the algo
+            # otherwise, the algorithm has fallen behind the feed 
+            # and processing per event is longer than time between events.
+            if event.dt >= self.current_dt:
+                self.run_algorithm()
+            
+            # tally the time spent on this iteration
+            self.last_iteration_dur = datetime.datetime.utcnow() - event_start
+            # move the algorithm's clock forward to include iteration time
+            self.current_dt = self.current_dt + self.last_iteration_dur
+        
             
     def run_algorithm(self):
         frame = self.get_frame()
@@ -114,14 +135,6 @@ class TradeSimulationClient(qmsg.Component):
         self.order_socket.send(str(zp.ORDER_PROTOCOL.DONE))
         
     def queue_event(self, event):
-        ##################################################################
-        # TODO: the next line of code need refactoring from RealDiehl
-        # the performance class needs to process each event, without skipping
-        # and any callbacks should wait until the performance has been 
-        # updated, so that down stream components can safely assume that
-        # performance is up to date.
-        ##################################################################
-        self.perf.process_event(event)
         if self.event_queue == None:
             self.event_queue = []
         series = event.as_series()
@@ -187,7 +200,7 @@ class OrderDataSource(qmsg.DataSource):
                 [self.order_socket],
                 #allow half the time of a heartbeat for the order
                 #timeout, so we have time to signal we are done.
-                #timeout=self.heartbeat_timeout/2000
+                timeout=self.heartbeat_timeout/2000
             ) 
         
             

zipline/lines.py

@@ -177,7 +177,7 @@ class SimulatedTrading(object):
         """
         :param config: A configuration object that is a dict with::
             - environment - a \
-            :py:class:`zipline.finance.trading.TradeSimulationClient`
+            :py:class:`zipline.finance.trading.TradingEnvironment`
             - allocator - a :py:class:`zipline.simulator.AddressAllocator`
             - sid - an integer, which will be used as the security ID. 
             - order_count - the number of orders the test algo will place,
@@ -279,8 +279,8 @@ class SimulatedTrading(object):
     
     def check_started(self):
         if self.started:
-            raise ZiplineException("TradeSimulation", "You cannot add sources \
-            after the simulation has begun.")
+            raise ZiplineException("TradeSimulation", "You cannot add \
+            components after the simulation has begun.")
     
     def get_cumulative_performance(self):
         return self.trading_client.perf.cumulative_performance.to_dict()

zipline/messaging.py

@@ -371,7 +371,10 @@ class MergedParallelBuffer(ParallelBuffer):
 
     def next(self):
         """Get the next merged message from the feed buffer."""
-        if(not(self.is_full() or self.draining)):
+        if not (self.is_full() or self.draining):
+            return
+        
+        if self.pending_messages() == 0:
             return
 
         #

zipline/sources.py

@@ -21,7 +21,6 @@ class TradeDataSource(zm.DataSource):
         
         event.source_id = self.get_id
         if event.sid in self.filter['SID']:            
-            
             message = zp.DATASOURCE_FRAME(event)
         else:
             message = zp.DATASOURCE_FRAME(None)
@@ -48,7 +47,7 @@ class RandomEquityTrades(TradeDataSource):
         zp.COMPONENT_TYPE.SOURCE
 
     def do_work(self):
-        if(self.incr == self.count):
+        if not self.incr < self.count:
             self.signal_done()
             return
 
@@ -62,7 +61,7 @@ class RandomEquityTrades(TradeDataSource):
             "volume"    : volume,
             "dt"        : self.trade_start + (self.minute * self.incr),
         })
-
+        self.send(event)
         self.incr += 1
         
 
@@ -74,8 +73,8 @@ class SpecificEquityTrades(TradeDataSource):
 
     def __init__(self, source_id, event_list):
         """
-        :event_list: should be a chronologically ordered list of dictionaries
-                     in the following form:
+        :param event_list: should be a chronologically ordered list of 
+                           dictionaries in the following form:
 
                 event = {
                     'sid'    : an integer for security id,
@@ -86,6 +85,7 @@ class SpecificEquityTrades(TradeDataSource):
         """
         zm.DataSource.__init__(self, source_id)
         self.event_list = event_list
+        self.count = 0
 
     def get_type(self):
         zp.COMPONENT_TYPE.SOURCE
@@ -97,5 +97,6 @@ class SpecificEquityTrades(TradeDataSource):
 
         event = self.event_list.pop(0)
         self.send(zp.namedict(event))
+        self.count +=1
 
 

zipline/test/algorithms.py

@@ -57,14 +57,10 @@ class TestAlgorithm():
         
     def handle_frame(self, frame):
         self.frame_count += 1
-        for dt, s in frame.iteritems():     
-            data = {}
-            data.update(s)
-            event = zp.namedict(data)
-            #place an order for 100 shares of sid:133
-            if self.incr < self.count:
-                self.order(self.sid, self.amount)
-                self.incr += 1
+        #place an order for 100 shares of sid:133
+        if self.incr < self.count:
+            self.order(self.sid, self.amount)
+            self.incr += 1
                 
     def get_sid_filter(self):
         return [self.sid]

zipline/test/factory.py

@@ -69,8 +69,6 @@ def get_next_trading_dt(current, interval, trading_calendar):
         next = next + interval
         if trading_calendar.is_trading_day(next):
             break
-        else:
-            next = next + timedelta(days=1)
     
     return next
 
@@ -84,6 +82,7 @@ def create_trade_history(sid, prices, amounts, start_time, interval, trading_cal
         trade = create_trade(sid, price, amount, current)
         trades.append(trade)
 
+    assert len(trades) == len(prices)
     return trades
 
 def create_txn(sid, price, amount, datetime, btrid=None):
@@ -108,17 +107,19 @@ def create_txn_history(sid, priceList, amtList, startTime, interval, trading_cal
 
 
 def create_returns(daycount, start, trading_calendar):
+    """
+    For the given number of calendar (not trading) days return all the trading
+    days between start and start + daycount.
+    """
     test_range = []
     current = start.replace(tzinfo=pytz.utc)
     one_day = timedelta(days = 1)
-    while i < daycount: 
-        current = get_next_trading_dt(
-            current, 
-            one_day, 
-            trading_calendar
-        )
-        r = risk.DailyReturn(current, random.random())
-        test_range.append(r)
+    
+    for day in range(daycount): 
+        current = current + one_day
+        if trading_calendar.is_trading_day(current):
+            r = risk.DailyReturn(current, random.random())
+            test_range.append(r)
         
     return test_range
     
@@ -128,12 +129,11 @@ def create_returns_from_range(start, end, trading_calendar):
     end = end.replace(tzinfo=pytz.utc)
     one_day = timedelta(days = 1)
     test_range = []
-    while current <= end: 
-        current = get_next_trading_dt(current, one_day, trading_calender)
+    while current <= end:
         r = risk.DailyReturn(current, random.random())
         test_range.append(r)
+        current = get_next_trading_dt(current, one_day, trading_calendar)
         
-
     return test_range
     
 def create_returns_from_list(returns, start, trading_calendar):
@@ -141,12 +141,16 @@ def create_returns_from_list(returns, start, trading_calendar):
     one_day = timedelta(days = 1)
     test_range = []
     
-    for return_val in returns: 
+    #sometimes the range starts with a non-trading day.
+    if not trading_calendar.is_trading_day(current):
         current = get_next_trading_dt(current, one_day, trading_calendar)
+    
+    for return_val in returns: 
         r = risk.DailyReturn(current, return_val)
         test_range.append(r)
+        current = get_next_trading_dt(current, one_day, trading_calendar)
         
-    return sorted(test_range, key=lambda(x):x.date)
+    return test_range
 
 def create_daily_trade_source(sids, trade_count, trading_environment):
     """

zipline/test/test_finance.py

@@ -108,6 +108,12 @@ class FinanceTestCase(TestCase):
             "Portfolio should have one position in " + str(SID)
         )
         
+        self.assertEqual(
+            zipline.sources['flat'].count,
+            self.zipline_test_config['trade_count'],
+            "The simulated trade source should send all trades."
+        )
+        
         self.assertEqual(
             zipline.algorithm.frame_count,
             self.zipline_test_config['trade_count'],

zipline/test/test_perf_tracking.py

@@ -5,6 +5,7 @@ import datetime
 import pytz
 
 import zipline.test.factory as factory
+import zipline.test.algorithms
 import zipline.util as qutil
 import zipline.finance.performance as perf
 import zipline.finance.risk as risk
@@ -541,6 +542,10 @@ shares in position"
         self.trading_environment.frame_index = ['sid', 'volume', 'dt', \
         'price', 'changed']
         client = TradeSimulationClient(self.trading_environment)
+        # the client expects an algorithm that fullfills the algorithm
+        # protocol, so we use the noop algorithm.
+        test_algo = zipline.test.algorithms.NoopAlgorithm()
+        client.set_algorithm(test_algo)
         
         for event in trade_history:
             #create a transaction for all but
@@ -556,7 +561,7 @@ shares in position"
             else:
                 txn = None
             event[zp.TRANSFORM_TYPE.TRANSACTION] = txn  
-            client.queue_event(event)
+            client.process_event(event)
             
         df = client.get_frame()
         

zipline/test/test_risk.py

@@ -13,7 +13,13 @@ class Risk(unittest.TestCase):
     
     def setUp(self):
         qutil.configure_logging()
-        start_date = datetime.datetime(year=2006, month=1, day=1, tzinfo=pytz.utc)
+        start_date = datetime.datetime(
+            year=2006, 
+            month=1, 
+            day=1, 
+            hour=0,
+            minute=0,
+            tzinfo=pytz.utc)
         end_date = datetime.datetime(year=2006, month=12, day=31, tzinfo=pytz.utc)
         
         self.benchmark_returns, self.treasury_curves = \
@@ -29,9 +35,16 @@ class Risk(unittest.TestCase):
         self.tradingday = datetime.timedelta(hours=6, minutes=30)
         self.dt = datetime.datetime.utcnow()
         
-        self.algo_returns_06 = factory.create_returns_from_list(RETURNS, start_date, self.trading_env)
+        self.algo_returns_06 = factory.create_returns_from_list(
+            RETURNS, 
+            start_date, 
+            self.trading_env
+        )
         
-        self.metrics_06 = risk.RiskReport(self.algo_returns_06, self.trading_env)
+        self.metrics_06 = risk.RiskReport(
+            self.algo_returns_06, 
+            self.trading_env
+        )
         
     def tearDown(self):
         return
@@ -204,10 +217,38 @@ class Risk(unittest.TestCase):
         self.check_metrics(metrics, total_months, start_date)
         
     def check_metrics(self, metrics, total_months, start_date):
-        self.assert_range_length(metrics.month_periods, total_months, 1, start_date)
-        self.assert_range_length(metrics.three_month_periods, total_months, 3, start_date)
-        self.assert_range_length(metrics.six_month_periods, total_months, 6, start_date)
-        self.assert_range_length(metrics.year_periods, total_months, 12, start_date)        
+        """
+        confirm that the right number of riskmetrics were calculated for each
+        window length.
+        """
+        self.assert_range_length(
+            metrics.month_periods, 
+            total_months, 
+            1, 
+            start_date
+        )
+        
+        self.assert_range_length(
+            metrics.three_month_periods, 
+            total_months, 
+            3, 
+            start_date
+        )
+        
+        self.assert_range_length(
+            metrics.six_month_periods, 
+            total_months, 
+            6, 
+            start_date
+        )
+        
+        self.assert_range_length(
+            metrics.year_periods, 
+            total_months, 
+            12, 
+            start_date
+        )        
+        
     def assert_last_day(self, period_end):
         #30 days has september, april, june and november
         if(period_end.month in [9,4,6,11]):
@@ -233,13 +274,16 @@ class Risk(unittest.TestCase):
         if(period_length > total_months):
             self.assertEqual(len(col), 0)
         else:
-            self.assertEqual(len(col), total_months - (period_length - 1), "mismatch for total months - expected:{total_months}/actual:{actual}, period:{period_length}, start:{start_date}, calculated end:{end}".format(
-                                                                            total_months=total_months,
-                                                                            period_length=period_length,
-                                                                            start_date=start_date,
-                                                                            end=col[-1].end_date,
-                                                                            actual=len(col)
-                                                                            ))
+            self.assertEqual(
+                len(col), 
+                total_months - (period_length - 1), 
+                "mismatch for total months - expected:{total_months}/actual:{actual}, period:{period_length}, start:{start_date}, calculated end:{end}".format(
+                        total_months=total_months,
+                        period_length=period_length,
+                        start_date=start_date,
+                        end=col[-1].end_date,
+                        actual=len(col)
+                    ))
             self.assert_month(start_date.month, col[-1].end_date.month)
             self.assert_last_day(col[-1].end_date)