added a new client component that calculates positions, performance, and risk.

zipline/finance/performance.py

@@ -0,0 +1,183 @@
+import datetime
+import pytz
+import math
+
+from zmq.core.poll import select
+
+import zipline.messaging as qmsg
+import zipline.util as qutil
+import zipline.protocol as zp
+import zipline.finance.risk as risk
+
+class PortfolioClient(qmsg.Component):
+    
+    def __init__(self, period_start, period_end, capital_base, trading_environment):
+        qmsg.Component.__init__(self)
+        self.trading_day            = datetime.timedelta(hours=6, minutes=30)
+        self.calendar_day           = datetime.timedelta(hours=24)
+        self.period_start           = period_start
+        self.period_end             = period_end
+        self.market_open            = self.period_start 
+        self.market_close           = self.market_open + self.trading_day
+        self.progress               = 0.0
+        self.total_days             = (self.period_end - self.period_start).days
+        self.day_count              = 0
+        self.cumulative_capital_used= 0.0
+        self.max_capital_used       = 0.0
+        self.capital_base           = capital_base
+        self.trading_environment    = trading_environment
+        self.returns                = []
+        self.cumulative_performance = PerformancePeriod(self.period_start, self.period_end, {}, 0, capital_base = capital_base)
+        self.todays_performance     = PerformancePeriod(self.market_open, self.market_close, {}, 0, capital_base = capital_base)
+    
+    @property
+    def get_id(self):
+        return str(zp.FINANCE_COMPONENT.PORTFOLIO_CLIENT)
+    
+    def open(self):
+        self.result_feed = self.connect_result()
+    
+    def do_work(self):
+        #next feed event
+        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("Portfolio Client is DONE!")
+                self.signal_done()
+                return
+            
+            event = zp.MERGE_UNFRAME(msg)
+            
+            if(event.dt >= self.market_close):
+                self.handle_market_close()
+            
+            if event.TRANSACTION != None:
+                self.cumulative_performance.execute_transaction(event.TRANSACTION)
+                self.todays_performance.execute_transaction(event.TRANSACTION)
+                
+                #we're adding a 10% cushion to the capital used, and then rounding to the nearest 5k
+                self.cumulative_capital_used += event.TRANSACTION.price * event.TRANSACTION.amount
+                if(math.fabs(self.cumulative_capital_used) > self.max_capital_used):
+                    self.max_capital_used = math.fabs(self.cumulative_capital_used)
+                self.max_capital_used = self.round_to_nearest(1.1 * self.max_capital_used, base=5000)
+                self.max_leverage = self.max_capital_used/self.capital_base
+            
+            #update last sale    
+            self.cumulative_performance.update_last_sale(event)
+            self.todays_performance.update_last_sale(event)
+            
+            #calculate performance as of last trade
+            self.cumulative_performance.calculate_performance()
+            self.todays_performance.calculate_performance()
+            
+            
+    #
+    def handle_market_close(self):
+        self.market_open = self.market_open + self.calendar_day
+        while not self.trading_environment.is_trading_day(self.market_open):
+            if self.market_open > self.trading_environment.trading_days[-1]:
+                raise Exception("Attempting to backtest beyond available history.")
+            self.market_open = self.market_open + self.calendar_day
+        self.market_close = self.market_open + self.trading_day   
+        self.day_count += 1.0
+        self.progress = self.day_count / self.total_days
+        self.returns.append(risk.daily_return(self.todays_performance.period_end.replace(hour=0, minute=0, second=0), self.todays_performance.returns))
+        self.cur_period_metrics = risk.periodmetrics(start_date=self.period_start, 
+                                                    end_date=self.todays_performance.period_end.replace(hour=0, minute=0, second=0), 
+                                                    returns=self.returns,
+                                                    trading_environment=self.trading_environment)
+        ###############################################
+        #######TODO: report/relay metrics here#########
+        ###############################################
+        #roll over positions to current day.
+        self.todays_performance = PerformancePeriod(self.market_open, 
+                                                    self.market_close, 
+                                                    self.todays_performance.positions, 
+                                                    self.todays_performance.ending_value, 
+                                                    self.capital_base)
+        
+    #
+    def round_to_nearest(self, x, base=5):
+        return int(base * round(float(x)/base))
+
+class Position():
+    sid         = None
+    amount      = None
+    cost_basis   = None
+    last_sale    = None
+    last_date    = None
+    
+    def __init__(self, sid):
+        self.sid = sid
+        self.amount = 0
+        self.cost_basis = 0.0 ##per share
+    
+    def update(self, txn):
+        if(self.sid != txn.sid):
+            raise NameError('attempt to update position with transaction in different sid')
+            #throw exception
+        
+        if(self.amount + txn.amount == 0): #we're covering a short or closing a position
+            self.cost_basis = 0.0
+            self.amount = 0
+        else:
+            self.cost_basis = (self.cost_basis*self.amount + (txn.amount*txn.price))/(self.amount + txn.amount)
+            self.amount = self.amount + txn.amount
+            
+    def currentValue(self):
+        return self.amount * self.last_sale
+        
+        
+    def __repr__(self):
+        return "sid: {sid}, amount: {amount}, cost_basis: {cost_basis}, last_sale: {last_sale}".format(
+        sid=self.sid, amount=self.amount, cost_basis=self.cost_basis, last_sale=self.last_sale)
+        
+class PerformancePeriod():
+    
+    def __init__(self, period_start, period_end, initial_positions, initial_value, capital_base = None):
+        self.ending_value        = 0.0
+        self.period_capital_used  = 0.0
+        self.period_start       = period_start
+        self.period_end         = period_end
+        self.positions          = initial_positions #sid => position object
+        self.starting_value      = initial_value
+        if(capital_base != None):
+            self.capital_base   = capital_base
+        else:
+            self.capital_base   = 0
+            
+    def calculate_performance(self):
+        self.ending_value = self.calculate_positions_value()
+        self.pnl = (self.ending_value - self.starting_value) - self.period_capital_used
+        if(self.capital_base != 0):
+            self.returns = self.pnl / self.capital_base
+        else:
+            self.returns = 0.0
+            
+    def execute_transaction(self, txn):
+        if(txn.dt > self.period_end):
+            raise Exception("transaction dated {dt} attempted for period ending {ending}".
+                            format(dt=txn.dt, ending=self.period_end))
+        if(not self.positions.has_key(txn.sid)):
+            self.positions[txn.sid] = Position(txn.sid)
+        self.positions[txn.sid].update(txn)
+        self.period_capital_used += -1 * txn.price * txn.amount
+        
+
+    def calculate_positions_value(self):
+        mktValue = 0.0
+        for key,pos in self.positions.iteritems():
+            mktValue += pos.currentValue()
+        return mktValue
+                
+    def update_last_sale(self, event):
+        if self.positions.has_key(event.sid):
+            self.positions[event.sid].last_sale = event.price 
+            self.positions[event.sid].last_date = event.dt
+        
+        
+
+    

