Source code for

Custom Momentum - MOM


**Supported environment variables**


    # to show debug, trace logging please export ``SHARED_LOG_CFG``
    # to a debug logger json file. To turn on debugging for this
    # library, you can export this variable to the repo's
    # included file with the command:
    export SHARED_LOG_CFG=/opt/sa/analysis_engine/log/debug-logging.json

import analysis_engine.ae_talib as ae_talib
import analysis_engine.consts as ae_consts
import analysis_engine.indicators.base_indicator as base_indicator

[docs]class IndicatorMOM(base_indicator.BaseIndicator): """IndicatorMOM""" def __init__( self, **kwargs): """__init__ Custom indicator example for showing a ``MOM`` within an algo for analyzing intraday minute datasets Please refer to the `analysis_engine.indicators.base_indicator.Ba seIndicator source code for the latest supported parameters <ht tps:// analysis_engine/indicators/>`__ :param kwargs: keyword arguments """ super().__init__(**kwargs) # end of __init__
[docs] def get_configurables( self, **kwargs): """get_configurables helper for setting up algorithm configs for this indicator and programmatically set the values based off the domain rules .. code-block:: python from \ import IndicatorMOM ind = IndicatorMOM(config_dict={ 'verbose': True }).get_configurables() :param kwargs: keyword args dictionary """ self.ind_confs = [] # common: self.build_base_configurables( ind_type='momentum', category='technical', uses_data=self.config.get( 'uses_data', 'minute'), version=1) # custom: self.ind_confs.append(self.build_configurable_node( name='num_points', conf_type='int', max_value=200, min_value=2, default_value=100, inc_interval=1)) self.ind_confs.append(self.build_configurable_node( name='buy_below_percent', conf_type='percent', min_value=3.0, max_value=100.0, default_value=5.0, inc_interval=1)) self.ind_confs.append(self.build_configurable_node( name='buy_above_percent', conf_type='percent', min_value=3.0, max_value=100.0, default_value=5.0, inc_interval=1)) self.ind_confs.append(self.build_configurable_node( name='sell_below_percent', conf_type='percent', min_value=3.0, max_value=100.0, default_value=5.0, inc_interval=1)) self.ind_confs.append(self.build_configurable_node( name='sell_above_percent', conf_type='percent', min_value=3.0, max_value=100.0, default_value=5.0, inc_interval=1)) # output / reporting: self.ind_confs.append(self.build_configurable_node( name='mom_value', conf_type='float', is_output_only=True)) self.ind_confs.append(self.build_configurable_node( name='amount_to_close', conf_type='float', is_output_only=True)) self.ind_confs.append(self.build_configurable_node( name='close', conf_type='float', is_output_only=True)) self.ind_confs.append(self.build_configurable_node( name='percent_value', conf_type='percent', is_output_only=True)) default_values_dict = {} for node in self.ind_confs: name = node['name'] default_value = node.get( 'default', None) default_values_dict[name] = default_value use_file = None try: if __file__: use_file = __file__ except Exception: use_file = None self.starter_dict = { 'name': self.__class__.__name__.lower().replace( 'indicator', ''), 'module_path': use_file, 'category': default_values_dict.get( 'category', 'momentum'), 'type': default_values_dict.get( 'type', 'technical'), 'uses_data': default_values_dict.get( 'uses_data', 'minute'), 'verbose': default_values_dict.get( 'verbose', False) } self.starter_dict.update(default_values_dict) self.lg( f'configurables={ae_consts.ppj(self.ind_confs)} for ' f'class={self.__class__.__name__} in file={use_file} ' f'starter:\n {ae_consts.ppj(self.starter_dict)}') return self.ind_confs
# end of get_configurables def get_starter_dict( self): if not self.starter_dict: self.get_configurables() return self.starter_dict # end of get_starter_dict
[docs] def process( self, algo_id, ticker, dataset): """process Derive custom indicator processing to determine buy and sell conditions before placing orders. Just implement your own ``process`` method. Please refer to the TA Lib guides for details on building indicators: - Overlap Studies - Momentum Indicators - Volume Indicators - Volatility Indicators - Price Transform - Cycle Indicators - Pattern Recognition - Statistic Functions - Math Transform - Math Operators :param algo_id: string - algo identifier label for debugging datasets during specific dates :param ticker: string - ticker :param dataset: dictionary of ``pandas.DataFrame(s)`` to process """ # set the algo config indicator 'uses_data' to 'day' or 'minute' df_status, self.use_df = self.get_subscribed_dataset( dataset=dataset) if df_status == ae_consts.EMPTY: self.lg('process end - no data found') return # notice the self.num_points is now a member variable # because the BaseIndicator class's __init__ # converts any self.config keys into useable # member variables automatically in your derived class self.lg( f'process - num_points={self.num_points} ' f'df={len(self.use_df.index)}') """ real = MOM(close, timeperiod=10) """ num_records = len(self.use_df.index) if num_records > self.num_points: cur_value = self.use_df['close'].iloc[-1] first_date = self.use_df['date'].iloc[0] end_date = self.use_df['date'].iloc[-1] start_row = num_records - self.num_points self.use_df = self.use_df[start_row:num_records] """ for idx, row in self.use_df[start_row:-1].iterrows(): high = row['high'] low = row['low'] open_val = row['open'] close = row['close'] row_date = row['date'] self.lg( f'{row_date} - high={high}, low={low}, ' f'close={close}, period={self.num_points}') """ closes = self.use_df['close'].values self.mom_value = ae_consts.to_f(ae_talib.MOM( close=closes, timeperiod=self.num_points)[-1]) """ Determine a buy or a sell as a label """ if cur_value <= 0: self.lg(f'invalid current_value={cur_value}') return self.close = cur_value self.amount_to_close = ae_consts.to_f( cur_value - self.mom_value) self.percent_value = ae_consts.to_f( self.amount_to_close / cur_value * 100.0) self.is_buy = ae_consts.INDICATOR_IGNORE self.is_sell = ae_consts.INDICATOR_IGNORE if (self.buy_above_percent != -1 and self.percent_value > self.buy_above_percent): self.is_buy = ae_consts.INDICATOR_BUY elif (self.buy_below_percent != -1 and self.percent_value > self.buy_below_percent): self.is_buy = ae_consts.INDICATOR_BUY if (self.sell_above_percent != -1 and self.percent_value > self.sell_above_percent): self.is_sell = ae_consts.INDICATOR_SELL elif (self.sell_below_percent != -1 and self.percent_value > self.sell_below_percent): self.is_sell = ae_consts.INDICATOR_SELL self.lg( f'process end - {first_date} to {end_date} ' f'buy_below={self.buy_below_percent} ' f'buy_above={self.buy_above_percent} is_buy={self.is_buy} ' f'sell_below={self.sell_below_percent} ' f'sell_above={self.sell_above_percent} is_sell={self.is_sell}') else: self.lg('process end')
# end of process
[docs] def reset_internals( self): """reset_internals""" self.is_buy = ae_consts.INDICATOR_RESET self.is_sell = ae_consts.INDICATOR_RESET
# end of reset_internals # end of IndicatorMOM
[docs]def get_indicator( **kwargs): """get_indicator Make sure to define the ``get_indicator`` for your custom algorithms to work as a backup with the ```` tool... Not anticipating issues, but if we do with importlib this is the backup plan. Please file an issue if you see something weird and would like some help: :param kwargs: dictionary of keyword arguments """ print('getting indicator') return IndicatorMOM(**kwargs)
# end of get_indicator