mirror of
https://github.com/mihakralj/QuanTAlib.git
synced 2026-07-28 09:47:43 +00:00
Miha Kralj
237 lines
7.8 KiB
Python
237 lines
7.8 KiB
Python
"""Shared wrapper helpers for quantalib indicator modules.
|
|
|
|
Auto-generated by generate_category_modules.py — DO NOT EDIT.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import numpy as np
|
|
from numpy.typing import NDArray
|
|
|
|
from ._bridge import _lib, _check, _dp, _ci, _cd
|
|
|
|
# Optional pandas support
|
|
try:
|
|
import pandas as pd # type: ignore[import-untyped]
|
|
except ImportError: # pragma: no cover
|
|
pd = None # type: ignore[assignment]
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Internal helpers
|
|
# ---------------------------------------------------------------------------
|
|
_F64 = np.float64
|
|
|
|
|
|
def _arr(x: object) -> tuple[NDArray[np.float64], object]:
|
|
"""Return (contiguous float64 array, original_index_or_None)."""
|
|
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)
|
|
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
|
|
|
|
|
|
def _ptr(a: NDArray[np.float64]): # noqa: ANN202
|
|
"""Get ctypes double* from array."""
|
|
return a.ctypes.data_as(_dp)
|
|
|
|
|
|
def _out(n: int) -> NDArray[np.float64]:
|
|
"""Allocate output array."""
|
|
return np.empty(n, dtype=_F64)
|
|
|
|
|
|
def _offset(arr: NDArray[np.float64], off: int) -> NDArray[np.float64]:
|
|
"""Apply offset (roll + NaN fill)."""
|
|
if off != 0:
|
|
arr = np.roll(arr, off)
|
|
if off > 0:
|
|
arr[:off] = np.nan
|
|
else:
|
|
arr[off:] = np.nan
|
|
return arr
|
|
|
|
|
|
def _wrap(
|
|
arr: NDArray[np.float64],
|
|
idx: object,
|
|
name: str,
|
|
category: str,
|
|
offset: int = 0,
|
|
):
|
|
"""Wrap result: apply offset, optionally convert to pd.Series."""
|
|
arr = _offset(arr, offset)
|
|
if idx is not None and pd is not None:
|
|
s = pd.Series(arr, index=idx, name=name)
|
|
s.attrs["category"] = category
|
|
return s
|
|
return arr
|
|
|
|
|
|
def _wrap_multi(
|
|
arrays: dict[str, NDArray[np.float64]],
|
|
idx: object,
|
|
category: str,
|
|
offset: int = 0,
|
|
):
|
|
"""Wrap multi-output result into tuple or DataFrame."""
|
|
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)
|
|
df.attrs["category"] = category
|
|
return df
|
|
return tuple(arrays.values())
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
# Generic pattern helpers
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
|
|
def _pa(
|
|
fn_name: str, close: object, length: int, offset: int,
|
|
default_length: int, label: str, category: str,
|
|
) -> object:
|
|
"""Generic Pattern A wrapper: single-input + period."""
|
|
length = int(length) if length is not None else default_length
|
|
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)
|
|
|
|
|
|
def _pa3(
|
|
fn_name: str, close: object, offset: int,
|
|
label: str, category: str,
|
|
) -> object:
|
|
"""Generic Pattern A3 wrapper: single-input, no params."""
|
|
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)))
|
|
return _wrap(dst, idx, label, category, offset)
|
|
|
|
|
|
def _pf(
|
|
fn_name: str, actual: object, predicted: object,
|
|
length: int, offset: int, default_length: int,
|
|
label: str, category: str,
|
|
) -> object:
|
|
"""Generic Pattern F wrapper: actual+predicted+period."""
|
|
length = int(length) if length is not None else default_length
|
|
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)
|
|
|
|
|
|
def _pg(
|
|
fn_name: str, close: object, volume: object,
|
|
offset: int, label: str, category: str,
|
|
) -> object:
|
|
"""Pattern G: source+volume, no 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)))
|
|
return _wrap(dst, idx, label, category, offset)
|
|
|
|
|
|
def _pg2(
|
|
fn_name: str, close: object, volume: object, length: int,
|
|
offset: int, default_length: int, label: str, category: str,
|
|
) -> object:
|
|
"""Pattern G2: source+volume+period."""
|
|
length = int(length) if length is not None else default_length
|
|
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)
|
|
|
|
|
|
def _ph(
|
|
fn_name: str, x: object, y: object, length: int,
|
|
offset: int, default_length: int, label: str, category: str,
|
|
) -> object:
|
|
"""Pattern H: X+Y+period."""
|
|
length = int(length) if length is not None else default_length
|
|
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)
|
|
|
|
|
|
def _ohlcv_bars_period(
|
|
fn_name: str, open: object, high: object, low: object,
|
|
close: object, volume: object, period: int,
|
|
offset: int, default_period: int, label: str, category: str,
|
|
) -> object:
|
|
"""OHLCV bars + period → single output (BuildBars pattern)."""
|
|
period = int(period) if period is not None else default_period
|
|
offset = int(offset) if offset is not None else 0
|
|
o, idx = _arr(open)
|
|
h, _ = _arr(high)
|
|
l, _ = _arr(low)
|
|
c, _ = _arr(close)
|
|
v, _ = _arr(volume)
|
|
n = len(o)
|
|
dst = _out(n)
|
|
_check(getattr(_lib, fn_name)(
|
|
_ptr(o), _ptr(h), _ptr(l), _ptr(c), _ptr(v), period, n, _ptr(dst)))
|
|
return _wrap(dst, idx, f"{label}_{period}", category, offset)
|
|
|
|
|
|
def _hlc_period(
|
|
fn_name: str, high: object, low: object, close: object,
|
|
period: int, offset: int, default_period: int,
|
|
label: str, category: str,
|
|
) -> object:
|
|
"""HLC + period → single output."""
|
|
period = int(period) if period is not None else default_period
|
|
offset = int(offset) if offset is not None else 0
|
|
h, idx = _arr(high)
|
|
l, _ = _arr(low)
|
|
c, _ = _arr(close)
|
|
n = len(h)
|
|
dst = _out(n)
|
|
_check(getattr(_lib, fn_name)(
|
|
_ptr(h), _ptr(l), _ptr(c), period, n, _ptr(dst)))
|
|
return _wrap(dst, idx, f"{label}_{period}", category, offset)
|
|
|
|
|
|
def _src_period(
|
|
fn_name: str, source: object, period: int,
|
|
offset: int, default_period: int, label: str, category: str,
|
|
) -> object:
|
|
"""source + period → single output (BuildSeries pattern, src,period,n,dst)."""
|
|
period = int(period) if period is not None else default_period
|
|
offset = int(offset) if offset is not None else 0
|
|
src, idx = _arr(source)
|
|
n = len(src)
|
|
dst = _out(n)
|
|
_check(getattr(_lib, fn_name)(_ptr(src), period, n, _ptr(dst)))
|
|
return _wrap(dst, idx, f"{label}_{period}", category, offset)
|