Source code for zipline.pipeline.factors.technical

"""
Technical Analysis Factors
--------------------------
"""
from __future__ import division

from numpy import (
    abs,
    average,
    clip,
    diff,
    dstack,
    inf,
)
from numexpr import evaluate

from zipline.pipeline.data import USEquityPricing
from zipline.pipeline.factors import CustomFactor
from zipline.pipeline.mixins import SingleInputMixin
from zipline.utils.input_validation import expect_bounded
from zipline.utils.math_utils import (
    nanargmax,
    nanargmin,
    nanmax,
    nanmean,
    nanstd,
    nanmin,
)
from zipline.utils.numpy_utils import rolling_window

from .basic import exponential_weights
from .basic import (  # noqa reexport
    # These are re-exported here for backwards compatibility with the old
    # definition site.
    LinearWeightedMovingAverage,
    MaxDrawdown,
    SimpleMovingAverage,
    VWAP,
    WeightedAverageValue
)


[docs]class RSI(CustomFactor, SingleInputMixin): """ Relative Strength Index **Default Inputs**: [USEquityPricing.close] **Default Window Length**: 15 """ window_length = 15 inputs = (USEquityPricing.close,) window_safe = True 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, )
[docs]class BollingerBands(CustomFactor): """ Bollinger Bands technical indicator. https://en.wikipedia.org/wiki/Bollinger_Bands **Default Inputs:** :data:`zipline.pipeline.data.USEquityPricing.close` Parameters ---------- inputs : length-1 iterable[BoundColumn] The expression over which to compute bollinger bands. window_length : int > 0 Length of the lookback window over which to compute the bollinger bands. k : float The number of standard deviations to add or subtract to create the upper and lower bands. """ params = ('k',) inputs = (USEquityPricing.close,) outputs = 'lower', 'middle', 'upper' def compute(self, today, assets, out, close, k): difference = k * nanstd(close, axis=0) out.middle = middle = nanmean(close, axis=0) out.upper = middle + difference out.lower = middle - difference
class Aroon(CustomFactor): """ Aroon technical indicator. https://www.fidelity.com/learning-center/trading-investing/technical-analysis/technical-indicator-guide/aroon-indicator # noqa **Defaults Inputs:** USEquityPricing.low, USEquityPricing.high Parameters ---------- window_length : int > 0 Length of the lookback window over which to compute the Aroon indicator. """ inputs = (USEquityPricing.low, USEquityPricing.high) outputs = ('down', 'up') def compute(self, today, assets, out, lows, highs): wl = self.window_length high_date_index = nanargmax(highs, axis=0) low_date_index = nanargmin(lows, axis=0) evaluate( '(100 * high_date_index) / (wl - 1)', local_dict={ 'high_date_index': high_date_index, 'wl': wl, }, out=out.up, ) evaluate( '(100 * low_date_index) / (wl - 1)', local_dict={ 'low_date_index': low_date_index, 'wl': wl, }, out=out.down, ) class FastStochasticOscillator(CustomFactor): """ Fast Stochastic Oscillator Indicator [%K, Momentum Indicator] https://wiki.timetotrade.eu/Stochastic This stochastic is considered volatile, and varies a lot when used in market analysis. It is recommended to use the slow stochastic oscillator or a moving average of the %K [%D]. **Default Inputs:** :data: `zipline.pipeline.data.USEquityPricing.close` :data: `zipline.pipeline.data.USEquityPricing.low` :data: `zipline.pipeline.data.USEquityPricing.high` **Default Window Length:** 14 Returns ------- out: %K oscillator """ inputs = (USEquityPricing.close, USEquityPricing.low, USEquityPricing.high) window_safe = True window_length = 14 def compute(self, today, assets, out, closes, lows, highs): highest_highs = nanmax(highs, axis=0) lowest_lows = nanmin(lows, axis=0) today_closes = closes[-1] evaluate( '((tc - ll) / (hh - ll)) * 100', local_dict={ 'tc': today_closes, 'll': lowest_lows, 'hh': highest_highs, }, global_dict={}, out=out, ) class IchimokuKinkoHyo(CustomFactor): """Compute the various metrics for the Ichimoku Kinko Hyo (Ichimoku Cloud). http://stockcharts.com/school/doku.php?id=chart_school:technical_indicators:ichimoku_cloud # noqa **Default Inputs:** :data:`zipline.pipeline.data.USEquityPricing.high` :data:`zipline.pipeline.data.USEquityPricing.low` :data:`zipline.pipeline.data.USEquityPricing.close` **Default Window Length:** 52 Parameters ---------- window_length : int > 0 The length the the window for the senkou span b. tenkan_sen_length : int >= 0, <= window_length The length of the window for the tenkan-sen. kijun_sen_length : int >= 0, <= window_length The length of the window for the kijou-sen. chikou_span_length : int >= 0, <= window_length The lag for the chikou span. """ params = { 'tenkan_sen_length': 9, 'kijun_sen_length': 26, 'chikou_span_length': 26, } inputs = (USEquityPricing.high, USEquityPricing.low, USEquityPricing.close) outputs = ( 'tenkan_sen', 'kijun_sen', 'senkou_span_a', 'senkou_span_b', 'chikou_span', ) window_length = 52 def _validate(self): super(IchimokuKinkoHyo, self)._validate() for k, v in self.params.items(): if v > self.window_length: raise ValueError( '%s must be <= the window_length: %s > %s' % ( k, v, self.window_length, ), ) def compute(self, today, assets, out, high, low, close, tenkan_sen_length, kijun_sen_length, chikou_span_length): out.tenkan_sen = tenkan_sen = ( high[-tenkan_sen_length:].max(axis=0) + low[-tenkan_sen_length:].min(axis=0) ) / 2 out.kijun_sen = kijun_sen = ( high[-kijun_sen_length:].max(axis=0) + low[-kijun_sen_length:].min(axis=0) ) / 2 out.senkou_span_a = (tenkan_sen + kijun_sen) / 2 out.senkou_span_b = (high.max(axis=0) + low.min(axis=0)) / 2 out.chikou_span = close[chikou_span_length] class RateOfChangePercentage(CustomFactor): """ Rate of change Percentage ROC measures the percentage change in price from one period to the next. The ROC calculation compares the current price with the price `n` periods ago. Formula for calculation: ((price - prevPrice) / prevPrice) * 100 price - the current price prevPrice - the price n days ago, equals window length """ def compute(self, today, assets, out, close): today_close = close[-1] prev_close = close[0] evaluate('((tc - pc) / pc) * 100', local_dict={ 'tc': today_close, 'pc': prev_close }, global_dict={}, out=out, ) class TrueRange(CustomFactor): """ True Range A technical indicator originally developed by J. Welles Wilder, Jr. Indicates the true degree of daily price change in an underlying. **Default Inputs:** :data:`zipline.pipeline.data.USEquityPricing.high` :data:`zipline.pipeline.data.USEquityPricing.low` :data:`zipline.pipeline.data.USEquityPricing.close` **Default Window Length:** 2 """ inputs = ( USEquityPricing.high, USEquityPricing.low, USEquityPricing.close, ) window_length = 2 def compute(self, today, assets, out, highs, lows, closes): high_to_low = highs[1:] - lows[1:] high_to_prev_close = abs(highs[1:] - closes[:-1]) low_to_prev_close = abs(lows[1:] - closes[:-1]) out[:] = nanmax( dstack(( high_to_low, high_to_prev_close, low_to_prev_close, )), 2 ) class MovingAverageConvergenceDivergenceSignal(CustomFactor): """ Moving Average Convergence/Divergence (MACD) Signal line https://en.wikipedia.org/wiki/MACD A technical indicator originally developed by Gerald Appel in the late 1970's. MACD shows the relationship between two moving averages and reveals changes in the strength, direction, momentum, and duration of a trend in a stock's price. **Default Inputs:** :data:`zipline.pipeline.data.USEquityPricing.close` Parameters ---------- fast_period : int > 0, optional The window length for the "fast" EWMA. Default is 12. slow_period : int > 0, > fast_period, optional The window length for the "slow" EWMA. Default is 26. signal_period : int > 0, < fast_period, optional The window length for the signal line. Default is 9. Notes ----- Unlike most pipeline expressions, this factor does not accept a ``window_length`` parameter. ``window_length`` is inferred from ``slow_period`` and ``signal_period``. """ inputs = (USEquityPricing.close,) # We don't use the default form of `params` here because we want to # dynamically calculate `window_length` from the period lengths in our # __new__. params = ('fast_period', 'slow_period', 'signal_period') @expect_bounded( __funcname='MACDSignal', fast_period=(1, None), # These must all be >= 1. slow_period=(1, None), signal_period=(1, None), ) def __new__(cls, fast_period=12, slow_period=26, signal_period=9, *args, **kwargs): if slow_period <= fast_period: raise ValueError( "'slow_period' must be greater than 'fast_period', but got\n" "slow_period={slow}, fast_period={fast}".format( slow=slow_period, fast=fast_period, ) ) return super(MovingAverageConvergenceDivergenceSignal, cls).__new__( cls, fast_period=fast_period, slow_period=slow_period, signal_period=signal_period, window_length=slow_period + signal_period - 1, *args, **kwargs ) def _ewma(self, data, length): decay_rate = 1.0 - (2.0 / (1.0 + length)) return average( data, axis=1, weights=exponential_weights(length, decay_rate) ) def compute(self, today, assets, out, close, fast_period, slow_period, signal_period): slow_EWMA = self._ewma( rolling_window(close, slow_period), slow_period ) fast_EWMA = self._ewma( rolling_window(close, fast_period)[-signal_period:], fast_period ) macd = fast_EWMA - slow_EWMA out[:] = self._ewma(macd.T, signal_period) # Convenience aliases. MACDSignal = MovingAverageConvergenceDivergenceSignal