gavindiaz/ferro-ta
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore: release v1.0.2

Browse Source
This commit is contained in:
Pratik Bhadane
2026-03-24 02:02:10 +05:30
parent 9011250f99
commit 2d5000262f
47 changed files with 3821 additions and 422 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/ferro_ta/analysis/backtest.py
+13 -8
View File
@@ -360,10 +360,10 @@ def backtest(
bar_returns[1:] = np.diff(c) / c[:-1]
strategy_returns = positions * bar_returns
position_changed = np.concatenate([[False], positions[1:] != positions[:-1]])
# Slippage: on each position change, reduce return by slippage_bps/10000 (one-way)
if slippage_bps > 0:
position_changed = np.concatenate([[False], positions[1:] != positions[:-1]])
strategy_returns = strategy_returns.copy()
strategy_returns[position_changed] -= slippage_bps / 10_000.0
@@ -371,13 +371,18 @@ def backtest(
if commission_per_trade <= 0:
equity = np.cumprod(1.0 + strategy_returns)
else:
equity = np.empty(len(c), dtype=np.float64)
equity[0] = 1.0
position_changed = np.concatenate([[False], positions[1:] != positions[:-1]])
for i in range(1, len(c)):
equity[i] = equity[i - 1] * (1.0 + strategy_returns[i])
if position_changed[i]:
equity[i] -= commission_per_trade
gross_equity = np.cumprod(1.0 + strategy_returns)
if np.any(gross_equity == 0.0):
equity = np.empty(len(c), dtype=np.float64)
equity[0] = 1.0
for i in range(1, len(c)):
equity[i] = equity[i - 1] * (1.0 + strategy_returns[i])
if position_changed[i]:
equity[i] -= commission_per_trade
else:
commissions = position_changed.astype(np.float64) * commission_per_trade
discounted_commissions = np.cumsum(commissions / gross_equity)
equity = gross_equity * (1.0 - discounted_commissions)
return BacktestResult(
signals=signals,
python/ferro_ta/analysis/features.py
+24 -59
View File
@@ -24,12 +24,23 @@ import numpy as np
from numpy.typing import NDArray
from ferro_ta._utils import _to_f64
from ferro_ta.core.registry import run as _registry_run
from ferro_ta.data.batch import compute_many
__all__ = [
"feature_matrix",
]
def _forward_fill_nan(arr: NDArray[np.float64]) -> NDArray[np.float64]:
mask = np.isnan(arr)
if not mask.any():
return arr
last_valid = np.where(~mask, np.arange(len(arr)), 0)
np.maximum.accumulate(last_valid, out=last_valid)
return arr[last_valid]
# ---------------------------------------------------------------------------
# feature_matrix
# ---------------------------------------------------------------------------
@@ -117,67 +128,23 @@ def feature_matrix(
n = len(close)
columns: dict[str, NDArray[np.float64]] = {}
# --- Indicators needing HLCV ---
_multi_input = {
"ATR",
"NATR",
"TRANGE",
"ADX",
"ADXR",
"PLUS_DI",
"MINUS_DI",
"PLUS_DM",
"MINUS_DM",
"DX",
"AROON",
"AROONOSC",
"CCI",
"MFI",
"STOCH",
"STOCHF",
"STOCHRSI",
"WILLR",
"AD",
"ADOSC",
"OBV",
"VWAP",
"DONCHIAN",
"ICHIMOKU",
}
results = compute_many(
indicators,
close=close,
high=high if high is not None else None,
low=low if low is not None else None,
volume=volume if volume is not None else None,
)
def _call_indicator(name: str, kwargs: dict[str, Any]) -> Any:
# Try with close only first; if that fails try with hlcv
try:
return _registry_run(name, close, **kwargs)
except (TypeError, Exception):
pass
# Build appropriate positional args from available arrays
if name in _multi_input and high is not None and low is not None:
try:
return _registry_run(name, high, low, close, **kwargs)
except Exception:
pass
if volume is not None:
try:
return _registry_run(name, high, low, close, volume, **kwargs)
except Exception:
pass
raise ValueError(
f"Cannot call indicator '{name}': insufficient data columns or incompatible parameters."
)
for spec in indicators:
for spec, result in zip(indicators, results):
if isinstance(spec, str):
name = spec
kwargs: dict[str, Any] = {}
out_key: Optional[Any] = None
elif len(spec) == 2:
name, kwargs = spec # type: ignore[misc]
name, _ = spec # type: ignore[misc]
out_key = None
else:
name, kwargs, out_key = spec # type: ignore[misc]
result = _call_indicator(name, kwargs)
name, _, out_key = spec # type: ignore[misc]
if isinstance(result, tuple):
if out_key is not None:
@@ -215,8 +182,6 @@ def feature_matrix(
mask &= ~np.isnan(arr)
return {k: v[mask] for k, v in columns.items()}
elif nan_policy == "fill":
for k, arr in columns.items():
for i in range(1, len(arr)):
if np.isnan(arr[i]):
arr[i] = arr[i - 1]
for key, arr in columns.items():
columns[key] = _forward_fill_nan(arr)
return columns
python/ferro_ta/analysis/options.py
+21 -20
View File
@@ -45,6 +45,7 @@ iv_zscore(iv_series, window)
from __future__ import annotations
import numpy as np
from numpy.lib.stride_tricks import sliding_window_view
from numpy.typing import ArrayLike, NDArray
from ferro_ta.core.exceptions import FerroTAInputError, FerroTAValueError
@@ -103,15 +104,15 @@ def iv_rank(
arr = _validate_iv(np.asarray(iv_series, dtype=np.float64), window)
n = len(arr)
out = np.full(n, np.nan, dtype=np.float64)
if window > n:
return out
for i in range(window - 1, n):
window_slice = arr[i - window + 1 : i + 1]
lo = float(np.nanmin(window_slice))
hi = float(np.nanmax(window_slice))
if hi == lo:
out[i] = 0.0
else:
out[i] = (arr[i] - lo) / (hi - lo)
windows = sliding_window_view(arr, window_shape=window)
lower = np.nanmin(windows, axis=1)
upper = np.nanmax(windows, axis=1)
current = arr[window - 1 :]
spread = upper - lower
out[window - 1 :] = np.where(spread == 0.0, 0.0, (current - lower) / spread)
return out
@@ -149,11 +150,12 @@ def iv_percentile(
arr = _validate_iv(np.asarray(iv_series, dtype=np.float64), window)
n = len(arr)
out = np.full(n, np.nan, dtype=np.float64)
if window > n:
return out
for i in range(window - 1, n):
window_slice = arr[i - window + 1 : i + 1]
current = arr[i]
out[i] = float(np.sum(window_slice <= current)) / window
windows = sliding_window_view(arr, window_shape=window)
current = arr[window - 1 :, None]
out[window - 1 :] = np.sum(windows <= current, axis=1, dtype=np.int64) / window
return out
@@ -192,14 +194,13 @@ def iv_zscore(
arr = _validate_iv(np.asarray(iv_series, dtype=np.float64), window)
n = len(arr)
out = np.full(n, np.nan, dtype=np.float64)
if window > n:
return out
for i in range(window - 1, n):
window_slice = arr[i - window + 1 : i + 1]
mu = float(np.nanmean(window_slice))
sigma = float(np.nanstd(window_slice, ddof=0))
if sigma == 0.0:
out[i] = np.nan
else:
out[i] = (arr[i] - mu) / sigma
windows = sliding_window_view(arr, window_shape=window)
mean = np.nanmean(windows, axis=1)
std = np.nanstd(windows, axis=1, ddof=0)
current = arr[window - 1 :]
out[window - 1 :] = np.where(std == 0.0, np.nan, (current - mean) / std)
return out
python/ferro_ta/analysis/signals.py
+2 -7
View File
@@ -33,6 +33,7 @@ import numpy as np
from numpy.typing import ArrayLike, NDArray
from ferro_ta._ferro_ta import bottom_n_indices as _rust_bottom_n
from ferro_ta._ferro_ta import compose_rank as _rust_compose_rank
from ferro_ta._ferro_ta import compose_weighted as _rust_compose_weighted
from ferro_ta._ferro_ta import rank_series as _rust_rank_series
from ferro_ta._ferro_ta import top_n_indices as _rust_top_n
@@ -131,13 +132,7 @@ def compose(
w = np.full(n_sigs, 1.0 / n_sigs)
return _rust_compose_weighted(arr, w)
elif method == "rank":
# Replace each column with its rank, then sum (ensure contiguous slices)
ranked = np.column_stack(
[_rust_rank_series(np.ascontiguousarray(arr[:, j])) for j in range(n_sigs)]
)
ranked = np.ascontiguousarray(ranked)
w = np.full(n_sigs, 1.0)
return _rust_compose_weighted(ranked, w)
return _rust_compose_rank(arr)
else:
# weighted (default)
if weights is None: