From a2fbb9ee7f1d4ddef34d1434861b063f7f46f449 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Miguel=20S=C3=A1nchez=20de=20Le=C3=B3n=20Peque?= Date: Tue, 23 May 2017 09:39:21 +0200 Subject: [PATCH 1/6] Add missing Python 3.5 references (now supported) Add this version to the Conda build matrix and to the setup.py file. --- etc/conda_build_matrix.py | 2 +- setup.py | 1 + 2 files changed, 2 insertions(+), 1 deletion(-) diff --git a/etc/conda_build_matrix.py b/etc/conda_build_matrix.py index 64dfb34c..5c0a0296 100644 --- a/etc/conda_build_matrix.py +++ b/etc/conda_build_matrix.py @@ -4,7 +4,7 @@ import subprocess import click -py_versions = ('2.7', '3.4') +py_versions = ('2.7', '3.4', '3.5') npy_versions = ('1.9', '1.10') zipline_path = os.path.join( os.path.dirname(__file__), diff --git a/setup.py b/setup.py index 860cd9d1..61f3d673 100644 --- a/setup.py +++ b/setup.py @@ -305,6 +305,7 @@ setup( 'Programming Language :: Python', 'Programming Language :: Python :: 2.7', 'Programming Language :: Python :: 3.4', + 'Programming Language :: Python :: 3.5', 'Operating System :: OS Independent', 'Intended Audience :: Science/Research', 'Topic :: Office/Business :: Financial', From c59518bbeb1f03e99c8185f5b6f2e741b50017b6 Mon Sep 17 00:00:00 2001 From: Ana Ruelas Date: Fri, 19 May 2017 13:59:41 -0400 Subject: [PATCH 2/6] ENH: Add function to concatenate list of dataframes with categoricals STY: Alphabetized import list --- tests/utils/test_date_utils.py | 29 ++++++++ tests/utils/test_pandas_utils.py | 100 +++++++++++++++++++++++++++- zipline/utils/calendars/__init__.py | 12 ++-- zipline/utils/date_utils.py | 21 ++++++ zipline/utils/pandas_utils.py | 43 ++++++++++++ 5 files changed, 198 insertions(+), 7 deletions(-) create mode 100644 tests/utils/test_date_utils.py create mode 100644 zipline/utils/date_utils.py diff --git a/tests/utils/test_date_utils.py b/tests/utils/test_date_utils.py new file mode 100644 index 00000000..ab43e11e --- /dev/null +++ b/tests/utils/test_date_utils.py @@ -0,0 +1,29 @@ +from pandas import Timestamp + +from nose_parameterized import parameterized + +from zipline.testing import ZiplineTestCase +from zipline.utils.calendars import get_calendar +from zipline.utils.date_utils import roll_dates_to_previous_session + + +class TestRollDatesToPreviousSession(ZiplineTestCase): + + @parameterized.expand([ + ( + Timestamp('05-19-2017', tz='UTC'), # actual trading date + Timestamp('05-19-2017', tz='UTC'), + ), + ( + Timestamp('07-04-2015', tz='UTC'), # weekend nyse holiday + Timestamp('07-02-2015', tz='UTC'), + ), + ( + Timestamp('01-16-2017', tz='UTC'), # weeknight nyse holiday + Timestamp('01-13-2017', tz='UTC'), + ), + ]) + def test_roll_dates_to_previous_session(self, date, expected_rolled_date): + calendar = get_calendar('NYSE') + result = roll_dates_to_previous_session(calendar, date) + self.assertEqual(result[0], expected_rolled_date) diff --git a/tests/utils/test_pandas_utils.py b/tests/utils/test_pandas_utils.py index e84ed3fe..870a99eb 100644 --- a/tests/utils/test_pandas_utils.py +++ b/tests/utils/test_pandas_utils.py @@ -4,7 +4,11 @@ Tests for zipline/utils/pandas_utils.py import pandas as pd from zipline.testing import parameter_space, ZiplineTestCase -from zipline.utils.pandas_utils import nearest_unequal_elements +from zipline.testing.predicates import assert_equal +from zipline.utils.pandas_utils import ( + categorical_df_concat, + nearest_unequal_elements +) class TestNearestUnequalElements(ZiplineTestCase): @@ -80,3 +84,97 @@ class TestNearestUnequalElements(ZiplineTestCase): str(e.exception), 'dts must be sorted in increasing order', ) + + +class TestCatDFConcat(ZiplineTestCase): + + def test_categorical_df_concat(self): + + inp = [ + pd.DataFrame( + { + 'A': pd.Series(['a', 'b', 'c'], dtype='category'), + 'B': pd.Series([100, 102, 103], dtype='int64'), + 'C': pd.Series(['x', 'x', 'x'], dtype='category'), + } + ), + pd.DataFrame( + { + 'A': pd.Series(['c', 'b', 'd'], dtype='category'), + 'B': pd.Series([103, 102, 104], dtype='int64'), + 'C': pd.Series(['y', 'y', 'y'], dtype='category'), + } + ), + pd.DataFrame( + { + 'A': pd.Series(['a', 'b', 'd'], dtype='category'), + 'B': pd.Series([101, 102, 104], dtype='int64'), + 'C': pd.Series(['z', 'z', 'z'], dtype='category'), + } + ), + ] + result = categorical_df_concat(inp) + + expected = pd.DataFrame( + { + 'A': pd.Series( + ['a', 'b', 'c', 'c', 'b', 'd', 'a', 'b', 'd'], + dtype='category' + ), + 'B': pd.Series( + [100, 102, 103, 103, 102, 104, 101, 102, 104], + dtype='int64' + ), + 'C': pd.Series( + ['x', 'x', 'x', 'y', 'y', 'y', 'z', 'z', 'z'], + dtype='category' + ), + }, + ) + expected.index = pd.Int64Index([0, 1, 2, 0, 1, 2, 0, 1, 2]) + assert_equal(expected, result) + assert_equal( + expected['A'].cat.categories, + result['A'].cat.categories + ) + assert_equal( + expected['C'].cat.categories, + result['C'].cat.categories + ) + + def test_categorical_df_concat_value_error(self): + + mismatched_dtypes = [ + pd.DataFrame( + { + 'A': pd.Series(['a', 'b', 'c'], dtype='category'), + 'B': pd.Series([100, 102, 103], dtype='int64'), + } + ), + pd.DataFrame( + { + 'A': pd.Series(['c', 'b', 'd'], dtype='category'), + 'B': pd.Series([103, 102, 104], dtype='float64'), + } + ), + ] + mismatched_column_names = [ + pd.DataFrame( + { + 'A': pd.Series(['a', 'b', 'c'], dtype='category'), + 'B': pd.Series([100, 102, 103], dtype='int64'), + } + ), + pd.DataFrame( + { + 'A': pd.Series(['c', 'b', 'd'], dtype='category'), + 'X': pd.Series([103, 102, 104], dtype='int64'), + } + ), + ] + + with self.assertRaises(ValueError): + categorical_df_concat(mismatched_dtypes) + + with self.assertRaises(ValueError): + categorical_df_concat(mismatched_column_names) diff --git a/zipline/utils/calendars/__init__.py b/zipline/utils/calendars/__init__.py index d3f3ecf0..61c22e6b 100644 --- a/zipline/utils/calendars/__init__.py +++ b/zipline/utils/calendars/__init__.py @@ -15,20 +15,20 @@ from .trading_calendar import TradingCalendar from .calendar_utils import ( - get_calendar, - register_calendar_alias, - register_calendar, - register_calendar_type, + clear_calendars, deregister_calendar, - clear_calendars + get_calendar, + register_calendar, + register_calendar_alias, + register_calendar_type, ) __all__ = [ - 'TradingCalendar', 'clear_calendars', 'deregister_calendar', 'get_calendar', 'register_calendar', 'register_calendar_alias', 'register_calendar_type', + 'TradingCalendar', ] diff --git a/zipline/utils/date_utils.py b/zipline/utils/date_utils.py new file mode 100644 index 00000000..07ceeda9 --- /dev/null +++ b/zipline/utils/date_utils.py @@ -0,0 +1,21 @@ +def roll_dates_to_previous_session(calendar, *dates): + """ + Roll ``dates`` to the next session of ``calendar``. + + Parameters + ---------- + calendar : zipline.utils.calendars.trading_calendar.TradingCalendar + The calendar to use as a reference. + *dates : pd.Timestamp + The dates for which the last trading date is needed. + + Returns + ------- + rolled_dates: pandas.tseries.index.DatetimeIndex + The last trading date of the input dates, inclusive. + + """ + all_sessions = calendar.all_sessions + + locs = [all_sessions.get_loc(dt, method='ffill') for dt in dates] + return all_sessions[locs] diff --git a/zipline/utils/pandas_utils.py b/zipline/utils/pandas_utils.py index ccac273a..39e4040b 100644 --- a/zipline/utils/pandas_utils.py +++ b/zipline/utils/pandas_utils.py @@ -2,6 +2,7 @@ Utilities for working with pandas objects. """ from contextlib import contextmanager +from copy import deepcopy from itertools import product import operator as op import warnings @@ -222,3 +223,45 @@ def clear_dataframe_indexer_caches(df): delattr(df, attr) except AttributeError: pass + + +def categorical_df_concat(df_list, inplace=False): + """ + Prepare list of pandas DataFrames to be used as input to pd.concat. + Ensure any columns of type 'category' have the same categories across each + dataframe. + + Parameters + ---------- + df_list : list + List of dataframes with same columns. + inplace : bool + True if input list can be modified. Default is False. + + Returns + ------- + concatenated : df + Dataframe of concatenated list. + """ + + if not inplace: + df_list = deepcopy(df_list) + + # Assert each dataframe has the same columns/dtypes + df = df_list[0] + if not all([(df.dtypes.equals(df_i.dtypes)) for df_i in df_list[1:]]): + raise ValueError("Input DataFrames must have the same columns/dtypes.") + + categorical_columns = df.columns[df.dtypes == 'category'] + + for col in categorical_columns: + new_categories = sorted( + set().union( + *(frame[col].cat.categories for frame in df_list) + ) + ) + + for df in df_list: + df[col].cat.set_categories(new_categories, inplace=True) + + return pd.concat(df_list) From 69b632db3774a9aa927818577134caf5c9898cf0 Mon Sep 17 00:00:00 2001 From: Ana Ruelas Date: Tue, 23 May 2017 16:56:05 -0400 Subject: [PATCH 3/6] ENH: Add function for running pipelines in chunks --- tests/pipeline/test_run_chunked_pipeline.py | 37 +++++++++++++ tests/utils/test_date_utils.py | 1 - zipline/pipeline/__init__.py | 37 +++++++++++++ zipline/utils/date_utils.py | 58 ++++++++++++++++++--- zipline/utils/pandas_utils.py | 2 +- 5 files changed, 126 insertions(+), 9 deletions(-) create mode 100644 tests/pipeline/test_run_chunked_pipeline.py diff --git a/tests/pipeline/test_run_chunked_pipeline.py b/tests/pipeline/test_run_chunked_pipeline.py new file mode 100644 index 00000000..533ccf06 --- /dev/null +++ b/tests/pipeline/test_run_chunked_pipeline.py @@ -0,0 +1,37 @@ +from zipline.pipeline import Pipeline, run_chunked_pipeline +from zipline.pipeline.data import USEquityPricing +from zipline.pipeline.factors import Returns +from zipline.testing import ZiplineTestCase +from zipline.testing.fixtures import WithEquityPricingPipelineEngine + + +class ChunkedPipelineTestCase(WithEquityPricingPipelineEngine, + ZiplineTestCase): + + def test_run_chunked_pipeline(self): + """ + Test that running a pipeline in chunks produces the same result as if + it were run all at once + """ + pipe = Pipeline( + columns={ + 'close': USEquityPricing.close.latest, + 'returns': Returns(window_length=2), + }, + ) + sessions = self.nyse_calendar.all_sessions + start_date = sessions[sessions.get_loc(self.START_DATE) + 2] + + pipeline_result = self.pipeline_engine.run_pipeline( + pipe, + start_date=start_date, + end_date=self.END_DATE, + ) + chunked_result = run_chunked_pipeline( + engine=self.pipeline_engine, + pipeline=pipe, + start_date=start_date, + end_date=self.END_DATE, + chunksize=22 + ) + self.assertTrue(chunked_result.equals(pipeline_result)) diff --git a/tests/utils/test_date_utils.py b/tests/utils/test_date_utils.py index ab43e11e..b9f3cd05 100644 --- a/tests/utils/test_date_utils.py +++ b/tests/utils/test_date_utils.py @@ -1,5 +1,4 @@ from pandas import Timestamp - from nose_parameterized import parameterized from zipline.testing import ZiplineTestCase diff --git a/zipline/pipeline/__init__.py b/zipline/pipeline/__init__.py index a169256b..16c429c2 100644 --- a/zipline/pipeline/__init__.py +++ b/zipline/pipeline/__init__.py @@ -1,5 +1,8 @@ from __future__ import print_function from zipline.assets import AssetFinder +from zipline.utils.calendars import get_calendar +from zipline.utils.date_utils import compute_date_range_chunks +from zipline.utils.pandas_utils import categorical_df_concat from .classifiers import Classifier, CustomClassifier from .engine import SimplePipelineEngine @@ -47,6 +50,39 @@ def engine_from_files(daily_bar_path, ) +def run_chunked_pipeline(engine, pipeline, start_date, end_date, chunksize): + """Run a pipeline to collect the results. + + Parameters + ---------- + engine : Engine + The pipeline engine. + pipeline : Pipeline + The pipeline to run. + start_date : pd.Timestamp + The start date to run the pipeline for. + end_date : pd.Timestamp + The end date to run the pipeline for. + chunksize : int or None + The number of days to execute at a time. If this is None, all the days + will be run at once. + + Returns + ------- + results : pd.DataFrame + The results for each output term in the pipeline. + """ + ranges = compute_date_range_chunks( + get_calendar('NYSE'), + start_date, + end_date, + chunksize, + ) + chunks = [engine.run_pipeline(pipeline, s, e) for s, e in ranges] + + return categorical_df_concat(chunks, inplace=True) + + __all__ = ( 'Classifier', 'CustomFactor', @@ -58,6 +94,7 @@ __all__ = ( 'Filter', 'Pipeline', 'SimplePipelineEngine', + 'run_chunked_pipeline', 'Term', 'TermGraph', ) diff --git a/zipline/utils/date_utils.py b/zipline/utils/date_utils.py index 07ceeda9..c59109a2 100644 --- a/zipline/utils/date_utils.py +++ b/zipline/utils/date_utils.py @@ -1,11 +1,15 @@ -def roll_dates_to_previous_session(calendar, *dates): +from toolz import partition_all + + +def roll_dates_to_previous_session(sessions, *dates): """ - Roll ``dates`` to the next session of ``calendar``. + Roll `dates` to the last session of `calendar`. Return input date if it + is a valid session. Parameters ---------- - calendar : zipline.utils.calendars.trading_calendar.TradingCalendar - The calendar to use as a reference. + sessions : pandas.tseries.index.DatetimeIndex + The list of valid session dates. *dates : pd.Timestamp The dates for which the last trading date is needed. @@ -15,7 +19,47 @@ def roll_dates_to_previous_session(calendar, *dates): The last trading date of the input dates, inclusive. """ - all_sessions = calendar.all_sessions + # Find the previous index value if there is no exact match. + locs = [sessions.get_loc(dt, method='ffill') for dt in dates] + return sessions[locs].tolist() - locs = [all_sessions.get_loc(dt, method='ffill') for dt in dates] - return all_sessions[locs] + +def compute_date_range_chunks(sessions, start_date, end_date, chunksize): + """Compute the start and end dates to run a pipeline for. + + Parameters + ---------- + sessions : DatetimeIndex + The available dates. + start_date : pd.Timestamp + The first date in the pipeline. + end_date : pd.Timestamp + The last date in the pipeline. + chunksize : int or None + The size of the chunks to run. Setting this to None returns one chunk. + + Returns + ------- + ranges : iterable[(np.datetime64, np.datetime64)] + A sequence of start and end dates to run the pipeline for. + """ + if start_date not in sessions: + raise KeyError("Start date %s is not found in calendar." % + (start_date.strftime("%Y-%m-%d"),)) + if end_date not in sessions: + raise KeyError("End date %s is not found in calendar." % + (end_date.strftime("%Y-%m-%d"),)) + if end_date < start_date: + raise ValueError("End date %s cannot precede start date %s." % + (end_date.strftime("%Y-%m-%d"), + start_date.strftime("%Y-%m-%d"))) + + if chunksize is None: + return [(start_date, end_date)] + + start_ix, end_ix = sessions.slice_locs(start_date, end_date) + return ( + (r[0], r[-1]) for r in partition_all( + chunksize, sessions[start_ix:end_ix] + ) + ) diff --git a/zipline/utils/pandas_utils.py b/zipline/utils/pandas_utils.py index 39e4040b..7323e245 100644 --- a/zipline/utils/pandas_utils.py +++ b/zipline/utils/pandas_utils.py @@ -258,7 +258,7 @@ def categorical_df_concat(df_list, inplace=False): new_categories = sorted( set().union( *(frame[col].cat.categories for frame in df_list) - ) + ) - {None} ) for df in df_list: From 2d56d253fa0b14e6f5062cd8f35cc48b6268daf1 Mon Sep 17 00:00:00 2001 From: Ana Ruelas Date: Thu, 1 Jun 2017 12:15:16 -0400 Subject: [PATCH 4/6] ENH: Add run_chunked_pipeline method to PipelineEngine --- tests/pipeline/test_engine.py | 34 ++++++++ tests/pipeline/test_run_chunked_pipeline.py | 37 -------- tests/utils/test_date_utils.py | 94 +++++++++++++++++---- tests/utils/test_pandas_utils.py | 12 ++- zipline/pipeline/__init__.py | 37 -------- zipline/pipeline/engine.py | 67 ++++++++++++++- zipline/testing/fixtures.py | 12 +-- zipline/utils/date_utils.py | 23 ----- 8 files changed, 191 insertions(+), 125 deletions(-) delete mode 100644 tests/pipeline/test_run_chunked_pipeline.py diff --git a/tests/pipeline/test_engine.py b/tests/pipeline/test_engine.py index 568b5442..b1e38920 100644 --- a/tests/pipeline/test_engine.py +++ b/tests/pipeline/test_engine.py @@ -51,6 +51,7 @@ from zipline.pipeline.factors import ( ExponentialWeightedMovingAverage, ExponentialWeightedMovingStdDev, MaxDrawdown, + Returns, SimpleMovingAverage, ) from zipline.pipeline.loaders.equity_pricing_loader import ( @@ -77,6 +78,7 @@ from zipline.testing import ( ) from zipline.testing.fixtures import ( WithAdjustmentReader, + WithEquityPricingPipelineEngine, WithSeededRandomPipelineEngine, WithTradingEnvironment, ZiplineTestCase, @@ -1497,3 +1499,35 @@ class PopulateInitialWorkspaceTestCase(WithConstantInputs, ZiplineTestCase): precomputed_term_value, ), ) + + +class ChunkedPipelineTestCase(WithEquityPricingPipelineEngine, + ZiplineTestCase): + + PIPELINE_START_DATE = Timestamp('2006-01-05', tz='UTC') + END_DATE = Timestamp('2006-12-29', tz='UTC') + + def test_run_chunked_pipeline(self): + """ + Test that running a pipeline in chunks produces the same result as if + it were run all at once + """ + pipe = Pipeline( + columns={ + 'close': USEquityPricing.close.latest, + 'returns': Returns(window_length=2), + 'categorical': USEquityPricing.close.latest.quantiles(5) + }, + ) + pipeline_result = self.pipeline_engine.run_pipeline( + pipe, + start_date=self.PIPELINE_START_DATE, + end_date=self.END_DATE, + ) + chunked_result = self.pipeline_engine.run_chunked_pipeline( + pipeline=pipe, + start_date=self.PIPELINE_START_DATE, + end_date=self.END_DATE, + chunksize=22 + ) + self.assertTrue(chunked_result.equals(pipeline_result)) diff --git a/tests/pipeline/test_run_chunked_pipeline.py b/tests/pipeline/test_run_chunked_pipeline.py deleted file mode 100644 index 533ccf06..00000000 --- a/tests/pipeline/test_run_chunked_pipeline.py +++ /dev/null @@ -1,37 +0,0 @@ -from zipline.pipeline import Pipeline, run_chunked_pipeline -from zipline.pipeline.data import USEquityPricing -from zipline.pipeline.factors import Returns -from zipline.testing import ZiplineTestCase -from zipline.testing.fixtures import WithEquityPricingPipelineEngine - - -class ChunkedPipelineTestCase(WithEquityPricingPipelineEngine, - ZiplineTestCase): - - def test_run_chunked_pipeline(self): - """ - Test that running a pipeline in chunks produces the same result as if - it were run all at once - """ - pipe = Pipeline( - columns={ - 'close': USEquityPricing.close.latest, - 'returns': Returns(window_length=2), - }, - ) - sessions = self.nyse_calendar.all_sessions - start_date = sessions[sessions.get_loc(self.START_DATE) + 2] - - pipeline_result = self.pipeline_engine.run_pipeline( - pipe, - start_date=start_date, - end_date=self.END_DATE, - ) - chunked_result = run_chunked_pipeline( - engine=self.pipeline_engine, - pipeline=pipe, - start_date=start_date, - end_date=self.END_DATE, - chunksize=22 - ) - self.assertTrue(chunked_result.equals(pipeline_result)) diff --git a/tests/utils/test_date_utils.py b/tests/utils/test_date_utils.py index b9f3cd05..67ac141a 100644 --- a/tests/utils/test_date_utils.py +++ b/tests/utils/test_date_utils.py @@ -3,26 +3,84 @@ from nose_parameterized import parameterized from zipline.testing import ZiplineTestCase from zipline.utils.calendars import get_calendar -from zipline.utils.date_utils import roll_dates_to_previous_session +from zipline.utils.date_utils import compute_date_range_chunks -class TestRollDatesToPreviousSession(ZiplineTestCase): +def T(s): + """ + Helpful function to improve readibility. + """ + return Timestamp(s, tz='UTC') + + +class TestDateUtils(ZiplineTestCase): + + @classmethod + def init_class_fixtures(cls): + super(TestDateUtils, cls).init_class_fixtures() + cls.calendar = get_calendar('NYSE') @parameterized.expand([ - ( - Timestamp('05-19-2017', tz='UTC'), # actual trading date - Timestamp('05-19-2017', tz='UTC'), - ), - ( - Timestamp('07-04-2015', tz='UTC'), # weekend nyse holiday - Timestamp('07-02-2015', tz='UTC'), - ), - ( - Timestamp('01-16-2017', tz='UTC'), # weeknight nyse holiday - Timestamp('01-13-2017', tz='UTC'), - ), + (None, [(T('2017-01-03'), T('2017-01-31'))]), + (10, [ + (T('2017-01-03'), T('2017-01-17')), + (T('2017-01-18'), T('2017-01-31')) + ]), + (15, [ + (T('2017-01-03'), T('2017-01-24')), + (T('2017-01-25'), T('2017-01-31')) + ]), ]) - def test_roll_dates_to_previous_session(self, date, expected_rolled_date): - calendar = get_calendar('NYSE') - result = roll_dates_to_previous_session(calendar, date) - self.assertEqual(result[0], expected_rolled_date) + def test_compute_date_range_chunks(self, chunksize, expected): + # This date range results in 20 business days + start_date = T('2017-01-03') + end_date = T('2017-01-31') + + date_ranges = compute_date_range_chunks( + self.calendar.all_sessions, + start_date, + end_date, + chunksize + ) + + self.assertListEqual(list(date_ranges), expected) + + def test_compute_date_range_chunks_invalid_input(self): + # Start date not found in calendar + with self.assertRaises(KeyError) as cm: + compute_date_range_chunks( + self.calendar.all_sessions, + T('2017-05-07'), # Sunday + T('2017-06-01'), + None + ) + self.assertEqual( + str(cm.exception), + "'Start date 2017-05-07 is not found in calendar.'" + ) + + # End date not found in calendar + with self.assertRaises(KeyError) as cm: + compute_date_range_chunks( + self.calendar.all_sessions, + T('2017-05-01'), + T('2017-05-27'), # Saturday + None + ) + self.assertEqual( + str(cm.exception), + "'End date 2017-05-27 is not found in calendar.'" + ) + + # End date before start date + with self.assertRaises(ValueError) as cm: + compute_date_range_chunks( + self.calendar.all_sessions, + T('2017-06-01'), + T('2017-05-01'), + None + ) + self.assertEqual( + str(cm.exception), + "End date 2017-05-01 cannot precede start date 2017-06-01." + ) diff --git a/tests/utils/test_pandas_utils.py b/tests/utils/test_pandas_utils.py index 870a99eb..68d1efb8 100644 --- a/tests/utils/test_pandas_utils.py +++ b/tests/utils/test_pandas_utils.py @@ -173,8 +173,16 @@ class TestCatDFConcat(ZiplineTestCase): ), ] - with self.assertRaises(ValueError): + with self.assertRaises(ValueError) as cm: categorical_df_concat(mismatched_dtypes) + self.assertEqual( + str(cm.exception), + "Input DataFrames must have the same columns/dtypes." + ) - with self.assertRaises(ValueError): + with self.assertRaises(ValueError) as cm: categorical_df_concat(mismatched_column_names) + self.assertEqual( + str(cm.exception), + "Input DataFrames must have the same columns/dtypes." + ) diff --git a/zipline/pipeline/__init__.py b/zipline/pipeline/__init__.py index 16c429c2..a169256b 100644 --- a/zipline/pipeline/__init__.py +++ b/zipline/pipeline/__init__.py @@ -1,8 +1,5 @@ from __future__ import print_function from zipline.assets import AssetFinder -from zipline.utils.calendars import get_calendar -from zipline.utils.date_utils import compute_date_range_chunks -from zipline.utils.pandas_utils import categorical_df_concat from .classifiers import Classifier, CustomClassifier from .engine import SimplePipelineEngine @@ -50,39 +47,6 @@ def engine_from_files(daily_bar_path, ) -def run_chunked_pipeline(engine, pipeline, start_date, end_date, chunksize): - """Run a pipeline to collect the results. - - Parameters - ---------- - engine : Engine - The pipeline engine. - pipeline : Pipeline - The pipeline to run. - start_date : pd.Timestamp - The start date to run the pipeline for. - end_date : pd.Timestamp - The end date to run the pipeline for. - chunksize : int or None - The number of days to execute at a time. If this is None, all the days - will be run at once. - - Returns - ------- - results : pd.DataFrame - The results for each output term in the pipeline. - """ - ranges = compute_date_range_chunks( - get_calendar('NYSE'), - start_date, - end_date, - chunksize, - ) - chunks = [engine.run_pipeline(pipeline, s, e) for s, e in ranges] - - return categorical_df_concat(chunks, inplace=True) - - __all__ = ( 'Classifier', 'CustomFactor', @@ -94,7 +58,6 @@ __all__ = ( 'Filter', 'Pipeline', 'SimplePipelineEngine', - 'run_chunked_pipeline', 'Term', 'TermGraph', ) diff --git a/zipline/pipeline/engine.py b/zipline/pipeline/engine.py index 1d28a834..65a8a6ab 100644 --- a/zipline/pipeline/engine.py +++ b/zipline/pipeline/engine.py @@ -12,6 +12,7 @@ from six import ( with_metaclass, ) from numpy import array +from odo.utils import copydoc from pandas import DataFrame, MultiIndex from toolz import groupby, juxt from toolz.curried.operator import getitem @@ -27,6 +28,9 @@ from zipline.utils.pandas_utils import explode from .term import AssetExists, InputDates, LoadableTerm +from zipline.utils.date_utils import compute_date_range_chunks +from zipline.utils.pandas_utils import categorical_df_concat + class PipelineEngine(with_metaclass(ABCMeta)): @@ -62,6 +66,45 @@ class PipelineEngine(with_metaclass(ABCMeta)): """ raise NotImplementedError("run_pipeline") + @abstractmethod + def run_chunked_pipeline(self, pipeline, start_date, end_date, chunksize): + """ + Compute values for `pipeline` in number of days equal to `chunksize` + and return stitched up result. Computing in chunks is useful for + pipelines computed over a long period of time. + + Parameters + ---------- + pipeline : Pipeline + The pipeline to run. + start_date : pd.Timestamp + The start date to run the pipeline for. + end_date : pd.Timestamp + The end date to run the pipeline for. + chunksize : int or None + The number of days to execute at a time. If None, then + results will be calculated for entire date range at once. + + Returns + ------- + result : pd.DataFrame + A frame of computed results. + + The columns `result` correspond to the entries of + `pipeline.columns`, which should be a dictionary mapping strings to + instances of `zipline.pipeline.term.Term`. + + For each date between `start_date` and `end_date`, `result` will + contain a row for each asset that passed `pipeline.screen`. A + screen of None indicates that a row should be returned for each + asset that existed each day. + + See Also + -------- + :meth:`PipelineEngine.run_pipeline` + """ + raise NotImplementedError("run_chunked_pipeline") + class NoEngineRegistered(Exception): """ @@ -80,6 +123,12 @@ class ExplodingPipelineEngine(PipelineEngine): "resources were registered." ) + def run_chunked_pipeline(self, pipeline, start_date, end_date, chunksize): + raise NoEngineRegistered( + "Attempted to run a chunked pipeline but no pipeline " + "resources were registered." + ) + def default_populate_initial_workspace(initial_workspace, root_mask_term, @@ -114,7 +163,7 @@ def default_populate_initial_workspace(initial_workspace, return initial_workspace -class SimplePipelineEngine(object): +class SimplePipelineEngine(PipelineEngine): """ PipelineEngine class that computes each term independently. @@ -146,7 +195,6 @@ class SimplePipelineEngine(object): '_root_mask_term', '_root_mask_dates_term', '_populate_initial_workspace', - '__weakref__', ) def __init__(self, @@ -210,7 +258,8 @@ class SimplePipelineEngine(object): See Also -------- - PipelineEngine.run_pipeline + :meth:`PipelineEngine.run_pipeline` + :meth:`PipelineEngine.run_chunked_pipeline` """ if end_date < start_date: raise ValueError( @@ -256,6 +305,18 @@ class SimplePipelineEngine(object): assets, ) + @copydoc(PipelineEngine.run_chunked_pipeline) + def run_chunked_pipeline(self, pipeline, start_date, end_date, chunksize): + ranges = compute_date_range_chunks( + self._calendar, + start_date, + end_date, + chunksize, + ) + chunks = [self.run_pipeline(pipeline, s, e) for s, e in ranges] + + return categorical_df_concat(chunks, inplace=True) + def _compute_root_mask(self, start_date, end_date, extra_rows): """ Compute a lifetimes matrix from our AssetFinder, then drop columns that diff --git a/zipline/testing/fixtures.py b/zipline/testing/fixtures.py index f1439d56..11bf5ace 100644 --- a/zipline/testing/fixtures.py +++ b/zipline/testing/fixtures.py @@ -1,4 +1,3 @@ -from itertools import repeat import os import sqlite3 from unittest import TestCase @@ -1333,12 +1332,15 @@ class WithEquityPricingPipelineEngine(WithAdjustmentReader, cls.bcolz_equity_daily_bar_reader, SQLiteAdjustmentReader(cls.adjustments_db_path), ) - dispatcher = dict( - zip(USEquityPricing.columns, repeat(loader)) - ).__getitem__ + + def get_loader(column): + if column in USEquityPricing.columns: + return loader + else: + raise AssertionError("No loader registered for %s" % column) cls.pipeline_engine = SimplePipelineEngine( - get_loader=dispatcher, + get_loader=get_loader, calendar=cls.nyse_sessions, asset_finder=cls.asset_finder, ) diff --git a/zipline/utils/date_utils.py b/zipline/utils/date_utils.py index c59109a2..22e57383 100644 --- a/zipline/utils/date_utils.py +++ b/zipline/utils/date_utils.py @@ -1,29 +1,6 @@ from toolz import partition_all -def roll_dates_to_previous_session(sessions, *dates): - """ - Roll `dates` to the last session of `calendar`. Return input date if it - is a valid session. - - Parameters - ---------- - sessions : pandas.tseries.index.DatetimeIndex - The list of valid session dates. - *dates : pd.Timestamp - The dates for which the last trading date is needed. - - Returns - ------- - rolled_dates: pandas.tseries.index.DatetimeIndex - The last trading date of the input dates, inclusive. - - """ - # Find the previous index value if there is no exact match. - locs = [sessions.get_loc(dt, method='ffill') for dt in dates] - return sessions[locs].tolist() - - def compute_date_range_chunks(sessions, start_date, end_date, chunksize): """Compute the start and end dates to run a pipeline for. From ff1c673a9d4823c06e261e9edec91ed31edbe99a Mon Sep 17 00:00:00 2001 From: Ana Ruelas Date: Fri, 2 Jun 2017 11:48:33 -0400 Subject: [PATCH 5/6] ENH: Include sharedoc function --- tests/utils/test_sharedoc.py | 21 +++++++++++++++++++++ zipline/pipeline/engine.py | 2 +- zipline/utils/sharedoc.py | 19 +++++++++++++++++++ 3 files changed, 41 insertions(+), 1 deletion(-) create mode 100644 tests/utils/test_sharedoc.py diff --git a/tests/utils/test_sharedoc.py b/tests/utils/test_sharedoc.py new file mode 100644 index 00000000..bed1cef9 --- /dev/null +++ b/tests/utils/test_sharedoc.py @@ -0,0 +1,21 @@ +from zipline.testing import ZiplineTestCase +from zipline.utils.sharedoc import copydoc + + +class TestSharedoc(ZiplineTestCase): + + def test_copydoc(self): + def original_docstring_function(): + """ + My docstring brings the boys to the yard. + """ + pass + + @copydoc(original_docstring_function) + def copied_docstring_function(): + pass + + self.assertEqual( + original_docstring_function.__doc__, + copied_docstring_function.__doc__ + ) diff --git a/zipline/pipeline/engine.py b/zipline/pipeline/engine.py index 65a8a6ab..00906f90 100644 --- a/zipline/pipeline/engine.py +++ b/zipline/pipeline/engine.py @@ -12,7 +12,6 @@ from six import ( with_metaclass, ) from numpy import array -from odo.utils import copydoc from pandas import DataFrame, MultiIndex from toolz import groupby, juxt from toolz.curried.operator import getitem @@ -30,6 +29,7 @@ from .term import AssetExists, InputDates, LoadableTerm from zipline.utils.date_utils import compute_date_range_chunks from zipline.utils.pandas_utils import categorical_df_concat +from zipline.utils.sharedoc import copydoc class PipelineEngine(with_metaclass(ABCMeta)): diff --git a/zipline/utils/sharedoc.py b/zipline/utils/sharedoc.py index d0d7248c..6ba53607 100644 --- a/zipline/utils/sharedoc.py +++ b/zipline/utils/sharedoc.py @@ -5,6 +5,7 @@ across different functions. import re from six import iteritems from textwrap import dedent +from toolz import curry PIPELINE_DOWNSAMPLING_FREQUENCY_DOC = dedent( """\ @@ -98,3 +99,21 @@ def templated_docstring(**docs): f.__doc__ = format_docstring(f.__name__, f.__doc__, docs) return f return decorator + + +@curry +def copydoc(from_, to): + """Copies the docstring from one function to another. + Parameters + ---------- + from_ : any + The object to copy the docstring from. + to : any + The object to copy the docstring to. + Returns + ------- + to : any + ``to`` with the docstring from ``from_`` + """ + to.__doc__ = from_.__doc__ + return to From ad3c72a8fb8b2ebc6e01fb2705328afc4dba7056 Mon Sep 17 00:00:00 2001 From: Ana Ruelas Date: Fri, 2 Jun 2017 12:00:04 -0400 Subject: [PATCH 6/6] ENH: Use context manager to suppress nan-categorical warning --- zipline/utils/pandas_utils.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/zipline/utils/pandas_utils.py b/zipline/utils/pandas_utils.py index 7323e245..2cae7f90 100644 --- a/zipline/utils/pandas_utils.py +++ b/zipline/utils/pandas_utils.py @@ -258,10 +258,11 @@ def categorical_df_concat(df_list, inplace=False): new_categories = sorted( set().union( *(frame[col].cat.categories for frame in df_list) - ) - {None} + ) ) - for df in df_list: - df[col].cat.set_categories(new_categories, inplace=True) + with ignore_pandas_nan_categorical_warning(): + for df in df_list: + df[col].cat.set_categories(new_categories, inplace=True) return pd.concat(df_list)