Add trailing stops and parameter sweeps
CI/CD / backend (pull_request) Successful in 6m6s
CI/CD / frontend (pull_request) Successful in 5m17s
CI/CD / build (pull_request) Skipped
CI/CD / deploy (pull_request) Skipped
CI/CD / cleanup (pull_request) Skipped

Trailing stops: strategies can pass trailing_stop_pct to enter_long/enter_short.
advance_trailing_stop in app/engine/portfolio.py ratchets the stop with the price
(up for longs, down for shorts) and never loosens; the engine persists the
ratcheted level and the backtester applies the same logic. Position gains a
trailing_stop_pct column (additive migration). momentum exposes it as a param.

Parameter sweeps: app/backtest/sweep.py runs a cartesian grid through the same
backtester and ranks by return minus max drawdown. POST /api/sweep plus a
dashboard Parameter sweep panel.
This commit is contained in:
matitos committed 2026-09-19 23:04:20 +00:00
1 parent d80e7b155e
commit 9dc00eba95
22 files changed
+575 -7

No files matched your search

+26
View File
@@ -62,6 +62,32 @@ def test_backtest_endpoint(runtime: Runtime, market: FakeMarketData) -> None:
assert data["metrics"]["num_trades"] >= 1
def test_sweep_endpoint(runtime: Runtime, market: FakeMarketData) -> None:
market.candles = FakeMarketData.sine_wave(200).candles
app = create_app(runtime.settings, runtime)
with TestClient(app) as client:
response = client.post(
"/api/sweep",
json={
"strategy_key": "sma_cross",
"grid": {"fast": [3, 5], "slow": [10, 20]},
"limit": 200,
},
)
assert response.status_code == 200
data = response.json()
assert data["combinations"] == 4
assert 1 <= len(data["rows"]) <= 4
assert "total_return_pct" in data["rows"][0]["metrics"]
def test_sweep_unknown_strategy_404(runtime: Runtime) -> None:
app = create_app(runtime.settings, runtime)
with TestClient(app) as client:
response = client.post("/api/sweep", json={"strategy_key": "nope"})
assert response.status_code == 404
def test_backtest_unknown_strategy_404(runtime: Runtime) -> None:
app = create_app(runtime.settings, runtime)
with TestClient(app) as client:
+30
View File
@@ -82,6 +82,36 @@ def test_stop_loss_is_honoured() -> None:
assert result.metrics.final_equity < 10_000
class TrailStrategy(Strategy):
key: ClassVar[str] = "test_trail"
name: ClassVar[str] = "Test Trail"
default_params: ClassVar[dict] = {}
def on_candle(self, ctx: StrategyContext) -> None:
if ctx.is_flat and not getattr(self, "_entered", False):
self._entered = True
ctx.enter_long(size=1.0, trailing_stop_pct=0.10)
def test_trailing_stop_locks_in_profit() -> None:
# Entry at 100, rise to 150, then a reversal. A 10% trail should exit near
# 135 (150 * 0.9) instead of round-tripping to the entry.
backtester = Backtester(fee_bps=0, slippage_bps=0, starting_balance=10_000)
candles = candles_from_closes([100, 120, 150, 130, 125, 120])
candles[2] = Candle(ts=2, open=120, high=150, low=120, close=150, volume=1.0)
candles[3] = Candle(ts=3, open=150, high=150, low=130, close=130, volume=1.0)
candles[4] = Candle(ts=4, open=130, high=132, low=125, close=125, volume=1.0)
candles[5] = Candle(ts=5, open=125, high=128, low=120, close=120, volume=1.0)
result = backtester.run(candles, TrailStrategy(), "BTC/USDT", "1h")
assert result.metrics.num_trades == 1
trade = result.trades[0]
assert "stop loss" in trade.reason
# 99 units, exit at the trailing stop (135), well above entry 100.
assert trade.exit_price == pytest.approx(135.0)
assert result.metrics.final_equity > 10_000
def test_strategy_error_is_surfaced() -> None:
class Boom(Strategy):
key = "test_boom"
+28
View File
@@ -4,6 +4,7 @@ import pytest
from app.engine.portfolio import (
account_equity,
advance_trailing_stop,
apply_close,
apply_open,
order_qty,
@@ -57,3 +58,30 @@ def test_risk_exit_short() -> None:
assert risk_exit(PositionSide.SHORT, 110, 90, high=111, low=100) == ("stop loss", 110)
assert risk_exit(PositionSide.SHORT, 110, 90, high=100, low=89) == ("take profit", 90)
assert risk_exit(PositionSide.SHORT, None, None, high=100, low=95) is None
def test_trailing_stop_long_ratchets_up_only() -> None:
# First update seeds the stop 2% below the high.
stop = advance_trailing_stop(PositionSide.LONG, None, 0.02, high=100, low=95)
assert stop == pytest.approx(98.0)
# A higher high raises it.
stop = advance_trailing_stop(PositionSide.LONG, stop, 0.02, high=110, low=100)
assert stop == pytest.approx(107.8)
# A pullback never loosens it.
stop = advance_trailing_stop(PositionSide.LONG, stop, 0.02, high=105, low=100)
assert stop == pytest.approx(107.8)
def test_trailing_stop_short_ratchets_down_only() -> None:
stop = advance_trailing_stop(PositionSide.SHORT, None, 0.02, high=100, low=95)
assert stop == pytest.approx(96.9)
stop = advance_trailing_stop(PositionSide.SHORT, stop, 0.02, high=95, low=90)
assert stop == pytest.approx(91.8)
# A bounce never loosens the short's stop.
stop = advance_trailing_stop(PositionSide.SHORT, stop, 0.02, high=95, low=92)
assert stop == pytest.approx(91.8)
def test_trailing_stop_disabled_is_noop() -> None:
assert advance_trailing_stop(PositionSide.LONG, 90.0, 0.0, high=100, low=80) == 90.0
assert advance_trailing_stop(PositionSide.LONG, None, 0.0, high=100, low=80) is None
+65
View File
@@ -0,0 +1,65 @@
from __future__ import annotations
from app.backtest.sweep import expand_grid, run_sweep
from app.schemas import Candle
def candles_from_closes(closes: list[float]) -> list[Candle]:
return [Candle(ts=i, open=c, high=c, low=c, close=c, volume=1.0) for i, c in enumerate(closes)]
def test_expand_grid() -> None:
grid = {"fast": [2, 3], "slow": [10, 20]}
combos = expand_grid(grid)
assert len(combos) == 4
assert {"fast": 2, "slow": 10} in combos
assert {"fast": 3, "slow": 20} in combos
def test_expand_grid_empty() -> None:
assert expand_grid({}) == [{}]
def test_expand_grid_single_axis() -> None:
assert expand_grid({"x": [1, 2, 3]}) == [{"x": 1}, {"x": 2}, {"x": 3}]
def test_run_sweep_ranks_and_merges_base_params() -> None:
# A trending series so SMA crossover trades.
prices = [100 + i for i in range(60)]
rows = run_sweep(
"sma_cross",
base_params={"size": 1.0},
grid={"fast": [2, 5], "slow": [10, 20]},
candles=candles_from_closes(prices),
symbol="BTC/USDT",
timeframe="1h",
top_n=10,
)
assert len(rows) == 4
# Every row carries the merged base params.
assert all(row.params["size"] == 1.0 for row in rows)
# Sorted by composite score descending.
scores = [row.metrics["total_return_pct"] - row.metrics["max_drawdown_pct"] for row in rows]
assert scores == sorted(scores, reverse=True)
def test_run_sweep_top_n_truncates() -> None:
prices = [100 + i for i in range(60)]
rows = run_sweep(
"sma_cross",
base_params={},
grid={"fast": [2, 3, 4], "slow": [10, 15, 20]},
candles=candles_from_closes(prices),
symbol="BTC/USDT",
timeframe="1h",
top_n=3,
)
assert len(rows) == 3
def test_run_sweep_unknown_strategy() -> None:
import pytest
with pytest.raises(KeyError):
run_sweep("nope", {}, {}, candles_from_closes([1, 2, 3]), "BTC/USDT", "1h")