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

feat(family-09): add 7 trailing stops (HiLo, Volty, Yo-Yo, Donchian, Pct, Step, Renko) (#46)

Browse Source 
* feat(family-09): add 7 trailing stops (HiLo, Volty, Yo-Yo, Donchian, Pct, Step, Renko)

Rounds out the Trailing Stops family from 5 to 12 indicators:

- HiLoActivator (Crabel): SMA-of-high/SMA-of-low trail with a one-bar
  lag; emits the opposite-side SMA as the trailing stop.
- VoltyStop (Cynthia Kase): ATR trail anchored on the extreme close
  since the trade was opened — tighter than AtrTrailingStop on
  pullbacks.
- YoyoExit: long-only ATR trail with an explicit re-entry trigger at
  trail + multiplier*ATR; exposes an in_trade flag.
- DonchianStop (Turtle): lowest low / highest high over the window;
  multi-output {stop_long, stop_short}.
- PercentageTrailingStop: fixed-percent trail that scales across
  instruments without per-asset tuning.
- StepTrailingStop: snaps to a step_size-aligned grid; mirrors
  discretionary stop-by-hand workflow.
- RenkoTrailingStop: block-anchored trail; only moves on full-block
  advances, ignores intra-block noise.

All seven are wired into wickra-core, the Python / Node / WASM
bindings, the indicator_update + indicator_update_candle fuzz targets,
the wickra bench harness, and the Python + Node test suites. README
counter bumps from 71 to 78; CHANGELOG entry under [Unreleased].

* fix(family-09): satisfy pedantic clippy lints

- hilo_activator: rewrite match-Some/None as if-let-else (single_match_else),
  add backticks around the HiLo identifier in module/struct doc (doc_markdown).
- percentage / step / renko trailing stop tests: use f64::from(i32) instead
  of `as f64` (cast_lossless).
- bench `benches()` is now >100 lines after Family 09 was wired in; allow
  too_many_lines (matches the python pymodule fn).
This commit is contained in:
kingchenc
2026-05-25 19:36:14 +02:00
committed by GitHub
parent 880a0e7430
commit f10b8c2e2d
22 changed files with 2965 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bindings/python/python/wickra/__init__.py
+14
View File
@@ -107,6 +107,13 @@ from ._wickra import (
ChandelierExit,
ChandeKrollStop,
AtrTrailingStop,
HiLoActivator,
VoltyStop,
YoyoExit,
DonchianStop,
PercentageTrailingStop,
StepTrailingStop,
RenkoTrailingStop,
TrueRange,
ChaikinVolatility,
RVIVolatility,
@@ -240,6 +247,13 @@ __all__ = [
"ChandelierExit",
"ChandeKrollStop",
"AtrTrailingStop",
"HiLoActivator",
"VoltyStop",
"YoyoExit",
"DonchianStop",
"PercentageTrailingStop",
"StepTrailingStop",
"RenkoTrailingStop",
"TrueRange",
"ChaikinVolatility",
"RVIVolatility",
bindings/python/src/lib.rs
+437
View File
@@ -5763,6 +5763,436 @@ impl PyAtrTrailingStop {
}
}
// ============================== HiLo Activator ==============================
#[pyclass(name = "HiLoActivator", module = "wickra._wickra", skip_from_py_object)]
#[derive(Clone)]
struct PyHiLoActivator {
inner: wc::HiLoActivator,
}
#[pymethods]
impl PyHiLoActivator {
#[new]
#[pyo3(signature = (period=3))]
fn new(period: usize) -> PyResult<Self> {
Ok(Self {
inner: wc::HiLoActivator::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>,
high: PyReadonlyArray1<'py, f64>,
low: PyReadonlyArray1<'py, f64>,
close: PyReadonlyArray1<'py, f64>,
) -> PyResult<Bound<'py, PyArray1<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))?;
let c = close
.as_slice()
.map_err(|_| PyValueError::new_err(NON_CONTIGUOUS))?;
if h.len() != l.len() || l.len() != c.len() {
return Err(PyValueError::new_err(
"high, low, close must be equal length",
));
}
let mut out = Vec::with_capacity(h.len());
for i in 0..h.len() {
let candle = wc::Candle::new(c[i], h[i], l[i], c[i], 0.0, 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!("HiLoActivator(period={})", self.inner.period())
}
}
// ============================== Volty Stop ==============================
#[pyclass(name = "VoltyStop", module = "wickra._wickra", skip_from_py_object)]
#[derive(Clone)]
struct PyVoltyStop {
inner: wc::VoltyStop,
}
#[pymethods]
impl PyVoltyStop {
#[new]
#[pyo3(signature = (atr_period=14, multiplier=2.0))]
fn new(atr_period: usize, multiplier: f64) -> PyResult<Self> {
Ok(Self {
inner: wc::VoltyStop::new(atr_period, multiplier).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>,
high: PyReadonlyArray1<'py, f64>,
low: PyReadonlyArray1<'py, f64>,
close: PyReadonlyArray1<'py, f64>,
) -> PyResult<Bound<'py, PyArray1<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))?;
let c = close
.as_slice()
.map_err(|_| PyValueError::new_err(NON_CONTIGUOUS))?;
if h.len() != l.len() || l.len() != c.len() {
return Err(PyValueError::new_err(
"high, low, close must be equal length",
));
}
let mut out = Vec::with_capacity(h.len());
for i in 0..h.len() {
let candle = wc::Candle::new(c[i], h[i], l[i], c[i], 0.0, 0).map_err(map_err)?;
out.push(self.inner.update(candle).unwrap_or(f64::NAN));
}
Ok(out.into_pyarray(py))
}
#[getter]
fn params(&self) -> (usize, f64) {
self.inner.params()
}
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, m) = self.inner.params();
format!("VoltyStop(atr_period={p}, multiplier={m})")
}
}
// ============================== Yo-Yo Exit ==============================
#[pyclass(name = "YoyoExit", module = "wickra._wickra", skip_from_py_object)]
#[derive(Clone)]
struct PyYoyoExit {
inner: wc::YoyoExit,
}
#[pymethods]
impl PyYoyoExit {
#[new]
#[pyo3(signature = (atr_period=14, multiplier=2.0))]
fn new(atr_period: usize, multiplier: f64) -> PyResult<Self> {
Ok(Self {
inner: wc::YoyoExit::new(atr_period, multiplier).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>,
high: PyReadonlyArray1<'py, f64>,
low: PyReadonlyArray1<'py, f64>,
close: PyReadonlyArray1<'py, f64>,
) -> PyResult<Bound<'py, PyArray1<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))?;
let c = close
.as_slice()
.map_err(|_| PyValueError::new_err(NON_CONTIGUOUS))?;
if h.len() != l.len() || l.len() != c.len() {
return Err(PyValueError::new_err(
"high, low, close must be equal length",
));
}
let mut out = Vec::with_capacity(h.len());
for i in 0..h.len() {
let candle = wc::Candle::new(c[i], h[i], l[i], c[i], 0.0, 0).map_err(map_err)?;
out.push(self.inner.update(candle).unwrap_or(f64::NAN));
}
Ok(out.into_pyarray(py))
}
#[getter]
fn params(&self) -> (usize, f64) {
self.inner.params()
}
#[getter]
fn in_trade(&self) -> bool {
self.inner.in_trade()
}
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, m) = self.inner.params();
format!("YoyoExit(atr_period={p}, multiplier={m})")
}
}
// ============================== Donchian Stop ==============================
#[pyclass(name = "DonchianStop", module = "wickra._wickra", skip_from_py_object)]
#[derive(Clone)]
struct PyDonchianStop {
inner: wc::DonchianStop,
}
#[pymethods]
impl PyDonchianStop {
#[new]
#[pyo3(signature = (period=10))]
fn new(period: usize) -> PyResult<Self> {
Ok(Self {
inner: wc::DonchianStop::new(period).map_err(map_err)?,
})
}
fn update(&mut self, candle: &Bound<'_, PyAny>) -> PyResult<Option<(f64, f64)>> {
let c = extract_candle(candle)?;
Ok(self.inner.update(c).map(|o| (o.stop_long, o.stop_short)))
}
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 * 2];
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 * 2] = o.stop_long;
out[i * 2 + 1] = o.stop_short;
}
}
Ok(numpy::ndarray::Array2::from_shape_vec((n, 2), out)
.expect("shape consistent")
.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!("DonchianStop(period={})", self.inner.period())
}
}
// ============================== Percentage Trailing Stop ==============================
#[pyclass(
name = "PercentageTrailingStop",
module = "wickra._wickra",
skip_from_py_object
)]
#[derive(Clone)]
struct PyPercentageTrailingStop {
inner: wc::PercentageTrailingStop,
}
#[pymethods]
impl PyPercentageTrailingStop {
#[new]
#[pyo3(signature = (percent=5.0))]
fn new(percent: f64) -> PyResult<Self> {
Ok(Self {
inner: wc::PercentageTrailingStop::new(percent).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 slice = prices
.as_slice()
.map_err(|_| PyValueError::new_err(NON_CONTIGUOUS))?;
Ok(flatten(self.inner.batch(slice)).into_pyarray(py))
}
#[getter]
fn percent(&self) -> f64 {
self.inner.percent()
}
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!("PercentageTrailingStop(percent={})", self.inner.percent())
}
}
// ============================== Step Trailing Stop ==============================
#[pyclass(
name = "StepTrailingStop",
module = "wickra._wickra",
skip_from_py_object
)]
#[derive(Clone)]
struct PyStepTrailingStop {
inner: wc::StepTrailingStop,
}
#[pymethods]
impl PyStepTrailingStop {
#[new]
#[pyo3(signature = (step_size=1.0))]
fn new(step_size: f64) -> PyResult<Self> {
Ok(Self {
inner: wc::StepTrailingStop::new(step_size).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 slice = prices
.as_slice()
.map_err(|_| PyValueError::new_err(NON_CONTIGUOUS))?;
Ok(flatten(self.inner.batch(slice)).into_pyarray(py))
}
#[getter]
fn step_size(&self) -> f64 {
self.inner.step_size()
}
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!("StepTrailingStop(step_size={})", self.inner.step_size())
}
}
// ============================== Renko Trailing Stop ==============================
#[pyclass(
name = "RenkoTrailingStop",
module = "wickra._wickra",
skip_from_py_object
)]
#[derive(Clone)]
struct PyRenkoTrailingStop {
inner: wc::RenkoTrailingStop,
}
#[pymethods]
impl PyRenkoTrailingStop {
#[new]
#[pyo3(signature = (block_size=1.0))]
fn new(block_size: f64) -> PyResult<Self> {
Ok(Self {
inner: wc::RenkoTrailingStop::new(block_size).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 slice = prices
.as_slice()
.map_err(|_| PyValueError::new_err(NON_CONTIGUOUS))?;
Ok(flatten(self.inner.batch(slice)).into_pyarray(py))
}
#[getter]
fn block_size(&self) -> f64 {
self.inner.block_size()
}
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!("RenkoTrailingStop(block_size={})", self.inner.block_size())
}
}
// ============================== Typical Price ==============================
#[pyclass(name = "TypicalPrice", module = "wickra._wickra", skip_from_py_object)]
@@ -7704,6 +8134,13 @@ fn _wickra(_py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()> {
m.add_class::<PyChandelierExit>()?;
m.add_class::<PyChandeKrollStop>()?;
m.add_class::<PyAtrTrailingStop>()?;
m.add_class::<PyHiLoActivator>()?;
m.add_class::<PyVoltyStop>()?;
m.add_class::<PyYoyoExit>()?;
m.add_class::<PyDonchianStop>()?;
m.add_class::<PyPercentageTrailingStop>()?;
m.add_class::<PyStepTrailingStop>()?;
m.add_class::<PyRenkoTrailingStop>()?;
m.add_class::<PyTypicalPrice>()?;
m.add_class::<PyMedianPrice>()?;
m.add_class::<PyWeightedClose>()?;
bindings/python/tests/test_known_values.py
+72
View File
@@ -332,6 +332,78 @@ def test_obv_cumulative_known_sequence():
np.testing.assert_allclose(out, [0.0, 20.0, -10.0, -10.0, 0.0])
def test_percentage_trailing_stop_seed_and_ratchet():
# 10% trail: first close 100 -> stop 90; next 110 -> stop max(90, 99) = 99.
s = ta.PercentageTrailingStop(10.0)
assert math.isclose(s.update(100.0), 90.0, abs_tol=1e-12)
assert math.isclose(s.update(110.0), 99.0, abs_tol=1e-12)
def test_step_trailing_stop_snaps_below_close():
# step 1: floor((100.4 - 1) / 1) = 99.
s = ta.StepTrailingStop(1.0)
assert math.isclose(s.update(100.4), 99.0, abs_tol=1e-12)
def test_renko_trailing_stop_holds_until_full_block():
# block 1: seed 100 -> stop 99; 100.5 still 99; 101 -> stop 100.
s = ta.RenkoTrailingStop(1.0)
assert math.isclose(s.update(100.0), 99.0, abs_tol=1e-12)
assert math.isclose(s.update(100.5), 99.0, abs_tol=1e-12)
assert math.isclose(s.update(101.0), 100.0, abs_tol=1e-12)
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)
# First 4 rows NaN, fifth row: stop_long = 0, stop_short = 5.
for i in range(4):
assert math.isnan(out[i, 0])
assert math.isnan(out[i, 1])
assert math.isclose(out[4, 0], 0.0, abs_tol=1e-12)
assert math.isclose(out[4, 1], 5.0, abs_tol=1e-12)
def test_hilo_activator_flat_market_holds_low_sma():
# Flat candles H=11, L=9, C=10 -> close (10) sits between bands, so the
# initial long seed is preserved: emitted stop = lo_sma = 9.
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)
# 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])
for i in range(3, 15):
assert math.isclose(out[i], 9.0, abs_tol=1e-12)
def test_volty_stop_flat_market_constant_level():
# ATR=2, mult=2 -> band 4; anchor stays at close 10 -> stop = 10 - 4 = 6.
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)
for i in range(4):
assert math.isnan(out[i])
for i in range(4, 20):
assert math.isclose(out[i], 6.0, abs_tol=1e-12)
def test_yoyo_exit_flat_market_constant_level():
# ATR=2, mult=2 -> band 4; trail = close - band = 10 - 4 = 6 and holds.
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)
for i in range(4):
assert math.isnan(out[i])
for i in range(4, 20):
assert math.isclose(out[i], 6.0, abs_tol=1e-12)
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
bindings/python/tests/test_new_indicators.py
+19
View File
@@ -73,6 +73,9 @@ SCALAR = [
(ta.VerticalHorizontalFilter, (28,)),
(ta.ZScore, (20,)),
(ta.LinRegAngle, (14,)),
(ta.PercentageTrailingStop, (5.0,)),
(ta.StepTrailingStop, (1.0,)),
(ta.RenkoTrailingStop, (1.0,)),
(ta.LaguerreRSI, (0.5,)),
(ta.ConnorsRSI, (3, 2, 100)),
(ta.RVIVolatility, (10,)),
@@ -199,6 +202,18 @@ CANDLE_SCALAR = {
lambda: ta.AtrTrailingStop(14, 3.0),
lambda ind, h, l, c, v: ind.batch(h, l, c),
),
"HiLoActivator": (
lambda: ta.HiLoActivator(3),
lambda ind, h, l, c, v: ind.batch(h, l, c),
),
"VoltyStop": (
lambda: ta.VoltyStop(14, 2.0),
lambda ind, h, l, c, v: ind.batch(h, l, c),
),
"YoyoExit": (
lambda: ta.YoyoExit(14, 2.0),
lambda ind, h, l, c, v: ind.batch(h, l, c),
),
"TypicalPrice": (
lambda: ta.TypicalPrice(),
lambda ind, h, l, c, v: ind.batch(h, l, c),
@@ -307,6 +322,10 @@ MULTI = {
lambda: ta.ChandeKrollStop(10, 1.0, 9),
lambda ind, h, l, c, v: ind.batch(h, l, c),
),
"DonchianStop": (
lambda: ta.DonchianStop(10),
lambda ind, h, l, c, v: ind.batch(h, l),
),
# Family 05 candle-input bands. Each entry is
# `(factory, batch_call, output_arity, streaming_fields)` where
# `streaming_fields` is the tuple shape returned by `update(...)`.