zipline.pipeline.factors.DollarVolume() - Code Metrics - Inspection of "ENH: Change DollarVolume to AverageDollarVolume." - quantopian/zipline - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#927)

unknown

created 2015-12-28 21:12 UTC

zipline.pipeline.factors.DollarVolume() A

↳ Parent: zipline.pipeline.factors.AverageDollarVolume

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	1
dl	0
loc	6
rs	9.4286

"""
Technical Analysis Factors
--------------------------
"""
from bottleneck import (
    nanargmax,
    nanmax,
    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.mixins import SingleInputMixin
from zipline.utils.control_flow import ignore_nanwarnings
from zipline.utils.input_validation import expect_types
from .factor import CustomFactor


class Returns(CustomFactor):
    """
    Calculates the percent change in close price over the given window_length.

    **Default Inputs**: [USEquityPricing.close]
    """
    inputs = [USEquityPricing.close]

    def compute(self, today, assets, out, close):
        out[:] = (close[-1] - close[0]) / close[0]


class RSI(CustomFactor, SingleInputMixin):
    """
    Relative Strength Index

    **Default Inputs**: [USEquityPricing.close]

    **Default Window Length**: 15
    """
    window_length = 15
    inputs = (USEquityPricing.close,)

    def compute(self, today, assets, out, closes):
        diffs = diff(closes, axis=0)
        ups = nanmean(clip(diffs, 0, inf), axis=0)
        downs = abs(nanmean(clip(diffs, -inf, 0), axis=0))
        return evaluate(
            "100 - (100 / (1 + (ups / downs)))",
            local_dict={'ups': ups, 'downs': downs},
            global_dict={},
            out=out,
        )


class SimpleMovingAverage(CustomFactor, SingleInputMixin):
    """
    Average Value of an arbitrary column

    **Default Inputs**: None

    **Default Window Length**: None
    """
    # numpy's nan functions throw warnings when passed an array containing only
    # nans, but they still returns the desired value (nan), so we ignore the
    # warning.
    ctx = ignore_nanwarnings()

    def compute(self, today, assets, out, data):
        out[:] = nanmean(data, axis=0)


class WeightedAverageValue(CustomFactor):
    """
    Helper for VWAP-like computations.

    **Default Inputs:** None

    **Default Window Length:** None
    """
    def compute(self, today, assets, out, base, weight):
        out[:] = nansum(base * weight, axis=0) / nansum(weight, axis=0)


class VWAP(WeightedAverageValue):
    """
    Volume Weighted Average Price

    **Default Inputs:** [USEquityPricing.close, USEquityPricing.volume]

    **Default Window Length:** None
    """
    inputs = (USEquityPricing.close, USEquityPricing.volume)


class MaxDrawdown(CustomFactor, SingleInputMixin):
    """
    Max Drawdown

    **Default Inputs:** None

    **Default Window Length:** None
    """
    ctx = ignore_nanwarnings()

    def compute(self, today, assets, out, data):
        drawdowns = fmax.accumulate(data, axis=0) - data
        drawdowns[isnan(drawdowns)] = NINF
        drawdown_ends = nanargmax(drawdowns, axis=0)

        # TODO: Accelerate this loop in Cython or Numba.
        for i, end in enumerate(drawdown_ends):
            peak = nanmax(data[:end + 1, i])
            out[i] = (peak - data[end, i]) / data[end, i]


class AverageDollarVolume(CustomFactor):
    """
    Average Daily Dollar Volume

    **Default Inputs:** [USEquityPricing.close, USEquityPricing.volume]

    **Default Window Length:** None
    """
    inputs = [USEquityPricing.close, USEquityPricing.volume]

    def compute(self, today, assets, out, close, volume):
        out[:] = nanmean(close * volume, axis=0)


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.

        Example
        -------
        .. code-block:: python

            # Equivalent to:
            # my_ewma = EWMA(
            #    inputs=[USEquityPricing.close],
            #    window_length=30,
            #    decay_rate=(1 - (2.0 / (1 + 15.0))),
            # )
            my_ewma = EWMA.from_span(
                inputs=[USEquityPricing.close],
                window_length=30,
                span=15,
            )

        Note
        ----
        This classmethod is provided by both
        :class:`ExponentialWeightedMovingAverage` and
        :class:`ExponentialWeightedMovingStdDev`.
        """
        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.

        Example
        -------
        .. code-block:: python

            # Equivalent to:
            # my_ewma = EWMA(
            #    inputs=[USEquityPricing.close],
            #    window_length=30,
            #    decay_rate=np.exp(np.log(0.5) / 15),
            # )
            my_ewma = EWMA.from_halflife(
                inputs=[USEquityPricing.close],
                window_length=30,
                halflife=15,
            )

        Note
        ----
        This classmethod is provided by both
        :class:`ExponentialWeightedMovingAverage` and
        :class:`ExponentialWeightedMovingStdDev`.
        """
        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 / 1 + center_of_mass)`.  This provides
        behavior equivalent to passing `center_of_mass` to pandas.ewma.

        Example
        -------
        .. code-block:: python

            # Equivalent to:
            # my_ewma = EWMA(
            #    inputs=[USEquityPricing.close],
            #    window_length=30,
            #    decay_rate=(1 - (1 / 15.0)),
            # )
            my_ewma = EWMA.from_center_of_mass(
                inputs=[USEquityPricing.close],
                window_length=30,
                center_of_mass=15,
            )

        Note
        ----
        This classmethod is provided by both
        :class:`ExponentialWeightedMovingAverage` and
        :class:`ExponentialWeightedMovingStdDev`.
        """
        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, ...

    Notes
    -----
    - This class can also be imported under the name ``EWMA``.

    See Also
    --------
    :func:`pandas.ewma`
    """
    def compute(self, today, assets, out, data, decay_rate):
        out[:] = average(
            data,
            axis=0,
            weights=self.weights(len(data), decay_rate),
        )


class ExponentialWeightedMovingStdDev(_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, ...

    Notes
    -----
    - This class can also be imported under the name ``EWMSTD``.

    See Also
    --------
    :func:`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 = ExponentialWeightedMovingStdDev


