zipline.utils.ensure_timezone() - Code Metrics - Inspection of "allows users to specify the cutoff time for data q..." - quantopian/zipline - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#947)

by Joe

created 2016-01-09 01:36 UTC

zipline.utils.ensure_timezone() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	3
dl	0
loc	23
rs	9.0857

# Copyright 2015 Quantopian, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from datetime import tzinfo
from operator import attrgetter

from numpy import dtype
from pytz import timezone
from six import iteritems, string_types, PY3
from toolz import valmap, complement, compose
import toolz.curried.operator as op

from zipline.utils.preprocess import preprocess


def ensure_upper_case(func, argname, arg):
    if isinstance(arg, string_types):
        return arg.upper()
    else:
        raise TypeError(
            "{0}() expected argument '{1}' to"
            " be a string, but got {2} instead.".format(
                func.__name__, argname, arg,)
        )


def ensure_dtype(func, argname, arg):
    """
    Argument preprocessor that converts the input into a numpy dtype.

    Usage
    -----
    >>> import numpy as np
    >>> from zipline.utils.preprocess import preprocess
    >>> @preprocess(dtype=ensure_dtype)
    ... def foo(dtype):
    ...     return dtype
    ...
    >>> foo(float)
    dtype('float64')
    """
    try:
        return dtype(arg)
    except TypeError:
        raise TypeError(
            "{func}() couldn't convert argument "
            "{argname}={arg!r} to a numpy dtype.".format(
                func=_qualified_name(func),
                argname=argname,
                arg=arg,
            ),
        )


def ensure_timezone(func, argname, arg):
    """Argument preprocessor that converts the input into a tzinfo object.

    Usage
    -----
    >>> from zipline.utils.preprocess import preprocess
    >>> @preprocess(tz=ensure_timezone)
    ... def foo(tz):
    ...     return tz
    >>> foo('utc')
    <UTC>
    """
    if isinstance(arg, tzinfo):
        return arg
    if isinstance(arg, string_types):
        return timezone(arg)

    raise TypeError(
        "{func}() couldn't convert argument "
        "{argname}={arg!r} to a timezone.".format(
            func=_qualified_name(func),
            argname=argname,
            arg=arg,
        ),
    )


def expect_dtypes(*_pos, **named):
    """
    Preprocessing decorator that verifies inputs have expected numpy dtypes.

    Usage
    -----
    >>> from numpy import dtype, arange
    >>> @expect_dtypes(x=dtype(int))
    ... def foo(x, y):
    ...    return x, y
    ...
    >>> foo(arange(3), 'foo')
    (array([0, 1, 2]), 'foo')
    >>> foo(arange(3, dtype=float), 'foo')
    Traceback (most recent call last):
       ...
    TypeError: foo() expected an argument with dtype 'int64' for argument 'x', but got dtype 'float64' instead.  # noqa
    """
    if _pos:
        raise TypeError("expect_dtypes() only takes keyword arguments.")

    for name, type_ in iteritems(named):
        if not isinstance(type_, (dtype, tuple)):
            raise TypeError(
                "expect_dtypes() expected a numpy dtype or tuple of dtypes"
                " for argument {name!r}, but got {dtype} instead.".format(
                    name=name, dtype=dtype,
                )
            )
    return preprocess(**valmap(_expect_dtype, named))


def _expect_dtype(_dtype_or_dtype_tuple):
    """
    Factory for dtype-checking functions that work the @preprocess decorator.
    """
    # Slightly different messages for dtype and tuple of dtypes.
    if isinstance(_dtype_or_dtype_tuple, tuple):
        allowed_dtypes = _dtype_or_dtype_tuple
    else:
        allowed_dtypes = (_dtype_or_dtype_tuple,)
    template = (
        "%(funcname)s() expected a value with dtype {dtype_str} "
        "for argument '%(argname)s', but got %(actual)r instead."
    ).format(dtype_str=' or '.join(repr(d.name) for d in allowed_dtypes))

    def check_dtype(value):
        return getattr(value, 'dtype', None) not in allowed_dtypes

    def display_bad_value(value):
        # If the bad value has a dtype, but it's wrong, show the dtype name.
        try:
            return value.dtype.name
        except AttributeError:
            return value

    return make_check(
        exc_type=TypeError,
        template=template,
        pred=check_dtype,
        actual=display_bad_value,
    )


