diff --git a/docs/indicator-chaining.md b/docs/indicator-chaining.md index cd53fa3b..166593d3 100644 --- a/docs/indicator-chaining.md +++ b/docs/indicator-chaining.md @@ -37,6 +37,8 @@ print(smoothed_rsi) There is no limit to the number of chained indicators, one can create a computation pipeline from as many indicators as needed. +## Input modifiers + Chaining of indicators assumes that output and input types of chained indicators match. In case they do not, talipp provides an option to specify a conversion function which will be applied to the output value before it is fed to the next indicator. The function is specified in indicator's `__init__` method via `input_modifier` attribute. To illustrate usage of input modifiers, imagine we want to create a new indicator based on [Bollinger Bands][talipp.indicators.BB.BB] which will calculate [EMA][talipp.indicators.EMA.EMA] of the upper band. With standard libraries you would first calculate `Bolliger Bands`, then extract the upper band and finally feed it to `EMA`. With indicator chaining we can do better (besides it gives much more efficient solution). The only issue is that while `EMA` expects `floats` as the input, `Bollinger Bands` produce [BBVal][talipp.indicators.BB.BBVal]. Input modifiers for the rescue. diff --git a/docs/overrides/main.html b/docs/overrides/main.html index 024af79e..0af326af 100644 --- a/docs/overrides/main.html +++ b/docs/overrides/main.html @@ -1,9 +1,5 @@ {% extends "base.html" %} -{% block announce %} -The documentation is currently in its beta and any feedback is welcome. To provide it please use GitHub issues. -{% endblock %} - {% block outdated %} You're not viewing the latest version. diff --git a/docs/timeframe-sampling.md b/docs/timeframe-sampling.md index 0d23a245..8dbca0e5 100644 --- a/docs/timeframe-sampling.md +++ b/docs/timeframe-sampling.md @@ -44,3 +44,21 @@ ohlcv.time = dt.replace(second=25) obv.add(ohlcv) # still within the same timeframe => no new value added, the last one updated print(len(obv)) # 2 ``` + +!!! tip + + If you want to apply auto-sampling to an indicator which accepts `float` input, e.g. [MACD][talipp.indicators.MACD] indicator, then wrap each input value in a "dummy" [OHLCV][talipp.ohlcv.OHLCV] object, populate its `close` and `time` components and finally provide [input modifier](indicator-chaining.md#input-modifiers) to extract the value + + !!! example + + ```python + from datetime import datetime + from talipp.indicators import MACD + from talipp.input import SamplingPeriodType + from talipp.ohlcv import OHLCV + + input_floats = [1.0, 2.0, ...] + dt = datetime(2024, 1, 1, 0, 0, 0) + input_ohlcv = [OHLCV(None, None, None, value, None, dt) for value in input_floats] + macd = MACD(input_values=input_ohlcv, input_modifier=lambda x: x.close, input_sampling=SamplingPeriodType.SEC_15) + ``` diff --git a/mkdocs.yml b/mkdocs.yml index 8bde1f17..70b69257 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -32,7 +32,7 @@ theme: icon: material/weather-night name: Switch to system preference features: - - announce.dismiss +# - announce.dismiss - search.suggest - search.highlight - search.share diff --git a/talipp/indicators/BB.py b/talipp/indicators/BB.py index 9172a0da..574b3c8a 100644 --- a/talipp/indicators/BB.py +++ b/talipp/indicators/BB.py @@ -53,8 +53,8 @@ def __init__(self, period: int, self.period = period self.std_dev_mult = std_dev_mult - self.central_band = MAFactory.get_ma(ma_type, period, input_modifier=input_modifier) - self.std_dev = StdDev(self.period, input_modifier=input_modifier) + self.central_band = MAFactory.get_ma(ma_type, period) + self.std_dev = StdDev(self.period) self.add_sub_indicator(self.central_band) self.add_sub_indicator(self.std_dev) diff --git a/talipp/indicators/Indicator.py b/talipp/indicators/Indicator.py index 794ef07d..73f5c8c9 100644 --- a/talipp/indicators/Indicator.py +++ b/talipp/indicators/Indicator.py @@ -101,29 +101,30 @@ def add(self, value: Any) -> None: and self.input_sampler.is_same_period(value, self.input_values[-1])): self.update(value) else: - for sub_indicator in self.sub_indicators: - sub_indicator.add(value) - if not isinstance(value, list): value = [value] for input_value in value: if input_value is not None and self.input_modifier is not None: input_value = self.input_modifier(input_value) + + for sub_indicator in self.sub_indicators: + sub_indicator.add(input_value) + self.input_values.append(input_value) if input_value is not None: - new_value = self._calculate_new_value() + new_output_value = self._calculate_new_value() else: - new_value = None + new_output_value = None - if new_value is None and len(self.output_values) > 0: - new_value = self.output_values[-1] + if new_output_value is None and len(self.output_values) > 0: + new_output_value = self.output_values[-1] - self._add_to_output_values(new_value) + self._add_to_output_values(new_output_value) for listener in self.output_listeners: - listener.add(new_value) + listener.add(new_output_value) def update_input_value(self, value: Any) -> None: """**Deprecated.** Use [update][talipp.indicators.Indicator.Indicator.update] method instead. diff --git a/test/test_indicator_chaining.py b/test/test_indicator_chaining.py index 32bcfa76..34293ceb 100644 --- a/test/test_indicator_chaining.py +++ b/test/test_indicator_chaining.py @@ -1,7 +1,8 @@ import unittest from TalippTest import TalippTest -from talipp.indicators import SMA +from talipp.indicators import SMA, MACD +from talipp.ohlcv import OHLCV class Test(TalippTest): @@ -109,6 +110,24 @@ def test_purge_oldest(self): self.assertSequenceEqual([], sma3) self.assertSequenceEqual([], sma4) + def test_composite_sub_indicator(self): + ohlcv_input = list(TalippTest.OHLCV_TMPL) + ind = MACD(12, 26, 9, ohlcv_input, input_modifier=lambda x: x.close) + + print(ind) + + self.assertAlmostEqual(ind[-3].macd, 0.293541, places=5) + self.assertAlmostEqual(ind[-3].signal, 0.098639, places=5) + self.assertAlmostEqual(ind[-3].histogram, 0.194901, places=5) + + self.assertAlmostEqual(ind[-2].macd, 0.326186, places=5) + self.assertAlmostEqual(ind[-2].signal, 0.144149, places=5) + self.assertAlmostEqual(ind[-2].histogram, 0.182037, places=5) + + self.assertAlmostEqual(ind[-1].macd, 0.329698, places=5) + self.assertAlmostEqual(ind[-1].signal, 0.181259, places=5) + self.assertAlmostEqual(ind[-1].histogram, 0.148439, places=5) + if __name__ == '__main__': unittest.main()