1			"""
2			Technical Analysis Factors
3			--------------------------
4			"""
5			from bottleneck import (
6			nanargmax,
7			nanmax,
8			nanmean,
9			nansum,
10			)
11			from numbers import Number
12			from numpy import (
13			abs,
14			arange,
15			average,
16			clip,
17			diff,
18			exp,
19			fmax,
20			full,
21			inf,
22			isnan,
23			log,
24			NINF,
25			sqrt,
26			sum as np_sum,
27			)
28			from numexpr import evaluate
29
30			from zipline.pipeline.data import USEquityPricing
31			from zipline.pipeline.mixins import SingleInputMixin
32			from zipline.utils.control_flow import ignore_nanwarnings
33			from zipline.utils.input_validation import expect_types
34			from .factor import CustomFactor
35
36
37			class Returns(CustomFactor):
38			"""
39			Calculates the percent change in close price over the given window_length.
40
41			Default Inputs: [USEquityPricing.close]
42			"""
43			inputs = [USEquityPricing.close]
44
45			def compute(self, today, assets, out, close):
46			out[:] = (close[-1] - close[0]) / close[0]
47
48
49			class RSI(CustomFactor, SingleInputMixin):
50			"""
51			Relative Strength Index
52
53			Default Inputs: [USEquityPricing.close]
54
55			Default Window Length: 15
56			"""
57			window_length = 15
58			inputs = (USEquityPricing.close,)
59
60			def compute(self, today, assets, out, closes):
61			diffs = diff(closes, axis=0)
62			ups = nanmean(clip(diffs, 0, inf), axis=0)
63			downs = abs(nanmean(clip(diffs, -inf, 0), axis=0))
64			return evaluate(
65			"100 - (100 / (1 + (ups / downs)))",
66			local_dict={'ups': ups, 'downs': downs},
67			global_dict={},
68			out=out,
69			)
70
71
72			class SimpleMovingAverage(CustomFactor, SingleInputMixin):
73			"""
74			Average Value of an arbitrary column
75
76			Default Inputs: None
77
78			Default Window Length: None
79			"""
80			# numpy's nan functions throw warnings when passed an array containing only
81			# nans, but they still returns the desired value (nan), so we ignore the
82			# warning.
83			ctx = ignore_nanwarnings()
84
85			def compute(self, today, assets, out, data):
86			out[:] = nanmean(data, axis=0)
87
88
89			class WeightedAverageValue(CustomFactor):
90			"""
91			Helper for VWAP-like computations.
92
93			Default Inputs: None
94
95			Default Window Length: None
96			"""
97			def compute(self, today, assets, out, base, weight):
98			out[:] = nansum(base * weight, axis=0) / nansum(weight, axis=0)
99
100
101			class VWAP(WeightedAverageValue):
102			"""
103			Volume Weighted Average Price
104
105			Default Inputs: [USEquityPricing.close, USEquityPricing.volume]
106
107			Default Window Length: None
108			"""
109			inputs = (USEquityPricing.close, USEquityPricing.volume)
110
111
112			class MaxDrawdown(CustomFactor, SingleInputMixin):
113			"""
114			Max Drawdown
115
116			Default Inputs: None
117
118			Default Window Length: None
119			"""
120			ctx = ignore_nanwarnings()
121
122			def compute(self, today, assets, out, data):
123			drawdowns = fmax.accumulate(data, axis=0) - data
124			drawdowns[isnan(drawdowns)] = NINF
125			drawdown_ends = nanargmax(drawdowns, axis=0)
126
127			# TODO: Accelerate this loop in Cython or Numba.
128			for i, end in enumerate(drawdown_ends):
129			peak = nanmax(data[:end + 1, i])
130			out[i] = (peak - data[end, i]) / data[end, i]
131
132
133			class AverageDollarVolume(CustomFactor):
134			"""
135			Average Daily Dollar Volume
136
137			Default Inputs: [USEquityPricing.close, USEquityPricing.volume]
138
139			Default Window Length: None
140			"""
141			inputs = [USEquityPricing.close, USEquityPricing.volume]
142
143			def compute(self, today, assets, out, close, volume):
144			out[:] = nanmean(close * volume, axis=0)
145
146
147			class _ExponentialWeightedFactor(SingleInputMixin, CustomFactor):
148			"""
149			Base class for factors implementing exponential-weighted operations.
150
151			Default Inputs: None
152
153			Default Window Length: None
154
155			Parameters
156			----------
157			inputs : length-1 list or tuple of BoundColumn
158			The expression over which to compute the average.
159			window_length : int > 0
160			Length of the lookback window over which to compute the average.
161			decay_rate : float, 0 < decay_rate <= 1
162			Weighting factor by which to discount past observations.
163
164			When calculating historical averages, rows are multiplied by the
165			sequence::
166
167			decay_rate, decay_rate 2, decay_rate 3, ...
168
169			Methods
170			-------
171			weights
172			from_span
173			from_halflife
174			from_center_of_mass
175			"""
176			params = ('decay_rate',)
177
178			@staticmethod
179			def weights(length, decay_rate):
180			"""
181			Return weighting vector for an exponential moving statistic on `length`
182			rows with a decay rate of `decay_rate`.
183			"""
184			return full(length, decay_rate) ** arange(length + 1, 1, -1)
185
186			@classmethod
187			@expect_types(span=Number)
188			def from_span(cls, inputs, window_length, span):
189			"""
190			Convenience constructor for passing `decay_rate` in terms of `span`.
191
192			Forwards `decay_rate` as `1 - (2.0 / (1 + span))`. This provides the
193			behavior equivalent to passing `span` to pandas.ewma.
194
195			Example
196			-------
197			.. code-block:: python
198
199			# Equivalent to:
200			# my_ewma = EWMA(
201			# inputs=[USEquityPricing.close],
202			# window_length=30,
203			# decay_rate=(1 - (2.0 / (1 + 15.0))),
204			# )
205			my_ewma = EWMA.from_span(
206			inputs=[USEquityPricing.close],
207			window_length=30,
208			span=15,
209			)
210
211			Note
212			----
213			This classmethod is provided by both
214			:class:`ExponentialWeightedMovingAverage` and
215			:class:`ExponentialWeightedMovingStdDev`.
216			"""
217			if span <= 1:
218			raise ValueError(
219			"`span` must be a positive number. %s was passed." % span
220			)
221
222			decay_rate = (1.0 - (2.0 / (1.0 + span)))
223			assert 0.0 < decay_rate <= 1.0
224
225			return cls(
226			inputs=inputs,
227			window_length=window_length,
228			decay_rate=decay_rate,
229			)
230
231			@classmethod
232			@expect_types(halflife=Number)
233			def from_halflife(cls, inputs, window_length, halflife):
234			"""
235			Convenience constructor for passing ``decay_rate`` in terms of half
236			life.
237
238			Forwards ``decay_rate`` as ``exp(log(.5) / halflife)``. This provides
239			the behavior equivalent to passing `halflife` to pandas.ewma.
240
241			Example
242			-------
243			.. code-block:: python
244
245			# Equivalent to:
246			# my_ewma = EWMA(
247			# inputs=[USEquityPricing.close],
248			# window_length=30,
249			# decay_rate=np.exp(np.log(0.5) / 15),
250			# )
251			my_ewma = EWMA.from_halflife(
252			inputs=[USEquityPricing.close],
253			window_length=30,
254			halflife=15,
255			)
256
257			Note
258			----
259			This classmethod is provided by both
260			:class:`ExponentialWeightedMovingAverage` and
261			:class:`ExponentialWeightedMovingStdDev`.
262			"""
263			if halflife <= 0:
264			raise ValueError(
265			"`span` must be a positive number. %s was passed." % halflife
266			)
267			decay_rate = exp(log(.5) / halflife)
268			assert 0.0 < decay_rate <= 1.0
269
270			return cls(
271			inputs=inputs,
272			window_length=window_length,
273			decay_rate=decay_rate,
274			)
275
276			@classmethod
277			def from_center_of_mass(cls, inputs, window_length, center_of_mass):
278			"""
279			Convenience constructor for passing `decay_rate` in terms of center of
280			mass.
281
282			Forwards `decay_rate` as `1 - (1 / 1 + center_of_mass)`. This provides
283			behavior equivalent to passing `center_of_mass` to pandas.ewma.
284
285			Example
286			-------
287			.. code-block:: python
288
289			# Equivalent to:
290			# my_ewma = EWMA(
291			# inputs=[USEquityPricing.close],
292			# window_length=30,
293			# decay_rate=(1 - (1 / 15.0)),
294			# )
295			my_ewma = EWMA.from_center_of_mass(
296			inputs=[USEquityPricing.close],
297			window_length=30,
298			center_of_mass=15,
299			)
300
301			Note
302			----
303			This classmethod is provided by both
304			:class:`ExponentialWeightedMovingAverage` and
305			:class:`ExponentialWeightedMovingStdDev`.
306			"""
307			return cls(
308			inputs=inputs,
309			window_length=window_length,
310			decay_rate=(1.0 - (1.0 / (1.0 + center_of_mass))),
311			)
312
313
314			class ExponentialWeightedMovingAverage(_ExponentialWeightedFactor):
315			"""
316			Exponentially Weighted Moving Average
317
318			Default Inputs: None
319
320			Default Window Length: None
321
322			Parameters
323			----------
324			inputs : length-1 list/tuple of BoundColumn
325			The expression over which to compute the average.
326			window_length : int > 0
327			Length of the lookback window over which to compute the average.
328			decay_rate : float, 0 < decay_rate <= 1
329			Weighting factor by which to discount past observations.
330
331			When calculating historical averages, rows are multiplied by the
332			sequence::
333
334			decay_rate, decay_rate 2, decay_rate 3, ...
335
336			Notes
337			-----
338			- This class can also be imported under the name ``EWMA``.
339
340			See Also
341			--------
342			:func:`pandas.ewma`
343			"""
344			def compute(self, today, assets, out, data, decay_rate):
345			out[:] = average(
346			data,
347			axis=0,
348			weights=self.weights(len(data), decay_rate),
349			)
350
351
352			class ExponentialWeightedMovingStdDev(_ExponentialWeightedFactor):
353			"""
354			Exponentially Weighted Moving Standard Deviation
355
356			Default Inputs: None
357
358			Default Window Length: None
359
360			Parameters
361			----------
362			inputs : length-1 list/tuple of BoundColumn
363			The expression over which to compute the average.
364			window_length : int > 0
365			Length of the lookback window over which to compute the average.
366			decay_rate : float, 0 < decay_rate <= 1
367			Weighting factor by which to discount past observations.
368
369			When calculating historical averages, rows are multiplied by the
370			sequence::
371
372			decay_rate, decay_rate 2, decay_rate 3, ...
373
374			Notes
375			-----
376			- This class can also be imported under the name ``EWMSTD``.
377
378			See Also
379			--------
380			:func:`pandas.ewmstd`
381			"""
382
383			def compute(self, today, assets, out, data, decay_rate):
384			weights = self.weights(len(data), decay_rate)
385
386			mean = average(data, axis=0, weights=weights)
387			variance = average((data - mean) ** 2, axis=0, weights=weights)
388
389			squared_weight_sum = (np_sum(weights) ** 2)
390			bias_correction = (
391			squared_weight_sum / (squared_weight_sum - np_sum(weights ** 2))
392			)
393			out[:] = sqrt(variance * bias_correction)
394
395
396			# Convenience aliases.
397			EWMA = ExponentialWeightedMovingAverage
398			EWMSTD = ExponentialWeightedMovingStdDev
399

quantopian / zipline

Pull Request — master (#927)

zipline.pipeline.factors.DollarVolume() A

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like