"""Tests for the advanced backtesting engine. Covers all 10 test groups from the plan: 1. backtest_ohlcv_core 2. compute_performance_metrics 3. extract_trades 4. backtest_multi_asset_core 5. monte_carlo_bootstrap 6. walk_forward_indices 7. kelly_fraction / half_kelly_fraction 8. BacktestEngine (Python API) 9. walk_forward() (Python API) 10. monte_carlo() (Python API) """ from __future__ import annotations import math import numpy as np import numpy.testing as npt import pytest from ferro_ta._ferro_ta import ( backtest_core, backtest_multi_asset_core, backtest_ohlcv_core, compute_performance_metrics, drawdown_series, half_kelly_fraction, kelly_fraction, monte_carlo_bootstrap, walk_forward_indices, ) from ferro_ta._ferro_ta import ( extract_trades_ohlcv as extract_trades, ) from ferro_ta.analysis.backtest import ( AdvancedBacktestResult, BacktestEngine, BacktestResult, MonteCarloResult, PortfolioBacktestResult, WalkForwardResult, backtest, backtest_portfolio, monte_carlo, rsi_strategy, walk_forward, ) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _make_ohlcv(n: int = 100, seed: int = 42) -> tuple: rng = np.random.default_rng(seed) close = np.cumprod(1 + rng.standard_normal(n) * 0.01) * 100.0 open_ = close * (1 - rng.uniform(0, 0.005, n)) high = close * (1 + rng.uniform(0, 0.01, n)) low = close * (1 - rng.uniform(0, 0.01, n)) signals = np.where(np.arange(n) % 20 < 10, 1.0, -1.0).astype(np.float64) return open_, high, low, close, signals def _all_finite(arr: np.ndarray) -> bool: return bool(np.all(np.isfinite(arr[~np.isnan(arr)]))) # =========================================================================== # Group 1: backtest_ohlcv_core # =========================================================================== class TestBacktestOhlcvCore: def test_returns_five_arrays(self): o, h, l, c, s = _make_ohlcv() result = backtest_ohlcv_core(o, h, l, c, s) assert len(result) == 5 def test_shapes_match_input(self): o, h, l, c, s = _make_ohlcv(n=80) pos, fp, br, sr, eq = backtest_ohlcv_core(o, h, l, c, s) for arr in (pos, fp, br, sr, eq): assert arr.shape == (80,) def test_equity_starts_at_one(self): o, h, l, c, s = _make_ohlcv() _, _, _, _, eq = backtest_ohlcv_core(o, h, l, c, s) assert eq[0] == pytest.approx(1.0, abs=1e-9) def test_no_lookahead_bias(self): """Position at bar 0 must always be 0 (signal not yet available).""" o, h, l, c, s = _make_ohlcv() pos, _, _, _, _ = backtest_ohlcv_core(o, h, l, c, s) assert pos[0] == 0.0 def test_stop_loss_reduces_equity_relative_to_no_stop(self): """With a tight stop-loss, equity should differ from no-stop run.""" o, h, l, c, s = _make_ohlcv(n=200) _, _, _, _, eq_no_stop = backtest_ohlcv_core(o, h, l, c, s) _, _, _, _, eq_with_stop = backtest_ohlcv_core( o, h, l, c, s, stop_loss_pct=0.005 ) # They should differ (stop-loss triggered on at least one bar) assert not np.allclose(eq_no_stop, eq_with_stop) def test_fill_prices_nan_when_flat(self): """fill_prices must be NaN whenever the position is 0.""" o, h, l, c, s = _make_ohlcv() pos, fp, _, _, _ = backtest_ohlcv_core(o, h, l, c, s) flat_mask = pos == 0.0 assert np.all(np.isnan(fp[flat_mask])) def test_market_close_mode_different_from_open(self): o, h, l, c, s = _make_ohlcv(n=150) _, _, _, sr_open, _ = backtest_ohlcv_core( o, h, l, c, s, fill_mode="market_open" ) _, _, _, sr_close, _ = backtest_ohlcv_core( o, h, l, c, s, fill_mode="market_close" ) # Different fill modes → different returns assert not np.allclose(sr_open, sr_close, equal_nan=True) def test_raises_on_mismatched_lengths(self): o, h, l, c, s = _make_ohlcv() with pytest.raises(Exception): backtest_ohlcv_core(o[:-1], h, l, c, s) # =========================================================================== # Group 2: compute_performance_metrics # =========================================================================== class TestComputePerformanceMetrics: EXPECTED_KEYS = { "total_return", "cagr", "annualized_vol", "sharpe", "sortino", "calmar", "max_drawdown", "avg_drawdown", "max_drawdown_duration_bars", "avg_drawdown_duration_bars", "ulcer_index", "omega_ratio", "win_rate", "profit_factor", "r_expectancy", "avg_win", "avg_loss", "tail_ratio", "skewness", "kurtosis", "best_bar", "worst_bar", "n_trades", } def _run(self, n: int = 200, seed: int = 0): rng = np.random.default_rng(seed) r = rng.standard_normal(n) * 0.01 eq = np.cumprod(1 + r) return compute_performance_metrics(r, eq) def test_all_expected_keys_present(self): m = self._run() assert self.EXPECTED_KEYS.issubset(set(m.keys())) def test_sharpe_all_positive_returns(self): """Constant +1% daily returns → Sharpe = (annualised) > 0.""" r = np.full(252, 0.01) eq = np.cumprod(1 + r) m = compute_performance_metrics(r, eq) assert m["sharpe"] > 0 def test_max_drawdown_matches_drawdown_series(self): rng = np.random.default_rng(7) r = rng.standard_normal(300) * 0.015 eq = np.cumprod(1 + r) m = compute_performance_metrics(r, eq) _, max_dd_ref = drawdown_series(eq) assert m["max_drawdown"] == pytest.approx(max_dd_ref, abs=1e-9) def test_cagr_formula(self): r = np.full(252, 0.01) eq = np.cumprod(1 + r) m = compute_performance_metrics(r, eq) # Rust computes CAGR as (eq[-1]/eq[0])^(ppy/n) - 1, treating eq[0] as start equity expected_cagr = (eq[-1] / eq[0]) ** (252.0 / len(r)) - 1.0 assert m["cagr"] == pytest.approx(expected_cagr, rel=1e-6) def test_win_rate_between_0_and_1(self): m = self._run() assert 0.0 <= m["win_rate"] <= 1.0 def test_max_drawdown_nonpositive(self): m = self._run() assert m["max_drawdown"] <= 0.0 def test_total_return_sign(self): r = np.full(100, 0.005) eq = np.cumprod(1 + r) m = compute_performance_metrics(r, eq) assert m["total_return"] > 0.0 def test_raises_on_short_input(self): with pytest.raises(Exception): compute_performance_metrics(np.array([0.01]), np.array([1.01])) def test_raises_on_mismatched_lengths(self): with pytest.raises(Exception): compute_performance_metrics(np.ones(10) * 0.01, np.ones(20)) # =========================================================================== # Group 3: extract_trades # =========================================================================== class TestExtractTrades: def _run_ohlcv(self, n: int = 100): o, h, l, c, s = _make_ohlcv(n=n) pos, fp, _, _, _ = backtest_ohlcv_core(o, h, l, c, s) return pos, fp, h, l def test_returns_nine_arrays(self): pos, fp, h, l = self._run_ohlcv() result = extract_trades(pos, fp, h, l) assert len(result) == 9 def test_all_arrays_same_length(self): pos, fp, h, l = self._run_ohlcv(n=200) arrays = extract_trades(pos, fp, h, l) lengths = {len(a) for a in arrays} assert len(lengths) == 1 # all same length def test_duration_bars_positive(self): pos, fp, h, l = self._run_ohlcv(n=200) _, _, _, _, _, _, dur, _, _ = extract_trades(pos, fp, h, l) assert np.all(dur >= 0) def test_exit_bar_gte_entry_bar(self): pos, fp, h, l = self._run_ohlcv(n=200) eb, xb, _, _, _, _, _, _, _ = extract_trades(pos, fp, h, l) assert np.all(xb >= eb) def test_direction_is_plus_minus_one(self): pos, fp, h, l = self._run_ohlcv(n=200) _, _, d, _, _, _, _, _, _ = extract_trades(pos, fp, h, l) if len(d) > 0: assert set(np.unique(d)).issubset({1.0, -1.0}) def test_mfe_gte_mae(self): """MFE (best) must always be >= MAE (worst) within the trade.""" pos, fp, h, l = self._run_ohlcv(n=200) _, _, _, _, _, _, _, mae, mfe = extract_trades(pos, fp, h, l) if len(mae) > 0: assert np.all(mfe >= mae) def test_raises_on_mismatched_lengths(self): pos, fp, h, l = self._run_ohlcv() with pytest.raises(Exception): extract_trades(pos[:-1], fp, h, l) # =========================================================================== # Group 4: backtest_multi_asset_core # =========================================================================== class TestBacktestMultiAssetCore: def test_single_asset_matches_backtest_core(self): """1-asset multi_asset == scalar backtest_core with same weights.""" rng = np.random.default_rng(99) n = 150 close = np.cumprod(1 + rng.standard_normal(n) * 0.01) * 100.0 signals = np.where(np.arange(n) % 15 < 7, 1.0, -1.0).astype(np.float64) # Single asset via multi_asset (weights = signals) close2d = close.reshape(n, 1) w2d = signals.reshape(n, 1) ar, pr, pe = backtest_multi_asset_core(close2d, w2d) # Same via backtest_core _, _, sr_ref, eq_ref = backtest_core(close, signals) npt.assert_allclose(pe, np.asarray(eq_ref), rtol=1e-6) def test_returns_shapes(self): n, k = 100, 5 rng = np.random.default_rng(0) c2d = np.cumprod(1 + rng.standard_normal((n, k)) * 0.01, axis=0) * 100 w2d = np.ones((n, k)) * 0.2 ar, pr, pe = backtest_multi_asset_core(c2d, w2d) assert ar.shape == (n, k) assert pr.shape == (n,) assert pe.shape == (n,) def test_parallel_equals_serial(self): n, k = 120, 4 rng = np.random.default_rng(1) c2d = np.cumprod(1 + rng.standard_normal((n, k)) * 0.01, axis=0) * 100 w2d = rng.choice([-1.0, 0.0, 1.0], size=(n, k)).astype(np.float64) _, _, pe_par = backtest_multi_asset_core(c2d, w2d, parallel=True) _, _, pe_ser = backtest_multi_asset_core(c2d, w2d, parallel=False) npt.assert_allclose(pe_par, pe_ser, rtol=1e-10) def test_raises_on_mismatched_shapes(self): c2d = np.ones((50, 3)) w2d = np.ones((50, 4)) # wrong n_assets with pytest.raises(Exception): backtest_multi_asset_core(c2d, w2d) def test_equity_starts_at_one(self): n, k = 50, 2 c2d = np.ones((n, k)) * 100.0 w2d = np.zeros((n, k)) _, _, pe = backtest_multi_asset_core(c2d, w2d) assert pe[0] == pytest.approx(1.0, abs=1e-9) # =========================================================================== # Group 5: monte_carlo_bootstrap # =========================================================================== class TestMonteCarloBootstrap: def _returns(self, n: int = 200, seed: int = 5): rng = np.random.default_rng(seed) return rng.standard_normal(n) * 0.01 def test_output_shape(self): r = self._returns() mc = monte_carlo_bootstrap(r, n_sims=50) assert mc.shape == (50, 200) def test_seed_reproducibility(self): r = self._returns() mc1 = monte_carlo_bootstrap(r, n_sims=100, seed=7) mc2 = monte_carlo_bootstrap(r, n_sims=100, seed=7) npt.assert_array_equal(mc1, mc2) def test_different_seeds_differ(self): r = self._returns() mc1 = monte_carlo_bootstrap(r, n_sims=50, seed=1) mc2 = monte_carlo_bootstrap(r, n_sims=50, seed=2) assert not np.allclose(mc1, mc2) def test_equity_starts_at_one(self): r = self._returns() mc = monte_carlo_bootstrap(r, n_sims=20) # Bootstrap resamples returns randomly, so mc[:,0] = 1 + random_return # All first-bar equity values must be in range of possible (1+r) values possible_first_bar = set(np.round(1.0 + r, 12)) for val in mc[:, 0]: assert any(abs(val - p) < 1e-9 for p in possible_first_bar) def test_block_bootstrap_shape(self): r = self._returns(n=100) mc = monte_carlo_bootstrap(r, n_sims=30, block_size=5) assert mc.shape == (30, 100) def test_raises_on_empty_input(self): with pytest.raises(Exception): monte_carlo_bootstrap(np.array([0.01]), n_sims=10) # =========================================================================== # Group 6: walk_forward_indices # =========================================================================== class TestWalkForwardIndices: def test_output_shape(self): idx = walk_forward_indices(500, 200, 50) assert idx.ndim == 2 assert idx.shape[1] == 4 def test_non_anchored_fixed_train_window(self): idx = walk_forward_indices(400, 200, 50) n_folds = idx.shape[0] assert n_folds >= 2 for fold in idx: tr_len = fold[1] - fold[0] assert tr_len == 200 def test_anchored_growing_train_window(self): idx = walk_forward_indices(400, 150, 50, anchored=True) for fold in idx: assert fold[0] == 0 # always starts at 0 train_lengths = idx[:, 1] - idx[:, 0] assert train_lengths[-1] >= train_lengths[0] def test_no_test_fold_overlap(self): idx = walk_forward_indices(500, 200, 50) # Test intervals should be non-overlapping (step = test_bars by default) for i in range(len(idx) - 1): assert idx[i, 3] <= idx[i + 1, 2] def test_all_test_folds_within_bounds(self): n = 600 idx = walk_forward_indices(n, 200, 100) assert np.all(idx[:, 0] >= 0) assert np.all(idx[:, 3] <= n) def test_step_bars_parameter(self): idx_default = walk_forward_indices(500, 200, 50) idx_step = walk_forward_indices(500, 200, 50, step_bars=25) # Smaller step → more folds assert idx_step.shape[0] >= idx_default.shape[0] def test_raises_when_no_folds_fit(self): with pytest.raises(Exception): walk_forward_indices(100, 80, 80) # 80+80 > 100 # =========================================================================== # Group 7: kelly_fraction / half_kelly_fraction # =========================================================================== class TestKellyFraction: def test_positive_expectancy(self): k = kelly_fraction(0.6, 0.02, 0.01) assert k > 0.0 def test_zero_edge_returns_zero(self): """win_rate = loss_rate AND avg_win = avg_loss → Kelly = 0.""" k = kelly_fraction(0.5, 0.01, 0.01) assert k == pytest.approx(0.0, abs=1e-9) def test_negative_expectancy_clamped_to_zero(self): k = kelly_fraction(0.3, 0.01, 0.02) assert k == 0.0 def test_half_kelly_is_half_of_kelly(self): k = kelly_fraction(0.6, 0.03, 0.015) hk = half_kelly_fraction(0.6, 0.03, 0.015) assert hk == pytest.approx(k / 2.0, rel=1e-9) def test_result_clamped_to_one(self): k = kelly_fraction(0.99, 0.5, 0.001) assert k <= 1.0 def test_raises_on_invalid_win_rate(self): with pytest.raises(Exception): kelly_fraction(1.5, 0.01, 0.01) def test_raises_on_nonpositive_avg_win(self): with pytest.raises(Exception): kelly_fraction(0.6, 0.0, 0.01) # =========================================================================== # Group 8: BacktestEngine (Python API) # =========================================================================== class TestBacktestEngine: def _close(self, n: int = 200, seed: int = 10) -> np.ndarray: rng = np.random.default_rng(seed) return np.cumprod(1 + rng.standard_normal(n) * 0.01) * 100.0 def test_run_returns_advanced_result(self): c = self._close() r = BacktestEngine().run(c, "rsi_30_70") assert isinstance(r, AdvancedBacktestResult) def test_advanced_result_is_backtest_result(self): c = self._close() r = BacktestEngine().run(c, "rsi_30_70") assert isinstance(r, BacktestResult) def test_chaining_returns_self(self): engine = BacktestEngine() assert engine.with_commission(0.001) is engine assert engine.with_slippage(5.0) is engine assert engine.with_stop_loss(0.02) is engine def test_all_metric_keys_present(self): c = self._close() r = BacktestEngine().run(c, "rsi_30_70") assert "sharpe" in r.metrics assert "max_drawdown" in r.metrics assert "cagr" in r.metrics def test_drawdown_series_shape(self): c = self._close() r = BacktestEngine().run(c) assert r.drawdown_series.shape == c.shape def test_drawdown_series_nonpositive(self): c = self._close() r = BacktestEngine().run(c) assert np.all(r.drawdown_series <= 0.0) def test_engine_close_only_matches_backtest_func(self): c = self._close() r_engine = BacktestEngine().run(c, "rsi_30_70") r_func = backtest(c, strategy="rsi_30_70") npt.assert_allclose(r_engine.equity, r_func.equity, rtol=1e-9) def test_ohlcv_mode_runs(self): c = self._close() h = c * 1.01 l = c * 0.99 o = c * 0.999 r = ( BacktestEngine() .with_ohlcv(high=h, low=l, open_=o) .with_stop_loss(0.02) .run(c) ) assert r.equity.shape == c.shape def test_trades_dataframe_columns(self): c = self._close() r = BacktestEngine().run(c, "sma_crossover") if r.trades is not None: expected_cols = { "entry_bar", "exit_bar", "direction", "entry_price", "exit_price", "pnl_pct", "duration_bars", "mae", "mfe", } assert expected_cols.issubset(set(r.trades.columns)) def test_invalid_fill_mode_raises(self): with pytest.raises(Exception): BacktestEngine().with_fill_mode("invalid") # =========================================================================== # Group 9: walk_forward() Python API # =========================================================================== class TestWalkForward: def _setup(self, n: int = 400): rng = np.random.default_rng(99) close = np.cumprod(1 + rng.standard_normal(n) * 0.01) * 100.0 param_grid = [{"timeperiod": p} for p in [10, 14, 20]] return close, param_grid def test_returns_walk_forward_result(self): c, pg = self._setup() r = walk_forward(c, rsi_strategy, pg, train_bars=200, test_bars=50) assert isinstance(r, WalkForwardResult) def test_fold_count_matches_indices(self): c, pg = self._setup() r = walk_forward(c, rsi_strategy, pg, train_bars=200, test_bars=50) assert len(r.fold_results) == r.fold_indices.shape[0] def test_oos_equity_length(self): c, pg = self._setup() r = walk_forward(c, rsi_strategy, pg, train_bars=200, test_bars=50) total_test_bars = sum( int(r.fold_indices[i, 3]) - int(r.fold_indices[i, 2]) for i in range(len(r.fold_results)) ) assert len(r.oos_equity) == total_test_bars def test_oos_metrics_has_sharpe(self): c, pg = self._setup() r = walk_forward(c, rsi_strategy, pg, train_bars=200, test_bars=50) assert "sharpe" in r.oos_metrics def test_anchored_mode(self): c, pg = self._setup() r = walk_forward( c, rsi_strategy, pg, train_bars=200, test_bars=50, anchored=True ) # In anchored mode, training always starts at 0 assert np.all(r.fold_indices[:, 0] == 0) def test_param_stability_populated(self): c, pg = self._setup() r = walk_forward(c, rsi_strategy, pg, train_bars=200, test_bars=50) assert "timeperiod" in r.param_stability assert "most_chosen" in r.param_stability["timeperiod"] # =========================================================================== # Group 10: monte_carlo() Python API # =========================================================================== class TestMonteCarlo: def _result(self, n: int = 200): rng = np.random.default_rng(77) c = np.cumprod(1 + rng.standard_normal(n) * 0.01) * 100.0 return BacktestEngine().run(c, "rsi_30_70") def test_returns_monte_carlo_result(self): r = self._result() mc = monte_carlo(r, n_sims=100) assert isinstance(mc, MonteCarloResult) def test_equity_curves_shape(self): r = self._result(n=150) mc = monte_carlo(r, n_sims=80) assert mc.equity_curves.shape == (80, 150) def test_confidence_bounds_cover_median(self): r = self._result() mc = monte_carlo(r, n_sims=500, confidence=0.95) assert np.all(mc.confidence_lower <= mc.median_curve + 1e-9) assert np.all(mc.confidence_upper >= mc.median_curve - 1e-9) def test_prob_profit_in_range(self): r = self._result() mc = monte_carlo(r, n_sims=200) assert 0.0 <= mc.prob_profit <= 1.0 def test_accepts_raw_array(self): rng = np.random.default_rng(3) returns = rng.standard_normal(100) * 0.01 mc = monte_carlo(returns, n_sims=50) assert isinstance(mc, MonteCarloResult) def test_seed_reproducibility(self): r = self._result() mc1 = monte_carlo(r, n_sims=50, seed=1) mc2 = monte_carlo(r, n_sims=50, seed=1) npt.assert_array_equal(mc1.equity_curves, mc2.equity_curves) def test_var_is_low_percentile_of_terminal_equity(self): r = self._result() mc = monte_carlo(r, n_sims=1000, confidence=0.95) # VaR = 5th percentile of terminal equity expected_var = float(np.percentile(mc.terminal_equity, 5.0)) assert mc.var == pytest.approx(expected_var, rel=1e-6) # =========================================================================== # Backward compatibility guard # =========================================================================== class TestBackwardCompat: def test_backtest_still_returns_backtest_result(self): rng = np.random.default_rng(0) c = np.cumprod(1 + rng.standard_normal(100) * 0.01) * 100.0 r = backtest(c, strategy="rsi_30_70") assert type(r) is BacktestResult def test_portfolio_backtest_result(self): rng = np.random.default_rng(0) n, k = 100, 3 c2d = np.cumprod(1 + rng.standard_normal((n, k)) * 0.01, axis=0) * 100.0 w2d = np.ones((n, k)) / k r = backtest_portfolio(c2d, w2d) assert isinstance(r, PortfolioBacktestResult) assert r.portfolio_equity.shape == (n,) # =========================================================================== # Sprint 1: Limit orders, time-based exit, pct_range slippage # =========================================================================== class TestLimitOrders: """Tests for limit-price order fill logic in backtest_ohlcv_core.""" def _ohlcv(self): n = 50 rng = np.random.default_rng(7) close = np.cumprod(1 + rng.standard_normal(n) * 0.005) * 100.0 open_ = close * (1 - rng.uniform(0, 0.003, n)) high = close * (1 + rng.uniform(0.002, 0.008, n)) low = close * (1 - rng.uniform(0.002, 0.008, n)) return open_, high, low, close, n def test_limit_nan_behaves_like_market(self): """NaN limit prices should give identical results to no limit array.""" o, h, l, c, n = self._ohlcv() signals = np.where(np.arange(n) % 10 < 5, 1.0, -1.0).astype(np.float64) lp_nan = np.full(n, np.nan) pos_mkt, fp_mkt, _, sr_mkt, eq_mkt = backtest_ohlcv_core(o, h, l, c, signals) pos_lim, fp_lim, _, sr_lim, eq_lim = backtest_ohlcv_core( o, h, l, c, signals, limit_prices=lp_nan ) npt.assert_array_almost_equal(pos_mkt, pos_lim) npt.assert_array_almost_equal(sr_mkt, sr_lim) npt.assert_array_almost_equal(eq_mkt, eq_lim) def test_buy_limit_fills_at_limit_price(self): """Buy limit fills when low <= limit_price and uses limit as fill price.""" n = 10 close = np.full(n, 100.0) open_ = np.full(n, 100.0) high = np.full(n, 102.0) low = np.full(n, 98.0) # Buy signal at bar 0, limit price 99 — low=98 <= 99 so should fill signals = np.zeros(n) signals[0] = 1.0 # want to go long at bar 1 limit_prices = np.full(n, np.nan) limit_prices[0] = 99.0 # limit for bar 1 execution _, fp, _, _, _ = backtest_ohlcv_core( open_, high, low, close, signals, fill_mode="market_close", limit_prices=limit_prices, ) # Bar 1 should have a fill at 99.0 (the limit price) assert fp[1] == pytest.approx(99.0, rel=1e-6) def test_buy_limit_not_hit_no_fill(self): """Buy limit is not filled when low > limit_price.""" n = 10 close = np.full(n, 100.0) open_ = np.full(n, 100.0) high = np.full(n, 102.0) low = np.full(n, 98.0) # low=98 signals = np.zeros(n) signals[0] = 1.0 # go long at bar 1 limit_prices = np.full(n, np.nan) limit_prices[0] = 97.0 # limit=97, but low=98 > 97 → no fill pos, fp, _, _, _ = backtest_ohlcv_core( open_, high, low, close, signals, fill_mode="market_close", limit_prices=limit_prices, ) # Position should stay 0 at bar 1 (limit not hit) assert pos[1] == pytest.approx(0.0) assert np.isnan(fp[1]) def test_sell_limit_fills_when_high_hits(self): """Sell limit fills when high >= limit_price.""" n = 10 close = np.full(n, 100.0) open_ = np.full(n, 100.0) high = np.full(n, 103.0) low = np.full(n, 97.0) signals = np.zeros(n) signals[0] = -1.0 # go short at bar 1 limit_prices = np.full(n, np.nan) limit_prices[0] = 102.0 # high=103 >= 102 → fill _, fp, _, _, _ = backtest_ohlcv_core( open_, high, low, close, signals, fill_mode="market_close", limit_prices=limit_prices, ) assert fp[1] == pytest.approx(102.0, rel=1e-6) def test_engine_with_limit_orders(self): """BacktestEngine.with_limit_orders with NaN limits matches market orders.""" o, h, l, c, n = self._ohlcv() # NaN limit prices = market orders; result must match engine without limit array limit_prices = np.full(n, np.nan) result_mkt = ( BacktestEngine() .with_ohlcv(high=h, low=l, open_=o) .run(c, strategy="sma_crossover", fast=5, slow=20) ) result_lim = ( BacktestEngine() .with_ohlcv(high=h, low=l, open_=o) .with_limit_orders(limit_prices) .run(c, strategy="sma_crossover", fast=5, slow=20) ) assert isinstance(result_lim, AdvancedBacktestResult) npt.assert_array_almost_equal(result_mkt.equity, result_lim.equity) class TestMaxHold: """Tests for time-based exit (max_hold_bars).""" def _flat_ohlcv(self, n=30): close = np.ones(n) * 100.0 open_ = close.copy() high = close * 1.005 low = close * 0.995 signals = np.ones(n) # always long signal return open_, high, low, close, signals def test_position_exits_after_n_bars(self): """Position should be closed after max_hold_bars regardless of signal.""" o, h, l, c, s = self._flat_ohlcv(n=20) max_hold = 5 pos, _, _, _, _ = backtest_ohlcv_core(o, h, l, c, s, max_hold_bars=max_hold) # Find first entry entry_bar = None for i in range(len(pos)): if pos[i] != 0.0: entry_bar = i break assert entry_bar is not None # Position should be 0 at entry_bar + max_hold exit_bar = entry_bar + max_hold if exit_bar < len(pos): assert pos[exit_bar] == pytest.approx(0.0), ( f"Expected exit at bar {exit_bar}, pos={pos[exit_bar]}" ) def test_max_hold_zero_is_disabled(self): """max_hold_bars=0 should not affect behaviour (disabled).""" o, h, l, c, s = self._flat_ohlcv(n=20) pos_no_hold, _, _, _, _ = backtest_ohlcv_core(o, h, l, c, s) pos_hold_0, _, _, _, _ = backtest_ohlcv_core(o, h, l, c, s, max_hold_bars=0) npt.assert_array_almost_equal(pos_no_hold, pos_hold_0) def test_engine_with_max_hold(self): """BacktestEngine.with_max_hold integrates correctly.""" rng = np.random.default_rng(99) n = 100 c = np.cumprod(1 + rng.standard_normal(n) * 0.01) * 100.0 h = c * 1.01 l = c * 0.99 o = c * 1.001 result = ( BacktestEngine() .with_ohlcv(high=h, low=l, open_=o) .with_max_hold(5) .run(c, strategy="rsi_30_70") ) assert isinstance(result, AdvancedBacktestResult) def test_max_hold_stop_takes_priority(self): """A stop-loss that triggers before max_hold should exit early.""" n = 20 close = np.array([100.0] * 5 + [95.0] * 15) # price drops on bar 5 open_ = close.copy() high = close * 1.002 low = np.array([100.0] * 5 + [93.0] * 15) # low hits stop at bar 5 signals = np.ones(n) # always long pos_sl, _, _, _, _ = backtest_ohlcv_core( open_, high, low, close, signals, stop_loss_pct=0.05, max_hold_bars=10, ) pos_hold, _, _, _, _ = backtest_ohlcv_core( open_, high, low, close, signals, stop_loss_pct=0.05, ) # Both should exit around the same time (stop triggers before hold limit) # At least the stop-loss exit should happen — position goes to 0 before bar 10+1 assert any(pos_sl[5:11] == 0.0), ( "Stop-loss should have triggered before max_hold" ) class TestSlippagePctRange: """Tests for pct_range slippage mode.""" def _ohlcv_wide_range(self, n=20): """OHLCV with a wide bar range to make pct_range slippage measurable.""" close = np.full(n, 100.0) open_ = np.full(n, 100.0) high = np.full(n, 110.0) # range = 10 (10%) low = np.full(n, 90.0) signals = np.where(np.arange(n) % 10 < 5, 1.0, -1.0).astype(np.float64) return open_, high, low, close, signals def test_pct_range_more_costly_than_zero_slippage(self): """With wide bar range, pct_range slip should reduce final equity vs no slip.""" o, h, l, c, s = self._ohlcv_wide_range() _, _, _, _, eq_no_slip = backtest_ohlcv_core(o, h, l, c, s) _, _, _, _, eq_pct = backtest_ohlcv_core(o, h, l, c, s, slippage_pct_range=0.10) # pct_range slippage = 0.10 × (110-90)/100 = 0.02 = 200bps per trade assert eq_pct[-1] < eq_no_slip[-1] def test_pct_range_more_costly_than_bps_equivalent(self): """pct_range with wide range should be costlier than modest bps slip.""" o, h, l, c, s = self._ohlcv_wide_range() # bps slip: 5bps = 0.05% of fill, small _, _, _, _, eq_bps = backtest_ohlcv_core(o, h, l, c, s, slippage_bps=5.0) # pct_range: 10% of 20-wide range = 2.0 absolute, or 2% of close=100 _, _, _, _, eq_pct = backtest_ohlcv_core(o, h, l, c, s, slippage_pct_range=0.10) assert eq_pct[-1] < eq_bps[-1] def test_pct_range_zero_equals_no_slippage(self): """slippage_pct_range=0 should give same result as no slippage.""" o, h, l, c, s = self._ohlcv_wide_range() _, _, _, _, eq_base = backtest_ohlcv_core(o, h, l, c, s) _, _, _, _, eq_zero = backtest_ohlcv_core(o, h, l, c, s, slippage_pct_range=0.0) npt.assert_array_almost_equal(eq_base, eq_zero) def test_engine_with_slippage_pct_range(self): """BacktestEngine.with_slippage_pct_range integrates correctly.""" rng = np.random.default_rng(17) n = 80 c = np.cumprod(1 + rng.standard_normal(n) * 0.01) * 100.0 h = c * 1.01 l = c * 0.99 o = c * 1.001 result = ( BacktestEngine() .with_ohlcv(high=h, low=l, open_=o) .with_slippage_pct_range(0.05) .run(c, strategy="sma_crossover", fast=5, slow=20) ) assert isinstance(result, AdvancedBacktestResult) # =========================================================================== # Group 11: Phase 1 Features (spread_bps, breakeven_stop, bracket order priority) # =========================================================================== from ferro_ta._ferro_ta import CommissionModel as RustCommissionModel class TestPhase1Features: """Tests for Phase 1 features: spread_bps, breakeven_pct, bracket order priority.""" def test_spread_bps_increases_cost(self): """CommissionModel with spread_bps=10 should produce lower equity than spread_bps=0.""" rng = np.random.default_rng(99) n = 200 c = np.cumprod(1 + rng.standard_normal(n) * 0.01) * 100.0 h = c * 1.005 l = c * 0.995 o = c * 0.999 signals = np.where(np.arange(n) % 20 < 10, 1.0, 0.0).astype(np.float64) # Build a commission model with spread_bps=0 cm_no_spread = RustCommissionModel() cm_no_spread.spread_bps = 0.0 # Build a commission model with spread_bps=10 cm_with_spread = RustCommissionModel() cm_with_spread.spread_bps = 10.0 _, _, _, _, eq_no_spread = backtest_ohlcv_core( o, h, l, c, signals, commission=cm_no_spread ) _, _, _, _, eq_with_spread = backtest_ohlcv_core( o, h, l, c, signals, commission=cm_with_spread ) # Spread adds cost on each trade leg → should produce lower or equal final equity assert eq_with_spread[-1] <= eq_no_spread[-1], ( f"spread equity {eq_with_spread[-1]:.6f} should be <= no-spread equity {eq_no_spread[-1]:.6f}" ) def test_spread_bps_getter_setter(self): """CommissionModel spread_bps getter/setter round-trip works correctly.""" m = RustCommissionModel() assert m.spread_bps == 0.0 m.spread_bps = 5.0 assert m.spread_bps == pytest.approx(5.0) def test_spread_bps_total_cost(self): """CommissionModel.total_cost includes spread cost at correct magnitude.""" m = RustCommissionModel() m.spread_bps = 20.0 # 20 bps total round-trip = 10 bps each leg trade_value = 100_000.0 cost = m.total_cost(trade_value, 1.0, True) # Expected: 10 bps = 0.001 * 100_000 = 100 per leg assert cost == pytest.approx(100.0, rel=1e-6) def test_breakeven_stop_prevents_loss(self): """With breakeven_pct=0.02, after price rises 3% then falls, exit should be near entry.""" # Build synthetic data: entry at bar 1, then price rises 3%, then falls below entry # Bar layout: [100, 103, 103, 101, 99, 99, 99, 99, 99] # We want a long signal from bar 0 onwards n = 20 # Create price data: starts at 100, rises to 103 at bar 3, then drops to 97 close = np.array([100.0] * 3 + [103.0] * 3 + [97.0] * (n - 6), dtype=np.float64) open_ = close.copy() high = close * 1.002 low = close * 0.998 # Set high of bar 3 to clearly trigger breakeven (>= 103 = entry * 1.03) # entry happens at bar 1 (open of bar 1 = 100), so entry_price ≈ 100 # breakeven triggers when h >= 100 * 1.02 = 102 → triggers at bar 3 (close=103, high≥103) high[3] = 103.5 # clearly above 102 (entry * 1.02) # At bar 6, low drops below entry (100), breakeven stop should trigger low[6] = 99.0 # below entry price 100 → breakeven stop fires signals = np.ones(n, dtype=np.float64) # always long _, fp, _, _, _ = backtest_ohlcv_core( open_, high, low, close, signals, stop_loss_pct=0.0, breakeven_pct=0.02, ) # Find first non-NaN fill price after the entry bar (entry at bar 1) # Breakeven exit should happen at or near entry price (100), not at a big loss exit_fps = fp[~np.isnan(fp)] # The breakeven stop exit should be at entry_price (~100), not at 97 or lower # Entry fill is at open of bar 1 = 100.0 # After breakeven activates, stop = entry (~100). So exit fill should be ~100 assert len(exit_fps) >= 1 # The exit fill from breakeven should be close to entry price (within 1%) # (first fill = entry, subsequent fills = exits) if len(exit_fps) >= 2: breakeven_exit = exit_fps[1] assert breakeven_exit >= 99.0, ( f"breakeven exit {breakeven_exit} should be >= 99 (near entry 100)" ) def test_bracket_order_tp_fires_before_sl(self): """When both SL and TP are breached in same bar, and open is near TP → TP fires.""" # Long trade: entry at price 100 # Bar where both trigger: open=109 (very close to TP=110), high=112, low=90 # SL = 100*(1-0.10) = 90, TP = 100*(1+0.10) = 110 # open=109 is closer to TP=110 (dist=1) than to SL=90 (dist=19) → TP fires entry_price = 100.0 close = np.array( [entry_price, entry_price, entry_price, 108.0, 108.0], dtype=np.float64 ) open_ = np.array( [entry_price, entry_price, entry_price, 109.0, 108.0], dtype=np.float64 ) high = np.array( [entry_price, entry_price, entry_price, 112.0, 108.0], dtype=np.float64 ) low = np.array( [entry_price, entry_price, entry_price, 88.0, 108.0], dtype=np.float64 ) signals = np.array([0.0, 1.0, 1.0, 1.0, 0.0], dtype=np.float64) _, fp, _, sr, _ = backtest_ohlcv_core( open_, high, low, close, signals, stop_loss_pct=0.10, take_profit_pct=0.10, ) # Bar 3 is where both trigger. TP=110. SL=90. Open=109 → TP fires. # fill price at bar 3 should be ~110 (TP), not 90 (SL) assert not np.isnan(fp[3]), "Expected a fill at bar 3" tp_level = entry_price * 1.10 # 110 assert fp[3] == pytest.approx(tp_level, rel=1e-6), ( f"Expected TP fill at ~{tp_level}, got {fp[3]}" ) def test_bracket_order_sl_fires_before_tp(self): """When both SL and TP are breached in same bar, and open is near SL → SL fires.""" # Long trade: entry at 100 # Bar where both trigger: open=91 (very close to SL=90), high=112, low=88 # SL=90, TP=110. open=91 is closer to SL=90 (dist=1) than to TP=110 (dist=19) → SL fires entry_price = 100.0 close = np.array( [entry_price, entry_price, entry_price, 95.0, 95.0], dtype=np.float64 ) open_ = np.array( [entry_price, entry_price, entry_price, 91.0, 95.0], dtype=np.float64 ) high = np.array( [entry_price, entry_price, entry_price, 112.0, 95.0], dtype=np.float64 ) low = np.array( [entry_price, entry_price, entry_price, 88.0, 95.0], dtype=np.float64 ) signals = np.array([0.0, 1.0, 1.0, 1.0, 0.0], dtype=np.float64) _, fp, _, sr, _ = backtest_ohlcv_core( open_, high, low, close, signals, stop_loss_pct=0.10, take_profit_pct=0.10, ) # Bar 3: both SL(90) and TP(110) are triggered. open=91 is close to SL → SL fires. assert not np.isnan(fp[3]), "Expected a fill at bar 3" sl_level = entry_price * 0.90 # 90 assert fp[3] == pytest.approx(sl_level, rel=1e-6), ( f"Expected SL fill at ~{sl_level}, got {fp[3]}" ) def test_breakeven_engine_integration(self): """BacktestEngine.with_breakeven_stop integrates correctly.""" rng = np.random.default_rng(77) n = 150 c = np.cumprod(1 + rng.standard_normal(n) * 0.01) * 100.0 h = c * 1.01 l = c * 0.99 o = c * 0.999 result = ( BacktestEngine() .with_ohlcv(high=h, low=l, open_=o) .with_breakeven_stop(0.02) .run(c, strategy="sma_crossover", fast=5, slow=20) ) assert isinstance(result, AdvancedBacktestResult) assert np.all(np.isfinite(result.equity)) # =========================================================================== # Phase 2: Portfolio & Risk Features # =========================================================================== class TestPhase2Features: """Tests for Phase 2: short borrow cost, margin/leverage, circuit breakers, and portfolio constraints.""" # ----------------------------------------------------------------------- # Helper: synthetic OHLCV with controllable direction # ----------------------------------------------------------------------- def _make_short_ohlcv(self, n: int = 100, seed: int = 7) -> tuple: """Produce OHLCV where the price trends downward (good for shorts).""" rng = np.random.default_rng(seed) # Steady downtrend close = 100.0 * np.cumprod(1 - np.abs(rng.standard_normal(n)) * 0.005) open_ = close * (1 + rng.uniform(-0.002, 0.002, n)) high = np.maximum(open_, close) * (1 + rng.uniform(0, 0.003, n)) low = np.minimum(open_, close) * (1 - rng.uniform(0, 0.003, n)) # Always short signals = np.full(n, -1.0, dtype=np.float64) return open_, high, low, close, signals # ----------------------------------------------------------------------- # 1. Short borrow cost # ----------------------------------------------------------------------- def test_short_borrow_cost_reduces_equity(self): """Short position with short_borrow_rate_annual=0.10 should produce lower final equity than the same run with no borrow cost.""" from ferro_ta._ferro_ta import CommissionModel o, h, l, c, signals = self._make_short_ohlcv(n=252) # Commission model without borrow cost cm_no_borrow = CommissionModel() # Commission model with 10% annual borrow cost cm_with_borrow = CommissionModel() cm_with_borrow.short_borrow_rate_annual = 0.10 _, _, _, _, eq_no_borrow = backtest_ohlcv_core( o, h, l, c, signals, commission=cm_no_borrow ) _, _, _, _, eq_with_borrow = backtest_ohlcv_core( o, h, l, c, signals, commission=cm_with_borrow ) # With borrow cost, final equity must be strictly lower assert float(eq_with_borrow[-1]) < float(eq_no_borrow[-1]), ( f"Expected borrow-cost equity {eq_with_borrow[-1]:.6f} < " f"no-borrow equity {eq_no_borrow[-1]:.6f}" ) def test_short_borrow_cost_getter_setter(self): """CommissionModel.short_borrow_rate_annual getter/setter works.""" from ferro_ta._ferro_ta import CommissionModel cm = CommissionModel() assert cm.short_borrow_rate_annual == pytest.approx(0.0) cm.short_borrow_rate_annual = 0.05 assert cm.short_borrow_rate_annual == pytest.approx(0.05) def test_short_borrow_zero_rate_no_effect(self): """With short_borrow_rate_annual=0, borrow cost should not affect equity.""" from ferro_ta._ferro_ta import CommissionModel o, h, l, c, signals = self._make_short_ohlcv(n=50) cm_zero = CommissionModel() cm_zero.short_borrow_rate_annual = 0.0 _, _, _, sr1, eq1 = backtest_ohlcv_core(o, h, l, c, signals) _, _, _, sr2, eq2 = backtest_ohlcv_core(o, h, l, c, signals, commission=cm_zero) npt.assert_allclose(eq1, eq2, rtol=1e-10) def test_short_borrow_engine_integration(self): """BacktestEngine with commission model including short_borrow_rate_annual runs.""" from ferro_ta._ferro_ta import CommissionModel o, h, l, c, sigs = self._make_short_ohlcv(n=80) cm = CommissionModel() cm.short_borrow_rate_annual = 0.08 result = ( BacktestEngine() .with_ohlcv(high=h, low=l, open_=o) .with_commission_model(cm) .run(c, lambda x: np.full(len(x), -1.0)) ) assert isinstance(result, AdvancedBacktestResult) assert np.all(np.isfinite(result.equity)) # ----------------------------------------------------------------------- # 2. Margin call force-close # ----------------------------------------------------------------------- def test_margin_call_force_closes_position(self): """A declining price sequence triggers a margin call and force-closes the long.""" n = 20 # Price drops sharply — enough to trigger a margin call on a long open_ = np.ones(n) * 100.0 high = np.ones(n) * 101.0 low = np.ones(n) * 99.0 close = np.ones(n) * 100.0 # After bar 5, price tanks sharply every bar for i in range(5, n): drop = 0.30 # 30% per bar — guaranteed to exceed margin open_[i] = open_[i - 1] * (1 - drop) high[i] = open_[i] * 1.001 low[i] = open_[i] * 0.999 close[i] = open_[i] # Always long signals = np.ones(n, dtype=np.float64) # margin_ratio=0.2 means 20% margin (5x leverage) # margin_call_pct=0.5 means call when equity hits 50% of initial margin _, _, _, sr_margin, eq_margin = backtest_ohlcv_core( open_, high, low, close, signals, margin_ratio=0.2, margin_call_pct=0.5, ) _, _, _, sr_no_margin, eq_no_margin = backtest_ohlcv_core( open_, high, low, close, signals, ) # Margin call should cause a forced exit, resulting in different equity # (the margin version stops losses earlier) assert not np.allclose(eq_margin, eq_no_margin), ( "Expected margin call to alter equity curve" ) def test_margin_disabled_when_ratio_zero(self): """margin_ratio=0 should behave identically to not passing the parameter.""" o, h, l, c, signals = _make_ohlcv(n=80) _, _, _, _, eq_default = backtest_ohlcv_core(o, h, l, c, signals) _, _, _, _, eq_zero_margin = backtest_ohlcv_core( o, h, l, c, signals, margin_ratio=0.0 ) npt.assert_allclose(eq_default, eq_zero_margin, rtol=1e-10) def test_margin_engine_builder(self): """BacktestEngine.with_leverage builder sets parameters without error.""" o, h, l, c, _ = _make_ohlcv(n=60) result = ( BacktestEngine() .with_ohlcv(high=h, low=l, open_=o) .with_leverage(margin_ratio=0.2, margin_call_pct=0.5) .run(c, lambda x: np.ones(len(x))) ) assert isinstance(result, AdvancedBacktestResult) assert np.all(np.isfinite(result.equity)) # ----------------------------------------------------------------------- # 3. Total loss limit (circuit breaker) # ----------------------------------------------------------------------- def test_total_loss_limit_halts_trading(self): """total_loss_limit=0.10 should halt trading once equity drops 10%.""" n = 100 # Construct a losing price sequence: steady decline close = 100.0 * np.cumprod(np.full(n, 0.99)) # -1% per bar open_ = close * 1.001 high = close * 1.005 low = close * 0.995 # Always long (so position loses money as price falls) signals = np.ones(n, dtype=np.float64) pos_with_limit, _, _, _, eq_with_limit = backtest_ohlcv_core( open_, high, low, close, signals, total_loss_limit=0.10, ) pos_no_limit, _, _, _, eq_no_limit = backtest_ohlcv_core( open_, high, low, close, signals, ) # After circuit break the position should be 0 # Check that at some point positions go to 0 in the limited version # while the unlimited version stays long assert np.any(pos_with_limit == 0.0), ( "Expected some bars with no position after circuit break" ) # Unlimited version should stay long throughout (except bar 0) assert np.all(pos_no_limit[1:] == 1.0), "No-limit should stay long" def test_total_loss_limit_does_not_trip_with_no_loss(self): """total_loss_limit does not trip on a profitable sequence.""" n = 60 close = 100.0 * np.cumprod(np.full(n, 1.005)) # +0.5% per bar open_ = close * 0.999 high = close * 1.003 low = close * 0.997 signals = np.ones(n, dtype=np.float64) pos, _, _, _, _ = backtest_ohlcv_core( open_, high, low, close, signals, total_loss_limit=0.20, ) # No circuit break should fire; position stays long assert np.all(pos[1:] == 1.0) # ----------------------------------------------------------------------- # 4. Daily (per-bar) loss limit circuit breaker # ----------------------------------------------------------------------- def test_daily_loss_limit_halts_after_large_bar_loss(self): """A single large losing bar triggers the daily_loss_limit circuit breaker.""" n = 30 close = np.ones(n) * 100.0 open_ = np.ones(n) * 100.0 high = np.ones(n) * 101.0 low = np.ones(n) * 99.0 # Create one very large losing bar at bar 10 (price drops 15%) # Strategy is long, so this is a large loss crash_bar = 10 close[crash_bar] = close[crash_bar - 1] * 0.85 open_[crash_bar] = close[crash_bar - 1] * 0.86 high[crash_bar] = open_[crash_bar] * 1.001 low[crash_bar] = close[crash_bar] * 0.999 signals = np.ones(n, dtype=np.float64) pos, _, _, sr, _ = backtest_ohlcv_core( open_, high, low, close, signals, daily_loss_limit=0.05, # 5% per-bar loss limit ) # After the crash bar, circuit breaker should fire and position should go to 0 # Check bars after crash_bar+1 have position 0 assert np.any(pos[crash_bar + 1 :] == 0.0), ( "Expected circuit breaker to zero out position after crash bar" ) def test_daily_loss_limit_zero_is_disabled(self): """daily_loss_limit=0 (default) should not change behavior.""" o, h, l, c, signals = _make_ohlcv(n=80) _, _, _, _, eq_default = backtest_ohlcv_core(o, h, l, c, signals) _, _, _, _, eq_zero_limit = backtest_ohlcv_core( o, h, l, c, signals, daily_loss_limit=0.0 ) npt.assert_allclose(eq_default, eq_zero_limit, rtol=1e-10) def test_loss_limits_engine_builder(self): """BacktestEngine.with_loss_limits builder sets parameters.""" o, h, l, c, _ = _make_ohlcv(n=80) result = ( BacktestEngine() .with_ohlcv(high=h, low=l, open_=o) .with_loss_limits(daily=0.05, total=0.20) .run(c, strategy="sma_crossover") ) assert isinstance(result, AdvancedBacktestResult) assert np.all(np.isfinite(result.equity)) # ----------------------------------------------------------------------- # 5. Portfolio constraints # ----------------------------------------------------------------------- def test_portfolio_max_asset_weight_clamps_signal(self): """max_asset_weight=0.5 should clamp signals from ±1 to ±0.5.""" rng = np.random.default_rng(99) n_bars, n_assets = 100, 3 close_2d = ( np.cumprod(1 + rng.standard_normal((n_bars, n_assets)) * 0.01, axis=0) * 100.0 ) # Alternating ±1 signals — shape (n_bars, n_assets) row_flags = (np.arange(n_bars) % 10 < 5)[:, None] # (n, 1) weights_2d = np.where(np.tile(row_flags, (1, n_assets)), 1.0, -1.0).astype( np.float64 ) # Run without constraint (unit signals) asset_ret_unconstrained, port_ret_unconstrained, _ = backtest_multi_asset_core( np.ascontiguousarray(close_2d), np.ascontiguousarray(weights_2d), max_asset_weight=1.0, ) # Run with max_asset_weight=0.5 asset_ret_constrained, port_ret_constrained, _ = backtest_multi_asset_core( np.ascontiguousarray(close_2d), np.ascontiguousarray(weights_2d), max_asset_weight=0.5, ) # Constrained returns should have smaller magnitude assert np.abs(port_ret_constrained).sum() < np.abs( port_ret_unconstrained ).sum() or np.allclose( np.abs(port_ret_constrained).sum(), np.abs(port_ret_unconstrained).sum() * 0.5, rtol=0.05, ), "max_asset_weight=0.5 should reduce absolute returns by ~50%" def test_portfolio_max_gross_exposure_constrains_sum(self): """max_gross_exposure=1.0 should limit total abs(weights).""" rng = np.random.default_rng(55) n_bars, n_assets = 80, 4 close_2d = ( np.cumprod(1 + rng.standard_normal((n_bars, n_assets)) * 0.01, axis=0) * 100.0 ) # Always long all assets = gross exposure of 4.0 weights_2d = np.ones((n_bars, n_assets), dtype=np.float64) # With max_gross_exposure=1.0, total abs weight should be normalized to 1 ar_constrained, pr_constrained, _ = backtest_multi_asset_core( np.ascontiguousarray(close_2d), np.ascontiguousarray(weights_2d), max_gross_exposure=1.0, ) ar_unconstrained, pr_unconstrained, _ = backtest_multi_asset_core( np.ascontiguousarray(close_2d), np.ascontiguousarray(weights_2d), ) # Constrained portfolio should have ~1/4 the returns magnitude ratio = np.abs(pr_constrained).sum() / (np.abs(pr_unconstrained).sum() + 1e-12) assert ratio < 0.5, ( f"Expected constrained to be much smaller, got ratio={ratio:.3f}" ) def test_portfolio_constraints_engine_builder(self): """BacktestEngine.with_portfolio_constraints stores the parameters.""" engine = BacktestEngine().with_portfolio_constraints( max_asset_weight=0.3, max_gross_exposure=1.5, max_net_exposure=0.5, ) assert engine._max_asset_weight == pytest.approx(0.3) assert engine._max_gross_exposure == pytest.approx(1.5) assert engine._max_net_exposure == pytest.approx(0.5) def test_backtest_portfolio_with_constraints(self): """backtest_portfolio accepts portfolio constraint kwargs.""" from ferro_ta.analysis.backtest import backtest_portfolio rng = np.random.default_rng(11) n_bars, n_assets = 60, 2 close_2d = ( np.cumprod(1 + rng.standard_normal((n_bars, n_assets)) * 0.01, axis=0) * 100.0 ) row_flags = (np.arange(n_bars) % 10 < 5)[:, None] weights_2d = np.where(np.tile(row_flags, (1, n_assets)), 1.0, -1.0).astype( np.float64 ) result = backtest_portfolio( close_2d, weights_2d, max_asset_weight=0.5, max_gross_exposure=0.8, ) assert isinstance(result, PortfolioBacktestResult) assert np.all(np.isfinite(result.portfolio_equity)) # =========================================================================== # Phase 3: Data & UX features # =========================================================================== class TestPhase3Features: """Tests for Phase 3: resample, adjust, multitf, and plot modules.""" # ----------------------------------------------------------------------- # Helpers # ----------------------------------------------------------------------- def _make_ohlcv(self, n=100, seed=42): rng = np.random.default_rng(seed) close = np.cumprod(1 + rng.standard_normal(n) * 0.005) * 100.0 open_ = close * (1 - rng.uniform(0, 0.002, n)) high = close * (1 + rng.uniform(0.001, 0.006, n)) low = close * (1 - rng.uniform(0.001, 0.006, n)) volume = rng.uniform(1_000, 10_000, n) return open_, high, low, close, volume # ----------------------------------------------------------------------- # 1. resample_ohlcv — factor=4, 20 bars → 5 coarse bars # ----------------------------------------------------------------------- def test_resample_ohlcv_factor4(self): from ferro_ta.analysis.resample import resample_ohlcv o, h, l, c, v = self._make_ohlcv(n=20) co, ch, cl, cc, cv = resample_ohlcv(o, h, l, c, v, factor=4) assert co.shape == (5,) assert ch.shape == (5,) assert cl.shape == (5,) assert cc.shape == (5,) assert cv.shape == (5,) # open = first bar of each group for i in range(5): assert co[i] == pytest.approx(o[i * 4]) # high = max of group for i in range(5): assert ch[i] == pytest.approx(h[i * 4 : i * 4 + 4].max()) # low = min of group for i in range(5): assert cl[i] == pytest.approx(l[i * 4 : i * 4 + 4].min()) # close = last bar of group for i in range(5): assert cc[i] == pytest.approx(c[i * 4 + 3]) # volume = sum of group for i in range(5): assert cv[i] == pytest.approx(v[i * 4 : i * 4 + 4].sum()) # ----------------------------------------------------------------------- # 2. resample_ohlcv — non-divisible length: 22 bars, factor=4 → 5 coarse bars # ----------------------------------------------------------------------- def test_resample_ohlcv_non_divisible(self): from ferro_ta.analysis.resample import resample_ohlcv o, h, l, c, v = self._make_ohlcv(n=22) co, ch, cl, cc, cv = resample_ohlcv(o, h, l, c, v, factor=4) # 22 // 4 = 5 complete bars, last 2 fine bars are dropped assert len(co) == 5 assert len(ch) == 5 # ----------------------------------------------------------------------- # 3. align_to_coarse — roundtrip test # ----------------------------------------------------------------------- def test_align_to_coarse_roundtrip(self): from ferro_ta.analysis.resample import align_to_coarse coarse = np.array([10.0, 20.0, 30.0, 40.0, 50.0]) factor = 4 n_fine = 20 fine = align_to_coarse(coarse, factor, n_fine) assert len(fine) == n_fine for i, val in enumerate(coarse): expected = np.full(factor, val) npt.assert_array_equal(fine[i * factor : i * factor + factor], expected) # ----------------------------------------------------------------------- # 4. adjust_for_splits — 2-for-1 split at bar 50 in 100-bar series # ----------------------------------------------------------------------- def test_adjust_for_splits_halves_historical(self): from ferro_ta.analysis.adjust import adjust_for_splits close = np.ones(100) * 100.0 adjusted = adjust_for_splits(close, split_factors=[2.0], split_indices=[50]) # Prices before split (bars 0-49) should be halved npt.assert_array_almost_equal(adjusted[:50], np.full(50, 50.0)) # Prices from split onwards unchanged npt.assert_array_almost_equal(adjusted[50:], np.full(50, 100.0)) # ----------------------------------------------------------------------- # 5. adjust_for_dividends — dividend at bar 50; prices before reduced # ----------------------------------------------------------------------- def test_adjust_for_dividends_reduces_historical(self): from ferro_ta.analysis.adjust import adjust_for_dividends close = np.ones(100) * 100.0 # bar 49 close = 100.0, dividend = 5.0 → factor = 95/100 = 0.95 adjusted = adjust_for_dividends(close, dividends=[5.0], ex_date_indices=[50]) # Prices before ex-date should be scaled by 0.95 expected_factor = (100.0 - 5.0) / 100.0 npt.assert_array_almost_equal( adjusted[:50], np.full(50, 100.0 * expected_factor) ) # Prices from ex-date onwards unchanged npt.assert_array_almost_equal(adjusted[50:], np.full(50, 100.0)) # ----------------------------------------------------------------------- # 6. adjust_ohlcv — volume doubles on 2-for-1 split (inverse adjustment) # ----------------------------------------------------------------------- def test_adjust_ohlcv_volume_increases_on_split(self): from ferro_ta.analysis.adjust import adjust_ohlcv n = 100 close = np.ones(n) * 100.0 open_ = close.copy() high = close.copy() low = close.copy() volume = np.ones(n) * 1000.0 ao, ah, al, ac, av = adjust_ohlcv( open_, high, low, close, volume, split_factors=[2.0], split_indices=[50], ) # Volume before the split is multiplied by factor (2x) — more shares pre-split npt.assert_array_almost_equal(av[:50], np.full(50, 2000.0)) # Volume at or after split unchanged npt.assert_array_almost_equal(av[50:], np.full(50, 1000.0)) # Prices before split halved npt.assert_array_almost_equal(ac[:50], np.full(50, 50.0)) npt.assert_array_almost_equal(ac[50:], np.full(50, 100.0)) # ----------------------------------------------------------------------- # 7. MultiTimeframeEngine — runs on 200 fine bars, returns valid result # ----------------------------------------------------------------------- def test_multitf_engine_runs(self): from ferro_ta.analysis.multitf import MultiTimeframeEngine rng = np.random.default_rng(99) n_fine = 200 close_fine = np.cumprod(1 + rng.standard_normal(n_fine) * 0.005) * 100.0 result = ( MultiTimeframeEngine(factor=4) .with_htf_strategy("rsi_30_70") .run(close_fine) ) assert isinstance(result, AdvancedBacktestResult) assert len(result.equity) == n_fine assert np.all(np.isfinite(result.equity)) assert result.equity[0] == pytest.approx(1.0, rel=1e-6) # ----------------------------------------------------------------------- # 8. plot_backtest — returns a plotly Figure (skip if plotly not installed) # ----------------------------------------------------------------------- def test_plot_backtest_returns_figure(self): pytest.importorskip("plotly", reason="plotly not installed") from plotly.graph_objects import Figure from ferro_ta.analysis.plot import plot_backtest rng = np.random.default_rng(7) n = 100 close = np.cumprod(1 + rng.standard_normal(n) * 0.005) * 100.0 high = close * 1.01 low = close * 0.99 open_ = close * 0.999 result = ( BacktestEngine() .with_ohlcv(high=high, low=low, open_=open_) .run(close, strategy="rsi_30_70") ) fig = plot_backtest(result, show=False, return_fig=True) assert isinstance(fig, Figure) # =========================================================================== # Phase 4: Regime Detection, Portfolio Optimization, PaperTrader # =========================================================================== class TestPhase4Features: """Tests for Phase 4 differentiation features.""" # ----------------------------------------------------------------------- # Helpers # ----------------------------------------------------------------------- def _make_close(self, n: int = 300, seed: int = 77) -> np.ndarray: rng = np.random.default_rng(seed) return np.cumprod(1 + rng.standard_normal(n) * 0.01) * 100.0 def _make_ohlcv_local(self, n: int = 300, seed: int = 77): rng = np.random.default_rng(seed) close = np.cumprod(1 + rng.standard_normal(n) * 0.01) * 100.0 open_ = close * (1 - rng.uniform(0, 0.005, n)) high = close * (1 + rng.uniform(0, 0.01, n)) low = close * (1 - rng.uniform(0, 0.01, n)) return open_, high, low, close # ----------------------------------------------------------------------- # 1. detect_volatility_regime # ----------------------------------------------------------------------- def test_volatility_regime_labels_three_states(self): from ferro_ta.analysis.regime import detect_volatility_regime close = self._make_close(300) labels = detect_volatility_regime(close, window=20, n_regimes=3) assert labels.shape == (300,) valid_values = {-1, 0, 1, 2} assert set(np.unique(labels)).issubset(valid_values) # Some valid (non-warmup) bars should be labeled assert np.any(labels >= 0) # ----------------------------------------------------------------------- # 2. detect_trend_regime # ----------------------------------------------------------------------- def test_trend_regime_bull_bear(self): from ferro_ta.analysis.regime import detect_trend_regime # Uptrend: price steadily rising n = 300 close_up = np.linspace(100, 200, n) labels_up = detect_trend_regime(close_up, fast=10, slow=50) valid = labels_up[labels_up != 0] assert len(valid) > 0, "Expected some labeled bars after warmup" # Most valid bars should be bull (1) bull_frac = (valid == 1).sum() / len(valid) assert bull_frac > 0.5, ( f"Expected mostly bull bars in uptrend, got {bull_frac:.2%}" ) # Downtrend: price steadily declining close_dn = np.linspace(200, 100, n) labels_dn = detect_trend_regime(close_dn, fast=10, slow=50) valid_dn = labels_dn[labels_dn != 0] assert len(valid_dn) > 0 bear_frac = (valid_dn == -1).sum() / len(valid_dn) assert bear_frac > 0.5, ( f"Expected mostly bear bars in downtrend, got {bear_frac:.2%}" ) # ----------------------------------------------------------------------- # 3. detect_combined_regime # ----------------------------------------------------------------------- def test_combined_regime_states(self): from ferro_ta.analysis.regime import detect_combined_regime close = self._make_close(500) labels = detect_combined_regime(close, vol_window=20, fast=20, slow=50) assert labels.shape == (500,) valid_values = {-1, 0, 1, 2, 3, 4, 5} assert set(np.unique(labels)).issubset(valid_values) # ----------------------------------------------------------------------- # 4. RegimeFilter # ----------------------------------------------------------------------- def test_regime_filter_zeros_disallowed(self): from ferro_ta.analysis.regime import RegimeFilter, detect_combined_regime n = 500 close = self._make_close(n) signals = np.ones(n) # Only allow regime 0 (bull + low vol) rf = RegimeFilter(allowed_regimes=[0], vol_window=20, fast=20, slow=50) filtered = rf.filter(signals, close) regimes = detect_combined_regime(close, vol_window=20, fast=20, slow=50) # Bars NOT in regime 0 should have filtered signal = 0 disallowed_mask = regimes != 0 assert np.all(filtered[disallowed_mask] == 0.0) # Bars in regime 0 should retain their signal allowed_mask = regimes == 0 if np.any(allowed_mask): assert np.all(filtered[allowed_mask] == 1.0) # ----------------------------------------------------------------------- # 5. mean_variance_optimize # ----------------------------------------------------------------------- def test_mean_variance_weights_sum_to_one(self): pytest.importorskip("scipy", reason="scipy not installed") from ferro_ta.analysis.optimize import mean_variance_optimize rng = np.random.default_rng(0) returns = rng.standard_normal((252, 4)) * 0.01 w = mean_variance_optimize(returns) assert w.shape == (4,) assert float(np.sum(w)) == pytest.approx(1.0, abs=1e-6) assert np.all(w >= -1e-9), "Weights should be non-negative (no short)" # ----------------------------------------------------------------------- # 6. risk_parity_optimize # ----------------------------------------------------------------------- def test_risk_parity_weights_sum_to_one(self): pytest.importorskip("scipy", reason="scipy not installed") from ferro_ta.analysis.optimize import risk_parity_optimize rng = np.random.default_rng(1) returns = rng.standard_normal((252, 3)) * 0.01 w = risk_parity_optimize(returns) assert w.shape == (3,) assert float(np.sum(w)) == pytest.approx(1.0, abs=1e-6) assert np.all(w >= 0.0) # ----------------------------------------------------------------------- # 7. max_sharpe_optimize # ----------------------------------------------------------------------- def test_max_sharpe_weights_sum_to_one(self): pytest.importorskip("scipy", reason="scipy not installed") from ferro_ta.analysis.optimize import max_sharpe_optimize rng = np.random.default_rng(2) returns = rng.standard_normal((252, 5)) * 0.01 w = max_sharpe_optimize(returns) assert w.shape == (5,) assert float(np.sum(w)) == pytest.approx(1.0, abs=1e-6) assert np.all(w >= -1e-9) # ----------------------------------------------------------------------- # 8. PortfolioOptimizer fluent builder # ----------------------------------------------------------------------- def test_portfolio_optimizer_fluent(self): pytest.importorskip("scipy", reason="scipy not installed") from ferro_ta.analysis.optimize import PortfolioOptimizer rng = np.random.default_rng(3) returns = rng.standard_normal((252, 3)) * 0.01 for method in ("min_variance", "risk_parity", "max_sharpe"): w = ( PortfolioOptimizer() .with_method(method) .with_lookback(100) .optimize(returns) ) assert w.shape == (3,) assert float(np.sum(w)) == pytest.approx(1.0, abs=1e-6) # ----------------------------------------------------------------------- # 9. PaperTrader: basic fills # ----------------------------------------------------------------------- def test_paper_trader_fills_on_signal(self): from ferro_ta.analysis.live import PaperTrader rng = np.random.default_rng(10) n = 20 close = np.cumprod(1 + rng.standard_normal(n) * 0.005) * 100.0 open_ = close * (1 - rng.uniform(0, 0.003, n)) high = close * (1 + rng.uniform(0.001, 0.005, n)) low = close * (1 - rng.uniform(0.001, 0.005, n)) trader = PaperTrader(initial_capital=100_000) signals = np.where(np.arange(n) % 6 < 3, 1.0, -1.0).astype(float) results = [] for i in range(n): r = trader.on_bar(open_[i], high[i], low[i], close[i], signals[i]) results.append(r) # Should have produced at least one fill after first bar fills = [r for r in results if r.filled] assert len(fills) > 0 # Equity curve length should match bars assert len(trader.equity_curve) == n # Final equity should be finite assert math.isfinite(trader.equity) # ----------------------------------------------------------------------- # 10. PaperTrader: stop-loss triggers # ----------------------------------------------------------------------- def test_paper_trader_stop_loss_triggers(self): from ferro_ta.analysis.live import PaperTrader # Price rises on entry then falls sharply — SL should trigger n = 20 close = np.array( [100.0] * 5 + [98.0, 96.0, 94.0, 92.0, 90.0] # declining + [88.0, 86.0, 84.0, 82.0, 80.0, 78.0, 76.0, 74.0, 72.0, 70.0], dtype=float, ) open_ = close * 1.001 high = close * 1.005 low = close * 0.99 # Low drops to trigger SL sl_pct = 0.03 # 3% stop-loss trader = PaperTrader(initial_capital=100_000, stop_loss_pct=sl_pct) # Signal: go long on bar 0 signals = np.zeros(n) signals[0] = 1.0 # enter long for i in range(n): trader.on_bar(open_[i], high[i], low[i], close[i], signals[i]) # With 3% SL and price dropping >3% below entry, we expect a trade to close # Final position should be 0 (SL triggered exit) assert trader.position == 0.0 or len(trader.trades) > 0 # ----------------------------------------------------------------------- # 11. PaperTrader: reset clears state # ----------------------------------------------------------------------- def test_paper_trader_reset_clears_state(self): from ferro_ta.analysis.live import PaperTrader rng = np.random.default_rng(20) n = 30 close = np.cumprod(1 + rng.standard_normal(n) * 0.01) * 100.0 open_ = close * 0.999 high = close * 1.01 low = close * 0.99 signals = np.where(np.arange(n) % 10 < 5, 1.0, -1.0).astype(float) trader = PaperTrader(initial_capital=50_000) for i in range(n): trader.on_bar(open_[i], high[i], low[i], close[i], signals[i]) assert len(trader.equity_curve) > 0 trader.reset() assert trader.position == 0.0 assert trader.equity == pytest.approx(1.0) assert len(trader.trades) == 0 assert len(trader.equity_curve) == 0 assert trader.equity_abs == pytest.approx(50_000.0) # ----------------------------------------------------------------------- # 12. PaperTrader equity matches backtest_ohlcv_core # ----------------------------------------------------------------------- def test_paper_trader_equity_matches_backtest(self): from ferro_ta.analysis.live import PaperTrader rng = np.random.default_rng(42) n = 50 close = np.cumprod(1 + rng.standard_normal(n) * 0.005) * 100.0 open_ = close * (1 - rng.uniform(0, 0.003, n)) high = close * (1 + rng.uniform(0.001, 0.005, n)) low = close * (1 - rng.uniform(0.001, 0.005, n)) signals = np.where(np.arange(n) % 10 < 5, 1.0, -1.0).astype(np.float64) # Vectorized Rust engine _, _, _, _, eq_rust = backtest_ohlcv_core(open_, high, low, close, signals) # PaperTrader bar-by-bar trader = PaperTrader(initial_capital=100_000) for i in range(n): trader.on_bar(open_[i], high[i], low[i], close[i], signals[i]) eq_paper = np.array(trader.equity_curve) assert eq_paper.shape == eq_rust.shape npt.assert_allclose( eq_paper, eq_rust, rtol=1e-6, atol=1e-9, err_msg="PaperTrader equity curve does not match backtest_ohlcv_core", )