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

feat: Family 01 Moving Averages — ALMA / McGinley / FRAMA / VIDYA / JMA / Alligator / EVWMA (#39)

Browse Source 
* feat(alma): add Arnaud Legoux Moving Average

Gaussian-weighted moving average with configurable centre (offset in
[0, 1]) and kernel width (sigma > 0). Pre-computes normalised weights
at construction so each update is a single rolling window dot product.

Reference: Arnaud Legoux and Dimitrios Kouzis-Loukas, 2009.

Touchpoints:
- crates/wickra-core: alma.rs + mod.rs + lib.rs re-export
- bindings/python: PyAlma + __init__.py + test_new_indicators +
  test_known_values reference
- bindings/node: AlmaNode + index.d.ts/index.js + indicators.test.js
  factory + reference value
- bindings/wasm: wasm_scalar_indicator! macro
- fuzz: indicator_update target covers ALMA(9, 0.85, 6.0)
- crates/wickra/benches: bench_scalar entry
- README + CHANGELOG: Moving Averages row + Unreleased entry

* feat(mcginley): add McGinley Dynamic moving average

John McGinley's self-adjusting moving average with the recurrence
MD + (price - MD) / (0.6 * period * (price / MD)^4). Speeds up when
price falls below the indicator and damps when price runs above the
indicator. Seeded with the simple average of the first period inputs.

Reference: McGinley, Technical Analysis of Stocks & Commodities, 1990.

Touchpoints:
- crates/wickra-core: mcginley_dynamic.rs + mod.rs + lib.rs re-export
- bindings/python: PyMcGinleyDynamic + __init__.py + test_new_indicators
  + test_known_values reference
- bindings/node: McGinleyDynamicNode (scalar macro) + index.d.ts/index.js
  + indicators.test.js factory + reference value
- bindings/wasm: wasm_scalar_indicator! macro
- fuzz: indicator_update target covers McGinleyDynamic(10)
- crates/wickra/benches: bench_scalar entry
- README + CHANGELOG: Moving Averages row + Unreleased entry

* feat(frama): add Fractal Adaptive Moving Average

Ehlers' FRAMA adapts its smoothing constant to the fractal dimension of
the recent window: tight tracking in trends, heavy smoothing in chop.
Uses the close-only variant where max/min over each window half drive
the dimension estimate. Period must be even (default 16).

Reference: Ehlers, Fractal Adaptive Moving Average, 2005.

Touchpoints:
- crates/wickra-core: frama.rs + mod.rs + lib.rs re-export
- bindings/python: PyFrama + __init__.py + test_new_indicators +
  test_known_values reference (constant series + uptrend tracking)
- bindings/node: FramaNode (scalar macro) + index.d.ts/index.js +
  indicators.test.js factory + reference value
- bindings/wasm: wasm_scalar_indicator! macro
- fuzz: indicator_update target covers Frama(16)
- crates/wickra/benches: bench_scalar entry
- README + CHANGELOG: Moving Averages row + Unreleased entry

* feat(vidya): add Variable Index Dynamic Average

Chande's VIDYA — an EMA whose alpha scales with |CMO(cmo_period)| / 100.
Strong directional momentum lifts the smoothing constant toward the
EMA-of-period rate; flat or choppy windows shrink it toward zero so
VIDYA coasts on its previous value. Two parameters: period (14) and
cmo_period (9). Reuses the existing wickra-core Cmo internally.

Reference: Chande, Stocks & Commodities, 1992.

Also fixes a silent gap from d37fbd1 (feat(frama)): the PyFrama Python
class wrapper and its add_class registration were dropped because the
two edits hit "File has not been read yet" errors that scrolled past
in a batch. Adds them here alongside VIDYA's bindings.

Touchpoints (VIDYA): vidya.rs + mod.rs + lib.rs re-export, PyVidya +
__init__.py + test_new_indicators + test_known_values reference,
VidyaNode (manual two-param binding) + index.d.ts/index.js +
indicators.test.js factory + reference, wasm_scalar_indicator! macro,
fuzz target, bench, README + CHANGELOG.

* feat(jma): add Jurik Moving Average

Three-stage filter reconstruction of Mark Jurik's adaptive MA (the
algorithm is proprietary; this is the form used by most open-source
ports since the 1999 TASC article). Parameters: period (14), phase in
[-100, 100] (0), power in 1..=4 (2). State is seeded by setting
e0 = JMA = first input so a constant input stream is reproduced exactly.

