kingchenc
3ea0f12b7a
* feat(rvi): add Relative Volatility Index
Donald Dorsey's RSI-shaped volatility gauge. Partitions the rolling
population standard deviation of close into "up" samples (close rose
since the previous bar) and "down" samples (close fell), Wilder-smooths
each side, and reports 100 * AvgUp / (AvgUp + AvgDown). Output bounded
on [0, 100]; saturates at 100 in pure uptrends, 0 in pure downtrends,
and falls back to 50 on a completely flat series (same undefined-RS
convention as RSI).
Single period parameter (default 10) drives both the stddev window and
the Wilder smoothing constant. First emit lands at index 2*period - 2
(2*period - 1 bars are needed: period to fill the stddev window plus
period - 1 to seed the Wilder averages, overlapping by one bar).
Touchpoints: rvi.rs + mod.rs + lib.rs re-export, PyRvi + __init__.py +
test_new_indicators SCALAR + test_known_values uptrend reference,
RviNode + index.d.ts/index.js + indicators.test.js factory +
reference, WasmRvi via scalar macro, scalar-fuzz target, bench_scalar
entry, README + CHANGELOG.
* feat(parkinson): add Parkinson Volatility
Michael Parkinson's (1980) high-low realised volatility estimator.
Under a driftless Geometric-Brownian-Motion assumption, the extreme
range of a bar carries roughly 5x the variance information of the
close-to-close estimator, so for a given statistical efficiency
Parkinson needs five times fewer samples.
Formula:
sigma^2 = (1 / (4n * ln 2)) * Sum_{i=1..n} (ln(H_i / L_i))^2
out = sqrt(sigma^2) * sqrt(trading_periods) * 100
The output is annualised to a percent in the same style as
HistoricalVolatility (pass `trading_periods = 1` for the raw per-bar
sigma * 100 figure). Two parameters: `period` (default 20) for the
rolling window, `trading_periods` (default 252) for the annualisation
factor. First emit at index `period - 1`.
Touchpoints: parkinson.rs + mod.rs + lib.rs re-export,
PyParkinsonVolatility + __init__.py + test_new_indicators CANDLE_SCALAR
+ test_known_values zero-range reference, ParkinsonVolatilityNode +
index.d.ts/index.js + indicators.test.js factory + reference,
WasmParkinsonVolatility hand-rolled, candle-fuzz target,
bench_candle_input entry, README + CHANGELOG.
* feat(garman-klass): add Garman-Klass Volatility
Garman & Klass (1980) OHLC realised-volatility estimator. Extends
Parkinson's high-low estimator with an open-to-close term, lifting
statistical efficiency from ~5x to ~7.4x relative to close-to-close
stddev under driftless Geometric Brownian Motion.
Formula (per bar):
s_t = 0.5 * (ln(H_t / L_t))^2 - (2*ln(2) - 1) * (ln(C_t / O_t))^2
out = sqrt(max(mean(s_t over `period`), 0)) * sqrt(trading_periods) * 100
The per-bar sample can be marginally negative when the bar has a small
range relative to its open-to-close move; a max(., 0) clamp on the
rolling mean absorbs that and the FP cancellation noise before the
square root.
Still biased on data with meaningful overnight drift -- use Yang-Zhang
when gaps matter. Defaults: `period = 20`, `trading_periods = 252`
(annualised percent, same convention as HistoricalVolatility).
Touchpoints: garman_klass.rs + mod.rs + lib.rs re-export,
PyGarmanKlassVolatility + __init__.py + test_new_indicators
CANDLE_SCALAR + test_known_values zero-movement reference,
GarmanKlassVolatilityNode + index.d.ts/index.js + indicators.test.js
factory + reference, WasmGarmanKlassVolatility hand-rolled,
candle-fuzz target, bench_candle_input entry, README + CHANGELOG.
* feat(rogers-satchell): add Rogers-Satchell Volatility
Rogers, Satchell & Yoon (1994) OHLC realised-volatility estimator.
Unlike Garman-Klass, the per-bar sample is exact under arbitrary
Brownian drift -- the drift component cancels algebraically.
Formula (per bar):
s_t = ln(H_t / C_t) * ln(H_t / O_t) + ln(L_t / C_t) * ln(L_t / O_t)
out = sqrt(max(mean(s_t over `period`), 0)) * sqrt(trading_periods) * 100
Each per-bar sample is also non-negative by construction: with
`Candle::new` guaranteeing H >= max(O, L, C) and L <= min(O, H, C), the
four log factors have predictable signs (ln(H/.) >= 0, ln(L/.) <= 0),
so both products contribute >= 0. The max(., 0) clamp on the rolling
mean is only there to absorb FP cancellation.
Defaults: `period = 20`, `trading_periods = 252` (annualised percent,
same convention as HistoricalVolatility / Parkinson / Garman-Klass).
Touchpoints: rogers_satchell.rs + mod.rs + lib.rs re-export,
PyRogersSatchellVolatility + __init__.py + test_new_indicators
CANDLE_SCALAR + test_known_values zero-movement reference,
RogersSatchellVolatilityNode + index.d.ts/index.js + indicators.test.js
factory + reference, WasmRogersSatchellVolatility hand-rolled,
candle-fuzz target, bench_candle_input entry, README + CHANGELOG.
* feat(yang-zhang): add Yang-Zhang Volatility
Yang & Zhang (2000) drift- and gap-robust OHLC realised-volatility
estimator. Combines three independent components into a single estimate
with minimum variance:
overnight = sample_var(ln(O_t / C_{t-1})) over n bars (close-to-open)
open_close = sample_var(ln(C_t / O_t)) over n bars
rs = mean(ln(H/C)*ln(H/O) + ln(L/C)*ln(L/O)) over n bars
sigma^2_YZ = overnight + k*open_close + (1-k)*rs
k = 0.34 / (1.34 + (n+1)/(n-1))
out = sqrt(max(sigma^2_YZ, 0)) * sqrt(trading_periods) * 100
The overnight and open-to-close variances use Bessel's correction (the
sample estimator, divisor n-1), same convention as
HistoricalVolatility. The blending factor `k` is the one that
minimises estimator variance under driftless Geometric Brownian Motion
with overnight gaps.
This is the gold-standard OHLC estimator for assets with both
close-to-open gaps and intraday drift: equities, futures, and any
market that does not trade continuously. For pure intraday data (where
O_t == C_{t-1} and the open-to-close return is constant), the
overnight and open-close terms vanish and the estimator collapses to
(1-k) * Rogers-Satchell -- this is the indicator's
intraday_data_collapses_to_rs_only unit test.
Period >= 2 (Bessel correction needs >= 2 samples). First emit at
index `period` (the (period+1)-th bar): one bar seeds prev_close, the
next `period` fill the rolling windows. Defaults: `period = 20`,
`trading_periods = 252`.
Touchpoints: yang_zhang.rs + mod.rs + lib.rs re-export,
PyYangZhangVolatility + __init__.py + test_new_indicators
CANDLE_SCALAR + test_known_values zero-movement reference,
YangZhangVolatilityNode + index.d.ts/index.js + indicators.test.js
factory + reference, WasmYangZhangVolatility hand-rolled, candle-fuzz
target, bench_candle_input entry, README + CHANGELOG.
* fix(rvi): rename to RviVolatility to avoid clash with family-02 RVI
Family 02 (PR #40) ships a separate `Rvi` struct for Relative Vigor
Index. The two indicators have nothing to do with each other beyond
sharing the acronym, so disambiguate by giving the volatility one a
longer name everywhere:
- Rust crate: `Rvi` -> `RviVolatility`
- Rust file: `rvi.rs` -> `rvi_volatility.rs`
- Python: `RVI` -> `RVIVolatility`
- Node: `RVI` -> `RVIVolatility`
- WASM: `RVI` -> `RVIVolatility`
Once the two PRs are both merged, callers get `wickra::Rvi` for Vigor
and `wickra::RviVolatility` for Volatility. The shorter `RVI` acronym
stays with the Momentum family per the existing wiki pages and the
implementation that shipped first.
Updates: rvi_volatility.rs (renamed), mod.rs, lib.rs re-export,
bindings/python/src/lib.rs + __init__.py + tests, bindings/node/src/lib.rs
+ index.d.ts + index.js + __tests__, bindings/wasm/src/lib.rs,
fuzz/fuzz_targets/indicator_update.rs, crates/wickra/benches/indicators.rs,
README family-table label, CHANGELOG entry.
* test(volatility): Rename test_rvi -> test_rvi_volatility + drop dead match arms
The Python test test_rvi_pure_uptrend_saturates_at_one_hundred was
calling ta.RVI() expecting the volatility version, but ta.RVI now
means Family 02's Relative Vigor Index (candle input). Renamed to
ta.RVIVolatility to match the binding rename done at merge time.
In all four OHLC volatility tests, the existing `match (r, a) { ...,
_ => panic!() }` arm is dead in passing runs (every aligned pair is
either (None, None) or (Some, Some)). Codecov flagged it as a patch
miss on each of parkinson / garman_klass / rogers_satchell /
yang_zhang. Refactored per CLAUDE.md cold-path guidance to
`assert_eq!(r.is_some(), a.is_some()); if let (Some, Some) ...`.
390 lines
14 KiB
Python
390 lines
14 KiB
Python
"""Reference-value tests that pin numerical behaviour from the Python side."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import math
|
||
|
||
import numpy as np
|
||
import pytest
|
||
|
||
import wickra as ta
|
||
|
||
|
||
def test_sma_constant_series():
|
||
out = 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)
|
||
|
||
|
||
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]))
|
||
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]))
|
||
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]))
|
||
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))
|
||
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))
|
||
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))
|
||
np.testing.assert_allclose(out[14:], 50.0)
|
||
|
||
|
||
def test_rsi_wilder_textbook_first_value():
|
||
"""Wilder's original 14-period example, ~70.46 at the first emit."""
|
||
prices = np.array(
|
||
[
|
||
44.34, 44.09, 44.15, 43.61, 44.33, 44.83, 45.10, 45.42, 45.84, 46.08,
|
||
45.89, 46.03, 45.61, 46.28, 46.28,
|
||
],
|
||
dtype=np.float64,
|
||
)
|
||
out = ta.RSI(14).batch(prices)
|
||
assert math.isclose(out[14], 70.464, abs_tol=0.05)
|
||
|
||
|
||
def test_inertia_constant_rvi_passes_through_linreg():
|
||
# Every bar identical (open, high, low, close) = (10, 11, 9, 10.5):
|
||
# 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(
|
||
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)
|
||
|
||
|
||
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))
|
||
ready = out[~np.isnan(out)]
|
||
assert ready.size > 0
|
||
assert ready.min() >= 0.0
|
||
assert ready.max() <= 100.0
|
||
|
||
|
||
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))
|
||
np.testing.assert_allclose(out, 50.0, atol=1e-12)
|
||
|
||
|
||
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))
|
||
# warmup_period = 5 + 3 + 3 - 2 = 9.
|
||
np.testing.assert_allclose(out[8:], 0.0, atol=1e-12)
|
||
|
||
|
||
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))
|
||
warmup = kst.warmup_period()
|
||
# Use NaN-safe comparison on the post-warmup tail.
|
||
tail = out[warmup - 1 :]
|
||
assert np.all(np.isfinite(tail))
|
||
np.testing.assert_allclose(tail, 0.0, atol=1e-12)
|
||
|
||
|
||
def test_pgo_flat_close_yields_zero():
|
||
# On a constant close the numerator (close − SMA) is zero, so PGO emits 0
|
||
# regardless of the TR-EMA in the denominator.
|
||
n = 20
|
||
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)
|
||
assert np.all(np.isnan(out[:4]))
|
||
np.testing.assert_allclose(out[4:], 0.0, atol=1e-12)
|
||
|
||
|
||
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(
|
||
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)
|
||
|
||
|
||
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_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))
|
||
# 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)
|
||
assert math.isclose(last[1], 0.0, abs_tol=1e-12)
|
||
assert math.isclose(last[2], 0.0, abs_tol=1e-12)
|
||
|
||
|
||
def test_awesome_oscillator_histogram_flat_series_converges_to_zero():
|
||
# Flat median price -> AO = 0 -> SMA(AO) = 0 -> AOHist = 0.
|
||
n = 50
|
||
high = np.full(n, 11.0)
|
||
low = np.full(n, 9.0)
|
||
out = ta.AwesomeOscillatorHistogram(3, 5, 3).batch(high, low)
|
||
# warmup = slow + sma - 1 = 5 + 3 - 1 = 7.
|
||
np.testing.assert_allclose(out[6:], 0.0, atol=1e-12)
|
||
|
||
|
||
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))
|
||
ready = out[~np.isnan(out)]
|
||
assert ready.size > 0
|
||
np.testing.assert_array_equal(ready[-5:], np.zeros(5))
|
||
|
||
|
||
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))
|
||
ready = out[~np.isnan(out)]
|
||
assert ready.size > 0
|
||
np.testing.assert_array_equal(ready, np.zeros_like(ready))
|
||
|
||
|
||
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)
|
||
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))
|
||
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))
|
||
# Last row's MACD and signal must be ~0.
|
||
last = out[-1]
|
||
assert math.isclose(last[0], 0.0, abs_tol=1e-9)
|
||
assert math.isclose(last[1], 0.0, abs_tol=1e-9)
|
||
assert math.isclose(last[2], 0.0, abs_tol=1e-9)
|
||
|
||
|
||
def test_bollinger_constant_series_zero_width():
|
||
out = 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))
|
||
ready = out[~np.isnan(out[:, 0])]
|
||
assert np.all(ready[:, 0] >= ready[:, 1])
|
||
assert np.all(ready[:, 1] >= ready[:, 2])
|
||
assert np.all(ready[:, 3] >= 0.0)
|
||
|
||
|
||
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)
|
||
# Once seeded, ATR equals the constant TR of 2.
|
||
np.testing.assert_allclose(out[13:], 2.0, atol=1e-12)
|
||
|
||
|
||
def test_stochastic_extremes():
|
||
# Close at the top of a 3-period range -> %K = 100.
|
||
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)
|
||
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)
|
||
np.testing.assert_allclose(out, [0.0, 20.0, -10.0, -10.0, 0.0])
|
||
|
||
|
||
def test_rvi_volatility_pure_uptrend_saturates_at_one_hundred():
|
||
# Strictly rising closes -> every stddev sample classified as "up" ->
|
||
# 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))
|
||
ready = out[~np.isnan(out)]
|
||
assert ready.size > 0
|
||
np.testing.assert_allclose(ready[-10:], 100.0, atol=1e-9)
|
||
|
||
|
||
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)
|
||
ready = out[~np.isnan(out)]
|
||
assert ready.size > 0
|
||
np.testing.assert_allclose(ready, 0.0, atol=1e-12)
|
||
|
||
|
||
def test_garman_klass_zero_movement_yields_zero():
|
||
# O == H == L == C every bar -> both log terms are zero -> sigma is zero.
|
||
o = np.full(30, 10.0)
|
||
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)
|
||
ready = out[~np.isnan(out)]
|
||
assert ready.size > 0
|
||
np.testing.assert_allclose(ready, 0.0, atol=1e-12)
|
||
|
||
|
||
def test_rogers_satchell_zero_movement_yields_zero():
|
||
o = np.full(30, 10.0)
|
||
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)
|
||
ready = out[~np.isnan(out)]
|
||
assert ready.size > 0
|
||
np.testing.assert_allclose(ready, 0.0, atol=1e-12)
|
||
|
||
|
||
def test_yang_zhang_zero_movement_yields_zero():
|
||
# O == H == L == C and constant across bars -> every sub-component is
|
||
# zero -> Yang-Zhang sigma is zero.
|
||
o = np.full(30, 10.0)
|
||
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)
|
||
ready = out[~np.isnan(out)]
|
||
assert ready.size > 0
|
||
np.testing.assert_allclose(ready, 0.0, atol=1e-12)
|