matitos
9dc00eba95
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.
66 lines
1.9 KiB
Python
66 lines
1.9 KiB
Python
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")
|