Pratik Bhadane
53566b9d82
Move several hot Python analysis paths to Rust-backed helpers. This adds Rust implementations for backtest strategy signal generation and the core portfolio loop, options and futures payoff aggregation, Greeks aggregation, ratio calculation, trade extraction, chunked close-only indicator runs, and forward-fill helpers. Wire the Python analysis and data modules to prefer these paths, and add coverage for the new batch fast path. Expand the WASM package to export WMA, ADX, and MFI from ferro_ta_core, refresh the Node examples, benchmarks, and README, and add a Node-vs-Python conformance test so the browser and node surface stays aligned with the main Python package. Introduce a generated cross-surface API manifest in docs/, along with scripts to rebuild and verify it from source exports. Enforce manifest freshness in the Python and WASM CI workflows so release candidates catch surface drift before push.
348 lines
12 KiB
Python
348 lines
12 KiB
Python
"""
|
|
Minimal Backtesting Harness
|
|
============================
|
|
|
|
A lightweight vectorized backtester that uses ferro_ta indicators as the engine.
|
|
|
|
**Scope** (minimal harness):
|
|
- Vectorized approach: compute indicators once, then apply a signal function over bars.
|
|
- Single-asset, long-only or long/short, no leverage.
|
|
- Optional **commission** (per trade) and **slippage** (basis points) for more realistic equity.
|
|
- Returns a :class:`BacktestResult` with signals, positions, and equity curve.
|
|
|
|
For production backtesting consider `backtrader`, `zipline`, or `vectorbt`.
|
|
|
|
Quick start
|
|
-----------
|
|
>>> import numpy as np
|
|
>>> from ferro_ta.analysis.backtest import backtest, rsi_strategy
|
|
>>>
|
|
>>> # Generate synthetic OHLCV data
|
|
>>> np.random.seed(42)
|
|
>>> n = 100
|
|
>>> close = np.cumprod(1 + np.random.randn(n) * 0.01) * 100
|
|
>>> volume = np.random.randint(1_000, 10_000, n).astype(float)
|
|
>>>
|
|
>>> result = backtest(close, volume=volume, strategy="rsi_30_70")
|
|
>>> print(result) # BacktestResult(bars=100, trades=…, final_equity=…)
|
|
|
|
API
|
|
---
|
|
backtest(close, *, high=None, low=None, open=None, volume=None,
|
|
strategy="rsi_30_70", commission_per_trade=0, slippage_bps=0, **kwargs)
|
|
Run the backtester and return a :class:`BacktestResult`. Optional
|
|
commission (subtracted from equity on each position change) and slippage
|
|
(basis points; applied as a cost on the bar where position changes).
|
|
|
|
rsi_strategy(close, timeperiod=14, oversold=30, overbought=70)
|
|
Built-in RSI oversold/overbought strategy; returns a signal array.
|
|
|
|
sma_crossover_strategy(close, fast=10, slow=30)
|
|
Built-in SMA crossover strategy; returns a signal array.
|
|
|
|
macd_crossover_strategy(close, fastperiod=12, slowperiod=26, signalperiod=9)
|
|
Built-in MACD line/signal crossover strategy; returns a signal array.
|
|
|
|
BacktestResult
|
|
Dataclass-like container with signals, positions, returns, equity arrays.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Callable
|
|
from typing import Optional, Union
|
|
|
|
import numpy as np
|
|
from numpy.typing import ArrayLike, NDArray
|
|
|
|
from ferro_ta._ferro_ta import backtest_core as _rust_backtest_core
|
|
from ferro_ta._ferro_ta import macd_crossover_signals as _rust_macd_crossover_signals
|
|
from ferro_ta._ferro_ta import rsi_threshold_signals as _rust_rsi_threshold_signals
|
|
from ferro_ta._ferro_ta import sma_crossover_signals as _rust_sma_crossover_signals
|
|
from ferro_ta.core.exceptions import FerroTAInputError, FerroTAValueError
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# BacktestResult
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class BacktestResult:
|
|
"""Container for backtesting output.
|
|
|
|
Attributes
|
|
----------
|
|
signals : NDArray[np.float64]
|
|
Array of +1 (long), -1 (short), or 0 (flat) for every bar.
|
|
positions : NDArray[np.float64]
|
|
Lagged signals — position held *during* each bar (shift by 1 to
|
|
avoid look-ahead bias).
|
|
bar_returns : NDArray[np.float64]
|
|
Per-bar return of the underlying close price (pct change).
|
|
strategy_returns : NDArray[np.float64]
|
|
``positions * bar_returns`` — strategy return at each bar.
|
|
equity : NDArray[np.float64]
|
|
Cumulative equity curve starting at 1.0.
|
|
n_trades : int
|
|
Number of position changes.
|
|
final_equity : float
|
|
Terminal equity value.
|
|
"""
|
|
|
|
__slots__ = (
|
|
"signals",
|
|
"positions",
|
|
"bar_returns",
|
|
"strategy_returns",
|
|
"equity",
|
|
"n_trades",
|
|
"final_equity",
|
|
)
|
|
|
|
def __init__(
|
|
self,
|
|
signals: NDArray[np.float64],
|
|
positions: NDArray[np.float64],
|
|
bar_returns: NDArray[np.float64],
|
|
strategy_returns: NDArray[np.float64],
|
|
equity: NDArray[np.float64],
|
|
) -> None:
|
|
self.signals = signals
|
|
self.positions = positions
|
|
self.bar_returns = bar_returns
|
|
self.strategy_returns = strategy_returns
|
|
self.equity = equity
|
|
self.n_trades = int(np.sum(np.diff(positions) != 0))
|
|
self.final_equity = float(equity[-1]) if len(equity) > 0 else 1.0
|
|
|
|
def __repr__(self) -> str: # pragma: no cover
|
|
return (
|
|
f"BacktestResult("
|
|
f"bars={len(self.signals)}, "
|
|
f"trades={self.n_trades}, "
|
|
f"final_equity={self.final_equity:.4f})"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Built-in strategies
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def rsi_strategy(
|
|
close: ArrayLike,
|
|
timeperiod: int = 14,
|
|
oversold: float = 30.0,
|
|
overbought: float = 70.0,
|
|
) -> NDArray[np.float64]:
|
|
"""RSI oversold / overbought signal generator.
|
|
|
|
Returns
|
|
-------
|
|
signals : ndarray of float64
|
|
+1 where RSI <= oversold (buy signal), -1 where RSI >= overbought
|
|
(sell signal), 0 otherwise. NaN during the RSI warm-up period.
|
|
|
|
Parameters
|
|
----------
|
|
close : array-like
|
|
Close prices.
|
|
timeperiod : int
|
|
RSI look-back period (default 14).
|
|
oversold : float
|
|
RSI level below which a long (+1) signal is generated (default 30).
|
|
overbought : float
|
|
RSI level above which a short (-1) signal is generated (default 70).
|
|
"""
|
|
if timeperiod < 1:
|
|
raise FerroTAValueError(f"timeperiod must be >= 1, got {timeperiod}")
|
|
|
|
c = np.asarray(close, dtype=np.float64)
|
|
return np.asarray(
|
|
_rust_rsi_threshold_signals(c, int(timeperiod), float(oversold), float(overbought)),
|
|
dtype=np.float64,
|
|
)
|
|
|
|
|
|
def sma_crossover_strategy(
|
|
close: ArrayLike,
|
|
fast: int = 10,
|
|
slow: int = 30,
|
|
) -> NDArray[np.float64]:
|
|
"""SMA fast/slow crossover strategy.
|
|
|
|
Returns
|
|
-------
|
|
signals : ndarray of float64
|
|
+1 when fast SMA > slow SMA (uptrend), -1 when fast SMA < slow SMA
|
|
(downtrend), NaN during the warm-up window.
|
|
|
|
Parameters
|
|
----------
|
|
close : array-like
|
|
Close prices.
|
|
fast : int
|
|
Fast SMA period (default 10).
|
|
slow : int
|
|
Slow SMA period (default 30).
|
|
"""
|
|
if fast < 1:
|
|
raise FerroTAValueError(f"fast must be >= 1, got {fast}")
|
|
if slow < 1:
|
|
raise FerroTAValueError(f"slow must be >= 1, got {slow}")
|
|
if fast >= slow:
|
|
raise FerroTAValueError(f"fast ({fast}) must be less than slow ({slow})")
|
|
|
|
c = np.asarray(close, dtype=np.float64)
|
|
return np.asarray(
|
|
_rust_sma_crossover_signals(c, int(fast), int(slow)),
|
|
dtype=np.float64,
|
|
)
|
|
|
|
|
|
def macd_crossover_strategy(
|
|
close: ArrayLike,
|
|
fastperiod: int = 12,
|
|
slowperiod: int = 26,
|
|
signalperiod: int = 9,
|
|
) -> NDArray[np.float64]:
|
|
"""MACD line / signal line crossover strategy.
|
|
|
|
Returns
|
|
-------
|
|
signals : ndarray of float64
|
|
+1 when MACD line > signal line (uptrend), -1 when MACD line < signal line
|
|
(downtrend), NaN during the MACD warm-up window.
|
|
|
|
Parameters
|
|
----------
|
|
close : array-like
|
|
Close prices.
|
|
fastperiod : int
|
|
Fast EMA period (default 12).
|
|
slowperiod : int
|
|
Slow EMA period (default 26).
|
|
signalperiod : int
|
|
Signal line EMA period (default 9).
|
|
"""
|
|
if fastperiod < 1 or slowperiod < 1 or signalperiod < 1:
|
|
raise FerroTAValueError("MACD periods must be >= 1")
|
|
if fastperiod >= slowperiod:
|
|
raise FerroTAValueError(
|
|
f"fastperiod ({fastperiod}) must be less than slowperiod ({slowperiod})"
|
|
)
|
|
|
|
c = np.asarray(close, dtype=np.float64)
|
|
return np.asarray(
|
|
_rust_macd_crossover_signals(
|
|
c, int(fastperiod), int(slowperiod), int(signalperiod)
|
|
),
|
|
dtype=np.float64,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Built-in strategy registry
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_BUILTIN_STRATEGIES: dict[str, Callable[..., NDArray[np.float64]]] = {
|
|
"rsi_30_70": rsi_strategy,
|
|
"sma_crossover": sma_crossover_strategy,
|
|
"macd_crossover": macd_crossover_strategy,
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Main backtest entry point
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def backtest(
|
|
close: ArrayLike,
|
|
*,
|
|
high: Optional[ArrayLike] = None,
|
|
low: Optional[ArrayLike] = None,
|
|
open: Optional[ArrayLike] = None,
|
|
volume: Optional[ArrayLike] = None,
|
|
strategy: Union[str, Callable[..., NDArray[np.float64]]] = "rsi_30_70",
|
|
commission_per_trade: float = 0.0,
|
|
slippage_bps: float = 0.0,
|
|
**strategy_kwargs: object,
|
|
) -> BacktestResult:
|
|
"""Run a vectorized backtest on *close* prices using *strategy*.
|
|
|
|
Parameters
|
|
----------
|
|
close : array-like
|
|
Close prices (required).
|
|
high, low, open, volume : array-like, optional
|
|
Additional OHLCV data. Passed to the strategy function if it accepts
|
|
them (via ``**strategy_kwargs``); currently unused by the built-in
|
|
strategies.
|
|
strategy : str or callable
|
|
Either a name of a built-in strategy (``"rsi_30_70"``,
|
|
``"sma_crossover"``, or ``"macd_crossover"``) or a callable with
|
|
signature ``(close, **kwargs) -> ndarray`` that returns a signal array.
|
|
commission_per_trade : float, optional
|
|
Fixed commission deducted from equity on each position change (default 0).
|
|
slippage_bps : float, optional
|
|
Slippage in basis points (1 bps = 0.01%) applied as a cost on the bar
|
|
where the position changes (default 0).
|
|
**strategy_kwargs
|
|
Extra keyword arguments forwarded to the strategy function
|
|
(e.g. ``timeperiod=14``, ``oversold=30``).
|
|
|
|
Returns
|
|
-------
|
|
BacktestResult
|
|
Container with signals, positions, equity curve, and trade count.
|
|
|
|
Raises
|
|
------
|
|
FerroTAValueError
|
|
If a named strategy is unknown.
|
|
FerroTAInputError
|
|
If ``close`` is too short (< 2 bars) or contains non-finite values.
|
|
|
|
Notes
|
|
-----
|
|
Commission is subtracted from equity immediately after each position change.
|
|
Slippage is applied by reducing the strategy return on the bar where the
|
|
position changes by ``slippage_bps / 10000`` (one-way).
|
|
"""
|
|
c = np.asarray(close, dtype=np.float64)
|
|
if c.ndim != 1:
|
|
raise FerroTAInputError("close must be a 1-D array.")
|
|
if len(c) < 2:
|
|
raise FerroTAInputError(f"close must have at least 2 bars, got {len(c)}.")
|
|
|
|
# ------------------------------------------------------------------
|
|
# Resolve strategy
|
|
# ------------------------------------------------------------------
|
|
if isinstance(strategy, str):
|
|
if strategy not in _BUILTIN_STRATEGIES:
|
|
raise FerroTAValueError(
|
|
f"Unknown strategy '{strategy}'. "
|
|
f"Available: {sorted(_BUILTIN_STRATEGIES)}"
|
|
)
|
|
strategy_fn: Callable[..., NDArray[np.float64]] = _BUILTIN_STRATEGIES[strategy]
|
|
elif callable(strategy):
|
|
strategy_fn = strategy
|
|
else:
|
|
raise FerroTAValueError("strategy must be a string name or a callable.")
|
|
|
|
# ------------------------------------------------------------------
|
|
# Compute signals
|
|
# ------------------------------------------------------------------
|
|
signals = np.asarray(strategy_fn(c, **strategy_kwargs), dtype=np.float64)
|
|
positions, bar_returns, strategy_returns, equity = _rust_backtest_core(
|
|
c, signals, float(commission_per_trade), float(slippage_bps)
|
|
)
|
|
|
|
return BacktestResult(
|
|
signals=signals,
|
|
positions=np.asarray(positions, dtype=np.float64),
|
|
bar_returns=np.asarray(bar_returns, dtype=np.float64),
|
|
strategy_returns=np.asarray(strategy_returns, dtype=np.float64),
|
|
equity=np.asarray(equity, dtype=np.float64),
|
|
)
|