wassname/catalyst
1
0
Fork 0
mirror of https://github.com/wassname/catalyst.git synced 2026-07-07 22:04:32 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #1046 from quantopian/classifiers

Browse Source 
Classifiers
This commit is contained in:
Scott Sanderson
2016-03-20 16:57:34 -04:00
parent bc09318ce3 396d2f4327
commit 14c1bb0cb5
22 changed files with 880 additions and 152 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/source/whatsnew/0.8.5.txt
+7
View File
@@ -32,6 +32,13 @@ Enhancements
factors use the new ``CashBuybackAuthorizations`` and
``ShareBuybackAuthorizations`` datasets, respectively. (:issue:`1022`).
* Added factor normalization methods:
:meth:`zipline.pipeline.Factor.demean` and
:meth:`zipline.pipeline.Factor.zscore`. (:issue:`1046`)
* Implemented :class:`zipline.pipeline.Classifier`, a new core pipeline API
term representing grouping keys. Classifiers are primarily used by passing
them as the ``groupby`` parameter to factor normalization methods.
Experimental Features
~~~~~~~~~~~~~~~~~~~~~
tests/pipeline/base.py
+18
View File
@@ -147,6 +147,24 @@ class BasePipelineTestCase(TestCase):
"""
return arange(prod(shape), dtype=dtype).reshape(shape)
@with_default_shape
def randn_data(self, seed, shape):
"""
Build a block of testing data from a seeded RandomState.
"""
return np.random.RandomState(seed).randn(*shape)
@with_default_shape
def eye_mask(self, shape):
"""
Build a mask using np.eye.
"""
return ~np.eye(*shape, dtype=bool)
@with_default_shape
def ones_mask(self, shape):
return np.ones(shape, dtype=bool)
class EventLoaderCommonMixin(object):
@abc.abstractproperty
tests/pipeline/test_factor.py
+151 -1
View File
@@ -5,19 +5,24 @@ from itertools import product
from nose_parameterized import parameterized
from numpy import (
apply_along_axis,
arange,
array,
datetime64,
empty,
eye,
nan,
nanmean,
nanstd,
ones,
where,
)
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.lib.normalize import naive_grouped_rowwise_apply as grouped_apply
from zipline.pipeline import Classifier, Factor, Filter, TermGraph
from zipline.pipeline.factors import (
Returns,
RSI,
@@ -43,6 +48,20 @@ class F(Factor):
window_length = 0
class C(Classifier):
dtype = int64_dtype
missing_value = -1
inputs = ()
window_length = 0
class OtherC(Classifier):
dtype = int64_dtype
missing_value = -1
inputs = ()
window_length = 0
class Mask(Filter):
inputs = ()
window_length = 0
@@ -403,3 +422,134 @@ class FactorTestCase(BasePipelineTestCase):
)
check_arrays(float_result, datetime_result)
@parameter_space(
seed_value=range(1, 2),
normalizer_name_and_func=[
('demean', lambda row: row - nanmean(row)),
('zscore', lambda row: (row - nanmean(row)) / nanstd(row)),
],
add_nulls_to_factor=(False, True,)
)
def test_normalizations(self,
seed_value,
normalizer_name_and_func,
add_nulls_to_factor):
name, func = normalizer_name_and_func
shape = (7, 7)
# All Trues.
nomask = self.ones_mask(shape=shape)
# Falses on main diagonal.
eyemask = self.eye_mask(shape=shape)
# Falses on other diagonal.
eyemask_T = eyemask.T
# Falses on both diagonals.
xmask = eyemask & eyemask_T
# Block of random data.
factor_data = self.randn_data(seed=seed_value, shape=shape)
if add_nulls_to_factor:
factor_data = where(eyemask, factor_data, nan)
# Cycles of 0, 1, 2, 0, 1, 2, ...
classifier_data = (
(self.arange_data(shape=shape, dtype=int) + seed_value) % 3
)
# With -1s on main diagonal.
classifier_data_eyenulls = where(eyemask, classifier_data, -1)
# With -1s on opposite diagonal.
classifier_data_eyenulls_T = where(eyemask_T, classifier_data, -1)
# With -1s on both diagonals.
classifier_data_xnulls = where(xmask, classifier_data, -1)
f = self.f
c = C()
c_with_nulls = OtherC()
m = Mask()
method = getattr(f, name)
terms = {
'vanilla': method(),
'masked': method(mask=m),
'grouped': method(groupby=c),
'grouped_with_nulls': method(groupby=c_with_nulls),
'both': method(mask=m, groupby=c),
'both_with_nulls': method(mask=m, groupby=c_with_nulls),
}
expected = {
'vanilla': apply_along_axis(func, 1, factor_data,),
'masked': where(
eyemask,
grouped_apply(factor_data, eyemask, func),
nan,
),
'grouped': grouped_apply(
factor_data,
classifier_data,
func,
),
# If the classifier has nulls, we should get NaNs in the
# corresponding locations in the output.
'grouped_with_nulls': where(
eyemask_T,
grouped_apply(factor_data, classifier_data_eyenulls_T, func),
nan,
),
# Passing a mask with a classifier should behave as though the
# classifier had nulls where the mask was False.
'both': where(
eyemask,
grouped_apply(
factor_data,
classifier_data_eyenulls,
func,
),
nan,
),
'both_with_nulls': where(
xmask,
grouped_apply(
factor_data,
classifier_data_xnulls,
func,
),
nan,
)
}
graph = TermGraph(terms)
results = self.run_graph(
graph,
initial_workspace={
f: factor_data,
c: classifier_data,
c_with_nulls: classifier_data_eyenulls_T,
Mask(): eyemask,
},
mask=self.build_mask(nomask),
)
for key in expected:
check_arrays(expected[key], results[key])
@parameter_space(method_name=['demean', 'zscore'])
def test_cant_normalize_non_float(self, method_name):
class DateFactor(Factor):
dtype = datetime64ns_dtype
inputs = ()
window_length = 0
d = DateFactor()
with self.assertRaises(TypeError) as e:
getattr(d, method_name)()
errmsg = str(e.exception)
expected = (
"{normalizer}() is only defined on Factors of dtype float64,"
" but it was called on a Factor of dtype datetime64[ns]."
).format(normalizer=method_name)
self.assertEqual(errmsg, expected)
tests/pipeline/test_numerical_expression.py
+6 -6
View File
@@ -153,7 +153,7 @@ class NumericalExpressionTestCase(TestCase):
message = e.exception.args[0]
expected = (
"Don't know how to compute datetime64[ns] + datetime64[ns].\n"
"Arithmetic operators are only supported on Factors of dtype "
"Arithmetic operators are only supported between Factors of dtype "
"'float64'."
)
self.assertEqual(message, expected)
@@ -164,7 +164,7 @@ class NumericalExpressionTestCase(TestCase):
message = e.exception.args[0]
expected = (
"Don't know how to compute datetime64[ns] * datetime64[ns].\n"
"Arithmetic operators are only supported on Factors of dtype "
"Arithmetic operators are only supported between Factors of dtype "
"'float64'."
)
self.assertEqual(message, expected)
@@ -178,8 +178,8 @@ class NumericalExpressionTestCase(TestCase):
message = e.exception.args[0]
expected = (
"Don't know how to compute float64 {sym} datetime64[ns].\n"
"Arithmetic operators are only supported on Factors of "
"dtype 'float64'."
"Arithmetic operators are only supported between Factors"
" of dtype 'float64'."
).format(sym=sym)
self.assertEqual(message, expected)
@@ -188,8 +188,8 @@ class NumericalExpressionTestCase(TestCase):
message = e.exception.args[0]
expected = (
"Don't know how to compute datetime64[ns] {sym} float64.\n"
"Arithmetic operators are only supported on Factors of "
"dtype 'float64'."
"Arithmetic operators are only supported between Factors"
" of dtype 'float64'."
).format(sym=sym)
self.assertEqual(message, expected)
tests/pipeline/test_term.py
+25 -2
View File
@@ -13,7 +13,7 @@ from zipline.errors import (
UnsupportedDType,
WindowLengthNotSpecified,
)
from zipline.pipeline import Factor, Filter, TermGraph
from zipline.pipeline import Classifier, Factor, Filter, TermGraph
from zipline.pipeline.data import Column, DataSet
from zipline.pipeline.data.testing import TestingDataSet
from zipline.pipeline.term import AssetExists, NotSpecified
@@ -343,11 +343,34 @@ class ObjectIdentityTestCase(TestCase):
with self.assertRaises(UnsupportedDType):
SomeFactor(dtype=complex128_dtype)
def test_require_super_call_in_validate(self):
class MyFactor(Factor):
inputs = ()
dtype = float64_dtype
window_length = 0
def _validate(self):
"Woops, I didn't call super()!"
with self.assertRaises(AssertionError) as e:
MyFactor()
errmsg = str(e.exception)
self.assertEqual(
errmsg,
"Term._validate() was not called.\n"
"This probably means that you overrode _validate"
" without calling super()."
)
def test_latest_on_different_dtypes(self):
factor_dtypes = (int64_dtype, float64_dtype, datetime64ns_dtype)
factor_dtypes = (float64_dtype, datetime64ns_dtype)
for column in TestingDataSet.columns:
if column.dtype == bool_dtype:
self.assertIsInstance(column.latest, Filter)
elif column.dtype == int64_dtype:
self.assertIsInstance(column.latest, Classifier)
elif column.dtype in factor_dtypes:
self.assertIsInstance(column.latest, Factor)
else:
tests/test_doctests.py
+4 -1
View File
@@ -4,7 +4,7 @@ import doctest
from unittest import TestCase
from zipline import testing
from zipline.lib import adjustment
from zipline.lib import adjustment, normalize
from zipline.pipeline import (
engine,
expression,
@@ -86,3 +86,6 @@ class DoctestTestCase(TestCase):
def test_functional_docs(self):
self._check_docs(functional)
def test_normalize_docs(self):
self._check_docs(normalize)
zipline/lib/normalize.py
+45
View File
@@ -0,0 +1,45 @@
import numpy as np
def naive_grouped_rowwise_apply(data, group_labels, func, out=None):
"""
Simple implementation of grouped row-wise function application.
Parameters
----------
data : ndarray[ndim=2]
Input array over which to apply a grouped function.
group_labels : ndarray[ndim=2, dtype=int64]
Labels to use to bucket inputs from array.
Should be the same shape as array.
func : function[ndarray[ndim=1]] -> function[ndarray[ndim=1]]
Function to apply to pieces of each row in array.
out : ndarray, optional
Array into which to write output. If not supplied, a new array of the
same shape as ``data`` is allocated and returned.
Example
-------
>>> data = np.array([[1., 2., 3.],
... [2., 3., 4.],
... [5., 6., 7.]])
>>> labels = np.array([[0, 0, 1],
... [0, 1, 0],
... [1, 0, 2]])
>>> naive_grouped_rowwise_apply(data, labels, lambda row: row - row.min())
array([[ 0., 1., 0.],
[ 0., 0., 2.],
[ 0., 0., 0.]])
>>> naive_grouped_rowwise_apply(data, labels, lambda row: row / row.sum())
array([[ 0.33333333, 0.66666667, 1. ],
[ 0.33333333, 1. , 0.66666667],
[ 1. , 1. , 1. ]])
"""
if out is None:
out = np.empty_like(data)
for (row, label_row, out_row) in zip(data, group_labels, out):
for label in np.unique(label_row):
locs = (label_row == label)
out_row[locs] = func(row[locs])
return out
zipline/pipeline/__init__.py
+4 -2
View File
@@ -1,10 +1,10 @@
from __future__ import print_function
from zipline.assets import AssetFinder
from .classifier import Classifier
from .classifiers import Classifier, CustomClassifier
from .engine import SimplePipelineEngine
from .factors import Factor, CustomFactor
from .filters import Filter
from .filters import Filter, CustomFilter
from .term import Term
from .graph import TermGraph
from .pipeline import Pipeline
@@ -53,6 +53,8 @@ def engine_from_files(daily_bar_path,
__all__ = (
'Classifier',
'CustomFactor',
'CustomFilter',
'CustomClassifier',
'engine_from_files',
'Factor',
'Filter',
zipline/pipeline/classifier.py
-9
View File
@@ -1,9 +0,0 @@
"""
classifier.py
"""
from zipline.pipeline.term import ComputableTerm
class Classifier(ComputableTerm):
pass
zipline/pipeline/classifiers/__init__.py
+8
View File
@@ -0,0 +1,8 @@
from .classifier import Classifier, CustomClassifier, Everything, Latest
__all__ = [
'Classifier',
'CustomClassifier',
'Everything',
'Latest',
]
zipline/pipeline/classifiers/classifier.py
+68
View File
@@ -0,0 +1,68 @@
"""
classifier.py
"""
from numpy import zeros, where
from zipline.pipeline.term import ComputableTerm
from zipline.utils.numpy_utils import int64_dtype
from ..mixins import (
CustomTermMixin,
LatestMixin,
PositiveWindowLengthMixin,
RestrictedDTypeMixin
)
class Classifier(RestrictedDTypeMixin, ComputableTerm):
"""
A Pipeline expression computing a categorical output.
Classifiers are most commonly useful for describing grouping keys for
complex transformations on Factor outputs. For example, Factor.demean() and
Factor.zscore() can be passed a Classifier in their ``groupby`` argument,
indicating that means/standard deviations should be computed on assets for
which the classifier produced the same label.
"""
ALLOWED_DTYPES = (int64_dtype,) # Used by RestrictedDTypeMixin
class Everything(Classifier):
"""
A trivial classifier that classifies everything the same.
"""
dtype = int64_dtype
window_length = 0
inputs = ()
missing_value = -1
def _compute(self, arrays, dates, assets, mask):
return where(
mask,
zeros(shape=mask.shape, dtype=int64_dtype),
self.missing_value,
)
class CustomClassifier(PositiveWindowLengthMixin, CustomTermMixin, Classifier):
"""
Base class for user-defined Classifiers.
See Also
--------
zipline.pipeline.CustomFactor
zipline.pipeline.CustomFilter
"""
pass
class Latest(LatestMixin, CustomClassifier):
"""
A classifier producing the latest value of an input.
See Also
--------
zipline.pipeline.data.dataset.BoundColumn.latest
zipline.pipeline.factors.factor.Latest
zipline.pipeline.filters.filter.Latest
"""
zipline/pipeline/data/dataset.py
+11 -4
View File
@@ -16,6 +16,7 @@ from zipline.pipeline.term import (
from zipline.utils.input_validation import ensure_dtype
from zipline.utils.numpy_utils import (
bool_dtype,
int64_dtype,
NoDefaultMissingValue,
)
from zipline.utils.preprocess import preprocess
@@ -93,16 +94,20 @@ class BoundColumn(LoadableTerm):
A column of data that's been concretely bound to a particular dataset.
Instances of this class are dynamically created upon access to attributes
of DataSets.
of DataSets (for example, USEquityPricing.close is an instance of this
class).
Attributes
----------
dtype : numpy.dtype
The dtype of data produced when this column is loaded.
latest : zipline.pipeline.data.Factor or zipline.pipeline.data.Filter
A Filter/Factor computing the most recently known value of this column
on each date. Produces a Filter if self.dtype == ``np.bool_``,
otherwise produces a Factor.
A Filter, Factor, or Classifier computing the most recently known value
of this column on each date.
Produces a Filter if self.dtype == ``np.bool_``.
Produces a Classifier if self.dtype == ``np.int64``
Otherwise produces a Factor.
dataset : zipline.pipeline.data.DataSet
The dataset to which this column is bound.
name : str
@@ -162,6 +167,8 @@ class BoundColumn(LoadableTerm):
def latest(self):
if self.dtype == bool_dtype:
from zipline.pipeline.filters import Latest
elif self.dtype == int64_dtype:
from zipline.pipeline.classifiers import Latest
else:
from zipline.pipeline.factors import Latest
return Latest(
zipline/pipeline/engine.py
+16 -15
View File
@@ -122,31 +122,32 @@ class SimplePipelineEngine(object):
The algorithm implemented here can be broken down into the following
stages:
0. Build a dependency graph of all terms in `terms`. Topologically
sort the graph to determine an order in which we can compute the terms.
0. Build a dependency graph of all terms in `pipeline`. Topologically
sort the graph to determine an order in which we can compute the
terms.
1. Ask our AssetFinder for a "lifetimes matrix", which should contain,
for each date between start_date and end_date, a boolean value for each
known asset indicating whether the asset existed on that date.
for each date between start_date and end_date, a boolean value for
each known asset indicating whether the asset existed on that date.
2. Compute each term in the dependency order determined in (0), caching
the results in a a dictionary to that they can be fed into future
terms.
the results in a a dictionary to that they can be fed into future
terms.
3. For each date, determine the number of assets passing **all**
filters. The sum, N, of all these values is the total number of rows in
our output frame, so we pre-allocate an output array of length N for
each factor in `terms`.
3. For each date, determine the number of assets passing
pipeline.screen. The sum, N, of all these values is the total
number of rows in our output frame, so we pre-allocate an output
array of length N for each factor in `terms`.
4. Fill in the arrays allocated in (3) by copying computed values from
our output cache into the corresponding rows.
our output cache into the corresponding rows.
5. Stick the values computed in (4) into a DataFrame and return it.
Step 0 is performed by `zipline.pipeline.graph.TermGraph`.
Step 1 is performed in `self._compute_root_mask`.
Step 2 is performed in `self.compute_chunk`.
Steps 3, 4, and 5 are performed in self._format_factor_matrix.
Step 0 is performed by ``Pipeline.to_graph``.
Step 1 is performed in ``SimplePipelineEngine._compute_root_mask``.
Step 2 is performed in ``SimplePipelineEngine.compute_chunk``.
Steps 3, 4, and 5 are performed in ``SimplePiplineEngine._to_narrow``.
See Also
--------
zipline/pipeline/expression.py
+1 -1
View File
@@ -223,7 +223,7 @@ class NumericalExpression(ComputableTerm):
expected_indices, expr_indices,
)
)
return super(NumericalExpression, self)._validate()
super(NumericalExpression, self)._validate()
def _compute(self, arrays, dates, assets, mask):
"""
zipline/pipeline/factors/__init__.py
+2 -2
View File
@@ -1,8 +1,8 @@
from .factor import (
Factor,
CustomFactor,
Factor,
Latest
)
from .latest import Latest
from .events import (
BusinessDaysSinceCashBuybackAuth,
BusinessDaysUntilNextEarnings,
zipline/pipeline/factors/factor.py
+388 -43
View File
@@ -5,20 +5,26 @@ from functools import wraps
from operator import attrgetter
from numbers import Number
from numpy import inf
from numpy import inf, where, nanstd
from toolz import curry
from zipline.errors import (
UnknownRankMethod,
UnsupportedDataType,
)
from zipline.errors import UnknownRankMethod
from zipline.lib.normalize import naive_grouped_rowwise_apply
from zipline.lib.rank import masked_rankdata_2d
from zipline.pipeline.classifiers import Classifier, Everything
from zipline.pipeline.mixins import (
CustomTermMixin,
LatestMixin,
PositiveWindowLengthMixin,
RestrictedDTypeMixin,
SingleInputMixin,
)
from zipline.pipeline.term import ComputableTerm, NotSpecified
from zipline.pipeline.term import (
ComputableTerm,
NotSpecified,
NotSpecifiedType,
Term,
)
from zipline.pipeline.expression import (
BadBinaryOperator,
COMPARISONS,
@@ -31,11 +37,13 @@ from zipline.pipeline.expression import (
unary_op_name,
)
from zipline.pipeline.filters import (
Filter,
NumExprFilter,
PercentileFilter,
NullFilter,
)
from zipline.utils.control_flow import nullctx
from zipline.utils.input_validation import expect_types
from zipline.utils.math_utils import nanmean
from zipline.utils.numpy_utils import (
bool_dtype,
coerce_to_dtype,
@@ -43,6 +51,7 @@ from zipline.utils.numpy_utils import (
float64_dtype,
int64_dtype,
)
from zipline.utils.preprocess import preprocess
_RANK_METHODS = frozenset(['average', 'min', 'max', 'dense', 'ordinal'])
@@ -140,7 +149,7 @@ def binop_return_dtype(op, left, right):
elif left != float64_dtype or right != float64_dtype:
raise TypeError(
"Don't know how to compute {left} {op} {right}.\n"
"Arithmetic operators are only supported on Factors of "
"Arithmetic operators are only supported between Factors of "
"dtype 'float64'.".format(
left=left.name,
op=op,
@@ -188,7 +197,7 @@ def binary_operator(op):
# inputs. Look up and call the appropriate reflected operator with
# ourself as the input.
return commuted_method_getter(other)(self)
elif isinstance(other, Factor):
elif isinstance(other, Term):
if self is other:
return return_type(
"x_0 {op} x_0".format(op=op),
@@ -204,7 +213,8 @@ def binary_operator(op):
return return_type(
"x_0 {op} ({constant})".format(op=op, constant=other),
binds=(self,),
# Interpret numeric literals as floats.
# .dtype access is safe here because coerce_numbers_to_my_dtype
# will convert any input numbers to numpy equivalents.
dtype=binop_return_dtype(op, self.dtype, other.dtype)
)
raise BadBinaryOperator(op, self, other)
@@ -318,35 +328,102 @@ def function_application(func):
return mathfunc
def if_not_float64_tell_caller_to_use_isnull(f):
def restrict_to_dtype(dtype, message_template):
"""
Factor method decorator that checks if self.dtype if float64.
A factory for decorators that restricting Factor methods to only be
callable on Factors with a specific dtype.
If the factor instance is of another dtype, this raises a TypeError
directing the user to `isnull` or `notnull` instead.
This is conceptually similar to
zipline.utils.input_validation.expect_dtypes, but provides more flexibility
for providing error messages that are specifically targeting Factor
methods.
Parameters
----------
dtype : numpy.dtype
The dtype on which the decorated method may be called.
message_template : str
A template for the error message to be raised.
`message_template.format` will be called with keyword arguments
`method_name`, `expected_dtype`, and `received_dtype`.
Usage
-----
@restrict_to_dtype(
dtype=float64_dtype,
message_template=(
"{method_name}() was called on a factor of dtype {received_dtype}."
"{method_name}() requires factors of dtype{expected_dtype}."
),
)
def some_factor_method(self, ...):
self.stuff_that_requires_being_float64(...)
"""
@wraps(f)
def wrapped_method(self):
if self.dtype != float64_dtype:
def processor(factor_method, _, factor_instance):
factor_dtype = factor_instance.dtype
if factor_dtype != dtype:
raise TypeError(
"{meth}() was called on a factor of dtype {dtype}.\n"
"{meth}() is only defined for dtype float64."
"To filter missing data, use isnull() or notnull().".format(
meth=f.__name__,
dtype=self.dtype,
),
message_template.format(
method_name=factor_method.__name__,
expected_dtype=dtype.name,
received_dtype=factor_dtype,
)
)
return f(self)
return wrapped_method
return factor_instance
return preprocess(self=processor)
# Decorators for Factor methods.
if_not_float64_tell_caller_to_use_isnull = restrict_to_dtype(
dtype=float64_dtype,
message_template=(
"{method_name}() was called on a factor of dtype {received_dtype}.\n"
"{method_name}() is only defined for dtype {expected_dtype}."
"To filter missing data, use isnull() or notnull()."
)
)
float64_only = restrict_to_dtype(
dtype=float64_dtype,
message_template=(
"{method_name}() is only defined on Factors of dtype {expected_dtype},"
" but it was called on a Factor of dtype {received_dtype}."
)
)
FACTOR_DTYPES = frozenset([datetime64ns_dtype, float64_dtype, int64_dtype])
class Factor(ComputableTerm):
class Factor(RestrictedDTypeMixin, ComputableTerm):
"""
Pipeline API expression producing numerically-valued outputs.
Pipeline API expression producing a numerical or date-valued output.
Factors are the most commonly-used Pipeline term, representing the result
of any computation producing a numerical result.
Factors can be combined, both with other Factors and with scalar values,
via any of the builtin mathematical operators (``+``, ``-``, ``*``, etc).
This makes it easy to write complex expressions that combine multiple
Factors. For example, constructing a Factor that computes the average of
two other Factors is simply::
>>> f1 = SomeFactor(...)
>>> f2 = SomeOtherFactor(...)
>>> average = (f1 + f2) / 2.0
Factors can also be converted into :class:`zipline.pipeline.Filter` objects
via comparison operators: (``<``, ``<=``, ``!=``, ``eq``, ``>``, ``>=``).
There are many natural operators defined on Factors besides the basic
numerical operators. These include methods identifying missing or
extreme-valued outputs (isnull, notnull, isnan, notnan), methods for
normalizing outputs (rank, demean, zscore), and methods for constructing
Filters based on rank-order properties of results (top, bottom,
percentile_between).
"""
ALLOWED_DTYPES = FACTOR_DTYPES # Used by RestrictedDTypeMixin
# Dynamically add functions for creating NumExprFactor/NumExprFilter
# instances.
clsdict = locals()
@@ -383,16 +460,189 @@ class Factor(ComputableTerm):
eq = binary_operator('==')
def _validate(self):
# Do superclass validation first so that `NotSpecified` dtypes get
# handled.
retval = super(Factor, self)._validate()
if self.dtype not in FACTOR_DTYPES:
raise UnsupportedDataType(
typename=type(self).__name__,
dtype=self.dtype
)
return retval
@expect_types(
mask=(Filter, NotSpecifiedType),
groupby=(Classifier, NotSpecifiedType),
)
@float64_only
def demean(self, mask=NotSpecified, groupby=NotSpecified):
"""
Construct a Factor that computes ``self`` and subtracts the mean from
row of the result.
If ``mask`` is supplied, ignore values where ``mask`` returns False
when computing row means, and output NaN anywhere the mask is False.
If ``groupby`` is supplied, compute by partitioning each row based on
the values produced by ``groupby``, de-meaning the partitioned arrays,
and stitching the sub-results back together.
Parameters
----------
mask : zipline.pipeline.Filter, optional
A Filter defining values to ignore when computing means.
groupby : zipline.pipeline.Classifier, optional
A classifier defining partitions over which to compute means.
Example
-------
Let ``f`` be a Factor which would produce the following output::
AAPL MSFT MCD BK
2017-03-13 1.0 2.0 3.0 4.0
2017-03-14 1.5 2.5 3.5 1.0
2017-03-15 2.0 3.0 4.0 1.5
2017-03-16 2.5 3.5 1.0 2.0
Let ``c`` be a Classifier producing the following output::
AAPL MSFT MCD BK
2017-03-13 1 1 2 2
2017-03-14 1 1 2 2
2017-03-15 1 1 2 2
2017-03-16 1 1 2 2
Let ``m`` be a Filter producing the following output::
AAPL MSFT MCD BK
2017-03-13 False True True True
2017-03-14 True False True True
2017-03-15 True True False True
2017-03-16 True True True False
Then ``f.demean()`` will subtract the mean from each row produced by
``f``.
::
AAPL MSFT MCD BK
2017-03-13 -1.500 -0.500 0.500 1.500
2017-03-14 -0.625 0.375 1.375 -1.125
2017-03-15 -0.625 0.375 1.375 -1.125
2017-03-16 0.250 1.250 -1.250 -0.250
``f.demean(mask=m)`` will subtract the mean from each row, but means
will be calculated ignoring values on the diagonal, and NaNs will
written to the diagonal in the output. Diagonal values are ignored
because they are the locations where the mask ``m`` produced False.
::
AAPL MSFT MCD BK
2017-03-13 NaN -1.000 0.000 1.000
2017-03-14 -0.500 NaN 1.500 -1.000
2017-03-15 -0.166 0.833 NaN -0.666
2017-03-16 0.166 1.166 -1.333 NaN
``f.demean(groupby=c)`` will subtract the group-mean of AAPL/MSFT and
MCD/BK from their respective entries. The AAPL/MSFT are grouped
together because both assets always produce 1 in the output of the
classifier ``c``. Similarly, MCD/BK are grouped together because they
always produce 2.
::
AAPL MSFT MCD BK
2017-03-13 -0.500 0.500 -0.500 0.500
2017-03-14 -0.500 0.500 1.250 -1.250
2017-03-15 -0.500 0.500 1.250 -1.250
2017-03-16 -0.500 0.500 -0.500 0.500
``f.demean(mask=m, groupby=c)`` will also subtract the group-mean of
AAPL/MSFT and MCD/BK, but means will be calculated ignoring values on
the diagonal , and NaNs will be written to the diagonal in the output.
::
AAPL MSFT MCD BK
2017-03-13 NaN 0.000 -0.500 0.500
2017-03-14 0.000 NaN 1.250 -1.250
2017-03-15 -0.500 0.500 NaN 0.000
2017-03-16 -0.500 0.500 0.000 NaN
Notes
-----
Mean is sensitive to the magnitudes of outliers. When working with
factor that can potentially produce large outliers, it is often useful
to use the ``mask`` parameter to discard values at the extremes of the
distribution::
>>> base = MyFactor(...)
>>> normalized = base.demean(mask=base.percentile_between(1, 99))
``demean()`` is only supported on Factors of dtype float64.
See Also
--------
:meth:`pandas.DataFrame.groupby`
"""
return GroupedRowTransform(
transform=lambda row: row - nanmean(row),
factor=self,
mask=mask,
groupby=groupby,
)
@expect_types(
mask=(Filter, NotSpecifiedType),
groupby=(Classifier, NotSpecifiedType),
)
@float64_only
def zscore(self, mask=NotSpecified, groupby=NotSpecified):
"""
Construct a Factor that Z-Scores each day's results.
The Z-Score of a row is defined as::
(row - row.mean()) / row.stddev()
If ``mask`` is supplied, ignore values where ``mask`` returns False
when computing row means and standard deviations, and output NaN
anywhere the mask is False.
If ``groupby`` is supplied, compute by partitioning each row based on
the values produced by ``groupby``, z-scoring the partitioned arrays,
and stitching the sub-results back together.
Parameters
----------
mask : zipline.pipeline.Filter, optional
A Filter defining values to ignore when Z-Scoring.
groupby : zipline.pipeline.Classifier, optional
A classifier defining partitions over which to compute Z-Scores.
Returns
-------
zscored : zipline.pipeline.Factor
A Factor producing that z-scores the output of self.
Notes
-----
Mean and standard deviation are sensitive to the magnitudes of
outliers. When working with factor that can potentially produce large
outliers, it is often useful to use the ``mask`` parameter to discard
values at the extremes of the distribution::
>>> base = MyFactor(...)
>>> normalized = base.zscore(mask=base.percentile_between(1, 99))
``zscore()`` is only supported on Factors of dtype float64.
Example
-------
See :meth:`~zipline.pipeline.factors.Factor.demean` for an in-depth
example of the semantics for ``mask`` and ``groupby``.
See Also
--------
:meth:`pandas.DataFrame.groupby`
"""
return GroupedRowTransform(
transform=lambda row: (row - nanmean(row)) / nanstd(row),
factor=self,
mask=mask,
groupby=groupby,
)
def rank(self, method='ordinal', ascending=True, mask=NotSpecified):
"""
@@ -430,9 +680,8 @@ class Factor(ComputableTerm):
See Also
--------
scipy.stats.rankdata
zipline.lib.rank.masked_rankdata_2d
zipline.pipeline.factors.factor.Rank
:func:`scipy.stats.rankdata`
:class:`zipline.pipeline.factors.factor.Rank`
"""
return Rank(self, method=method, ascending=ascending, mask=mask)
@@ -591,6 +840,90 @@ class NumExprFactor(NumericalExpression, Factor):
pass
class GroupedRowTransform(Factor):
"""
A Factor that transforms an input factor by applying a row-wise
shape-preserving transformation on classifier-defined groups of that
Factor.
This is most often useful for normalization operators like ``zscore`` or
``demean``.
Parameters
----------
transform : function[ndarray[ndim=1] -> ndarray[ndim=1]]
Function to apply over each row group.
factor : zipline.pipeline.Factor
The factor providing baseline data to transform.
mask : zipline.pipeline.Filter
Mask of entries to ignore when calculating transforms.
groupby : zipline.pipeline.Classifier
Classifier partitioning ``factor`` into groups to use when calculating
means.
Notes
-----
Users should rarely construct instances of this factor directly. Instead,
they should construct instances via factor normalization methods like
``zscore`` and ``demean``.
See Also
--------
zipline.pipeline.factors.Factor.zscore
zipline.pipeline.factors.Factor.demean
"""
window_length = 0
def __new__(cls, transform, factor, mask, groupby):
if mask is NotSpecified:
mask = factor.mask
else:
mask = mask & factor.mask
if groupby is NotSpecified:
groupby = Everything(mask=mask)
return super(GroupedRowTransform, cls).__new__(
GroupedRowTransform,
transform=transform,
inputs=(factor, groupby),
missing_value=factor.missing_value,
mask=mask,
dtype=factor.dtype,
)
def _init(self, transform, *args, **kwargs):
self._transform = transform
return super(GroupedRowTransform, self)._init(*args, **kwargs)
@classmethod
def static_identity(cls, transform, *args, **kwargs):
return (
super(GroupedRowTransform, cls).static_identity(*args, **kwargs),
transform,
)
def _compute(self, arrays, dates, assets, mask):
data = arrays[0]
null_group_value = self.inputs[1].missing_value
group_labels = where(
mask,
arrays[1],
null_group_value,
)
return where(
group_labels != null_group_value,
naive_grouped_rowwise_apply(
data=data,
group_labels=group_labels,
func=self._transform,
),
self.missing_value,
)
class Rank(SingleInputMixin, Factor):
"""
A Factor representing the row-wise rank data of another Factor.
@@ -606,8 +939,8 @@ class Rank(SingleInputMixin, Factor):
See Also
--------
scipy.stats.rankdata : Underlying ranking algorithm.
zipline.factors.Factor.rank : Method-style interface to same functionality.
:func:`scipy.stats.rankdata`
:class:`Factor.rank`
Notes
-----
@@ -777,4 +1110,16 @@ class CustomFactor(PositiveWindowLengthMixin, CustomTermMixin, Factor):
median_low15 = MedianValue([USEquityPricing.low], window_length=15)
'''
dtype = float64_dtype
ctx = nullctx()
class Latest(LatestMixin, CustomFactor):
"""
Factor producing the most recently-known value of `inputs[0]` on each day.
The `.latest` attribute of DataSet columns returns an instance of this
Factor.
"""
window_length = 1
def compute(self, today, assets, out, data):
out[:] = data[-1]
zipline/pipeline/factors/latest.py
-18
View File
@@ -1,18 +0,0 @@
"""
Factor that produces the most most recently-known value of Column.
"""
from .factor import CustomFactor
from ..mixins import SingleInputMixin
class Latest(SingleInputMixin, CustomFactor):
"""
Factor producing the most recently-known value of `inputs[0]` on each day.
The `.latest` attribute of DataSet columns returns an instance of this
Factor.
"""
window_length = 1
def compute(self, today, assets, out, data):
out[:] = data[-1]
zipline/pipeline/filters/__init__.py
+10 -3
View File
@@ -1,10 +1,17 @@
from .filter import Filter, NumExprFilter, NullFilter, PercentileFilter
from .latest import Latest
from .filter import (
CustomFilter,
Filter,
Latest,
NullFilter,
NumExprFilter,
PercentileFilter,
)
__all__ = [
'CustomFilter',
'Filter',
'Latest',
'NumExprFilter',
'NullFilter',
'NumExprFilter',
'PercentileFilter',
]
zipline/pipeline/filters/filter.py
+61 -7
View File
@@ -16,17 +16,18 @@ from zipline.errors import (
from zipline.lib.rank import ismissing
from zipline.pipeline.mixins import (
CustomTermMixin,
LatestMixin,
PositiveWindowLengthMixin,
RestrictedDTypeMixin,
SingleInputMixin,
)
from zipline.pipeline.term import ComputableTerm
from zipline.pipeline.term import ComputableTerm, Term
from zipline.pipeline.expression import (
BadBinaryOperator,
FILTER_BINOPS,
method_name_for_op,
NumericalExpression,
)
from zipline.utils.control_flow import nullctx
from zipline.utils.numpy_utils import bool_dtype
@@ -67,7 +68,9 @@ def binary_operator(op):
# merging of inputs. Look up and call the appropriate
# right-binding operator with ourself as the input.
return commuted_method_getter(other)(self)
elif isinstance(other, Filter):
elif isinstance(other, Term):
if other.dtype != bool_dtype:
raise BadBinaryOperator(op, self, other)
if self is other:
return NumExprFilter.create(
"x_0 {op} x_0".format(op=op),
@@ -112,10 +115,55 @@ def unary_operator(op):
return unary_operator
class Filter(ComputableTerm):
class Filter(RestrictedDTypeMixin, ComputableTerm):
"""
Pipeline API expression producing boolean-valued outputs.
Pipeline expression computing a boolean output.
Filters are most commonly useful for describing sets of assets to include
or exclude for some particular purpose. Many Pipeline API functions accept
a ``mask`` argument, which can be supplied a Filter indicating that only
values passing the Filter should be considered when performing the
requested computation. For example, :meth:`zipline.pipeline.Factor.top`
accepts a mask indicating that ranks should be computed only on assets that
passed the specified Filter.
The most common way to construct a Filter is via one of the comparison
operators (``<``, ``<=``, ``!=``, ``eq``, ``>``, ``>=``) of
:class:`~zipline.pipeline.Factor`. For example, a natural way to construct
a Filter for stocks with a 10-day VWAP less than $20.0 is to first
construct a Factor computing 10-day VWAP and compare it to the scalar value
20.0::
>>> from zipline.pipeline.factors import VWAP
>>> vwap_10 = VWAP(window_length=10)
>>> vwaps_under_20 = (vwap_10 <= 20)
Filters can also be constructed via comparisons between two Factors. For
example, to construct a Filter producing True for asset/date pairs where
the asset's 10-day VWAP was greater than it's 30-day VWAP::
>>> short_vwap = VWAP(window_length=10)
>>> long_vwap = VWAP(window_length=30)
>>> higher_short_vwap = (short_vwap > long_vwap)
Filters can be combined via the ``&`` (and) and ``|`` (or) operators.
``&``-ing together two filters produces a new Filter that produces True if
**both** of the inputs produced True.
``|``-ing together two filters produces a new Filter that produces True if
**either** of its inputs produced True.
The ``~`` operator can be used to invert a Filter, swapping all True values
with Falses and vice-versa.
Filters may be set as the ``screen`` attribute of a Pipeline, indicating
asset/date pairs for which the filter produces False should be excluded
from the Pipeline's output. This is useful both for reducing noise in the
output of a Pipeline and for reducing memory consumption of Pipeline
results.
"""
ALLOWED_DTYPES = (bool_dtype,) # Used by RestrictedDTypeMixin
dtype = bool_dtype
clsdict = locals()
@@ -180,7 +228,7 @@ class NullFilter(SingleInputMixin, Filter):
Parameters
----------
factor : zipline.pipeline.factor.Factor
factor : zipline.pipeline.Factor
The factor to compare against its missing_value.
"""
window_length = 0
@@ -322,4 +370,10 @@ class CustomFilter(PositiveWindowLengthMixin, CustomTermMixin, Filter):
--------
zipline.pipeline.factors.factor.CustomFactor
"""
ctx = nullctx()
class Latest(LatestMixin, CustomFilter):
"""
Filter producing the most recently-known value of `inputs[0]` on each day.
"""
pass
zipline/pipeline/filters/latest.py
-29
View File
@@ -1,29 +0,0 @@
"""
Filter that produces the most most recently-known value of a boolean-valued
Column.
"""
from zipline.utils.numpy_utils import bool_dtype
from .filter import CustomFilter
from ..mixins import SingleInputMixin
class Latest(SingleInputMixin, CustomFilter):
"""
Filter producing the most recently-known value of `inputs[0]` on each day.
"""
window_length = 1
def compute(self, today, assets, out, data):
out[:] = data[-1]
def _validate(self):
if self.inputs[0].dtype != bool_dtype:
raise TypeError(
"{name} expected an input of dtype bool, "
"but got {not_bool} instead.".format(
name=type(self).__name__,
not_bool=self.inputs[0].dtype,
)
)
super(Latest, self)._validate()
zipline/pipeline/mixins.py
+49 -3
View File
@@ -2,7 +2,9 @@
Mixins classes for use with Filters and Factors.
"""
from numpy import full_like
from zipline.errors import WindowLengthNotPositive
from zipline.utils.control_flow import nullctx
from zipline.errors import WindowLengthNotPositive, UnsupportedDataType
from .term import NotSpecified
@@ -12,9 +14,9 @@ class PositiveWindowLengthMixin(object):
Validation mixin enforcing that a Term gets a positive WindowLength
"""
def _validate(self):
super(PositiveWindowLengthMixin, self)._validate()
if not self.windowed:
raise WindowLengthNotPositive(window_length=self.window_length)
return super(PositiveWindowLengthMixin, self)._validate()
class SingleInputMixin(object):
@@ -22,6 +24,7 @@ class SingleInputMixin(object):
Validation mixin enforcing that a Term gets a length-1 inputs list.
"""
def _validate(self):
super(SingleInputMixin, self)._validate()
num_inputs = len(self.inputs)
if num_inputs != 1:
raise ValueError(
@@ -31,7 +34,26 @@ class SingleInputMixin(object):
num_inputs=num_inputs
)
)
return super(SingleInputMixin, self)._validate()
class RestrictedDTypeMixin(object):
"""
Validation mixin enforcing that a term has a specific dtype.
"""
ALLOWED_DTYPES = NotSpecified
def _validate(self):
super(RestrictedDTypeMixin, self)._validate()
assert self.ALLOWED_DTYPES is not NotSpecified, (
"ALLOWED_DTYPES not supplied on subclass "
"of RestrictedDTypeMixin: %s." % type(self).__name__
)
if self.dtype not in self.ALLOWED_DTYPES:
raise UnsupportedDataType(
typename=type(self.__name__),
dtype=self.dtype,
)
class CustomTermMixin(object):
@@ -43,6 +65,8 @@ class CustomTermMixin(object):
Used by CustomFactor, CustomFilter, CustomClassifier, etc.
"""
ctx = nullctx()
def __new__(cls,
inputs=NotSpecified,
window_length=NotSpecified,
@@ -101,3 +125,25 @@ class CustomTermMixin(object):
def short_repr(self):
return type(self).__name__ + '(%d)' % self.window_length
class LatestMixin(SingleInputMixin):
"""
Mixin for behavior shared by Custom{Factor,Filter,Classifier}.
"""
window_length = 1
def compute(self, today, assets, out, data):
out[:] = data[-1]
def _validate(self):
super(LatestMixin, self)._validate()
if self.inputs[0].dtype != self.dtype:
raise TypeError(
"{name} expected an input of dtype {expected}, "
"but got {actual} instead.".format(
name=type(self).__name__,
expected=self.dtype,
actual=self.inputs[0].dtype,
)
)
zipline/pipeline/term.py
+6 -6
View File
@@ -28,6 +28,8 @@ NotSpecified = sentinel(
'Singleton sentinel value used for Term defaults.',
)
NotSpecifiedType = type(NotSpecified)
class Term(with_metaclass(ABCMeta, object)):
"""
@@ -396,14 +398,14 @@ class ComputableTerm(Term):
)
def _validate(self):
"""
Assert that this term is well-formed. This should be called exactly
once, at the end of Term._init().
"""
super(ComputableTerm, self)._validate()
if self.inputs is NotSpecified:
raise TermInputsNotSpecified(termname=type(self).__name__)
if self.window_length is NotSpecified:
raise WindowLengthNotSpecified(termname=type(self).__name__)
if self.mask is NotSpecified:
# This isn't user error, this is a bug in our code.
raise AssertionError("{term} has no mask".format(term=self))
@@ -413,8 +415,6 @@ class ComputableTerm(Term):
if child.windowed:
raise WindowedInputToWindowedTerm(parent=self, child=child)
return super(ComputableTerm, self)._validate()
def _compute(self, inputs, dates, assets, mask):
"""
Subclasses should implement this to perform actual computation.