From 2235a53581cba0c74f5a49647243e5abba8d6d57 Mon Sep 17 00:00:00 2001 From: Scott Sanderson Date: Thu, 10 Dec 2015 17:16:29 -0500 Subject: [PATCH 1/4] ENH: Add EWMA and `DollarVolume` factors. --- tests/pipeline/test_engine.py | 167 ++++++++++++++++++++++++ tests/pipeline/test_factor.py | 5 +- tests/pipeline/test_term.py | 29 +++++ zipline/errors.py | 11 ++ zipline/pipeline/factors/__init__.py | 8 +- zipline/pipeline/factors/factor.py | 9 +- zipline/pipeline/factors/latest.py | 2 +- zipline/pipeline/factors/technical.py | 124 +++++++++++++++++- zipline/pipeline/filters/filter.py | 8 +- zipline/pipeline/mixins.py | 101 +++++++++++++++ zipline/pipeline/term.py | 179 ++++++++++++++------------ 11 files changed, 545 insertions(+), 98 deletions(-) create mode 100644 zipline/pipeline/mixins.py diff --git a/tests/pipeline/test_engine.py b/tests/pipeline/test_engine.py index 7587d663..6ce9bb9b 100644 --- a/tests/pipeline/test_engine.py +++ b/tests/pipeline/test_engine.py @@ -6,7 +6,9 @@ from collections import OrderedDict from unittest import TestCase from itertools import product +from nose_parameterized import parameterized from numpy import ( + arange, array, full, nan, @@ -14,12 +16,16 @@ from numpy import ( zeros, float32, concatenate, + log, ) +from numpy.testing import assert_almost_equal from pandas import ( DataFrame, date_range, + ewma, Int64Index, MultiIndex, + rolling_apply, rolling_mean, Series, Timestamp, @@ -46,8 +52,12 @@ from zipline.pipeline.loaders.equity_pricing_loader import ( from zipline.pipeline.engine import SimplePipelineEngine from zipline.pipeline import CustomFactor from zipline.pipeline.factors import ( + DollarVolume, MaxDrawdown, SimpleMovingAverage, + EWMA, + ExponentialWeightedMovingAverage, + DollarVolume, ) from zipline.utils.memoize import lazyval from zipline.utils.test_utils import ( @@ -767,3 +777,160 @@ class SyntheticBcolzTestCase(TestCase): result = results['drawdown'].unstack() assert_frame_equal(expected, result) + + +class ParameterizedFactorTestCase(TestCase): + @classmethod + def setUpClass(cls): + cls.env = TradingEnvironment() + day = cls.env.trading_day + + cls.sids = sids = Int64Index([1, 2, 3]) + cls.dates = dates = date_range( + '2015-02-01', + '2015-02-28', + freq=day, + tz='UTC', + ) + + asset_info = make_simple_equity_info( + cls.sids, + start_date=Timestamp('2015-01-31', tz='UTC'), + end_date=Timestamp('2015-03-01', tz='UTC'), + ) + cls.env.write_data(equities_df=asset_info) + cls.asset_finder = cls.env.asset_finder + + cls.raw_data = DataFrame( + data=arange(len(dates) * len(sids), dtype=float).reshape( + len(dates), len(sids), + ), + index=dates, + columns=cls.asset_finder.retrieve_all(sids), + ) + + close_loader = DataFrameLoader(USEquityPricing.close, cls.raw_data) + volume_loader = DataFrameLoader( + USEquityPricing.volume, + cls.raw_data * 2, + ) + + cls.engine = SimplePipelineEngine( + { + USEquityPricing.close: close_loader, + USEquityPricing.volume: volume_loader, + }.__getitem__, + cls.dates, + cls.asset_finder, + ) + + @classmethod + def tearDownClass(cls): + del cls.env + del cls.asset_finder + + def expected_ewma(self, window_length, decay_rate): + alpha = 1 - decay_rate + span = (2 / alpha) - 1 + return rolling_apply( + self.raw_data, + window_length, + lambda window: ewma(window, span=span)[-1], + )[window_length:] + + @parameterized.expand([ + (3,), + (5,), + ]) + def test_ewma(self, window_length): + def ewma_name(decay_rate): + return 'ewma_%s' % decay_rate + + decay_rates = [0.25, 0.5, 0.75] + ewmas = { + ewma_name(decay_rate): ExponentialWeightedMovingAverage( + inputs=(USEquityPricing.close,), + window_length=window_length, + decay_rate=decay_rate, + ) + for decay_rate in decay_rates + } + + all_results = self.engine.run_pipeline( + Pipeline(columns=ewmas), + self.dates[window_length], + self.dates[-1], + ) + + for decay_rate in decay_rates: + result = all_results[ewma_name(decay_rate)].unstack() + expected = self.expected_ewma(window_length, decay_rate) + assert_frame_equal(result, expected) + + @staticmethod + def decay_rate_to_span(decay_rate): + alpha = 1 - decay_rate + return (2 / alpha) - 1 + + @staticmethod + def decay_rate_to_com(decay_rate): + alpha = 1 - decay_rate + return (1 / alpha) - 1 + + @staticmethod + def decay_rate_to_halflife(decay_rate): + return log(.5) / log(decay_rate) + + @parameterized.expand([ + (3,), + (5,), + (10,), + ]) + def test_from_span(self, span): + from_span = EWMA.from_span( + inputs=[USEquityPricing.close], + window_length=20, + span=span, + ) + implied_span = self.decay_rate_to_span(from_span.params['decay_rate']) + assert_almost_equal(span, implied_span) + + @parameterized.expand([ + (3,), + (5,), + (10,), + ]) + def test_from_halflife(self, halflife): + from_hl = EWMA.from_halflife( + inputs=[USEquityPricing.close], + window_length=20, + halflife=halflife, + ) + implied_hl = self.decay_rate_to_halflife(from_hl.params['decay_rate']) + assert_almost_equal(halflife, implied_hl) + + @parameterized.expand([ + (3,), + (5,), + (10,), + ]) + def test_from_com(self, com): + from_com = EWMA.from_center_of_mass( + inputs=[USEquityPricing.close], + window_length=20, + center_of_mass=com, + ) + implied_com = self.decay_rate_to_com(from_com.params['decay_rate']) + assert_almost_equal(com, implied_com) + + def test_ewma_aliasing(self): + self.assertIs(ExponentialWeightedMovingAverage, EWMA) + + def test_dollar_volume(self): + results = self.engine.run_pipeline( + Pipeline(columns={'dv': DollarVolume()}), + self.dates[0], + self.dates[-1], + )['dv'].unstack() + expected = (self.raw_data ** 2) * 2 + assert_frame_equal(results, expected) diff --git a/tests/pipeline/test_factor.py b/tests/pipeline/test_factor.py index 0be70e5e..22a2242e 100644 --- a/tests/pipeline/test_factor.py +++ b/tests/pipeline/test_factor.py @@ -18,7 +18,10 @@ from numpy.random import randn, seed from zipline.errors import UnknownRankMethod from zipline.lib.rank import masked_rankdata_2d from zipline.pipeline import Factor, Filter, TermGraph -from zipline.pipeline.factors import RSI, Returns +from zipline.pipeline.factors import ( + Returns, + RSI, +) from zipline.utils.test_utils import check_allclose, check_arrays from zipline.utils.numpy_utils import datetime64ns_dtype, float64_dtype, np_NaT diff --git a/tests/pipeline/test_term.py b/tests/pipeline/test_term.py index 23bf1753..1a4f94af 100644 --- a/tests/pipeline/test_term.py +++ b/tests/pipeline/test_term.py @@ -1,6 +1,7 @@ """ Tests for Term. """ +from collections import Counter from itertools import product from unittest import TestCase @@ -169,6 +170,13 @@ class ObjectIdentityTestCase(TestCase): for obj in objs: self.assertIs(first, obj) + def assertDifferentObjects(self, *objs): + id_counts = Counter(map(id, objs)) + ((most_common_id, count),) = id_counts.most_common(1) + if count > 1: + dupe = [o for o in objs if id(o) == most_common_id][0] + self.fail("%s appeared %d times in %s" % (dupe, count, objs)) + def test_instance_caching(self): self.assertSameObject(*gen_equivalent_factors()) @@ -259,6 +267,27 @@ class ObjectIdentityTestCase(TestCase): method = getattr(f, funcname) self.assertIs(method(), method()) + def test_parameterized_term(self): + + class SomeFactorParameterized(SomeFactor): + params = ('a', 'b') + + f = SomeFactorParameterized(a=1, b=2) + self.assertEqual(f.params, {'a': 1, 'b': 2}) + + g = SomeFactorParameterized(a=1, b=3) + h = SomeFactorParameterized(a=2, b=2) + self.assertDifferentObjects(f, g, h) + + f2 = SomeFactorParameterized(a=1, b=2) + f3 = SomeFactorParameterized(b=2, a=1) + self.assertSameObject(f, f2, f3) + + self.assertEqual(f.params['a'], 1) + self.assertEqual(f.params['b'], 2) + self.assertEqual(f.window_length, SomeFactor.window_length) + self.assertEqual(f.inputs, tuple(SomeFactor.inputs)) + def test_bad_input(self): class SomeFactor(Factor): diff --git a/zipline/errors.py b/zipline/errors.py index 5b611e0f..43dc0809 100644 --- a/zipline/errors.py +++ b/zipline/errors.py @@ -375,6 +375,17 @@ class WindowLengthNotSpecified(ZiplineError): ) +class InvalidTermParams(ZiplineError): + """ + Raised if a user attempts to construct a Term using ParameterizedTermMixin + without specifying a `params` list in the class body. + """ + msg = ( + "Expected a list of strings as a class-level attribute for " + "{termname}.params, but got {value} instead." + ) + + class DTypeNotSpecified(ZiplineError): """ Raised if a user attempts to construct a term without specifying dtype and diff --git a/zipline/pipeline/factors/__init__.py b/zipline/pipeline/factors/__init__.py index 5b5da33d..3236a019 100644 --- a/zipline/pipeline/factors/__init__.py +++ b/zipline/pipeline/factors/__init__.py @@ -8,6 +8,9 @@ from .events import ( BusinessDaysUntilNextEarnings, ) from .technical import ( + DollarVolume, + EWMA, + ExponentialWeightedMovingAverage, MaxDrawdown, RSI, Returns, @@ -17,9 +20,12 @@ from .technical import ( ) __all__ = [ - 'CustomFactor', 'BusinessDaysSincePreviousEarnings', 'BusinessDaysUntilNextEarnings', + 'CustomFactor', + 'DollarVolume', + 'EWMA', + 'ExponentialWeightedMovingAverage', 'Factor', 'Latest', 'MaxDrawdown', diff --git a/zipline/pipeline/factors/factor.py b/zipline/pipeline/factors/factor.py index 3c46febb..d7cb647a 100644 --- a/zipline/pipeline/factors/factor.py +++ b/zipline/pipeline/factors/factor.py @@ -12,13 +12,12 @@ from zipline.errors import ( UnsupportedDataType, ) from zipline.lib.rank import masked_rankdata_2d -from zipline.pipeline.term import ( +from zipline.pipeline.mixins import ( CustomTermMixin, - NotSpecified, - RequiredWindowLengthMixin, + PositiveWindowLengthMixin, SingleInputMixin, - CompositeTerm, ) +from zipline.pipeline.term import CompositeTerm, NotSpecified from zipline.pipeline.expression import ( BadBinaryOperator, COMPARISONS, @@ -606,7 +605,7 @@ class Rank(SingleInputMixin, Factor): ) -class CustomFactor(RequiredWindowLengthMixin, CustomTermMixin, Factor): +class CustomFactor(PositiveWindowLengthMixin, CustomTermMixin, Factor): ''' Base class for user-defined Factors. diff --git a/zipline/pipeline/factors/latest.py b/zipline/pipeline/factors/latest.py index 199074ec..0b134f9b 100644 --- a/zipline/pipeline/factors/latest.py +++ b/zipline/pipeline/factors/latest.py @@ -2,7 +2,7 @@ Factor that produces the most most recently-known value of Column. """ from .factor import CustomFactor -from ..term import SingleInputMixin +from ..mixins import SingleInputMixin class Latest(SingleInputMixin, CustomFactor): diff --git a/zipline/pipeline/factors/technical.py b/zipline/pipeline/factors/technical.py index 688c9b80..d82cb9c7 100644 --- a/zipline/pipeline/factors/technical.py +++ b/zipline/pipeline/factors/technical.py @@ -8,20 +8,27 @@ from bottleneck import ( nanmean, nansum, ) +from numbers import Number from numpy import ( abs, + arange, + average, clip, diff, + exp, fmax, + full, inf, isnan, + log, NINF, ) from numexpr import evaluate from zipline.pipeline.data import USEquityPricing -from zipline.pipeline.term import SingleInputMixin +from zipline.pipeline.mixins import SingleInputMixin from zipline.utils.control_flow import ignore_nanwarnings +from zipline.utils.input_validation import expect_types from .factor import CustomFactor @@ -119,3 +126,118 @@ class MaxDrawdown(CustomFactor, SingleInputMixin): for i, end in enumerate(drawdown_ends): peak = nanmax(data[:end + 1, i]) out[i] = (peak - data[end, i]) / data[end, i] + + +def DollarVolume(): + """ + Returns a Factor computing the product of most recent close price and + volume. + """ + return USEquityPricing.close.latest * USEquityPricing.volume.latest + + +def exponential_decay_weights(length, decay_rate): + """ + Return weighting vector for an exponential moving statistic on `length` + rows with a decay rate of `decay_rate`. + """ + return full(length, decay_rate) ** arange(length + 1, 1, -1) + + +class ExponentialWeightedMovingAverage(SingleInputMixin, CustomFactor): + """ + Exponentially Weighted Moving Average + + **Default Inputs:** None + **Default Window Length:** None + + Parameters + ---------- + inputs : length-1 list or tuple of BoundColumn + The expression over which to compute the average. + window_length : int > 0 + Length of the lookback window over which to compute the average. + decay_rate : float, 0 < decay_rate <= 1 + Weighting factor by which to discount past observations. + + When calculating historical averages, rows are multiplied by the + sequence:: + + decay_rate, decay_rate ** 2, decay_rate ** 3, ... + + See Also + -------- + pandas.ewma + """ + params = ('decay_rate',) + + @classmethod + @expect_types(span=Number) + def from_span(cls, inputs, window_length, span): + """ + Convenience constructor for passing `decay_rate` in terms of `span`. + + Forwards `decay_rate` as `1 - (2.0 / (1 + span))`. This provides the + behavior equivalent to passing `span` to pandas.ewma. + """ + if span <= 1: + raise ValueError( + "`span` must be a positive number. %s was passed." % span + ) + + decay_rate = (1.0 - (2.0 / (1.0 + span))) + assert 0.0 < decay_rate <= 1.0 + + return cls( + inputs=inputs, + window_length=window_length, + decay_rate=decay_rate, + ) + + @classmethod + @expect_types(halflife=Number) + def from_halflife(cls, inputs, window_length, halflife): + """ + Convenience constructor for passing `decay_rate` in terms of half life. + + Forwards `decay_rate` as `exp(log(.5) / halflife)`. This provides + the behavior equivalent to passing `halflife` to pandas.ewma. + """ + if halflife <= 0: + raise ValueError( + "`span` must be a positive number. %s was passed." % halflife + ) + decay_rate = exp(log(.5) / halflife) + assert 0.0 < decay_rate <= 1.0 + + return cls( + inputs=inputs, + window_length=window_length, + decay_rate=decay_rate, + ) + + @classmethod + def from_center_of_mass(cls, inputs, window_length, center_of_mass): + """ + Convenience constructor for passing `decay_rate` in terms of center of + mass. + + Forwards `decay_rate` as `1 - (1 / center_of_mass)`. This provides + behavior equivalent to passing `center_of_mass` to pandas.ewma. + """ + return cls( + inputs=inputs, + window_length=window_length, + decay_rate=(1.0 - (1.0 / (1.0 + center_of_mass))), + ) + + def compute(self, today, assets, out, data, decay_rate): + out[:] = average( + data, + axis=0, + weights=exponential_decay_weights(len(data), decay_rate), + ) + + +# Convenience alias. +EWMA = ExponentialWeightedMovingAverage diff --git a/zipline/pipeline/filters/filter.py b/zipline/pipeline/filters/filter.py index 34a96204..fa450254 100644 --- a/zipline/pipeline/filters/filter.py +++ b/zipline/pipeline/filters/filter.py @@ -13,12 +13,12 @@ from zipline.errors import ( BadPercentileBounds, UnsupportedDataType, ) -from zipline.pipeline.term import ( - CompositeTerm, +from zipline.pipeline.mixins import ( CustomTermMixin, - RequiredWindowLengthMixin, + PositiveWindowLengthMixin, SingleInputMixin, ) +from zipline.pipeline.term import CompositeTerm from zipline.pipeline.expression import ( BadBinaryOperator, FILTER_BINOPS, @@ -243,7 +243,7 @@ class PercentileFilter(SingleInputMixin, Filter): return (lower_bounds <= data) & (data <= upper_bounds) -class CustomFilter(RequiredWindowLengthMixin, CustomTermMixin, Filter): +class CustomFilter(PositiveWindowLengthMixin, CustomTermMixin, Filter): """ Filter analog to ``CustomFactor``. """ diff --git a/zipline/pipeline/mixins.py b/zipline/pipeline/mixins.py new file mode 100644 index 00000000..e3ebf815 --- /dev/null +++ b/zipline/pipeline/mixins.py @@ -0,0 +1,101 @@ +""" +Mixins classes for use with Filters and Factors. +""" +from numpy import full_like +from zipline.errors import WindowLengthNotPositive + +from .term import NotSpecified + + +class PositiveWindowLengthMixin(object): + """ + Validation mixin enforcing that a Term gets a positive WindowLength + """ + def _validate(self): + if not self.windowed: + raise WindowLengthNotPositive(window_length=self.window_length) + return super(PositiveWindowLengthMixin, self)._validate() + + +class SingleInputMixin(object): + """ + Validation mixin enforcing that a Term gets a length-1 inputs list. + """ + def _validate(self): + num_inputs = len(self.inputs) + if num_inputs != 1: + raise ValueError( + "{typename} expects only one input, " + "but received {num_inputs} instead.".format( + typename=type(self).__name__, + num_inputs=num_inputs + ) + ) + return super(SingleInputMixin, self)._validate() + + +class CustomTermMixin(object): + """ + Mixin for user-defined rolling-window Terms. + + Implements `_compute` in terms of a user-defined `compute` function, which + is mapped over the input windows. + + Used by CustomFactor, CustomFilter, CustomClassifier, etc. + """ + def __new__(cls, + inputs=NotSpecified, + window_length=NotSpecified, + dtype=NotSpecified, + **kwargs): + + unexpected_keys = set(kwargs) - set(cls.params) + if unexpected_keys: + raise TypeError( + "{termname} received unexpected keyword " + "arguments {unexpected}".format( + termname=cls.__name__, + unexpected={k: kwargs[k] for k in unexpected_keys}, + ) + ) + + return super(CustomTermMixin, cls).__new__( + cls, + inputs=inputs, + window_length=window_length, + dtype=dtype, + **kwargs + ) + + def compute(self, today, assets, out, *arrays): + """ + Override this method with a function that writes a value into `out`. + """ + raise NotImplementedError() + + def _compute(self, windows, dates, assets, mask): + """ + Call the user's `compute` function on each window with a pre-built + output array. + """ + # TODO: Make mask available to user's `compute`. + compute = self.compute + missing_value = self.missing_value + params = self.params + out = full_like(mask, missing_value, dtype=self.dtype) + with self.ctx: + # TODO: Consider pre-filtering columns that are all-nan at each + # time-step? + for idx, date in enumerate(dates): + compute( + date, + assets, + out[idx], + *(next(w) for w in windows), + **params + ) + out[~mask] = missing_value + return out + + def short_repr(self): + return type(self).__name__ + '(%d)' % self.window_length diff --git a/zipline/pipeline/term.py b/zipline/pipeline/term.py index 2498cfc9..5f427bc7 100644 --- a/zipline/pipeline/term.py +++ b/zipline/pipeline/term.py @@ -4,7 +4,7 @@ Base class for Filters, Factors and Classifiers from abc import ABCMeta, abstractproperty from weakref import WeakValueDictionary -from numpy import full_like, dtype as dtype_class +from numpy import dtype as dtype_class from six import with_metaclass from zipline.errors import ( @@ -12,7 +12,6 @@ from zipline.errors import ( InputTermNotAtomic, InvalidDType, TermInputsNotSpecified, - WindowLengthNotPositive, WindowLengthNotSpecified, ) from zipline.utils.memoize import lazyval @@ -34,11 +33,16 @@ class Term(with_metaclass(ABCMeta, object)): dtype = NotSpecified domain = NotSpecified + # Subclasses aren't required to provide `params`. The default behavior is + # no params. + params = () + _term_cache = WeakValueDictionary() def __new__(cls, - domain=NotSpecified, - dtype=NotSpecified, + domain=domain, + dtype=dtype, + # params is explicitly not allowed to be passed to an instance. *args, **kwargs): """ @@ -56,11 +60,14 @@ class Term(with_metaclass(ABCMeta, object)): if domain is NotSpecified: domain = cls.domain + dtype = cls._validate_dtype(dtype) + params = cls._pop_params(kwargs) identity = cls.static_identity( domain=domain, dtype=dtype, + params=params, *args, **kwargs ) @@ -71,10 +78,59 @@ class Term(with_metaclass(ABCMeta, object)): super(Term, cls).__new__(cls)._init( domain=domain, dtype=dtype, + params=params, *args, **kwargs ) return new_instance + @classmethod + def _pop_params(cls, kwargs): + """ + Pop entries from the `kwargs` passed to cls.__new__ based on the values + in `cls.params`. + + Parameters + ---------- + kwargs : dict + The kwargs passed to cls.__new__. + + Returns + ------- + params : list[(str, object)] + A list of string, value pairs containing the entries in cls.params. + + Raises + ------ + TypeError + Raised if any parameter values are not passed or not hashable. + """ + param_values = [] + for key in cls.params: + try: + value = kwargs.pop(key) + # Check here that the value is hashable so that we fail here + # instead of trying to hash the param values tuple later. + hash(key) + param_values.append(value) + except KeyError: + raise TypeError( + "{typename} expected a keyword parameter {name!r}.".format( + typename=cls.__name__, + name=key + ) + ) + except TypeError: + # Value wasn't hashable. + raise TypeError( + "{typename} expected a hashable value for parameter " + "{name!r}, but got {value!r} instead.".format( + typename=cls.__name__, + name=key, + value=value, + ) + ) + return tuple(zip(cls.params, param_values)) + @classmethod def _validate_dtype(cls, passed_dtype): """ @@ -127,7 +183,7 @@ class Term(with_metaclass(ABCMeta, object)): pass @classmethod - def static_identity(cls, domain, dtype): + def static_identity(cls, domain, dtype, params): """ Return the identity of the Term that would be constructed from the given arguments. @@ -139,12 +195,37 @@ class Term(with_metaclass(ABCMeta, object)): This is a classmethod so that it can be called from Term.__new__ to determine whether to produce a new instance. """ - return (cls, domain, dtype) + return (cls, domain, dtype, params) - def _init(self, domain, dtype): + def _init(self, domain, dtype, params): + """ + Parameters + ---------- + domain : object + Unused placeholder. + dtype : np.dtype + Dtype of this term's output. + params : tuple[(str, hashable)] + Tuple of key/value pairs of additional parameters. + """ self.domain = domain self.dtype = dtype + for name, value in params: + if hasattr(self, name): + raise TypeError( + "Parameter {name!r} conflicts with already-present" + "attribute with value {value!r}.".format( + name=name, + value=getattr(self, name), + ) + ) + # TODO: Consider setting these values as attributes and replacing + # the boilerplate in NumericalExpression, Rank, and + # PercentileFilter. + + self.params = dict(params) + # Make sure that subclasses call super() in their _validate() methods # by setting this flag. The base class implementation of _validate # should set this flag to True. @@ -217,92 +298,20 @@ class AssetExists(Term): return "AssetExists()" -# TODO: Move mixins to a separate file? -class SingleInputMixin(object): - - def _validate(self): - num_inputs = len(self.inputs) - if num_inputs != 1: - raise ValueError( - "{typename} expects only one input, " - "but received {num_inputs} instead.".format( - typename=type(self).__name__, - num_inputs=num_inputs - ) - ) - return super(SingleInputMixin, self)._validate() - - -class RequiredWindowLengthMixin(object): - def _validate(self): - if not self.windowed: - raise WindowLengthNotPositive(window_length=self.window_length) - return super(RequiredWindowLengthMixin, self)._validate() - - -class CustomTermMixin(object): - """ - Mixin for user-defined rolling-window Terms. - - Implements `_compute` in terms of a user-defined `compute` function, which - is mapped over the input windows. - - Used by CustomFactor, CustomFilter, CustomClassifier, etc. - """ - - def __new__(cls, - inputs=NotSpecified, - window_length=NotSpecified, - dtype=NotSpecified): - return super(CustomTermMixin, cls).__new__( - cls, - inputs=inputs, - window_length=window_length, - dtype=dtype, - ) - - def compute(self, today, assets, out, *arrays): - """ - Override this method with a function that writes a value into `out`. - """ - raise NotImplementedError() - - def _compute(self, windows, dates, assets, mask): - """ - Call the user's `compute` function on each window with a pre-built - output array. - """ - # TODO: Make mask available to user's `compute`. - compute = self.compute - missing_value = self.missing_value - out = full_like(mask, missing_value, dtype=self.dtype) - with self.ctx: - # TODO: Consider pre-filtering columns that are all-nan at each - # time-step? - for idx, date in enumerate(dates): - compute( - date, - assets, - out[idx], - *(next(w) for w in windows) - ) - out[~mask] = missing_value - return out - - def short_repr(self): - return type(self).__name__ + '(%d)' % self.window_length - - class CompositeTerm(Term): inputs = NotSpecified window_length = NotSpecified mask = NotSpecified - def __new__(cls, inputs=NotSpecified, window_length=NotSpecified, - mask=NotSpecified, *args, **kwargs): + def __new__(cls, + inputs=inputs, + window_length=window_length, + mask=mask, + *args, **kwargs): if inputs is NotSpecified: inputs = cls.inputs + # Having inputs = NotSpecified is an error, but we handle it later # in self._validate rather than here. if inputs is not NotSpecified: From b91f9697b9632059277ba10bab735de26882a190 Mon Sep 17 00:00:00 2001 From: Scott Sanderson Date: Fri, 11 Dec 2015 22:09:19 -0500 Subject: [PATCH 2/4] ENH: Add ExponentialWeightedStandardDeviation. --- tests/pipeline/test_engine.py | 81 +++++++++++++-------- zipline/pipeline/factors/__init__.py | 4 + zipline/pipeline/factors/technical.py | 101 ++++++++++++++++++++++---- 3 files changed, 142 insertions(+), 44 deletions(-) diff --git a/tests/pipeline/test_engine.py b/tests/pipeline/test_engine.py index 6ce9bb9b..480ea4c0 100644 --- a/tests/pipeline/test_engine.py +++ b/tests/pipeline/test_engine.py @@ -23,6 +23,7 @@ from pandas import ( DataFrame, date_range, ewma, + ewmstd, Int64Index, MultiIndex, rolling_apply, @@ -34,6 +35,7 @@ from pandas.compat.chainmap import ChainMap from pandas.util.testing import assert_frame_equal from six import iteritems, itervalues from testfixtures import TempDirectory +from toolz import merge from zipline.data.us_equity_pricing import BcolzDailyBarReader from zipline.finance.trading import TradingEnvironment @@ -53,11 +55,12 @@ from zipline.pipeline.engine import SimplePipelineEngine from zipline.pipeline import CustomFactor from zipline.pipeline.factors import ( DollarVolume, + EWMA, + EWMSTD, + ExponentialWeightedMovingAverage, + ExponentialWeightedStandardDeviation, MaxDrawdown, SimpleMovingAverage, - EWMA, - ExponentialWeightedMovingAverage, - DollarVolume, ) from zipline.utils.memoize import lazyval from zipline.utils.test_utils import ( @@ -838,17 +841,39 @@ class ParameterizedFactorTestCase(TestCase): lambda window: ewma(window, span=span)[-1], )[window_length:] + def expected_ewmstd(self, window_length, decay_rate): + alpha = 1 - decay_rate + span = (2 / alpha) - 1 + return rolling_apply( + self.raw_data, + window_length, + lambda window: ewmstd(window, span=span)[-1], + )[window_length:] + @parameterized.expand([ (3,), (5,), ]) - def test_ewma(self, window_length): + def test_ewm_stats(self, window_length): + def ewma_name(decay_rate): return 'ewma_%s' % decay_rate + def ewmstd_name(decay_rate): + return 'ewmstd_%s' % decay_rate + decay_rates = [0.25, 0.5, 0.75] ewmas = { - ewma_name(decay_rate): ExponentialWeightedMovingAverage( + ewma_name(decay_rate): EWMA( + inputs=(USEquityPricing.close,), + window_length=window_length, + decay_rate=decay_rate, + ) + for decay_rate in decay_rates + } + + ewmstds = { + ewmstd_name(decay_rate): EWMSTD( inputs=(USEquityPricing.close,), window_length=window_length, decay_rate=decay_rate, @@ -857,15 +882,19 @@ class ParameterizedFactorTestCase(TestCase): } all_results = self.engine.run_pipeline( - Pipeline(columns=ewmas), + Pipeline(columns=merge(ewmas, ewmstds)), self.dates[window_length], self.dates[-1], ) for decay_rate in decay_rates: - result = all_results[ewma_name(decay_rate)].unstack() - expected = self.expected_ewma(window_length, decay_rate) - assert_frame_equal(result, expected) + ewma_result = all_results[ewma_name(decay_rate)].unstack() + ewma_expected = self.expected_ewma(window_length, decay_rate) + assert_frame_equal(ewma_result, ewma_expected) + + ewmstd_result = all_results[ewmstd_name(decay_rate)].unstack() + ewmstd_expected = self.expected_ewmstd(window_length, decay_rate) + assert_frame_equal(ewmstd_result, ewmstd_expected) @staticmethod def decay_rate_to_span(decay_rate): @@ -881,13 +910,12 @@ class ParameterizedFactorTestCase(TestCase): def decay_rate_to_halflife(decay_rate): return log(.5) / log(decay_rate) - @parameterized.expand([ - (3,), - (5,), - (10,), - ]) - def test_from_span(self, span): - from_span = EWMA.from_span( + def ewm_cases(): + return product([EWMSTD, EWMA], [3, 5, 10]) + + @parameterized.expand(ewm_cases()) + def test_from_span(self, type_, span): + from_span = type_.from_span( inputs=[USEquityPricing.close], window_length=20, span=span, @@ -895,12 +923,8 @@ class ParameterizedFactorTestCase(TestCase): implied_span = self.decay_rate_to_span(from_span.params['decay_rate']) assert_almost_equal(span, implied_span) - @parameterized.expand([ - (3,), - (5,), - (10,), - ]) - def test_from_halflife(self, halflife): + @parameterized.expand(ewm_cases()) + def test_from_halflife(self, type_, halflife): from_hl = EWMA.from_halflife( inputs=[USEquityPricing.close], window_length=20, @@ -909,12 +933,8 @@ class ParameterizedFactorTestCase(TestCase): implied_hl = self.decay_rate_to_halflife(from_hl.params['decay_rate']) assert_almost_equal(halflife, implied_hl) - @parameterized.expand([ - (3,), - (5,), - (10,), - ]) - def test_from_com(self, com): + @parameterized.expand(ewm_cases()) + def test_from_com(self, type_, com): from_com = EWMA.from_center_of_mass( inputs=[USEquityPricing.close], window_length=20, @@ -923,8 +943,11 @@ class ParameterizedFactorTestCase(TestCase): implied_com = self.decay_rate_to_com(from_com.params['decay_rate']) assert_almost_equal(com, implied_com) - def test_ewma_aliasing(self): + del ewm_cases + + def test_ewm_aliasing(self): self.assertIs(ExponentialWeightedMovingAverage, EWMA) + self.assertIs(ExponentialWeightedStandardDeviation, EWMSTD) def test_dollar_volume(self): results = self.engine.run_pipeline( diff --git a/zipline/pipeline/factors/__init__.py b/zipline/pipeline/factors/__init__.py index 3236a019..f9924679 100644 --- a/zipline/pipeline/factors/__init__.py +++ b/zipline/pipeline/factors/__init__.py @@ -10,7 +10,9 @@ from .events import ( from .technical import ( DollarVolume, EWMA, + EWMSTD, ExponentialWeightedMovingAverage, + ExponentialWeightedStandardDeviation, MaxDrawdown, RSI, Returns, @@ -25,7 +27,9 @@ __all__ = [ 'CustomFactor', 'DollarVolume', 'EWMA', + 'EWMSTD', 'ExponentialWeightedMovingAverage', + 'ExponentialWeightedStandardDeviation', 'Factor', 'Latest', 'MaxDrawdown', diff --git a/zipline/pipeline/factors/technical.py b/zipline/pipeline/factors/technical.py index d82cb9c7..3233d5d9 100644 --- a/zipline/pipeline/factors/technical.py +++ b/zipline/pipeline/factors/technical.py @@ -22,6 +22,8 @@ from numpy import ( isnan, log, NINF, + sqrt, + sum as np_sum, ) from numexpr import evaluate @@ -136,17 +138,9 @@ def DollarVolume(): return USEquityPricing.close.latest * USEquityPricing.volume.latest -def exponential_decay_weights(length, decay_rate): +class _ExponentialWeightedFactor(SingleInputMixin, CustomFactor): """ - Return weighting vector for an exponential moving statistic on `length` - rows with a decay rate of `decay_rate`. - """ - return full(length, decay_rate) ** arange(length + 1, 1, -1) - - -class ExponentialWeightedMovingAverage(SingleInputMixin, CustomFactor): - """ - Exponentially Weighted Moving Average + Base class for factors implementing exponential-weighted operations. **Default Inputs:** None **Default Window Length:** None @@ -165,12 +159,23 @@ class ExponentialWeightedMovingAverage(SingleInputMixin, CustomFactor): decay_rate, decay_rate ** 2, decay_rate ** 3, ... - See Also - -------- - pandas.ewma + Methods + ------- + weights + from_span + from_halflife + from_center_of_mass """ params = ('decay_rate',) + @staticmethod + def weights(length, decay_rate): + """ + Return weighting vector for an exponential moving statistic on `length` + rows with a decay rate of `decay_rate`. + """ + return full(length, decay_rate) ** arange(length + 1, 1, -1) + @classmethod @expect_types(span=Number) def from_span(cls, inputs, window_length, span): @@ -231,13 +236,79 @@ class ExponentialWeightedMovingAverage(SingleInputMixin, CustomFactor): decay_rate=(1.0 - (1.0 / (1.0 + center_of_mass))), ) + +class ExponentialWeightedMovingAverage(_ExponentialWeightedFactor): + """ + Exponentially Weighted Moving Average + + **Default Inputs:** None + **Default Window Length:** None + + Parameters + ---------- + inputs : length-1 list/tuple of BoundColumn + The expression over which to compute the average. + window_length : int > 0 + Length of the lookback window over which to compute the average. + decay_rate : float, 0 < decay_rate <= 1 + Weighting factor by which to discount past observations. + + When calculating historical averages, rows are multiplied by the + sequence:: + + decay_rate, decay_rate ** 2, decay_rate ** 3, ... + + See Also + -------- + pandas.ewma + """ def compute(self, today, assets, out, data, decay_rate): out[:] = average( data, axis=0, - weights=exponential_decay_weights(len(data), decay_rate), + weights=self.weights(len(data), decay_rate), ) -# Convenience alias. +class ExponentialWeightedStandardDeviation(_ExponentialWeightedFactor): + """ + Exponentially Weighted Moving Standard Deviation + + **Default Inputs:** None + **Default Window Length:** None + + Parameters + ---------- + inputs : length-1 list/tuple of BoundColumn + The expression over which to compute the average. + window_length : int > 0 + Length of the lookback window over which to compute the average. + decay_rate : float, 0 < decay_rate <= 1 + Weighting factor by which to discount past observations. + + When calculating historical averages, rows are multiplied by the + sequence:: + + decay_rate, decay_rate ** 2, decay_rate ** 3, ... + + See Also + -------- + pandas.ewmstd + """ + + def compute(self, today, assets, out, data, decay_rate): + weights = self.weights(len(data), decay_rate) + + mean = average(data, axis=0, weights=weights) + variance = average((data - mean) ** 2, axis=0, weights=weights) + + squared_weight_sum = (np_sum(weights) ** 2) + bias_correction = ( + squared_weight_sum / (squared_weight_sum - np_sum(weights ** 2)) + ) + out[:] = sqrt(variance * bias_correction) + + +# Convenience aliases. EWMA = ExponentialWeightedMovingAverage +EWMSTD = ExponentialWeightedStandardDeviation From 7996c0710747b410a17188e52469035dab86dede Mon Sep 17 00:00:00 2001 From: Scott Sanderson Date: Fri, 11 Dec 2015 22:20:38 -0500 Subject: [PATCH 3/4] DOC: Add whatsnew. --- docs/source/whatsnew/0.8.4.txt | 15 +++++++++++++-- 1 file changed, 13 insertions(+), 2 deletions(-) diff --git a/docs/source/whatsnew/0.8.4.txt b/docs/source/whatsnew/0.8.4.txt index 94cc6548..3b0376de 100644 --- a/docs/source/whatsnew/0.8.4.txt +++ b/docs/source/whatsnew/0.8.4.txt @@ -17,7 +17,6 @@ Highlights * :class:`~zipline.assets.assets.AssetFinder` speedups (:issue:`830` and :issue:`817`). - Enhancements ~~~~~~~~~~~~ @@ -53,6 +52,13 @@ Enhancements calculates the percent change in close price over the given window_length. (:issue:`884`). +* Added a new built-in factor: + :class:`~zipline.pipeline.factors.DollarVolume`. (:issue:`910`). + +* Added :class:`~zipline.pipeline.factors.ExponentialWeightedMovingAverage` and + :class:`~zipline.pipeline.factors.ExponentialWeightedStandardDeviation` + factors. (:issue:`910`). + Experimental Features ~~~~~~~~~~~~~~~~~~~~~ @@ -60,7 +66,11 @@ Experimental Features Experimental features are subject to change. -None +* Added support for parameterized ``Factor`` subclasses. Factors may specify + ``params`` as a class-level attribute containing a tuple of parameter names. + These values are then accepted by the constructor and forwarded by name to + the factor's ``compute`` function. This API is experimental, and may change + in future releases. Bug Fixes ~~~~~~~~~ @@ -71,6 +81,7 @@ Bug Fixes * Fixes an error raised in calculating beta when benchmark data were sparse. Instead `numpy.nan` is returned (:issue:`859`). + * Fixed an issue pickling :func:`~zipline.utils.sentinel.sentinel` objects (:issue:`872`). From 7305f0c3b95651ced444feec7c7d8cf75d88b108 Mon Sep 17 00:00:00 2001 From: Scott Sanderson Date: Fri, 11 Dec 2015 22:28:42 -0500 Subject: [PATCH 4/4] DOC: Miscellaneous docs updates. --- docs/source/appendix.rst | 8 ++++++++ zipline/pipeline/factors/events.py | 5 +++-- zipline/pipeline/factors/factor.py | 6 +++--- zipline/pipeline/factors/technical.py | 3 +++ 4 files changed, 17 insertions(+), 5 deletions(-) diff --git a/docs/source/appendix.rst b/docs/source/appendix.rst index 2d161ed3..7119bf91 100644 --- a/docs/source/appendix.rst +++ b/docs/source/appendix.rst @@ -60,6 +60,14 @@ Pipeline API .. autoclass:: zipline.pipeline.factors.WeightedAverageValue :members: +.. autoclass:: zipline.pipeline.factors.ExponentialWeightedMovingAverage + :members: + +.. autoclass:: zipline.pipeline.factors.ExponentialWeightedStandardDeviation + :members: + +.. autofunction:: zipline.pipeline.factors.DollarVolume + .. autoclass:: zipline.pipeline.filters.Filter :members: __and__, __or__ :exclude-members: dtype diff --git a/zipline/pipeline/factors/events.py b/zipline/pipeline/factors/events.py index 77dce728..42db1fbf 100644 --- a/zipline/pipeline/factors/events.py +++ b/zipline/pipeline/factors/events.py @@ -33,9 +33,10 @@ class BusinessDaysUntilNextEarnings(Factor): Assets for which `EarningsCalendar.next_announcement` is `NaT` will produce a value of `NaN`. + See Also -------- - BusinessDaysSincePreviousEarnings + zipline.pipeline.factors.BusinessDaysSincePreviousEarnings """ inputs = [EarningsCalendar.next_announcement] window_length = 0 @@ -71,7 +72,7 @@ class BusinessDaysSincePreviousEarnings(Factor): See Also -------- - BusinessDaysUntilNextEarnings + zipline.pipeline.factors.BusinessDaysUntilNextEarnings """ inputs = [EarningsCalendar.previous_announcement] window_length = 0 diff --git a/zipline/pipeline/factors/factor.py b/zipline/pipeline/factors/factor.py index d7cb647a..2e6463b3 100644 --- a/zipline/pipeline/factors/factor.py +++ b/zipline/pipeline/factors/factor.py @@ -394,8 +394,8 @@ class Factor(CompositeTerm): See Also -------- scipy.stats.rankdata - zipline.lib.rank - zipline.pipeline.factors.Rank + zipline.lib.rank.masked_rankdata_2d + zipline.pipeline.factors.factor.Rank """ return Rank(self, method=method, ascending=ascending, mask=mask) @@ -466,7 +466,7 @@ class Factor(CompositeTerm): See Also -------- - zipline.pipeline.filters.PercentileFilter + zipline.pipeline.filters.filter.PercentileFilter """ return PercentileFilter( self, diff --git a/zipline/pipeline/factors/technical.py b/zipline/pipeline/factors/technical.py index 3233d5d9..e8329bd8 100644 --- a/zipline/pipeline/factors/technical.py +++ b/zipline/pipeline/factors/technical.py @@ -143,6 +143,7 @@ class _ExponentialWeightedFactor(SingleInputMixin, CustomFactor): Base class for factors implementing exponential-weighted operations. **Default Inputs:** None + **Default Window Length:** None Parameters @@ -242,6 +243,7 @@ class ExponentialWeightedMovingAverage(_ExponentialWeightedFactor): Exponentially Weighted Moving Average **Default Inputs:** None + **Default Window Length:** None Parameters @@ -275,6 +277,7 @@ class ExponentialWeightedStandardDeviation(_ExponentialWeightedFactor): Exponentially Weighted Moving Standard Deviation **Default Inputs:** None + **Default Window Length:** None Parameters