Touchpoints: jma.rs + mod.rs + lib.rs re-export, PyJma + __init__.py +
test_new_indicators + test_known_values reference, JmaNode (manual
three-param binding) + index.d.ts/index.js + indicators.test.js factory
+ reference, wasm_scalar_indicator! macro, fuzz target, bench, README +
CHANGELOG.

* feat(alligator): add Bill Williams Alligator

Three SMMA lines (Jaw / Teeth / Lips) over the median price
(high + low) / 2 with default periods 13 / 8 / 5. Multi-output
indicator returning AlligatorOutput { jaw, teeth, lips }. The
original chart variant shifts each line forward for display; we
publish the unshifted SMMA values and leave the visual shift to
the consumer.

Reference: Bill Williams, Trading Chaos, 1995.

Touchpoints: alligator.rs + mod.rs + lib.rs re-export, PyAlligator
(Candle input, returns 3-tuple, ndarray (n, 3) batch) + __init__.py
+ test_new_indicators + test_known_values reference, AlligatorNode +
AlligatorValue + index.d.ts/index.js + indicators.test.js multi
factory + reference, WasmAlligator (manual JsValue object) +
candle-fuzz target + README + CHANGELOG.

* feat(evwma): add Elastic Volume-Weighted Moving Average

Christian P. Fries' elastic recurrence where the smoothing weight is the
bar's volume relative to the running window total:

  V_sum_t = sum of volumes over the last period candles
  EVWMA_t = ((V_sum_t - v_t) * EVWMA_{t-1} + v_t * close_t) / V_sum_t

A bar whose volume is small barely moves the average; a bar that
dominates the window pulls it strongly toward that bar's close. Seeded
with the close of the first full window; holds its previous value if
the entire window has zero volume.

Reference: Fries, Wilmott Magazine, 2001.

Touchpoints: evwma.rs + mod.rs + lib.rs re-export, PyEvwma (close +
volume batch) + __init__.py + test_new_indicators CANDLE_SCALAR +
test_known_values reference, EvwmaNode + index.d.ts/index.js +
indicators.test.js candleScalar factory + reference, WasmEvwma,
candle-fuzz target + README + CHANGELOG.

* ci: Force local wheel install in Python jobs

