gavindiaz/wickra
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

python: drop the NumPy runtime dependency (zero third-party deps) (#317)

Browse Source 
Makes the Python binding truly dependency-free (Task 5b). `pip install wickra` now pulls **zero** third-party packages, and `import wickra` never imports NumPy (verified: `"numpy" not in sys.modules` after a batch call).

## What changed
- **Inputs** accept any sequence or buffer of numbers — `array.array`, `memoryview`, a NumPy `ndarray`, or a plain `list` — via `Vec` extraction (newtypes `Buf1`/`BufI64`). `PyBuffer` is unavailable under `abi3-py39`, and `unsafe_code = forbid` rules out a zero-copy slice, so inputs are copied once (negligible vs. compute).
- **Single-output `batch(...)`** returns a stdlib `array.array('d')` (native buffer protocol → `numpy.asarray` zero-copy).
- **Multi-output `batch(...)`** returns a buffer-protocol `Matrix` preserving `.shape`, integer-row and `[i, j]` access, and `.tolist()`.
- **NumPy** moves to an optional extra (`pip install wickra[numpy]`); it is never required.
- Streaming `update(...)` is unchanged; results are numerically identical.

## Implementation
- Trait `IntoPyData` + a `matrix()`/`f64_array()` helper collapse the ~400 batch call sites; `array.array` is built via `bytemuck` (Zlib/MIT/Apache — `cargo deny check licenses` ok).
- Tests migrated to `array.array`/`Matrix` (1-D `.shape`→`len()`, `.dtype`→`.typecode`; numeric comparisons normalize through a `_to_np` helper). The non-contiguous-input test now asserts acceptance instead of rejection.

## Verification (local)
- `pytest bindings/python/tests` — **1991 passed**.
- `cargo fmt --all`, `cargo clippy --workspace --all-targets --all-features -- -D warnings`, `cargo test --workspace --all-features`, `cargo deny check licenses` — all green.

**BREAKING for Python**: batch return types change from NumPy arrays to `array.array`/`Matrix`. Documented in CHANGELOG; ships with the data-layer release bundle (no separate tag).
This commit is contained in:
kingchenc
2026-06-17 03:19:50 +02:00
committed by GitHub
parent 677ea37402
commit 8228be7069
12 changed files with 1910 additions and 2713 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+16
View File
@@ -7,6 +7,22 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
## [Unreleased]
### Changed
- **Python: zero third-party dependencies — NumPy is no longer required**
(breaking). `pip install wickra` now pulls nothing else. Batch inputs accept any
sequence or buffer of numbers (`array.array`, `memoryview`, a NumPy array, or a
plain `list`); single-output `batch(...)` now returns a stdlib `array.array('d')`
and multi-output indicators return a buffer-protocol `Matrix` (with `.shape`,
integer-row and `[i, j]` element access, `.tolist()`) instead of 1-D / 2-D NumPy
arrays. Both expose the buffer protocol, so `numpy.asarray(result)` still wraps a
1-D result zero-copy when NumPy is installed — it is now an optional extra
(`pip install wickra[numpy]`). Streaming `update(...)` is unchanged, and results
are numerically identical. The other nine languages were already dependency-free.
### Removed
- **Python: `numpy` runtime dependency** (see *Changed*). NumPy moves to the
optional `numpy`/`test`/`bench` extras.
### Fixed
- **Binance kline feed: add the missing `3d` and `1M` intervals.** The
`live-binance` `Interval` enum was missing three-day (`3d`) and one-month (`1M`)
Cargo.lock
Generated
+7 -81
View File
@@ -100,6 +100,12 @@ version = "3.20.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5d20789868f4b01b2f2caec9f5c4e0213b41e3e5702a50157d699ae31ced2fcb"
[[package]]
name = "bytemuck"
version = "1.25.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c8efb64bd706a16a1bdde310ae86b351e4d21550d98d056f22f8a7f7a2183fec"
[[package]]
name = "bytes"
version = "1.11.1"
@@ -716,16 +722,6 @@ version = "0.4.29"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5e5032e24019045c762d3c0f28f5b6b8bbf38563a65908389bf7978758920897"
[[package]]
name = "matrixmultiply"
version = "0.3.10"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a06de3016e9fae57a36fd14dba131fccf49f74b40b7fbdb472f96e361ec71a08"
dependencies = [
"autocfg",
"rawpointer",
]
[[package]]
name = "memchr"
version = "2.8.0"
@@ -827,21 +823,6 @@ dependencies = [
"tempfile",
]
[[package]]
name = "ndarray"
version = "0.16.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "882ed72dce9365842bf196bdeedf5055305f11fc8c03dee7bb0194a6cad34841"
dependencies = [
"matrixmultiply",
"num-complex",
"num-integer",
"num-traits",
"portable-atomic",
"portable-atomic-util",
"rawpointer",
]
[[package]]
name = "nu-ansi-term"
version = "0.50.3"
@@ -851,24 +832,6 @@ dependencies = [
"windows-sys",
]
[[package]]
name = "num-complex"
version = "0.4.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "73f88a1307638156682bada9d7604135552957b7818057dcef22705b4d509495"
dependencies = [
"num-traits",
]
[[package]]
name = "num-integer"
version = "0.1.46"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7969661fd2958a5cb096e56c8e1ad0444ac2bbcd0061bd28660485a44879858f"
dependencies = [
"num-traits",
]
[[package]]
name = "num-traits"
version = "0.2.19"
@@ -901,22 +864,6 @@ dependencies = [
"syn",
]
[[package]]
name = "numpy"
version = "0.29.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6a5b15d63a5ff39e378daed0e1340d3a5964703ea9712eb09a0dc66fade996f4"
dependencies = [
"libc",
"ndarray",
"num-complex",
"num-integer",
"num-traits",
"pyo3",
"pyo3-build-config",
"rustc-hash",
]
[[package]]
name = "once_cell"
version = "1.21.4"
@@ -1034,15 +981,6 @@ version = "1.13.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c33a9471896f1c69cecef8d20cbe2f7accd12527ce60845ff44c153bb2a21b49"
[[package]]
name = "portable-atomic-util"
version = "0.2.7"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c2a106d1259c23fac8e543272398ae0e3c0b8d33c88ed73d0cc71b0f1d902618"
dependencies = [
"portable-atomic",
]
[[package]]
name = "potential_utf"
version = "0.1.5"
@@ -1214,12 +1152,6 @@ dependencies = [
"rand_core",
]
[[package]]
name = "rawpointer"
version = "0.2.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "60a357793950651c4ed0f3f52338f53b2f809f32d83a07f72909fa13e4c6c1e3"
[[package]]
name = "rayon"
version = "1.12.0"
@@ -1269,12 +1201,6 @@ version = "0.8.10"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "dc897dd8d9e8bd1ed8cdad82b5966c3e0ecae09fb1907d58efaa013543185d0a"
[[package]]
name = "rustc-hash"
version = "2.1.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "94300abf3f1ae2e2b8ffb7b58043de3d399c73fa6f4b73826402a5c457614dbe"
[[package]]
name = "rustix"
version = "1.1.4"
@@ -1934,7 +1860,7 @@ dependencies = [
name = "wickra-python"
version = "0.9.2"
dependencies = [
"numpy",
"bytemuck",
"pyo3",
"tokio",
"wickra-core",
bindings/python/Cargo.toml
+3 -1
View File
@@ -24,6 +24,8 @@ workspace = true
wickra-core = { workspace = true }
wickra-data = { workspace = true, features = ["live-binance"] }
pyo3 = { workspace = true }
numpy = { workspace = true }
# Reinterprets `&[f64]` as bytes when building stdlib `array.array('d')` results,
# keeping the binding free of any NumPy runtime dependency.
bytemuck = "1"
# Drives the async Binance feed behind the blocking, GIL-releasing poll.
tokio = { version = "1", features = ["rt", "net", "time"] }
bindings/python/pyproject.toml
+9 -3
View File
@@ -25,11 +25,17 @@ classifiers = [
"Topic :: Office/Business :: Financial :: Investment",
"Topic :: Scientific/Engineering :: Mathematics",
]
dependencies = [
"numpy>=1.22",
]
# Zero third-party runtime dependencies: batch results are stdlib `array.array`
# objects (and a buffer-protocol `Matrix` for multi-output indicators), so a plain
# `pip install wickra` pulls nothing else.
dependencies = []
[project.optional-dependencies]
# Optional NumPy interop. Results expose the buffer protocol, so `numpy.asarray(...)`
# wraps them zero-copy when NumPy is installed — but it is never required.
numpy = [
"numpy>=1.22",
]
test = [
"pytest>=7",
"numpy>=1.22",
bindings/python/python/wickra/__init__.py
+16 -5
View File
@@ -1,24 +1,35 @@
"""Wickra: streaming-first technical indicators.
Every indicator is available both in streaming mode (call ``update(value)`` per
new data point) and batch mode (call ``batch(numpy_array)`` over a full series).
Warmup positions in batch output are returned as ``NaN`` so the shape always
new data point) and batch mode (call ``batch(series)`` over a full series).
Warmup positions in batch output are returned as ``NaN`` so the length always
matches the input.
Wickra has **zero third-party dependencies**. Batch inputs accept any sequence or
buffer of numbers (``array.array``, ``memoryview``, a NumPy array, or a plain
``list``); single-output batches return a stdlib ``array.array('d')`` and
multi-output indicators return a ``Matrix`` (with ``.shape`` and ``[i, j]``
access). Both expose the buffer protocol, so ``numpy.asarray(result)`` wraps a
1-D result zero-copy when NumPy is installed — but NumPy is never required.
Example::
import numpy as np
import array
import wickra as ta
prices = np.linspace(100, 200, 1000)
prices = array.array("d", [100.0 + i * 0.1 for i in range(1000)])
rsi = ta.RSI(14)
values = rsi.batch(prices) # numpy array, NaN during warmup
values = rsi.batch(prices) # array.array('d'), NaN during warmup
# Or streaming:
rsi = ta.RSI(14)
for p in prices:
v = rsi.update(p) # None during warmup, then float
# Optional NumPy interop (zero-copy):
# import numpy as np
# np.asarray(values)
"""
from __future__ import annotations
bindings/python/src/lib.rs
+1592 -2404
View File
File diff suppressed because it is too large Load Diff
bindings/python/tests/test_input_validation.py
+8 -6
View File
@@ -8,20 +8,22 @@ import pytest
import wickra as ta
def test_non_contiguous_array_raises_value_error():
# A strided view is not C-contiguous; batch() must reject it cleanly.
def test_non_contiguous_array_is_accepted():
# Inputs are read through the buffer/sequence protocol, so a strided (non
# C-contiguous) view is accepted and yields the same result as a dense copy.
base = np.linspace(1.0, 100.0, 60)
non_contiguous = base[::2]
assert not non_contiguous.flags["C_CONTIGUOUS"]
with pytest.raises(ValueError):
ta.SMA(5).batch(non_contiguous)
strided = np.asarray(ta.SMA(5).batch(non_contiguous))
dense = np.asarray(ta.SMA(5).batch(np.ascontiguousarray(non_contiguous)))
np.testing.assert_array_equal(strided, dense)
def test_ascontiguousarray_recovers():
def test_dense_array_batches():
base = np.linspace(1.0, 100.0, 60)
fixed = np.ascontiguousarray(base[::2])
out = ta.SMA(5).batch(fixed)
assert out.shape == fixed.shape
assert len(out) == len(fixed)
def test_unequal_length_candle_batch_raises(ohlc_series):
bindings/python/tests/test_known_values.py
+97 -88
View File
@@ -10,8 +10,17 @@ import pytest
import wickra as ta
def _to_np(x) -> np.ndarray:
"""Normalize a Wickra batch result (``array.array`` or ``Matrix``) to an ndarray."""
if isinstance(x, np.ndarray):
return x.astype(np.float64)
if hasattr(x, "tolist") and hasattr(x, "shape"):
return np.asarray(x.tolist(), dtype=np.float64)
return np.asarray(x, dtype=np.float64)
def test_sma_constant_series():
out = ta.SMA(5).batch(np.full(20, 42.0, dtype=np.float64))
out = _to_np(ta.SMA(5).batch(np.full(20, 42.0, dtype=np.float64)))
# First 4 are warmup -> NaN; rest equal 42.
assert np.all(np.isnan(out[:4]))
assert np.allclose(out[4:], 42.0)
@@ -19,37 +28,37 @@ def test_sma_constant_series():
def test_sma_known_window():
# SMA(3) of [2, 4, 6, 8, 10] -> [_, _, 4, 6, 8]
out = ta.SMA(3).batch(np.array([2.0, 4.0, 6.0, 8.0, 10.0]))
out = _to_np(ta.SMA(3).batch(np.array([2.0, 4.0, 6.0, 8.0, 10.0])))
assert math.isnan(out[0]) and math.isnan(out[1])
np.testing.assert_allclose(out[2:], [4.0, 6.0, 8.0])
def test_ema_seed_equals_simple_mean_of_first_window():
# EMA(5) seed = mean([10, 20, 30, 40, 50]) = 30
out = ta.EMA(5).batch(np.array([10.0, 20.0, 30.0, 40.0, 50.0]))
out = _to_np(ta.EMA(5).batch(np.array([10.0, 20.0, 30.0, 40.0, 50.0])))
assert math.isnan(out[0])
assert math.isclose(out[4], 30.0, abs_tol=1e-12)
def test_wma_known_window():
# WMA(4) of [1, 2, 3, 4] = (1*1 + 2*2 + 3*3 + 4*4)/10 = 3
out = ta.WMA(4).batch(np.array([1.0, 2.0, 3.0, 4.0]))
out = _to_np(ta.WMA(4).batch(np.array([1.0, 2.0, 3.0, 4.0])))
assert math.isnan(out[0]) and math.isnan(out[1]) and math.isnan(out[2])
assert math.isclose(out[3], 3.0, abs_tol=1e-12)
def test_rsi_pure_uptrend_is_100():
out = ta.RSI(14).batch(np.arange(1.0, 21.0, dtype=np.float64))
out = _to_np(ta.RSI(14).batch(np.arange(1.0, 21.0, dtype=np.float64)))
np.testing.assert_allclose(out[14:], 100.0)
def test_rsi_pure_downtrend_is_0():
out = ta.RSI(14).batch(np.arange(20.0, 0.0, -1.0))
out = _to_np(ta.RSI(14).batch(np.arange(20.0, 0.0, -1.0)))
np.testing.assert_allclose(out[14:], 0.0)
def test_rsi_flat_series_is_50():
out = ta.RSI(14).batch(np.full(30, 100.0))
out = _to_np(ta.RSI(14).batch(np.full(30, 100.0)))
np.testing.assert_allclose(out[14:], 50.0)
@@ -62,13 +71,13 @@ def test_rsi_wilder_textbook_first_value():
],
dtype=np.float64,
)
out = ta.RSI(14).batch(prices)
out = _to_np(ta.RSI(14).batch(prices))
assert math.isclose(out[14], 70.464, abs_tol=0.05)
def test_anchored_rsi_cumulative_reference():
"""Cumulative anchored RSI: 10 -> 11 (+1) -> 9 (-2) -> 12 (+3)."""
out = ta.AnchoredRSI().batch(np.array([10.0, 11.0, 9.0, 12.0]))
out = _to_np(ta.AnchoredRSI().batch(np.array([10.0, 11.0, 9.0, 12.0])))
assert math.isclose(out[1], 100.0, abs_tol=1e-9)
assert math.isclose(out[2], 100.0 - 100.0 / 1.5, abs_tol=1e-6)
assert math.isclose(out[3], 100.0 - 100.0 / 3.0, abs_tol=1e-6)
@@ -79,9 +88,9 @@ def test_inertia_constant_rvi_passes_through_linreg():
# RVI = (c-o) / (h-l) = 0.5 / 2 = 0.25 every bar. LinReg of a constant
# series equals that constant after warmup.
n = 60
out = ta.Inertia(3, 4).batch(
out = _to_np(ta.Inertia(3, 4).batch(
np.full(n, 10.0), np.full(n, 11.0), np.full(n, 9.0), np.full(n, 10.5)
)
))
# warmup_period = 3 + 4 - 1 = 6.
np.testing.assert_allclose(out[5:], 0.25, atol=1e-12)
@@ -90,7 +99,7 @@ def test_connors_rsi_output_is_bounded():
# CRSI is the average of three [0, 100] components, so the aggregate must
# also sit in [0, 100] after warmup.
prices = 100.0 + 20.0 * np.sin(np.linspace(0, 30, 250))
out = ta.ConnorsRSI(3, 2, 100).batch(prices.astype(np.float64))
out = _to_np(ta.ConnorsRSI(3, 2, 100).batch(prices.astype(np.float64)))
ready = out[~np.isnan(out)]
assert ready.size > 0
assert ready.min() >= 0.0
@@ -101,7 +110,7 @@ def test_laguerre_rsi_constant_series_stays_at_mid_band():
# All four Laguerre stages seed to the first input, so subsequent flat
# inputs keep them equal and the up/down accumulator is 0 — Wickra maps
# that to the neutral 50.
out = ta.LaguerreRSI(0.5).batch(np.full(40, 42.0, dtype=np.float64))
out = _to_np(ta.LaguerreRSI(0.5).batch(np.full(40, 42.0, dtype=np.float64)))
np.testing.assert_allclose(out, 50.0, atol=1e-12)
@@ -109,7 +118,7 @@ def test_smi_close_at_centre_yields_zero():
# Close at the midpoint of a flat high/low range -> displacement is
# always zero -> SMI converges to 0.
n = 60
out = ta.SMI(5, 3, 3).batch(np.full(n, 11.0), np.full(n, 9.0), np.full(n, 10.0))
out = _to_np(ta.SMI(5, 3, 3).batch(np.full(n, 11.0), np.full(n, 9.0), np.full(n, 10.0)))
# warmup_period = 5 + 3 + 3 - 2 = 9.
np.testing.assert_allclose(out[8:], 0.0, atol=1e-12)
@@ -118,7 +127,7 @@ def test_kst_constant_series_yields_zero():
# ROC is zero on a flat input, so every RCMA is zero, so KST and its
# signal SMA are both zero after warmup.
kst = ta.KST(10, 15, 20, 30, 10, 10, 10, 15, 9)
out = kst.batch(np.full(80, 42.0, dtype=np.float64))
out = _to_np(kst.batch(np.full(80, 42.0, dtype=np.float64)))
warmup = kst.warmup_period()
# Use NaN-safe comparison on the post-warmup tail.
tail = out[warmup - 1 :]
@@ -133,7 +142,7 @@ def test_pgo_flat_close_yields_zero():
high = np.full(n, 11.0)
low = np.full(n, 9.0)
close = np.full(n, 10.0)
out = ta.PGO(5).batch(high, low, close)
out = _to_np(ta.PGO(5).batch(high, low, close))
assert np.all(np.isnan(out[:4]))
np.testing.assert_allclose(out[4:], 0.0, atol=1e-12)
@@ -141,12 +150,12 @@ def test_pgo_flat_close_yields_zero():
def test_rvi_reference_value_period_2():
# Two bars: (open, high, low, close) = (10, 11, 9, 10.5), (10.5, 11.5, 10, 11).
# num = (0.5 + 0.5) = 1.0; den = (2.0 + 1.5) = 3.5; RVI = 1 / 3.5.
out = ta.RVI(2).batch(
out = _to_np(ta.RVI(2).batch(
np.array([10.0, 10.5]),
np.array([11.0, 11.5]),
np.array([9.0, 10.0]),
np.array([10.5, 11.0]),
)
))
assert math.isnan(out[0])
assert math.isclose(out[1], 1.0 / 3.5, abs_tol=1e-12)
@@ -154,7 +163,7 @@ def test_rvi_reference_value_period_2():
def test_alma_constant_series_yields_the_constant():
# ALMA's Gaussian weights are normalised, so any constant series is
# reproduced exactly after warmup.
out = ta.ALMA(9, 0.85, 6.0).batch(np.full(30, 42.0, dtype=np.float64))
out = _to_np(ta.ALMA(9, 0.85, 6.0).batch(np.full(30, 42.0, dtype=np.float64)))
assert np.all(np.isnan(out[:8]))
np.testing.assert_allclose(out[8:], 42.0, atol=1e-12)
@@ -162,7 +171,7 @@ def test_alma_constant_series_yields_the_constant():
def test_alma_reference_value_period_3():
# ALMA(period=3, offset=0.85, sigma=6) on [10, 20, 30].
# m = 0.85 * 2 = 1.7; s = 3 / 6 = 0.5; 2*s^2 = 0.5.
out = ta.ALMA(3, 0.85, 6.0).batch(np.array([10.0, 20.0, 30.0]))
out = _to_np(ta.ALMA(3, 0.85, 6.0).batch(np.array([10.0, 20.0, 30.0])))
assert math.isnan(out[0]) and math.isnan(out[1])
# Independently compute the expected Gaussian-weighted sum.
w = np.exp(-((np.arange(3, dtype=np.float64) - 1.7) ** 2) / 0.5)
@@ -175,7 +184,7 @@ def test_alma_reference_value_period_3():
def test_mcginley_dynamic_constant_series_yields_the_constant():
# ratio = 1, so the recurrence collapses to MD + 0 / divisor = MD.
out = ta.McGinleyDynamic(5).batch(np.full(30, 42.0, dtype=np.float64))
out = _to_np(ta.McGinleyDynamic(5).batch(np.full(30, 42.0, dtype=np.float64)))
assert np.all(np.isnan(out[:4]))
np.testing.assert_allclose(out[4:], 42.0, atol=1e-12)
@@ -183,7 +192,7 @@ def test_mcginley_dynamic_constant_series_yields_the_constant():
def test_mcginley_dynamic_reference_value():
# Period 3, seed = SMA([10, 20, 30]) = 20.0. Next price 40.0:
# ratio = 2; divisor = 0.6 * 3 * 16 = 28.8; next = 20 + 20/28.8.
out = ta.McGinleyDynamic(3).batch(np.array([10.0, 20.0, 30.0, 40.0]))
out = _to_np(ta.McGinleyDynamic(3).batch(np.array([10.0, 20.0, 30.0, 40.0])))
assert math.isnan(out[0]) and math.isnan(out[1])
assert math.isclose(out[2], 20.0, abs_tol=1e-12)
expected = 20.0 + 20.0 / (0.6 * 3.0 * 16.0)
@@ -193,21 +202,21 @@ def test_mcginley_dynamic_reference_value():
def test_frama_constant_series_yields_the_constant():
# Flat input -> degenerate ranges -> alpha clamps to 0.01 and the EMA
# recurrence holds the seed value.
out = ta.FRAMA(4).batch(np.full(20, 42.0, dtype=np.float64))
out = _to_np(ta.FRAMA(4).batch(np.full(20, 42.0, dtype=np.float64)))
assert np.all(np.isnan(out[:3]))
np.testing.assert_allclose(out[3:], 42.0, atol=1e-12)
def test_frama_pure_uptrend_hugs_latest():
# Monotonic uptrend -> alpha pushed toward 1.0, FRAMA tracks close.
out = ta.FRAMA(4).batch(np.arange(1.0, 9.0, dtype=np.float64))
out = _to_np(ta.FRAMA(4).batch(np.arange(1.0, 9.0, dtype=np.float64)))
assert math.isclose(out[-1], 8.0, abs_tol=0.05)
def test_jma_constant_series_yields_the_constant():
# JMA seeds e0 and the output to the first input, so a constant series
# is reproduced exactly from the first sample.
out = ta.JMA(14, 0.0, 2).batch(np.full(30, 42.0, dtype=np.float64))
out = _to_np(ta.JMA(14, 0.0, 2).batch(np.full(30, 42.0, dtype=np.float64)))
np.testing.assert_allclose(out, 42.0, atol=1e-12)
@@ -215,7 +224,7 @@ def test_evwma_reference_value_period_2():
# EVWMA(2). Bars: (close, volume) = (10, 1), (20, 3), (30, 1).
# Bar 2: sum_v = 4, seeded prev = 20, EVWMA = (1*20 + 3*20)/4 = 20.
# Bar 3: sum_v = 4 (drops 1, gains 1), EVWMA = (3*20 + 1*30)/4 = 22.5.
out = ta.EVWMA(2).batch(np.array([10.0, 20.0, 30.0]), np.array([1.0, 3.0, 1.0]))
out = _to_np(ta.EVWMA(2).batch(np.array([10.0, 20.0, 30.0]), np.array([1.0, 3.0, 1.0])))
assert math.isnan(out[0])
assert math.isclose(out[1], 20.0, abs_tol=1e-12)
assert math.isclose(out[2], 22.5, abs_tol=1e-12)
@@ -227,7 +236,7 @@ def test_alligator_constant_series_holds_at_median_price():
n = 30
high = np.full(n, 11.0)
low = np.full(n, 9.0)
out = ta.Alligator(13, 8, 5).batch(high, low)
out = _to_np(ta.Alligator(13, 8, 5).batch(high, low))
assert out.shape == (n, 3)
for row in out[12:]:
assert math.isclose(row[0], 10.0, abs_tol=1e-12)
@@ -237,7 +246,7 @@ def test_alligator_constant_series_holds_at_median_price():
def test_vidya_constant_series_holds_seed():
# CMO = 0 on a flat series -> alpha = 0 -> VIDYA holds its seed value.
out = ta.VIDYA(14, 4).batch(np.full(20, 42.0, dtype=np.float64))
out = _to_np(ta.VIDYA(14, 4).batch(np.full(20, 42.0, dtype=np.float64)))
assert np.all(np.isnan(out[:4]))
np.testing.assert_allclose(out[4:], 42.0, atol=1e-12)
@@ -245,7 +254,7 @@ def test_vidya_constant_series_holds_seed():
def test_zero_lag_macd_constant_series_converges_to_zero():
# Each inner ZLEMA reproduces a constant, so macd, signal and histogram
# are all 0 once the slowest branch warms up.
out = ta.ZeroLagMACD(3, 5, 3).batch(np.full(60, 42.0, dtype=np.float64))
out = _to_np(ta.ZeroLagMACD(3, 5, 3).batch(np.full(60, 42.0, dtype=np.float64)))
# Take the last row and verify all three columns are 0.
last = out[-1]
assert math.isclose(last[0], 0.0, abs_tol=1e-12)
@@ -265,7 +274,7 @@ def test_awesome_oscillator_histogram_flat_series_converges_to_zero():
def test_stc_constant_series_yields_zero():
# Flat input collapses both stochastic stages to zero -> STC stays at 0.
out = ta.STC(3, 5, 4, 0.5).batch(np.full(60, 42.0, dtype=np.float64))
out = _to_np(ta.STC(3, 5, 4, 0.5).batch(np.full(60, 42.0, dtype=np.float64)))
ready = out[~np.isnan(out)]
assert ready.size > 0
np.testing.assert_array_equal(ready[-5:], np.zeros(5))
@@ -273,7 +282,7 @@ def test_stc_constant_series_yields_zero():
def test_elder_impulse_constant_series_is_neutral():
# Flat input -> neither EMA nor MACD histogram moves -> Impulse stays at 0.
out = ta.ElderImpulse(13, 12, 26, 9).batch(np.full(120, 42.0, dtype=np.float64))
out = _to_np(ta.ElderImpulse(13, 12, 26, 9).batch(np.full(120, 42.0, dtype=np.float64)))
ready = out[~np.isnan(out)]
assert ready.size > 0
np.testing.assert_array_equal(ready, np.zeros_like(ready))
@@ -281,19 +290,19 @@ def test_elder_impulse_constant_series_is_neutral():
def test_cfo_perfect_linear_series_yields_zero():
# LinReg of a perfectly linear series fits exactly, so CFO = 0 after warmup.
out = ta.CFO(5).batch(np.arange(1.0, 21.0, dtype=np.float64) * 2.0)
out = _to_np(ta.CFO(5).batch(np.arange(1.0, 21.0, dtype=np.float64) * 2.0))
np.testing.assert_allclose(out[4:], 0.0, atol=1e-9)
def test_apo_constant_series_converges_to_zero():
# Both EMAs reproduce a constant exactly, so APO = 0 after warmup.
out = ta.APO(3, 5).batch(np.full(30, 42.0, dtype=np.float64))
out = _to_np(ta.APO(3, 5).batch(np.full(30, 42.0, dtype=np.float64)))
assert np.all(np.isnan(out[:4]))
np.testing.assert_allclose(out[4:], 0.0, atol=1e-12)
def test_macd_constant_series_converges_to_zero():
out = ta.MACD().batch(np.full(200, 100.0))
out = _to_np(ta.MACD().batch(np.full(200, 100.0)))
# Last row's MACD and signal must be ~0.
last = out[-1]
assert math.isclose(last[0], 0.0, abs_tol=1e-9)
@@ -302,13 +311,13 @@ def test_macd_constant_series_converges_to_zero():
def test_bollinger_constant_series_zero_width():
out = ta.BollingerBands(20, 2.0).batch(np.full(50, 100.0))
out = _to_np(ta.BollingerBands(20, 2.0).batch(np.full(50, 100.0)))
row = out[-1]
np.testing.assert_allclose(row, [100.0, 100.0, 100.0, 0.0], atol=1e-12)
def test_bollinger_upper_middle_lower_ordering():
out = ta.BollingerBands(20, 2.0).batch(np.linspace(50.0, 150.0, 100))
out = _to_np(ta.BollingerBands(20, 2.0).batch(np.linspace(50.0, 150.0, 100)))
ready = out[~np.isnan(out[:, 0])]
assert np.all(ready[:, 0] >= ready[:, 1])
assert np.all(ready[:, 1] >= ready[:, 2])
@@ -319,7 +328,7 @@ def test_atr_constant_range_constant_output():
high = np.full(30, 11.0)
low = np.full(30, 9.0)
close = np.full(30, 10.0)
out = ta.ATR(14).batch(high, low, close)
out = _to_np(ta.ATR(14).batch(high, low, close))
# Once seeded, ATR equals the constant TR of 2.
np.testing.assert_allclose(out[13:], 2.0, atol=1e-12)
@@ -329,14 +338,14 @@ def test_stochastic_extremes():
high = np.array([10.0, 11.0, 12.0])
low = np.array([8.0, 9.0, 10.0])
close = np.array([9.0, 10.0, 12.0])
out = ta.Stochastic(3, 1).batch(high, low, close)
out = _to_np(ta.Stochastic(3, 1).batch(high, low, close))
assert math.isclose(out[2, 0], 100.0, abs_tol=1e-12)
def test_obv_cumulative_known_sequence():
close = np.array([10.0, 11.0, 10.5, 10.5, 12.0])
volume = np.array([100.0, 20.0, 30.0, 40.0, 10.0])
out = ta.OBV().batch(close, volume)
out = _to_np(ta.OBV().batch(close, volume))
np.testing.assert_allclose(out, [0.0, 20.0, -10.0, -10.0, 0.0])
@@ -346,7 +355,7 @@ def test_obv_cumulative_known_sequence():
def test_sharpe_ratio_known_window():
# returns [0.01, 0.02, 0.03, 0.04], rf = 0; mean = 0.025;
# sample-var = 0.000166...; Sharpe = 0.025 / sqrt(var).
out = ta.SharpeRatio(4, 0.0).batch(np.array([0.01, 0.02, 0.03, 0.04]))
out = _to_np(ta.SharpeRatio(4, 0.0).batch(np.array([0.01, 0.02, 0.03, 0.04])))
expected = 0.025 / math.sqrt(0.000_166_666_666_666_666_67)
assert math.isclose(out[3], expected, rel_tol=1e-9)
@@ -354,49 +363,49 @@ def test_sharpe_ratio_known_window():
def test_sortino_ratio_known_window():
# returns [-0.02, 0.01, -0.01, 0.03], mar = 0; mean = 0.0025;
# downside_sq = 0.0005; dd = sqrt(0.0005/4); Sortino = 0.0025/dd.
out = ta.SortinoRatio(4, 0.0).batch(np.array([-0.02, 0.01, -0.01, 0.03]))
out = _to_np(ta.SortinoRatio(4, 0.0).batch(np.array([-0.02, 0.01, -0.01, 0.03])))
expected = 0.0025 / math.sqrt(0.000_125)
assert math.isclose(out[3], expected, rel_tol=1e-9)
def test_max_drawdown_known_window():
# window [100, 120, 90] -> peak 120, trough 90 -> 25% drawdown.
out = ta.MaxDrawdown(3).batch(np.array([100.0, 120.0, 90.0]))
out = _to_np(ta.MaxDrawdown(3).batch(np.array([100.0, 120.0, 90.0])))
assert math.isclose(out[2], 0.25, abs_tol=1e-12)
def test_pain_index_known_window():
# dd[0..2] = 0, 0, 0.25; mean = 0.25/3.
out = ta.PainIndex(3).batch(np.array([100.0, 120.0, 90.0]))
out = _to_np(ta.PainIndex(3).batch(np.array([100.0, 120.0, 90.0])))
assert math.isclose(out[2], 0.25 / 3.0, abs_tol=1e-12)
def test_profit_factor_known_window():
# gains 0.05, losses 0.03 -> PF = 5/3.
out = ta.ProfitFactor(4).batch(np.array([0.02, -0.01, 0.03, -0.02]))
out = _to_np(ta.ProfitFactor(4).batch(np.array([0.02, -0.01, 0.03, -0.02])))
assert math.isclose(out[3], 5.0 / 3.0, rel_tol=1e-9)
def test_gain_loss_ratio_known_window():
# avg_win 0.03, avg_loss 0.02 -> GLR = 1.5.
out = ta.GainLossRatio(4).batch(np.array([0.02, -0.01, 0.04, -0.03]))
out = _to_np(ta.GainLossRatio(4).batch(np.array([0.02, -0.01, 0.04, -0.03])))
assert math.isclose(out[3], 1.5, rel_tol=1e-9)
def test_omega_ratio_known_window():
# gains 0.04, losses 0.03 -> Omega = 4/3.
out = ta.OmegaRatio(4, 0.0).batch(np.array([-0.02, 0.01, -0.01, 0.03]))
out = _to_np(ta.OmegaRatio(4, 0.0).batch(np.array([-0.02, 0.01, -0.01, 0.03])))
assert math.isclose(out[3], 4.0 / 3.0, rel_tol=1e-9)
def test_kelly_criterion_known_window():
# n_win=n_loss=2, payoff=2 -> Kelly = 0.5 - 0.5/2 = 0.25.
out = ta.KellyCriterion(4).batch(np.array([0.02, 0.04, -0.01, -0.02]))
out = _to_np(ta.KellyCriterion(4).batch(np.array([0.02, 0.04, -0.01, -0.02])))
assert math.isclose(out[3], 0.25, rel_tol=1e-9)
def test_drawdown_duration_under_water_counter():
out = ta.DrawdownDuration().batch(np.array([100.0, 95.0, 90.0, 85.0]))
out = _to_np(ta.DrawdownDuration().batch(np.array([100.0, 95.0, 90.0, 85.0])))
np.testing.assert_allclose(out, [0.0, 1.0, 2.0, 3.0])
@@ -404,35 +413,35 @@ def test_recovery_factor_known_path():
# Start 100, peak 110, trough 88 -> max_dd = 0.20; end 130 ->
# net_return = 0.30 -> Recovery = 1.5.
prices = np.array([100.0, 110.0, 105.0, 95.0, 88.0, 100.0, 120.0, 130.0])
out = ta.RecoveryFactor().batch(prices)
out = _to_np(ta.RecoveryFactor().batch(prices))
assert math.isclose(out[-1], 1.5, rel_tol=1e-9)
def test_alpha_perfect_capm_fit_yields_zero():
bench = np.array([0.01 * i for i in range(1, 21)])
asset = 2.0 * bench
out = ta.Alpha(20, 0.0).batch(asset, bench)
out = _to_np(ta.Alpha(20, 0.0).batch(asset, bench))
assert math.isclose(out[-1], 0.0, abs_tol=1e-12)
def test_alpha_additive_offset_recovered():
bench = np.array([0.01 * i for i in range(1, 21)])
asset = bench + 0.005
out = ta.Alpha(20, 0.0).batch(asset, bench)
out = _to_np(ta.Alpha(20, 0.0).batch(asset, bench))
assert math.isclose(out[-1], 0.005, rel_tol=1e-9)
def test_treynor_ratio_known_window():
bench = np.array([0.01 * i for i in range(1, 21)])
asset = 2.0 * bench
out = ta.TreynorRatio(20, 0.0).batch(asset, bench)
out = _to_np(ta.TreynorRatio(20, 0.0).batch(asset, bench))
assert math.isclose(out[-1], bench.mean(), rel_tol=1e-9)
def test_information_ratio_known_window():
asset = np.array([0.02, 0.04, 0.06, 0.08])
bench = np.array([0.01, 0.02, 0.03, 0.04])
out = ta.InformationRatio(4).batch(asset, bench)
out = _to_np(ta.InformationRatio(4).batch(asset, bench))
expected = 0.025 / math.sqrt(0.000_166_666_666_666_666_67)
assert math.isclose(out[-1], expected, rel_tol=1e-9)
@@ -443,7 +452,7 @@ def test_pairwise_beta_squared_price_is_two():
# and an undefined slope, which the indicator reports as 0).
b = np.array([100.0 + 10.0 * math.sin(i * 0.5) for i in range(20)])
a = b**2
out = ta.PairwiseBeta(5).batch(a, b)
out = _to_np(ta.PairwiseBeta(5).batch(a, b))
assert math.isclose(out[-1], 2.0, rel_tol=1e-9)
@@ -451,7 +460,7 @@ def test_pairwise_beta_inverse_price_is_minus_one():
# a = 1/b ⇒ a's log-returns are −1× b's ⇒ pairwise beta = −1.
b = np.array([100.0 + 10.0 * math.sin(i * 0.5) for i in range(20)])
a = 1.0 / b
out = ta.PairwiseBeta(5).batch(a, b)
out = _to_np(ta.PairwiseBeta(5).batch(a, b))
assert math.isclose(out[-1], -1.0, rel_tol=1e-9)
@@ -460,7 +469,7 @@ def test_pair_spread_zscore_flat_benchmark_sign():
# collapses to the sign of the last move: rising a ⇒ +1, falling a ⇒ −1.
a = np.array([100.0, 100.0, 110.0, 105.0, 130.0])
b = np.full_like(a, 100.0)
out = ta.PairSpreadZScore(2, 2).batch(a, b)
out = _to_np(ta.PairSpreadZScore(2, 2).batch(a, b))
assert math.isclose(out[-1], 1.0, abs_tol=1e-9)
assert math.isclose(out[-2], -1.0, abs_tol=1e-9)
@@ -473,7 +482,7 @@ def test_lead_lag_cross_correlation_negative_lead():
n = 60
a = np.array([sig(t - 2) for t in range(n)])
b = np.array([sig(t) for t in range(n)])
out = ta.LeadLagCrossCorrelation(12, 5).batch(a, b)
out = _to_np(ta.LeadLagCrossCorrelation(12, 5).batch(a, b))
assert int(out[-1, 0]) == -2
assert out[-1, 1] > 0.99
@@ -482,7 +491,7 @@ def test_cointegration_perfect_pair():
# a = 2*b + 5 exactly ⇒ hedge ratio 2, zero spread, degenerate ADF ⇒ 0.
b = np.array([100.0 + t for t in range(40)])
a = 2.0 * b + 5.0
out = ta.Cointegration(20, 1).batch(a, b)
out = _to_np(ta.Cointegration(20, 1).batch(a, b))
assert math.isclose(out[-1, 0], 2.0, rel_tol=1e-9)
assert math.isclose(out[-1, 1], 0.0, abs_tol=1e-6)
assert math.isclose(out[-1, 2], 0.0, abs_tol=1e-12)
@@ -493,7 +502,7 @@ def test_relative_strength_rising_ratio_is_overbought():
n = 20
a = np.array([100.0 + 2.0 * t for t in range(n)])
b = np.full(n, 100.0)
out = ta.RelativeStrengthAB(5, 5).batch(a, b)
out = _to_np(ta.RelativeStrengthAB(5, 5).batch(a, b))
assert out[-1, 0] > 1.0
assert math.isclose(out[-1, 2], 100.0, abs_tol=1e-9)
@@ -501,21 +510,21 @@ def test_relative_strength_rising_ratio_is_overbought():
def test_value_at_risk_known_window():
# returns -5..4 *0.01; q=0.05*9=0.45 -> -0.0455; VaR = 0.0455.
returns = np.array([i * 0.01 for i in range(-5, 5)])
out = ta.ValueAtRisk(10, 0.95).batch(returns)
out = _to_np(ta.ValueAtRisk(10, 0.95).batch(returns))
assert math.isclose(out[-1], 0.0455, rel_tol=1e-9)
def test_conditional_value_at_risk_known_window():
# tail = {-0.10}; CVaR = 0.10.
returns = np.array([i * 0.01 for i in range(-10, 10)])
out = ta.ConditionalValueAtRisk(20, 0.95).batch(returns)
out = _to_np(ta.ConditionalValueAtRisk(20, 0.95).batch(returns))
assert math.isclose(out[-1], 0.10, rel_tol=1e-9)
def test_calmar_ratio_known_path():
# returns [0.10, -0.20, 0.05]; equity 1.0->1.10->0.88->0.924;
# mdd = 0.20; mean = -0.01666...; Calmar = mean / 0.20.
out = ta.CalmarRatio(3).batch(np.array([0.10, -0.20, 0.05]))
out = _to_np(ta.CalmarRatio(3).batch(np.array([0.10, -0.20, 0.05])))
expected = ((0.10 - 0.20 + 0.05) / 3.0) / 0.20
assert math.isclose(out[-1], expected, rel_tol=1e-9)
@@ -523,7 +532,7 @@ def test_calmar_ratio_known_path():
def test_average_drawdown_known_window():
# window [100, 120, 90, 110]: one drawdown episode (peak 120, trough 90),
# never recovering -> depth (120-90)/120 = 0.25; one episode -> AvgDD = 0.25.
out = ta.AverageDrawdown(4).batch(np.array([100.0, 120.0, 90.0, 110.0]))
out = _to_np(ta.AverageDrawdown(4).batch(np.array([100.0, 120.0, 90.0, 110.0])))
expected = 0.25
assert math.isclose(out[-1], expected, rel_tol=1e-12)
@@ -536,7 +545,7 @@ def test_value_area_concentrated_volume_locates_poc():
high = np.array([100.5, 100.5, 100.5, 100.5, 110.5])
low = np.array([99.5, 99.5, 99.5, 99.5, 109.5])
volume = np.array([1.0, 1.0, 1.0, 1.0, 1000.0])
out = ta.ValueArea(5, 50, 0.70).batch(high, low, volume)
out = _to_np(ta.ValueArea(5, 50, 0.70).batch(high, low, volume))
poc = out[-1, 0]
assert 109.5 <= poc <= 110.5
# VAH >= POC >= VAL.
@@ -547,7 +556,7 @@ def test_initial_balance_locks_after_period():
# First two bars set IB = [99, 103]. Third bar (extreme) must be ignored.
high = np.array([102.0, 103.0, 200.0])
low = np.array([100.0, 99.0, 50.0])
out = ta.InitialBalance(2).batch(high, low)
out = _to_np(ta.InitialBalance(2).batch(high, low))
# Bar 0: IB = [100, 102]; Bar 1: IB locked at [99, 103]; Bar 2: unchanged.
np.testing.assert_allclose(out[0], [102.0, 100.0])
np.testing.assert_allclose(out[1], [103.0, 99.0])
@@ -560,7 +569,7 @@ def test_opening_range_breakout_distance_signed():
high = np.array([102.0, 103.0, 110.0, 110.0])
low = np.array([100.0, 101.0, 102.0, 90.0])
close = np.array([101.0, 102.0, 105.0, 95.0])
out = ta.OpeningRange(2).batch(high, low, close)
out = _to_np(ta.OpeningRange(2).batch(high, low, close))
assert math.isclose(out[2, 0], 103.0)
assert math.isclose(out[2, 1], 100.0)
assert math.isclose(out[2, 2], 105.0 - 101.5)
@@ -578,19 +587,19 @@ def test_inverse_fisher_saturates_for_large_input():
def test_super_smoother_constant_input_is_constant():
out = ta.SuperSmoother(20).batch(np.full(200, 50.0))
out = _to_np(ta.SuperSmoother(20).batch(np.full(200, 50.0)))
# Steady-state gain is 1, so a flat input stays flat.
np.testing.assert_allclose(out[-50:], 50.0, atol=1e-9)
def test_decycler_oscillator_flat_series_is_zero():
out = ta.DecyclerOscillator(10, 30).batch(np.full(80, 42.0))
out = _to_np(ta.DecyclerOscillator(10, 30).batch(np.full(80, 42.0)))
ready = out[~np.isnan(out)]
np.testing.assert_allclose(ready, 0.0, atol=1e-9)
def test_mama_constant_series_both_lines_converge_to_price():
out = ta.MAMA().batch(np.full(200, 100.0))
out = _to_np(ta.MAMA().batch(np.full(200, 100.0)))
last = out[-1]
# MAMA and FAMA both track price closely on a flat series.
assert abs(last[0] - 100.0) < 1.0
@@ -606,7 +615,7 @@ def test_td_setup_buy_setup_completes_at_minus_9_uptrend():
h = np.arange(2.0, 22.0)
l = h - 1.0
c = h - 0.5
out = ta.TDSetup(4, 9).batch(h, l, c)
out = _to_np(ta.TDSetup(4, 9).batch(h, l, c))
assert out[12] == pytest.approx(-9.0)
assert out[-1] == pytest.approx(-9.0)
@@ -615,7 +624,7 @@ def test_td_demarker_downtrend_pegs_at_zero():
n = 20
h = np.arange(30.0, 30.0 - n, -1.0)
l = h - 2.0
out = ta.TDDeMarker(5).batch(h, l)
out = _to_np(ta.TDDeMarker(5).batch(h, l))
assert out[-1] == pytest.approx(0.0)
@@ -626,7 +635,7 @@ def test_td_pressure_pure_bearish_yields_minus_100():
low = np.full(n, 9.0)
close = np.full(n, 9.0)
volume = np.full(n, 100.0)
out = ta.TDPressure(5).batch(open_, high, low, close, volume)
out = _to_np(ta.TDPressure(5).batch(open_, high, low, close, volume))
assert out[-1] == pytest.approx(-100.0)
@@ -638,7 +647,7 @@ def test_td_combo_uptrend_completes_to_minus_13():
high = np.arange(1.0, 1.0 + n) + 0.5
low = high - 1.0
close = high - 0.5
out = ta.TDCombo().batch(high, low, close)
out = _to_np(ta.TDCombo().batch(high, low, close))
assert out[-1] == pytest.approx(-13.0)
@@ -647,7 +656,7 @@ def test_td_countdown_uptrend_completes_to_minus_13():
high = np.arange(1.0, 1.0 + n) + 0.5
low = high - 1.0
close = high - 0.5
out = ta.TDCountdown().batch(high, low, close)
out = _to_np(ta.TDCountdown().batch(high, low, close))
assert out[-1] == pytest.approx(-13.0)
@@ -655,9 +664,9 @@ def test_td_range_projection_doji_reference():
# open=close=10, high=12, low=9 -> doji branch.
# pivot_sum = 12 + 9 + 2*10 = 41; half = 20.5.
# projHigh = 20.5 - 9 = 11.5; projLow = 20.5 - 12 = 8.5.
out = ta.TDRangeProjection().batch(
out = _to_np(ta.TDRangeProjection().batch(
np.array([10.0]), np.array([12.0]), np.array([9.0]), np.array([10.0])
)
))
assert out[0, 0] == pytest.approx(11.5)
assert out[0, 1] == pytest.approx(8.5)
@@ -685,7 +694,7 @@ def test_td_lines_uptrend_support_reference():
high = np.arange(1.0, 1.0 + n) + 0.5
low = high - 1.0
close = high - 0.5
out = ta.TDLines().batch(high, low, close)
out = _to_np(ta.TDLines().batch(high, low, close))
assert math.isnan(out[-1, 0])
assert out[-1, 1] == pytest.approx(4.5)
@@ -699,7 +708,7 @@ def test_td_risk_level_uptrend_sell_risk_reference():
high = np.arange(1.0, 1.0 + n) + 0.5
low = high - 1.0
close = high - 0.5
out = ta.TDRiskLevel().batch(high, low, close)
out = _to_np(ta.TDRiskLevel().batch(high, low, close))
assert math.isnan(out[12, 0])
assert out[12, 1] == pytest.approx(15.0)
@@ -729,7 +738,7 @@ def test_donchian_stop_window_extremes():
# 5-bar window of highs 1..5 and lows 0..4.
high = np.array([1.0, 2.0, 3.0, 4.0, 5.0])
low = np.array([0.0, 1.0, 2.0, 3.0, 4.0])
out = ta.DonchianStop(5).batch(high, low)
out = _to_np(ta.DonchianStop(5).batch(high, low))
# First 4 rows NaN, fifth row: stop_long = 0, stop_short = 5.
for i in range(4):
assert math.isnan(out[i, 0])
@@ -744,7 +753,7 @@ def test_hilo_activator_flat_market_holds_low_sma():
h = np.full(15, 11.0)
l = np.full(15, 9.0)
c = np.full(15, 10.0)
out = ta.HiLoActivator(3).batch(h, l, c)
out = _to_np(ta.HiLoActivator(3).batch(h, l, c))
# warmup_period == period + 1 == 4, so indices 0..2 are NaN; index 3 onwards is 9.
for i in range(3):
assert math.isnan(out[i])
@@ -757,7 +766,7 @@ def test_volty_stop_flat_market_constant_level():
h = np.full(20, 11.0)
l = np.full(20, 9.0)
c = np.full(20, 10.0)
out = ta.VoltyStop(5, 2.0).batch(h, l, c)
out = _to_np(ta.VoltyStop(5, 2.0).batch(h, l, c))
for i in range(4):
assert math.isnan(out[i])
for i in range(4, 20):
@@ -769,7 +778,7 @@ def test_yoyo_exit_flat_market_constant_level():
h = np.full(20, 11.0)
l = np.full(20, 9.0)
c = np.full(20, 10.0)
out = ta.YoyoExit(5, 2.0).batch(h, l, c)
out = _to_np(ta.YoyoExit(5, 2.0).batch(h, l, c))
for i in range(4):
assert math.isnan(out[i])
for i in range(4, 20):
@@ -781,7 +790,7 @@ def test_rvi_volatility_pure_uptrend_saturates_at_one_hundred():
# RVIVolatility saturates at 100. Renamed from the original ta.RVI in
# PR 42 to disambiguate from Family 02's Relative Vigor Index, which
# now owns the short ta.RVI name (candle input).
out = ta.RVIVolatility(5).batch(np.arange(1.0, 41.0, dtype=np.float64))
out = _to_np(ta.RVIVolatility(5).batch(np.arange(1.0, 41.0, dtype=np.float64)))
ready = out[~np.isnan(out)]
assert ready.size > 0
np.testing.assert_allclose(ready[-10:], 100.0, atol=1e-9)
@@ -791,7 +800,7 @@ def test_parkinson_volatility_zero_range_yields_zero():
# H == L every bar -> ln(H/L) = 0 -> Parkinson sigma is zero.
h = np.full(30, 10.0)
l = np.full(30, 10.0)
out = ta.ParkinsonVolatility(14, 252).batch(h, l)
out = _to_np(ta.ParkinsonVolatility(14, 252).batch(h, l))
ready = out[~np.isnan(out)]
assert ready.size > 0
np.testing.assert_allclose(ready, 0.0, atol=1e-12)
@@ -803,7 +812,7 @@ def test_garman_klass_zero_movement_yields_zero():
h = np.full(30, 10.0)
l = np.full(30, 10.0)
c = np.full(30, 10.0)
out = ta.GarmanKlassVolatility(14, 252).batch(o, h, l, c)
out = _to_np(ta.GarmanKlassVolatility(14, 252).batch(o, h, l, c))
ready = out[~np.isnan(out)]
assert ready.size > 0
np.testing.assert_allclose(ready, 0.0, atol=1e-12)
@@ -814,7 +823,7 @@ def test_rogers_satchell_zero_movement_yields_zero():
h = np.full(30, 10.0)
l = np.full(30, 10.0)
c = np.full(30, 10.0)
out = ta.RogersSatchellVolatility(14, 252).batch(o, h, l, c)
out = _to_np(ta.RogersSatchellVolatility(14, 252).batch(o, h, l, c))
ready = out[~np.isnan(out)]
assert ready.size > 0
np.testing.assert_allclose(ready, 0.0, atol=1e-12)
@@ -827,7 +836,7 @@ def test_yang_zhang_zero_movement_yields_zero():
h = np.full(30, 10.0)
l = np.full(30, 10.0)
c = np.full(30, 10.0)
out = ta.YangZhangVolatility(14, 252).batch(o, h, l, c)
out = _to_np(ta.YangZhangVolatility(14, 252).batch(o, h, l, c))
ready = out[~np.isnan(out)]
assert ready.size > 0
np.testing.assert_allclose(ready, 0.0, atol=1e-12)
@@ -952,7 +961,7 @@ def test_kyles_lambda_recovers_constant_impact():
size.append(sz)
is_buy.append(buy)
mids.append(mid)
out = ta.KylesLambda(6).batch(price, size, is_buy, mids)
out = _to_np(ta.KylesLambda(6).batch(price, size, is_buy, mids))
assert out[-1] == pytest.approx(0.5, abs=1e-9)
bindings/python/tests/test_new_indicators.py
+123 -110
View File
@@ -16,10 +16,23 @@ import pytest
import wickra as ta
def _eq_nan(a: np.ndarray, b: np.ndarray, tol: float = 1e-9) -> bool:
def _to_np(x) -> np.ndarray:
"""Normalize a Wickra batch result to a float64 ndarray.
Scalar batches return a stdlib ``array.array`` (1-D, wrapped zero-copy);
multi-output batches return a ``Matrix`` exposing ``.shape``/``.tolist()``.
"""
if isinstance(x, np.ndarray):
return x.astype(np.float64)
if hasattr(x, "tolist") and hasattr(x, "shape"):
return np.asarray(x.tolist(), dtype=np.float64)
return np.asarray(x, dtype=np.float64)
def _eq_nan(a, b, tol: float = 1e-9) -> bool:
"""Compare two float arrays treating NaN and matching-sign inf positions as equal."""
a = np.asarray(a, dtype=np.float64)
b = np.asarray(b, dtype=np.float64)
a = _to_np(a)
b = _to_np(b)
if a.shape != b.shape:
return False
both_nan = np.isnan(a) & np.isnan(b)
@@ -212,8 +225,8 @@ SCALAR_MULTI = {
@pytest.mark.parametrize("cls, args", SCALAR, ids=[c.__name__ for c, _ in SCALAR])
def test_scalar_streaming_matches_batch(cls, args, sine_prices):
batch = cls(*args).batch(sine_prices)
assert batch.shape == sine_prices.shape
assert batch.dtype == np.float64
assert len(batch) == len(sine_prices)
assert batch.typecode == "d"
streamer = cls(*args)
streamed = []
@@ -250,8 +263,8 @@ def test_pair_streaming_matches_batch(cls, args, sine_prices):
asset = np.ascontiguousarray(sine_prices.astype(np.float64))
bench = np.ascontiguousarray((sine_prices * 0.7 + 0.001).astype(np.float64))
batch = cls(*args).batch(asset, bench)
assert batch.shape == asset.shape
assert batch.dtype == np.float64
assert len(batch) == len(asset)
assert batch.typecode == "d"
streamer = cls(*args)
streamed = []
@@ -270,7 +283,7 @@ def test_lead_lag_detects_lead():
a = np.array([_ll_signal(t) for t in range(n)])
# b is a delayed by 3 ⇒ a leads b ⇒ lag = +3, correlation ≈ 1.
b = np.array([_ll_signal(t - 3) for t in range(n)])
out = ta.LeadLagCrossCorrelation(12, 5).batch(a, b)
out = _to_np(ta.LeadLagCrossCorrelation(12, 5).batch(a, b))
assert out.shape == (n, 2)
assert int(out[-1, 0]) == 3
assert out[-1, 1] > 0.99
@@ -298,7 +311,7 @@ def test_cointegration_detects_mean_reverting_pair():
b = np.array([50.0 + 0.5 * t for t in range(n)])
# a tracks 2*b with a small mean-reverting wobble ⇒ cointegrated.
a = 2.0 * b + 1.0 + 0.5 * np.sin(np.arange(n) * 0.6)
out = ta.Cointegration(40, 1).batch(a, b)
out = _to_np(ta.Cointegration(40, 1).batch(a, b))
assert out.shape == (n, 3)
assert abs(out[-1, 0] - 2.0) < 0.1 # hedge ratio
assert out[-1, 2] < -2.0 # ADF statistic: strongly mean-reverting
@@ -361,7 +374,7 @@ def test_relative_strength_constant_ratio():
n = 30
a = np.full(n, 200.0)
b = np.full(n, 100.0) # ratio is a constant 2
out = ta.RelativeStrengthAB(5, 5).batch(a, b)
out = _to_np(ta.RelativeStrengthAB(5, 5).batch(a, b))
assert out.shape == (n, 3)
assert math.isclose(out[-1, 0], 2.0, abs_tol=1e-12) # ratio
assert math.isclose(out[-1, 1], 2.0, abs_tol=1e-12) # ratio MA
@@ -1009,7 +1022,7 @@ def test_candle_scalar_streaming_matches_batch(name, ohlcv):
make, batch_call = CANDLE_SCALAR[name]
batch = batch_call(make(), high, low, close, volume)
assert batch.shape == close.shape
assert len(batch) == len(close)
streamer = make()
streamed = []
@@ -1527,7 +1540,7 @@ def test_td_pressure_streaming_matches_batch(ohlcv):
high, low, close, volume = ohlcv
open_ = close.copy() # TD Pressure needs open; reuse close as the open column.
batch = ta.TDPressure(5).batch(open_, high, low, close, volume)
assert batch.shape == close.shape
assert len(batch) == len(close)
streamer = ta.TDPressure(5)
streamed = []
@@ -1629,7 +1642,7 @@ def test_plus_dm_reference():
high = np.array([11.0, 12.0, 13.0, 14.0, 15.0])
low = np.array([9.0, 9.5, 10.0, 10.5, 11.0])
close = np.array([10.0, 11.0, 12.0, 13.0, 14.0])
out = ta.PLUS_DM(3).batch(high, low, close)
out = _to_np(ta.PLUS_DM(3).batch(high, low, close))
assert math.isnan(out[0]) and math.isnan(out[2])
assert out[3] == pytest.approx(3.0)
assert out[4] == pytest.approx(3.0)
@@ -1641,7 +1654,7 @@ def test_minus_dm_reference():
high = np.array([20.0, 19.5, 19.0, 18.5, 18.0])
low = np.array([18.0, 17.0, 16.0, 15.0, 14.0])
close = np.array([19.0, 18.0, 17.0, 16.0, 15.0])
out = ta.MINUS_DM(3).batch(high, low, close)
out = _to_np(ta.MINUS_DM(3).batch(high, low, close))
assert math.isnan(out[0]) and math.isnan(out[2])
assert out[3] == pytest.approx(3.0)
assert out[4] == pytest.approx(3.0)
@@ -1652,7 +1665,7 @@ def test_plus_di_reference():
high = np.array([101.0, 103.0, 105.0, 107.0, 109.0, 111.0])
low = np.array([99.5, 101.5, 103.5, 105.5, 107.5, 109.5])
close = np.array([100.5, 102.5, 104.5, 106.5, 108.5, 110.5])
out = ta.PLUS_DI(3).batch(high, low, close)
out = _to_np(ta.PLUS_DI(3).batch(high, low, close))
assert 0.0 < out[-1] <= 100.0
@@ -1661,7 +1674,7 @@ def test_minus_di_reference():
high = np.array([111.0, 109.0, 107.0, 105.0, 103.0, 101.0])
low = np.array([109.5, 107.5, 105.5, 103.5, 101.5, 99.5])
close = np.array([110.5, 108.5, 106.5, 104.5, 102.5, 100.5])
out = ta.MINUS_DI(3).batch(high, low, close)
out = _to_np(ta.MINUS_DI(3).batch(high, low, close))
assert 0.0 < out[-1] <= 100.0
@@ -1670,7 +1683,7 @@ def test_dx_reference():
high = np.array([101.0, 103.0, 105.0, 107.0, 109.0, 111.0])
low = np.array([99.5, 101.5, 103.5, 105.5, 107.5, 109.5])
close = np.array([100.5, 102.5, 104.5, 106.5, 108.5, 110.5])
out = ta.DX(3).batch(high, low, close)
out = _to_np(ta.DX(3).batch(high, low, close))
assert 50.0 < out[-1] <= 100.0
@@ -1679,13 +1692,13 @@ def test_mid_price_reference():
high = np.array([12.0, 14.0, 16.0])
low = np.array([8.0, 9.0, 10.0])
close = np.array([10.0, 11.0, 12.0])
out = ta.MIDPRICE(3).batch(high, low, close)
out = _to_np(ta.MIDPRICE(3).batch(high, low, close))
assert out[-1] == pytest.approx(12.0)
def test_mid_point_reference():
# Window {8, 12, 10}: (12 + 8) / 2 = 10.
out = ta.MIDPOINT(3).batch(np.array([8.0, 12.0, 10.0]))
out = _to_np(ta.MIDPOINT(3).batch(np.array([8.0, 12.0, 10.0])))
assert out[-1] == pytest.approx(10.0)
@@ -1711,8 +1724,8 @@ def test_linreg_intercept_and_tsf_reference():
def test_macdfix_matches_macd():
# MACDFIX(signal) is exactly MACD(12, 26, signal).
prices = 100.0 + np.sin(np.arange(80) * 0.3) * 5.0
fix = ta.MACDFIX(9).batch(prices)
classic = ta.MACD(12, 26, 9).batch(prices)
fix = _to_np(ta.MACDFIX(9).batch(prices))
classic = _to_np(ta.MACD(12, 26, 9).batch(prices))
np.testing.assert_allclose(fix, classic, equal_nan=True)
@@ -1720,7 +1733,7 @@ def test_nvi_reference():
# closes [10, 11], volumes [200, 100]: volume contracts -> NVI absorbs +10%.
# 1000 * (1 + 0.1) = 1100.
nvi = ta.NVI()
out = nvi.batch(np.array([10.0, 11.0]), np.array([200.0, 100.0]))
out = _to_np(nvi.batch(np.array([10.0, 11.0]), np.array([200.0, 100.0])))
assert out[0] == pytest.approx(1000.0)
assert out[1] == pytest.approx(1100.0)
@@ -1728,7 +1741,7 @@ def test_nvi_reference():
def test_pvi_reference():
# closes [10, 11], volumes [100, 200]: volume expands -> PVI absorbs +10%.
pvi = ta.PVI()
out = pvi.batch(np.array([10.0, 11.0]), np.array([100.0, 200.0]))
out = _to_np(pvi.batch(np.array([10.0, 11.0]), np.array([100.0, 200.0])))
assert out[0] == pytest.approx(1000.0)
assert out[1] == pytest.approx(1100.0)
@@ -1737,7 +1750,7 @@ def test_volume_oscillator_reference():
# fast=2, slow=4 over volumes [10, 20, 30, 40, 50]:
# bar 4 -> fast=(30+40)/2=35, slow=(10+20+30+40)/4=25 -> VO = 100*(35-25)/25 = 40.
vo = ta.VolumeOscillator(2, 4)
out = vo.batch(np.array([10.0, 20.0, 30.0, 40.0, 50.0]))
out = _to_np(vo.batch(np.array([10.0, 20.0, 30.0, 40.0, 50.0])))
assert math.isnan(out[2])
assert out[3] == pytest.approx(40.0)
assert out[4] == pytest.approx(1000.0 / 35.0)
@@ -1751,7 +1764,7 @@ def test_kvo_constant_series_is_zero():
low = np.full(60, 10.0)
close = np.full(60, 10.0)
volume = np.full(60, 100.0)
out = kvo.batch(high, low, close, volume)
out = _to_np(kvo.batch(high, low, close, volume))
for v in out[~np.isnan(out)]:
assert v == pytest.approx(0.0, abs=1e-12)
@@ -1766,7 +1779,7 @@ def test_wad_reference():
high = np.array([11.0, 13.0, 11.0])
low = np.array([9.0, 8.0, 7.0])
close = np.array([10.0, 12.0, 7.0])
out = ad.batch(high, low, close)
out = _to_np(ad.batch(high, low, close))
assert math.isnan(out[0])
assert out[1] == pytest.approx(4.0)
assert out[2] == pytest.approx(-1.0)
@@ -1780,7 +1793,7 @@ def test_anchored_vwap_reference():
low = np.array([10.0, 20.0, 30.0])
close = np.array([10.0, 20.0, 30.0])
volume = np.array([1.0, 1.0, 1.0])
out = avwap.batch(high, low, close, volume)
out = _to_np(avwap.batch(high, low, close, volume))
assert out[2] == pytest.approx(20.0)
@@ -1801,7 +1814,7 @@ def test_anchored_rsi_reference():
# bar2: sum_gain=1, sum_loss=2 -> rs=0.5 -> 100 - 100/1.5 = 33.3333
# bar3: sum_gain=4, sum_loss=2 -> rs=2.0 -> 100 - 100/3 = 66.6667
rsi = ta.AnchoredRSI()
out = rsi.batch(np.array([10.0, 11.0, 9.0, 12.0]))
out = _to_np(rsi.batch(np.array([10.0, 11.0, 9.0, 12.0])))
assert np.isnan(out[0])
assert out[1] == pytest.approx(100.0)
assert out[2] == pytest.approx(33.333333, abs=1e-4)
@@ -1843,7 +1856,7 @@ def test_tsv_reference():
tsv = ta.TSV(3)
close = np.array([10.0, 11.0, 13.0, 12.0, 14.0, 15.0])
volume = np.array([50.0, 100.0, 200.0, 150.0, 50.0, 200.0])
out = tsv.batch(close, volume)
out = _to_np(tsv.batch(close, volume))
assert math.isnan(out[0]) and math.isnan(out[1]) and math.isnan(out[2])
assert out[3] == pytest.approx(350.0)
assert out[4] == pytest.approx(350.0)
@@ -1857,7 +1870,7 @@ def test_vzo_strictly_rising_saturates_to_plus_100():
vzo = ta.VZO(5)
close = np.array([10.0 + i for i in range(60)])
volume = np.full(60, 100.0)
out = vzo.batch(close, volume)
out = _to_np(vzo.batch(close, volume))
last = out[~np.isnan(out)][-1]
assert last == pytest.approx(100.0)
@@ -1868,7 +1881,7 @@ def test_market_facilitation_index_reference():
high = np.array([12.0])
low = np.array([8.0])
volume = np.array([200.0])
out = mfi_bw.batch(high, low, volume)
out = _to_np(mfi_bw.batch(high, low, volume))
assert out[0] == pytest.approx(0.02)
@@ -1879,7 +1892,7 @@ def test_demand_index_constant_series_is_zero():
low = np.full(60, 10.0)
close = np.full(60, 10.0)
volume = np.full(60, 100.0)
out = di.batch(high, low, close, volume)
out = _to_np(di.batch(high, low, close, volume))
for v in out[~np.isnan(out)]:
assert v == pytest.approx(0.0, abs=1e-12)
@@ -1891,13 +1904,13 @@ def test_chaikin_money_flow_reference():
def test_linear_regression_reference():
out = ta.LinearRegression(3).batch(np.array([1.0, 2.0, 9.0]))
out = _to_np(ta.LinearRegression(3).batch(np.array([1.0, 2.0, 9.0])))
assert math.isnan(out[0]) and math.isnan(out[1])
assert out[2] == pytest.approx(8.0)
def test_linreg_slope_reference():
out = ta.LinRegSlope(3).batch(np.array([1.0, 2.0, 9.0]))
out = _to_np(ta.LinRegSlope(3).batch(np.array([1.0, 2.0, 9.0])))
assert math.isnan(out[0]) and math.isnan(out[1])
assert out[2] == pytest.approx(4.0)
@@ -1916,7 +1929,7 @@ def test_true_range_reference():
def test_linreg_angle_reference():
# A series rising by 1 per step has slope 1, and atan(1) = 45 degrees.
out = ta.LinRegAngle(5).batch(np.array([1.0, 2.0, 3.0, 4.0, 5.0, 6.0]))
out = _to_np(ta.LinRegAngle(5).batch(np.array([1.0, 2.0, 3.0, 4.0, 5.0, 6.0])))
assert out[4] == pytest.approx(45.0)
@@ -1924,16 +1937,16 @@ def test_wave_trend_flat_market_yields_zero():
# On a perfectly flat market the flat-tolerance guard keeps both lines
# at exactly zero (otherwise the ratio ci = (ap - esa) / (0.015 * d)
# would explode on the first esa ULP).
out = ta.WaveTrend.classic().batch(
out = _to_np(ta.WaveTrend.classic().batch(
np.full(80, 10.0), np.full(80, 10.0), np.full(80, 10.0)
)
))
last = out[~np.isnan(out[:, 0])][-1]
assert last[0] == 0.0
assert last[1] == 0.0
def test_kst_classic_constants_yield_zero():
out = ta.KST.classic().batch(np.full(120, 100.0))
out = _to_np(ta.KST.classic().batch(np.full(120, 100.0)))
last_row = out[~np.isnan(out[:, 0])][-1]
assert last_row[0] == pytest.approx(0.0)
assert last_row[1] == pytest.approx(0.0)
@@ -1943,13 +1956,13 @@ def test_tii_pure_uptrend_saturates_at_100():
# On a strictly increasing series every close sits above the lagging
# SMA, so every deviation is positive and TII reaches 100.
prices = np.arange(80, dtype=np.float64) + 100.0
out = ta.TII(10, 5).batch(prices)
out = _to_np(ta.TII(10, 5).batch(prices))
last = out[~np.isnan(out)][-1]
assert last == pytest.approx(100.0)
def test_tii_flat_market_yields_50():
out = ta.TII(5, 4).batch(np.full(30, 10.0))
out = _to_np(ta.TII(5, 4).batch(np.full(30, 10.0)))
last = out[~np.isnan(out)][-1]
assert last == 50.0
@@ -1961,7 +1974,7 @@ def test_rwi_reference_uptrend_dominates_low_line():
high = base + 1.0
low = base - 0.5
close = base + 0.5
out = ta.RWI(14).batch(high, low, close)
out = _to_np(ta.RWI(14).batch(high, low, close))
last_row = out[~np.isnan(out[:, 0])][-1]
assert last_row[0] > last_row[1], f"RWI_High {last_row[0]} must dominate RWI_Low {last_row[1]}"
assert last_row[0] > 1.0
@@ -1975,14 +1988,14 @@ def test_adxr_reference_on_pure_uptrend():
high = base + 1.0
low = base - 0.5
close = base + 0.5
out = ta.ADXR(5).batch(high, low, close)
out = _to_np(ta.ADXR(5).batch(high, low, close))
last = out[~np.isnan(out)][-1]
assert last == pytest.approx(100.0)
def test_z_score_reference():
# Window [1, 3]: mean 2, population stddev 1; latest 3 -> z = 1.
out = ta.ZScore(2).batch(np.array([1.0, 3.0]))
out = _to_np(ta.ZScore(2).batch(np.array([1.0, 3.0])))
assert math.isnan(out[0])
assert out[1] == pytest.approx(1.0)
@@ -1992,44 +2005,44 @@ def test_z_score_reference():
def test_variance_reference():
# Variance(3) of [2, 4, 6]: mean 4, variance (4 + 0 + 4) / 3 = 8/3.
out = ta.Variance(3).batch(np.array([2.0, 4.0, 6.0]))
out = _to_np(ta.Variance(3).batch(np.array([2.0, 4.0, 6.0])))
assert math.isnan(out[1])
assert out[2] == pytest.approx(8.0 / 3.0)
def test_coefficient_of_variation_reference():
# CV(3) of [2, 4, 6]: sd / mean = sqrt(8/3) / 4.
out = ta.CoefficientOfVariation(3).batch(np.array([2.0, 4.0, 6.0]))
out = _to_np(ta.CoefficientOfVariation(3).batch(np.array([2.0, 4.0, 6.0])))
assert out[2] == pytest.approx(math.sqrt(8.0 / 3.0) / 4.0)
def test_skewness_symmetric_window_is_zero():
# Symmetric window has zero Pearson skewness.
out = ta.Skewness(5).batch(np.array([-2.0, -1.0, 0.0, 1.0, 2.0]))
out = _to_np(ta.Skewness(5).batch(np.array([-2.0, -1.0, 0.0, 1.0, 2.0])))
assert out[4] == pytest.approx(0.0, abs=1e-9)
def test_kurtosis_two_point_distribution_minimum():
# Alternating {-1, 1} has m4/m2² = 1, so excess kurtosis = -2.
out = ta.Kurtosis(4).batch(np.array([-1.0, 1.0, -1.0, 1.0]))
out = _to_np(ta.Kurtosis(4).batch(np.array([-1.0, 1.0, -1.0, 1.0])))
assert out[3] == pytest.approx(-2.0, abs=1e-9)
def test_standard_error_perfect_line_is_zero():
# Residuals are zero on a perfectly linear series.
out = ta.StandardError(5).batch(np.linspace(1.0, 20.0, num=20, dtype=np.float64))
out = _to_np(ta.StandardError(5).batch(np.linspace(1.0, 20.0, num=20, dtype=np.float64)))
finite = out[~np.isnan(out)]
assert np.allclose(finite, 0.0, atol=1e-9)
def test_detrended_std_dev_perfect_line_is_zero():
out = ta.DetrendedStdDev(5).batch(np.linspace(1.0, 20.0, num=20, dtype=np.float64))
out = _to_np(ta.DetrendedStdDev(5).batch(np.linspace(1.0, 20.0, num=20, dtype=np.float64)))
finite = out[~np.isnan(out)]
assert np.allclose(finite, 0.0, atol=1e-9)
def test_r_squared_perfect_line_is_one():
out = ta.RSquared(5).batch(np.linspace(1.0, 20.0, num=20, dtype=np.float64))
out = _to_np(ta.RSquared(5).batch(np.linspace(1.0, 20.0, num=20, dtype=np.float64)))
finite = out[~np.isnan(out)]
assert np.allclose(finite, 1.0, atol=1e-9)
@@ -2037,35 +2050,35 @@ def test_r_squared_perfect_line_is_one():
def test_median_absolute_deviation_ignores_single_outlier():
# 9 equal values + 1 huge outlier: MAD is still 0 (more than half agree).
prices = np.array([5.0] * 9 + [1000.0], dtype=np.float64)
out = ta.MedianAbsoluteDeviation(10).batch(prices)
out = _to_np(ta.MedianAbsoluteDeviation(10).batch(prices))
assert out[9] == pytest.approx(0.0, abs=1e-12)
def test_autocorrelation_alternating_series_negative():
# ±1 alternating: lag-1 ACF must be strongly negative.
prices = np.array([-1.0 if i % 2 == 0 else 1.0 for i in range(20)], dtype=np.float64)
out = ta.Autocorrelation(10, 1).batch(prices)
out = _to_np(ta.Autocorrelation(10, 1).batch(prices))
assert out[-1] < -0.5
def test_hurst_exponent_trending_above_half():
# A clean monotone ramp is the textbook persistent series.
prices = np.arange(200, dtype=np.float64)
out = ta.HurstExponent(100, 4).batch(prices)
out = _to_np(ta.HurstExponent(100, 4).batch(prices))
assert out[-1] > 0.5
def test_pearson_correlation_perfect_positive_is_one():
x = np.arange(10, dtype=np.float64)
y = 2.0 * x + 3.0
out = ta.PearsonCorrelation(5).batch(x, y)
out = _to_np(ta.PearsonCorrelation(5).batch(x, y))
assert out[-1] == pytest.approx(1.0, abs=1e-9)
def test_beta_perfect_two_to_one():
benchmark = np.arange(10, dtype=np.float64)
asset = 2.0 * benchmark
out = ta.Beta(5).batch(asset, benchmark)
out = _to_np(ta.Beta(5).batch(asset, benchmark))
assert out[-1] == pytest.approx(2.0, abs=1e-9)
@@ -2073,7 +2086,7 @@ def test_spearman_correlation_monotone_nonlinear_is_one():
# y = x^3 is monotone non-linear; Spearman = 1 (Pearson would not be).
x = np.arange(1.0, 11.0, dtype=np.float64)
y = x**3
out = ta.SpearmanCorrelation(5).batch(x, y)
out = _to_np(ta.SpearmanCorrelation(5).batch(x, y))
assert out[-1] == pytest.approx(1.0, abs=1e-9)
@@ -2109,32 +2122,32 @@ def test_mama_batch_shape_and_streaming_equivalence(sine_prices):
def test_inverse_fisher_transform_zero_input_yields_zero():
out = ta.InverseFisherTransform(1.0).batch(np.array([0.0, 0.0, 0.0]))
out = _to_np(ta.InverseFisherTransform(1.0).batch(np.array([0.0, 0.0, 0.0])))
np.testing.assert_allclose(out, [0.0, 0.0, 0.0], atol=1e-12)
def test_fisher_transform_flat_series_is_zero():
# Zero range -> the normaliser yields 0, and tanh(0) chain stays at 0.
out = ta.FisherTransform(5).batch(np.full(20, 42.0))
out = _to_np(ta.FisherTransform(5).batch(np.full(20, 42.0)))
ready = out[~np.isnan(out)]
assert np.all(np.abs(ready) < 1e-6)
def test_decycler_flat_series_passes_through():
# High-pass of a flat input is zero, so the decycler equals the input.
out = ta.Decycler(20).batch(np.full(30, 100.0))
out = _to_np(ta.Decycler(20).batch(np.full(30, 100.0)))
ready = out[~np.isnan(out)]
np.testing.assert_allclose(ready, 100.0, atol=1e-9)
def test_center_of_gravity_flat_series_is_zero():
out = ta.CenterOfGravity(5).batch(np.full(20, 7.0))
out = _to_np(ta.CenterOfGravity(5).batch(np.full(20, 7.0)))
ready = out[~np.isnan(out)]
np.testing.assert_allclose(ready, 0.0, atol=1e-12)
def test_super_smoother_first_two_outputs_equal_inputs():
out = ta.SuperSmoother(10).batch(np.array([100.0, 101.0, 102.0]))
out = _to_np(ta.SuperSmoother(10).batch(np.array([100.0, 101.0, 102.0])))
# The 2-pole filter is seeded with raw values for the first two bars.
assert out[0] == pytest.approx(100.0)
assert out[1] == pytest.approx(101.0)
@@ -2145,7 +2158,7 @@ def test_td_setup_pure_uptrend_reaches_minus_9():
h = np.arange(2.0, 22.0)
l = h - 1.0
c = h - 0.5
out = ta.TDSetup(4, 9).batch(h, l, c)
out = _to_np(ta.TDSetup(4, 9).batch(h, l, c))
# Setup completes at index 12 (warmup is 5 -> first emit at index 4 with
# value -1, increments to -9 at index 12).
assert out[12] == pytest.approx(-9.0)
@@ -2156,7 +2169,7 @@ def test_td_demarker_uptrend_pegs_at_one():
# -> indicator == 1 after warmup.
h = np.arange(11.0, 31.0)
l = h - 2.0
out = ta.TDDeMarker(5).batch(h, l)
out = _to_np(ta.TDDeMarker(5).batch(h, l))
assert out[-1] == pytest.approx(1.0)
@@ -2164,7 +2177,7 @@ def test_td_demarker_flat_market_emits_05():
# All highs and lows equal -> denominator is zero -> neutral fallback 0.5.
h = np.full(20, 11.0)
l = np.full(20, 9.0)
out = ta.TDDeMarker(5).batch(h, l)
out = _to_np(ta.TDDeMarker(5).batch(h, l))
assert out[-1] == pytest.approx(0.5)
@@ -2177,7 +2190,7 @@ def test_td_pressure_pure_bullish_yields_100():
low = np.full(n, 9.0)
close = np.full(n, 11.0)
volume = np.full(n, 100.0)
out = ta.TDPressure(5).batch(open_, high, low, close, volume)
out = _to_np(ta.TDPressure(5).batch(open_, high, low, close, volume))
assert out[-1] == pytest.approx(100.0)
@@ -2189,7 +2202,7 @@ def test_td_combo_uptrend_saturates_at_minus_13():
high = np.arange(1.0, 1.0 + n) + 0.5
low = high - 1.0
close = high - 0.5
out = ta.TDCombo().batch(high, low, close)
out = _to_np(ta.TDCombo().batch(high, low, close))
assert out[-1] == pytest.approx(-13.0)
@@ -2198,7 +2211,7 @@ def test_td_countdown_uptrend_saturates_at_minus_13():
high = np.arange(1.0, 1.0 + n) + 0.5
low = high - 1.0
close = high - 0.5
out = ta.TDCountdown().batch(high, low, close)
out = _to_np(ta.TDCountdown().batch(high, low, close))
assert out[-1] == pytest.approx(-13.0)
@@ -2209,7 +2222,7 @@ def test_td_lines_uptrend_sets_support_at_first_run_low():
high = np.arange(1.0, 1.0 + n) + 0.5
low = high - 1.0
close = high - 0.5
out = ta.TDLines().batch(high, low, close)
out = _to_np(ta.TDLines().batch(high, low, close))
# support is column 1; resistance is NaN at -1.
assert math.isnan(out[-1, 0])
# low at idx 4 = 5 + 0.5 - 1.0 = 4.5.
@@ -2220,9 +2233,9 @@ def test_td_range_projection_bullish_bar_reference():
# open=10, high=12, low=9, close=11 (close > open) ->
# pivot_sum = 2*12 + 9 + 11 = 44; half = 22.
# projHigh = 22 - 9 = 13; projLow = 22 - 12 = 10.
out = ta.TDRangeProjection().batch(
out = _to_np(ta.TDRangeProjection().batch(
np.array([10.0]), np.array([12.0]), np.array([9.0]), np.array([11.0])
)
))
assert out[0, 0] == pytest.approx(13.0)
assert out[0, 1] == pytest.approx(10.0)
@@ -2252,7 +2265,7 @@ def test_td_risk_level_uptrend_sets_sell_risk():
high = np.arange(1.0, 1.0 + n) + 0.5
low = high - 1.0
close = high - 0.5
out = ta.TDRiskLevel().batch(high, low, close)
out = _to_np(ta.TDRiskLevel().batch(high, low, close))
# buy_risk is column 0; sell_risk is column 1.
assert math.isnan(out[12, 0])
assert out[12, 1] == pytest.approx(15.0)
@@ -2346,7 +2359,7 @@ def test_zigzag_confirms_after_threshold_reversal():
def test_ma_envelope_reference():
# SMA([10, 20, 30]) = 20; with percent = 0.10: upper = 22, lower = 18.
out = ta.MaEnvelope(3, 0.10).batch(np.array([10.0, 20.0, 30.0]))
out = _to_np(ta.MaEnvelope(3, 0.10).batch(np.array([10.0, 20.0, 30.0])))
assert math.isnan(out[0, 0]) and math.isnan(out[1, 0])
assert out[2, 0] == pytest.approx(22.0) # upper
assert out[2, 1] == pytest.approx(20.0) # middle
@@ -2363,9 +2376,9 @@ def test_acceleration_bands_reference():
def test_atr_bands_reference():
# Five identical bars (h=11, l=9, c=10) → ATR=2, close=10, mult=3:
# upper=16, middle=10, lower=4.
out = ta.AtrBands(5, 3.0).batch(
out = _to_np(ta.AtrBands(5, 3.0).batch(
np.array([11.0] * 5), np.array([9.0] * 5), np.array([10.0] * 5)
)
))
assert math.isnan(out[3, 0])
assert out[4, 0] == pytest.approx(16.0)
assert out[4, 1] == pytest.approx(10.0)
@@ -2374,9 +2387,9 @@ def test_atr_bands_reference():
def test_hurst_channel_reference():
# Five identical (h=12, l=8, c=10): SMA(close)=10, range=4, mult=0.5.
out = ta.HurstChannel(5, 0.5).batch(
out = _to_np(ta.HurstChannel(5, 0.5).batch(
np.array([12.0] * 5), np.array([8.0] * 5), np.array([10.0] * 5)
)
))
assert out[4, 0] == pytest.approx(12.0)
assert out[4, 1] == pytest.approx(10.0)
assert out[4, 2] == pytest.approx(8.0)
@@ -2385,7 +2398,7 @@ def test_hurst_channel_reference():
def test_linreg_channel_reference():
# period 3 over [1, 2, 9]: line y=4x, endpoint=8, residuals=[1, -2, 1],
# population sigma=sqrt(2); mult=2 → upper=8+2√2, lower=8-2√2.
out = ta.LinRegChannel(3, 2.0).batch(np.array([1.0, 2.0, 9.0]))
out = _to_np(ta.LinRegChannel(3, 2.0).batch(np.array([1.0, 2.0, 9.0])))
s = math.sqrt(2.0)
assert out[2, 0] == pytest.approx(8.0 + 2.0 * s)
assert out[2, 1] == pytest.approx(8.0)
@@ -2395,7 +2408,7 @@ def test_linreg_channel_reference():
def test_standard_error_bands_reference():
# Same [1, 2, 9] with n=3: SSE=6, n-2=1, stderr=sqrt(6); mult=2 →
# upper=8+2√6, lower=8-2√6.
out = ta.StandardErrorBands(3, 2.0).batch(np.array([1.0, 2.0, 9.0]))
out = _to_np(ta.StandardErrorBands(3, 2.0).batch(np.array([1.0, 2.0, 9.0])))
s = math.sqrt(6.0)
assert out[2, 0] == pytest.approx(8.0 + 2.0 * s)
assert out[2, 1] == pytest.approx(8.0)
@@ -2404,9 +2417,9 @@ def test_standard_error_bands_reference():
def test_double_bollinger_orders_bands():
# On a non-trivial dispersion, outer >= inner >= middle >= -inner >= -outer.
out = ta.DoubleBollinger(5, 1.0, 2.0).batch(
out = _to_np(ta.DoubleBollinger(5, 1.0, 2.0).batch(
np.array([1.0, 5.0, 2.0, 4.0, 3.0, 6.0])
)
))
v = out[5]
assert v[0] >= v[1] >= v[2] >= v[3] >= v[4]
@@ -2427,7 +2440,7 @@ def test_ttm_squeeze_flat_market():
# Zero volatility: BB and KC both collapse to a point → squeeze=1.0,
# momentum=0.0.
candles_h = np.array([10.0] * 25)
out = ta.TtmSqueeze(20, 2.0, 1.5).batch(candles_h, candles_h, candles_h)
out = _to_np(ta.TtmSqueeze(20, 2.0, 1.5).batch(candles_h, candles_h, candles_h))
assert out[24, 0] == pytest.approx(1.0)
assert out[24, 1] == pytest.approx(0.0)
@@ -2436,7 +2449,7 @@ def test_fractal_chaos_bands_detects_peak_and_trough():
# Sequence that creates one fractal high (i=2) and one low (i=3).
h = np.array([1.0, 2.0, 5.0, 3.0, 2.0, 1.0, 2.0])
l = np.array([1.0, 2.0, 3.0, 0.5, 2.0, 1.0, 2.0])
out = ta.FractalChaosBands(2).batch(h, l)
out = _to_np(ta.FractalChaosBands(2).batch(h, l))
# First bar with both bands set is index 5.
assert math.isnan(out[4, 0])
assert out[5, 0] == pytest.approx(5.0)
@@ -2989,7 +3002,7 @@ def test_spread_ar1_coefficient_reference():
# Spread a - b grows by exactly 1 each bar (unit root) => rho == 1.
a = np.array([2.0 * i for i in range(40)])
b = np.array([float(i) for i in range(40)])
out = ta.SpreadAr1Coefficient(20).batch(a, b)
out = _to_np(ta.SpreadAr1Coefficient(20).batch(a, b))
assert math.isclose(out[-1], 1.0, abs_tol=1e-9)
@@ -2998,7 +3011,7 @@ def test_elder_ray_reference():
high = np.array([11.0, 13.0, 16.0])
low = np.array([9.0, 11.0, 13.0])
close = np.array([10.0, 12.0, 14.0])
out = er.batch(high, low, close)
out = _to_np(er.batch(high, low, close))
# EMA(3) seeds at the third bar with mean close 12; bar high 16 -> bull 4,
# low 13 -> bear 1.
assert out[2][0] == pytest.approx(4.0)
@@ -3011,7 +3024,7 @@ def test_imi_reference():
high = np.array([12.0, 12.0, 13.0])
low = np.array([9.0, 9.0, 9.0])
close = np.array([11.0, 10.0, 12.0])
out = imi.batch(open_, high, low, close)
out = _to_np(imi.batch(open_, high, low, close))
# bodies +1, -1, +2 -> gain 3, loss 1 -> 100 * 3 / 4 = 75.
assert math.isnan(out[0])
assert math.isnan(out[1])
@@ -3022,7 +3035,7 @@ def test_qstick_reference():
q = ta.Qstick(3)
open_ = np.array([10.0, 10.0, 10.0])
close = np.array([11.0, 11.0, 11.0])
out = q.batch(open_, close)
out = _to_np(q.batch(open_, close))
# Each body is close - open = 1; SMA(3) of [1, 1, 1] = 1.
assert math.isnan(out[0])
assert math.isnan(out[1])
@@ -3034,7 +3047,7 @@ def test_ttm_trend_reference():
high = np.array([13.0, 13.0, 13.0])
low = np.array([9.0, 9.0, 9.0])
close = np.array([12.0, 12.0, 12.0])
out = t.batch(high, low, close)
out = _to_np(t.batch(high, low, close))
# Median (13 + 9) / 2 = 11; close 12 is above the SMA(3) reference -> +1.
assert math.isnan(out[0])
assert out[2] == pytest.approx(1.0)
@@ -3043,7 +3056,7 @@ def test_ttm_trend_reference():
def test_trend_strength_index_reference():
tsi = ta.TREND_STRENGTH_INDEX(10)
closes = np.arange(10, dtype=float)
out = tsi.batch(closes)
out = _to_np(tsi.batch(closes))
# A clean ramp is a perfect uptrend -> signed r^2 = +1.
assert math.isclose(out[-1], 1.0, abs_tol=1e-9)
@@ -3051,7 +3064,7 @@ def test_trend_strength_index_reference():
def test_polarized_fractal_efficiency_reference():
pfe = ta.POLARIZED_FRACTAL_EFFICIENCY(5, 3)
closes = np.arange(20, dtype=float)
out = pfe.batch(closes)
out = _to_np(pfe.batch(closes))
# On a straight ramp the path equals the diagonal -> efficiency 1 -> +100.
assert math.isclose(out[-1], 100.0, abs_tol=1e-9)
@@ -3059,7 +3072,7 @@ def test_polarized_fractal_efficiency_reference():
def test_wave_pm_reference():
wpm = ta.WAVE_PM(10, 3)
closes = np.arange(60, dtype=float) * 5.0
out = wpm.batch(closes)
out = _to_np(wpm.batch(closes))
# Constant-slope ramp: momentum equals its energy -> 100 * (1 - e^-0.5).
baseline = 100.0 * (1.0 - math.exp(-0.5))
assert math.isclose(out[-1], baseline, abs_tol=1e-9)
@@ -3071,7 +3084,7 @@ def test_gator_oscillator_reference():
high = np.full(n, 11.0)
low = np.full(n, 9.0)
close = np.full(n, 10.0)
out = g.batch(high, low, close)
out = _to_np(g.batch(high, low, close))
# Constant median collapses all three Alligator lines -> both bars zero.
assert out[-1][0] == pytest.approx(0.0)
assert out[-1][1] == pytest.approx(0.0)
@@ -3081,7 +3094,7 @@ def test_kase_permission_stochastic_reference():
k = ta.KasePermissionStochastic(4, 2)
n = 20
flat = np.full(n, 10.0)
out = k.batch(flat, flat, flat)
out = _to_np(k.batch(flat, flat, flat))
# HH == LL -> raw %K defaults to the neutral 50 -> both lines at 50.
assert out[-1][0] == pytest.approx(50.0)
assert out[-1][1] == pytest.approx(50.0)
@@ -3416,7 +3429,7 @@ def test_new_indicators_expose_lifecycle():
def test_ichimoku_batch_shape_and_warmup(ohlcv):
high, low, close, _ = ohlcv
ichi = ta.Ichimoku()
out = ichi.batch(high, low, close)
out = _to_np(ichi.batch(high, low, close))
assert out.shape == (close.size, 5)
# Warmup is 77 for the classic (9, 26, 52, 26) configuration.
assert ichi.warmup_period() == 77
@@ -3454,7 +3467,7 @@ def test_ichimoku_chikou_is_close_displacement_back():
close = np.full(n, 100.0)
high = close + 1.0
low = close - 1.0
out = ta.Ichimoku().batch(high, low, close)
out = _to_np(ta.Ichimoku().batch(high, low, close))
# Displacement = 26, so chikou is defined from bar 25 onwards.
for i in range(25, n):
assert out[i, 4] == pytest.approx(100.0)
@@ -3706,7 +3719,7 @@ def test_orderbook_indicators_streaming_equals_batch():
streamed = np.array(
[streamer.update(*snap) for snap in snaps], dtype=np.float64
)
assert batch.shape == (len(snaps),)
assert len(batch) == len(snaps)
assert _eq_nan(batch, streamed)
@@ -3731,7 +3744,7 @@ def test_tradeflow_indicators_streaming_equals_batch():
[streamer.update(price[i], size[i], is_buy[i]) for i in range(n)],
dtype=np.float64,
)
assert batch.shape == (n,)
assert len(batch) == n
assert _eq_nan(batch, streamed)
@@ -3779,7 +3792,7 @@ def test_price_impact_indicators_streaming_equals_batch():
[streamer.update(price[i], size[i], is_buy[i], mid[i]) for i in range(n)],
dtype=np.float64,
)
assert batch.shape == (n,)
assert len(batch) == n
assert _eq_nan(batch, streamed)
@@ -3793,7 +3806,7 @@ def test_footprint_streaming_equals_batch():
assert len(batch) == n
for i in range(n):
streamed = streamer.update(price[i], size[i], is_buy[i])
assert np.array_equal(streamed, batch[i])
assert np.array_equal(_to_np(streamed), _to_np(batch[i]))
def test_funding_indicators_streaming_equals_batch():
@@ -3809,7 +3822,7 @@ def test_funding_indicators_streaming_equals_batch():
streamed = np.array(
[streamer.update(rate[i]) for i in range(n)], dtype=np.float64
)
assert batch.shape == (n,)
assert len(batch) == n
assert _eq_nan(batch, streamed)
@@ -3833,7 +3846,7 @@ def test_advance_decline_streaming_equals_batch():
],
dtype=np.float64,
)
assert batch.shape == (3,)
assert len(batch) == 3
assert _eq_nan(batch, streamed)
# Cumulative line: +2 -> 0 -> 0.
assert list(batch) == [2.0, 0.0, 0.0]
@@ -3856,7 +3869,7 @@ def _breadth_streaming_equals_batch(indicator, change, volume, new_high, new_low
],
dtype=np.float64,
)
assert batch.shape == (len(change),)
assert len(batch) == len(change)
assert _eq_nan(batch, streamed)
return batch
@@ -4040,7 +4053,7 @@ def test_funding_basis_streaming_equals_batch():
streamed = np.array(
[streamer.update(mark[i], index[i]) for i in range(n)], dtype=np.float64
)
assert batch.shape == (n,)
assert len(batch) == n
assert _eq_nan(batch, streamed)
@@ -4050,7 +4063,7 @@ def test_open_interest_delta_streaming_equals_batch():
batch = ta.OpenInterestDelta().batch(oi)
streamer = ta.OpenInterestDelta()
streamed = np.array([streamer.update(oi[i]) for i in range(n)], dtype=np.float64)
assert batch.shape == (n,)
assert len(batch) == n
assert _eq_nan(batch, streamed)
@@ -4067,7 +4080,7 @@ def test_oi_flow_indicators_streaming_equals_batch():
streamed = np.array(
[streamer.update(oi[i], mark[i]) for i in range(n)], dtype=np.float64
)
assert batch.shape == (n,)
assert len(batch) == n
assert _eq_nan(batch, streamed)
# OIWeighted; update(mark_price, open_interest).
@@ -4121,7 +4134,7 @@ def test_basis_indicators_streaming_equals_batch():
streamed = np.array(
[streamer.update(futures[i], index[i]) for i in range(n)], dtype=np.float64
)
assert batch.shape == (n,)
assert len(batch) == n
assert _eq_nan(batch, streamed)
# CalendarSpread; update(futures_price, mark_price).
@@ -4166,7 +4179,7 @@ def test_b16_derivatives_streaming_equals_batch():
streamed = np.array(
[streamer.update(oi[i], long_sz[i], short_sz[i]) for i in range(n)], dtype=np.float64
)
assert batch.shape == (n,)
assert len(batch) == n
assert _eq_nan(batch, streamed)
# OiToVolumeRatio; update(open_interest, taker_buy_volume, taker_sell_volume).
@@ -4210,7 +4223,7 @@ def test_renko_bars_reference():
def test_renko_bars_batch_shape():
r = ta.RenkoBars(1.0)
out = r.batch(np.array([10.0, 11.0, 12.0, 13.0]))
out = _to_np(r.batch(np.array([10.0, 11.0, 12.0, 13.0])))
assert out.shape == (3, 3)
np.testing.assert_allclose(out[:, 2], [1.0, 1.0, 1.0])
@@ -4247,7 +4260,7 @@ def test_range_bars_reference():
def test_range_bars_batch_shape():
rb = ta.RangeBars(1.0)
out = rb.batch(np.array([10.0, 11.0, 12.0, 13.0]))
out = _to_np(rb.batch(np.array([10.0, 11.0, 12.0, 13.0])))
assert out.shape == (3, 3)
np.testing.assert_allclose(out[:, 2], [1.0, 1.0, 1.0])
@@ -4264,7 +4277,7 @@ def test_tick_bars_batch_shape():
tb = ta.TickBars(2)
col = np.array([10.0, 10.0, 10.0, 10.0])
vol = np.array([1.0, 1.0, 1.0, 1.0])
out = tb.batch(col, col, col, col, vol)
out = _to_np(tb.batch(col, col, col, col, vol))
assert out.shape == (2, 5)
@@ -4315,5 +4328,5 @@ def test_three_line_break_bars_reference():
def test_three_line_break_bars_batch_shape():
tlb = ta.ThreeLineBreakBars(3)
out = tlb.batch(np.array([10.0, 11.0, 12.0, 13.0]))
out = _to_np(tlb.batch(np.array([10.0, 11.0, 12.0, 13.0])))
assert out.shape[1] == 3
bindings/python/tests/test_seasonality.py
+11 -2
View File
@@ -14,6 +14,15 @@ import wickra as ta
HOUR_MS = 3_600_000
def _to_np(x) -> np.ndarray:
"""Normalize a Wickra batch result (``array.array`` or ``Matrix``) to an ndarray."""
if isinstance(x, np.ndarray):
return x.astype(np.float64)
if hasattr(x, "tolist") and hasattr(x, "shape"):
return np.asarray(x.tolist(), dtype=np.float64)
return np.asarray(x, dtype=np.float64)
@pytest.fixture(scope="module")
def candle_columns():
"""240 hourly candles (10 days) with valid OHLCV and epoch-ms timestamps."""
@@ -43,7 +52,7 @@ def _check_scalar(make, cols):
[np.nan if (v := a.update(c)) is None else v for c in candles],
dtype=np.float64,
)
batch = np.asarray(b.batch(*cols))
batch = _to_np(b.batch(*cols))
np.testing.assert_allclose(stream, batch, equal_nan=True, rtol=1e-9, atol=1e-9)
@@ -55,7 +64,7 @@ def _check_matrix(make, k, cols):
out = a.update(c)
rows.append(np.full(k, np.nan) if out is None else np.asarray(out, dtype=float))
stream = np.vstack(rows)
batch = np.asarray(b.batch(*cols))
batch = _to_np(b.batch(*cols))
assert batch.shape == (len(candles), k)
np.testing.assert_allclose(stream, batch, equal_nan=True, rtol=1e-9, atol=1e-9)
bindings/python/tests/test_smoke.py
+12 -12
View File
@@ -24,8 +24,8 @@ def test_version_is_a_nonempty_string():
)
def test_scalar_batch_returns_same_length(cls, args, sine_prices):
out = cls(*args).batch(sine_prices)
assert out.shape == sine_prices.shape
assert out.dtype == np.float64
assert len(out) == len(sine_prices)
assert out.typecode == "d"
def test_macd_batch_returns_n_by_3(sine_prices):
@@ -41,7 +41,7 @@ def test_bollinger_batch_returns_n_by_4(sine_prices):
def test_atr_batch_shape(ohlc_series):
high, low, close = ohlc_series
out = ta.ATR(14).batch(high, low, close)
assert out.shape == close.shape
assert len(out) == len(close)
def test_stochastic_batch_shape(ohlc_series):
@@ -54,7 +54,7 @@ def test_obv_batch_shape(ohlc_series):
_, _, close = ohlc_series
volume = np.ones_like(close)
out = ta.OBV().batch(close, volume)
assert out.shape == close.shape
assert len(out) == len(close)
def test_value_area_batch_shape(ohlc_series):
@@ -91,7 +91,7 @@ def test_heikin_ashi_batch_returns_n_by_4(ohlc_series):
def test_ehlers_super_smoother_batch_shape(sine_prices):
out = ta.SuperSmoother(10).batch(sine_prices)
assert out.shape == sine_prices.shape
assert len(out) == len(sine_prices)
def test_mama_batch_shape(sine_prices):
@@ -118,8 +118,8 @@ def test_orderbook_indicators_construct_and_emit():
def test_orderbook_batch_returns_one_value_per_snapshot():
snapshots = [([100.0], [3.0], [101.0], [1.0])] * 5
out = ta.OrderBookImbalanceTop1().batch(snapshots)
assert out.shape == (5,)
assert out.dtype == np.float64
assert len(out) == 5
assert out.typecode == "d"
def test_tradeflow_indicators_construct_and_emit():
@@ -134,8 +134,8 @@ def test_tradeflow_batch_returns_one_value_per_trade():
size = np.array([1.0, 2.0, 3.0, 1.0, 2.0, 3.0])
is_buy = [True, False, True, False, True, False]
out = ta.CumulativeVolumeDelta().batch(price, size, is_buy)
assert out.shape == (6,)
assert out.dtype == np.float64
assert len(out) == 6
assert out.typecode == "d"
def test_price_impact_indicators_construct_and_emit():
@@ -152,14 +152,14 @@ def test_price_impact_batch_returns_one_value_per_trade():
mid = np.full(4, 100.0)
for ind in (ta.EffectiveSpread(), ta.RealizedSpread(2), ta.KylesLambda(2)):
out = ind.batch(price, size, is_buy, mid)
assert out.shape == (4,)
assert out.dtype == np.float64
assert len(out) == 4
assert out.typecode == "d"
def test_footprint_constructs_and_emits():
out = ta.Footprint(1.0).update(100.2, 2.0, True)
assert out.shape == (1, 3)
assert out.dtype == np.float64
assert isinstance(out[0, 0], float)
def test_footprint_batch_returns_list_of_arrays():
bindings/python/tests/test_streaming_vs_batch.py
+16 -1
View File
@@ -15,8 +15,23 @@ import pytest
import wickra as ta
def _equal_with_nan(a: np.ndarray, b: np.ndarray, tol: float = 1e-9) -> bool:
def _to_np(x) -> np.ndarray:
"""Normalize a Wickra batch result to a float64 ndarray.
Scalar batches return a stdlib ``array.array``; multi-output batches return a
``Matrix`` exposing ``.shape``/``.tolist()``.
"""
if isinstance(x, np.ndarray):
return x.astype(np.float64)
if hasattr(x, "tolist") and hasattr(x, "shape"):
return np.asarray(x.tolist(), dtype=np.float64)
return np.asarray(x, dtype=np.float64)
def _equal_with_nan(a, b, tol: float = 1e-9) -> bool:
"""NumPy ``==`` treats NaN as not-equal; emulate ``equal_nan`` for floats."""
a = _to_np(a)
b = _to_np(b)
if a.shape != b.shape:
return False
both_nan = np.isnan(a) & np.isnan(b)