zipline/finance/risk.py

@@ -17,17 +17,16 @@ class daily_return():
         return str(self.date) + " - " + str(self.returns)
         
 class periodmetrics():
-    def __init__(self, start_date, end_date, returns, benchmark_returns, treasury_curves, trading_calendar):
+    def __init__(self, start_date, end_date, returns, trading_environment):
         """
         :param treasury_curves: {datetime in utc -> {duration label -> interest rate}}
         """
         
-        self.treasury_curves = treasury_curves
         self.start_date = start_date
         self.end_date = end_date
-        self.trading_calendar = trading_calendar
+        self.trading_environment = trading_environment
         self.algorithm_period_returns, self.algorithm_returns = self.calculate_period_returns(returns)
-        self.benchmark_period_returns, self.benchmark_returns = self.calculate_period_returns(benchmark_returns)
+        self.benchmark_period_returns, self.benchmark_returns = self.calculate_period_returns(trading_environment.benchmark_returns)
         if(len(self.benchmark_returns) != len(self.algorithm_returns)):
             raise Exception("Mismatch between benchmark_returns ({bm_count}) and algorithm_returns ({algo_count}) in range {start} : {end}".format(
                                                                                                                      bm_count=len(self.benchmark_returns),
@@ -53,7 +52,7 @@ class periodmetrics():
         return '\n'.join(statements)
         
     def calculate_period_returns(self, daily_returns):
-        returns = [x.returns for x in daily_returns if x.date >= self.start_date and x.date <= self.end_date and self.trading_calendar.is_trading_day(x.date)]
+        returns = [x.returns for x in daily_returns if x.date >= self.start_date and x.date <= self.end_date and self.trading_environment.is_trading_day(x.date)]
         #qutil.LOGGER.debug("using {count} daily returns out of {total}".format(count=len(returns),total=len(daily_returns)))
         period_returns = 1.0
         for r in returns:
@@ -146,9 +145,8 @@ class periodmetrics():
         curve = None
         #in case end date is not a trading day, search for the next market day for an interest rate
         for i in range(7): 
-            if(self.treasury_curves.has_key(self.end_date + i * one_day)):
-                #qutil.LOGGER.info(self.treasury_curves[self.end_date + i * one_day])
-                curve = self.treasury_curves[self.end_date + i * one_day]
+            if(self.trading_environment.treasury_curves.has_key(self.end_date + i * one_day)):
+                curve = self.trading_environment.treasury_curves[self.end_date + i * one_day]
                 break
         
         if curve:
@@ -165,13 +163,14 @@ class periodmetrics():
         
 class riskmetrics():
     
-    def __init__(self, algorithm_returns, benchmark_returns, treasury_curves, trading_calendar):
+    def __init__(self, algorithm_returns, trading_environment):
         """algorithm_returns needs to be a list of daily_return objects sorted in date ascending order"""
         
         self.algorithm_returns = algorithm_returns
-        self.bm_returns = [x for x in benchmark_returns if x.date >= self.algorithm_returns[0].date and x.date <= self.algorithm_returns[-1].date]
-        self.treasury_curves = treasury_curves
-        self.trading_calendar = trading_calendar
+        self.trading_environment = trading_environment
+        self.bm_returns = [x for x in self.trading_environmentself.benchmark_returns if x.date >= self.algorithm_returns[0].date and x.date <= self.algorithm_returns[-1].date]
+        self.treasury_curves = self.trading_environment.treasury_curves
+        
         
         qutil.LOGGER.debug("#### {start} thru {end} with {count} trading_days of {total} possible".format(start=self.algorithm_returns[0].date, 
                                                                                            end=self.algorithm_returns[-1].date,
@@ -179,20 +178,20 @@ class riskmetrics():
                                                                                            total=len(benchmark_returns)))
         
         #calculate month ends
-        self.month_periods          = self.periodsInRange(1, self.algorithm_returns[0].date, self.algorithm_returns[-1].date)
+        self.month_periods          = self.periods_in_range(1, self.algorithm_returns[0].date, self.algorithm_returns[-1].date)
         #calculate 3 month ends
-        self.three_month_periods    = self.periodsInRange(3, self.algorithm_returns[0].date, self.algorithm_returns[-1].date)
+        self.three_month_periods    = self.periods_in_range(3, self.algorithm_returns[0].date, self.algorithm_returns[-1].date)
         #calculate 6 month ends
-        self.six_month_periods      = self.periodsInRange(6, self.algorithm_returns[0].date, self.algorithm_returns[-1].date)
+        self.six_month_periods      = self.periods_in_range(6, self.algorithm_returns[0].date, self.algorithm_returns[-1].date)
         #calculate 1 year ends
-        self.year_periods           = self.periodsInRange(12, self.algorithm_returns[0].date, self.algorithm_returns[-1].date)
+        self.year_periods           = self.periods_in_range(12, self.algorithm_returns[0].date, self.algorithm_returns[-1].date)
         #calculate 3 year ends
-        self.three_year_periods     = self.periodsInRange(36, self.algorithm_returns[0].date, self.algorithm_returns[-1].date)
+        self.three_year_periods     = self.periods_in_range(36, self.algorithm_returns[0].date, self.algorithm_returns[-1].date)
         #calculate 5 year ends
-        self.five_year_periods      = self.periodsInRange(60, self.algorithm_returns[0].date, self.algorithm_returns[-1].date)
+        self.five_year_periods      = self.periods_in_range(60, self.algorithm_returns[0].date, self.algorithm_returns[-1].date)
         
         
-    def periodsInRange(self, months_per, start, end):
+    def periods_in_range(self, months_per, start, end):
         one_day = datetime.timedelta(days = 1)
         ends = []
         cur_start = start.replace(day=1)
@@ -208,7 +207,7 @@ class riskmetrics():
                                                returns=self.algorithm_returns, 
                                                benchmark_returns=self.bm_returns, 
                                                treasury_curves=self.treasury_curves,
-                                               trading_calendar=self.trading_calendar)
+                                               trading_environment=self.trading_environment)
             ends.append(cur_period_metrics)
             cur_start = advance_by_months(cur_start, 1)
             