Use --no-index --no-deps so the Python matrix installs the freshly
built wheel from dist/ and never falls back to PyPI. Previously pip
sometimes picked the released 0.2.x wheel on macOS / Windows when its
platform tag was a wider match than the local build, which made the
job test the released package and miss any new symbols added in the
PR (e.g. AttributeError: module 'wickra' has no attribute 'ALMA').
numpy is already installed by the preceding pip step, so --no-deps
is safe.
This commit is contained in:
kingchenc
2026-05-25 15:01:14 +02:00
committed by GitHub
parent 178fbfd68e
commit 466faddd87
23 changed files with 2765 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bindings/python/python/wickra/__init__.py
+14
View File
@@ -38,6 +38,13 @@ from ._wickra import (
ZLEMA,
T3,
VWMA,
ALMA,
McGinleyDynamic,
FRAMA,
VIDYA,
JMA,
Alligator,
EVWMA,
# Momentum
RSI,
MACD,
@@ -119,6 +126,13 @@ __all__ = [
"ZLEMA",
"T3",
"VWMA",
"ALMA",
"McGinleyDynamic",
"FRAMA",
"VIDYA",
"JMA",
"Alligator",
"EVWMA",
# Momentum
"RSI",
"MACD",
bindings/python/src/lib.rs
+386
View File
@@ -812,6 +812,385 @@ impl PyKama {
}
}
// ============================== FRAMA ==============================
#[pyclass(name = "FRAMA", module = "wickra._wickra", skip_from_py_object)]
#[derive(Clone)]
struct PyFrama {
inner: wc::Frama,
}
#[pymethods]
impl PyFrama {
#[new]
#[pyo3(signature = (period=16))]
fn new(period: usize) -> PyResult<Self> {
Ok(Self {
inner: wc::Frama::new(period).map_err(map_err)?,
})
}
fn update(&mut self, value: f64) -> Option<f64> {
self.inner.update(value)
}
fn batch<'py>(
&mut self,
py: Python<'py>,
prices: PyReadonlyArray1<'py, f64>,
) -> PyResult<Bound<'py, PyArray1<f64>>> {
let s = prices
.as_slice()
.map_err(|_| PyValueError::new_err(NON_CONTIGUOUS))?;
Ok(flatten(self.inner.batch(s)).into_pyarray(py))
}
#[getter]
fn period(&self) -> usize {
self.inner.period()
}
fn reset(&mut self) {
self.inner.reset();
}
fn is_ready(&self) -> bool {
self.inner.is_ready()
}
fn warmup_period(&self) -> usize {
self.inner.warmup_period()
}
fn __repr__(&self) -> String {
format!("FRAMA(period={})", self.inner.period())
}
}
// ============================== EVWMA ==============================
#[pyclass(name = "EVWMA", module = "wickra._wickra", skip_from_py_object)]
#[derive(Clone)]
struct PyEvwma {
inner: wc::Evwma,
}
#[pymethods]
impl PyEvwma {
#[new]
#[pyo3(signature = (period=20))]
fn new(period: usize) -> PyResult<Self> {
Ok(Self {
inner: wc::Evwma::new(period).map_err(map_err)?,
})
}
fn update(&mut self, candle: &Bound<'_, PyAny>) -> PyResult<Option<f64>> {
let c = extract_candle(candle)?;
Ok(self.inner.update(c))
}
fn batch<'py>(
&mut self,
py: Python<'py>,
close: PyReadonlyArray1<'py, f64>,
volume: PyReadonlyArray1<'py, f64>,
) -> PyResult<Bound<'py, PyArray1<f64>>> {
let c = close
.as_slice()
.map_err(|_| PyValueError::new_err(NON_CONTIGUOUS))?;
let v = volume
.as_slice()
.map_err(|_| PyValueError::new_err(NON_CONTIGUOUS))?;
if c.len() != v.len() {
return Err(PyValueError::new_err(
"close and volume must be equal length",
));
}
let mut out = Vec::with_capacity(c.len());
for i in 0..c.len() {
let candle = wc::Candle::new(c[i], c[i], c[i], c[i], v[i], 0).map_err(map_err)?;
out.push(self.inner.update(candle).unwrap_or(f64::NAN));
}
Ok(out.into_pyarray(py))
}
#[getter]
fn period(&self) -> usize {
self.inner.period()
}
fn reset(&mut self) {
self.inner.reset();
}
fn is_ready(&self) -> bool {
self.inner.is_ready()
}
fn warmup_period(&self) -> usize {
self.inner.warmup_period()
}
fn __repr__(&self) -> String {
format!("EVWMA(period={})", self.inner.period())
}
}
// ============================== Alligator ==============================
#[pyclass(name = "Alligator", module = "wickra._wickra", skip_from_py_object)]
#[derive(Clone)]
struct PyAlligator {
inner: wc::Alligator,
}
#[pymethods]
impl PyAlligator {
#[new]
#[pyo3(signature = (jaw=13, teeth=8, lips=5))]
fn new(jaw: usize, teeth: usize, lips: usize) -> PyResult<Self> {
Ok(Self {
inner: wc::Alligator::new(jaw, teeth, lips).map_err(map_err)?,
})
}
fn update(&mut self, candle: &Bound<'_, PyAny>) -> PyResult<Option<(f64, f64, f64)>> {
let c = extract_candle(candle)?;
Ok(self.inner.update(c).map(|o| (o.jaw, o.teeth, o.lips)))
}
fn batch<'py>(
&mut self,
py: Python<'py>,
high: PyReadonlyArray1<'py, f64>,
low: PyReadonlyArray1<'py, f64>,
) -> PyResult<Bound<'py, PyArray2<f64>>> {
let h = high
.as_slice()
.map_err(|_| PyValueError::new_err(NON_CONTIGUOUS))?;
let l = low
.as_slice()
.map_err(|_| PyValueError::new_err(NON_CONTIGUOUS))?;
if h.len() != l.len() {
return Err(PyValueError::new_err("high and low must be equal length"));
}
let n = h.len();
let mut out = vec![f64::NAN; n * 3];
for i in 0..n {
let candle = wc::Candle::new(l[i], h[i], l[i], l[i], 0.0, 0).map_err(map_err)?;
if let Some(o) = self.inner.update(candle) {
out[i * 3] = o.jaw;
out[i * 3 + 1] = o.teeth;
out[i * 3 + 2] = o.lips;
}
}
Ok(numpy::ndarray::Array2::from_shape_vec((n, 3), out)
.expect("shape consistent")
.into_pyarray(py))
}
fn reset(&mut self) {
self.inner.reset();
}
fn is_ready(&self) -> bool {
self.inner.is_ready()
}
fn warmup_period(&self) -> usize {
self.inner.warmup_period()
}
fn __repr__(&self) -> String {
let (j, t, l) = self.inner.periods();
format!("Alligator(jaw={j}, teeth={t}, lips={l})")
}
}
// ============================== JMA ==============================
#[pyclass(name = "JMA", module = "wickra._wickra", skip_from_py_object)]
#[derive(Clone)]
struct PyJma {
inner: wc::Jma,
}
#[pymethods]
impl PyJma {
#[new]
#[pyo3(signature = (period=14, phase=0.0, power=2))]
fn new(period: usize, phase: f64, power: u32) -> PyResult<Self> {
Ok(Self {
inner: wc::Jma::new(period, phase, power).map_err(map_err)?,
})
}
fn update(&mut self, value: f64) -> Option<f64> {
self.inner.update(value)
}
fn batch<'py>(
&mut self,
py: Python<'py>,
prices: PyReadonlyArray1<'py, f64>,
) -> PyResult<Bound<'py, PyArray1<f64>>> {
let s = prices
.as_slice()
.map_err(|_| PyValueError::new_err(NON_CONTIGUOUS))?;
Ok(flatten(self.inner.batch(s)).into_pyarray(py))
}
fn reset(&mut self) {
self.inner.reset();
}
fn is_ready(&self) -> bool {
self.inner.is_ready()
}
fn warmup_period(&self) -> usize {
self.inner.warmup_period()
}
fn __repr__(&self) -> String {
let (p, ph, pw) = self.inner.params();
format!("JMA(period={p}, phase={ph}, power={pw})")
}
}
// ============================== VIDYA ==============================
#[pyclass(name = "VIDYA", module = "wickra._wickra", skip_from_py_object)]
#[derive(Clone)]
struct PyVidya {
inner: wc::Vidya,
}
#[pymethods]
impl PyVidya {
#[new]
#[pyo3(signature = (period=14, cmo_period=9))]
fn new(period: usize, cmo_period: usize) -> PyResult<Self> {
Ok(Self {
inner: wc::Vidya::new(period, cmo_period).map_err(map_err)?,
})
}
fn update(&mut self, value: f64) -> Option<f64> {
self.inner.update(value)
}
fn batch<'py>(
&mut self,
py: Python<'py>,
prices: PyReadonlyArray1<'py, f64>,
) -> PyResult<Bound<'py, PyArray1<f64>>> {
let s = prices
.as_slice()
.map_err(|_| PyValueError::new_err(NON_CONTIGUOUS))?;
Ok(flatten(self.inner.batch(s)).into_pyarray(py))
}
fn reset(&mut self) {
self.inner.reset();
}
fn is_ready(&self) -> bool {
self.inner.is_ready()
}
fn warmup_period(&self) -> usize {
self.inner.warmup_period()
}
fn __repr__(&self) -> String {
let (p, c) = self.inner.periods();
format!("VIDYA(period={p}, cmo_period={c})")
}
}
// ============================== McGinley Dynamic ==============================
#[pyclass(
name = "McGinleyDynamic",
module = "wickra._wickra",
skip_from_py_object
)]
#[derive(Clone)]
struct PyMcGinleyDynamic {
inner: wc::McGinleyDynamic,
}
#[pymethods]
impl PyMcGinleyDynamic {
#[new]
#[pyo3(signature = (period=10))]
fn new(period: usize) -> PyResult<Self> {
Ok(Self {
inner: wc::McGinleyDynamic::new(period).map_err(map_err)?,
})
}
fn update(&mut self, value: f64) -> Option<f64> {
self.inner.update(value)
}
fn batch<'py>(
&mut self,
py: Python<'py>,
prices: PyReadonlyArray1<'py, f64>,
) -> PyResult<Bound<'py, PyArray1<f64>>> {
let s = prices
.as_slice()
.map_err(|_| PyValueError::new_err(NON_CONTIGUOUS))?;
Ok(flatten(self.inner.batch(s)).into_pyarray(py))
}
#[getter]
fn period(&self) -> usize {
self.inner.period()
}
fn reset(&mut self) {
self.inner.reset();
}
fn is_ready(&self) -> bool {
self.inner.is_ready()
}
fn warmup_period(&self) -> usize {
self.inner.warmup_period()
}
fn __repr__(&self) -> String {
format!("McGinleyDynamic(period={})", self.inner.period())
}
}
// ============================== ALMA ==============================
#[pyclass(name = "ALMA", module = "wickra._wickra", skip_from_py_object)]
#[derive(Clone)]
struct PyAlma {
inner: wc::Alma,
}
#[pymethods]
impl PyAlma {
#[new]
#[pyo3(signature = (period=9, offset=0.85, sigma=6.0))]
fn new(period: usize, offset: f64, sigma: f64) -> PyResult<Self> {
Ok(Self {
inner: wc::Alma::new(period, offset, sigma).map_err(map_err)?,
})
}
fn update(&mut self, value: f64) -> Option<f64> {
self.inner.update(value)
}
fn batch<'py>(
&mut self,
py: Python<'py>,
prices: PyReadonlyArray1<'py, f64>,
) -> PyResult<Bound<'py, PyArray1<f64>>> {
let s = prices
.as_slice()
.map_err(|_| PyValueError::new_err(NON_CONTIGUOUS))?;
Ok(flatten(self.inner.batch(s)).into_pyarray(py))
}
#[getter]
fn period(&self) -> usize {
self.inner.period()
}
#[getter]
fn offset(&self) -> f64 {
self.inner.offset()
}
#[getter]
fn sigma(&self) -> f64 {
self.inner.sigma()
}
fn reset(&mut self) {
self.inner.reset();
}
fn is_ready(&self) -> bool {
self.inner.is_ready()
}
fn warmup_period(&self) -> usize {
self.inner.warmup_period()
}
fn __repr__(&self) -> String {
format!(
"ALMA(period={}, offset={}, sigma={})",
self.inner.period(),
self.inner.offset(),
self.inner.sigma()
)
}
}
// ============================== CCI ==============================
#[pyclass(name = "CCI", module = "wickra._wickra", skip_from_py_object)]
@@ -4494,6 +4873,13 @@ fn _wickra(_py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()> {
m.add_class::<PyTema>()?;
m.add_class::<PyHma>()?;
m.add_class::<PyKama>()?;
m.add_class::<PyAlma>()?;
m.add_class::<PyMcGinleyDynamic>()?;
m.add_class::<PyFrama>()?;
m.add_class::<PyVidya>()?;
m.add_class::<PyJma>()?;
m.add_class::<PyAlligator>()?;
m.add_class::<PyEvwma>()?;
m.add_class::<PyCci>()?;
m.add_class::<PyRoc>()?;
m.add_class::<PyWilliamsR>()?;
bindings/python/tests/test_known_values.py
+91
View File
@@ -66,6 +66,97 @@ def test_rsi_wilder_textbook_first_value():
assert math.isclose(out[14], 70.464, abs_tol=0.05)
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))
assert np.all(np.isnan(out[:8]))
np.testing.assert_allclose(out[8:], 42.0, atol=1e-12)
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]))
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)
expected = float(np.dot([10.0, 20.0, 30.0], w) / w.sum())
assert math.isclose(out[2], expected, abs_tol=1e-12)
# Sanity: heavy offset toward the newest sample lifts the average above
# the simple mean of 20.
assert out[2] > 20.0
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))
assert np.all(np.isnan(out[:4]))
np.testing.assert_allclose(out[4:], 42.0, atol=1e-12)
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]))
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)
assert math.isclose(out[3], expected, abs_tol=1e-12)
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))
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))
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))
np.testing.assert_allclose(out, 42.0, atol=1e-12)
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]))
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)
def test_alligator_constant_series_holds_at_median_price():
# Median price = (11 + 9) / 2 = 10 on every candle, so all three SMMAs
# seed at 10 and stay there.
n = 30
high = np.full(n, 11.0)
low = np.full(n, 9.0)
out = 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)
assert math.isclose(row[1], 10.0, abs_tol=1e-12)
assert math.isclose(row[2], 10.0, abs_tol=1e-12)
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))
assert np.all(np.isnan(out[:4]))
np.testing.assert_allclose(out[4:], 42.0, atol=1e-12)
def test_macd_constant_series_converges_to_zero():
out = ta.MACD().batch(np.full(200, 100.0))
# Last row's MACD and signal must be ~0.
bindings/python/tests/test_new_indicators.py
+25
View File
@@ -44,6 +44,11 @@ SCALAR = [
(ta.SMMA, (14,)),
(ta.TRIMA, (20,)),
(ta.ZLEMA, (14,)),
(ta.ALMA, (9, 0.85, 6.0)),
(ta.McGinleyDynamic, (10,)),
(ta.FRAMA, (16,)),
(ta.VIDYA, (14, 9)),
(ta.JMA, (14, 0.0, 2)),
(ta.T3, (5, 0.7)),
(ta.MOM, (10,)),
(ta.CMO, (14,)),
@@ -87,6 +92,7 @@ def test_scalar_streaming_matches_batch(cls, args, sine_prices):
CANDLE_SCALAR = {
"VWMA": (lambda: ta.VWMA(20), lambda ind, h, l, c, v: ind.batch(c, v)),
"EVWMA": (lambda: ta.EVWMA(20), lambda ind, h, l, c, v: ind.batch(c, v)),
"UltimateOscillator": (
lambda: ta.UltimateOscillator(7, 14, 28),
lambda ind, h, l, c, v: ind.batch(h, l, c),
@@ -226,6 +232,24 @@ def test_multi_streaming_matches_batch(name, ohlcv):
assert _eq_nan(batch, np.array(rows, dtype=np.float64)), f"{name} mismatch"
# --- Alligator (3-tuple output) -------------------------------------------
def test_alligator_streaming_matches_batch(ohlcv):
high, low, _, _ = ohlcv
alligator = ta.Alligator(13, 8, 5)
batch = alligator.batch(high, low)
assert batch.shape == (high.size, 3)
streamer = ta.Alligator(13, 8, 5)
rows = []
for i in range(high.size):
candle = (float(low[i]), float(high[i]), float(low[i]), float(low[i]), 0.0, i)
v = streamer.update(candle)
rows.append([math.nan, math.nan, math.nan] if v is None else list(v))
assert _eq_nan(batch, np.array(rows, dtype=np.float64)), "Alligator mismatch"
# --- Reference values -----------------------------------------------------
@@ -296,6 +320,7 @@ def test_new_indicators_expose_lifecycle():
instances = [make() for make, _ in CANDLE_SCALAR.values()]
instances += [make() for make, _ in MULTI.values()]
instances += [cls(*args) for cls, args in SCALAR]
instances.append(ta.Alligator(13, 8, 5))
for ind in instances:
assert ind.is_ready() is False
assert ind.warmup_period() >= 1