For code/output blocks: Use ``` (aka backtick or grave accent) in a single line before and after the block. See: http://commonmark.org/help/

Indicator unit test using new pytest fixture



  • I'm starting to look and practice with Backtrader. However the way I'm developing is BDD base, so i looked at how I could perform this with backtester to write new indicator.
    After some try and fail I finally set on using pytest fixture to drive unit test of indicatorthat.
    This allow me to use array as input and array as ouput without having to depend on any external file
    Could you give me your input about it and if you think this is worth having ?

    I pasted below to code of the fixture and how I use it with a simple sma.

    If you like that I will create an pip library that contain the fixture and may look at fixture for other component as well

    @pytest.fixture
    def indicator_runner():
        def _run_test(indicator: bt.Indicator,
                      result_columns: List[str],
                      open: List[float] = [],
                      high: List[float] = [],
                      low: List[float] = [],
                      close: List[float] = [],
                      **indicator_kwargs):
            max_len = max(len(open), len(high), len(low), len(close))
    
            open_list = [0 for _ in range(0, max_len)] if len(open) == 0 else open
            high_list = [0 for _ in range(0, max_len)] if len(high) == 0 else high
            low_list = [0 for _ in range(0, max_len)] if len(low) == 0 else low
            close_list = [0 for _ in range(0, max_len)] if len(close) == 0 else close
    
            if len(open_list) != max_len:
                raise Exception(
                    "open len is smaller than the max one. open list len = {}, max_len = {}".format(len(open_list),
                                                                                                    max_len))
            if len(high_list) != max_len:
                raise Exception(
                    "high len is smaller than the max one. high list len = {}, max_len = {}".format(len(high_list),
                                                                                                    max_len))
            if len(low_list) != max_len:
                raise Exception(
                    "low len is smaller than the max one. low list len = {}, max_len = {}".format(len(low_list), max_len))
            if len(close_list) != max_len:
                raise Exception(
                    "close len is smaller than the max one. close list len = {}, max_len = {}".format(len(close_list),
                                                                                                      max_len))
    
            raw_data = [{
                "open": val[0],
                "high": val[1],
                "low": val[2],
                "close": val[3],
                "datetime": dt.datetime(2018, 1, 1) + dt.timedelta(days=idx)
            } for idx, val in enumerate(zip(open_list, high_list, low_list, close_list))]
    
            df_raw_data = pd.DataFrame(raw_data, columns=['close', 'open', 'high', 'low', 'datetime'])
            df_raw_data.set_index("datetime", drop=True, inplace=True)
            data_feed = PandasData(dataname=df_raw_data)
            indicator.csv = True
    
            cerebro = bt.Cerebro()
            wrkwargs = {"csv": True}
            cerebro.addwriter(bt.WriterStringIO, **wrkwargs)
            cerebro.adddata(data_feed)
            cerebro.addindicator(indicator, **indicator_kwargs)
            cerebro.addstrategy(bt.Strategy)
            cerebro.run()
            result = _get_dataframe_from_writer(cerebro.runwriters[0])
    
            return {column: result[column].dropna().values for column in result_columns}
    
        def _get_dataframe_from_writer(writer):
            output_csv = StringIO()
            find_start = False
            for l in writer.out:
                if l.startswith('======') and not find_start:
                    find_start = True
                    continue
                if l.startswith('======') and find_start:
                    break
    
                output_csv.write(l)
    
            output_csv.seek(0)
            return pd.read_csv(output_csv)
    
        return _run_test
    

    and the way I'm using is like that

    @pytest.mark.parametrize("input_data, result_data, indicator, indicator_args, indicator_name", [
            (
            [1, 2, 3, 4, 5, 4, 3, 2, 1, 2, 3, 4, 5, 4, 3, 2, 1, 2, 3, 4],
            [1.5, 2.5, 3.5, 4.5, 4.5, 3.5, 2.5, 1.5, 1.5, 2.5, 3.5, 4.5, 4.5, 3.5, 2.5, 1.5, 1.5, 2.5, 3.5],
            MovingAverageSimple,
            {"period": 2},
            "sma"
        )
    ])
    def test_indicator(indicator_runner, input_data: List[float], result_data: List[float], indicator: bt.Indicator,
                       indicator_args: Dict[str, object], indicator_name: str):
        result = indicator_runner(indicator, [indicator_name], close=input_data, **indicator_args)
        indicators_value = result[indicator_name]
        assert (result_data == indicators_value).all()
    
    

  • administrators

    I do fail to grasp what you are trying to accomplish in general (it may be too early in 2019 and this may come later though).

    In any case this seems like a bit of over engineering:

    @davzucky said in Indicator unit test using new pytest fixture:

        def _get_dataframe_from_writer(writer):
            output_csv = StringIO()
            find_start = False
            for l in writer.out:
                if l.startswith('======') and not find_start:
                    find_start = True
                    continue
                if l.startswith('======') and find_start:
                    break
    
                output_csv.write(l)
    
            output_csv.seek(0)
            return pd.read_csv(output_csv)
    

    It would seem you try to capture the values of the indicator using the output of the writer. You could simply read the values directly from the array(s) which holds the output (i.e.: the "line(s)" defined by the indicator) once the execution is finished.



  • Thank you about that information. I didn't catch that cerebro contain the indicator and that I can query them . This is indeed more simpler. Will update the code and may publish that code as a pytest fixture anyway.