Merge pull request #910 from quantopian/new-factors

New factors

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

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`).
 

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,17 @@ from numpy import (
     zeros,
     float32,
     concatenate,
+    log,
 )
+from numpy.testing import assert_almost_equal
 from pandas import (
     DataFrame,
     date_range,
+    ewma,
+    ewmstd,
     Int64Index,
     MultiIndex,
+    rolling_apply,
     rolling_mean,
     Series,
     Timestamp,
@@ -28,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
@@ -46,6 +54,11 @@ 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,
+    EWMA,
+    EWMSTD,
+    ExponentialWeightedMovingAverage,
+    ExponentialWeightedStandardDeviation,
     MaxDrawdown,
     SimpleMovingAverage,
 )
@@ -767,3 +780,180 @@ 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:]
+
+    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_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): 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,
+            )
+            for decay_rate in decay_rates
+        }
+
+        all_results = self.engine.run_pipeline(
+            Pipeline(columns=merge(ewmas, ewmstds)),
+            self.dates[window_length],
+            self.dates[-1],
+        )
+
+        for decay_rate in decay_rates:
+            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):
+        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)
+
+    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,
+        )
+        implied_span = self.decay_rate_to_span(from_span.params['decay_rate'])
+        assert_almost_equal(span, implied_span)
+
+    @parameterized.expand(ewm_cases())
+    def test_from_halflife(self, type_, 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(ewm_cases())
+    def test_from_com(self, type_, 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)
+
+    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(
+            Pipeline(columns={'dv': DollarVolume()}),
+            self.dates[0],
+            self.dates[-1],
+        )['dv'].unstack()
+        expected = (self.raw_data ** 2) * 2
+        assert_frame_equal(results, expected)

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
 

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):

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

zipline/pipeline/factors/__init__.py

@@ -8,6 +8,11 @@ from .events import (
     BusinessDaysUntilNextEarnings,
 )
 from .technical import (
+    DollarVolume,
+    EWMA,
+    EWMSTD,
+    ExponentialWeightedMovingAverage,
+    ExponentialWeightedStandardDeviation,
     MaxDrawdown,
     RSI,
     Returns,
@@ -17,9 +22,14 @@ from .technical import (
 )
 
 __all__ = [
-    'CustomFactor',
     'BusinessDaysSincePreviousEarnings',
     'BusinessDaysUntilNextEarnings',
+    'CustomFactor',
+    'DollarVolume',
+    'EWMA',
+    'EWMSTD',
+    'ExponentialWeightedMovingAverage',
+    'ExponentialWeightedStandardDeviation',
     'Factor',
     'Latest',
     'MaxDrawdown',

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

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,
@@ -395,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)
 
@@ -467,7 +466,7 @@ class Factor(CompositeTerm):
 
         See Also
         --------
-        zipline.pipeline.filters.PercentileFilter
+        zipline.pipeline.filters.filter.PercentileFilter
         """
         return PercentileFilter(
             self,
@@ -606,7 +605,7 @@ class Rank(SingleInputMixin, Factor):
         )
 
 
-class CustomFactor(RequiredWindowLengthMixin, CustomTermMixin, Factor):
+class CustomFactor(PositiveWindowLengthMixin, CustomTermMixin, Factor):
     '''
     Base class for user-defined Factors.
 

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):

zipline/pipeline/factors/technical.py

@@ -8,20 +8,29 @@ from bottleneck import (
     nanmean,
     nansum,
 )
+from numbers import Number
 from numpy import (
     abs,
+    arange,
+    average,
     clip,
     diff,
+    exp,
     fmax,
+    full,
     inf,
     isnan,
+    log,
     NINF,
+    sqrt,
+    sum as np_sum,
 )
 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 +128,190 @@ 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
+
+
+class _ExponentialWeightedFactor(SingleInputMixin, CustomFactor):
+    """
+    Base class for factors implementing exponential-weighted operations.
+
+    **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, ...
+
+    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):
+        """
+        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))),
+        )
+
+
+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=self.weights(len(data), decay_rate),
+        )
+
+
+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

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``.
     """

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

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: