feat(stream): handles over TA-Lib C 0.8.1's streaming API (#757)

talib.stream.SMA(close) returns a handle, not a value: .value is the value at
the last history bar, .update(bar) is O(1) and returns that bar's, .peek(bar)
evaluates a forming bar without committing, .copy() forks it, and
.open_and_fill() returns the handle plus the Function API's series in one pass.
Multi-output functions answer with a named tuple; .out_range and .advance()
carry the C contract's bar range.

BREAKING: talib.stream_X and the old value-returning talib.stream.X are gone,
along with the stream_* stubs in _ta_lib.pyi -- talib/stream.pyi types the
handles instead. Migrating is stream.X(...) -> stream.X(...).value, and the
compiler will not find the sites: `if stream.CDLDOJI(o, h, l, c):` used to test
the pattern and now tests a handle, which is always true.

Adds talib.InsufficientHistory, raised when an open gets too little history.

talib/_stream.pxi and talib/stream.pyi are generated by
tools/generate_stream.py (--stub for the second).

Author: mario4tier

CHANGELOG

@@ -3,6 +3,24 @@
 
 - [NEW]: Support TA-Lib C 0.8.1, which is now the minimum required version.
 
+- [CHANGE]: ``talib.stream`` is now the real streaming API of TA-Lib C 0.8.1:
+  ``stream.SMA(close)`` returns a handle, not a value.  ``handle.value`` is the
+  value at the last history bar, ``handle.update(bar)`` costs O(1) and returns
+  that bar's value, ``handle.peek(bar)`` evaluates a forming bar without
+  committing it, and ``handle.copy()`` forks it.  ``stream.SMA.open_and_fill()``
+  returns the handle and the Function API's series in one pass.  A multi-output
+  function answers with a named tuple.  The old last-value functions --
+  ``talib.stream.SMA``, ``talib.stream_SMA``, and their ``_ta_lib.pyi`` stubs --
+  are gone; ``talib/stream.pyi`` types the handles instead.
+
+  Migrating is ``stream.X(...)`` -> ``stream.X(...).value``, and the compiler
+  cannot find the sites for you: ``if stream.CDLDOJI(o, h, l, c):`` used to test
+  the pattern and now tests a handle, which is always true.
+
+- [NEW]: ``talib.InsufficientHistory``, raised when a stream is opened with
+  too little history.  It is the library's one recoverable error, so it is
+  catchable on its own rather than as a bare ``Exception``.
+
 - [NEW]: The 40 functions TA-Lib C added since 0.7.1: AC, ADR, AO, CMF, CMOU,
   COPPOCK, CUMSUM, CVI, DONCHIAN, DPO, EFI, ER, ERI, FOSC, FRACTAL, HA, HMA,
   KC, KDJ, MARKETFI, MASSI, NVI, PERCENTILE, PERCENTRANK, PVI, PVO, PVT,

DEVELOPMENT

@@ -38,11 +38,15 @@ talib/_ta_lib.pyx
   need to use in the above pyx files.
 
 talib/_stream.pxi
-  This file contains code for interfacing a "streaming" interface to TA-Lib.
+  This file is generated automatically by tools/generate_stream.py: one handle
+  class per indicator, over TA-Lib C's streaming API.
+
+talib/stream.pyi
+  Type stubs for those handles, generated by tools/generate_stream.py --stub.
 
 tools/generate_func.py,generate_stream.py
-  Scripts that generate and print _func.pxi or _stream.pxi to stdout. Gets information
-  about all functions from the C headers of the installed TA-Lib.
+  Scripts that generate and print _func.pxi, _stream.pxi or stream.pyi to stdout.
+  Gets information about all functions from the C headers of the installed TA-Lib.
 
 If you are interested in developing new indicator functions or whatnot on
 the underlying TA-Lib, you must install TA-Lib from git.

MANIFEST.in

