Reject MACD signalperiod=1 for look-ahead affected TA-Lib paths

talib/__init__.py

@@ -106,6 +106,24 @@ def wrapper(*args, **kwds):
     _wrapper = lambda x: x
 
 
+def _validate_macd_signalperiod(func_name, kwargs):
+    signalperiod = kwargs.get('signalperiod')
+    if signalperiod == 1:
+        raise ValueError(
+            f"signalperiod=1 is not supported for {func_name} because the underlying TA-Lib "
+            "implementation can produce look-ahead affected results; use signalperiod >= 2 instead."
+        )
+
+
+def _macd_wrapper(func_name, func):
+    @wraps(func)
+    def wrapper(*args, **kwds):
+        _validate_macd_signalperiod(func_name, kwds)
+        return func(*args, **kwds)
+
+    return wrapper
+
+
 from ._ta_lib import (
     _ta_initialize, _ta_shutdown, MA_Type, __ta_version__,
     _ta_set_unstable_period as set_unstable_period,
@@ -122,13 +140,17 @@ def wrapper(*args, **kwds):
 func = __import__("_ta_lib", globals(), locals(), __TA_FUNCTION_NAMES__, level=1)
 for func_name in __TA_FUNCTION_NAMES__:
     wrapped_func = _wrapper(getattr(func, func_name))
+    if func_name in ('MACD', 'MACDFIX'):
+        wrapped_func = _macd_wrapper(func_name, wrapped_func)
     setattr(func, func_name, wrapped_func)
     globals()[func_name] = wrapped_func
 
 stream_func_names = ['stream_%s' % fname for fname in __TA_FUNCTION_NAMES__]
 stream = __import__("stream", globals(), locals(), stream_func_names, level=1)
 for func_name, stream_func_name in zip(__TA_FUNCTION_NAMES__, stream_func_names):
     wrapped_func = _wrapper(getattr(stream, func_name))
+    if func_name in ('MACD', 'MACDFIX'):
+        wrapped_func = _macd_wrapper(func_name, wrapped_func)
     setattr(stream, func_name, wrapped_func)
     globals()[stream_func_name] = wrapped_func
 

talib/abstract.py

@@ -10,13 +10,37 @@
 }
 
 
+class _FunctionProxy:
+    def __init__(self, func_name, func_obj):
+        object.__setattr__(self, '_func_name', func_name)
+        object.__setattr__(self, '_func_obj', func_obj)
+
+    def __call__(self, *args, **kwargs):
+        if self._func_name in ('MACD', 'MACDFIX') and kwargs.get('signalperiod') == 1:
+            raise ValueError(
+                f"signalperiod=1 is not supported for {self._func_name} because the underlying TA-Lib "
+                "implementation can produce look-ahead affected results; use signalperiod >= 2 instead."
+            )
+        return self._func_obj(*args, **kwargs)
+
+    def __getattr__(self, item):
+        return getattr(self._func_obj, item)
+
+    def __setattr__(self, key, value):
+        if key in {'_func_name', '_func_obj'}:
+            object.__setattr__(self, key, value)
+            return
+        setattr(self._func_obj, key, value)
+
+
 def Function(function_name, *args, **kwargs):
     func_name = function_name.upper()
     if func_name not in _func_obj_mapping:
         raise Exception('%s not supported by TA-LIB.' % func_name)
 
-    return _Function(
-        func_name, _func_obj_mapping[func_name], *args, **kwargs
+    return _FunctionProxy(
+        func_name,
+        _Function(func_name, _func_obj_mapping[func_name], *args, **kwargs)
     )
 
 

tests/test_abstract.py

@@ -25,6 +25,16 @@ def test_pararmeters():
     assert all(type(v) == int for k, v in parameters.items())
 
 
+def test_macd_signalperiod_one_rejected():
+    values = np.linspace(1.0, 100.0, 100, dtype=float)
+
+    with pytest.raises(ValueError, match="signalperiod=1 is not supported for MACD"):
+        abstract.MACD(values, signalperiod=1)
+
+    with pytest.raises(ValueError, match="signalperiod=1 is not supported for MACDFIX"):
+        abstract.MACDFIX(values, signalperiod=1)
+
+
 def test_pandas(ford_2012):
     import pandas
     input_df = pandas.DataFrame(ford_2012)

tests/test_func.py

@@ -63,6 +63,19 @@ def test_unstable_period():
     talib.set_unstable_period('EMA', 0)
 
 
+def test_macd_signalperiod_one_rejected():
+    values = np.linspace(1.0, 100.0, 100, dtype=float)
+
+    with pytest.raises(ValueError, match="signalperiod=1 is not supported for MACD"):
+        talib.MACD(values, signalperiod=1)
+
+    with pytest.raises(ValueError, match="signalperiod=1 is not supported for MACD"):
+        func.MACD(values, signalperiod=1)
+
+    with pytest.raises(ValueError, match="signalperiod=1 is not supported for MACDFIX"):
+        talib.MACDFIX(values, signalperiod=1)
+
+
 def test_compatibility():
     a = np.arange(10, dtype=float)
     talib.set_compatibility(0)

tests/test_stream.py

@@ -1,5 +1,6 @@
 import numpy as np
 import pandas as pd
+import pytest
 
 from talib import stream
 
@@ -64,3 +65,13 @@ def test_MAXINDEX():
     a = np.array([1., 2, 3, 4, 5, 6, 7, 8, 7, 7, 3, 4, 5, 6, 7, 8, 9, 2, 3, 4, 5, 15])
     r = stream.MAXINDEX(a, 10)
     assert r == 21
+
+
+def test_stream_macd_signalperiod_one_rejected():
+    values = np.linspace(1.0, 100.0, 100, dtype=float)
+
+    with pytest.raises(ValueError, match="signalperiod=1 is not supported for MACD"):
+        stream.MACD(values, signalperiod=1)
+
+    with pytest.raises(ValueError, match="signalperiod=1 is not supported for MACDFIX"):
+        stream.MACDFIX(values, signalperiod=1)