text: '''
  Workflow Serialization Unit Tests
  
  To run unittests:
  
  # Using standard library unittest
  
  python -m unittest -v
  python -m unittest tests/unit/test_indicator.py -v
  
  or
  
  python -m unittest discover <test_directory>
  python -m unittest discover -s <directory> -p 'test_*.py'
  
  # Using pytest
  # "conda install pytest" or "pip install pytest"
  pytest -v tests
  pytest -v tests/unit/test_indicator.py
  
  '''
  import warnings
  import pandas as pd
  import unittest
  import pathlib
  import cudf
  import os
  from gquant.dataframe_flow.task import load_modules
  load_modules(os.getenv('MODULEPATH')+'/rapids_modules/')
  import rapids_modules.cuindicator as gi
  from . import technical_indicators as ti
  from .utils import make_orderer, error_function
  import numpy as np
  
  ordered, compare = make_orderer()
  unittest.defaultTestLoader.sortTestMethodsUsing = compare
  
  
  class TestIndicator(unittest.TestCase):
  
      def setUp(self):
          # ignore importlib warnings.
          path = pathlib.Path(__file__)
          self._pandas_data = pd.read_csv(str(path.parent)+'/testdata.csv.gz')
          self._pandas_data['Volume'] /= 1000.0
          self._cudf_data = cudf.from_pandas(self._pandas_data)
          warnings.simplefilter('ignore', category=ImportWarning)
          warnings.simplefilter('ignore', category=DeprecationWarning)
  
      def tearDown(self):
          pass
  
      @ordered
      def test_rate_of_return(self):
          '''Test rate of return calculation'''
          r_cudf = gi.rate_of_change(self._cudf_data['Close'], 2)
          r_pandas = ti.rate_of_change(self._pandas_data, 2)
          err = error_function(r_cudf, r_pandas.ROC_2)
          msg = "bad error %f\n" % (err,)
          self.assertTrue(np.isclose(err, 0, atol=1e-6), msg)
  
      @ordered
      def test_trix(self):
          """ test the trix calculation"""
          r_cudf = gi.trix(self._cudf_data['Close'], 3)
          r_pandas = ti.trix(self._pandas_data, 3)
          err = error_function(r_cudf, r_pandas.Trix_3)
          msg = "bad error %f\n" % (err,)
          self.assertTrue(np.isclose(err, 0, atol=1e-6), msg)
  
      @ordered
      def test_bollinger_bands(self):
          """ test the bollinger_bands """
          r_cudf = gi.bollinger_bands(self._cudf_data['Close'], 20)
          r_pandas = ti.bollinger_bands(self._pandas_data, 20)
          err = error_function(r_cudf.b1, r_pandas['BollingerB_20'])
          msg = "bad error %f\n" % (err,)
          self.assertTrue(np.isclose(err, 0, atol=1e-6), msg)
          err = error_function(r_cudf.b2, r_pandas['Bollinger%b_20'])
          msg = "bad error %f\n" % (err,)
          self.assertTrue(np.isclose(err, 0, atol=1e-6), msg)
  
      @ordered
      def test_macd(self):
          """ test the macd """
          n_fast = 10
          n_slow = 20
          r_cudf = gi.macd(self._cudf_data['Close'], n_fast, n_slow)
          r_pandas = ti.macd(self._pandas_data, n_fast, n_slow)
          err = error_function(r_cudf.MACD, r_pandas['MACD_10_20'])
          msg = "bad error %f\n" % (err,)
          self.assertTrue(np.isclose(err, 0, atol=1e-6), msg)
          err = error_function(r_cudf.MACDdiff, r_pandas['MACDdiff_10_20'])
          msg = "bad error %f\n" % (err,)
          self.assertTrue(np.isclose(err, 0, atol=1e-6), msg)
          err = error_function(r_cudf.MACDsign, r_pandas['MACDsign_10_20'])
          msg = "bad error %f\n" % (err,)
          self.assertTrue(np.isclose(err, 0, atol=1e-6), msg)
  
      @ordered
      def test_average_true_range(self):
          """ test the average true range """
          r_cudf = gi.average_true_range(self._cudf_data['High'],
                                         self._cudf_data['Low'],
                                         self._cudf_data['Close'], 10)
          r_pandas = ti.average_true_range(self._pandas_data, 10)
          err = error_function(r_cudf, r_pandas['ATR_10'])
          msg = "bad error %f\n" % (err,)
          self.assertTrue(np.isclose(err, 0, atol=1e-6), msg)
  
      @ordered
      def test_ppsr(self):
          """ test the ppsr """
          r_cudf = gi.ppsr(self._cudf_data['High'], self._cudf_data['Low'],
                           self._cudf_data['Close'])
          r_pandas = ti.ppsr(self._pandas_data)
          err = error_function(r_cudf.PP, r_pandas['PP'])
          msg = "bad error %f\n" % (err,)
          self.assertTrue(np.isclose(err, 0, atol=1e-6), msg)
          err = error_function(r_cudf.R1, r_pandas['R1'])
          msg = "bad error %f\n" % (err,)
          self.assertTrue(np.isclose(err, 0, atol=1e-6), msg)
          err = error_function(r_cudf.S1, r_pandas['S1'])
          msg = "bad error %f\n" % (err,)
          self.assertTrue(np.isclose(err, 0, atol=1e-6), msg)
          err = error_function(r_cudf.R2, r_pandas['R2'])
          msg = "bad error %f\n" % (err,)
          self.assertTrue(np.isclose(err, 0, atol=1e-6), msg)
          err = error_function(r_cudf.S2, r_pandas['S2'])
          msg = "bad error %f\n" % (err,)
          self.assertTrue(np.isclose(err, 0, atol=1e-6), msg)
          err = error_function(r_cudf.R3, r_pandas['R3'])
          msg = "bad error %f\n" % (err,)
          self.assertTrue(np.isclose(err, 0, atol=1e-6), msg)
          err = error_function(r_cudf.S3, r_pandas['S3'])
          msg = "bad error %f\n" % (err,)
          self.assertTrue(np.isclose(err, 0, atol=1e-6), msg)
  
      @ordered
      def test_stochastic_oscillator_k(self):
          """ test the stochastic oscillator k """
          r_cudf = gi.stochastic_oscillator_k(self._cudf_data['High'],
                                              self._cudf_data['Low'],
                                              self._cudf_data['Close'])
          r_pandas = ti.stochastic_oscillator_k(self._pandas_data)
          err = error_function(r_cudf, r_pandas['SO%k'])
          msg = "bad error %f\n" % (err,)
          self.assertTrue(np.isclose(err, 0, atol=1e-6), msg)
  
      @ordered
      def test_stochastic_oscillator_d(self):
          """ test the stochastic oscillator d """
          r_cudf = gi.stochastic_oscillator_d(self._cudf_data['High'],
                                              self._cudf_data['Low'],
                                              self._cudf_data['Close'], 10)
          r_pandas = ti.stochastic_oscillator_d(self._pandas_data, 10)
          err = error_function(r_cudf, r_pandas['SO%d_10'])
          msg = "bad error %f\n" % (err,)
          self.assertTrue(np.isclose(err, 0, atol=1e-6), msg)
  
      @ordered
      def test_average_directional_movement_index(self):
          """ test the average_directional_movement_index """
          r_cudf = gi.average_directional_movement_index(
              self._cudf_data['High'],
              self._cudf_data['Low'],
              self._cudf_data['Close'],
              10, 20)
          r_pandas = ti.average_directional_movement_index(self._pandas_data,
                                                           10, 20)
          err = error_function(r_cudf, r_pandas['ADX_10_20'])
          msg = "bad error %f\n" % (err,)
          self.assertTrue(np.isclose(err, 0, atol=1e-6), msg)
  
      @ordered
      def test_vortex_indicator(self):
          """ test the vortex_indicator """
          r_cudf = gi.vortex_indicator(self._cudf_data['High'],
                                       self._cudf_data['Low'],
                                       self._cudf_data['Close'], 10)
          r_pandas = ti.vortex_indicator(self._pandas_data, 10)
          err = error_function(r_cudf, r_pandas['Vortex_10'])
          msg = "bad error %f\n" % (err,)
          self.assertTrue(np.isclose(err, 0, atol=1e-6), msg)
  
      @ordered
      def test_kst_oscillator(self):
          """ test the kst_oscillator """
          r_cudf = gi.kst_oscillator(self._cudf_data['Close'],
                                     3, 4, 5, 6, 7, 8, 9, 10)
          r_pandas = ti.kst_oscillator(self._pandas_data,
                                       3, 4, 5, 6, 7, 8, 9, 10)
          err = error_function(r_cudf, r_pandas['KST_3_4_5_6_7_8_9_10'])
          msg = "bad error %f\n" % (err,)
          self.assertTrue(np.isclose(err, 0, atol=1e-6), msg)
  
      @ordered
      def test_relative_strength_index(self):
          """ test the relative_strength_index """
          r_cudf = gi.relative_strength_index(self._cudf_data['High'],
                                              self._cudf_data['Low'], 10)
          r_pandas = ti.relative_strength_index(self._pandas_data, 10)
          err = error_function(r_cudf, r_pandas['RSI_10'])
          msg = "bad error %f\n" % (err,)
          self.assertTrue(np.isclose(err, 0, atol=1e-6), msg)
  
      @ordered
      def test_mass_index(self):
          """ test the mass_index """
          r_cudf = gi.mass_index(self._cudf_data['High'],
                                 self._cudf_data['Low'], 9, 25)
          r_pandas = ti.mass_index(self._pandas_data)
          err = error_function(r_cudf, r_pandas['Mass Index'])
          msg = "bad error %f\n" % (err,)
          self.assertTrue(np.isclose(err, 0, atol=1e-6), msg)
  
      @ordered
      def test_true_strength_index(self):
          """ test the true_strength_index """
          r_cudf = gi.true_strength_index(self._cudf_data['Close'], 5, 8)
          r_pandas = ti.true_strength_index(self._pandas_data, 5, 8)
          err = error_function(r_cudf, r_pandas['TSI_5_8'])
          msg = "bad error %f\n" % (err,)
          self.assertTrue(np.isclose(err, 0, atol=1e-6), msg)
  
      @ordered
      def test_chaikin_oscillator(self):
          """ test the chaikin_oscillator """
          r_cudf = gi.chaikin_oscillator(self._cudf_data['High'],
                                         self._cudf_data['Low'],
                                         self._cudf_data['Close'],
                                         self._cudf_data['Volume'],  3, 10)
          r_pandas = ti.chaikin_oscillator(self._pandas_data)
          err = error_function(r_cudf, r_pandas['Chaikin'])
          msg = "bad error %f\n" % (err,)
          self.assertTrue(np.isclose(err, 0, atol=1e-6), msg)
  
      @ordered
      def test_money_flow_index(self):
          """ test the money_flow_index """
          r_cudf = gi.money_flow_index(self._cudf_data['High'],
                                       self._cudf_data['Low'],
                                       self._cudf_data['Close'],
                                       self._cudf_data['Volume'], 10)
          r_pandas = ti.money_flow_index(self._pandas_data, 10)
          err = error_function(r_cudf, r_pandas['MFI_10'])
          msg = "bad error %f\n" % (err,)
          self.assertTrue(np.isclose(err, 0, atol=1e-6), msg)
  
      @ordered
      def test_on_balance_volume(self):
          """ test the on_balance_volume """
          r_cudf = gi.on_balance_volume(self._cudf_data['Close'],
                                        self._cudf_data['Volume'], 10)
          r_pandas = ti.on_balance_volume(self._pandas_data, 10)
          err = error_function(r_cudf, r_pandas['OBV_10'])
          msg = "bad error %f\n" % (err,)
          self.assertTrue(np.isclose(err, 0, atol=1e-6), msg)
  
      @ordered
      def test_force_index(self):
          """ test the force index """
          r_cudf = gi.force_index(self._cudf_data['Close'],
                                  self._cudf_data['Volume'], 10)
          r_pandas = ti.force_index(self._pandas_data, 10)
          err = error_function(r_cudf, r_pandas['Force_10'])
          msg = "bad error %f\n" % (err,)
          self.assertTrue(np.isclose(err, 0, atol=1e-6), msg)
  
      @ordered
      def test_ease_of_movement(self):
          """ test the ease_of_movement """
          r_cudf = gi.ease_of_movement(self._cudf_data['High'],
                                       self._cudf_data['Low'],
                                       self._cudf_data['Volume'], 10)
          r_pandas = ti.ease_of_movement(self._pandas_data, 10)
          err = error_function(r_cudf, r_pandas['EoM_10'])
          msg = "bad error %f\n" % (err,)
          self.assertTrue(np.isclose(err, 0, atol=1e-6), msg)
  
      @ordered
      def test_ultimate_oscillator(self):
          """ test the ultimate_oscillator """
          r_cudf = gi.ultimate_oscillator(self._cudf_data['High'],
                                          self._cudf_data['Low'],
                                          self._cudf_data['Close'])
          r_pandas = ti.ultimate_oscillator(self._pandas_data)
          err = error_function(r_cudf, r_pandas['Ultimate_Osc'])
          msg = "bad error %f\n" % (err,)
          self.assertTrue(np.isclose(err, 0, atol=1e-6), msg)
  
      @ordered
      def test_donchian_channel(self):
          """ test the donchian_channel """
          r_cudf = gi.donchian_channel(self._cudf_data['High'],
                                       self._cudf_data['Low'], 10)
          r_pandas = ti.donchian_channel(self._pandas_data, 10)
          err = error_function(r_cudf[:-1], r_pandas['Donchian_10'][:-1])
          msg = "bad error %f\n" % (err,)
          self.assertTrue(np.isclose(err, 0, atol=1e-6), msg)
  
      @ordered
      def test_keltner_channel(self):
          """ test the keltner_channel """
          r_cudf = gi.keltner_channel(self._cudf_data['High'],
                                      self._cudf_data['Low'],
                                      self._cudf_data['Close'], 10)
          r_pandas = ti.keltner_channel(self._pandas_data, 10)
          err = error_function(r_cudf.KelChD, r_pandas['KelChD_10'])
          msg = "bad error %f\n" % (err,)
          self.assertTrue(np.isclose(err, 0, atol=1e-6), msg)
          err = error_function(r_cudf.KelChM, r_pandas['KelChM_10'])
          msg = "bad error %f\n" % (err,)
          self.assertTrue(np.isclose(err, 0, atol=1e-6), msg)
          err = error_function(r_cudf.KelChU, r_pandas['KelChU_10'])
          msg = "bad error %f\n" % (err,)
          self.assertTrue(np.isclose(err, 0, atol=1e-6), msg)
  
      @ordered
      def test_coppock_curve(self):
          """ test the coppock_curve """
          r_cudf = gi.coppock_curve(self._cudf_data['Close'], 10)
          r_pandas = ti.coppock_curve(self._pandas_data, 10)
          err = error_function(r_cudf, r_pandas['Copp_10'])
          msg = "bad error %f\n" % (err,)
          self.assertTrue(np.isclose(err, 0, atol=1e-6), msg)
  
      @ordered
      def test_accumulation_distribution(self):
          """ test the accumulation_distribution """
          r_cudf = gi.accumulation_distribution(self._cudf_data['High'],
                                                self._cudf_data['Low'],
                                                self._cudf_data['Close'],
                                                self._cudf_data['Volume'], 10)
          r_pandas = ti.accumulation_distribution(self._pandas_data, 10)
          err = error_function(r_cudf, r_pandas['Acc/Dist_ROC_10'])
          msg = "bad error %f\n" % (err,)
          self.assertTrue(np.isclose(err, 0, atol=1e-6), msg)
  
      @ordered
      def test_commodity_channel_index(self):
          """ test the commodity_channel_index """
          r_cudf = gi.commodity_channel_index(self._cudf_data['High'],
                                              self._cudf_data['Low'],
                                              self._cudf_data['Close'], 10)
          r_pandas = ti.commodity_channel_index(self._pandas_data, 10)
          err = error_function(r_cudf, r_pandas['CCI_10'])
          msg = "bad error %f\n" % (err,)
          self.assertTrue(np.isclose(err, 0, atol=1e-6), msg)
  
      @ordered
      def test_momentum(self):
          """ test the momentum """
          r_cudf = gi.momentum(self._cudf_data['Close'], 10)
          r_pandas = ti.momentum(self._pandas_data, 10)
          err = error_function(r_cudf, r_pandas['Momentum_10'])
          msg = "bad error %f\n" % (err,)
          self.assertTrue(np.isclose(err, 0, atol=1e-6), msg)
  
      @ordered
      def test_moving_average(self):
          """ test the moving average """
          r_cudf = gi.moving_average(self._cudf_data['Close'], 10)
          r_pandas = ti.moving_average(self._pandas_data, 10)
          err = error_function(r_cudf, r_pandas['MA_10'])
          msg = "bad error %f\n" % (err,)
          self.assertTrue(np.isclose(err, 0, atol=1e-6), msg)
  
      @ordered
      def test_exponential_moving_average(self):
          """ test the exponential moving average """
          r_cudf = gi.exponential_moving_average(self._cudf_data['Close'], 10)
          r_pandas = ti.exponential_moving_average(self._pandas_data, 10)
          err = error_function(r_cudf, r_pandas['EMA_10'])
          msg = "bad error %f\n" % (err,)
          self.assertTrue(np.isclose(err, 0, atol=1e-6), msg)
  
  
  if __name__ == '__main__':
      unittest.main()
  