Implements IntradayRange RelativeIntradayRange using higher level ind…

…icator FunctionCallerIndicator which is create using IndicatorFactory

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])

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)

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)

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

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()