def expect_types(*_pos, **named):
    """
    Preprocessing decorator that verifies inputs have expected types.

    Usage
    -----
    >>> @expect_types(x=int, y=str)
    ... def foo(x, y):
    ...    return x, y
    ...
    >>> foo(2, '3')
    (2, '3')
    >>> foo(2.0, '3')
    Traceback (most recent call last):
       ...
    TypeError: foo() expected an argument of type 'int' for argument 'x', but got float instead.  # noqa
    """
    if _pos:
        raise TypeError("expect_types() only takes keyword arguments.")

    for name, type_ in iteritems(named):
        if not isinstance(type_, (type, tuple)):
            raise TypeError(
                "expect_types() expected a type or tuple of types for "
                "argument '{name}', but got {type_} instead.".format(
                    name=name, type_=type_,
                )
            )

    return preprocess(**valmap(_expect_type, named))


if PY3:
    _qualified_name = attrgetter('__qualname__')
else:
    def _qualified_name(obj):
        """
        Return the fully-qualified name (ignoring inner classes) of a type.
        """
        module = obj.__module__
        if module in ('__builtin__', '__main__', 'builtins'):
            return obj.__name__
        return '.'.join([module, obj.__name__])


def make_check(exc_type, template, pred, actual):
    """
    Factory for making preprocessing functions that check a predicate on the
    input value.

    Parameters
    ----------
    exc_type : Exception
        The exception type to raise if the predicate fails.
    template : str
        A template string to use to create error messages.
        Should have %-style named template parameters for 'funcname',
        'argname', and 'actual'.
    pred : function[object -> bool]
        A function to call on the argument being preprocessed.  If the
        predicate returns `True`, we raise an instance of `exc_type`.
    actual : function[object -> object]
        A function to call on bad values to produce the value to display in the
        error message.
    """

    def _check(func, argname, argvalue):
        if pred(argvalue):
            raise exc_type(
                template % {
                    'funcname': _qualified_name(func),
                    'argname': argname,
                    'actual': actual(argvalue),
                },
            )
        return argvalue
    return _check


def _expect_type(type_):
    """
    Factory for type-checking functions that work the @preprocess decorator.
    """
    # Slightly different messages for type and tuple of types.
    _template = (
        "%(funcname)s() expected a value of type {type_or_types} "
        "for argument '%(argname)s', but got %(actual)s instead."
    )
    if isinstance(type_, tuple):
        template = _template.format(
            type_or_types=' or '.join(map(_qualified_name, type_))
        )
    else:
        template = _template.format(type_or_types=_qualified_name(type_))

    return make_check(
        TypeError,
        template,
        lambda v: not isinstance(v, type_),
        compose(_qualified_name, type),
    )


def optional(type_):
    """
    Helper for use with `expect_types` when an input can be `type_` or `None`.

    Returns an object such that both `None` and instances of `type_` pass
    checks of the form `isinstance(obj, optional(type_))`.

    Parameters
    ----------
    type_ : type
       Type for which to produce an option.

    Examples
    --------
    >>> isinstance({}, optional(dict))
    True
    >>> isinstance(None, optional(dict))
    True
    >>> isinstance(1, optional(dict))
    False
    """
    return (type_, type(None))


def expect_element(*_pos, **named):
    """
    Preprocessing decorator that verifies inputs are elements of some
    expected collection.

    Usage
    -----
    >>> @expect_element(x=('a', 'b'))
    ... def foo(x):
    ...    return x.upper()
    ...
    >>> foo('a')
    'A'
    >>> foo('b')
    'B'
    >>> foo('c')
    Traceback (most recent call last):
       ...
    ValueError: foo() expected a value in ('a', 'b') for argument 'x', but got 'c' instead.  # noqa

    Notes
    -----
    This uses the `in` operator (__contains__) to make the containment check.
    This allows us to use any custom container as long as the object supports
    the container protocol.
    """
    if _pos:
        raise TypeError("expect_element() only takes keyword arguments.")

    return preprocess(**valmap(_expect_element, named))


def _expect_element(collection):
    template = (
        "%(funcname)s() expected a value in {collection} "
        "for argument '%(argname)s', but got %(actual)s instead."
    ).format(collection=collection)
    return make_check(
        ValueError,
        template,
        complement(op.contains(collection)),
        repr,
    )


