From 815b3ddf7c38b1739da9c08041475b9a511bac94 Mon Sep 17 00:00:00 2001 From: Justin Gu <97915@qq.com> Date: Thu, 11 Jun 2026 01:53:11 +0800 Subject: [PATCH] feat: implement stop-loss/take-profit execution in backtest engine - Track SL/TP conditions from BUY signals in _generate_signals loop - Check active conditions against each bar's high/low price range - Auto-generate SELL signal at trigger price when condition is met - Modify OrderSimulator to respect signal.price for direct execution (previously signal.price was stored but never used in execution) - SL/TP activates on bar AFTER BUY signal (consistent with next_open) - Stop-loss checked before take-profit (conservative for holder) - Add 4 tests: SL trigger, TP trigger, no-trigger, priority over manual sell --- src/easy_tdx/backtest/engine.py | 116 +++++++++++++++++++++++- src/easy_tdx/backtest/orders.py | 24 +++-- tests/unit/test_backtest_engine.py | 139 +++++++++++++++++++++++++++++ 3 files changed, 268 insertions(+), 11 deletions(-) diff --git a/src/easy_tdx/backtest/engine.py b/src/easy_tdx/backtest/engine.py index 2a492bd..e1fa2a3 100644 --- a/src/easy_tdx/backtest/engine.py +++ b/src/easy_tdx/backtest/engine.py @@ -5,6 +5,7 @@ Coordinates Strategy → OrderSimulator → PortfolioTracker → PerformanceAnal from __future__ import annotations +from dataclasses import dataclass from typing import Any import pandas as pd @@ -16,6 +17,19 @@ from easy_tdx.backtest.strategy import Strategy from easy_tdx.backtest.types import BacktestResult, Signal, Trade +@dataclass +class _StopCondition: + """Active stop-loss / take-profit condition tied to an open position. + + Attributes: + stop_loss: Price below which a SELL is triggered (None = disabled) + take_profit: Price above which a SELL is triggered (None = disabled) + """ + + stop_loss: float | None + take_profit: float | None + + class BacktestEngine: """Orchestrate backtest execution pipeline. @@ -171,18 +185,114 @@ class BacktestEngine: # Generate signals bar by bar all_signals: list[Signal] = [] + active_stops: list[_StopCondition] = [] + + high_arr = df["high"].to_numpy() + low_arr = df["low"].to_numpy() + for i in range(len(df)): strat._set_bar_index(i) strat._call_next() bar_signals = strat._clear_signals() - # Update strategy position state so next bar sees current holdings - self._update_strategy_position(strat, bar_signals, close_arr[i]) + # Check existing SL/TP conditions against current bar's price range + sl_tp_signals = self._check_stop_conditions( + active_stops, high_arr[i], low_arr[i], close_arr[i], i, df + ) - all_signals.extend(bar_signals) + # Combine strategy signals with SL/TP signals + combined = bar_signals + sl_tp_signals + + # Register new SL/TP from BUY signals (after checking, so they + # activate on the NEXT bar — consistent with next_open execution) + for sig in bar_signals: + if sig.direction == "BUY" and ( + sig.stop_loss is not None or sig.take_profit is not None + ): + active_stops.append( + _StopCondition(stop_loss=sig.stop_loss, take_profit=sig.take_profit) + ) + + # Clear conditions when a SELL occurs (strategy or SL/TP triggered) + for sig in combined: + if sig.direction == "SELL" and active_stops: + active_stops.clear() + + # Update strategy position state so next bar sees current holdings + self._update_strategy_position(strat, combined, close_arr[i]) + + all_signals.extend(combined) return all_signals + def _check_stop_conditions( + self, + active_stops: list[_StopCondition], + bar_high: float, + bar_low: float, + bar_close: float, + bar_index: int, + df: pd.DataFrame, + ) -> list[Signal]: + """Check active SL/TP conditions against current bar's price range. + + If triggered, generates a SELL signal at the trigger price and removes + the condition. Stop-loss is checked first (conservative: assume the + worst case for the holder). + + Args: + active_stops: List of active stop conditions + bar_high: Current bar's high price + bar_low: Current bar's low price + bar_close: Current bar's close price + bar_index: Current bar index + df: Price DataFrame (for datetime extraction) + + Returns: + List of SELL signals triggered by SL/TP conditions + """ + if not active_stops: + return [] + + signals: list[Signal] = [] + remaining: list[_StopCondition] = [] + + for cond in active_stops: + triggered = False + trigger_price = 0.0 + + # Check stop-loss first (worst case for holder) + if cond.stop_loss is not None and bar_low <= cond.stop_loss: + triggered = True + trigger_price = cond.stop_loss + # Then check take-profit + elif cond.take_profit is not None and bar_high >= cond.take_profit: + triggered = True + trigger_price = cond.take_profit + + if triggered: + # Get datetime for this bar + dt_val = df["datetime"].iloc[bar_index] + if hasattr(dt_val, "strftime"): + dt_int = int(dt_val.strftime("%Y%m%d")) + else: + dt_int = int(dt_val) + + signals.append( + Signal( + datetime=dt_int, + direction="SELL", + size=0, # full position close + price=trigger_price, + ) + ) + else: + remaining.append(cond) + + active_stops.clear() + active_stops.extend(remaining) + return signals + def _update_strategy_position( self, strat: Strategy, signals: list[Signal], est_price: float ) -> None: diff --git a/src/easy_tdx/backtest/orders.py b/src/easy_tdx/backtest/orders.py index b8fee7f..638e948 100644 --- a/src/easy_tdx/backtest/orders.py +++ b/src/easy_tdx/backtest/orders.py @@ -72,15 +72,23 @@ class OrderSimulator: if bar_idx is None: continue - # 确定成交的 K 线索引 - exec_idx = self._resolve_exec_index(bar_idx) - if exec_idx is None or exec_idx >= len(self.df): - continue + # 当信号指定了价格（止损/止盈/限价单）， + # 直接在信号所在 bar 以信号价格成交 + if signal.price is not None: + exec_idx: int = bar_idx + price: float = signal.price + else: + # 确定成交的 K 线索引 + exec_idx_raw = self._resolve_exec_index(bar_idx) + if exec_idx_raw is None or exec_idx_raw >= len(self.df): + continue + exec_idx = exec_idx_raw - # 获取成交价 - price = self._get_price(exec_idx, signal.direction) - if price is None: - continue + # 获取成交价 + price_raw = self._get_price(exec_idx, signal.direction) + if price_raw is None: + continue + price = price_raw # 执行交易 if signal.direction == "BUY": diff --git a/tests/unit/test_backtest_engine.py b/tests/unit/test_backtest_engine.py index 4927961..98b6488 100644 --- a/tests/unit/test_backtest_engine.py +++ b/tests/unit/test_backtest_engine.py @@ -357,3 +357,142 @@ def test_position_aware_no_duplicate_buys(): f"Suspicious tiny buy {cur_size} after {prev_size} — " f"position feedback may be broken" ) + + +# ── Stop-Loss / Take-Profit ────────────────────────────────────────────────── + + +def _make_flat_df(n: int = 30, base_price: float = 100.0) -> pd.DataFrame: + """Generate flat OHLCV data at constant price for SL/TP testing.""" + dates = pd.date_range("2024-01-01", periods=n, freq="D") + return pd.DataFrame( + { + "datetime": dates, + "open": [base_price] * n, + "high": [base_price + 2.0] * n, + "low": [base_price - 2.0] * n, + "close": [base_price] * n, + "vol": [1000000] * n, + "amount": [100000000] * n, + } + ) + + +class StopLossStrategy(Strategy): + """Strategy that buys with stop-loss.""" + + def init(self) -> None: + pass + + def next(self) -> None: + if self._bar_index == 5 and self.position["size"] == 0: + self.buy(size=0, stop_loss=95.0) + + +class TakeProfitStrategy(Strategy): + """Strategy that buys with take-profit.""" + + def init(self) -> None: + pass + + def next(self) -> None: + if self._bar_index == 5 and self.position["size"] == 0: + self.buy(size=0, take_profit=110.0) + + +class StopLossAndTakeProfitStrategy(Strategy): + """Strategy that buys with both stop-loss and take-profit.""" + + def init(self) -> None: + pass + + def next(self) -> None: + if self._bar_index == 5 and self.position["size"] == 0: + self.buy(size=0, stop_loss=95.0, take_profit=110.0) + + +def test_stop_loss_triggers_sell(): + """Test stop-loss triggers auto SELL when price drops below stop.""" + df = _make_flat_df(n=30) + # Bar 12 drops low below stop_loss=95.0 + df.loc[12, "low"] = 93.0 + df.loc[12, "high"] = 96.0 + df.loc[12, "close"] = 94.0 + df.loc[12, "open"] = 97.0 + + engine = BacktestEngine(StopLossStrategy, cash=100000) + result = engine.run(df) + + trades = result.trades[~result.trades["rejected"]] + sell_trades = trades[trades["direction"] == "SELL"] + + # Should have at least one SELL triggered by stop-loss + assert len(sell_trades) >= 1, "Expected stop-loss sell" + # Sell price should be at stop_loss price (95.0) + assert sell_trades.iloc[0]["price"] == 95.0 + + +def test_take_profit_triggers_sell(): + """Test take-profit triggers auto SELL when price rises above target.""" + df = _make_flat_df(n=30) + # Bar 12 rises above take_profit=110.0 + df.loc[12, "high"] = 112.0 + df.loc[12, "low"] = 108.0 + df.loc[12, "close"] = 111.0 + df.loc[12, "open"] = 109.0 + + engine = BacktestEngine(TakeProfitStrategy, cash=100000) + result = engine.run(df) + + trades = result.trades[~result.trades["rejected"]] + sell_trades = trades[trades["direction"] == "SELL"] + + # Should have at least one SELL triggered by take-profit + assert len(sell_trades) >= 1, "Expected take-profit sell" + # Sell price should be at take_profit price (110.0) + assert sell_trades.iloc[0]["price"] == 110.0 + + +def test_stop_loss_not_triggered_when_price_stays_above(): + """Test no SL sell when price never drops to stop level.""" + df = _make_flat_df(n=30, base_price=100.0) + # low is always 98.0 (> stop_loss=95.0), so SL never triggers + + engine = BacktestEngine(StopLossStrategy, cash=100000) + result = engine.run(df) + + trades = result.trades[~result.trades["rejected"]] + sell_trades = trades[trades["direction"] == "SELL"] + + # No SELL should be triggered by SL (low=98 > stop_loss=95) + assert len(sell_trades) == 0, "SL should not trigger when price stays above" + + +def test_stop_loss_takes_priority_over_strategy_sell(): + """SL-triggered sell prevents duplicate strategy sell.""" + df = _make_flat_df(n=30) + + class SLThenManualSell(Strategy): + def init(self) -> None: + pass + + def next(self) -> None: + if self._bar_index == 5 and self.position["size"] == 0: + self.buy(size=0, stop_loss=95.0) + # Manual sell at bar 15 — but SL should have fired first + if self._bar_index == 15 and self.position["size"] > 0: + self.sell(size=0) + + # Bar 10 triggers stop-loss + df.loc[10, "low"] = 93.0 + df.loc[10, "close"] = 94.0 + + engine = BacktestEngine(SLThenManualSell, cash=100000) + result = engine.run(df) + + trades = result.trades[~result.trades["rejected"]] + sell_trades = trades[trades["direction"] == "SELL"] + + # Should have exactly 1 SELL (from SL, not the manual one at bar 15) + assert len(sell_trades) == 1, f"Expected 1 SL sell, got {len(sell_trades)}" + assert sell_trades.iloc[0]["price"] == 95.0