diff --git a/etc/jenkins.sh b/etc/jenkins.sh index 5b304092..96d4745f 100755 --- a/etc/jenkins.sh +++ b/etc/jenkins.sh @@ -29,7 +29,7 @@ pip freeze paver apidocs html #run all the tests in test. see setup.cfg for flags. -nosetests +nosetests --config=jenkins_setup.cfg #run pylint checks cp ./pylint.rcfile /mnt/jenkins/.pylintrc #default location for config file... diff --git a/jenkins_setup.cfg b/jenkins_setup.cfg new file mode 100644 index 00000000..a6f24289 --- /dev/null +++ b/jenkins_setup.cfg @@ -0,0 +1,12 @@ +[nosetests] +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 + + diff --git a/notebooks/Experimenting with Frames.ipynb b/notebooks/Experimenting with Frames.ipynb new file mode 100644 index 00000000..50e5b5a5 --- /dev/null +++ b/notebooks/Experimenting with Frames.ipynb @@ -0,0 +1,352 @@ +{ + "metadata": { + "name": "Experimenting with Frames" + }, + "nbformat": 3, + "worksheets": [ + { + "cells": [ + { + "cell_type": "heading", + "source": [ + "Performance Tracking" + ] + }, + { + "cell_type": "code", + "collapsed": false, + "input": [ + "import datetime", + "import pandas", + "import pytz", + "", + "import zipline.test.factory as factory", + "import zipline.finance.performance as perf", + "import zipline.protocol as zp", + "import zipline.finance.risk as risk", + "import zipline.finance.trading as trading" + ], + "language": "python", + "outputs": [], + "prompt_number": 38 + }, + { + "cell_type": "heading", + "source": [ + "Create a simulated trade history using the test factory" + ] + }, + { + "cell_type": "markdown", + "source": [ + "For any backtesting, zipline relies on a TradingEnvironment object. Trading environment holds essential facts: ", + " ", + " - start and end times for the simulation.", + " - historical daily returns for your benchmark.", + " - historical treasury curves", + " - an assumed capital base for your portfolio", + " - a calendar of trading days based on your benchmark", + "", + "zipline ships with a compressed archives of the S&P daily returns, and US treasury curves to facilitate standalone development and testing. In the next cell we instantiate the environment using these defaults. You can see more of this in zipline/test/test_perf_tracking.py" + ] + }, + { + "cell_type": "code", + "collapsed": false, + "input": [ + "benchmark_returns, treasury_curves = factory.load_market_data()", + " ", + "trading_environment = risk.TradingEnvironment(benchmark_returns, treasury_curves)" + ], + "language": "python", + "outputs": [], + "prompt_number": 39 + }, + { + "cell_type": "code", + "collapsed": false, + "input": [ + "trade_count = 100", + "sid = 133", + "price = 10.1 ", + "price_list = [price] * trade_count", + "volume = [100] * trade_count", + "start_date = datetime.datetime.strptime(\"01/01/2011\",\"%m/%d/%Y\")", + "start_date = start_date.replace(tzinfo=pytz.utc)", + "trade_time_increment = datetime.timedelta(days=1)", + "", + "trade_history = factory.create_trade_history( ", + " sid, ", + " price_list, ", + " volume, ", + " start_date, ", + " trade_time_increment, ", + " trading_environment ", + ")", + "", + "sid2 = 134", + "price2 = 12.12", + "price2_list = [price2] * trade_count ", + "trade_history2 = factory.create_trade_history( ", + " sid2, ", + " price2_list, ", + " volume, ", + " start_date, ", + " trade_time_increment, ", + " trading_environment ", + ")", + " ", + "trade_history.extend(trade_history2) ", + "trade_history = sorted(trade_history, key=lambda x: x.dt)" + ], + "language": "python", + "outputs": [], + "prompt_number": 40 + }, + { + "cell_type": "markdown", + "source": [ + "Now that we have a simulated history of trades for two companies and a corresponding trading environment, we can create a dataframe of trades." + ] + }, + { + "cell_type": "code", + "collapsed": false, + "input": [ + "df = pandas.DataFrame(index = ['price', 'volume', 'dt'])", + "for event in trade_history:", + " series = event.as_series()", + " #df.index = df.index.tolist().append(event.sid)", + " #series.name = event.sid", + " df[event.sid] = series" + ], + "language": "python", + "outputs": [], + "prompt_number": 92 + }, + { + "cell_type": "code", + "collapsed": false, + "input": [ + "df" + ], + "language": "python", + "outputs": [ + { + "output_type": "pyout", + "prompt_number": 93, + "text": [ + " 133 134", + "price 10.1 12.12", + "volume 100 100", + "dt 2011-04-08 00:00:00+00:00 2011-04-08 00:00:00+00:00" + ] + } + ], + "prompt_number": 93 + }, + { + "cell_type": "code", + "collapsed": false, + "input": [ + "df_t = df.transpose()", + "df_t" + ], + "language": "python", + "outputs": [ + { + "output_type": "pyout", + "prompt_number": 94, + "text": [ + " price volume dt", + "133 10.1 100 2011-04-08 00:00:00+00:00", + "134 12.12 100 2011-04-08 00:00:00+00:00" + ] + } + ], + "prompt_number": 94 + }, + { + "cell_type": "code", + "collapsed": false, + "input": [ + "df[133]" + ], + "language": "python", + "outputs": [ + { + "output_type": "pyout", + "prompt_number": 56, + "text": [ + "sid 133", + "volume 100", + "dt 2011-04-08 00:00:00+00:00", + "price 10.1", + "changed NaN", + "Name: 133" + ] + } + ], + "prompt_number": 56 + }, + { + "cell_type": "code", + "collapsed": false, + "input": [ + "df_t['price']" + ], + "language": "python", + "outputs": [ + { + "output_type": "pyout", + "prompt_number": 57, + "text": [ + "133 10.1", + "134 12.12", + "Name: price" + ] + } + ], + "prompt_number": 57 + }, + { + "cell_type": "code", + "collapsed": false, + "input": [ + "df_t['price'].max()" + ], + "language": "python", + "outputs": [ + { + "output_type": "pyout", + "prompt_number": 50, + "text": [ + "12.12" + ] + } + ], + "prompt_number": 50 + }, + { + "cell_type": "code", + "collapsed": true, + "input": [ + "last = trade_history[23].dt" + ], + "language": "python", + "outputs": [], + "prompt_number": 51 + }, + { + "cell_type": "code", + "collapsed": false, + "input": [ + "df_t['changed'] = df_t['dt'] > last" + ], + "language": "python", + "outputs": [], + "prompt_number": 53 + }, + { + "cell_type": "code", + "collapsed": false, + "input": [ + "df_t" + ], + "language": "python", + "outputs": [ + { + "output_type": "pyout", + "prompt_number": 54, + "text": [ + " sid volume dt price changed", + "133 133 100 2011-04-08 00:00:00+00:00 10.1 True", + "134 134 100 2011-04-08 00:00:00+00:00 12.12 True" + ] + } + ], + "prompt_number": 54 + }, + { + "cell_type": "code", + "collapsed": false, + "input": [ + "df_t.index" + ], + "language": "python", + "outputs": [ + { + "output_type": "pyout", + "prompt_number": 59, + "text": [ + "Int64Index([133, 134])" + ] + } + ], + "prompt_number": 59 + }, + { + "cell_type": "code", + "collapsed": false, + "input": [ + "df.index" + ], + "language": "python", + "outputs": [ + { + "output_type": "pyout", + "prompt_number": 60, + "text": [ + "Index([sid, volume, dt, price, changed], dtype=object)" + ] + } + ], + "prompt_number": 60 + }, + { + "cell_type": "code", + "collapsed": false, + "input": [ + "df.columns" + ], + "language": "python", + "outputs": [ + { + "output_type": "pyout", + "prompt_number": 61, + "text": [ + "Int64Index([133, 134])" + ] + } + ], + "prompt_number": 61 + }, + { + "cell_type": "code", + "collapsed": false, + "input": [ + "df_t.columns" + ], + "language": "python", + "outputs": [ + { + "output_type": "pyout", + "prompt_number": 62, + "text": [ + "Index([sid, volume, dt, price, changed], dtype=object)" + ] + } + ], + "prompt_number": 62 + }, + { + "cell_type": "code", + "collapsed": true, + "input": [], + "language": "python", + "outputs": [] + } + ] + } + ] +} \ No newline at end of file diff --git a/setup.cfg b/setup.cfg index d14c258b..740ef396 100644 --- a/setup.cfg +++ b/setup.cfg @@ -2,15 +2,6 @@ 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 - - # Drop into debugger on failure #pdb=0 #pdb-failures=0 - diff --git a/zipline/component.py b/zipline/component.py index 4595c192..db9b7bc8 100644 --- a/zipline/component.py +++ b/zipline/component.py @@ -14,7 +14,8 @@ import humanhash from datetime import datetime import zipline.util as qutil -from zipline.protocol import CONTROL_PROTOCOL, COMPONENT_STATE +from zipline.protocol import CONTROL_PROTOCOL, COMPONENT_STATE, \ + COMPONENT_FAILURE, BACKTEST_STATE class Component(object): """ @@ -66,6 +67,7 @@ class Component(object): self.controller = None self.heartbeat_timeout = 2000 self.state_flag = COMPONENT_STATE.OK + self.error_state = COMPONENT_FAILURE.NOFAILURE self.on_done = None self._exception = None @@ -254,8 +256,17 @@ class Component(object): # Internal Maintenance # ---------------------- - def signal_exception(self, exc=None): + def signal_exception(self, exc=None, scope=None): + + if scope == 'algo': + self.error_state = COMPONENT_FAILURE.ALGOEXCEPT + else: + self.error_state = COMPONENT_FAILURE.HOSTEXCEPT + self.state_flag = COMPONENT_STATE.EXCEPTION + # mark the time of failure so we can track the failure + # progogation through the system. + self.stop_tic = time.time() self._exception = exc diff --git a/zipline/finance/performance.py b/zipline/finance/performance.py index f5f675ee..3a80171d 100644 --- a/zipline/finance/performance.py +++ b/zipline/finance/performance.py @@ -1,58 +1,15 @@ -import datetime -import pytz -import math -import pandas +""" -from zmq.core.poll import select +Performance Tracking +==================== -import zipline.messaging as qmsg -import zipline.util as qutil -import zipline.protocol as zp -import zipline.finance.risk as risk - -class PerformanceTracker(): - - def __init__(self, period_start, period_end, capital_base, trading_environment): - 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.txn_count = 0 - self.event_count = 0 - self.cumulative_performance = PerformancePeriod( - {}, - capital_base, - starting_cash = capital_base - ) - - self.todays_performance = PerformancePeriod( - {}, - capital_base, - starting_cash = capital_base - ) - - def to_dict(self): - """ - Creates a dictionary representing the state of this tracker. - Returns a dict object of the form: - +-----------------+----------------------------------------------------+ | key | value | +=================+====================================================+ | period_start | The beginning of the period to be tracked. datetime| | | in pytz.utc timezone. Will always be 0:00 on the | | | date in UTC. The fact that the time may be on the | - | | prior day in the exchange's local time is ignored | + | | prior day in the exchange's local time is ignored | +-----------------+----------------------------------------------------+ | period_end | The end of the period to be tracked. datetime | | | in pytz.utc timezone. Will always be 23:59 on the | @@ -62,7 +19,7 @@ class PerformanceTracker(): | progress | percentage of test completed | +-----------------+----------------------------------------------------+ | cumulative_capti| The net capital used (positive is spent) through | - | al_used | the course of all the events sent to this tracker | + | al_used | the course of all the events sent to this tracker | +-----------------+----------------------------------------------------+ | max_capital_used| The maximum amount of capital deployed through the | | | course of all the events sent to this tracker | @@ -97,168 +54,18 @@ class PerformanceTracker(): | | overkill. | +-----------------+----------------------------------------------------+ | cumulative_risk | A dictionary representing the risk metrics | - | _metrics | calculated based on the positions aggregated | + | _metrics | calculated based on the positions aggregated | | | through all the events delivered to this tracker. | | | For details look at the comments for | | | :py:meth:`zipline.finance.risk.RiskMetrics.to_dict`| +-----------------+----------------------------------------------------+ - - - - """ - returns_list = [x.to_dict() for x in self.returns] - d = { - 'period_start' : self.period_start, - 'period_end' : self.period_end, - 'progress' : self.progress, - 'cumulative_captial_used' : self.cumulative_captial_used, - 'max_capital_used' : self.max_capital_used, - 'last_close' : self.market_close, - 'last_open' : self.market_open, - 'capital_base' : self.capital_base, - 'returns' : returns_list, - 'cumulative_perf' : self.cumulative_perf.to_dict(), - 'todays_perf' : self.todays_perf.to_dict(), - 'cumulative_risk_metrics' : self.cumulative_risk_metrics.to_dict() - } - - def update(self, event_frame): - for dt, event_series in event_frame.iteritems(): - data = {} - data.update(event_series) - event = zp.namedict(data) - self.process_event(event) - - def process_event(self, event): - qutil.LOGGER.debug("series is " + str(event)) - self.event_count += 1 - if(event.dt >= self.market_close): - self.handle_market_close() - - if not pandas.isnull(event.TRANSACTION): - self.txn_count += 1 - 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 - transaction_cost = event.TRANSACTION.price * event.TRANSACTION.amount - self.cumulative_capital_used += transaction_cost - if(math.fabs(self.cumulative_capital_used) > self.max_capital_used): - self.max_capital_used = math.fabs(self.cumulative_capital_used) - - cushioned_capital = 1.1 * self.max_capital_used - self.max_capital_used = self.round_to_nearest( - cushioned_capital, - 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): - #add the return results from today to the list of DailyReturn objects. - todays_date = self.market_close.replace(hour=0, minute=0, second=0) - todays_return_obj = risk.DailyReturn( - todays_date, - self.todays_performance.returns - ) - self.returns.append(todays_return_obj) - - #calculate risk metrics for cumulative performance - self.cumulative_risk_metrics = risk.RiskMetrics( - start_date=self.period_start, - end_date=self.market_close.replace(hour=0, minute=0, second=0), - returns=self.returns, - trading_environment=self.trading_environment - ) - - #move the market day markers forward - 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("Attempt 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 - - #calculate progress of test - self.progress = self.day_count / self.total_days - - #################################################################### - #######TODO: relay the results of self.to_dict() ########### - #################################################################### - - #roll over positions to current day. - self.todays_performance.calculate_performance() - self.todays_performance = PerformancePeriod( - self.todays_performance.positions, - self.todays_performance.ending_value, - self.todays_performance.ending_cash - ) + | timestamp | System time evevent occurs in zipilne | + +-----------------+----------------------------------------------------+ - def handle_simulation_end(self): - self.risk_report = risk.RiskReport( - self.returns, - self.trading_environment - ) - - #################################################################### - #######TODO: relay the results of self.risk_report.to_dict() ####### - #################################################################### - - def round_to_nearest(self, x, base=5): - return int(base * round(float(x)/base)) -class Position(): - - def __init__(self, sid): - self.sid = sid - self.amount = 0 - self.cost_basis = 0.0 ##per share - self.last_sale_price = None - self.last_sale_date = None - - def update(self, txn): - if(self.sid != txn.sid): - raise NameError('updating position with txn for a 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: - prev_cost = self.cost_basis*self.amount - txn_cost = txn.amount*txn.price - total_cost = prev_cost + txn_cost - total_shares = self.amount + txn.amount - self.cost_basis = total_cost/total_shares - self.amount = self.amount + txn.amount - - def currentValue(self): - return self.amount * self.last_sale_price - - - def __repr__(self): - template = "sid: {sid}, amount: {amount}, cost_basis: {cost_basis}, \ - last_sale_price: {last_sale_price}" - return template.format( - sid=self.sid, - amount=self.amount, - cost_basis=self.cost_basis, - last_sale_price=self.last_sale_price - ) - - def to_dict(self): - """ - Creates a dictionary representing the state of this position. - Returns a dict object of the form: +Position Tracking +================= + +-----------------+----------------------------------------------------+ | key | value | +=================+====================================================+ @@ -272,97 +79,338 @@ class Position(): | last_sale_date | datetime of the last trade of the position's | | | security on the exchange | +-----------------+----------------------------------------------------+ - """ - state = { - 'sid':self.sid, - 'amount':self.amount, - 'cost_basis':self.cost_basis, - 'last_sale_price':self.last_sale_price, - 'last_sale_date':self.last_sale_date - } - return state + | timestamp | System time evevent occurs in zipilne | + +-----------------+----------------------------------------------------+ + +Performance Period +================== + + +---------------+------------------------------------------------------+ + | key | value | + +===============+======================================================+ + | ending_value | the total market value of the positions held at the | + | | end of the period | + +---------------+------------------------------------------------------+ + | capital_used | the net capital consumed (positive means spent) by | + | | buying and selling securities in the period | + +---------------+------------------------------------------------------+ + | starting_value| the total market value of the positions held at the | + | | start of the period | + +---------------+------------------------------------------------------+ + | starting_cash | cash on hand at the beginning of the period | + +---------------+------------------------------------------------------+ + | ending_cash | cash on hand at the end of the period | + +---------------+------------------------------------------------------+ + | positions | a list of dicts representing positions, see | + | | :py:meth:`Position.to_dict()` | + | | for details on the contents of the dict | + +---------------+------------------------------------------------------+ + | timestamp | System time evevent occurs in zipilne | + +---------------+------------------------------------------------------+ + +""" +import datetime +import msgpack +import pandas +import math + +import zmq + +import zipline.util as qutil +import zipline.protocol as zp +import zipline.finance.risk as risk + +class PerformanceTracker(): + """ + + Tracks the performance of the zipstream as it is running in + the simulotr, relays this out to the Deluge broker and then + to the client. + + +--------------------+ Result Stream +--------+ + | PerformanceTracker | ----------------> | Deluge | + +--------------------+ +--------+ + + """ + + def __init__(self, trading_environment): + + self.trading_environment = trading_environment + self.trading_day = datetime.timedelta(hours = 6, minutes = 30) + self.calendar_day = datetime.timedelta(hours = 24) + + self.period_start = self.trading_environment.period_start + self.period_end = self.trading_environment.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 = self.trading_environment.capital_base + self.returns = [] + self.txn_count = 0 + self.event_count = 0 + self.result_stream = None + + self.cumulative_performance = PerformancePeriod( + {}, + self.capital_base, + starting_cash = self.capital_base + ) + + self.todays_performance = PerformancePeriod( + {}, + self.capital_base, + starting_cash = self.capital_base + ) + + + def publish_to(self, zmq_socket, context=None): + """ + Publish the performance results asynchronously to a + socket. + """ + ctx = context or zmq.Context.instance() + sock = ctx.socket(zmq.PUSH) + sock.connect(zmq_socket) + + self.result_stream = sock + + def to_dict(self): + """ + Creates a dictionary representing the state of this tracker. + Returns a dict object of the form: + """ + + returns_list = [x.to_dict() for x in self.returns] + + return { + 'period_start' : self.period_start, + 'period_end' : self.period_end, + 'progress' : self.progress, + 'cumulative_captial_used' : self.cumulative_capital_used, + 'max_capital_used' : self.max_capital_used, + 'last_close' : self.market_close, + 'last_open' : self.market_open, + 'capital_base' : self.capital_base, + 'returns' : returns_list, + 'cumulative_perf' : self.cumulative_performance.to_dict(), + 'todays_perf' : self.todays_performance.to_dict(), + 'cumulative_risk_metrics' : self.cumulative_risk_metrics.to_dict(), + 'timestamp' : datetime.datetime.now(), + } + + def process_event(self, event): + self.event_count += 1 + + if(event.dt >= self.market_close): + self.handle_market_close() + + if not pandas.isnull(event.TRANSACTION): + self.txn_count += 1 + 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 + transaction_cost = event.TRANSACTION.price * event.TRANSACTION.amount + self.cumulative_capital_used += transaction_cost + + if math.fabs(self.cumulative_capital_used) > self.max_capital_used: + self.max_capital_used = math.fabs(self.cumulative_capital_used) + + cushioned_capital = 1.1 * self.max_capital_used + self.max_capital_used = self.round_to_nearest( + cushioned_capital, + 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): + #add the return results from today to the list of DailyReturn objects. + todays_date = self.market_close.replace(hour=0, minute=0, second=0) + todays_return_obj = risk.DailyReturn( + todays_date, + self.todays_performance.returns + ) + self.returns.append(todays_return_obj) + + #calculate risk metrics for cumulative performance + self.cumulative_risk_metrics = risk.RiskMetrics( + start_date=self.period_start, + end_date=self.market_close.replace(hour=0, minute=0, second=0), + returns=self.returns, + trading_environment=self.trading_environment + ) + + #move the market day markers forward + 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("Attempt 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 + + #calculate progress of test + self.progress = self.day_count / self.total_days + + # Output Results + if self.result_stream: + # TODO: proper framing + self.result_stream.send_pyobj(self.to_dict()) + + #roll over positions to current day. + self.todays_performance.calculate_performance() + self.todays_performance = PerformancePeriod( + self.todays_performance.positions, + self.todays_performance.ending_value, + self.todays_performance.ending_cash + ) + + def handle_simulation_end(self): + assert False + + 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()) + + self.result_stream.send_pyobj(None) + + def round_to_nearest(self, x, base=5): + return int(base * round(float(x)/base)) + + +class Position(): + + def __init__(self, sid): + self.sid = sid + self.amount = 0 + self.cost_basis = 0.0 ##per share + self.last_sale_price = None + self.last_sale_date = None + + def update(self, txn): + if(self.sid != txn.sid): + raise NameError('updating position with txn for a different sid') + + #we're covering a short or closing a position + if(self.amount + txn.amount == 0): + self.cost_basis = 0.0 + self.amount = 0 + else: + prev_cost = self.cost_basis*self.amount + txn_cost = txn.amount*txn.price + total_cost = prev_cost + txn_cost + total_shares = self.amount + txn.amount + self.cost_basis = total_cost/total_shares + self.amount = self.amount + txn.amount + + def currentValue(self): + return self.amount * self.last_sale_price + + + def __repr__(self): + template = "sid: {sid}, amount: {amount}, cost_basis: {cost_basis}, \ + last_sale_price: {last_sale_price}" + return template.format( + sid=self.sid, + amount=self.amount, + cost_basis=self.cost_basis, + last_sale_price=self.last_sale_price + ) + + def to_dict(self): + """ + Creates a dictionary representing the state of this position. + Returns a dict object of the form: + """ + return { + 'sid' : self.sid, + 'amount' : self.amount, + 'cost_basis' : self.cost_basis, + 'last_sale_price' : self.last_sale_price, + 'last_sale_date' : self.last_sale_date, + 'timestamp' : datetime.datetime.now(), + } + + class PerformancePeriod(): - + def __init__(self, initial_positions, starting_value, starting_cash): self.ending_value = 0.0 self.period_capital_used = 0.0 self.pnl = 0.0 #sid => position object - self.positions = initial_positions + self.positions = initial_positions self.starting_value = starting_value #cash balance at start of period self.starting_cash = starting_cash self.ending_cash = starting_cash - + def calculate_performance(self): self.ending_value = self.calculate_positions_value() - + total_at_start = self.starting_cash + self.starting_value self.ending_cash = self.starting_cash + self.period_capital_used total_at_end = self.ending_cash + self.ending_value - + self.pnl = total_at_end - total_at_start if(total_at_start != 0): self.returns = self.pnl / total_at_start else: self.returns = 0.0 - + def execute_transaction(self, txn): 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): is_trade = event.type == zp.DATASOURCE_TYPE.TRADE if self.positions.has_key(event.sid) and is_trade: - self.positions[event.sid].last_sale_price = event.price + self.positions[event.sid].last_sale_price = event.price self.positions[event.sid].last_sale_date = event.dt - + def to_dict(self): """ Creates a dictionary representing the state of this performance period Returns a dict object of the form: - -+---------------+-----------------------------------------------------------+ -| key | value | -+===============+===========================================================+ -| ending_value | the total market value of the positions held at the | -| | end of the period | -+---------------+-----------------------------------------------------------+ -| capital_used | the net capital consumed (positive means spent) by | -| | buying and selling securities in the period | -+---------------+-----------------------------------------------------------+ -| starting_value| the total market value of the positions held at the | -| | start of the period | -+---------------+-----------------------------------------------------------+ -| starting_cash | cash on hand at the beginning of the period | -+---------------+-----------------------------------------------------------+ -| ending_cash | cash on hand at the end of the period | -+---------------+-----------------------------------------------------------+ -| positions | a list of dicts representing positions, see | -| | :py:meth:`Position.to_dict()` | -| | for details on the contents of the dict | -+---------------+-----------------------------------------------------------+ + """ - d = { - 'ending_value':self.ending_value, - 'capital_used':self.capital_used, - 'starting_value':self.starting_value, - 'starting_cash':self.starting_cash, - 'ending_cash':self.ending_cash + + return { + 'ending_value' : self.ending_value, + 'capital_used' : self.period_capital_used, + 'starting_value' : self.starting_value, + 'starting_cash' : self.starting_cash, + 'ending_cash' : self.ending_cash, + 'positions' : self.positions, + 'timestamp' : datetime.datetime.now(), } - - position_list = [] - for pos in self.positions: - position_list.append(pos.to_dict()) - - d['positions'] = positions_list - return d \ No newline at end of file diff --git a/zipline/finance/risk.py b/zipline/finance/risk.py index cfca36bf..28dc35f0 100644 --- a/zipline/finance/risk.py +++ b/zipline/finance/risk.py @@ -5,7 +5,6 @@ import numpy as np import numpy.linalg as la import zipline.util as qutil import zipline.protocol as zp -from pymongo import ASCENDING, DESCENDING class DailyReturn(): @@ -46,6 +45,7 @@ class RiskMetrics(): ) raise Exception(messge) + self.trading_days = len(self.benchmark_returns) self.benchmark_volatility = self.calculate_volatility(self.benchmark_returns) self.algorithm_volatility = self.calculate_volatility(self.algorithm_returns) @@ -90,10 +90,10 @@ class RiskMetrics(): | | and self.end_date. | +-----------------+----------------------------------------------------+ """ - d = { + return { 'trading_days' : self.trading_days, 'benchmark_volatility' : self.benchmark_volatility, - 'algo_volatility' : self.algo_volatility, + 'algo_volatility' : self.algorithm_volatility, 'treasury_period_return': self.treasury_period_return, 'sharpe' : self.sharpe, 'beta' : self.beta, @@ -101,7 +101,7 @@ class RiskMetrics(): 'excess_return' : self.excess_return, 'max_drawdown' : self.max_drawdown } - + def __repr__(self): statements = [] for metric in [ @@ -137,7 +137,6 @@ class RiskMetrics(): return period_returns, returns def calculate_volatility(self, daily_returns): - #qutil.LOGGER.debug("trading days {td}".format(td=self.trading_days)) return np.std(daily_returns, ddof=1) * math.sqrt(self.trading_days) def calculate_sharpe(self): @@ -326,11 +325,24 @@ def advance_by_months(dt, jump_in_months): class TradingEnvironment(object): - def __init__(self, benchmark_returns, treasury_curves): + def __init__( + self, + benchmark_returns, + treasury_curves, + period_start=None, + period_end=None, + capital_base=None, + frame_index=None + ): + self.trading_days = [] self.trading_day_map = {} self.treasury_curves = treasury_curves self.benchmark_returns = benchmark_returns + self.frame_index = frame_index + self.period_start = period_start + self.period_end = period_end + self.capital_base = capital_base for bm in benchmark_returns: self.trading_days.append(bm.date) self.trading_day_map[bm.date] = bm diff --git a/zipline/finance/trading.py b/zipline/finance/trading.py index 0dd77da4..6c3d1318 100644 --- a/zipline/finance/trading.py +++ b/zipline/finance/trading.py @@ -8,19 +8,27 @@ from zmq.core.poll import select import zipline.messaging as qmsg import zipline.util as qutil import zipline.protocol as zp +import zipline.finance.performance as perf class TradeSimulationClient(qmsg.Component): - def __init__(self, simulation_dt): + def __init__(self, trading_environment): qmsg.Component.__init__(self) - self.received_count = 0 - self.prev_dt = None - self.event_queue = None - self.event_callbacks = [] - self.txn_count = 0 - self.current_dt = simulation_dt - self.last_iteration_duration = datetime.timedelta(seconds=0) - self.event_frame = None + self.received_count = 0 + self.prev_dt = None + self.event_queue = None + self.event_callbacks = [] + self.txn_count = 0 + self.trading_environment = trading_environment + self.current_dt = trading_environment.period_start + self.last_iteration_dur = datetime.timedelta(seconds=0) + + assert self.trading_environment.frame_index != None + self.event_frame = pandas.DataFrame( + index=self.trading_environment.frame_index + ) + + self.perf = perf.PerformanceTracker(self.trading_environment) @property def get_id(self): @@ -67,9 +75,9 @@ class TradeSimulationClient(qmsg.Component): self.run_callbacks() #update time based on receipt of the order - self.last_iteration_duration = datetime.datetime.utcnow() - event_start + self.last_iteration_dur = datetime.datetime.utcnow() - event_start - self.current_dt = self.current_dt + self.last_iteration_duration + self.current_dt = self.current_dt + self.last_iteration_dur #signal done to order source. self.order_socket.send(str(zp.ORDER_PROTOCOL.BREAK)) @@ -95,15 +103,16 @@ class TradeSimulationClient(qmsg.Component): self.order_socket.send(str(zp.ORDER_PROTOCOL.DONE)) def queue_event(self, event): + self.perf.process_event(event) if self.event_queue == None: - self.event_queue = {} + self.event_queue = [] series = event.as_series() - self.event_queue[event.dt] = series + self.event_queue.append(series) def get_frame(self): - frame = pandas.DataFrame(self.event_queue) - self.event_queue = None - return frame + for event in self.event_queue: + self.event_frame[event['sid']] = event + return self.event_frame class OrderDataSource(qmsg.DataSource): """DataSource that relays orders from the client""" diff --git a/zipline/protocol.py b/zipline/protocol.py index 74fbfbba..e60177aa 100644 --- a/zipline/protocol.py +++ b/zipline/protocol.py @@ -119,98 +119,13 @@ import numbers import datetime import pytz import copy -import pandas from collections import namedtuple -import zipline.util as qutil +from protocol_utils import Enum, FrameExceptionFactory, namedict + #import ujson #import ultrajson_numpy -from ctypes import Structure, c_ubyte - -def Enum(*options): - """ - Fast enums are very important when we want really tight zmq - loops. These are probably going to evolve into pure C structs - anyways so might as well get going on that. - """ - class cstruct(Structure): - _fields_ = [(o, c_ubyte) for o in options] - return cstruct(*range(len(options))) - -def FrameExceptionFactory(name): - """ - Exception factory with a closure around the frame class 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): - """ - So that you can use:: - - foo.BAR - -- or -- - foo['BAR'] - - For more complex structs use collections.namedtuple: - """ - - 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 __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): - # !!!!!!!!!!!!!!!!!!!! - # !!!! DANGEROUS !!!!! - # !!!!!!!!!!!!!!!!!!!! - return other != None and self.__dict__ == other.__dict__ - - def has_attr(self, name): - return self.__dict__.has_key(name) - - def as_series(self): - s = pandas.Series(self.__dict__, self.__dict__.keys()) - return s - # ================ # Control Protocol # ================ @@ -279,6 +194,27 @@ COMPONENT_STATE = Enum( 'EXCEPTION' , # 2 ) +# NOFAILURE - Component is either not running or has not failed +# ALGOEXCEPT - Exception thrown in the given algorithm +# HOSTEXCEPT - Exception thrown on our end. +# INTERRUPT - Manually interuptted by user + +COMPONENT_FAILURE = Enum( + 'NOFAILURE' , + 'ALGOEXCEPT' , + 'HOSTEXCEPT' , + 'INTERRUPT' , +) + +BACKTEST_STATE = Enum( + 'IDLE' , + 'QUEUED' , + 'INPROGRESS' , + 'CANCELLED' , # cancelled ( before natural completion ) + 'EXCEPTION' , # failure ( due to unnatural causes ) + 'DONE' , # done ( naturally completed ) +) + # ================== # Datasource Protocol # ================== diff --git a/zipline/protocol_utils.py b/zipline/protocol_utils.py new file mode 100644 index 00000000..ba805b3b --- /dev/null +++ b/zipline/protocol_utils.py @@ -0,0 +1,93 @@ +import copy +import pandas +from ctypes import Structure, c_ubyte + +def Enum(*options): + """ + Fast enums are very important when we want really tight zmq + loops. These are probably going to evolve into pure C structs + anyways so might as well get going on that. + """ + class cstruct(Structure): + _fields_ = [(o, c_ubyte) for o in options] + return cstruct(*range(len(options))) + +def FrameExceptionFactory(name): + """ + Exception factory with a closure around the frame class 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): + """ + + 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 + + For more complex structs use collections.namedtuple: + """ + + 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 __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): + # !!!!!!!!!!!!!!!!!!!! + # !!!! DANGEROUS !!!!! + # !!!!!!!!!!!!!!!!!!!! + return other != None and self.__dict__ == other.__dict__ + + def has_attr(self, name): + return self.__dict__.has_key(name) + + def as_series(self): + s = pandas.Series(self.__dict__) + s.name = self.sid + return s diff --git a/zipline/test/client.py b/zipline/test/client.py index f9ddf443..1ebf22f3 100644 --- a/zipline/test/client.py +++ b/zipline/test/client.py @@ -84,9 +84,8 @@ class TestAlgorithm(): event = zp.namedict(data) #place an order for 100 shares of sid:133 if self.incr < self.count: - if event.source_id != zp.FINANCE_COMPONENT.ORDER_SOURCE: - self.trading_client.order(self.sid, self.amount) - self.incr += 1 + self.trading_client.order(self.sid, self.amount) + self.incr += 1 elif not self.done: self.trading_client.signal_order_done() self.done = True diff --git a/zipline/test/test_finance.py b/zipline/test/test_finance.py index 2542aa2e..b1413800 100644 --- a/zipline/test/test_finance.py +++ b/zipline/test/test_finance.py @@ -207,7 +207,13 @@ class FinanceTestCase(TestCase): set1 = SpecificEquityTrades("flat-133", trade_history) - trading_client = TradeSimulationClient(start_date) + self.trading_environment.period_start = trade_history[0].dt + self.trading_environment.period_end = trade_history[-1].dt + self.trading_environment.capital_base = 10000 + self.trading_environment.frame_index = ['sid', 'volume', 'dt', \ + 'price', 'changed'] + + trading_client = TradeSimulationClient(self.trading_environment) #client will send 10 orders for 100 shares of 133 test_algo = TestAlgorithm(133, 100, 10, trading_client) @@ -280,25 +286,25 @@ class FinanceTestCase(TestCase): volume, start_date, trade_time_increment, - self.trading_environment ) - + self.trading_environment + ) + + + self.trading_environment.period_start = trade_history[0].dt + self.trading_environment.period_end = trade_history[-1].dt + self.trading_environment.capital_base = 10000 + self.trading_environment.frame_index = ['sid', 'volume', 'dt', \ + 'price', 'changed'] + set1 = SpecificEquityTrades("flat-133", trade_history) #client sill send 10 orders for 100 shares of 133 - trading_client = TradeSimulationClient(start_date) + trading_client = TradeSimulationClient(self.trading_environment) test_algo = TestAlgorithm(133, 100, 10, trading_client) order_source = OrderDataSource() transaction_sim = TransactionSimulator() - perf_tracker = perf.PerformanceTracker( - trade_history[0]['dt'], - trade_history[-1]['dt'], - 1000000.0, - self.trading_environment) - #register perf_tracker to receive callbacks from the client. - trading_client.add_event_callback(perf_tracker.update) - sim.register_components([ trading_client, order_source, @@ -339,19 +345,19 @@ class FinanceTestCase(TestCase): self.assertEqual( transaction_sim.txn_count, - perf_tracker.txn_count, + trading_client.perf.txn_count, "The perf tracker should handle the same number of transactions \ as the simulator emits." ) self.assertEqual( - len(perf_tracker.cumulative_performance.positions), + len(trading_client.perf.cumulative_performance.positions), 1, "Portfolio should have one position." ) self.assertEqual( - perf_tracker.cumulative_performance.positions[133].sid, + trading_client.perf.cumulative_performance.positions[133].sid, 133, "Portfolio should have one position in 133." ) diff --git a/zipline/test/test_perf_tracking.py b/zipline/test/test_perf_tracking.py index 218f3113..36a99ae8 100644 --- a/zipline/test/test_perf_tracking.py +++ b/zipline/test/test_perf_tracking.py @@ -506,34 +506,46 @@ shares in position" trade_count = 100 sid = 133 - price = [10.1] * trade_count + price = 10.1 + price_list = [price] * trade_count volume = [100] * trade_count start_date = datetime.datetime.strptime("01/01/2011","%m/%d/%Y") start_date = start_date.replace(tzinfo=pytz.utc) trade_time_increment = datetime.timedelta(days=1) trade_history = factory.create_trade_history( sid, - price, + price_list, volume, start_date, trade_time_increment, self.trading_environment ) - trade_client = TradeSimulationClient(start_date) - start = trade_history[0].dt - end = trade_history[-1].dt - tracker = perf.PerformanceTracker( - start, - end, - 1000.0, - self.trading_environment + sid2 = 134 + price2 = 12.12 + price2_list = [price2] * trade_count + trade_history2 = factory.create_trade_history( + sid2, + price2_list, + volume, + start_date, + trade_time_increment, + self.trading_environment ) + trade_history.extend(trade_history2) + + self.trading_environment.period_start = trade_history[0].dt + self.trading_environment.period_end = trade_history[-1].dt + self.trading_environment.capital_base = 1000.0 + self.trading_environment.frame_index = ['sid', 'volume', 'dt', \ + 'price', 'changed'] + client = TradeSimulationClient(self.trading_environment) + for event in trade_history: #create a transaction for all but - #one trade, to simulate None transaction - if(event.dt != start): + #first trade in each sid, to simulate None transaction + if(event.dt != self.trading_environment.period_start): txn = zp.namedict({ 'sid' : event.sid, 'amount' : -25, @@ -543,17 +555,19 @@ shares in position" }) else: txn = None - event[zp.TRANSFORM_TYPE.TRANSACTION] = txn - trade_client.queue_event(event) + event[zp.TRANSFORM_TYPE.TRANSACTION] = txn + client.queue_event(event) - df = trade_client.get_frame() - tracker.update(df) + df = client.get_frame() - #we skip one trade, to test case of None transaction - txn_count = len(trade_history) - 1 - self.assertEqual(tracker.txn_count, txn_count) + self.assertEqual(df[133]['price'], price) + self.assertEqual(df[134]['price'], price2) - cumulative_pos = tracker.cumulative_performance.positions[sid] - expected_size = txn_count * -25 + #we skip two trades, to test case of None transaction + txn_count = len(trade_history) - 2 + self.assertEqual(client.perf.txn_count, txn_count) + + cumulative_pos = client.perf.cumulative_performance.positions[sid] + expected_size = txn_count / 2 * -25 self.assertEqual(cumulative_pos.amount, expected_size) \ No newline at end of file diff --git a/zipline/test/test_risk.py b/zipline/test/test_risk.py index 0d293b5e..57264ad6 100644 --- a/zipline/test/test_risk.py +++ b/zipline/test/test_risk.py @@ -11,6 +11,9 @@ class Risk(unittest.TestCase): def setUp(self): qutil.configure_logging() + start_date = datetime.datetime(year=2006, month=1, day=1, tzinfo=pytz.utc) + end_date = datetime.datetime(year=2006, month=12, day=31, tzinfo=pytz.utc) + self.benchmark_returns, self.treasury_curves = \ factory.load_market_data() @@ -23,9 +26,9 @@ class Risk(unittest.TestCase): self.oneday = datetime.timedelta(days=1) self.tradingday = datetime.timedelta(hours=6, minutes=30) self.dt = datetime.datetime.utcnow() - start_date = datetime.datetime(year=2006, month=1, day=1, tzinfo=pytz.utc) + self.algo_returns_06 = factory.create_returns_from_list(RETURNS, start_date, self.trading_calendar) - end_date = datetime.datetime(year=2006, month=12, day=31, tzinfo=pytz.utc) + self.metrics_06 = risk.RiskReport(self.algo_returns_06, self.trading_calendar) def tearDown(self): diff --git a/zipline/zmq_utils.py b/zipline/zmq_utils.py new file mode 100644 index 00000000..e7c09047 --- /dev/null +++ b/zipline/zmq_utils.py @@ -0,0 +1,16 @@ +import gevent +from gevent_zeromq import zmq + +def ZmqConsole(sock_typ, socket_addr, sock_conn=None, context=None): + + context = context or zmq.Context.instance() + socket = context.socket(zmq.PULL) + socket.connect('tcp://127.0.0.1:3141') + + def console(): + while True: + msg = socket.recv() + print msg + import pdb; pdb.set_trace() + + return gevent.spawn(console)