diff --git a/alpaca_paper/client.py b/alpaca_paper/client.py index 9666b6a..3f501e7 100644 --- a/alpaca_paper/client.py +++ b/alpaca_paper/client.py @@ -125,8 +125,34 @@ class AlpacaPaperClient: def cancel_order(self, order_id: str) -> None: self._trading("DELETE", f"/orders/{order_id}") - def close_position(self, symbol: str) -> dict: - return self._trading("DELETE", f"/positions/{symbol}") + def replace_order( + self, order_id: str, stop_price: float | None = None, + limit_price: float | None = None, qty: int | None = None, + ) -> dict: + """PATCH an open order — e.g. move a bracket STOP leg up to breakeven or + trail it — WITHOUT cancelling it. Cancelling a bracket leg tears down the + OCO and leaves the position naked (the bug we already fixed once); a + replace keeps the protection intact and just changes the price.""" + body: dict = {} + if stop_price is not None: + body["stop_price"] = str(round(stop_price, 2)) + if limit_price is not None: + body["limit_price"] = str(round(limit_price, 2)) + if qty is not None: + body["qty"] = str(int(qty)) + if not body: + return {} + return self._trading("PATCH", f"/orders/{order_id}", body=body) + + def close_position(self, symbol: str, qty: int | None = None, + percentage: float | None = None) -> dict: + params = {} + if qty is not None: + params["qty"] = str(int(qty)) + elif percentage is not None: + params["percentage"] = str(round(percentage, 4)) + return self._trading("DELETE", f"/positions/{symbol}", + params=params or None) # -- market data API ---------------------------------------------------- diff --git a/run_live_paper.py b/run_live_paper.py index 7dd3597..46b19a2 100644 --- a/run_live_paper.py +++ b/run_live_paper.py @@ -50,8 +50,10 @@ from research.ingestion.scheduler import Scheduler from research.ingestion.watcher_task import ResearchWatchlistProvider from research.ingestion.discovery import run_discovery, screen_universe from research.multi_schema import open_research_db +from runtime.exit_manager import LiveExitManager from runtime.triggers import ArmedTriggerBook from runtime.watcher import Watcher, WatcherConfig +from strategy.exits import ExitConfig from storage.event_schema import EventMode, ModuleTickEvent from storage.event_store import EventStore from trading_execution import ExecutionSettings, TradingExecutionService @@ -467,7 +469,8 @@ def main(argv: list[str] | None = None) -> int: gappers = scan_gappers( rt["research_con"], session_date, min_gap_pct=0.0, min_relative_volume=0.0, - price_min=price_min_v, price_max=price_max_v, limit=12, + price_min=price_min_v, price_max=price_max_v, + limit=max(12, book.max_armed + 6), ) for g in gappers: bars = rq.query_minute_bars(rt["research_con"], g.symbol, session_date) @@ -559,6 +562,29 @@ def main(argv: list[str] | None = None) -> int: except Exception as exc: # noqa: BLE001 print(f"[trigger] error: {exc}", flush=True) + # --- live exit management (breakeven / trail / scale / first-red) ------- + exit_cfg = ExitConfig.from_env() + + def _rth_bars(sym): + from research import query as rq + b = rq.query_minute_bars(rt["research_con"], sym, _now_session_date()) + if b is not None and len(b) and "is_regular_hours" in b.columns: + b = b[b["is_regular_hours"] == True].reset_index(drop=True) # noqa: E712 + return b + + exit_mgr = ( + LiveExitManager(client, rt["store"], _rth_bars, cfg=exit_cfg, + session_id=rt["session_id"]) + if client else None + ) + print(f"[boot] exit management: {exit_cfg.describe()}") + + def step_manage_exits(): + if exit_mgr is None: + return None + acts = exit_mgr.manage() + return ", ".join(acts) if acts else None + scheduler = Scheduler() scheduler.add("discover", step_discover, float(os.environ.get("DISCOVER_INTERVAL_SECONDS", "300")), @@ -575,13 +601,15 @@ def main(argv: list[str] | None = None) -> int: enabled=_flag("TRADING_EXECUTION_ENABLED", "1")) scheduler.add("guard", step_guard, float(os.environ.get("TRADING_ENTRY_GUARD_INTERVAL_SECONDS", "5")), enabled=_flag("TRADING_EXECUTION_ENABLED", "1")) + scheduler.add("exits", step_manage_exits, float(os.environ.get("TRADING_EXIT_MANAGE_INTERVAL_SECONDS", "12")), + enabled=_flag("TRADING_EXECUTION_ENABLED", "1")) def run_pass(): # force every task due, in pipeline order for task in scheduler.tasks: task.last_run = -10**9 results = {} - for name in ("discover", "ingest", "watch", "arm", "sync", "execute"): + for name in ("discover", "ingest", "watch", "arm", "sync", "execute", "exits"): for task in scheduler.tasks: if task.name == name and task.enabled: results.update(scheduler_run_one(task)) diff --git a/runtime/exit_manager.py b/runtime/exit_manager.py new file mode 100644 index 0000000..48f1b84 --- /dev/null +++ b/runtime/exit_manager.py @@ -0,0 +1,168 @@ +"""Live exit manager — actively manages OPEN positions per strategy/exits. + +The static bracket (fixed stop + fixed target) is replaced/augmented here with +the SAME managed-exit rules the backtest swept (strategy/exits.manage_live): +move the stop to breakeven, trail it, scale out, or exit on a first red candle. + +Critically, the stop is RATCHETED via a broker order REPLACE (PATCH), never a +cancel — cancelling a bracket leg strips the protection (the naked-stop bug). +Runs on the main loop (not the fast trigger thread): stops update off closed +bars, so 10-15s cadence is plenty, and it keeps slow broker calls off the hot +path. +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass + +import pandas as pd + +from storage.event_schema import EventMode, RiskRuleTriggeredEvent +from strategy.exits import ExitConfig, TRAIL_NONE, manage_live + +logger = logging.getLogger(__name__) + + +@dataclass +class _Tracked: + entry_price: float + init_stop: float # original opening-range-low stop (for the R math) + entry_ts: object # timestamp we first saw the position (~entry) + last_stop: float # last stop we set at the broker + scaled: bool = False + + +def _is_active(cfg: ExitConfig) -> bool: + return bool(cfg and (cfg.breakeven_at_r or cfg.trail_mode != TRAIL_NONE + or cfg.scale_out_r or cfg.first_red_exit)) + + +class LiveExitManager: + def __init__(self, client, store, bars_fn, cfg: ExitConfig | None = None, + session_id=None, mode: EventMode = EventMode.PAPER): + self.client = client + self.store = store + self.bars_fn = bars_fn # (symbol) -> today's RTH minute bars DataFrame + self.cfg = cfg or ExitConfig.from_env() + self.session_id = session_id + self.mode = mode + self._tracked: dict[str, _Tracked] = {} + + # -- broker helpers ---------------------------------------------------- + + def _stop_leg(self, symbol: str): + """(order_id, stop_price) of the live sell-stop protecting symbol.""" + try: + orders = self.client.get_orders(status="open", limit=200, nested=True) + except Exception: # noqa: BLE001 + return (None, None) + for o in orders: + for c in [o, *(o.get("legs") or [])]: + sym = c.get("symbol") or o.get("symbol") + if (sym == symbol and c.get("type") in ("stop", "stop_limit") + and c.get("side") == "sell"): + sp = c.get("stop_price") + return (c.get("id"), float(sp) if sp else None) + return (None, None) + + def _emit(self, symbol: str, message: str, rule: str, state: dict) -> None: + from datetime import datetime + self.store.emit(RiskRuleTriggeredEvent( + timestamp=datetime.now(), mode=self.mode, correlation_id=self.session_id, + message=message, rule_type=rule, rule_value=0.0, + current_state={"symbol": symbol, **state}, action_taken=rule, + )) + + # -- main pass --------------------------------------------------------- + + def manage(self) -> list[str]: + if not _is_active(self.cfg): + return [] # static bracket — broker OCO handles it, nothing to do + try: + positions = self.client.get_positions() + except Exception: # noqa: BLE001 + return [] + held = {p.get("symbol"): p for p in positions if p.get("symbol")} + for sym in list(self._tracked): # stop tracking closed names + if sym not in held: + self._tracked.pop(sym, None) + + actions: list[str] = [] + for sym, p in held.items(): + try: + bars = self.bars_fn(sym) + except Exception: # noqa: BLE001 + continue + if bars is None or len(bars) == 0: + continue + entry = float(p.get("avg_entry_price") or 0.0) + if entry <= 0: + continue + if sym not in self._tracked: + _leg, cur_stop = self._stop_leg(sym) + if cur_stop is None or cur_stop >= entry: + continue # no usable protective stop yet (or above entry) + self._tracked[sym] = _Tracked( + entry, cur_stop, _last_ts(bars), cur_stop) + tr = self._tracked[sym] + + since = _bars_since(bars, tr.entry_ts) + if len(since) == 0: + since = bars + d = manage_live(tr.entry_price, tr.init_stop, since, self.cfg, tr.scaled) + + # (1) first-red: exit the whole position now + if d.exit_now: + try: + self.client.close_position(sym) + self._emit(sym, f"first-red exit {sym} @~mkt", "exit_first_red", + {"reason": d.reason}) + actions.append(f"{sym} first-red exit") + self._tracked.pop(sym, None) + continue + except Exception as exc: # noqa: BLE001 + logger.warning("first-red close failed %s: %s", sym, exc) + + # (2) scale-out a fraction into strength + if d.scale_out_frac > 0 and not tr.scaled: + qty = int(abs(float(p.get("qty") or 0))) + sell = int(qty * d.scale_out_frac) + if sell >= 1: + try: + self.client.close_position(sym, qty=sell) + tr.scaled = True + self._emit(sym, f"scaled out {sell} {sym}", "exit_scale_out", + {"qty": sell}) + actions.append(f"{sym} scaled {sell}") + except Exception as exc: # noqa: BLE001 + logger.warning("scale-out failed %s: %s", sym, exc) + + # (3) ratchet the protective stop UP (breakeven / trail) via replace + leg_id, cur_stop = self._stop_leg(sym) + if leg_id and cur_stop is not None and d.desired_stop > cur_stop + 0.01: + try: + self.client.replace_order(leg_id, stop_price=round(d.desired_stop, 2)) + tr.last_stop = d.desired_stop + self._emit(sym, f"stop {sym} {cur_stop:.2f} -> {d.desired_stop:.2f}", + "stop_moved", {"from": cur_stop, "to": d.desired_stop}) + actions.append(f"{sym} stop->{d.desired_stop:.2f}") + except Exception as exc: # noqa: BLE001 + logger.warning("stop move failed %s: %s", sym, exc) + return actions + + +def _last_ts(bars: pd.DataFrame): + try: + return pd.Timestamp(bars["timestamp"].iloc[-1]) + except Exception: # noqa: BLE001 + return None + + +def _bars_since(bars: pd.DataFrame, ts) -> pd.DataFrame: + if ts is None or "timestamp" not in bars.columns: + return bars + try: + return bars[pd.to_datetime(bars["timestamp"]) >= pd.Timestamp(ts)].reset_index(drop=True) + except Exception: # noqa: BLE001 + return bars diff --git a/strategy/backtest/engine.py b/strategy/backtest/engine.py index fbf201f..8f7d5ff 100644 --- a/strategy/backtest/engine.py +++ b/strategy/backtest/engine.py @@ -13,6 +13,7 @@ import pandas as pd from config import BacktestConfig, Config from strategy.evaluation.setup_evaluator import evaluate_setup +from strategy.exits import ExitConfig, simulate_exit from strategy.risk.position_sizing import PositionSizingConfig, calculate_position_size @@ -31,6 +32,7 @@ class Trade: realized_pnl: float | None = None r_multiple: float | None = None setup_type: str | None = None + entry_index: int | None = None # bar position of the entry-fill bar @dataclass @@ -68,6 +70,7 @@ class BacktestEngine: ready_score_pct: float = 60.0, criteria=None, min_bars: int = 10, + exit_config: ExitConfig | None = None, ): self.config = config or Config() self.bt: BacktestConfig = self.config.backtest @@ -78,6 +81,8 @@ class BacktestEngine: self.ready_score_pct = ready_score_pct self.criteria = criteria self.min_bars = min_bars + # default reproduces the original static bracket (stop / +target_r / EOD) + self.exit_config = exit_config or ExitConfig(target_r=target_r) def _entry_costs(self, price: float, quantity: int) -> tuple[float, float]: slip = price * self.bt.slippage.base_spread_pct @@ -95,114 +100,111 @@ class BacktestEngine: ) -> BacktestResult: result = BacktestResult(symbol=symbol) bars = bars.reset_index(drop=True) - open_trade: Trade | None = None + cfg = self.exit_config + n = len(bars) i = self.warmup_bars - while i < len(bars) - 1: - bar = bars.iloc[i] - - if open_trade is not None: - low, high = float(bar["low"]), float(bar["high"]) - exit_price = None - reason = None - if low <= open_trade.stop_price: - exit_price, reason = open_trade.stop_price, "stop_loss" - elif high >= open_trade.target_price: - exit_price, reason = open_trade.target_price, "target" - elif i == len(bars) - 2: - exit_price, reason = float(bar["close"]), "session_end" - if exit_price is not None: - self._close(open_trade, bar, exit_price, reason, result) - open_trade = None + while i < n - 1: + if i % self.eval_every != 0: i += 1 continue - if i % self.eval_every == 0: - window = bars.iloc[: i + 1] - # Evaluate as of the current bar's timestamp, not wall-clock - # now() — otherwise every setup is judged against the real - # clock and marked "late" past the entry cutoff (so a Sunday - # backtest would show zero trades). Bars are stored UTC-naive; - # the cutoff reasons in US/Eastern, so convert. - eval_time = None - try: - from datetime import timezone as _tz - from zoneinfo import ZoneInfo as _ZI + window = bars.iloc[: i + 1] + # Evaluate as of the current bar's timestamp, not wall-clock now() — + # otherwise every setup is judged against the real clock and marked + # "late" past the entry cutoff (so a Sunday backtest would show zero + # trades). Bars are stored UTC-naive; the cutoff reasons in + # US/Eastern, so convert. + eval_time = None + try: + from datetime import timezone as _tz + from zoneinfo import ZoneInfo as _ZI - last_ts = pd.Timestamp(window["timestamp"].iloc[-1]).to_pydatetime() - if last_ts.tzinfo is None: - last_ts = last_ts.replace(tzinfo=_tz.utc) - eval_time = last_ts.astimezone(_ZI("America/New_York")).replace( - tzinfo=None - ) - except Exception: # noqa: BLE001 - fall back to now() - eval_time = None - - evaluation = evaluate_setup( - window, - previous_close=previous_close, - avg_daily_volume=avg_daily_volume, - evaluation_time=eval_time, - ready_score_pct=self.ready_score_pct, - criteria=self.criteria, - min_bars=self.min_bars, + last_ts = pd.Timestamp(window["timestamp"].iloc[-1]).to_pydatetime() + if last_ts.tzinfo is None: + last_ts = last_ts.replace(tzinfo=_tz.utc) + eval_time = last_ts.astimezone(_ZI("America/New_York")).replace( + tzinfo=None ) - result.evaluations += 1 - if evaluation.status == "ready" and evaluation.setups: - result.signals += 1 - setup = evaluation.setups[0] - next_bar = bars.iloc[i + 1] - raw_entry = ( - float(next_bar["open"]) - if self.bt.entry_mode == "next_bar" - else float(bar["close"]) - ) - stop = float(setup["stop_loss_price"]) - sizing = calculate_position_size( - raw_entry, stop, equity=self.equity, - config=PositionSizingConfig(), - ) - if sizing.position_size > 0: - slip, commission = self._entry_costs( - raw_entry, sizing.position_size - ) - entry = raw_entry + slip - risk = entry - stop - open_trade = Trade( - trade_id=str(uuid.uuid4()), - symbol=symbol, - entry_time=str(next_bar.get("timestamp", i + 1)), - entry_price=round(entry, 4), - stop_price=round(stop, 4), - target_price=round(entry + self.target_r * risk, 4), - quantity=sizing.position_size, - setup_type=setup.get("setup_type"), - ) - open_trade.realized_pnl = -commission - i += 1 # consumed next bar for entry - i += 1 + except Exception: # noqa: BLE001 - fall back to now() + eval_time = None - if open_trade is not None: - last = bars.iloc[-1] - self._close(open_trade, last, float(last["close"]), "session_end", result) + evaluation = evaluate_setup( + window, + previous_close=previous_close, + avg_daily_volume=avg_daily_volume, + evaluation_time=eval_time, + ready_score_pct=self.ready_score_pct, + criteria=self.criteria, + min_bars=self.min_bars, + ) + result.evaluations += 1 + if not (evaluation.status == "ready" and evaluation.setups): + i += 1 + continue + + result.signals += 1 + setup = evaluation.setups[0] + entry_idx = i + 1 + next_bar = bars.iloc[entry_idx] + raw_entry = ( + float(next_bar["open"]) + if self.bt.entry_mode == "next_bar" + else float(bars.iloc[i]["close"]) + ) + stop = float(setup["stop_loss_price"]) + sizing = calculate_position_size( + raw_entry, stop, equity=self.equity, config=PositionSizingConfig(), + ) + if sizing.position_size <= 0: + i += 1 + continue + slip, entry_commission = self._entry_costs(raw_entry, sizing.position_size) + entry = raw_entry + slip + risk = entry - stop + if risk <= 0: + i += 1 + continue + + # Simulate the FULL managed exit forward with the shared rules (the + # same logic the live exit manager applies), then resume scanning + # after the exit bar so trades never overlap. + bars_after = bars.iloc[entry_idx + 1:] + res = simulate_exit(entry, stop, bars_after, cfg) + qty = sizing.position_size + gross = res.r_multiple * risk * qty + total_commission = entry_commission + exit_slip = 0.0 + for fll in res.fills: + fqty = max(1, int(round(fll.frac * qty))) + total_commission += max( + self.bt.commission.minimum, fqty * self.bt.commission.per_share + ) + exit_slip += fll.price * self.bt.slippage.base_spread_pct * fqty + realized = gross - total_commission - exit_slip + + exit_pos = min(entry_idx + 1 + res.exit_index, n - 1) if len(bars_after) else entry_idx + trade = Trade( + trade_id=str(uuid.uuid4()), + symbol=symbol, + entry_time=str(next_bar.get("timestamp", entry_idx)), + entry_price=round(entry, 4), + stop_price=round(stop, 4), + target_price=round(entry + cfg.target_r * risk, 4) if cfg.target_r else round(entry, 4), + quantity=qty, + exit_time=str(bars.iloc[exit_pos].get("timestamp", exit_pos)), + exit_price=round(res.fills[-1].price, 4) if res.fills else None, + exit_reason=res.reason, + realized_pnl=round(realized, 2), + r_multiple=round(res.r_multiple, 3), + setup_type=setup.get("setup_type"), + entry_index=entry_idx, + ) + result.trades.append(trade) + result.total_pnl += trade.realized_pnl + i = exit_pos + 1 # resume after the exit if result.trades: wins = sum(1 for t in result.trades if (t.realized_pnl or 0) > 0) result.win_rate = wins / len(result.trades) return result - - def _close( - self, trade: Trade, bar, exit_price: float, reason: str | None, - result: BacktestResult, - ) -> None: - slip, commission = self._entry_costs(exit_price, trade.quantity) - fill = exit_price - slip - gross = (fill - trade.entry_price) * trade.quantity - trade.exit_time = str(bar.get("timestamp", "end")) - trade.exit_price = round(fill, 4) - trade.exit_reason = reason - trade.realized_pnl = round((trade.realized_pnl or 0.0) + gross - commission, 2) - risk = trade.entry_price - trade.stop_price - trade.r_multiple = round((fill - trade.entry_price) / risk, 3) if risk > 0 else 0.0 - result.trades.append(trade) - result.total_pnl += trade.realized_pnl diff --git a/strategy/exits.py b/strategy/exits.py new file mode 100644 index 0000000..3a39511 --- /dev/null +++ b/strategy/exits.py @@ -0,0 +1,235 @@ +"""Trade-exit management — ONE source of truth for the backtest and live. + +The original exit was a static bracket: a fixed stop at the opening-range low +(-1R) and a fixed take-profit at +target_r. Once filled, nothing moved, so a +trade could run to +1.8R and round-trip to a full loss, and losers were always +taken at the full stop. + +This module adds active management as pure, parameterised logic so the exact +same rules drive the backtest (where we sweep them) and the live manager (where +we apply the winner). Rules, each independently toggleable: + + * breakeven — once price reaches +breakeven_at_r, move the stop to entry + * trail — after +trail_after_r, trail the stop under each bar's low + (prior_low) or a % off the high-water mark (pct) + * scale-out — sell scale_out_pct of the position at +scale_out_r, run the rest + * first-red — exit on the first bar that closes below the prior bar's low + * target_r — final take-profit (whatever remains) + +Intrabar convention is conservative: within a bar the STOP is assumed to fill +before the target/scale, so the backtest never flatters a trade. +""" + +from __future__ import annotations + +import os +from dataclasses import dataclass, field + +import pandas as pd + +TRAIL_NONE = "none" +TRAIL_PRIOR_LOW = "prior_low" +TRAIL_PCT = "pct" + + +@dataclass +class ExitConfig: + target_r: float = 2.0 # final take-profit in R (0 = no fixed target) + breakeven_at_r: float = 0.0 # move stop to entry once +this R is reached (0 = off) + trail_mode: str = TRAIL_NONE # none | prior_low | pct + trail_pct: float = 0.0 # pct trail: stop = high_water * (1 - trail_pct) + trail_after_r: float = 1.0 # only start trailing once +this R is reached + scale_out_r: float = 0.0 # sell scale_out_pct at +this R (0 = off) + scale_out_pct: float = 0.5 # fraction sold at scale_out_r + first_red_exit: bool = False # exit on first close below the prior bar's low + + @classmethod + def from_env(cls, env: dict | None = None) -> "ExitConfig": + v = dict(os.environ) + if env: + v.update(env) + + def f(key, default): + try: + return float(v.get(key, default)) + except (TypeError, ValueError): + return float(default) + + def b(key, default): + return v.get(key, default).strip().lower() in {"1", "true", "yes", "on"} + + return cls( + target_r=f("TRADING_REWARD_MULTIPLE", "2.0"), + breakeven_at_r=f("TRADING_EXIT_BREAKEVEN_R", "0.0"), + trail_mode=v.get("TRADING_EXIT_TRAIL_MODE", TRAIL_NONE).strip().lower(), + trail_pct=f("TRADING_EXIT_TRAIL_PCT", "0.0"), + trail_after_r=f("TRADING_EXIT_TRAIL_AFTER_R", "1.0"), + scale_out_r=f("TRADING_EXIT_SCALE_OUT_R", "0.0"), + scale_out_pct=f("TRADING_EXIT_SCALE_OUT_PCT", "0.5"), + first_red_exit=b("TRADING_EXIT_FIRST_RED", "0"), + ) + + def describe(self) -> str: + parts = [f"target {self.target_r:g}R"] + if self.breakeven_at_r: + parts.append(f"BE@{self.breakeven_at_r:g}R") + if self.trail_mode != TRAIL_NONE: + how = f"{self.trail_pct:.1%}" if self.trail_mode == TRAIL_PCT else "prior-low" + parts.append(f"trail {how} after {self.trail_after_r:g}R") + if self.scale_out_r: + parts.append(f"scale {self.scale_out_pct:.0%}@{self.scale_out_r:g}R") + if self.first_red_exit: + parts.append("first-red") + return ", ".join(parts) + + +@dataclass +class ExitFill: + frac: float # fraction of the original position closed by this fill + price: float + reason: str + + +@dataclass +class ExitResult: + r_multiple: float # realized R, weighted across partial fills + reason: str # reason of the FINAL (position-closing) fill + exit_index: int # index in bars_after where the position fully closed + fills: list[ExitFill] = field(default_factory=list) + + +def _trail_stop(stop: float, entry: float, high_water: float, + bar_low: float, reached_r: float, cfg: ExitConfig) -> float: + """The stop after processing one bar — only ever ratchets UP.""" + new = stop + if cfg.breakeven_at_r and reached_r >= cfg.breakeven_at_r: + new = max(new, entry) + if reached_r >= cfg.trail_after_r: + if cfg.trail_mode == TRAIL_PRIOR_LOW: + new = max(new, bar_low) + elif cfg.trail_mode == TRAIL_PCT and cfg.trail_pct > 0: + new = max(new, high_water * (1.0 - cfg.trail_pct)) + return new + + +def simulate_exit( + entry_price: float, init_stop: float, bars_after: pd.DataFrame, cfg: ExitConfig +) -> ExitResult: + """Walk the bars AFTER entry and return the realized R for the whole trade. + + ``bars_after`` are the OHLC bars strictly after the entry bar, in order. + The same per-bar rules are used live (see ``manage_live``). + """ + risk = entry_price - init_stop + if risk <= 0 or bars_after is None or len(bars_after) == 0: + return ExitResult(0.0, "invalid", 0, []) + + target = entry_price + cfg.target_r * risk if cfg.target_r else None + stop = init_stop + high_water = entry_price + remaining = 1.0 + realized_r = 0.0 + scaled = False + fills: list[ExitFill] = [] + prev_low: float | None = None + + n = len(bars_after) + for idx in range(n): + bar = bars_after.iloc[idx] + high = float(bar["high"]); low = float(bar["low"]); close = float(bar["close"]) + high_water = max(high_water, high) + reached_r = (high_water - entry_price) / risk + + # (1) stop first (conservative intrabar ordering) + if low <= stop: + r = (stop - entry_price) / risk + realized_r += remaining * r + reason = "breakeven" if abs(stop - entry_price) < 1e-9 else ( + "trail_stop" if stop > init_stop else "stop_loss") + fills.append(ExitFill(remaining, stop, reason)) + return ExitResult(realized_r, reason, idx, fills) + + # (2) scale-out into strength + if (cfg.scale_out_r and not scaled + and high >= entry_price + cfg.scale_out_r * risk): + frac = min(cfg.scale_out_pct, remaining) + realized_r += frac * cfg.scale_out_r + remaining -= frac + scaled = True + fills.append(ExitFill(frac, entry_price + cfg.scale_out_r * risk, "scale_out")) + if remaining <= 1e-9: + return ExitResult(realized_r, "scale_out", idx, fills) + + # (3) final target on the remainder + if target is not None and high >= target: + realized_r += remaining * cfg.target_r + fills.append(ExitFill(remaining, target, "target")) + return ExitResult(realized_r, "target", idx, fills) + + # (4) first red candle that breaks the prior bar's low + if cfg.first_red_exit and prev_low is not None and close < prev_low: + r = (close - entry_price) / risk + realized_r += remaining * r + fills.append(ExitFill(remaining, close, "first_red")) + return ExitResult(realized_r, "first_red", idx, fills) + + # (5) ratchet the stop for the NEXT bar + stop = _trail_stop(stop, entry_price, high_water, low, reached_r, cfg) + prev_low = low + + # session end on the remainder + last_close = float(bars_after.iloc[-1]["close"]) + realized_r += remaining * ((last_close - entry_price) / risk) + fills.append(ExitFill(remaining, last_close, "session_end")) + return ExitResult(realized_r, "session_end", n - 1, fills) + + +@dataclass +class ExitDecision: + desired_stop: float # where the protective stop SHOULD be now + scale_out_frac: float # fraction to sell at market now (0 if none) + exit_now: bool # full market exit now (first-red) + reason: str + + +def manage_live( + entry_price: float, init_stop: float, bars_since_entry: pd.DataFrame, + cfg: ExitConfig, already_scaled: bool = False, +) -> ExitDecision: + """Compute the management action for an OPEN live position. + + Uses the same ratchet/scale/first-red rules as ``simulate_exit`` so the live + behaviour matches the backtest. ``bars_since_entry`` are the closed bars from + the entry bar onward. Returns the stop the broker order should be moved to + (only ever up), whether to scale out now, and whether to exit outright. + """ + risk = entry_price - init_stop + if risk <= 0 or bars_since_entry is None or len(bars_since_entry) == 0: + return ExitDecision(init_stop, 0.0, False, "hold") + + stop = init_stop + high_water = entry_price + for idx in range(len(bars_since_entry)): + bar = bars_since_entry.iloc[idx] + high_water = max(high_water, float(bar["high"])) + reached_r = (high_water - entry_price) / risk + stop = _trail_stop(stop, entry_price, high_water, float(bar["low"]), + reached_r, cfg) + + last = bars_since_entry.iloc[-1] + last_close = float(last["close"]) + reached_r = (high_water - entry_price) / risk + + # first-red: latest bar closed below the prior bar's low + if cfg.first_red_exit and len(bars_since_entry) >= 2: + prior_low = float(bars_since_entry.iloc[-2]["low"]) + if last_close < prior_low and reached_r >= cfg.trail_after_r: + return ExitDecision(stop, 0.0, True, "first_red") + + # scale-out: high reached the scale level and we haven't yet + scale_frac = 0.0 + if (cfg.scale_out_r and not already_scaled + and high_water >= entry_price + cfg.scale_out_r * risk): + scale_frac = cfg.scale_out_pct + + return ExitDecision(stop, scale_frac, False, "trail") diff --git a/sweep_exits.py b/sweep_exits.py new file mode 100644 index 0000000..1147537 --- /dev/null +++ b/sweep_exits.py @@ -0,0 +1,177 @@ +#!/usr/bin/env python3 +"""Sweep trade-EXIT variants over the stored gapper sessions. + +Entry params are held FIXED at the live ORB config (so we isolate the effect of +the exit rules), and each variant backtests the SAME shared exit logic the live +manager uses (strategy/exits.py) — so a variant that wins here transfers live. + +Sessions are split by date into TRAIN (earlier) and TEST (later) so a winner can +be checked out-of-sample (guards against overfitting to a handful of days). + + PYTHONPATH=. python sweep_exits.py --list-names + PYTHONPATH=. python sweep_exits.py --split train --json + PYTHONPATH=. python sweep_exits.py --split test --names static_2R,be1_t3 --json +""" + +from __future__ import annotations + +import argparse +import json +import os +from collections import Counter + +from dotenv import load_dotenv + +from config import Config +from research.multi_schema import open_research_db +from strategy.backtest.engine import BacktestEngine +from strategy.exits import ExitConfig, TRAIL_PRIOR_LOW, TRAIL_PCT, simulate_exit +from strategy.models import SetupCriteria + +load_dotenv() +os.environ["STRATEGY_SETUPS"] = "opening_range_break" # isolate the ORB + +# Fixed ORB entry config (mirrors the live learned params) so only EXITS vary. +ENTRY = dict(warmup=3, eval_every=1, ready=55, min_bars=3, gap_min=0.05, rvol_min=2.0) + + +def variants() -> dict[str, ExitConfig]: + """Named exit variants to sweep. Keep this the single grid definition.""" + v: dict[str, ExitConfig] = {} + # --- fixed bracket targets (the current style) --- + for r in (1.5, 2.0, 2.5, 3.0): + v[f"static_{r:g}R"] = ExitConfig(target_r=r) + # --- move to breakeven, then let a far target run --- + for be in (0.5, 0.75, 1.0): + v[f"be{be:g}_t3"] = ExitConfig(target_r=3.0, breakeven_at_r=be) + # --- trail under prior-bar lows (target far so the trail is the exit) --- + for after in (0.5, 1.0, 1.5): + v[f"trailLow_a{after:g}"] = ExitConfig( + target_r=10.0, trail_mode=TRAIL_PRIOR_LOW, trail_after_r=after) + # --- percent trailing off the high-water mark --- + for pct in (0.04, 0.06, 0.10): + v[f"trailPct{int(pct*100)}"] = ExitConfig( + target_r=10.0, trail_mode=TRAIL_PCT, trail_pct=pct, trail_after_r=1.0) + # --- scale out into strength, run the rest --- + v["scale50_1R_t3"] = ExitConfig(target_r=3.0, scale_out_r=1.0, scale_out_pct=0.5) + v["scale50_1.5R_t4"] = ExitConfig(target_r=4.0, scale_out_r=1.5, scale_out_pct=0.5) + v["scale33_1R_trailLow"] = ExitConfig( + target_r=10.0, scale_out_r=1.0, scale_out_pct=0.34, + trail_mode=TRAIL_PRIOR_LOW, trail_after_r=1.0) + # --- first red candle exits --- + v["firstRed_t3"] = ExitConfig(target_r=3.0, first_red_exit=True, trail_after_r=0.0) + v["be1_firstRed_t4"] = ExitConfig( + target_r=4.0, breakeven_at_r=1.0, first_red_exit=True, trail_after_r=1.0) + # --- composites: breakeven + trail (+ scale) --- + v["be1_trailLow_t4"] = ExitConfig( + target_r=4.0, breakeven_at_r=1.0, trail_mode=TRAIL_PRIOR_LOW, trail_after_r=1.0) + v["scale50_be1_trailLow"] = ExitConfig( + target_r=10.0, breakeven_at_r=1.0, scale_out_r=1.0, scale_out_pct=0.5, + trail_mode=TRAIL_PRIOR_LOW, trail_after_r=1.0) + return v + + +def load_split(con, which: str): + """Return (train, test, all) session lists split by date (70/30 by day).""" + from learn_params import load_sessions + sessions = load_sessions(con) + days = sorted({s for _, s, _, _, _ in sessions}) + cut = days[int(len(days) * 0.7)] if len(days) >= 4 else (days[-1] if days else None) + train = [s for s in sessions if s[1] < cut] if cut else sessions + test = [s for s in sessions if s[1] >= cut] if cut else [] + if which == "train": + return train or sessions + if which == "test": + return test or sessions + return sessions + + +def extract_entries(sessions) -> list[tuple]: + """Run the entry detection ONCE (with a reference static exit) and return the + fixed entry set [(entry_price, init_stop, qty, bars_after), ...]. Every exit + variant is then scored against the SAME entries — far faster than re-running + the expensive per-bar evaluation per variant, and the fair comparison (same + trades, only the management differs). + """ + ref = BacktestEngine( + Config(), target_r=2.0, warmup_bars=ENTRY["warmup"], + eval_every=ENTRY["eval_every"], ready_score_pct=ENTRY["ready"], + min_bars=ENTRY["min_bars"], exit_config=ExitConfig(target_r=2.0), + criteria=SetupCriteria(gap_pct_min=ENTRY["gap_min"], + relative_volume_min=ENTRY["rvol_min"]), + ) + entries: list[tuple] = [] + for symbol, _sess, bars, pc, adv in sessions: + bars_r = bars.reset_index(drop=True) + for t in ref.run(bars, symbol, previous_close=pc, avg_daily_volume=adv).trades: + if t.entry_index is None: + continue + bars_after = bars_r.iloc[t.entry_index + 1:] + entries.append((t.entry_price, t.stop_price, t.quantity, bars_after)) + return entries + + +def run_variant(entries, cfg: ExitConfig) -> dict: + rs, pnls, reasons = [], [], Counter() + for entry, stop, qty, bars_after in entries: + risk = entry - stop + if risk <= 0: + continue + res = simulate_exit(entry, stop, bars_after, cfg) + rs.append(res.r_multiple) + pnls.append(res.r_multiple * risk * qty) # gross; costs ~const across variants + for f in res.fills: + reasons[f.reason] += 1 + n = len(rs) + if not n: + return {"trades": 0, "win": 0.0, "pnl": 0.0, "avgR": 0.0, "reasons": {}} + return { + "trades": n, + "win": round(sum(1 for r in rs if r > 0) / n, 3), + "pnl": round(sum(pnls), 0), + "avgR": round(sum(rs) / n, 3), + "reasons": dict(reasons), + } + + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument("--split", default="all", choices=["train", "test", "all"]) + ap.add_argument("--names", default="", help="comma-separated variant names (default: all)") + ap.add_argument("--list-names", action="store_true") + ap.add_argument("--json", action="store_true") + args = ap.parse_args() + + grid = variants() + if args.list_names: + print("\n".join(grid)) + return + + con = open_research_db("market") + sessions = load_split(con, args.split) + entries = extract_entries(sessions) + names = [s.strip() for s in args.names.split(",") if s.strip()] or list(grid) + out = [] + for name in names: + cfg = grid.get(name) + if cfg is None: + continue + m = run_variant(entries, cfg) + m["name"] = name + m["desc"] = cfg.describe() + out.append(m) + out.sort(key=lambda m: m["avgR"], reverse=True) + + if args.json: + print(json.dumps({"split": args.split, "n_sessions": len(sessions), + "n_entries": len(entries), "results": out})) + return + print(f"=== exit sweep ({args.split}, {len(sessions)} sessions, {len(entries)} entries) ===") + print(f"{'variant':22s} {'trades':>6} {'win%':>5} {'avgR':>6} {'pnl':>9} desc") + for m in out: + print(f"{m['name']:22s} {m['trades']:>6} {m['win']*100:>4.0f}% {m['avgR']:>+6.3f} " + f"{m['pnl']:>9.0f} {m['desc']}") + + +if __name__ == "__main__": + main() diff --git a/tests/unit/test_exit_manager.py b/tests/unit/test_exit_manager.py new file mode 100644 index 0000000..fea8863 --- /dev/null +++ b/tests/unit/test_exit_manager.py @@ -0,0 +1,70 @@ +"""Live exit manager: ratchet the stop UP via replace (never cancel = never naked).""" + +import pandas as pd + +from runtime.exit_manager import LiveExitManager +from storage.event_store import EventStore +from strategy.exits import ExitConfig, TRAIL_PRIOR_LOW + + +class _FakeBroker: + def __init__(self, positions, stop_leg): + self._positions = positions + self._stop_leg = stop_leg + self.replaced = [] + self.closed = [] + self.canceled = [] + + def get_positions(self): + return self._positions + + def get_orders(self, status="open", limit=200, nested=True): + return [self._stop_leg] + + def replace_order(self, order_id, stop_price=None, **kw): + self.replaced.append((order_id, stop_price)) + return {"id": order_id} + + def cancel_order(self, order_id): + self.canceled.append(order_id) + + def close_position(self, symbol, qty=None, percentage=None): + self.closed.append((symbol, qty)) + return {"id": "c"} + + +def _bars(seq): + base = pd.Timestamp("2026-06-17 13:30:00") + return pd.DataFrame([ + {"timestamp": base + pd.Timedelta(minutes=i), "high": h, "low": lo, "close": c} + for i, (h, lo, c) in enumerate(seq) + ]) + + +def test_manager_ratchets_stop_up_and_never_cancels(): + store = EventStore(":memory:") + pos = [{"symbol": "AAA", "avg_entry_price": "10.0", "qty": "100"}] + leg = {"id": "stop1", "symbol": "AAA", "type": "stop", "side": "sell", "stop_price": "9.0"} + broker = _FakeBroker(pos, leg) + bars = _bars([(10.5, 9.8, 10.4), (11.5, 10.5, 11.4), (12.2, 11.5, 12.0)]) + mgr = LiveExitManager( + broker, store, lambda s: bars, + cfg=ExitConfig(breakeven_at_r=1.0, trail_mode=TRAIL_PRIOR_LOW, trail_after_r=1.0), + session_id="t", + ) + mgr.manage() + assert broker.replaced, "expected the stop to be moved up" + order_id, new_stop = broker.replaced[-1] + assert order_id == "stop1" and new_stop > 9.0 # ratcheted UP + assert broker.canceled == [] # never cancels the protective leg + + +def test_manager_noop_for_static_bracket(): + # no active rules -> the broker OCO handles everything; manager does nothing + broker = _FakeBroker([{"symbol": "AAA", "avg_entry_price": "10", "qty": "100"}], + {"id": "s", "symbol": "AAA", "type": "stop", "side": "sell", + "stop_price": "9"}) + mgr = LiveExitManager(broker, EventStore(":memory:"), lambda s: _bars([(11, 10, 10.9)]), + cfg=ExitConfig(target_r=2.0), session_id="t") + assert mgr.manage() == [] + assert not broker.replaced and not broker.closed diff --git a/tests/unit/test_exits.py b/tests/unit/test_exits.py new file mode 100644 index 0000000..54ed22e --- /dev/null +++ b/tests/unit/test_exits.py @@ -0,0 +1,82 @@ +"""Shared exit logic: simulate_exit (backtest) and manage_live must agree. + +Locks in the rule behaviours we sweep and then run live: fixed target, move to +breakeven, trail under prior lows / by percent, scale-out, and first-red exit. +""" + +import pandas as pd + +from strategy.exits import ( + ExitConfig, + TRAIL_PCT, + TRAIL_PRIOR_LOW, + manage_live, + simulate_exit, +) + + +def _bars(seq): + return pd.DataFrame([{"high": h, "low": lo, "close": c} for h, lo, c in seq]) + + +# runs 10 -> 12.2, then collapses to 9.4 on the last bar +RUN_THEN_DROP = _bars([(10.5, 9.8, 10.4), (11.0, 10.2, 10.9), + (12.2, 11.5, 12.0), (12.0, 9.4, 9.6)]) + + +def test_static_target_hits_2r(): + r = simulate_exit(10.0, 9.0, RUN_THEN_DROP, ExitConfig(target_r=2.0)) + assert r.reason == "target" + assert round(r.r_multiple, 2) == 2.0 + + +def test_breakeven_saves_a_runner_from_full_loss(): + # far target so it can't take profit; breakeven@1R moves stop to entry, + # so the final collapse exits at ~0R instead of -1R + r = simulate_exit(10.0, 9.0, RUN_THEN_DROP, + ExitConfig(target_r=5.0, breakeven_at_r=1.0)) + assert r.reason == "breakeven" + assert abs(r.r_multiple) < 1e-6 + + +def test_trail_prior_low_captures_partial_gain(): + r = simulate_exit(10.0, 9.0, RUN_THEN_DROP, + ExitConfig(target_r=5.0, trail_mode=TRAIL_PRIOR_LOW, trail_after_r=1.0)) + assert r.reason == "trail_stop" + assert r.r_multiple > 1.0 # locked in more than breakeven + + +def test_scale_out_blends_realized_r(): + r = simulate_exit(10.0, 9.0, RUN_THEN_DROP, + ExitConfig(target_r=3.0, scale_out_r=1.0, scale_out_pct=0.5)) + # half booked at +1R, the rest rides to session end (close 9.6 = -0.4R) + assert any(f.reason == "scale_out" for f in r.fills) + assert round(r.r_multiple, 2) == round(0.5 * 1.0 + 0.5 * -0.4, 2) + + +def test_first_red_exits_on_break_of_prior_low(): + bars = _bars([(10.5, 10.1, 10.4), (11.0, 10.6, 10.9), (10.8, 9.9, 10.0)]) + # bar 3 closes 10.0 < bar 2 low 10.6 -> first red exit + r = simulate_exit(10.0, 9.0, bars, ExitConfig(target_r=9.0, first_red_exit=True)) + assert r.reason == "first_red" + + +def test_stop_is_conservative_intrabar(): + # a bar whose range spans BOTH stop and target -> assume stop fills first + bars = _bars([(12.5, 8.9, 10.0)]) + r = simulate_exit(10.0, 9.0, bars, ExitConfig(target_r=2.0)) + assert r.reason == "stop_loss" + assert round(r.r_multiple, 2) == -1.0 + + +def test_manage_live_ratchets_stop_to_breakeven(): + bars = _bars([(11.2, 10.1, 11.0)]) # reached +1.2R + d = manage_live(10.0, 9.0, bars, ExitConfig(breakeven_at_r=1.0)) + assert d.desired_stop >= 10.0 # moved up to (at least) entry + + +def test_manage_live_pct_trail(): + bars = _bars([(12.0, 10.5, 11.8)]) # high-water 12 + d = manage_live(10.0, 9.0, bars, + ExitConfig(trail_mode=TRAIL_PCT, trail_pct=0.05, trail_after_r=1.0)) + assert round(d.desired_stop, 2) == round(12.0 * 0.95, 2)