<s> """<STR_LIT>""" <EOL> from numbers import Number <EOL> from numpy import ( <EOL> abs , <EOL> arange , <EOL> average , <EOL> clip , <EOL> diff , <EOL> exp , <EOL> fmax , <EOL> full , <EOL> inf , <EOL> isnan , <EOL> log , <EOL> NINF , <EOL> sqrt , <EOL> sum as np_sum , <EOL> ) <EOL> from numexpr import evaluate <EOL> from zipline . pipeline . data import USEquityPricing <EOL> from zipline . pipeline . mixins import SingleInputMixin <EOL> from zipline . utils . numpy_utils import ignore_nanwarnings <EOL> from zipline . utils . input_validation import expect_types <EOL> from zipline . utils . math_utils import ( <EOL> nanargmax , <EOL> nanmax , <EOL> nanmean , <EOL> nansum , <EOL> ) <EOL> from . factor import CustomFactor <EOL> class Returns ( CustomFactor ) : <EOL> """<STR_LIT>""" <EOL> inputs = [ USEquityPricing . close ] <EOL> def compute ( self , today , assets , out , close ) : <EOL> out [ : ] = ( close [ - <NUM_LIT:1> ] - close [ <NUM_LIT:0> ] ) / close [ <NUM_LIT:0> ] <EOL> class RSI ( CustomFactor , SingleInputMixin ) : <EOL> """<STR_LIT>""" <EOL> window_length = <NUM_LIT:15> <EOL> inputs = ( USEquityPricing . close , ) <EOL> def compute ( self , today , assets , out , closes ) : <EOL> diffs = diff ( closes , axis = <NUM_LIT:0> ) <EOL> ups = nanmean ( clip ( diffs , <NUM_LIT:0> , inf ) , axis = <NUM_LIT:0> ) <EOL> downs = abs ( nanmean ( clip ( diffs , - inf , <NUM_LIT:0> ) , axis = <NUM_LIT:0> ) ) <EOL> return evaluate ( <EOL> "<STR_LIT>" , <EOL> local_dict = { '<STR_LIT>' : ups , '<STR_LIT>' : downs } , <EOL> global_dict = { } , <EOL> out = out , <EOL> ) <EOL> class SimpleMovingAverage ( CustomFactor , SingleInputMixin ) : <EOL> """<STR_LIT>""" <EOL> ctx = ignore_nanwarnings ( ) <EOL> def compute ( self , today , assets , out , data ) : <EOL> out [ : ] = nanmean ( data , axis = <NUM_LIT:0> ) <EOL> class WeightedAverageValue ( CustomFactor ) : <EOL> """<STR_LIT>""" <EOL> def compute ( self , today , assets , out , base , weight ) : <EOL> out [ : ] = nansum ( base * weight , axis = <NUM_LIT:0> ) / nansum ( weight , axis = <NUM_LIT:0> ) <EOL> class VWAP ( WeightedAverageValue ) : <EOL> """<STR_LIT>""" <EOL> inputs = ( USEquityPricing . close , USEquityPricing . volume ) <EOL> class MaxDrawdown ( CustomFactor , SingleInputMixin ) : <EOL> """<STR_LIT>""" <EOL> ctx = ignore_nanwarnings ( ) <EOL> def compute ( self , today , assets , out , data ) : <EOL> drawdowns = fmax . accumulate ( data , axis = <NUM_LIT:0> ) - data <EOL> drawdowns [ isnan ( drawdowns ) ] = NINF <EOL> drawdown_ends = nanargmax ( drawdowns , axis = <NUM_LIT:0> ) <EOL> for i , end in enumerate ( drawdown_ends ) : <EOL> peak = nanmax ( data [ : end + <NUM_LIT:1> , i ] ) <EOL> out [ i ] = ( peak - data [ end , i ] ) / data [ end , i ] <EOL> class AverageDollarVolume ( CustomFactor ) : <EOL> """<STR_LIT>""" <EOL> inputs = [ USEquityPricing . close , USEquityPricing . volume ] <EOL> def compute ( self , today , assets , out , close , volume ) : <EOL> out [ : ] = nanmean ( close * volume , axis = <NUM_LIT:0> ) <EOL> class _ExponentialWeightedFactor ( SingleInputMixin , CustomFactor ) : <EOL> """<STR_LIT>""" <EOL> params = ( '<STR_LIT>' , ) <EOL> @ staticmethod <EOL> def weights ( length , decay_rate ) : <EOL> """<STR_LIT>""" <EOL> return full ( length , decay_rate , float ) ** arange ( length + <NUM_LIT:1> , <NUM_LIT:1> , - <NUM_LIT:1> ) <EOL> @ classmethod <EOL> @ expect_types ( span = Number ) <EOL> def from_span ( cls , inputs , window_length , span ) : <EOL> """<STR_LIT>""" <EOL> if span <= <NUM_LIT:1> : <EOL> raise ValueError ( <EOL> "<STR_LIT>" % span <EOL> ) <EOL> decay_rate = ( <NUM_LIT:1.0> - ( <NUM_LIT> / ( <NUM_LIT:1.0> + span ) ) ) <EOL> assert <NUM_LIT:0.0> < decay_rate <= <NUM_LIT:1.0> <EOL> return cls ( <EOL> inputs = inputs , <EOL> window_length = window_length , <EOL> decay_rate = decay_rate , <EOL> ) <EOL> @ classmethod <EOL> @ expect_types ( halflife = Number ) <EOL> def from_halflife ( cls , inputs , window_length , halflife ) : <EOL> """<STR_LIT>""" <EOL> if halflife <= <NUM_LIT:0> : <EOL> raise ValueError ( <EOL> "<STR_LIT>" % halflife <EOL> ) <EOL> decay_rate = exp ( log ( <NUM_LIT> ) / halflife ) <EOL> assert <NUM_LIT:0.0> < decay_rate <= <NUM_LIT:1.0> <EOL> return cls ( <EOL> inputs = inputs , <EOL> window_length = window_length , <EOL> decay_rate = decay_rate , <EOL> ) <EOL> @ classmethod <EOL> def from_center_of_mass ( cls , inputs , window_length , center_of_mass ) : <EOL> """<STR_LIT>""" <EOL> return cls ( <EOL> inputs = inputs , <EOL> window_length = window_length , <EOL> decay_rate = ( <NUM_LIT:1.0> - ( <NUM_LIT:1.0> / ( <NUM_LIT:1.0> + center_of_mass ) ) ) , <EOL> ) <EOL> class ExponentialWeightedMovingAverage ( _ExponentialWeightedFactor ) : <EOL> """<STR_LIT>""" <EOL> def compute ( self , today , assets , out , data , decay_rate ) : <EOL> out [ : ] = average ( <EOL> data , <EOL> axis = <NUM_LIT:0> , <EOL> weights = self . weights ( len ( data ) , decay_rate ) , <EOL> ) <EOL> class ExponentialWeightedMovingStdDev ( _ExponentialWeightedFactor ) : <EOL> """<STR_LIT>""" <EOL> def compute ( self , today , assets , out , data , decay_rate ) : <EOL> weights = self . weights ( len ( data ) , decay_rate ) <EOL> mean = average ( data , axis = <NUM_LIT:0> , weights = weights ) <EOL> variance = average ( ( data - mean ) ** <NUM_LIT:2> , axis = <NUM_LIT:0> , weights = weights ) <EOL> squared_weight_sum = ( np_sum ( weights ) ** <NUM_LIT:2> ) <EOL> bias_correction = ( <EOL> squared_weight_sum / ( squared_weight_sum - np_sum ( weights ** <NUM_LIT:2> ) ) <EOL> ) <EOL> out [ : ] = sqrt ( variance * bias_correction ) <EOL> EWMA = ExponentialWeightedMovingAverage <EOL> EWMSTD = ExponentialWeightedMovingStdDev </s>