@@ -236,12 +235,13 @@ def advance_by_months(dt, jump_in_months):
     r = dt.replace(year = dt.year + years, month = month)
     return r
 
-class TradingCalendar(object):
+class TradingEnvironment(object):
 
     def __init__(self, benchmark_returns, treasury_curves):
         self.trading_days = []
         self.trading_day_map = {}
         self.treasury_curves = treasury_curves
+        self.benchmark_returns = benchmark_returns
         for bm in benchmark_returns:
             self.trading_days.append(bm.date)
             self.trading_day_map[bm.date] = bm

zipline/test/factory.py

@@ -7,13 +7,13 @@ import zipline.finance.risk as risk
 import zipline.protocol as zp
 
 def load_market_data():
-    fp_bm = open("./etc/benchmark.msgpack", "rb")
+    fp_bm = open("./zipline/test/benchmark.msgpack", "rb")
     bm_map = msgpack.loads(fp_bm.read())
     bm_returns = []
     for epoch, returns in bm_map.iteritems():
         bm_returns.append(risk.daily_return(date=datetime.datetime.fromtimestamp(epoch).replace(hour=0, minute=0, second=0, tzinfo=pytz.utc), returns=returns))
     bm_returns = sorted(bm_returns, key=lambda(x): x.date) 
-    fp_tr = open("./etc/treasury_curves.msgpack", "rb")
+    fp_tr = open("./zipline/test/treasury_curves.msgpack", "rb")
     tr_map = msgpack.loads(fp_tr.read())
     tr_curves = {}
     for epoch, curve in tr_map.iteritems():