@@ -5,4 +5,5 @@ include talib/*.c
 include talib/*.pyx
 include talib/*.pxd
 include talib/*.pxi
+include talib/*.pyi
 include tests/*.py

Makefile

@@ -12,7 +12,13 @@ talib/_func.pxi: tools/generate_func.py
 talib/_stream.pxi: tools/generate_stream.py
 	python3 tools/generate_stream.py > talib/_stream.pxi
 
-generate: talib/_func.pxi talib/_stream.pxi
+talib/stream.pyi: tools/generate_stream.py
+	python3 tools/generate_stream.py --stub > talib/stream.pyi
+
+talib/abstract.pyi: tools/generate_abstract_stub.py
+	python3 tools/generate_abstract_stub.py > talib/abstract.pyi
+
+generate: talib/_func.pxi talib/_stream.pxi talib/stream.pyi talib/abstract.pyi
 
 cython:
 	cython talib/_ta_lib.pyx

README.md

@@ -532,27 +532,82 @@ slowk, slowd = STOCH(inputs, 5, 3, 0, 3, 0, prices=['high', 'low', 'open'])
 
 ## Streaming API
 
-An experimental Streaming API was added that allows users to compute the latest
-value of an indicator.  This can be faster than using the Function API, for
-example in an application that receives streaming data, and wants to know just
-the most recent updated indicator value.
+The Streaming API keeps a handle per indicator instead of recomputing from the
+whole array. Opening one costs a pass over the history; every bar after that is
+O(1), and each value it produces is identical to the one the Function API
+reports for that bar.
 
 ```python
 import talib
 from talib import stream
 
-close = np.random.random(100)
-
-# the Function API
+# the Function API: the whole series, from the whole array
 output = talib.SMA(close)
 
-# the Streaming API
-latest = stream.SMA(close)
+# the Streaming API: a handle, positioned at the end of the history
+s = stream.SMA(close)
+assert s.value == output[-1]
 
-# the latest value is the same as the last output value
-assert (output[-1] - latest) < 0.00001
+for price in feed:
+    latest = s.update(price)      # one closed bar in, its value out
+
+s.peek(forming)                   # what update would return, committing nothing
+fork = s.copy()                   # an independent handle at the same bar
+```
+
+`stream.SMA` takes exactly the arguments `talib.SMA` takes. A single-output
+function answers with a `float` (an `int` where the Function API returns an
+integer array); a multi-output one with a named tuple that still unpacks like
+the Function API's tuple:
+
+```python
+m = stream.MACD(close)
+macd, macdsignal, macdhist = m.update(price)
+m.value.macdhist
 ```
 
+Opening needs at least `lookback + 1` bars, which `abstract` knows, and a little
+more where a function's seeding does -- so rather than computing the number,
+treat a short history as "not yet":
+
+```python
+from talib import abstract
+
+need = abstract.Function('RSI', timeperiod=14).lookback + 1   # 15, usually enough
+
+try:
+    s = stream.RSI(history, timeperiod=14)
+except talib.InsufficientHistory:
+    ...                                                       # collect more bars
+```
+
+Leading bars that are NaN in any input are not history. They are skipped, as the
+Function API skips them, and do not count toward the warm-up. A NaN or an
+infinity anywhere else in the history is undefined behaviour in TA-Lib C, and
+for a few window functions a handle and the Function API do then disagree.
+
+A bar that is not finite is likewise rejected: `update` raises and the handle is
+left exactly as it was, neither its value nor its range moved. For a bar you
+mean to skip rather than re-feed, say so with `advance()`, or two handles on one
+feed drift a bar apart.
+
+If you want the series over the history as well, one pass gives both:
+
+```python
+s, rsi = stream.RSI.open_and_fill(history, timeperiod=14)     # rsi == talib.RSI(history)
+```
+
+A handle also reports the range it has an output for, in the input series'
+coordinates, and can be told about a bar it was not fed:
+
+```python
+s.out_range                       # (begidx, nbelement), as the Function API's output
+s.advance()                       # count a skipped bar: the range moves, the value holds
+```
+
+A handle points into the TA-Lib C library, so it cannot be pickled or shared
+with another process. Keep the history and re-open instead.
+
 ## Supported Indicators and Functions 📋
 
 We can show all the TA functions supported by TA-Lib, either as a `list` or

pyproject.toml

@@ -40,4 +40,4 @@ requires-python = '>=3.9'
 
 [tool.setuptools]
 packages = ["talib"]
-package-data = {"talib" = ["_ta_lib.pyi", "py.typed", "abstract.pyi"]}
+package-data = {"talib" = ["_ta_lib.pyi", "py.typed", "abstract.pyi", "stream.pyi"]}

talib/__init__.py

@@ -79,12 +79,6 @@ def wrapper(*args, **kwds):
 
             result = func(*_args, **_kwds)
 
-            # check to see if we got a streaming result
-            first_result = result[0] if isinstance(result, tuple) else result
-            is_streaming_fn_result = not hasattr(first_result, '__len__')
-            if is_streaming_fn_result:
-                return result
-
             # Series was passed in, Series gets out
             if use_pl:
                 if isinstance(result, tuple):
@@ -116,6 +110,7 @@ def wrapper(*args, **kwds):
         _ta_get_unstable_period as get_unstable_period,
         _ta_set_compatibility as set_compatibility,
         _ta_get_compatibility as get_compatibility,
+        InsufficientHistory,
         __TA_FUNCTION_NAMES__
     )
 except ImportError as error:
@@ -141,12 +136,7 @@ def wrapper(*args, **kwds):
     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))
-    setattr(stream, func_name, wrapped_func)
-    globals()[stream_func_name] = wrapped_func
+from . import stream
 
 __version__ = '0.7.1'
 
@@ -400,4 +390,4 @@ def get_function_groups():
     """
     return __function_groups__.copy()
 
-__all__ = ['get_functions', 'get_function_groups'] + __TA_FUNCTION_NAMES__ + ["stream_%s" % name for name in __TA_FUNCTION_NAMES__]
+__all__ = ['get_functions', 'get_function_groups', 'InsufficientHistory', 'stream'] + __TA_FUNCTION_NAMES__

talib/_common.pxi

@@ -5,6 +5,13 @@ from _ta_lib cimport TA_RetCode, TA_FuncUnstId
 
 __ta_version__ = lib.TA_GetVersionString()
 
+
+class InsufficientHistory(Exception):
+    """Not enough history to open a stream (TA_INSUFFICIENT_HISTORY).
+
+    Recoverable: collect more bars and try again."""
+
+
 cpdef _ta_check_success(str function_name, TA_RetCode ret_code):
     if ret_code == 0:
         return True
@@ -48,7 +55,8 @@ cpdef _ta_check_success(str function_name, TA_RetCode ret_code):
         description = 'Unknown Error (TA_UNKNOWN_ERR)'
     else:
         description = 'Unknown Error'
-    raise Exception('%s function failed with error code %s: %s' % (
+    error = InsufficientHistory if ret_code == 17 else Exception
+    raise error('%s function failed with error code %s: %s' % (
         function_name, ret_code, description))
 
 def _ta_initialize():