From f84ad70b5d9d32161ae07e28a43597c37a8a63fb Mon Sep 17 00:00:00 2001 From: femtotrader Date: Sat, 13 Jul 2024 12:03:07 +0200 Subject: [PATCH] Implements IntradayRange RelativeIntradayRange using higher level indicator FunctionCallerIndicator which is create using IndicatorFactory --- talipp/indicators/FunctionCallerIndicator.py | 23 ++++++++ talipp/indicators/IR.py | 33 ++++++++++++ talipp/indicators/IndicatorFactory.py | 15 ++++++ talipp/indicators/__init__.py | 2 + test/test_IR.py | 55 ++++++++++++++++++++ 5 files changed, 128 insertions(+) create mode 100644 talipp/indicators/FunctionCallerIndicator.py create mode 100644 talipp/indicators/IR.py create mode 100644 talipp/indicators/IndicatorFactory.py create mode 100644 test/test_IR.py diff --git a/talipp/indicators/FunctionCallerIndicator.py b/talipp/indicators/FunctionCallerIndicator.py new file mode 100644 index 00000000..84ef3176 --- /dev/null +++ b/talipp/indicators/FunctionCallerIndicator.py @@ -0,0 +1,23 @@ +from typing import Any +from collections.abc import Callable +from talipp.indicators.Indicator import Indicator + + +class FunctionCallerIndicator(Indicator): + def __init__(self, func: Callable): + super().__init__() + + self._func = func + + def __call__( + self, + input_values=None, + input_indicator=None, + *args: Any, + **kwargs + ) -> Any: + self.initialize(input_values, input_indicator) + return self + + def _calculate_new_value(self) -> Any: + return self._func(self.input_values[-1]) diff --git a/talipp/indicators/IR.py b/talipp/indicators/IR.py new file mode 100644 index 00000000..f585c08e --- /dev/null +++ b/talipp/indicators/IR.py @@ -0,0 +1,33 @@ +from talipp.indicators.IndicatorFactory import IndicatorFactory + + +""" +Intraday Range + +Input type: [OHLCV][talipp.ohlcv.OHLCV] + +Output type: `float` (as absolute value) + +Args: + input_values: List of input values. + input_indicator: Input indicator. + input_modifier: Input modifier. + input_sampling: Input sampling type. +""" +IR = IndicatorFactory.get_function_caller(lambda input: input.high - input.low) + + +""" +Relative Intraday Range + +Input type: [OHLCV][talipp.ohlcv.OHLCV] + +Output type: `float` (as relative value between 0 and 1) + +Args: + input_values: List of input values. + input_indicator: Input indicator. + input_modifier: Input modifier. + input_sampling: Input sampling type. +""" +RIR = IndicatorFactory.get_function_caller(lambda input: (input.high - input.low) / input.open) diff --git a/talipp/indicators/IndicatorFactory.py b/talipp/indicators/IndicatorFactory.py new file mode 100644 index 00000000..42af8b22 --- /dev/null +++ b/talipp/indicators/IndicatorFactory.py @@ -0,0 +1,15 @@ +from collections.abc import Callable +from talipp.indicators.FunctionCallerIndicator import FunctionCallerIndicator + + +class IndicatorFactory: + """Indicator factory.""" + + @staticmethod + def get_function_caller(func: Callable): + """ + Return a function caller indicator + Args: + func: Function which is called when input occurs. + """ + return FunctionCallerIndicator(func) diff --git a/talipp/indicators/__init__.py b/talipp/indicators/__init__.py index e10ffd29..36d52224 100644 --- a/talipp/indicators/__init__.py +++ b/talipp/indicators/__init__.py @@ -21,6 +21,7 @@ from .HMA import HMA as HMA from .IBS import IBS as IBS from .Ichimoku import Ichimoku as Ichimoku +from .IR import IR as IR from .KAMA import KAMA as KAMA from .KeltnerChannels import KeltnerChannels as KeltnerChannels from .KST import KST as KST @@ -31,6 +32,7 @@ from .MeanDev import MeanDev as MeanDev from .OBV import OBV as OBV from .PivotsHL import PivotsHL as PivotsHL +from .IR import RIR as RIR from .ROC import ROC as ROC from .RogersSatchell import RogersSatchell as RogersSatchell from .RSI import RSI as RSI diff --git a/test/test_IR.py b/test/test_IR.py new file mode 100644 index 00000000..224cd567 --- /dev/null +++ b/test/test_IR.py @@ -0,0 +1,55 @@ +import unittest + +from talipp.indicators import IR, RIR + +from TalippTest import TalippTest + + +class TestIR(TalippTest): + def setUp(self) -> None: + self.input_values = list(TalippTest.OHLCV_TMPL) + + def test_init(self): + ind = IR(self.input_values) + + print(ind) + + self.assertAlmostEqual(ind[-3], 0.670000, places = 5) + self.assertAlmostEqual(ind[-2], 0.620000, places = 5) + self.assertAlmostEqual(ind[-1], 0.770000, places = 5) + + def test_update(self): + self.assertIndicatorUpdate(IR(self.input_values)) + + def test_delete(self): + self.assertIndicatorDelete(IR(self.input_values)) + + def test_purge_oldest(self): + self.assertIndicatorPurgeOldest(IR(self.input_values)) + + +class TestRIR(TalippTest): + def setUp(self) -> None: + self.input_values = list(TalippTest.OHLCV_TMPL) + + def test_init(self): + ind = RIR(self.input_values) + + print(ind) + + self.assertAlmostEqual(ind[-3], 0.065111, places = 5) + self.assertAlmostEqual(ind[-2], 0.057567, places = 5) + self.assertAlmostEqual(ind[-1], 0.074903, places = 5) + + def test_update(self): + self.assertIndicatorUpdate(RIR(self.input_values)) + + def test_delete(self): + self.assertIndicatorDelete(RIR(self.input_values)) + + def test_purge_oldest(self): + self.assertIndicatorPurgeOldest(RIR(self.input_values)) + + +if __name__ == '__main__': + unittest.main()