zipline/test/test_finance.py

@@ -8,6 +8,7 @@ import zipline.test.factory as factory
 import zipline.util as qutil
 import zipline.finance.risk as risk
 import zipline.protocol as zp
+import zipline.finance.performance as perf
 
 from zipline.test.client import TestTradingClient
 from zipline.sources import SpecificEquityTrades
@@ -20,13 +21,11 @@ class FinanceTestCase(TestCase):
 
     def setUp(self):
         qutil.configure_logging()
-        self.benchmark_returns, self.treasury_curves = factory.load_market_data()
-        self.trading_calendar = risk.TradingCalendar(self.benchmark_returns, self.treasury_curves)
+        benchmark_returns, treasury_curves = factory.load_market_data()
+        self.trading_env = risk.TradingEnvironment(benchmark_returns, treasury_curves)
         
 
     def test_trade_feed_protocol(self):
-
-        # TODO: Perhaps something more self-documenting for variables names?
         sid = 133
         price = [10.0] * 4
         volume = [100] * 4
@@ -34,7 +33,7 @@ class FinanceTestCase(TestCase):
         start_date = datetime.strptime("02/15/2012","%m/%d/%Y")
         one_day_td = timedelta(days=1)
 
-        trades = factory.create_trade_history(sid, price, volume, start_date, one_day_td, self.trading_calendar)
+        trades = factory.create_trade_history(sid, price, volume, start_date, one_day_td, self.trading_env)
 
         for trade in trades:
             #simulate data source sending frame
@@ -149,7 +148,7 @@ class FinanceTestCase(TestCase):
         start_date = datetime.strptime("02/1/2012","%m/%d/%Y")
         trade_time_increment = timedelta(days=1)
 
-        trade_history = factory.create_trade_history( sid, price, volume, start_date, trade_time_increment, self.trading_calendar )
+        trade_history = factory.create_trade_history( sid, price, volume, start_date, trade_time_increment, self.trading_env )
 
         set1 = SpecificEquityTrades("flat-133", trade_history)
 
@@ -173,3 +172,66 @@ class FinanceTestCase(TestCase):
         self.assertEqual(sim.feed.pending_messages(), 0, \
             "The feed should be drained of all messages, found {n} remaining." \
             .format(n=sim.feed.pending_messages()))
+
+
+    def test_performance(self):
+
+        # verify order -> transaction -> portfolio position.
+        # --------------
+
+        # Allocate sockets for the simulator components
+        allocator = AddressAllocator(8)
+        sockets = allocator.lease(8)
+
+        addresses = {
+            'sync_address'   : sockets[0],
+            'data_address'   : sockets[1],
+            'feed_address'   : sockets[2],
+            'merge_address'  : sockets[3],
+            'result_address' : sockets[4],
+            'order_address'  : sockets[5]
+        }
+
+        con = Controller(
+            sockets[6],
+            sockets[7],
+            logging = qutil.LOGGER
+        )
+
+        sim = Simulator(addresses)
+
+        # Simulation Components
+        # ---------------------
+
+        # TODO: Perhaps something more self-documenting for variables names?
+        sid = 133
+        price = [10.1] * 16
+        volume = [100] * 16
+        start_date = datetime.strptime("02/1/2012","%m/%d/%Y")
+        trade_time_increment = timedelta(days=1)
+
+        trade_history = factory.create_trade_history( sid, price, volume, start_date, trade_time_increment, self.trading_env )
+
+        set1 = SpecificEquityTrades("flat-133", trade_history)
+
+        #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()
+        portfolio_client = perf.PortfolioClient(trade_history[0]['dt'], trade_history[-1]['dt'], 1000000.0, self.trading_env)
+
+        sim.register_components([client, order_source, transaction_sim, set1, portfolio_client])
+        sim.register_controller( con )
+
+        # Simulation
+        # ----------
+        sim_context = sim.simulate()
+        sim_context.join()
+
+
+        # 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()))