gavindiaz/QuanTAlib
1
0
Fork 0
mirror of https://github.com/mihakralj/QuanTAlib.git synced 2026-08-03 11:47:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

docs: add license rationale, Python/PineScript guides, API updates

Browse Source 
- Add docs/license.md with Apache 2.0 rationale and patent protection analysis
- Add docs/python.md and docs/pinescript.md platform guides
- Expand README license section with disclosure and link to rationale
- Update docs/api.md and docs/architecture.md
- Update Python bindings: helpers, all indicator modules, pyproject.toml
- Add Python tests for Arrow and Polars integration
- Update TValue core type and documentation
- Add fix_length_to_period tooling script
This commit is contained in:
Miha Kralj
2026-03-03 22:11:35 -08:00
parent 6f4e083811
commit f10baa6dfb
28 changed files with 2050 additions and 542 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/quantalib/_helpers.py
+146 -38
View File
@@ -9,35 +9,103 @@ from numpy.typing import NDArray
from ._bridge import _lib, _check, _dp, _ci, _cd
# Optional pandas support
# ---------------------------------------------------------------------------
# Optional dataframe library imports (all optional; numpy is the only hard dep)
# ---------------------------------------------------------------------------
try:
import pandas as pd # type: ignore[import-untyped]
except ImportError: # pragma: no cover
pd = None # type: ignore[assignment]
try:
import polars as pl # type: ignore[import-untyped]
except ImportError: # pragma: no cover
pl = None # type: ignore[assignment]
try:
import pyarrow as pa # type: ignore[import-untyped]
except ImportError: # pragma: no cover
pa = None # type: ignore[assignment]
# ---------------------------------------------------------------------------
# Origin token — opaque marker passed from _arr() → _wrap()/_wrap_multi()
#
# Callers in category modules simply do:
# src, origin = _arr(close)
# ...
# return _wrap(dst, origin, name, category, offset)
#
# The token carries enough info to reconstruct the original container type.
# ---------------------------------------------------------------------------
_ORIGIN_NUMPY = "numpy"
_ORIGIN_PANDAS = "pandas"
_ORIGIN_POLARS = "polars"
_ORIGIN_PYARROW = "pyarrow"
class _Origin:
"""Lightweight tag recording the input container type + metadata."""
__slots__ = ("kind", "meta")
def __init__(self, kind: str, meta: object = None) -> None:
self.kind = kind
self.meta = meta # pandas.Index | polars series name | None
# ---------------------------------------------------------------------------
# Internal helpers
# ---------------------------------------------------------------------------
_F64 = np.float64
def _arr(x: object) -> tuple[NDArray[np.float64], object]:
"""Return (contiguous float64 array, original_index_or_None)."""
def _arr(x: object) -> tuple[NDArray[np.float64], _Origin | None]:
"""Return *(contiguous float64 array, origin_token)*.
Supported input types
---------------------
* ``numpy.ndarray``
* ``pandas.Series`` / ``pandas.DataFrame`` (first column)
* ``polars.Series`` / ``polars.DataFrame`` (first column)
* ``pyarrow.Array`` / ``pyarrow.ChunkedArray``
* Any sequence coercible via ``np.asarray``
"""
if x is None:
raise ValueError("Input array must not be None")
idx = None
if pd is not None and isinstance(x, pd.Series):
idx = x.index
x = x.to_numpy(dtype=_F64, copy=False)
elif pd is not None and isinstance(x, pd.DataFrame):
idx = x.index
x = x.iloc[:, 0].to_numpy(dtype=_F64, copy=False)
origin: _Origin | None = None
# ── pandas ──────────────────────────────────────────────────────────
if pd is not None and isinstance(x, (pd.Series, pd.DataFrame)):
if isinstance(x, pd.DataFrame):
origin = _Origin(_ORIGIN_PANDAS, x.index)
x = x.iloc[:, 0].to_numpy(dtype=_F64, copy=False)
else:
origin = _Origin(_ORIGIN_PANDAS, x.index)
x = x.to_numpy(dtype=_F64, copy=False)
# ── polars ──────────────────────────────────────────────────────────
elif pl is not None and isinstance(x, (pl.Series, pl.DataFrame)):
if isinstance(x, pl.DataFrame):
col = x.get_column(x.columns[0])
origin = _Origin(_ORIGIN_POLARS, col.name)
x = col.cast(pl.Float64).to_numpy(allow_copy=True)
else:
origin = _Origin(_ORIGIN_POLARS, x.name)
x = x.cast(pl.Float64).to_numpy(allow_copy=True)
# ── pyarrow ─────────────────────────────────────────────────────────
elif pa is not None and isinstance(x, (pa.Array, pa.ChunkedArray)):
origin = _Origin(_ORIGIN_PYARROW)
x = x.to_numpy(zero_copy_only=False).astype(_F64, copy=False)
# ── coerce to numpy ────────────────────────────────────────────────
arr = np.asarray(x, dtype=_F64)
if not arr.flags["C_CONTIGUOUS"]:
arr = np.ascontiguousarray(arr)
if arr.ndim == 0 or len(arr) == 0:
raise ValueError("Input array must not be empty")
return arr, idx
return arr, origin
def _ptr(a: NDArray[np.float64]): # noqa: ANN202
@@ -63,33 +131,73 @@ def _offset(arr: NDArray[np.float64], off: int) -> NDArray[np.float64]:
def _wrap(
arr: NDArray[np.float64],
idx: object,
origin: _Origin | None,
name: str,
category: str,
offset: int = 0,
):
"""Wrap result: apply offset, optionally convert to pd.Series."""
"""Wrap result: apply offset, reconstruct the original container type.
Return types by origin
----------------------
* numpy → ``np.ndarray``
* pandas → ``pd.Series`` with original index
* polars → ``pl.Series``
* pyarrow → ``pa.Array`` (float64)
"""
arr = _offset(arr, offset)
if idx is not None and pd is not None:
s = pd.Series(arr, index=idx, name=name)
if origin is None:
return arr
if origin.kind == _ORIGIN_PANDAS and pd is not None:
s = pd.Series(arr, index=origin.meta, name=name)
s.attrs["category"] = category
return s
if origin.kind == _ORIGIN_POLARS and pl is not None:
return pl.Series(name=name, values=arr)
if origin.kind == _ORIGIN_PYARROW and pa is not None:
return pa.array(arr, type=pa.float64())
return arr
def _wrap_multi(
arrays: dict[str, NDArray[np.float64]],
idx: object,
origin: _Origin | None,
category: str,
offset: int = 0,
):
"""Wrap multi-output result into tuple or DataFrame."""
"""Wrap multi-output result into the original container type.
Return types by origin
----------------------
* numpy → ``tuple[np.ndarray, ...]``
* pandas → ``pd.DataFrame`` with original index
* polars → ``pl.DataFrame``
* pyarrow → ``dict[str, pa.Array]``
"""
for k in arrays:
arrays[k] = _offset(arrays[k], offset)
if idx is not None and pd is not None:
df = pd.DataFrame(arrays, index=idx)
if origin is None:
return tuple(arrays.values())
if origin.kind == _ORIGIN_PANDAS and pd is not None:
df = pd.DataFrame(arrays, index=origin.meta)
df.attrs["category"] = category
return df
if origin.kind == _ORIGIN_POLARS and pl is not None:
return pl.DataFrame(
{k: pl.Series(name=k, values=v) for k, v in arrays.items()}
)
if origin.kind == _ORIGIN_PYARROW and pa is not None:
return {k: pa.array(v, type=pa.float64()) for k, v in arrays.items()}
return tuple(arrays.values())
@@ -98,17 +206,17 @@ def _wrap_multi(
# ═══════════════════════════════════════════════════════════════════════════
def _pa(
fn_name: str, close: object, length: int, offset: int,
default_length: int, label: str, category: str,
fn_name: str, close: object, period: int, offset: int,
default_period: int, label: str, category: str,
) -> object:
"""Generic Pattern A wrapper: single-input + period."""
length = int(length) if length is not None else default_length
period = int(period) if period is not None else default_period
offset = int(offset) if offset is not None else 0
src, idx = _arr(close)
n = len(src)
dst = _out(n)
_check(getattr(_lib, fn_name)(_ptr(src), n, _ptr(dst), length))
return _wrap(dst, idx, f"{label}_{length}", category, offset)
_check(getattr(_lib, fn_name)(_ptr(src), n, _ptr(dst), period))
return _wrap(dst, idx, f"{label}_{period}", category, offset)
def _pa3(
@@ -126,18 +234,18 @@ def _pa3(
def _pf(
fn_name: str, actual: object, predicted: object,
length: int, offset: int, default_length: int,
period: int, offset: int, default_period: int,
label: str, category: str,
) -> object:
"""Generic Pattern F wrapper: actual+predicted+period."""
length = int(length) if length is not None else default_length
period = int(period) if period is not None else default_period
offset = int(offset) if offset is not None else 0
a, idx = _arr(actual)
p, _ = _arr(predicted)
n = len(a)
dst = _out(n)
_check(getattr(_lib, fn_name)(_ptr(a), _ptr(p), n, _ptr(dst), length))
return _wrap(dst, idx, f"{label}_{length}", category, offset)
_check(getattr(_lib, fn_name)(_ptr(a), _ptr(p), n, _ptr(dst), period))
return _wrap(dst, idx, f"{label}_{period}", category, offset)
def _pg(
@@ -155,33 +263,33 @@ def _pg(
def _pg2(
fn_name: str, close: object, volume: object, length: int,
offset: int, default_length: int, label: str, category: str,
fn_name: str, close: object, volume: object, period: int,
offset: int, default_period: int, label: str, category: str,
) -> object:
"""Pattern G2: source+volume+period."""
length = int(length) if length is not None else default_length
period = int(period) if period is not None else default_period
offset = int(offset) if offset is not None else 0
c, idx = _arr(close)
v, _ = _arr(volume)
n = len(c)
dst = _out(n)
_check(getattr(_lib, fn_name)(_ptr(c), _ptr(v), n, _ptr(dst), length))
return _wrap(dst, idx, f"{label}_{length}", category, offset)
_check(getattr(_lib, fn_name)(_ptr(c), _ptr(v), n, _ptr(dst), period))
return _wrap(dst, idx, f"{label}_{period}", category, offset)
def _ph(
fn_name: str, x: object, y: object, length: int,
offset: int, default_length: int, label: str, category: str,
fn_name: str, x: object, y: object, period: int,
offset: int, default_period: int, label: str, category: str,
) -> object:
"""Pattern H: X+Y+period."""
length = int(length) if length is not None else default_length
period = int(period) if period is not None else default_period
offset = int(offset) if offset is not None else 0
xarr, idx = _arr(x)
yarr, _ = _arr(y)
n = len(xarr)
dst = _out(n)
_check(getattr(_lib, fn_name)(_ptr(xarr), _ptr(yarr), n, _ptr(dst), length))
return _wrap(dst, idx, f"{label}_{length}", category, offset)
_check(getattr(_lib, fn_name)(_ptr(xarr), _ptr(yarr), n, _ptr(dst), period))
return _wrap(dst, idx, f"{label}_{period}", category, offset)
def _ohlcv_bars_period(
python/quantalib/channels.py
+8 -8
View File
@@ -315,15 +315,15 @@ def vwapsd(price: object, volume: object, numDevs: float = 2.0, offset: int = 0,
return _wrap_multi({"upper": upper, "lower": lower, "vwap": vwap, "stdDev": stdDev}, idx, "channels", offset)
def bbands(close: object, length: int = 20, std: float = 2.0,
def bbands(close: object, period: int = 20, std: float = 2.0,
offset: int = 0, **kwargs) -> object:
"""Bollinger Bands -> (upper, mid, lower) or DataFrame."""
length = int(length); std = float(std); offset = int(offset)
period = int(kwargs.get("length", period)); std = float(std); offset = int(offset)
src, idx = _arr(close); n = len(src)
upper = _out(n); mid = _out(n); lower = _out(n)
_check(_lib.qtl_bbands(_ptr(src), n, _ptr(upper), _ptr(mid), _ptr(lower), length, std))
_check(_lib.qtl_bbands(_ptr(src), n, _ptr(upper), _ptr(mid), _ptr(lower), period, std))
return _wrap_multi(
{f"BBU_{length}_{std}": upper, f"BBM_{length}_{std}": mid, f"BBL_{length}_{std}": lower},
{f"BBU_{period}_{std}": upper, f"BBM_{period}_{std}": mid, f"BBL_{period}_{std}": lower},
idx, "channels", offset)
@@ -341,12 +341,12 @@ def atrbands(high: object, low: object, close: object,
idx, "channels", offset)
def apchannel(high: object, low: object, length: int = 20,
def apchannel(high: object, low: object, period: int = 20,
offset: int = 0, **kwargs) -> object:
"""Average Price Channel -> (upper, lower) or DataFrame."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
h, idx = _arr(high); l, _ = _arr(low)
n = len(h)
upper = _out(n); lower = _out(n)
_check(_lib.qtl_apchannel(_ptr(h), _ptr(l), n, _ptr(upper), _ptr(lower), float(length)))
return _wrap_multi({f"APCU_{length}": upper, f"APCL_{length}": lower}, idx, "channels", offset)
_check(_lib.qtl_apchannel(_ptr(h), _ptr(l), n, _ptr(upper), _ptr(lower), float(period)))
return _wrap_multi({f"APCU_{period}": upper, f"APCL_{period}": lower}, idx, "channels", offset)
python/quantalib/cycles.py
+12 -12
View File
@@ -108,29 +108,29 @@ def ssfdsp(close: object, period: int = 14, offset: int = 0, **kwargs) -> object
_check(_lib.qtl_ssfdsp(_ptr(src), _ptr(output), n, period))
return _wrap(output, idx, f"SSFDSP_{period}", "cycles", offset)
def cg(close: object, length: int = 10, offset: int = 0, **kwargs) -> object:
def cg(close: object, period: int = 10, offset: int = 0, **kwargs) -> object:
"""Center of Gravity."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_cg(_ptr(src), n, _ptr(dst), length))
return _wrap(dst, idx, f"CG_{length}", "cycles", offset)
_check(_lib.qtl_cg(_ptr(src), n, _ptr(dst), period))
return _wrap(dst, idx, f"CG_{period}", "cycles", offset)
def dsp(close: object, length: int = 20, offset: int = 0, **kwargs) -> object:
def dsp(close: object, period: int = 20, offset: int = 0, **kwargs) -> object:
"""Dominant Cycle Period (DSP)."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_dsp(_ptr(src), n, _ptr(dst), length))
return _wrap(dst, idx, f"DSP_{length}", "cycles", offset)
_check(_lib.qtl_dsp(_ptr(src), n, _ptr(dst), period))
return _wrap(dst, idx, f"DSP_{period}", "cycles", offset)
def ccor(close: object, length: int = 20, alpha: float = 0.07,
def ccor(close: object, period: int = 20, alpha: float = 0.07,
offset: int = 0, **kwargs) -> object:
"""Circular Correlation."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_ccor(_ptr(src), n, _ptr(dst), length, float(alpha)))
return _wrap(dst, idx, f"CCOR_{length}", "cycles", offset)
_check(_lib.qtl_ccor(_ptr(src), n, _ptr(dst), period, float(alpha)))
return _wrap(dst, idx, f"CCOR_{period}", "cycles", offset)
def ebsw(close: object, hp_length: int = 40, ssf_length: int = 10,
python/quantalib/errors.py
+16 -16
View File
@@ -282,41 +282,41 @@ def wrmse(actual: object, predicted: object, period: int = 14, offset: int = 0,
_check(_lib.qtl_wrmse(_ptr(a), _ptr(p), _ptr(output), n, period))
return _wrap(output, idx, f"WRMSE_{period}", "errors", offset)
def mse(actual: object, predicted: object, length: int = 20,
def mse(actual: object, predicted: object, period: int = 20,
offset: int = 0, **kwargs) -> object:
"""Mean Squared Error."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
a, idx = _arr(actual); p, _ = _arr(predicted)
n = len(a); dst = _out(n)
_check(_lib.qtl_mse(_ptr(a), _ptr(p), n, _ptr(dst), length))
return _wrap(dst, idx, f"MSE_{length}", "errors", offset)
_check(_lib.qtl_mse(_ptr(a), _ptr(p), n, _ptr(dst), period))
return _wrap(dst, idx, f"MSE_{period}", "errors", offset)
def rmse(actual: object, predicted: object, length: int = 20,
def rmse(actual: object, predicted: object, period: int = 20,
offset: int = 0, **kwargs) -> object:
"""Root Mean Squared Error."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
a, idx = _arr(actual); p, _ = _arr(predicted)
n = len(a); dst = _out(n)
_check(_lib.qtl_rmse(_ptr(a), _ptr(p), n, _ptr(dst), length))
return _wrap(dst, idx, f"RMSE_{length}", "errors", offset)
_check(_lib.qtl_rmse(_ptr(a), _ptr(p), n, _ptr(dst), period))
return _wrap(dst, idx, f"RMSE_{period}", "errors", offset)
def mae(actual: object, predicted: object, length: int = 20,
def mae(actual: object, predicted: object, period: int = 20,
offset: int = 0, **kwargs) -> object:
"""Mean Absolute Error."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
a, idx = _arr(actual); p, _ = _arr(predicted)
n = len(a); dst = _out(n)
_check(_lib.qtl_mae(_ptr(a), _ptr(p), n, _ptr(dst), length))
return _wrap(dst, idx, f"MAE_{length}", "errors", offset)
_check(_lib.qtl_mae(_ptr(a), _ptr(p), n, _ptr(dst), period))
return _wrap(dst, idx, f"MAE_{period}", "errors", offset)
def mape(actual: object, predicted: object, length: int = 20,
def mape(actual: object, predicted: object, period: int = 20,
offset: int = 0, **kwargs) -> object:
"""Mean Absolute Percentage Error."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
a, idx = _arr(actual); p, _ = _arr(predicted)
n = len(a); dst = _out(n)
_check(_lib.qtl_mape(_ptr(a), _ptr(p), n, _ptr(dst), length))
return _wrap(dst, idx, f"MAPE_{length}", "errors", offset)
_check(_lib.qtl_mape(_ptr(a), _ptr(p), n, _ptr(dst), period))
return _wrap(dst, idx, f"MAPE_{period}", "errors", offset)
python/quantalib/filters.py
+56 -56
View File
@@ -57,15 +57,15 @@ def gauss(close: object, sigma: float = 1.0, offset: int = 0, **kwargs) -> objec
return _wrap(output, idx, "GAUSS", "filters", offset)
def hann(close: object, length: int = 14, offset: int = 0, **kwargs) -> object:
def hann(close: object, period: int = 14, offset: int = 0, **kwargs) -> object:
"""Hann Filter."""
length = int(length)
period = int(kwargs.get("length", period))
offset = int(offset)
src, idx = _arr(close)
n = len(src)
output = _out(n)
_check(_lib.qtl_hann(_ptr(src), _ptr(output), n, length))
return _wrap(output, idx, f"HANN_{length}", "filters", offset)
_check(_lib.qtl_hann(_ptr(src), _ptr(output), n, period))
return _wrap(output, idx, f"HANN_{period}", "filters", offset)
def hp(close: object, lam: float = 1600.0, offset: int = 0, **kwargs) -> object:
@@ -79,15 +79,15 @@ def hp(close: object, lam: float = 1600.0, offset: int = 0, **kwargs) -> object:
return _wrap(output, idx, "HP", "filters", offset)
def hpf(close: object, length: int = 40, offset: int = 0, **kwargs) -> object:
def hpf(close: object, period: int = 40, offset: int = 0, **kwargs) -> object:
"""High-Pass Filter."""
length = int(length)
period = int(kwargs.get("length", period))
offset = int(offset)
src, idx = _arr(close)
n = len(src)
output = _out(n)
_check(_lib.qtl_hpf(_ptr(src), _ptr(output), n, length))
return _wrap(output, idx, f"HPF_{length}", "filters", offset)
_check(_lib.qtl_hpf(_ptr(src), _ptr(output), n, period))
return _wrap(output, idx, f"HPF_{period}", "filters", offset)
def kalman(close: object, q: float = 0.01, r: float = 0.1, offset: int = 0, **kwargs) -> object:
@@ -317,106 +317,106 @@ def wiener(close: object, period: int = 14, smoothPeriod: int = 10, offset: int
_check(_lib.qtl_wiener(_ptr(src), _ptr(destination), n, period, smoothPeriod))
return _wrap(destination, idx, f"WIENER_{period}", "filters", offset)
def bessel(close: object, length: int = 14, offset: int = 0, **kwargs) -> object:
def bessel(close: object, period: int = 14, offset: int = 0, **kwargs) -> object:
"""Bessel Filter."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_bessel(_ptr(src), n, _ptr(dst), length))
return _wrap(dst, idx, f"BESSEL_{length}", "filters", offset)
_check(_lib.qtl_bessel(_ptr(src), n, _ptr(dst), period))
return _wrap(dst, idx, f"BESSEL_{period}", "filters", offset)
def butter2(close: object, length: int = 14, gain: float = 1.0,
def butter2(close: object, period: int = 14, gain: float = 1.0,
offset: int = 0, **kwargs) -> object:
"""2nd-order Butterworth."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_butter2(_ptr(src), n, _ptr(dst), length, float(gain)))
return _wrap(dst, idx, f"BUTTER2_{length}", "filters", offset)
_check(_lib.qtl_butter2(_ptr(src), n, _ptr(dst), period, float(gain)))
return _wrap(dst, idx, f"BUTTER2_{period}", "filters", offset)
def butter3(close: object, length: int = 14, gain: float = 1.0,
def butter3(close: object, period: int = 14, gain: float = 1.0,
offset: int = 0, **kwargs) -> object:
"""3rd-order Butterworth."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_butter3(_ptr(src), n, _ptr(dst), length, float(gain)))
return _wrap(dst, idx, f"BUTTER3_{length}", "filters", offset)
_check(_lib.qtl_butter3(_ptr(src), n, _ptr(dst), period, float(gain)))
return _wrap(dst, idx, f"BUTTER3_{period}", "filters", offset)
def cheby1(close: object, length: int = 14, ripple: float = 0.5,
def cheby1(close: object, period: int = 14, ripple: float = 0.5,
offset: int = 0, **kwargs) -> object:
"""Chebyshev Type I."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_cheby1(_ptr(src), n, _ptr(dst), length, float(ripple)))
return _wrap(dst, idx, f"CHEBY1_{length}", "filters", offset)
_check(_lib.qtl_cheby1(_ptr(src), n, _ptr(dst), period, float(ripple)))
return _wrap(dst, idx, f"CHEBY1_{period}", "filters", offset)
def cheby2(close: object, length: int = 14, ripple: float = 0.5,
def cheby2(close: object, period: int = 14, ripple: float = 0.5,
offset: int = 0, **kwargs) -> object:
"""Chebyshev Type II."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_cheby2(_ptr(src), n, _ptr(dst), length, float(ripple)))
return _wrap(dst, idx, f"CHEBY2_{length}", "filters", offset)
_check(_lib.qtl_cheby2(_ptr(src), n, _ptr(dst), period, float(ripple)))
return _wrap(dst, idx, f"CHEBY2_{period}", "filters", offset)
def elliptic(close: object, length: int = 14, offset: int = 0, **kwargs) -> object:
def elliptic(close: object, period: int = 14, offset: int = 0, **kwargs) -> object:
"""Elliptic (Cauer) Filter."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_elliptic(_ptr(src), n, _ptr(dst), length))
return _wrap(dst, idx, f"ELLIPTIC_{length}", "filters", offset)
_check(_lib.qtl_elliptic(_ptr(src), n, _ptr(dst), period))
return _wrap(dst, idx, f"ELLIPTIC_{period}", "filters", offset)
def edcf(close: object, length: int = 14, offset: int = 0, **kwargs) -> object:
def edcf(close: object, period: int = 14, offset: int = 0, **kwargs) -> object:
"""Ehlers Distance Coefficient Filter."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_edcf(_ptr(src), n, _ptr(dst), length))
return _wrap(dst, idx, f"EDCF_{length}", "filters", offset)
_check(_lib.qtl_edcf(_ptr(src), n, _ptr(dst), period))
return _wrap(dst, idx, f"EDCF_{period}", "filters", offset)
def bpf(close: object, length: int = 14, bandwidth: int = 5,
def bpf(close: object, period: int = 14, bandwidth: int = 5,
offset: int = 0, **kwargs) -> object:
"""Bandpass Filter."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_bpf(_ptr(src), n, _ptr(dst), length, int(bandwidth)))
return _wrap(dst, idx, f"BPF_{length}", "filters", offset)
_check(_lib.qtl_bpf(_ptr(src), n, _ptr(dst), period, int(bandwidth)))
return _wrap(dst, idx, f"BPF_{period}", "filters", offset)
def alaguerre(close: object, length: int = 20, order: int = 5,
def alaguerre(close: object, period: int = 20, order: int = 5,
offset: int = 0, **kwargs) -> object:
"""Adaptive Laguerre Filter."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_alaguerre(_ptr(src), n, _ptr(dst), length, int(order)))
return _wrap(dst, idx, f"ALAGUERRE_{length}", "filters", offset)
_check(_lib.qtl_alaguerre(_ptr(src), n, _ptr(dst), period, int(order)))
return _wrap(dst, idx, f"ALAGUERRE_{period}", "filters", offset)
def bilateral(close: object, length: int = 14, sigma_s: float = 0.5,
def bilateral(close: object, period: int = 14, sigma_s: float = 0.5,
sigma_r: float = 1.0, offset: int = 0, **kwargs) -> object:
"""Bilateral Filter."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_bilateral(_ptr(src), n, _ptr(dst), length, float(sigma_s), float(sigma_r)))
return _wrap(dst, idx, f"BILATERAL_{length}", "filters", offset)
_check(_lib.qtl_bilateral(_ptr(src), n, _ptr(dst), period, float(sigma_s), float(sigma_r)))
return _wrap(dst, idx, f"BILATERAL_{period}", "filters", offset)
def baxterking(close: object, length: int = 12, min_period: int = 6,
def baxterking(close: object, period: int = 12, min_period: int = 6,
max_period: int = 32, offset: int = 0, **kwargs) -> object:
"""Baxter-King Filter."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_baxterking(_ptr(src), n, _ptr(dst), length, int(min_period), int(max_period)))
return _wrap(dst, idx, f"BAXTERKING_{length}", "filters", offset)
_check(_lib.qtl_baxterking(_ptr(src), n, _ptr(dst), period, int(min_period), int(max_period)))
return _wrap(dst, idx, f"BAXTERKING_{period}", "filters", offset)
def cfitz(close: object, length: int = 6, bw_period: int = 32,
def cfitz(close: object, period: int = 6, bw_period: int = 32,
offset: int = 0, **kwargs) -> object:
"""Christiano-Fitzgerald Filter."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_cfitz(_ptr(src), n, _ptr(dst), length, int(bw_period)))
return _wrap(dst, idx, f"CFITZ_{length}", "filters", offset)
_check(_lib.qtl_cfitz(_ptr(src), n, _ptr(dst), period, int(bw_period)))
return _wrap(dst, idx, f"CFITZ_{period}", "filters", offset)
python/quantalib/momentum.py
+24 -24
View File
@@ -145,36 +145,36 @@ def vel(close: object, period: int = 14, offset: int = 0, **kwargs) -> object:
_check(_lib.qtl_vel(_ptr(src), _ptr(output), n, period))
return _wrap(output, idx, f"VEL_{period}", "momentum", offset)
def rsi(close: object, length: int = 14, offset: int = 0, **kwargs) -> object:
def rsi(close: object, period: int = 14, offset: int = 0, **kwargs) -> object:
"""Relative Strength Index."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_rsi(_ptr(src), n, _ptr(dst), length))
return _wrap(dst, idx, f"RSI_{length}", "momentum", offset)
_check(_lib.qtl_rsi(_ptr(src), n, _ptr(dst), period))
return _wrap(dst, idx, f"RSI_{period}", "momentum", offset)
def roc(close: object, length: int = 10, offset: int = 0, **kwargs) -> object:
def roc(close: object, period: int = 10, offset: int = 0, **kwargs) -> object:
"""Rate of Change."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_roc(_ptr(src), n, _ptr(dst), length))
return _wrap(dst, idx, f"ROC_{length}", "momentum", offset)
_check(_lib.qtl_roc(_ptr(src), n, _ptr(dst), period))
return _wrap(dst, idx, f"ROC_{period}", "momentum", offset)
def mom(close: object, length: int = 10, offset: int = 0, **kwargs) -> object:
def mom(close: object, period: int = 10, offset: int = 0, **kwargs) -> object:
"""Momentum."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_mom(_ptr(src), n, _ptr(dst), length))
return _wrap(dst, idx, f"MOM_{length}", "momentum", offset)
_check(_lib.qtl_mom(_ptr(src), n, _ptr(dst), period))
return _wrap(dst, idx, f"MOM_{period}", "momentum", offset)
def cmo(close: object, length: int = 14, offset: int = 0, **kwargs) -> object:
def cmo(close: object, period: int = 14, offset: int = 0, **kwargs) -> object:
"""Chande Momentum Oscillator."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_cmo(_ptr(src), n, _ptr(dst), length))
return _wrap(dst, idx, f"CMO_{length}", "momentum", offset)
_check(_lib.qtl_cmo(_ptr(src), n, _ptr(dst), period))
return _wrap(dst, idx, f"CMO_{period}", "momentum", offset)
def tsi(close: object, long_period: int = 25, short_period: int = 13,
@@ -195,20 +195,20 @@ def apo(close: object, fast: int = 12, slow: int = 26,
return _wrap(dst, idx, f"APO_{fast}_{slow}", "momentum", offset)
def bias(close: object, length: int = 26, offset: int = 0, **kwargs) -> object:
def bias(close: object, period: int = 26, offset: int = 0, **kwargs) -> object:
"""Bias."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_bias(_ptr(src), n, _ptr(dst), length))
return _wrap(dst, idx, f"BIAS_{length}", "momentum", offset)
_check(_lib.qtl_bias(_ptr(src), n, _ptr(dst), period))
return _wrap(dst, idx, f"BIAS_{period}", "momentum", offset)
def cfo(close: object, length: int = 14, offset: int = 0, **kwargs) -> object:
def cfo(close: object, period: int = 14, offset: int = 0, **kwargs) -> object:
"""Chande Forecast Oscillator."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_cfo(_ptr(src), n, _ptr(dst), length))
return _wrap(dst, idx, f"CFO_{length}", "momentum", offset)
_check(_lib.qtl_cfo(_ptr(src), n, _ptr(dst), period))
return _wrap(dst, idx, f"CFO_{period}", "momentum", offset)
def cfb(close: object, lengths: list | None = None,
python/quantalib/numerics.py
+20 -20
View File
@@ -269,12 +269,12 @@ def weibulldist(close: object, k: float = 1.5, lam: float = 1.0, period: int = 1
_check(_lib.qtl_weibulldist(_ptr(src), _ptr(output), n, k, lam, period))
return _wrap(output, idx, f"WEIBULLDIST_{period}", "numerics", offset)
def change(close: object, length: int = 1, offset: int = 0, **kwargs) -> object:
def change(close: object, period: int = 1, offset: int = 0, **kwargs) -> object:
"""Price Change."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_change(_ptr(src), n, _ptr(dst), length))
return _wrap(dst, idx, f"CHANGE_{length}", "numerics", offset)
_check(_lib.qtl_change(_ptr(src), n, _ptr(dst), period))
return _wrap(dst, idx, f"CHANGE_{period}", "numerics", offset)
def exptrans(close: object, offset: int = 0, **kwargs) -> object:
@@ -285,31 +285,31 @@ def exptrans(close: object, offset: int = 0, **kwargs) -> object:
return _wrap(dst, idx, "EXPTRANS", "numerics", offset)
def betadist(close: object, length: int = 50, alpha: float = 2.0,
def betadist(close: object, period: int = 50, alpha: float = 2.0,
beta: float = 2.0, offset: int = 0, **kwargs) -> object:
"""Beta Distribution."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_betadist(_ptr(src), n, _ptr(dst), length, float(alpha), float(beta)))
return _wrap(dst, idx, f"BETADIST_{length}", "numerics", offset)
_check(_lib.qtl_betadist(_ptr(src), n, _ptr(dst), period, float(alpha), float(beta)))
return _wrap(dst, idx, f"BETADIST_{period}", "numerics", offset)
def expdist(close: object, length: int = 50, lam: float = 3.0,
def expdist(close: object, period: int = 50, lam: float = 3.0,
offset: int = 0, **kwargs) -> object:
"""Exponential Distribution."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_expdist(_ptr(src), n, _ptr(dst), length, float(lam)))
return _wrap(dst, idx, f"EXPDIST_{length}", "numerics", offset)
_check(_lib.qtl_expdist(_ptr(src), n, _ptr(dst), period, float(lam)))
return _wrap(dst, idx, f"EXPDIST_{period}", "numerics", offset)
def binomdist(close: object, length: int = 50, trials: int = 20,
def binomdist(close: object, period: int = 50, trials: int = 20,
threshold: int = 10, offset: int = 0, **kwargs) -> object:
"""Binomial Distribution."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_binomdist(_ptr(src), n, _ptr(dst), length, int(trials), int(threshold)))
return _wrap(dst, idx, f"BINOMDIST_{length}", "numerics", offset)
_check(_lib.qtl_binomdist(_ptr(src), n, _ptr(dst), period, int(trials), int(threshold)))
return _wrap(dst, idx, f"BINOMDIST_{period}", "numerics", offset)
def cwt(close: object, scale: float = 10.0, omega: float = 6.0,
@@ -321,10 +321,10 @@ def cwt(close: object, scale: float = 10.0, omega: float = 6.0,
return _wrap(dst, idx, "CWT", "numerics", offset)
def dwt(close: object, length: int = 4, levels: int = 0,
def dwt(close: object, period: int = 4, levels: int = 0,
offset: int = 0, **kwargs) -> object:
"""Discrete Wavelet Transform."""
length = int(length); levels = int(levels); offset = int(offset)
period = int(kwargs.get("length", period)); levels = int(levels); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_dwt(_ptr(src), n, _ptr(dst), length, levels))
return _wrap(dst, idx, f"DWT_{length}", "numerics", offset)
_check(_lib.qtl_dwt(_ptr(src), n, _ptr(dst), period, levels))
return _wrap(dst, idx, f"DWT_{period}", "numerics", offset)
python/quantalib/oscillators.py
+59 -59
View File
@@ -156,9 +156,9 @@ def imi(open: object, high: object, low: object, close: object, volume: object,
return _wrap(dst, idx, f"IMI_{period}", "oscillators", offset)
def kdj(high: object, low: object, close: object, length: int = 14, signal: int = 3, offset: int = 0, **kwargs) -> object:
def kdj(high: object, low: object, close: object, period: int = 14, signal: int = 3, offset: int = 0, **kwargs) -> object:
"""KDJ Indicator."""
length = int(length)
period = int(kwargs.get("length", period))
signal = int(signal)
offset = int(offset)
h, idx = _arr(high); l, _ = _arr(low); c, _ = _arr(close)
@@ -166,7 +166,7 @@ def kdj(high: object, low: object, close: object, length: int = 14, signal: int
kOut = _out(n)
dOut = _out(n)
jOut = _out(n)
_check(_lib.qtl_kdj(_ptr(h), _ptr(l), _ptr(c), _ptr(kOut), _ptr(dOut), _ptr(jOut), n, length, signal))
_check(_lib.qtl_kdj(_ptr(h), _ptr(l), _ptr(c), _ptr(kOut), _ptr(dOut), _ptr(jOut), n, period, signal))
return _wrap_multi({"kOut": kOut, "dOut": dOut, "jOut": jOut}, idx, "oscillators", offset)
@@ -377,100 +377,100 @@ def willr(high: object, low: object, close: object, period: int = 14, offset: in
_check(_lib.qtl_willr(_ptr(h), _ptr(l), _ptr(c), _ptr(output), n, period))
return _wrap(output, idx, f"WILLR_{period}", "oscillators", offset)
def fisher(close: object, length: int = 9, offset: int = 0, **kwargs) -> object:
def fisher(close: object, period: int = 9, offset: int = 0, **kwargs) -> object:
"""Fisher Transform."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_fisher(_ptr(src), n, _ptr(dst), length))
return _wrap(dst, idx, f"FISHER_{length}", "oscillators", offset)
_check(_lib.qtl_fisher(_ptr(src), n, _ptr(dst), period))
return _wrap(dst, idx, f"FISHER_{period}", "oscillators", offset)
def fisher04(close: object, length: int = 9, offset: int = 0, **kwargs) -> object:
def fisher04(close: object, period: int = 9, offset: int = 0, **kwargs) -> object:
"""Fisher Transform (0.4 variant)."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_fisher04(_ptr(src), n, _ptr(dst), length))
return _wrap(dst, idx, f"FISHER04_{length}", "oscillators", offset)
_check(_lib.qtl_fisher04(_ptr(src), n, _ptr(dst), period))
return _wrap(dst, idx, f"FISHER04_{period}", "oscillators", offset)
def dpo(close: object, length: int = 20, offset: int = 0, **kwargs) -> object:
def dpo(close: object, period: int = 20, offset: int = 0, **kwargs) -> object:
"""Detrended Price Oscillator."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_dpo(_ptr(src), n, _ptr(dst), length))
return _wrap(dst, idx, f"DPO_{length}", "oscillators", offset)
_check(_lib.qtl_dpo(_ptr(src), n, _ptr(dst), period))
return _wrap(dst, idx, f"DPO_{period}", "oscillators", offset)
def trix(close: object, length: int = 18, offset: int = 0, **kwargs) -> object:
def trix(close: object, period: int = 18, offset: int = 0, **kwargs) -> object:
"""Triple EMA Rate of Change."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_trix(_ptr(src), n, _ptr(dst), length))
return _wrap(dst, idx, f"TRIX_{length}", "oscillators", offset)
_check(_lib.qtl_trix(_ptr(src), n, _ptr(dst), period))
return _wrap(dst, idx, f"TRIX_{period}", "oscillators", offset)
def inertia(close: object, length: int = 20, offset: int = 0, **kwargs) -> object:
def inertia(close: object, period: int = 20, offset: int = 0, **kwargs) -> object:
"""Inertia."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_inertia(_ptr(src), n, _ptr(dst), length))
return _wrap(dst, idx, f"INERTIA_{length}", "oscillators", offset)
_check(_lib.qtl_inertia(_ptr(src), n, _ptr(dst), period))
return _wrap(dst, idx, f"INERTIA_{period}", "oscillators", offset)
def rsx(close: object, length: int = 14, offset: int = 0, **kwargs) -> object:
def rsx(close: object, period: int = 14, offset: int = 0, **kwargs) -> object:
"""Relative Strength Xtra."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_rsx(_ptr(src), n, _ptr(dst), length))
return _wrap(dst, idx, f"RSX_{length}", "oscillators", offset)
_check(_lib.qtl_rsx(_ptr(src), n, _ptr(dst), period))
return _wrap(dst, idx, f"RSX_{period}", "oscillators", offset)
def er(close: object, length: int = 10, offset: int = 0, **kwargs) -> object:
def er(close: object, period: int = 10, offset: int = 0, **kwargs) -> object:
"""Efficiency Ratio."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_er(_ptr(src), n, _ptr(dst), length))
return _wrap(dst, idx, f"ER_{length}", "oscillators", offset)
_check(_lib.qtl_er(_ptr(src), n, _ptr(dst), period))
return _wrap(dst, idx, f"ER_{period}", "oscillators", offset)
def cti(close: object, length: int = 12, offset: int = 0, **kwargs) -> object:
def cti(close: object, period: int = 12, offset: int = 0, **kwargs) -> object:
"""Correlation Trend Indicator."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_cti(_ptr(src), n, _ptr(dst), length))
return _wrap(dst, idx, f"CTI_{length}", "oscillators", offset)
_check(_lib.qtl_cti(_ptr(src), n, _ptr(dst), period))
return _wrap(dst, idx, f"CTI_{period}", "oscillators", offset)
def reflex(close: object, length: int = 20, offset: int = 0, **kwargs) -> object:
def reflex(close: object, period: int = 20, offset: int = 0, **kwargs) -> object:
"""Reflex."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_reflex(_ptr(src), n, _ptr(dst), length))
return _wrap(dst, idx, f"REFLEX_{length}", "oscillators", offset)
_check(_lib.qtl_reflex(_ptr(src), n, _ptr(dst), period))
return _wrap(dst, idx, f"REFLEX_{period}", "oscillators", offset)
def trendflex(close: object, length: int = 20, offset: int = 0, **kwargs) -> object:
def trendflex(close: object, period: int = 20, offset: int = 0, **kwargs) -> object:
"""Trendflex."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_trendflex(_ptr(src), n, _ptr(dst), length))
return _wrap(dst, idx, f"TRENDFLEX_{length}", "oscillators", offset)
_check(_lib.qtl_trendflex(_ptr(src), n, _ptr(dst), period))
return _wrap(dst, idx, f"TRENDFLEX_{period}", "oscillators", offset)
def kri(close: object, length: int = 20, offset: int = 0, **kwargs) -> object:
def kri(close: object, period: int = 20, offset: int = 0, **kwargs) -> object:
"""Kairi Relative Index."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_kri(_ptr(src), n, _ptr(dst), length))
return _wrap(dst, idx, f"KRI_{length}", "oscillators", offset)
_check(_lib.qtl_kri(_ptr(src), n, _ptr(dst), period))
return _wrap(dst, idx, f"KRI_{period}", "oscillators", offset)
def psl(close: object, length: int = 12, offset: int = 0, **kwargs) -> object:
def psl(close: object, period: int = 12, offset: int = 0, **kwargs) -> object:
"""Psychological Line."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_psl(_ptr(src), n, _ptr(dst), length))
return _wrap(dst, idx, f"PSL_{length}", "oscillators", offset)
_check(_lib.qtl_psl(_ptr(src), n, _ptr(dst), period))
return _wrap(dst, idx, f"PSL_{period}", "oscillators", offset)
def deco(close: object, short_period: int = 30, long_period: int = 60,
@@ -515,13 +515,13 @@ def crsi(close: object, rsi_period: int = 3, streak_period: int = 2,
return _wrap(dst, idx, f"CRSI_{rsi_period}", "oscillators", offset)
def bbb(close: object, length: int = 20, mult: float = 2.0,
def bbb(close: object, period: int = 20, mult: float = 2.0,
offset: int = 0, **kwargs) -> object:
"""Bollinger Band Bounce."""
length = int(length); mult = float(mult); offset = int(offset)
period = int(kwargs.get("length", period)); mult = float(mult); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_bbb(_ptr(src), n, _ptr(dst), length, mult))
return _wrap(dst, idx, f"BBB_{length}", "oscillators", offset)
_check(_lib.qtl_bbb(_ptr(src), n, _ptr(dst), period, mult))
return _wrap(dst, idx, f"BBB_{period}", "oscillators", offset)
def bbi(close: object, p1: int = 3, p2: int = 6, p3: int = 12, p4: int = 24,
@@ -533,14 +533,14 @@ def bbi(close: object, p1: int = 3, p2: int = 6, p3: int = 12, p4: int = 24,
return _wrap(dst, idx, "BBI", "oscillators", offset)
def dem(high: object, low: object, length: int = 14,
def dem(high: object, low: object, period: int = 14,
offset: int = 0, **kwargs) -> object:
"""DeMarker."""
length = int(length)
period = int(kwargs.get("length", period))
h, idx = _arr(high); l, _ = _arr(low)
n = len(h); dst = _out(n)
_check(_lib.qtl_dem(_ptr(h), _ptr(l), n, _ptr(dst), length))
return _wrap(dst, idx, f"DEM_{length}", "oscillators", int(offset))
_check(_lib.qtl_dem(_ptr(h), _ptr(l), n, _ptr(dst), period))
return _wrap(dst, idx, f"DEM_{period}", "oscillators", int(offset))
def brar(open: object, high: object, low: object, close: object,
python/quantalib/statistics.py
+20 -20
View File
@@ -340,12 +340,12 @@ def ztest(close: object, period: int = 14, mu0: float = 0.0, offset: int = 0, **
_check(_lib.qtl_ztest(_ptr(src), _ptr(output), n, period, mu0))
return _wrap(output, idx, f"ZTEST_{period}", "statistics", offset)
def zscore(close: object, length: int = 20, offset: int = 0, **kwargs) -> object:
def zscore(close: object, period: int = 20, offset: int = 0, **kwargs) -> object:
"""Z-Score."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_zscore(_ptr(src), n, _ptr(dst), length))
return _wrap(dst, idx, f"ZSCORE_{length}", "statistics", offset)
_check(_lib.qtl_zscore(_ptr(src), n, _ptr(dst), period))
return _wrap(dst, idx, f"ZSCORE_{period}", "statistics", offset)
def cma(close: object, offset: int = 0, **kwargs) -> object:
@@ -356,39 +356,39 @@ def cma(close: object, offset: int = 0, **kwargs) -> object:
return _wrap(dst, idx, "CMA", "statistics", offset)
def entropy(close: object, length: int = 10, offset: int = 0, **kwargs) -> object:
def entropy(close: object, period: int = 10, offset: int = 0, **kwargs) -> object:
"""Shannon Entropy."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_entropy(_ptr(src), n, _ptr(dst), length))
return _wrap(dst, idx, f"ENTROPY_{length}", "statistics", offset)
_check(_lib.qtl_entropy(_ptr(src), n, _ptr(dst), period))
return _wrap(dst, idx, f"ENTROPY_{period}", "statistics", offset)
def correlation(x: object, y: object, length: int = 20,
def correlation(x: object, y: object, period: int = 20,
offset: int = 0, **kwargs) -> object:
"""Pearson Correlation."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
xarr, idx = _arr(x); yarr, _ = _arr(y)
n = len(xarr); dst = _out(n)
_check(_lib.qtl_correlation(_ptr(xarr), _ptr(yarr), n, _ptr(dst), length))
return _wrap(dst, idx, f"CORR_{length}", "statistics", offset)
_check(_lib.qtl_correlation(_ptr(xarr), _ptr(yarr), n, _ptr(dst), period))
return _wrap(dst, idx, f"CORR_{period}", "statistics", offset)
def covariance(x: object, y: object, length: int = 20,
def covariance(x: object, y: object, period: int = 20,
is_sample: bool = True, offset: int = 0, **kwargs) -> object:
"""Covariance."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
xarr, idx = _arr(x); yarr, _ = _arr(y)
n = len(xarr); dst = _out(n)
_check(_lib.qtl_covariance(_ptr(xarr), _ptr(yarr), n, _ptr(dst), length, int(is_sample)))
return _wrap(dst, idx, f"COV_{length}", "statistics", offset)
_check(_lib.qtl_covariance(_ptr(xarr), _ptr(yarr), n, _ptr(dst), period, int(is_sample)))
return _wrap(dst, idx, f"COV_{period}", "statistics", offset)
def cointegration(x: object, y: object, length: int = 20,
def cointegration(x: object, y: object, period: int = 20,
offset: int = 0, **kwargs) -> object:
"""Cointegration."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
xarr, idx = _arr(x); yarr, _ = _arr(y)
n = len(xarr); dst = _out(n)
_check(_lib.qtl_cointegration(_ptr(xarr), _ptr(yarr), n, _ptr(dst), length))
return _wrap(dst, idx, f"COINT_{length}", "statistics", offset)
_check(_lib.qtl_cointegration(_ptr(xarr), _ptr(yarr), n, _ptr(dst), period))
return _wrap(dst, idx, f"COINT_{period}", "statistics", offset)
python/quantalib/trends_fir.py
+80 -80
View File
@@ -195,117 +195,117 @@ def rwma(high: object, low: object, close: object, period: int = 14, offset: int
_check(_lib.qtl_rwma(_ptr(c), _ptr(h), _ptr(l), _ptr(output), n, period))
return _wrap(output, idx, f"RWMA_{period}", "trends_fir", offset)
def sma(close: object, length: int = 10, offset: int = 0, **kwargs) -> object:
def sma(close: object, period: int = 10, offset: int = 0, **kwargs) -> object:
"""Simple Moving Average."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_sma(_ptr(src), n, _ptr(dst), length))
return _wrap(dst, idx, f"SMA_{length}", "trends_fir", offset)
_check(_lib.qtl_sma(_ptr(src), n, _ptr(dst), period))
return _wrap(dst, idx, f"SMA_{period}", "trends_fir", offset)
def wma(close: object, length: int = 10, offset: int = 0, **kwargs) -> object:
def wma(close: object, period: int = 10, offset: int = 0, **kwargs) -> object:
"""Weighted Moving Average."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_wma(_ptr(src), n, _ptr(dst), length))
return _wrap(dst, idx, f"WMA_{length}", "trends_fir", offset)
_check(_lib.qtl_wma(_ptr(src), n, _ptr(dst), period))
return _wrap(dst, idx, f"WMA_{period}", "trends_fir", offset)
def hma(close: object, length: int = 9, offset: int = 0, **kwargs) -> object:
def hma(close: object, period: int = 9, offset: int = 0, **kwargs) -> object:
"""Hull Moving Average."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_hma(_ptr(src), n, _ptr(dst), length))
return _wrap(dst, idx, f"HMA_{length}", "trends_fir", offset)
_check(_lib.qtl_hma(_ptr(src), n, _ptr(dst), period))
return _wrap(dst, idx, f"HMA_{period}", "trends_fir", offset)
def trima(close: object, length: int = 10, offset: int = 0, **kwargs) -> object:
def trima(close: object, period: int = 10, offset: int = 0, **kwargs) -> object:
"""Triangular Moving Average."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_trima(_ptr(src), n, _ptr(dst), length))
return _wrap(dst, idx, f"TRIMA_{length}", "trends_fir", offset)
_check(_lib.qtl_trima(_ptr(src), n, _ptr(dst), period))
return _wrap(dst, idx, f"TRIMA_{period}", "trends_fir", offset)
def swma(close: object, length: int = 10, offset: int = 0, **kwargs) -> object:
def swma(close: object, period: int = 10, offset: int = 0, **kwargs) -> object:
"""Symmetric Weighted Moving Average."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_swma(_ptr(src), n, _ptr(dst), length))
return _wrap(dst, idx, f"SWMA_{length}", "trends_fir", offset)
_check(_lib.qtl_swma(_ptr(src), n, _ptr(dst), period))
return _wrap(dst, idx, f"SWMA_{period}", "trends_fir", offset)
def dwma(close: object, length: int = 10, offset: int = 0, **kwargs) -> object:
def dwma(close: object, period: int = 10, offset: int = 0, **kwargs) -> object:
"""Double Weighted Moving Average."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_dwma(_ptr(src), n, _ptr(dst), length))
return _wrap(dst, idx, f"DWMA_{length}", "trends_fir", offset)
_check(_lib.qtl_dwma(_ptr(src), n, _ptr(dst), period))
return _wrap(dst, idx, f"DWMA_{period}", "trends_fir", offset)
def blma(close: object, length: int = 10, offset: int = 0, **kwargs) -> object:
def blma(close: object, period: int = 10, offset: int = 0, **kwargs) -> object:
"""Blackman Moving Average."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_blma(_ptr(src), n, _ptr(dst), length))
return _wrap(dst, idx, f"BLMA_{length}", "trends_fir", offset)
_check(_lib.qtl_blma(_ptr(src), n, _ptr(dst), period))
return _wrap(dst, idx, f"BLMA_{period}", "trends_fir", offset)
def alma(close: object, length: int = 10, alma_offset: float = 0.85,
def alma(close: object, period: int = 10, alma_offset: float = 0.85,
sigma: float = 6.0, offset: int = 0, **kwargs) -> object:
"""Arnaud Legoux Moving Average."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_alma(_ptr(src), n, _ptr(dst), length, float(alma_offset), float(sigma)))
return _wrap(dst, idx, f"ALMA_{length}", "trends_fir", offset)
_check(_lib.qtl_alma(_ptr(src), n, _ptr(dst), period, float(alma_offset), float(sigma)))
return _wrap(dst, idx, f"ALMA_{period}", "trends_fir", offset)
def lsma(close: object, length: int = 25, offset: int = 0, **kwargs) -> object:
def lsma(close: object, period: int = 25, offset: int = 0, **kwargs) -> object:
"""Least Squares Moving Average."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_lsma(_ptr(src), n, _ptr(dst), length, 0, 1.0))
return _wrap(dst, idx, f"LSMA_{length}", "trends_fir", offset)
_check(_lib.qtl_lsma(_ptr(src), n, _ptr(dst), period, 0, 1.0))
return _wrap(dst, idx, f"LSMA_{period}", "trends_fir", offset)
def sgma(close: object, length: int = 10, offset: int = 0, **kwargs) -> object:
def sgma(close: object, period: int = 10, offset: int = 0, **kwargs) -> object:
"""Savitzky-Golay Moving Average."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_sgma(_ptr(src), n, _ptr(dst), length))
return _wrap(dst, idx, f"SGMA_{length}", "trends_fir", offset)
_check(_lib.qtl_sgma(_ptr(src), n, _ptr(dst), period))
return _wrap(dst, idx, f"SGMA_{period}", "trends_fir", offset)
def sinema(close: object, length: int = 10, offset: int = 0, **kwargs) -> object:
def sinema(close: object, period: int = 10, offset: int = 0, **kwargs) -> object:
"""Sine-weighted Moving Average."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_sinema(_ptr(src), n, _ptr(dst), length))
return _wrap(dst, idx, f"SINEMA_{length}", "trends_fir", offset)
_check(_lib.qtl_sinema(_ptr(src), n, _ptr(dst), period))
return _wrap(dst, idx, f"SINEMA_{period}", "trends_fir", offset)
def hanma(close: object, length: int = 10, offset: int = 0, **kwargs) -> object:
def hanma(close: object, period: int = 10, offset: int = 0, **kwargs) -> object:
"""Hann-weighted Moving Average."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_hanma(_ptr(src), n, _ptr(dst), length))
return _wrap(dst, idx, f"HANMA_{length}", "trends_fir", offset)
_check(_lib.qtl_hanma(_ptr(src), n, _ptr(dst), period))
return _wrap(dst, idx, f"HANMA_{period}", "trends_fir", offset)
def parzen(close: object, length: int = 10, offset: int = 0, **kwargs) -> object:
def parzen(close: object, period: int = 10, offset: int = 0, **kwargs) -> object:
"""Parzen-weighted Moving Average."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_parzen(_ptr(src), n, _ptr(dst), length))
return _wrap(dst, idx, f"PARZEN_{length}", "trends_fir", offset)
_check(_lib.qtl_parzen(_ptr(src), n, _ptr(dst), period))
return _wrap(dst, idx, f"PARZEN_{period}", "trends_fir", offset)
def tsf(close: object, length: int = 14, offset: int = 0, **kwargs) -> object:
def tsf(close: object, period: int = 14, offset: int = 0, **kwargs) -> object:
"""Time Series Forecast."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_tsf(_ptr(src), n, _ptr(dst), length))
return _wrap(dst, idx, f"TSF_{length}", "trends_fir", offset)
_check(_lib.qtl_tsf(_ptr(src), n, _ptr(dst), period))
return _wrap(dst, idx, f"TSF_{period}", "trends_fir", offset)
def conv(close: object, kernel: list | None = None,
@@ -320,52 +320,52 @@ def conv(close: object, kernel: list | None = None,
return _wrap(dst, idx, "CONV", "trends_fir", offset)
def bwma(close: object, length: int = 10, order: int = 0,
def bwma(close: object, period: int = 10, order: int = 0,
offset: int = 0, **kwargs) -> object:
"""Butterworth-weighted Moving Average."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_bwma(_ptr(src), n, _ptr(dst), length, int(order)))
return _wrap(dst, idx, f"BWMA_{length}", "trends_fir", offset)
_check(_lib.qtl_bwma(_ptr(src), n, _ptr(dst), period, int(order)))
return _wrap(dst, idx, f"BWMA_{period}", "trends_fir", offset)
def crma(close: object, length: int = 10, volume_factor: float = 1.0,
def crma(close: object, period: int = 10, volume_factor: float = 1.0,
offset: int = 0, **kwargs) -> object:
"""Cosine-Ramp Moving Average."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_crma(_ptr(src), n, _ptr(dst), length, float(volume_factor)))
return _wrap(dst, idx, f"CRMA_{length}", "trends_fir", offset)
_check(_lib.qtl_crma(_ptr(src), n, _ptr(dst), period, float(volume_factor)))
return _wrap(dst, idx, f"CRMA_{period}", "trends_fir", offset)
def sp15(close: object, length: int = 15, offset: int = 0, **kwargs) -> object:
def sp15(close: object, period: int = 15, offset: int = 0, **kwargs) -> object:
"""SP-15 Moving Average."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_sp15(_ptr(src), n, _ptr(dst), length))
return _wrap(dst, idx, f"SP15_{length}", "trends_fir", offset)
_check(_lib.qtl_sp15(_ptr(src), n, _ptr(dst), period))
return _wrap(dst, idx, f"SP15_{period}", "trends_fir", offset)
def tukey_w(close: object, length: int = 10, offset: int = 0, **kwargs) -> object:
def tukey_w(close: object, period: int = 10, offset: int = 0, **kwargs) -> object:
"""Tukey-windowed Moving Average."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_tukey_w(_ptr(src), n, _ptr(dst), length))
return _wrap(dst, idx, f"TUKEY_{length}", "trends_fir", offset)
_check(_lib.qtl_tukey_w(_ptr(src), n, _ptr(dst), period))
return _wrap(dst, idx, f"TUKEY_{period}", "trends_fir", offset)
def rain(close: object, length: int = 10, offset: int = 0, **kwargs) -> object:
def rain(close: object, period: int = 10, offset: int = 0, **kwargs) -> object:
"""RAIN Moving Average."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_rain(_ptr(src), n, _ptr(dst), length))
return _wrap(dst, idx, f"RAIN_{length}", "trends_fir", offset)
_check(_lib.qtl_rain(_ptr(src), n, _ptr(dst), period))
return _wrap(dst, idx, f"RAIN_{period}", "trends_fir", offset)
def afirma(close: object, length: int = 10, window_type: int = 0,
def afirma(close: object, period: int = 10, window_type: int = 0,
use_simd: bool = False, offset: int = 0, **kwargs) -> object:
"""Adaptive FIR Moving Average."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_afirma(_ptr(src), n, _ptr(dst), length, int(window_type), int(use_simd)))
return _wrap(dst, idx, f"AFIRMA_{length}", "trends_fir", offset)
_check(_lib.qtl_afirma(_ptr(src), n, _ptr(dst), period, int(window_type), int(use_simd)))
return _wrap(dst, idx, f"AFIRMA_{period}", "trends_fir", offset)
python/quantalib/trends_iir.py
+40 -40
View File
@@ -357,12 +357,12 @@ def zltema(close: object, period: int = 14, offset: int = 0, **kwargs) -> object
_check(_lib.qtl_zltema(_ptr(src), _ptr(output), n, period))
return _wrap(output, idx, f"ZLTEMA_{period}", "trends_iir", offset)
def ema(close: object, length: int = 10, offset: int = 0, **kwargs) -> object:
def ema(close: object, period: int = 10, offset: int = 0, **kwargs) -> object:
"""Exponential Moving Average."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_ema(_ptr(src), n, _ptr(dst), length))
return _wrap(dst, idx, f"EMA_{length}", "trends_iir", offset)
_check(_lib.qtl_ema(_ptr(src), n, _ptr(dst), period))
return _wrap(dst, idx, f"EMA_{period}", "trends_iir", offset)
def ema_alpha(close: object, alpha: float = 0.1, offset: int = 0, **kwargs) -> object:
@@ -373,12 +373,12 @@ def ema_alpha(close: object, alpha: float = 0.1, offset: int = 0, **kwargs) -> o
return _wrap(dst, idx, f"EMA_a{alpha:.4f}", "trends_iir", offset)
def dema(close: object, length: int = 10, offset: int = 0, **kwargs) -> object:
def dema(close: object, period: int = 10, offset: int = 0, **kwargs) -> object:
"""Double Exponential Moving Average."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_dema(_ptr(src), n, _ptr(dst), length))
return _wrap(dst, idx, f"DEMA_{length}", "trends_iir", offset)
_check(_lib.qtl_dema(_ptr(src), n, _ptr(dst), period))
return _wrap(dst, idx, f"DEMA_{period}", "trends_iir", offset)
def dema_alpha(close: object, alpha: float = 0.1, offset: int = 0, **kwargs) -> object:
@@ -389,71 +389,71 @@ def dema_alpha(close: object, alpha: float = 0.1, offset: int = 0, **kwargs) ->
return _wrap(dst, idx, f"DEMA_a{alpha:.4f}", "trends_iir", offset)
def tema(close: object, length: int = 10, offset: int = 0, **kwargs) -> object:
def tema(close: object, period: int = 10, offset: int = 0, **kwargs) -> object:
"""Triple Exponential Moving Average."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_tema(_ptr(src), n, _ptr(dst), length))
return _wrap(dst, idx, f"TEMA_{length}", "trends_iir", offset)
_check(_lib.qtl_tema(_ptr(src), n, _ptr(dst), period))
return _wrap(dst, idx, f"TEMA_{period}", "trends_iir", offset)
def lema(close: object, length: int = 10, offset: int = 0, **kwargs) -> object:
def lema(close: object, period: int = 10, offset: int = 0, **kwargs) -> object:
"""Laguerre-based EMA."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_lema(_ptr(src), n, _ptr(dst), length))
return _wrap(dst, idx, f"LEMA_{length}", "trends_iir", offset)
_check(_lib.qtl_lema(_ptr(src), n, _ptr(dst), period))
return _wrap(dst, idx, f"LEMA_{period}", "trends_iir", offset)
def hema(close: object, length: int = 10, offset: int = 0, **kwargs) -> object:
def hema(close: object, period: int = 10, offset: int = 0, **kwargs) -> object:
"""Henderson EMA."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_hema(_ptr(src), n, _ptr(dst), length))
return _wrap(dst, idx, f"HEMA_{length}", "trends_iir", offset)
_check(_lib.qtl_hema(_ptr(src), n, _ptr(dst), period))
return _wrap(dst, idx, f"HEMA_{period}", "trends_iir", offset)
def ahrens(close: object, length: int = 10, offset: int = 0, **kwargs) -> object:
def ahrens(close: object, period: int = 10, offset: int = 0, **kwargs) -> object:
"""Ahrens Moving Average."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_ahrens(_ptr(src), n, _ptr(dst), length))
return _wrap(dst, idx, f"AHRENS_{length}", "trends_iir", offset)
_check(_lib.qtl_ahrens(_ptr(src), n, _ptr(dst), period))
return _wrap(dst, idx, f"AHRENS_{period}", "trends_iir", offset)
def decycler(close: object, length: int = 20, offset: int = 0, **kwargs) -> object:
def decycler(close: object, period: int = 20, offset: int = 0, **kwargs) -> object:
"""Simple Decycler."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_decycler(_ptr(src), n, _ptr(dst), length))
return _wrap(dst, idx, f"DECYCLER_{length}", "trends_iir", offset)
_check(_lib.qtl_decycler(_ptr(src), n, _ptr(dst), period))
return _wrap(dst, idx, f"DECYCLER_{period}", "trends_iir", offset)
def dsma(close: object, length: int = 10, factor: float = 0.5,
def dsma(close: object, period: int = 10, factor: float = 0.5,
offset: int = 0, **kwargs) -> object:
"""Deviation-Scaled Moving Average."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_dsma(_ptr(src), n, _ptr(dst), length, float(factor)))
return _wrap(dst, idx, f"DSMA_{length}", "trends_iir", offset)
_check(_lib.qtl_dsma(_ptr(src), n, _ptr(dst), period, float(factor)))
return _wrap(dst, idx, f"DSMA_{period}", "trends_iir", offset)
def gdema(close: object, length: int = 10, vfactor: float = 1.0,
def gdema(close: object, period: int = 10, vfactor: float = 1.0,
offset: int = 0, **kwargs) -> object:
"""Generalized DEMA."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_gdema(_ptr(src), n, _ptr(dst), length, float(vfactor)))
return _wrap(dst, idx, f"GDEMA_{length}", "trends_iir", offset)
_check(_lib.qtl_gdema(_ptr(src), n, _ptr(dst), period, float(vfactor)))
return _wrap(dst, idx, f"GDEMA_{period}", "trends_iir", offset)
def coral(close: object, length: int = 10, friction: float = 0.4,
def coral(close: object, period: int = 10, friction: float = 0.4,
offset: int = 0, **kwargs) -> object:
"""CORAL Trend."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_coral(_ptr(src), n, _ptr(dst), length, float(friction)))
return _wrap(dst, idx, f"CORAL_{length}", "trends_iir", offset)
_check(_lib.qtl_coral(_ptr(src), n, _ptr(dst), period, float(friction)))
return _wrap(dst, idx, f"CORAL_{period}", "trends_iir", offset)
def agc(close: object, alpha: float = 0.1, offset: int = 0, **kwargs) -> object:
python/quantalib/volatility.py
+32 -32
View File
@@ -252,57 +252,57 @@ def tr(high: object, low: object, close: object, offset: int = 0, **kwargs) -> o
return _wrap(dst, idx, "TR", "volatility", int(offset))
def bbw(close: object, length: int = 20, mult: float = 2.0,
def bbw(close: object, period: int = 20, mult: float = 2.0,
offset: int = 0, **kwargs) -> object:
"""Bollinger Band Width."""
length = int(length); mult = float(mult); offset = int(offset)
period = int(kwargs.get("length", period)); mult = float(mult); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_bbw(_ptr(src), n, _ptr(dst), length, mult))
return _wrap(dst, idx, f"BBW_{length}", "volatility", offset)
_check(_lib.qtl_bbw(_ptr(src), n, _ptr(dst), period, mult))
return _wrap(dst, idx, f"BBW_{period}", "volatility", offset)
def bbwn(close: object, length: int = 20, mult: float = 2.0,
def bbwn(close: object, period: int = 20, mult: float = 2.0,
lookback: int = 252, offset: int = 0, **kwargs) -> object:
"""Bollinger Band Width Normalized."""
length = int(length); mult = float(mult); lookback = int(lookback); offset = int(offset)
period = int(kwargs.get("length", period)); mult = float(mult); lookback = int(lookback); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_bbwn(_ptr(src), n, _ptr(dst), length, mult, lookback))
return _wrap(dst, idx, f"BBWN_{length}", "volatility", offset)
_check(_lib.qtl_bbwn(_ptr(src), n, _ptr(dst), period, mult, lookback))
return _wrap(dst, idx, f"BBWN_{period}", "volatility", offset)
def bbwp(close: object, length: int = 20, mult: float = 2.0,
def bbwp(close: object, period: int = 20, mult: float = 2.0,
lookback: int = 252, offset: int = 0, **kwargs) -> object:
"""Bollinger Band Width Percentile."""
length = int(length); mult = float(mult); lookback = int(lookback); offset = int(offset)
period = int(kwargs.get("length", period)); mult = float(mult); lookback = int(lookback); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_bbwp(_ptr(src), n, _ptr(dst), length, mult, lookback))
return _wrap(dst, idx, f"BBWP_{length}", "volatility", offset)
_check(_lib.qtl_bbwp(_ptr(src), n, _ptr(dst), period, mult, lookback))
return _wrap(dst, idx, f"BBWP_{period}", "volatility", offset)
def stddev(close: object, length: int = 20, offset: int = 0, **kwargs) -> object:
def stddev(close: object, period: int = 20, offset: int = 0, **kwargs) -> object:
"""Standard Deviation."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_stddev(_ptr(src), n, _ptr(dst), length))
return _wrap(dst, idx, f"STDDEV_{length}", "volatility", offset)
_check(_lib.qtl_stddev(_ptr(src), n, _ptr(dst), period))
return _wrap(dst, idx, f"STDDEV_{period}", "volatility", offset)
def variance(close: object, length: int = 20, offset: int = 0, **kwargs) -> object:
def variance(close: object, period: int = 20, offset: int = 0, **kwargs) -> object:
"""Variance."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_variance(_ptr(src), n, _ptr(dst), length))
return _wrap(dst, idx, f"VAR_{length}", "volatility", offset)
_check(_lib.qtl_variance(_ptr(src), n, _ptr(dst), period))
return _wrap(dst, idx, f"VAR_{period}", "volatility", offset)
def etherm(high: object, low: object, length: int = 14,
def etherm(high: object, low: object, period: int = 14,
offset: int = 0, **kwargs) -> object:
"""Elder Thermometer."""
length = int(length)
period = int(kwargs.get("length", period))
h, idx = _arr(high); l, _ = _arr(low)
n = len(h); dst = _out(n)
_check(_lib.qtl_etherm(_ptr(h), _ptr(l), n, _ptr(dst), length))
return _wrap(dst, idx, f"ETHERM_{length}", "volatility", int(offset))
_check(_lib.qtl_etherm(_ptr(h), _ptr(l), n, _ptr(dst), period))
return _wrap(dst, idx, f"ETHERM_{period}", "volatility", int(offset))
def ccv(close: object, short_period: int = 20, long_period: int = 1,
@@ -314,13 +314,13 @@ def ccv(close: object, short_period: int = 20, long_period: int = 1,
return _wrap(dst, idx, f"CCV_{short_period}", "volatility", offset)
def cv(close: object, length: int = 20, min_vol: float = 0.2,
def cv(close: object, period: int = 20, min_vol: float = 0.2,
max_vol: float = 0.7, offset: int = 0, **kwargs) -> object:
"""Coefficient of Variation."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_cv(_ptr(src), n, _ptr(dst), length, float(min_vol), float(max_vol)))
return _wrap(dst, idx, f"CV_{length}", "volatility", offset)
_check(_lib.qtl_cv(_ptr(src), n, _ptr(dst), period, float(min_vol), float(max_vol)))
return _wrap(dst, idx, f"CV_{period}", "volatility", offset)
def cvi(close: object, ema_period: int = 10, roc_period: int = 10,
@@ -332,10 +332,10 @@ def cvi(close: object, ema_period: int = 10, roc_period: int = 10,
return _wrap(dst, idx, f"CVI_{ema_period}", "volatility", offset)
def ewma(close: object, length: int = 20, is_pop: int = 1,
def ewma(close: object, period: int = 20, is_pop: int = 1,
ann_factor: int = 252, offset: int = 0, **kwargs) -> object:
"""Exponentially Weighted Moving Average (volatility)."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
src, idx = _arr(close); n = len(src); dst = _out(n)
_check(_lib.qtl_ewma(_ptr(src), n, _ptr(dst), length, int(is_pop), int(ann_factor)))
return _wrap(dst, idx, f"EWMA_{length}", "volatility", offset)
_check(_lib.qtl_ewma(_ptr(src), n, _ptr(dst), period, int(is_pop), int(ann_factor)))
return _wrap(dst, idx, f"EWMA_{period}", "volatility", offset)
python/quantalib/volume.py
+20 -20
View File
@@ -216,54 +216,54 @@ def pvi(close: object, volume: object, offset: int = 0, **kwargs) -> object:
return _wrap(dst, idx, "PVI", "volume", offset)
def tvi(close: object, volume: object, length: int = 14,
def tvi(close: object, volume: object, period: int = 14,
offset: int = 0, **kwargs) -> object:
"""Trade Volume Index."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
c, idx = _arr(close); v, _ = _arr(volume)
n = len(c); dst = _out(n)
_check(_lib.qtl_tvi(_ptr(c), _ptr(v), n, _ptr(dst), length))
return _wrap(dst, idx, f"TVI_{length}", "volume", offset)
_check(_lib.qtl_tvi(_ptr(c), _ptr(v), n, _ptr(dst), period))
return _wrap(dst, idx, f"TVI_{period}", "volume", offset)
def pvd(close: object, volume: object, length: int = 14,
def pvd(close: object, volume: object, period: int = 14,
offset: int = 0, **kwargs) -> object:
"""Price Volume Divergence."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
c, idx = _arr(close); v, _ = _arr(volume)
n = len(c); dst = _out(n)
_check(_lib.qtl_pvd(_ptr(c), _ptr(v), n, _ptr(dst), length))
return _wrap(dst, idx, f"PVD_{length}", "volume", offset)
_check(_lib.qtl_pvd(_ptr(c), _ptr(v), n, _ptr(dst), period))
return _wrap(dst, idx, f"PVD_{period}", "volume", offset)
def vwma(close: object, volume: object, length: int = 20,
def vwma(close: object, volume: object, period: int = 20,
offset: int = 0, **kwargs) -> object:
"""Volume Weighted Moving Average."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
c, idx = _arr(close); v, _ = _arr(volume)
n = len(c); dst = _out(n)
_check(_lib.qtl_vwma(_ptr(c), _ptr(v), n, _ptr(dst), length))
return _wrap(dst, idx, f"VWMA_{length}", "volume", offset)
_check(_lib.qtl_vwma(_ptr(c), _ptr(v), n, _ptr(dst), period))
return _wrap(dst, idx, f"VWMA_{period}", "volume", offset)
def evwma(close: object, volume: object, length: int = 20,
def evwma(close: object, volume: object, period: int = 20,
offset: int = 0, **kwargs) -> object:
"""Elastic Volume Weighted Moving Average."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
c, idx = _arr(close); v, _ = _arr(volume)
n = len(c); dst = _out(n)
_check(_lib.qtl_evwma(_ptr(c), _ptr(v), n, _ptr(dst), length))
return _wrap(dst, idx, f"EVWMA_{length}", "volume", offset)
_check(_lib.qtl_evwma(_ptr(c), _ptr(v), n, _ptr(dst), period))
return _wrap(dst, idx, f"EVWMA_{period}", "volume", offset)
def efi(close: object, volume: object, length: int = 13,
def efi(close: object, volume: object, period: int = 13,
offset: int = 0, **kwargs) -> object:
"""Elder Force Index."""
length = int(length); offset = int(offset)
period = int(kwargs.get("length", period)); offset = int(offset)
c, idx = _arr(close); v, _ = _arr(volume)
n = len(c); dst = _out(n)
_check(_lib.qtl_efi(_ptr(c), _ptr(v), n, _ptr(dst), length))
return _wrap(dst, idx, f"EFI_{length}", "volume", offset)
_check(_lib.qtl_efi(_ptr(c), _ptr(v), n, _ptr(dst), period))
return _wrap(dst, idx, f"EFI_{period}", "volume", offset)
def aobv(close: object, volume: object, offset: int = 0, **kwargs) -> object: