From 7a57c27295e7ee7042472c29b71f86a04b378fe1 Mon Sep 17 00:00:00 2001 From: fawce Date: Tue, 20 Mar 2012 23:10:24 -0400 Subject: [PATCH] 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 | 6 +- etc/requirements_dev.txt | 11 +++- pylint.rcfile | 2 +- zipline/finance/risk.py | 15 +++-- zipline/finance/trading.py | 89 +++++++++++++++++------------- zipline/lines.py | 6 +- zipline/messaging.py | 5 +- zipline/sources.py | 11 ++-- zipline/test/algorithms.py | 12 ++-- zipline/test/factory.py | 34 +++++++----- zipline/test/test_finance.py | 6 ++ zipline/test/test_perf_tracking.py | 7 ++- zipline/test/test_risk.py | 72 +++++++++++++++++++----- 13 files changed, 179 insertions(+), 97 deletions(-) diff --git a/etc/jenkins.sh b/etc/jenkins.sh index 96d4745f..489bf7a6 100755 --- a/etc/jenkins.sh +++ b/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 diff --git a/etc/requirements_dev.txt b/etc/requirements_dev.txt index dc3a0691..ea7f2b7f 100644 --- a/etc/requirements_dev.txt +++ b/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 \ No newline at end of file +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 diff --git a/pylint.rcfile b/pylint.rcfile index 55e39e25..892e121e 100644 --- a/pylint.rcfile +++ b/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 diff --git a/zipline/finance/risk.py b/zipline/finance/risk.py index 38e8159e..8d4e1417 100644 --- a/zipline/finance/risk.py +++ b/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) diff --git a/zipline/finance/trading.py b/zipline/finance/trading.py index 35eec0d1..d88ea6f9 100644 --- a/zipline/finance/trading.py +++ b/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 ) diff --git a/zipline/lines.py b/zipline/lines.py index dbef3930..640885d4 100644 --- a/zipline/lines.py +++ b/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() diff --git a/zipline/messaging.py b/zipline/messaging.py index 8f1a394a..12724023 100644 --- a/zipline/messaging.py +++ b/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 # diff --git a/zipline/sources.py b/zipline/sources.py index e1c2a122..e3ec8800 100644 --- a/zipline/sources.py +++ b/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 diff --git a/zipline/test/algorithms.py b/zipline/test/algorithms.py index 37444919..e1e0f57b 100644 --- a/zipline/test/algorithms.py +++ b/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] diff --git a/zipline/test/factory.py b/zipline/test/factory.py index af362fbc..1bfddc07 100644 --- a/zipline/test/factory.py +++ b/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): """ diff --git a/zipline/test/test_finance.py b/zipline/test/test_finance.py index 31b5b511..be658cca 100644 --- a/zipline/test/test_finance.py +++ b/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'], diff --git a/zipline/test/test_perf_tracking.py b/zipline/test/test_perf_tracking.py index 0038d009..568833bc 100644 --- a/zipline/test/test_perf_tracking.py +++ b/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() diff --git a/zipline/test/test_risk.py b/zipline/test/test_risk.py index e42ae205..57576223 100644 --- a/zipline/test/test_risk.py +++ b/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)