1			# Copyright 2015 Quantopian, Inc.
2			#
3			# Licensed under the Apache License, Version 2.0 (the "License");
4			# you may not use this file except in compliance with the License.
5			# You may obtain a copy of the License at
6			#
7			# http://www.apache.org/licenses/LICENSE-2.0
8			#
9			# Unless required by applicable law or agreed to in writing, software
10			# distributed under the License is distributed on an "AS IS" BASIS,
11			# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12			# See the License for the specific language governing permissions and
13			# limitations under the License.
14			from datetime import tzinfo
15			from operator import attrgetter
16
17			from numpy import dtype
18			from pytz import timezone
19			from six import iteritems, string_types, PY3
20			from toolz import valmap, complement, compose
21			import toolz.curried.operator as op
22
23			from zipline.utils.preprocess import preprocess
24
25
26			def ensure_upper_case(func, argname, arg):
27			if isinstance(arg, string_types):
28			return arg.upper()
29			else:
30			raise TypeError(
31			"{0}() expected argument '{1}' to"
32			" be a string, but got {2} instead.".format(
33			func.__name__, argname, arg,)
34			)
35
36
37			def ensure_dtype(func, argname, arg):
38			"""
39			Argument preprocessor that converts the input into a numpy dtype.
40
41			Usage
42			-----
43			>>> import numpy as np
44			>>> from zipline.utils.preprocess import preprocess
45			>>> @preprocess(dtype=ensure_dtype)
46			... def foo(dtype):
47			... return dtype
48			...
49			>>> foo(float)
50			dtype('float64')
51			"""
52			try:
53			return dtype(arg)
54			except TypeError:
55			raise TypeError(
56			"{func}() couldn't convert argument "
57			"{argname}={arg!r} to a numpy dtype.".format(
58			func=_qualified_name(func),
59			argname=argname,
60			arg=arg,
61			),
62			)
63
64
65			def ensure_timezone(func, argname, arg):
66			"""Argument preprocessor that converts the input into a tzinfo object.
67
68			Usage
69			-----
70			>>> from zipline.utils.preprocess import preprocess
71			>>> @preprocess(tz=ensure_timezone)
72			... def foo(tz):
73			... return tz
74			>>> foo('utc')
75			<UTC>
76			"""
77			if isinstance(arg, tzinfo):
78			return arg
79			if isinstance(arg, string_types):
80			return timezone(arg)
81
82			raise TypeError(
83			"{func}() couldn't convert argument "
84			"{argname}={arg!r} to a timezone.".format(
85			func=_qualified_name(func),
86			argname=argname,
87			arg=arg,
88			),
89			)
90
91
92			def expect_dtypes(_pos, *named):
93			"""
94			Preprocessing decorator that verifies inputs have expected numpy dtypes.
95
96			Usage
97			-----
98			>>> from numpy import dtype, arange
99			>>> @expect_dtypes(x=dtype(int))
100			... def foo(x, y):
101			... return x, y
102			...
103			>>> foo(arange(3), 'foo')
104			(array([0, 1, 2]), 'foo')
105			>>> foo(arange(3, dtype=float), 'foo')
106			Traceback (most recent call last):
107			...
108			TypeError: foo() expected an argument with dtype 'int64' for argument 'x', but got dtype 'float64' instead. # noqa
109			"""
110			if _pos:
111			raise TypeError("expect_dtypes() only takes keyword arguments.")
112
113			for name, type_ in iteritems(named):
114			if not isinstance(type_, (dtype, tuple)):
115			raise TypeError(
116			"expect_dtypes() expected a numpy dtype or tuple of dtypes"
117			" for argument {name!r}, but got {dtype} instead.".format(
118			name=name, dtype=dtype,
119			)
120			)
121			return preprocess(**valmap(_expect_dtype, named))
122
123
124			def _expect_dtype(_dtype_or_dtype_tuple):
125			"""
126			Factory for dtype-checking functions that work the @preprocess decorator.
127			"""
128			# Slightly different messages for dtype and tuple of dtypes.
129			if isinstance(_dtype_or_dtype_tuple, tuple):
130			allowed_dtypes = _dtype_or_dtype_tuple
131			else:
132			allowed_dtypes = (_dtype_or_dtype_tuple,)
133			template = (
134			"%(funcname)s() expected a value with dtype {dtype_str} "
135			"for argument '%(argname)s', but got %(actual)r instead."
136			).format(dtype_str=' or '.join(repr(d.name) for d in allowed_dtypes))
137
138			def check_dtype(value):
139			return getattr(value, 'dtype', None) not in allowed_dtypes
140
141			def display_bad_value(value):
142			# If the bad value has a dtype, but it's wrong, show the dtype name.
143			try:
144			return value.dtype.name
145			except AttributeError:
146			return value
147
148			return make_check(
149			exc_type=TypeError,
150			template=template,
151			pred=check_dtype,
152			actual=display_bad_value,
153			)
154
155
156			def expect_types(_pos, *named):
157			"""
158			Preprocessing decorator that verifies inputs have expected types.
159
160			Usage
161			-----
162			>>> @expect_types(x=int, y=str)
163			... def foo(x, y):
164			... return x, y
165			...
166			>>> foo(2, '3')
167			(2, '3')
168			>>> foo(2.0, '3')
169			Traceback (most recent call last):
170			...
171			TypeError: foo() expected an argument of type 'int' for argument 'x', but got float instead. # noqa
172			"""
173			if _pos:
174			raise TypeError("expect_types() only takes keyword arguments.")
175
176			for name, type_ in iteritems(named):
177			if not isinstance(type_, (type, tuple)):
178			raise TypeError(
179			"expect_types() expected a type or tuple of types for "
180			"argument '{name}', but got {type_} instead.".format(
181			name=name, type_=type_,
182			)
183			)
184
185			return preprocess(**valmap(_expect_type, named))
186
187
188			if PY3:
189			_qualified_name = attrgetter('__qualname__')
190			else:
191			def _qualified_name(obj):
192			"""
193			Return the fully-qualified name (ignoring inner classes) of a type.
194			"""
195			module = obj.__module__
196			if module in ('__builtin__', '__main__', 'builtins'):
197			return obj.__name__
198			return '.'.join([module, obj.__name__])
199
200
201			def make_check(exc_type, template, pred, actual):
202			"""
203			Factory for making preprocessing functions that check a predicate on the
204			input value.
205
206			Parameters
207			----------
208			exc_type : Exception
209			The exception type to raise if the predicate fails.
210			template : str
211			A template string to use to create error messages.
212			Should have %-style named template parameters for 'funcname',
213			'argname', and 'actual'.
214			pred : function[object -> bool]
215			A function to call on the argument being preprocessed. If the
216			predicate returns `True`, we raise an instance of `exc_type`.
217			actual : function[object -> object]
218			A function to call on bad values to produce the value to display in the
219			error message.
220			"""
221
222			def _check(func, argname, argvalue):
223			if pred(argvalue):
224			raise exc_type(
225			template % {
226			'funcname': _qualified_name(func),
227			'argname': argname,
228			'actual': actual(argvalue),
229			},
230			)
231			return argvalue
232			return _check
233
234
235			def _expect_type(type_):
236			"""
237			Factory for type-checking functions that work the @preprocess decorator.
238			"""
239			# Slightly different messages for type and tuple of types.
240			_template = (
241			"%(funcname)s() expected a value of type {type_or_types} "
242			"for argument '%(argname)s', but got %(actual)s instead."
243			)
244			if isinstance(type_, tuple):
245			template = _template.format(
246			type_or_types=' or '.join(map(_qualified_name, type_))
247			)
248			else:
249			template = _template.format(type_or_types=_qualified_name(type_))
250
251			return make_check(
252			TypeError,
253			template,
254			lambda v: not isinstance(v, type_),
255			compose(_qualified_name, type),
256			)
257
258
259			def optional(type_):
260			"""
261			Helper for use with `expect_types` when an input can be `type_` or `None`.
262
263			Returns an object such that both `None` and instances of `type_` pass
264			checks of the form `isinstance(obj, optional(type_))`.
265
266			Parameters
267			----------
268			type_ : type
269			Type for which to produce an option.
270
271			Examples
272			--------
273			>>> isinstance({}, optional(dict))
274			True
275			>>> isinstance(None, optional(dict))
276			True
277			>>> isinstance(1, optional(dict))
278			False
279			"""
280			return (type_, type(None))
281
282
283			def expect_element(_pos, *named):
284			"""
285			Preprocessing decorator that verifies inputs are elements of some
286			expected collection.
287
288			Usage
289			-----
290			>>> @expect_element(x=('a', 'b'))
291			... def foo(x):
292			... return x.upper()
293			...
294			>>> foo('a')
295			'A'
296			>>> foo('b')
297			'B'
298			>>> foo('c')
299			Traceback (most recent call last):
300			...
301			ValueError: foo() expected a value in ('a', 'b') for argument 'x', but got 'c' instead. # noqa
302
303			Notes
304			-----
305			This uses the `in` operator (__contains__) to make the containment check.
306			This allows us to use any custom container as long as the object supports
307			the container protocol.
308			"""
309			if _pos:
310			raise TypeError("expect_element() only takes keyword arguments.")
311
312			return preprocess(**valmap(_expect_element, named))
313
314
315			def _expect_element(collection):
316			template = (
317			"%(funcname)s() expected a value in {collection} "
318			"for argument '%(argname)s', but got %(actual)s instead."
319			).format(collection=collection)
320			return make_check(
321			ValueError,
322			template,
323			complement(op.contains(collection)),
324			repr,
325			)
326

quantopian / zipline

Pull Request — master (#947)

zipline.utils.ensure_timezone() A

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like