From ce6a91adb6dec9d30baa88fe79d8e8230e2868e6 Mon Sep 17 00:00:00 2001 From: fawce Date: Thu, 29 Mar 2012 17:13:28 -0400 Subject: [PATCH] added a frame to get performance data written to the result stream. not complete, but functional. --- zipline/finance/performance.py | 30 ++++++++------ zipline/finance/risk.py | 5 ++- zipline/protocol.py | 72 ++++++++++++++++++++++++++++++++++ 3 files changed, 93 insertions(+), 14 deletions(-) diff --git a/zipline/finance/performance.py b/zipline/finance/performance.py index ac459697..90b84cbc 100644 --- a/zipline/finance/performance.py +++ b/zipline/finance/performance.py @@ -272,10 +272,10 @@ class PerformanceTracker(): # Output Results if self.result_stream: - # TODO: proper framing - self.result_stream.send_pyobj(self.to_dict()) + msg = zp.PERF_FRAME(self.to_dict()) + self.result_stream.send(msg) - #roll over positions to current day. + # Roll over positions to current day. self.todays_performance.calculate_performance() self.todays_performance = PerformancePeriod( self.todays_performance.positions, @@ -284,18 +284,22 @@ class PerformanceTracker(): ) def handle_simulation_end(self): + """ + When the simulation is complete, run the full period risk report + and send it out on the result_stream. + """ + self.risk_report = risk.RiskReport( + self.returns, + self.trading_environment + ) - #self.risk_report = risk.RiskReport( - #self.returns, - #self.trading_environment - #) - - # Output Results - #if self.result_stream: - ## TODO: proper framing - #self.result_stream.send_pyobj(self.risk_report.to_dict()) if self.result_stream: - self.result_stream.send_pyobj(None) + qutil.LOGGER.info("about to stream the risk report...") + report = self.risk_report.to_dict() + msg = zp.RISK_FRAME(report) + self.result_stream.send(msg) + # this signals that the simulation is complete. + self.result_stream.send("DONE") def round_to_nearest(self, x, base=5): return int(base * round(float(x)/base)) diff --git a/zipline/finance/risk.py b/zipline/finance/risk.py index 8d4e1417..4a335990 100644 --- a/zipline/finance/risk.py +++ b/zipline/finance/risk.py @@ -187,6 +187,9 @@ class RiskMetrics(): """ http://en.wikipedia.org/wiki/Sharpe_ratio """ + if self.algorithm_volatility == 0: + return None + return ( (self.algorithm_period_returns - self.treasury_period_return) / self.algorithm_volatility ) @@ -338,7 +341,7 @@ class RiskReport(): """ return { '1_month' : [x.to_dict() for x in self.month_periods], - '3_month' : [x.to_dict() for x in self.three_year_periods], + '3_month' : [x.to_dict() for x in self.three_month_periods], '6_month' : [x.to_dict() for x in self.six_month_periods], '12_month' : [x.to_dict() for x in self.month_periods] } diff --git a/zipline/protocol.py b/zipline/protocol.py index fb5cb5e8..23eeb39a 100644 --- a/zipline/protocol.py +++ b/zipline/protocol.py @@ -118,6 +118,8 @@ import msgpack import numbers import datetime import pytz +import numpy +import time import copy from collections import namedtuple @@ -599,11 +601,79 @@ def ORDER_SOURCE_UNFRAME(msg): raise INVALID_ORDER_FRAME(msg) except ValueError: raise INVALID_ORDER_FRAME(msg) + +# ----------------------- +# Performance and Risk +# ----------------------- + +def PERF_FRAME(perf): + """ + Frame the performance update created at the end of each simulated trading + day. The msgpack is a tuple with the first element statically set to 'PERF'. + Frames prepared by this method are sent via the same socket as + Frames prepared by RISK_FRAME. So, both methods prefix the payload with + a shorthand for their type. That way, all messages received from the socket + can be PERF_UNFRAMED(), whether they are risk or perf. + + :param perf: the dictionary created by zipline.trade_client.perf + :rvalue: a msgpack string + """ + #pull some special fields from the perf for easy access + date = perf['last_close'] + tp = perf['todays_perf'] + risk = perf['cumulative_risk_metrics'] + + #create the daily nested message + daily_perf = dict( + date=EPOCH(date), + returns=perf['returns'][-1]['returns'], + #TODO: add daily PnL in dollars + pnl=0.0, + portfolio_value=tp['ending_value'] + ) + + cumulative_perf = dict( + alpha=risk['alpha'], + beta=risk['beta'], + sharpe=risk['sharpe'], + #TODO: add total returns to the message from perf + total_returns=0.0, + volatility=risk['algo_volatility'], + benchmark_volatility=risk['benchmark_volatility'], + #TODO: add total bm returns to the message from perf + benchmark_returns=0, + max_drawdown=risk['max_drawdown'], + #TODO: add daily PnL in dollars + pnl=0.0 + ) + + result = {} + #TODO: perf needs to track start time of the bt + result['started_at'] = 0 + result['daily'] = [daily_perf] + result['percent_complete'] = perf['progress'] + result['cumulative'] = cumulative_perf + #TODO: pass the cursor value in. + result['cursor'] = 0 + + return msgpack.dumps(tuple(['PERF', result])) + + +def RISK_FRAME(risk): + return msgpack.dumps(tuple(['RISK', risk])) + + +def PERF_UNFRAME(msg): + prefix, payload = msgpack.loads(msg) + return dict(prefix=prefix, payload=payload) # ----------------------- # Date Helpers # ----------------------- +def EPOCH(some_date): + return time.mktime(some_date.timetuple()) + def PACK_DATE(event): """ Packs the datetime property of event into msgpack'able longs. @@ -676,3 +746,5 @@ FINANCE_COMPONENT = namedict({ 'ORDER_SOURCE' : 'ORDER_SOURCE', 'TRANSACTION_SIM' : 'TRANSACTION_SIM' }) + +